fix: exception-safe aggregation backfill + generation-verified adoption + loud open-path guards
- Aggregation backfill walks build into a STAGING map and swap in atomically on completion. A mid-walk failure drops the staging map — the previous live state keeps serving, the aggregate stays flagged pending, and the storage error surfaces to the failing query. Previously the walk wiped live state before a scan that could throw, never cleared the pending flag on failure, and re-ran a full walk on every subsequent query: a silent wipe/walk/throw loop at the caller's retry rate. - Failed walks are latched: retries within a 30s cooldown rethrow the recorded error instantly instead of re-walking, so a tight caller-side retry loop costs one loud error per query, never a full store walk per query. - Persisted aggregation state is stamped with the store's committed generation at flush; reopen adoption requires stamp equality. Stale state (unclean shutdown) or over-counting state (a log truncation on a copied store pulled the watermark back) triggers exactly one loud rescan, never a silent adopt. - The backfill/adoption path narrates: adoption decisions, walk start/finish with counts and duration, and failures all log by default. - getNouns/getVerbs refuse a supplied-but-undecodable pagination cursor with a loud error instead of silently restarting the walk at offset 0 (which re-served page 1 forever to any while(hasMore) caller). - The graph cold-load verb walk aborts loudly on a missing or non-advancing cursor with hasMore=true. - A versioned index provider whose generation is AHEAD of the committed watermark (torn copy / crash-recovery truncation) is now named loudly at open, alongside the existing behind-direction message.
This commit is contained in:
parent
01a7f3dd01
commit
a77b064bd7
7 changed files with 335 additions and 36 deletions
38
RELEASES.md
38
RELEASES.md
|
|
@ -10,6 +10,44 @@ Full auto-generated changelog: `CHANGELOG.md` · Releases: https://github.com/so
|
|||
|
||||
---
|
||||
|
||||
## v8.5.2 — 2026-07-17 (aggregation backfill: exception-safe, generation-verified, and loud)
|
||||
|
||||
Hardening patch from a migration incident (a byte-copied store on new hardware; the service
|
||||
entered a silent full-CPU loop at boot). Four changes, all in the aggregation engine's
|
||||
backfill/adoption path:
|
||||
|
||||
- **Backfill walks are exception-safe and non-destructive.** A rescan now builds into a
|
||||
staging map and swaps in atomically on completion; a mid-walk failure drops the staging map,
|
||||
keeps the previous live state serving, and surfaces the storage error to the failing query.
|
||||
Previously the walk wiped live state *before* a scan that could throw, never cleared the
|
||||
pending flag on failure, and re-ran a full walk on every subsequent query — a silent
|
||||
wipe/walk/throw loop at the caller's retry rate.
|
||||
- **Failed walks are latched.** After a walk fails, retries within a 30-second cooldown rethrow
|
||||
the recorded error instantly instead of re-walking — a tight caller-side retry loop now costs
|
||||
one loud error per query, never a full store walk per query.
|
||||
- **Adoption is generation-verified.** Persisted aggregation state is stamped with the store's
|
||||
committed generation at flush; reopen adoption requires the stamp to equal the current
|
||||
watermark. Stale state (unclean shutdown) or over-counting state (a fact-log truncation on a
|
||||
copied store pulled the watermark back) triggers exactly one loud rescan — never a silent
|
||||
adopt. Pre-8.5.2 state on generation-aware stores rescans once after upgrade, then is stamped.
|
||||
- **The path narrates.** Adoption decisions, no-adoptable-state outcomes, walk start/finish
|
||||
(entity count + duration), and walk failures all log by default; a non-advancing storage
|
||||
pagination cursor aborts the walk loudly instead of looping forever.
|
||||
|
||||
Plus three guards from a full audit of every loop in the open/init path:
|
||||
|
||||
- **Invalid pagination cursors fail loudly.** A supplied-but-undecodable resume token to
|
||||
`getNouns`/`getVerbs` used to silently restart the walk at offset 0 — to a `while(hasMore)`
|
||||
caller that re-serves page 1 forever (an unbounded silent CPU loop). It now throws with a
|
||||
clear message instead.
|
||||
- **The graph cold-load verb walk has a stall guard.** `hasMore=true` with a missing or
|
||||
non-advancing cursor aborts loudly instead of re-reading the same page forever.
|
||||
- **A derived index AHEAD of the store is named at open.** Brainy already surfaced a provider
|
||||
generation *behind* the committed watermark; the *ahead* direction (the signature of a
|
||||
byte-copy of a live service, or a log truncation during crash recovery) now logs a loud
|
||||
warning explaining what happened and that `brain.repairIndex()` forces a heal — instead of
|
||||
passing unnamed into whatever the derived index does next.
|
||||
|
||||
## v8.5.1 — 2026-07-17 (aggregation state survives restarts + the query-cap ratchet removed)
|
||||
|
||||
Patch release from a production incident (aggregate/count paths taking 40–90 s on an idle box
|
||||
|
|
|
|||
|
|
@ -29,6 +29,7 @@ import { matchesMetadataFilter } from '../utils/metadataFilter.js'
|
|||
import { compareCodePoints } from '../utils/collation.js'
|
||||
import { bucketTimestamp } from './timeWindows.js'
|
||||
import { NounType } from '../types/graphTypes.js'
|
||||
import { prodLog } from '../utils/logger.js'
|
||||
|
||||
/** Persistence key for aggregate definitions */
|
||||
const DEFINITIONS_KEY = '__aggregation_definitions__'
|
||||
|
|
@ -343,6 +344,14 @@ export class AggregationIndex {
|
|||
*/
|
||||
private pendingAdopt = new Set<string>()
|
||||
|
||||
/**
|
||||
* In-flight rescan targets. While a name has a staging map, ALL
|
||||
* contributions (the walk's and concurrent write hooks') land there instead
|
||||
* of the live map; the live map keeps serving until {@link finishBackfill}
|
||||
* swaps the staging map in atomically.
|
||||
*/
|
||||
private backfillStaging = new Map<string, Map<string, AggregateGroupState>>()
|
||||
|
||||
constructor(storage: StorageAdapter, nativeProvider?: AggregationProvider) {
|
||||
this.storage = storage
|
||||
this.nativeProvider = nativeProvider
|
||||
|
|
@ -391,10 +400,37 @@ export class AggregationIndex {
|
|||
* adopt (or init never ran / failed) — it must backfill.
|
||||
*/
|
||||
private resolvePendingAdoptToBackfill(): void {
|
||||
if (this.pendingAdopt.size > 0) {
|
||||
prodLog.info(
|
||||
`[Aggregation] no adoptable persisted state for: ${Array.from(this.pendingAdopt).join(', ')} — flagged for backfill`
|
||||
)
|
||||
}
|
||||
for (const name of this.pendingAdopt) this.needsBackfill.add(name)
|
||||
this.pendingAdopt.clear()
|
||||
}
|
||||
|
||||
/**
|
||||
* May this persisted state be ADOPTED? When the store exposes its committed
|
||||
* watermark, the state's `sourceGeneration` must EQUAL it: behind means
|
||||
* later writes are missing from the state (unclean shutdown); ahead means
|
||||
* it counts writes that no longer exist (e.g. a fact-log truncation on a
|
||||
* copied store pulled the watermark back). Either way: one exact rescan,
|
||||
* said out loud — never a silent adopt. Stores without the capability (and
|
||||
* pre-stamp state on them) fall back to hash-only adoption.
|
||||
*/
|
||||
private stateGenerationAdoptable(name: string, stateData: unknown): boolean {
|
||||
const committed = this.storage.committedGeneration?.() ?? null
|
||||
if (committed === null) return true
|
||||
const raw = (stateData as Record<string, unknown>).sourceGeneration
|
||||
const stamped = typeof raw === 'number' ? raw : null
|
||||
if (stamped === committed) return true
|
||||
prodLog.warn(
|
||||
`[Aggregation] '${name}': persisted state is at generation ${stamped ?? 'unstamped'} ` +
|
||||
`but the store's committed generation is ${committed} — rescanning instead of adopting`
|
||||
)
|
||||
return false
|
||||
}
|
||||
|
||||
private async loadPersisted(): Promise<void> {
|
||||
// Load persisted definitions
|
||||
const savedDefs = await this.storage.getMetadata(DEFINITIONS_KEY)
|
||||
|
|
@ -413,7 +449,11 @@ export class AggregationIndex {
|
|||
const appHash = this.definitionHashes.get(def.name) || ''
|
||||
if (appHash === savedHash && this.pendingAdopt.has(def.name)) {
|
||||
const stateData = await this.storage.getMetadata(`${STATE_KEY_PREFIX}${def.name}__`)
|
||||
if (stateData && stateData.groups) {
|
||||
if (
|
||||
stateData &&
|
||||
stateData.groups &&
|
||||
this.stateGenerationAdoptable(def.name, stateData)
|
||||
) {
|
||||
const groupMap = new Map<string, AggregateGroupState>()
|
||||
for (const group of stateData.groups as AggregateGroupState[]) {
|
||||
groupMap.set(serializeGroupKey(group.groupKey), group)
|
||||
|
|
@ -421,6 +461,9 @@ export class AggregationIndex {
|
|||
this.states.set(def.name, groupMap)
|
||||
this.pendingAdopt.delete(def.name)
|
||||
this.needsBackfill.delete(def.name)
|
||||
prodLog.info(
|
||||
`[Aggregation] '${def.name}': adopted persisted state (${groupMap.size} groups) — no rescan`
|
||||
)
|
||||
}
|
||||
// No/invalid persisted state: stays in pendingAdopt and resolves
|
||||
// to backfill when init settles.
|
||||
|
|
@ -434,7 +477,12 @@ export class AggregationIndex {
|
|||
const currentHash = hashDefinition(def)
|
||||
|
||||
const stateData = await this.storage.getMetadata(`${STATE_KEY_PREFIX}${def.name}__`)
|
||||
if (stateData && stateData.groups && savedHash === currentHash) {
|
||||
if (
|
||||
stateData &&
|
||||
stateData.groups &&
|
||||
savedHash === currentHash &&
|
||||
this.stateGenerationAdoptable(def.name, stateData)
|
||||
) {
|
||||
// Definition unchanged — load state
|
||||
const groupMap = new Map<string, AggregateGroupState>()
|
||||
for (const group of stateData.groups as AggregateGroupState[]) {
|
||||
|
|
@ -443,6 +491,9 @@ export class AggregationIndex {
|
|||
}
|
||||
this.states.set(def.name, groupMap)
|
||||
this.needsBackfill.delete(def.name)
|
||||
prodLog.info(
|
||||
`[Aggregation] '${def.name}': restored definition + adopted persisted state (${groupMap.size} groups)`
|
||||
)
|
||||
} else {
|
||||
// Definition changed or no saved state — start fresh and backfill from
|
||||
// existing entities (the owner drains needsBackfill on first query).
|
||||
|
|
@ -483,14 +534,22 @@ export class AggregationIndex {
|
|||
}))
|
||||
await this.storage.saveMetadata(DEFINITIONS_KEY, { definitions: defsToSave })
|
||||
|
||||
// Persist dirty states
|
||||
// Persist dirty states, stamped with the committed generation they
|
||||
// reflect. The stamp is what makes reopen-adoption verifiable: state at a
|
||||
// different generation than the store's committed watermark is stale (an
|
||||
// unclean shutdown after later writes) or over-counts (a fact-log
|
||||
// truncation on a copied store pulled the watermark BACK below the
|
||||
// stamp) — either way the answer is one exact rescan, never a silent
|
||||
// adopt. Read the generation after collecting groups so any racing
|
||||
// commit resolves toward rescan, not wrong-adopt.
|
||||
for (const name of this.dirty) {
|
||||
const stateMap = this.states.get(name)
|
||||
if (stateMap) {
|
||||
const groups = Array.from(stateMap.values())
|
||||
const sourceGeneration = this.storage.committedGeneration?.() ?? null
|
||||
await this.storage.saveMetadata(
|
||||
`${STATE_KEY_PREFIX}${name}__`,
|
||||
{ groups }
|
||||
sourceGeneration === null ? { groups } : { groups, sourceGeneration }
|
||||
)
|
||||
}
|
||||
}
|
||||
|
|
@ -609,9 +668,17 @@ export class AggregationIndex {
|
|||
return Array.from(this.needsBackfill)
|
||||
}
|
||||
|
||||
/** Clear an aggregate's state so a full rescan cannot double-count. */
|
||||
/**
|
||||
* Begin a rescan into a STAGING map. The live state is not touched — it
|
||||
* keeps serving (possibly stale, but flagged pending) until the rescan
|
||||
* completes and swaps in atomically. A mid-walk failure drops the staging
|
||||
* map via {@link abortBackfill} and loses nothing: wiping live state before
|
||||
* a scan that could throw was the destructive-before-durable defect.
|
||||
* Contributions (walk + concurrent write hooks) land in staging while it
|
||||
* exists, so the swapped-in result reflects writes that raced the walk.
|
||||
*/
|
||||
beginBackfill(name: string): void {
|
||||
this.states.set(name, new Map())
|
||||
this.backfillStaging.set(name, new Map())
|
||||
// Reset native provider state for this aggregate too, if present.
|
||||
const def = this.definitions.get(name)
|
||||
if (def && this.nativeProvider?.removeAggregate && this.nativeProvider?.defineAggregate) {
|
||||
|
|
@ -620,6 +687,15 @@ export class AggregationIndex {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Abandon an in-flight rescan after a failure: drop the staging map, keep
|
||||
* the live state serving, leave the aggregate flagged as pending so a later
|
||||
* attempt rescans. The failure itself must be surfaced loudly by the owner.
|
||||
*/
|
||||
abortBackfill(name: string): void {
|
||||
this.backfillStaging.delete(name)
|
||||
}
|
||||
|
||||
/** Feed one already-stored entity into a single aggregate during backfill. */
|
||||
backfillEntity(name: string, entity: Record<string, unknown>): void {
|
||||
if (isAggregateEntity(entity)) return
|
||||
|
|
@ -633,8 +709,13 @@ export class AggregationIndex {
|
|||
}
|
||||
}
|
||||
|
||||
/** Mark an aggregate's backfill complete; rebuilt state persists on next flush(). */
|
||||
/** Swap the rebuilt staging state in atomically; persists on next flush(). */
|
||||
finishBackfill(name: string): void {
|
||||
const staged = this.backfillStaging.get(name)
|
||||
if (staged) {
|
||||
this.states.set(name, staged)
|
||||
this.backfillStaging.delete(name)
|
||||
}
|
||||
this.needsBackfill.delete(name)
|
||||
this.dirty.add(name)
|
||||
}
|
||||
|
|
@ -873,7 +954,7 @@ export class AggregationIndex {
|
|||
def: AggregateDefinition,
|
||||
entity: Record<string, unknown>
|
||||
): void {
|
||||
const stateMap = this.states.get(aggName)!
|
||||
const stateMap = (this.backfillStaging.get(aggName) ?? this.states.get(aggName))!
|
||||
|
||||
// Fan out: an unnest dimension makes one entity contribute to several groups.
|
||||
for (const groupKey of computeGroupKeys(entity, def.groupBy)) {
|
||||
|
|
@ -931,7 +1012,7 @@ export class AggregationIndex {
|
|||
def: AggregateDefinition,
|
||||
entity: Record<string, unknown>
|
||||
): void {
|
||||
const stateMap = this.states.get(aggName)!
|
||||
const stateMap = (this.backfillStaging.get(aggName) ?? this.states.get(aggName))!
|
||||
|
||||
// Fan out: reverse the entity's contribution from every group it joined.
|
||||
for (const groupKey of computeGroupKeys(entity, def.groupBy)) {
|
||||
|
|
@ -987,7 +1068,7 @@ export class AggregationIndex {
|
|||
* Apply results from native provider back into the state maps.
|
||||
*/
|
||||
private applyNativeResults(aggName: string, results: AggregateGroupState[]): void {
|
||||
const stateMap = this.states.get(aggName)!
|
||||
const stateMap = (this.backfillStaging.get(aggName) ?? this.states.get(aggName))!
|
||||
for (const group of results) {
|
||||
const serialized = serializeGroupKey(group.groupKey)
|
||||
stateMap.set(serialized, group)
|
||||
|
|
|
|||
104
src/brainy.ts
104
src/brainy.ts
|
|
@ -384,6 +384,14 @@ class InsertPreconditionExistsSignal extends Error {
|
|||
*/
|
||||
export type IndexFamily = 'vector' | 'metadata' | 'graph'
|
||||
|
||||
/**
|
||||
* How long a failed aggregation-backfill walk suppresses fresh walk attempts.
|
||||
* Within the window, queries rethrow the recorded failure instantly (loud,
|
||||
* cheap); after it, one new attempt is allowed. Bounds the damage of a
|
||||
* caller-side tight retry loop against a deterministically-failing store.
|
||||
*/
|
||||
const AGGREGATION_BACKFILL_RETRY_COOLDOWN_MS = 30_000
|
||||
|
||||
/**
|
||||
* The main Brainy class - Clean, Beautiful, Powerful
|
||||
* REAL IMPLEMENTATION - No stubs, no mocks
|
||||
|
|
@ -599,6 +607,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
private _pendingMigrationRunner?: MigrationRunner // Deferred migration runner for large datasets
|
||||
private _aggregationIndex?: AggregationIndex // Incremental aggregation engine
|
||||
private _aggregationBackfillFlight: Promise<void> | null = null // Single-flight backfill walk
|
||||
// A failed walk latches its error: retries within the cooldown rethrow it
|
||||
// instantly instead of re-walking, so a tight caller-side retry loop costs
|
||||
// one loud error per query, never a full store walk per query.
|
||||
private _aggregationBackfillFailure: { at: number; error: Error } | null = null
|
||||
private _materializer?: AggregateMaterializer // Debounced materialization of aggregate results
|
||||
/**
|
||||
* Fields registered via `brain.trackField()` — drives optional value validation on
|
||||
|
|
@ -1182,6 +1194,22 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
`(storage committed: ${committed}) — provider replays the gap per ` +
|
||||
`the post-commit applier contract`
|
||||
)
|
||||
} else if (providerGen > committed) {
|
||||
// The AHEAD direction is incoherence, not a replay gap: the provider's
|
||||
// persisted index claims writes the store no longer has — the signature
|
||||
// of a torn copy or a log truncation that pulled the committed
|
||||
// watermark back (crash recovery, byte-copy of a live store). A replay
|
||||
// can never converge on it and index answers may reference vanished
|
||||
// writes. Name it loudly at open so it is never diagnosed from a
|
||||
// silent journal; the provider's own coherence check / heal walk (or
|
||||
// brain.repairIndex()) is the cure.
|
||||
prodLog.warn(
|
||||
`[Brainy] Versioned index provider is AHEAD of the store: provider ` +
|
||||
`generation ${providerGen} vs committed ${committed}. This store was ` +
|
||||
`likely copied from a live service or truncated during crash recovery. ` +
|
||||
`Derived-index answers may reference rolled-back writes until the ` +
|
||||
`provider heals from canonical (brain.repairIndex() forces it).`
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -15828,6 +15856,18 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
// the in-flight walk snapshotted its batch before `name` became pending —
|
||||
// the next iteration starts a fresh walk that includes it.
|
||||
while (index.getPendingBackfills().includes(name)) {
|
||||
// Failure latch: a deterministically-failing walk must not be re-run at
|
||||
// the caller's retry rate — that is a silent CPU loop wearing a retry
|
||||
// loop's clothes. Within the cooldown, rethrow the recorded failure
|
||||
// immediately; after it, one fresh attempt is allowed.
|
||||
const failure = this._aggregationBackfillFailure
|
||||
if (failure && Date.now() - failure.at < AGGREGATION_BACKFILL_RETRY_COOLDOWN_MS) {
|
||||
throw new Error(
|
||||
`Aggregation backfill for '${name}' is in failure cooldown (retry in ` +
|
||||
`${Math.ceil((AGGREGATION_BACKFILL_RETRY_COOLDOWN_MS - (Date.now() - failure.at)) / 1000)}s). ` +
|
||||
`Last failure: ${failure.error.message}`
|
||||
)
|
||||
}
|
||||
if (!this._aggregationBackfillFlight) {
|
||||
this._aggregationBackfillFlight = this.runAggregationBackfillWalk()
|
||||
.finally(() => {
|
||||
|
|
@ -15850,30 +15890,60 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
const names = index.getPendingBackfills()
|
||||
if (names.length === 0) return
|
||||
|
||||
prodLog.info(`[Aggregation] backfill walk starting for: ${names.join(', ')}`)
|
||||
const startedAt = Date.now()
|
||||
for (const n of names) index.beginBackfill(n)
|
||||
|
||||
const PAGE = 500
|
||||
let offset = 0
|
||||
let cursor: string | undefined
|
||||
for (;;) {
|
||||
const page = await this.storage.getNouns({
|
||||
pagination: cursor ? { limit: PAGE, cursor } : { limit: PAGE, offset }
|
||||
})
|
||||
for (const noun of page.items) {
|
||||
const record = noun as unknown as Record<string, unknown>
|
||||
for (const n of names) {
|
||||
index.backfillEntity(n, record)
|
||||
let scanned = 0
|
||||
try {
|
||||
const PAGE = 500
|
||||
let offset = 0
|
||||
let cursor: string | undefined
|
||||
for (;;) {
|
||||
const page = await this.storage.getNouns({
|
||||
pagination: cursor ? { limit: PAGE, cursor } : { limit: PAGE, offset }
|
||||
})
|
||||
for (const noun of page.items) {
|
||||
const record = noun as unknown as Record<string, unknown>
|
||||
for (const n of names) {
|
||||
index.backfillEntity(n, record)
|
||||
}
|
||||
}
|
||||
scanned += page.items.length
|
||||
if (!page.hasMore || page.items.length === 0) break
|
||||
if (page.nextCursor) {
|
||||
if (page.nextCursor === cursor) {
|
||||
// A non-advancing cursor with hasMore=true would loop this walk at
|
||||
// CPU speed forever, silently. That is a storage pagination defect —
|
||||
// fail the waiting queries loudly instead of spinning.
|
||||
throw new Error(
|
||||
`Aggregation backfill aborted: storage pagination returned a non-advancing cursor ` +
|
||||
`after ${scanned} entities with hasMore=true — the storage adapter's getNouns cursor is broken.`
|
||||
)
|
||||
}
|
||||
cursor = page.nextCursor
|
||||
} else {
|
||||
offset += page.items.length
|
||||
}
|
||||
}
|
||||
if (!page.hasMore || page.items.length === 0) break
|
||||
if (page.nextCursor) {
|
||||
cursor = page.nextCursor
|
||||
} else {
|
||||
offset += page.items.length
|
||||
}
|
||||
} catch (err) {
|
||||
// Non-destructive failure: drop the staging maps (live state keeps
|
||||
// serving), keep the aggregates flagged pending, latch the error so
|
||||
// retries within the cooldown fail fast, and say all of it out loud.
|
||||
for (const n of names) index.abortBackfill(n)
|
||||
this._aggregationBackfillFailure = { at: Date.now(), error: err as Error }
|
||||
prodLog.warn(
|
||||
`[Aggregation] backfill walk FAILED after ${scanned} entities: ${(err as Error).message} — ` +
|
||||
`prior aggregate state preserved; retries suppressed for ${AGGREGATION_BACKFILL_RETRY_COOLDOWN_MS / 1000}s`
|
||||
)
|
||||
throw err
|
||||
}
|
||||
|
||||
for (const n of names) index.finishBackfill(n)
|
||||
this._aggregationBackfillFailure = null
|
||||
prodLog.info(
|
||||
`[Aggregation] backfill walk finished: ${scanned} entities → ${names.length} aggregate(s) in ${Date.now() - startedAt}ms`
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -282,6 +282,16 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
|
|||
}
|
||||
|
||||
hasMore = result.hasMore
|
||||
if (hasMore && (!result.nextCursor || result.nextCursor === cursor)) {
|
||||
// A stalled cursor with hasMore=true would re-read the same page
|
||||
// forever — a silent full-CPU loop at cold open. Abort loudly; a
|
||||
// graph read failing beats a process that spins without a log line.
|
||||
throw new Error(
|
||||
`GraphAdjacencyIndex: verb walk stalled after ${count} verbs — storage returned ` +
|
||||
`hasMore=true with ${result.nextCursor ? 'a non-advancing' : 'no'} cursor. ` +
|
||||
`Aborting the cold-load; run brain.repairIndex() if this persists.`
|
||||
)
|
||||
}
|
||||
cursor = result.nextCursor
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -2068,11 +2068,20 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
// Cursor (8.0): resume token carrying the (shard, nounId) of the last returned
|
||||
// noun — the noun mirror of getVerbsWithPagination. When present it supersedes
|
||||
// `offset` and resumes the shard walk immediately AFTER that position, so a full
|
||||
// walk is O(N) instead of the O(N²) of offset paging. Malformed/foreign tokens
|
||||
// decode to null → offset fallback. (Previously the cursor was ignored, which
|
||||
// was latent — the only multi-page consumer used a single big page — until small
|
||||
// chunk sizes needed page 2 and an offset-0-on-every-call walk never terminated.)
|
||||
// walk is O(N) instead of the O(N²) of offset paging. (Previously the cursor was
|
||||
// ignored, which was latent — the only multi-page consumer used a single big
|
||||
// page — until small chunk sizes needed page 2 and an offset-0-on-every-call
|
||||
// walk never terminated.)
|
||||
const cursor = this.decodeNounWalkCursor(options.cursor)
|
||||
if (options.cursor && cursor === null) {
|
||||
// A supplied-but-undecodable resume token must FAIL, not silently restart
|
||||
// at offset 0 — to a while(hasMore) caller the silent fallback re-serves
|
||||
// page 1 forever: an unbounded CPU loop wearing pagination's clothes.
|
||||
throw BrainyError.storage(
|
||||
`getNouns: invalid pagination cursor '${options.cursor}' — cannot resume this walk. ` +
|
||||
`Restart it without a cursor.`
|
||||
)
|
||||
}
|
||||
const collected: Array<{ noun: HNSWNounWithMetadata; shard: number }> = []
|
||||
|
||||
// Peek one past the window so `hasMore` is decidable. Cursor mode collects one
|
||||
|
|
@ -2366,8 +2375,17 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
// (shard, verbId) of the last returned verb. When present it SUPERSEDES `offset`
|
||||
// and resumes the shard walk immediately AFTER that position, so a full walk is
|
||||
// O(N) total instead of the O(N²) of offset paging (which re-scans from shard 0
|
||||
// every page). Malformed / foreign tokens decode to null → offset fallback.
|
||||
// every page).
|
||||
const cursor = this.decodeVerbWalkCursor(options.cursor)
|
||||
if (options.cursor && cursor === null) {
|
||||
// A supplied-but-undecodable resume token must FAIL, not silently restart
|
||||
// at offset 0 — to a while(hasMore) caller the silent fallback re-serves
|
||||
// page 1 forever: an unbounded CPU loop wearing pagination's clothes.
|
||||
throw BrainyError.storage(
|
||||
`getVerbs: invalid pagination cursor '${options.cursor}' — cannot resume this walk. ` +
|
||||
`Restart it without a cursor.`
|
||||
)
|
||||
}
|
||||
|
||||
// Each collected entry remembers its shard so nextCursor can point at the exact
|
||||
// (shard, id) resume position.
|
||||
|
|
|
|||
|
|
@ -209,6 +209,82 @@ describe('aggregation state persistence — boot-order contract', () => {
|
|||
await brain2.close()
|
||||
})
|
||||
|
||||
it('generation-mismatched persisted state is rescanned once, loudly — never adopted', async () => {
|
||||
const brain1 = await open()
|
||||
brain1.defineAggregate(SPENDING)
|
||||
await seed(brain1)
|
||||
await brain1.queryAggregate('spending')
|
||||
await brain1.close()
|
||||
|
||||
// Simulate the copied-store incident class: a fact-log truncation (or an
|
||||
// unclean shutdown) leaves the committed watermark different from the
|
||||
// generation the flushed state was stamped with.
|
||||
const tamper: any = await open()
|
||||
const key = '__aggregation_state_spending__'
|
||||
const stored = await tamper.storage.getMetadata(key)
|
||||
expect(typeof stored.sourceGeneration).toBe('number') // the stamp is really persisted
|
||||
await tamper.storage.saveMetadata(key, {
|
||||
...stored,
|
||||
sourceGeneration: stored.sourceGeneration + 5
|
||||
})
|
||||
await tamper.close()
|
||||
|
||||
const warnSpy = vi.spyOn(prodLog, 'warn')
|
||||
const brain2 = await open()
|
||||
brain2.defineAggregate(SPENDING)
|
||||
await brain2.getNounCount()
|
||||
const walks = countWalks(brain2)
|
||||
|
||||
const rows = await brain2.queryAggregate('spending')
|
||||
|
||||
expect(walks.count()).toBe(1) // exactly ONE rescan — no silent adopt, no spin
|
||||
const food = rows.find((r: any) => r.groupKey.category === 'food')
|
||||
expect(food.metrics.count).toBe(6) // rescan produced exact results
|
||||
expect(
|
||||
warnSpy.mock.calls.some(args => String(args[0]).includes('rescanning instead of adopting'))
|
||||
).toBe(true) // and it said so out loud
|
||||
warnSpy.mockRestore()
|
||||
await brain2.close()
|
||||
})
|
||||
|
||||
it('a failing walk is loud, non-destructive, and latched — never a silent retry loop', async () => {
|
||||
// Fresh define + seeded writes: the write hooks have populated LIVE state,
|
||||
// and the first-query rescan is still pending. The incident shape
|
||||
// (wipe-before-scan + no try/catch + per-query re-walk) would have wiped
|
||||
// that live state and silently re-walked on every query.
|
||||
const brain: any = await open()
|
||||
brain.defineAggregate(SPENDING)
|
||||
await seed(brain)
|
||||
expect(brain._aggregationIndex.queryAggregate({ name: 'spending' }).length).toBe(2)
|
||||
|
||||
const storage = brain.storage
|
||||
const origGetNouns = storage.getNouns.bind(storage)
|
||||
let walkAttempts = 0
|
||||
storage.getNouns = async () => {
|
||||
walkAttempts++
|
||||
throw new Error('injected storage failure')
|
||||
}
|
||||
|
||||
// First query: the walk fails LOUDLY with the storage error.
|
||||
await expect(brain.queryAggregate('spending')).rejects.toThrow('injected storage failure')
|
||||
expect(walkAttempts).toBe(1)
|
||||
|
||||
// Live state was NOT destroyed by the failed walk (staging was dropped).
|
||||
expect(brain._aggregationIndex.queryAggregate({ name: 'spending' }).length).toBe(2)
|
||||
|
||||
// Second query inside the cooldown: instant loud failure, NO new walk.
|
||||
await expect(brain.queryAggregate('spending')).rejects.toThrow('failure cooldown')
|
||||
expect(walkAttempts).toBe(1)
|
||||
|
||||
// Heal the storage + expire the cooldown: one fresh walk succeeds exactly.
|
||||
storage.getNouns = origGetNouns
|
||||
brain._aggregationBackfillFailure.at = Date.now() - 60_000
|
||||
const rows = await brain.queryAggregate('spending')
|
||||
const food = rows.find((r: any) => r.groupKey.category === 'food')
|
||||
expect(food.metrics.count).toBe(6)
|
||||
await brain.close()
|
||||
})
|
||||
|
||||
it('aggregation persistence keys never log "Unknown key format"', async () => {
|
||||
const warnSpy = vi.spyOn(prodLog, 'warn')
|
||||
const brain1 = await open()
|
||||
|
|
|
|||
|
|
@ -95,9 +95,15 @@ describe('verb cursor pagination (graph-perf #2)', () => {
|
|||
expect(new Set(cursorSeen)).toEqual(new Set(offsetSeen))
|
||||
})
|
||||
|
||||
it('a foreign/malformed cursor falls back gracefully (no throw, starts from the beginning)', async () => {
|
||||
const page = await storage.getVerbs({ pagination: { limit: 5, cursor: 'not-a-cv1-token' } })
|
||||
expect(page.items.length).toBe(5)
|
||||
expect(page.hasMore).toBe(true)
|
||||
it('a foreign/malformed cursor FAILS LOUDLY — never a silent restart from page 1', async () => {
|
||||
// The old behavior (decode-null → silent offset-0 fallback) re-served page 1
|
||||
// forever to any while(hasMore) walker: an unbounded CPU loop with no log
|
||||
// line. An undecodable resume token now refuses the walk instead.
|
||||
await expect(
|
||||
storage.getVerbs({ pagination: { limit: 5, cursor: 'not-a-cv1-token' } })
|
||||
).rejects.toThrow('invalid pagination cursor')
|
||||
await expect(
|
||||
storage.getNouns({ pagination: { limit: 5, cursor: 'not-a-cv1-token' } })
|
||||
).rejects.toThrow('invalid pagination cursor')
|
||||
})
|
||||
})
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue