Merge branch 'fix/pending-embed-low-water' into rel/10.4.9-candidate

src/brainy.ts

@@ -1820,16 +1820,22 @@ export class Brainy<T = any> implements BrainyInterface<T> {
       // a deferred write's ack and its background embed DELAYED a vector;
       // this is where it lands.
       if (!this.isReadOnly) {
-        // BEHIND THE DOORS (the open pays nothing here): the bridge + the
+        // Foreground, as the crash-recovery contract pins it: a reopened brain
-        // recovery fold run as one latched background task; the embed worker
+        // has its markers re-armed when open() returns. The low-water mark
-        // starts when it settles. A pending embed's outcome was always
+        // bounds this to the log's tail on any brain that has ever drained —
-        // eventual — moving its recovery off the open's foreground changes
+        // milliseconds — so the foreground cost is the unmarked first open
-        // when the worker starts, never whether a marker is honored.
+        // only, once per upgraded brain.
-        // awaitPendingEmbeds() and close() wait on the latch first.
-        this._pendingEmbedRecovery = (async () => {
         try {
-            await this.bridgeLegacyPendingEmbedSidecars()
+          await step(
-            await this.recoverPendingEmbedsFromLog()
+            'bridge-pending-embed-sidecars',
+            'migrating any pre-log deferred-embed marker files into the generation log',
+            () => this.bridgeLegacyPendingEmbedSidecars()
+          )
+          await step(
+            'recover-pending-embeds',
+            'folding the generation log\'s deferred-embed markers (from the low-water mark) into the pending set',
+            () => this.recoverPendingEmbedsFromLog()
+          )
           if (this._pendingEmbedIds.size > 0) {
             prodLog.info(
               `[Brainy] ${this._pendingEmbedIds.size} deferred embed(s) pending from a previous ` +
@@ -1844,7 +1850,6 @@ export class Brainy<T = any> implements BrainyInterface<T> {
               `the log's markers remain durable; recovery retries next open`
           )
         }
-        })()
       }
 
       // PHASE 4 of 5 — "VFS bootstrap": shutdown-hook registration, blob
@@ -2418,8 +2423,6 @@ export class Brainy<T = any> implements BrainyInterface<T> {
    */
   private static readonly PENDING_EMBED_LOWWATER_PATH = '_system/pending_embeds_lowwater.json'
 
-  /** Resolves when the background pending-embed recovery fold has settled (open arms it). */
-  private _pendingEmbedRecovery: Promise<void> | null = null
 
   /**
    * @description Mark a deferred embed pending (MT5): the id joins the
@@ -2497,9 +2500,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
    * pending set last drained to empty — so a settled brain reads only the
    * facts since then, not its whole history. Without a mark (first open
    * after upgrade) it scans from generation 1, once; a stale-low mark costs
-   * a longer scan, never a marker. The fold runs BEHIND the doors (open
+   * a longer scan, never a marker. The fold stays on the open's foreground —
-   * arms it as a background task and the embed worker starts when it
+   * the crash-recovery contract pins that a reopened brain has its markers
-   * settles); {@link awaitPendingEmbeds} and close() wait for it first.
+   * re-armed when open() returns — and the mark is what makes that cheap.
    * It is SKIPPED WHOLESALE when the log has never had a v2 tail
    * ({@link FactLog.hasV2History} — v1 facts cannot carry marker records),
    * so pre-cutover brains pay nothing; on a mixed log the scan still reads
@@ -2710,7 +2713,6 @@ export class Brainy<T = any> implements BrainyInterface<T> {
    * before I proceed" callers use this; nothing else ever needs to wait.
    */
   public async awaitPendingEmbeds(): Promise<void> {
-    if (this._pendingEmbedRecovery) await this._pendingEmbedRecovery
     while (this._pendingEmbedIds.size > 0 || this._embedWorkerFlight) {
       this.kickEmbedWorker()
       await (this._embedWorkerFlight ?? Promise.resolve())
@@ -19584,18 +19586,6 @@ export class Brainy<T = any> implements BrainyInterface<T> {
    *   terminal releases have run.
    */
   async close(): Promise<void> {
-    if (this._pendingEmbedRecovery) {
-      // Settle the background marker fold before the durable steps — its scan
-      // is bounded by the low-water mark (a full scan happens at most once,
-      // on the first open after upgrade).
-      const settleStart = Date.now()
-      await this._pendingEmbedRecovery
-      const settleMs = Date.now() - settleStart
-      if (settleMs >= 1000) {
-        prodLog.info(`[Brainy] close: pending-embed recovery settled in ${settleMs}ms`)
-      }
-      this._pendingEmbedRecovery = null
-    }
     if (this._pendingEmbedIds.size === 0) await this.writeEmbedLowWater()
     let closeFailure: unknown = null
     try {

tests/integration/pending-embed-low-water.test.ts

@@ -6,9 +6,8 @@
  * on the open's foreground — O(whole history) per open on long-lived brains.
  * Now: an advisory low-water mark (`_system/pending_embeds_lowwater.json`)
  * records the committed generation whenever the pending set drains to empty,
- * recovery scans from `mark + 1`, and the fold runs behind the doors as a
+ * recovery scans from `mark + 1` on the open's foreground — the crash-recovery
- * latched background task the worker, `awaitPendingEmbeds()` and `close()`
+ * contract keeps markers re-armed when open() returns. The mark is advisory: stale-low costs a longer scan, never a
- * wait on. The mark is advisory: stale-low costs a longer scan, never a
  * marker — a pending embed enqueued before a crash is still recovered.
  */
 import { describe, it, expect, afterEach, vi } from 'vitest'
@@ -20,7 +19,7 @@ import { NounType } from '../../src/types/graphTypes'
 
 const LOWWATER_PATH = '_system/pending_embeds_lowwater.json'
 
-describe('pending-embed recovery: bounded by the low-water mark, behind the doors', () => {
+describe('pending-embed recovery: bounded by the low-water mark', () => {
   const roots: string[] = []
   const dir = (): string => {
     const d = mkdtempSync(join(tmpdir(), 'brainy-lowwater-'))
@@ -100,7 +99,6 @@ describe('pending-embed recovery: bounded by the low-water mark, behind the door
     await (brain as any).storage.releaseWriterLock()
 
     const brain2 = await open(root)
-    await (brain2 as any)._pendingEmbedRecovery
     expect(brain2.pendingEmbedCount()).toBeGreaterThan(0)
     await brain2.awaitPendingEmbeds()
     expect(brain2.pendingEmbedCount()).toBe(0)
@@ -110,7 +108,7 @@ describe('pending-embed recovery: bounded by the low-water mark, behind the door
     await brain.close().catch(() => undefined)
   })
 
-  it('open arms the fold as a background latch; awaitPendingEmbeds waits on it', async () => {
+  it('a reopened brain has its pending set settled when open() returns', async () => {
     const root = dir()
     const brain = await open(root)
     await brain.add({ id: 'a-row', data: 'some data', type: NounType.Thing })
@@ -118,10 +116,8 @@ describe('pending-embed recovery: bounded by the low-water mark, behind the door
     await brain.close()
 
     const brain2 = await open(root)
-    // The latch exists the moment init() returns (writable filesystem brain)…
+    // The crash-recovery contract: markers are re-armed by open itself —
-    expect((brain2 as any)._pendingEmbedRecovery).not.toBeNull()
+    // no latch, no background race. (Here the drain landed, so zero.)
-    // …and the barrier settles it before answering.
-    await brain2.awaitPendingEmbeds()
     expect(brain2.pendingEmbedCount()).toBe(0)
     await brain2.close()
   })