Merge branches 'fix/connected-find-order', 'fix/pending-embed-low-water' and 'fix/related-verb-array' into rel/10.4.9-candidate

src/brainy.ts

@@ -1820,31 +1820,31 @@ 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) {
-        try {
-          await step(
-            '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 back into the pending set',
-            () => this.recoverPendingEmbedsFromLog()
-          )
-          if (this._pendingEmbedIds.size > 0) {
-            prodLog.info(
-              `[Brainy] ${this._pendingEmbedIds.size} deferred embed(s) pending from a previous ` +
-                `session — resuming in the background`
+        // BEHIND THE DOORS (the open pays nothing here): the bridge + the
+        // recovery fold run as one latched background task; the embed worker
+        // starts when it settles. A pending embed's outcome was always
+        // eventual — moving its recovery off the open's foreground changes
+        // when the worker starts, never whether a marker is honored.
+        // awaitPendingEmbeds() and close() wait on the latch first.
+        this._pendingEmbedRecovery = (async () => {
+          try {
+            await this.bridgeLegacyPendingEmbedSidecars()
+            await this.recoverPendingEmbedsFromLog()
+            if (this._pendingEmbedIds.size > 0) {
+              prodLog.info(
+                `[Brainy] ${this._pendingEmbedIds.size} deferred embed(s) pending from a previous ` +
+                  `session — resuming in the background`
+              )
+              const t = setTimeout(() => this.kickEmbedWorker(), 0)
+              ;(t as { unref?: () => void }).unref?.()
+            }
+          } catch (err) {
+            prodLog.warn(
+              `[Brainy] pending-embed recovery failed: ${(err as Error).message} — ` +
+                `the log's markers remain durable; recovery retries next open`
             )
-            const t = setTimeout(() => this.kickEmbedWorker(), 0)
-            ;(t as { unref?: () => void }).unref?.()
           }
-        } catch (err) {
-          prodLog.warn(
-            `[Brainy] pending-embed recovery failed: ${(err as Error).message} — ` +
-              `the log's markers remain durable; recovery retries next open`
-          )
-        }
+        })()
       }
 
       // PHASE 4 of 5 — "VFS bootstrap": shutdown-hook registration, blob
@@ -2408,6 +2408,19 @@ export class Brainy<T = any> implements BrainyInterface<T> {
    */
   private static readonly PENDING_EMBED_PREFIX = '_system/pending_embeds/'
 
+  /**
+   * Storage-root-relative path of the ADVISORY pending-embed low-water mark:
+   * `{ generation, writtenAt }`, written whenever the pending set drains to
+   * empty (and at clean close when empty). Every marker in facts at or below
+   * `generation` is consumed, so recovery scans from `generation + 1`. The
+   * mark is advisory and monotone-safe: stale-low costs a longer scan, never
+   * a lost marker; it is never required for correctness.
+   */
+  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
    * in-memory fast-path set and the returned `embed.pending` record is
@@ -2435,6 +2448,40 @@ export class Brainy<T = any> implements BrainyInterface<T> {
    */
   private clearPendingEmbed(id: string): void {
     this._pendingEmbedIds.delete(id)
+    if (this._pendingEmbedIds.size === 0) this.maybeWriteEmbedLowWater()
+  }
+
+  /**
+   * @description Advance the advisory low-water mark: called at drain-to-empty
+   * (and at clean close when empty), it records the fact log's CURRENT head —
+   * with the set empty, every marker at or below the head has been consumed,
+   * so the next open's recovery fold scans only what comes after. Fire-and-
+   * forget at the drain (close() awaits the core); loud on failure: a missed
+   * write costs the next open a longer scan, never a marker. No-op without a
+   * fact log (no durable markers exist there) and on read-only opens.
+   */
+  private maybeWriteEmbedLowWater(): void {
+    void this.writeEmbedLowWater()
+  }
+
+  /** The awaitable core of {@link maybeWriteEmbedLowWater} — close() awaits it. */
+  private async writeEmbedLowWater(): Promise<void> {
+    if (this.isReadOnly) return
+    const log = this.generationStore ? this.generationStore.getFactLog() : null
+    if (!log) return
+    const generation = log.headGeneration()
+    if (!(generation > 0)) return
+    try {
+      await this.storage.writeRawObject(Brainy.PENDING_EMBED_LOWWATER_PATH, {
+        generation,
+        writtenAt: Date.now()
+      })
+    } catch (err) {
+      prodLog.warn(
+        `[Brainy] pending-embed low-water write failed at generation ${generation}: ` +
+          `${(err as Error).message} — the next open scans from the previous mark`
+      )
+    }
   }
 
   /**
@@ -2445,9 +2492,14 @@ export class Brainy<T = any> implements BrainyInterface<T> {
    * survives the fold is exactly the set of acknowledged deferred writes
    * whose vectors have not landed.
    *
-   * BOUND (honest): no durable low-water mark exists for the earliest
-   * unconsumed pending, so the fold scans the log's committed facts from
-   * generation 1 — a sequential read of the log at open, O(log bytes).
+   * BOUND: the scan starts at the advisory low-water mark
+   * ({@link Brainy.PENDING_EMBED_LOWWATER_PATH}) — the log head at which the
+   * 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
+   * arms it as a background task and the embed worker starts when it
+   * settles); {@link awaitPendingEmbeds} and close() wait for it first.
    * 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
@@ -2460,7 +2512,18 @@ export class Brainy<T = any> implements BrainyInterface<T> {
   private async recoverPendingEmbedsFromLog(): Promise<void> {
     const log = this.generationStore.getFactLog()
     if (!log || !log.hasV2History()) return
-    const scan = log.scanFacts({ fromGeneration: 1 })
+    let fromGeneration = 1
+    try {
+      const mark = (await this.storage.readRawObject(Brainy.PENDING_EMBED_LOWWATER_PATH)) as {
+        generation?: number
+      } | null
+      if (mark && typeof mark.generation === 'number' && mark.generation > 0) {
+        fromGeneration = mark.generation + 1
+      }
+    } catch {
+      // No mark (or unreadable): scan from 1 — correctness over cost.
+    }
+    const scan = log.scanFacts({ fromGeneration })
     for await (const batch of scan.batches()) {
       for (const fact of batch.facts) {
         for (const record of fact.records ?? []) {
@@ -2647,6 +2710,7 @@ 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())
@@ -19520,6 +19584,19 @@ 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 {
       await this.closeDurableSteps()

src/storage/baseStorage.ts

@@ -2942,19 +2942,33 @@ export abstract class BaseStorage extends BaseStorageAdapter {
         !options.filter.service &&
         !options.filter.metadata
       ) {
-        const sourceId = Array.isArray(options.filter.sourceId)
-          ? options.filter.sourceId[0]
-          : options.filter.sourceId
+        const sourceIds = Array.isArray(options.filter.sourceId)
+          ? options.filter.sourceId
+          : [options.filter.sourceId]
 
-        const verbType = Array.isArray(options.filter.verbType)
-          ? options.filter.verbType[0]
-          : options.filter.verbType
+        // EVERY requested verb type is honoured — an array used to collapse to
+        // its first element here, silently dropping the rest of the ask.
+        const verbTypes = new Set(
+          Array.isArray(options.filter.verbType)
+            ? options.filter.verbType
+            : [options.filter.verbType]
+        )
 
-        // Get verbs by source, then filter by type (O(1) graph lookup + O(n) type filter),
-        // then apply the subtype / visibility metadata filters on the candidate set.
-        const verbsBySource = await this.getVerbsBySource_internal(sourceId)
+        // Get verbs by source (union over every requested source), filter by the
+        // requested type SET (O(1) graph lookup + O(n) type filter), then apply
+        // the subtype / visibility metadata filters on the candidate set.
+        const bySource: HNSWVerbWithMetadata[] = []
+        const seenVerbIds = new Set<string>()
+        for (const oneSource of sourceIds) {
+          for (const v of await this.getVerbsBySource_internal(oneSource)) {
+            if (!seenVerbIds.has(v.id)) {
+              seenVerbIds.add(v.id)
+              bySource.push(v)
+            }
+          }
+        }
         const filteredVerbs = this.applyVerbMetadataFilters(
-          verbsBySource.filter(v => v.verb === verbType),
+          bySource.filter(v => verbTypes.has(v.verb)),
           options.filter
         )
 
@@ -2985,16 +2999,22 @@ export abstract class BaseStorage extends BaseStorageAdapter {
         !options.filter.service &&
         !options.filter.metadata
       ) {
-        const sourceId = Array.isArray(options.filter.sourceId)
-          ? options.filter.sourceId[0]
-          : options.filter.sourceId
-
-        // Get verbs by source directly (hydrated with metadata), then apply the
-        // subtype / visibility metadata filters on the O(degree) candidate set.
-        const verbsBySource = this.applyVerbMetadataFilters(
-          await this.getVerbsBySource_internal(sourceId),
-          options.filter
-        )
+        // EVERY requested source is honoured — an array used to collapse to
+        // its first element here, silently dropping the rest of the ask.
+        const onlySourceIds = Array.isArray(options.filter.sourceId)
+          ? options.filter.sourceId
+          : [options.filter.sourceId]
+        const sourceUnion: HNSWVerbWithMetadata[] = []
+        const seenSourceVerbIds = new Set<string>()
+        for (const oneSource of onlySourceIds) {
+          for (const v of await this.getVerbsBySource_internal(oneSource)) {
+            if (!seenSourceVerbIds.has(v.id)) {
+              seenSourceVerbIds.add(v.id)
+              sourceUnion.push(v)
+            }
+          }
+        }
+        const verbsBySource = this.applyVerbMetadataFilters(sourceUnion, options.filter)
 
         // Apply pagination
         const paginatedVerbs = verbsBySource.slice(offset, offset + limit)
@@ -3023,16 +3043,22 @@ export abstract class BaseStorage extends BaseStorageAdapter {
         !options.filter.service &&
         !options.filter.metadata
       ) {
-        const targetId = Array.isArray(options.filter.targetId)
-          ? options.filter.targetId[0]
-          : options.filter.targetId
-
-        // Get verbs by target directly (hydrated with metadata), then apply the
-        // subtype / visibility metadata filters on the O(degree) candidate set.
-        const verbsByTarget = this.applyVerbMetadataFilters(
-          await this.getVerbsByTarget_internal(targetId),
-          options.filter
-        )
+        // EVERY requested target is honoured — an array used to collapse to
+        // its first element here, silently dropping the rest of the ask.
+        const onlyTargetIds = Array.isArray(options.filter.targetId)
+          ? options.filter.targetId
+          : [options.filter.targetId]
+        const targetUnion: HNSWVerbWithMetadata[] = []
+        const seenTargetVerbIds = new Set<string>()
+        for (const oneTarget of onlyTargetIds) {
+          for (const v of await this.getVerbsByTarget_internal(oneTarget)) {
+            if (!seenTargetVerbIds.has(v.id)) {
+              seenTargetVerbIds.add(v.id)
+              targetUnion.push(v)
+            }
+          }
+        }
+        const verbsByTarget = this.applyVerbMetadataFilters(targetUnion, options.filter)
 
         // Apply pagination
         const paginatedVerbs = verbsByTarget.slice(offset, offset + limit)
@@ -3061,16 +3087,25 @@ export abstract class BaseStorage extends BaseStorageAdapter {
         !options.filter.service &&
         !options.filter.metadata
       ) {
-        const verbType = Array.isArray(options.filter.verbType)
-          ? options.filter.verbType[0]
-          : options.filter.verbType
+        // EVERY requested verb type is honoured — an array used to collapse to
+        // its first element here, silently dropping the rest of the ask.
+        const verbTypes = Array.isArray(options.filter.verbType)
+          ? options.filter.verbType
+          : [options.filter.verbType]
 
-        // Get verbs by type directly (hydrated with metadata), then apply the
-        // subtype / visibility metadata filters on the candidate set.
-        const verbsByType = this.applyVerbMetadataFilters(
-          await this.getVerbsByType_internal(verbType),
-          options.filter
-        )
+        // Get verbs by each requested type (hydrated with metadata), deduped by
+        // id, then apply the subtype / visibility metadata filters on the set.
+        const byType: HNSWVerbWithMetadata[] = []
+        const seenTypeVerbIds = new Set<string>()
+        for (const oneType of verbTypes) {
+          for (const v of await this.getVerbsByType_internal(oneType)) {
+            if (!seenTypeVerbIds.has(v.id)) {
+              seenTypeVerbIds.add(v.id)
+              byType.push(v)
+            }
+          }
+        }
+        const verbsByType = this.applyVerbMetadataFilters(byType, options.filter)
 
         // Apply pagination
         const paginatedVerbs = verbsByType.slice(offset, offset + limit)