diff --git a/RELEASES.md b/RELEASES.md index 457b65d0..920f0d79 100644 --- a/RELEASES.md +++ b/RELEASES.md @@ -10,6 +10,39 @@ Full auto-generated changelog: `CHANGELOG.md` · Releases: https://github.com/so --- +## 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 +while vector search stayed fast, and every `find({ limit: 5000 })` suddenly failing against an +"auto-configured query limit of 1000"). Three fixes, one cosmetic: + +- **Aggregation state is actually adopted on reopen.** The boot pattern `defineAggregate()` → + query raced the engine's async state load: the synchronous define always won, flagged a + backfill, and the first query then wiped the just-loaded persisted state and re-walked the + entire store — every restart, forever. Reopening with an unchanged definition now adopts the + persisted state directly (zero scans); a backfill runs only on a real definition change, a + missing/failed state load, or a write that landed before adoption (exactness wins). Apps that + rely on persisted definitions without re-defining at boot also no longer race a spurious + "Aggregate not defined". +- **Backfills are single-flight and batched.** Concurrent queries on a cold aggregate used to + each wipe the others' partial state and start their own full store walk — under steady query + arrival the store never converged (the 40–90 s loop). Now all concurrent queries share one + walk, and one walk fills every aggregate pending backfill (M aggregates ≠ M scans). +- **The query-cap "learning" ratchet is removed.** `maxLimit` is a memory-protection bound, but + a hidden tuner shrank it 20 % per recorded query while the lifetime-average query time + exceeded 1 s — down to a floor of 1000, below the documented 10 000 auto floor, with no + practical recovery, and the resulting error blamed "available free memory" (stale basis + label). The cap now comes from its construction-time basis (or your explicit + `maxQueryLimit` / `reservedQueryMemory`) alone and never changes at runtime; query timing is + recorded for diagnostics only. +- **Cosmetic:** the engine's own persistence keys (`__aggregation_*`, `brainy:entityIdMapper`) + no longer log `[Storage] Unknown key format` at boot — they were always routed correctly; + they're now recognized before the warning fires. + +Operationally: if a host was bitten, upgrading and restarting is the whole fix — no repair +ritual needed. Setting `maxQueryLimit` explicitly remains the valve that bypasses auto-detection +entirely. + ## v8.5.0 — 2026-07-15 (provider access to the fact log + the shared stamp verifier) Small additive follow-up to 8.4.0, from the native accelerator's first consumption pass: diff --git a/docs/guides/aggregation.md b/docs/guides/aggregation.md index e58eb2ce..11d86ec8 100644 --- a/docs/guides/aggregation.md +++ b/docs/guides/aggregation.md @@ -11,7 +11,13 @@ No batch jobs. No scheduled recalculations. Aggregates stay current with every w **Defining over existing data:** if you define an aggregate on a store that already holds matching entities, Brainy backfills it from those entities on the first query (a one-time scan, then purely incremental). So `defineAggregate()` behaves the same whether you define it before -or after the data exists — including when a persisted brain reopens already populated. +or after the data exists. + +**Reopening a persisted brain:** aggregate state persists across restarts. Re-defining the +same aggregate at boot (the normal declarative pattern) adopts the persisted state directly — +no rescan. A backfill scan runs only when the definition actually changed, when no persisted +state exists, or when the state failed to load; and however many aggregates need backfilling, +they share a single scan. ## Quick Start diff --git a/docs/guides/find-limits.md b/docs/guides/find-limits.md index 66418b80..4c7fd252 100644 --- a/docs/guides/find-limits.md +++ b/docs/guides/find-limits.md @@ -40,6 +40,10 @@ Brainy picks `maxLimit` from the first of these that's available: Worked example: a 4 GB Cloud Run container picks priority 3 → `floor(4 GB × 0.25 / 25 KB) = floor(40 960) = 40 000` results. A 900 MB free-memory box on priority 4 gets `floor(900 MB / 25 KB) = ~36 000`. +The cap is fixed at construction and never changes at runtime. Query timing is recorded +for diagnostics only — a burst of slow queries cannot silently shrink the cap, and the +auto-detected tiers (3 and 4) never go below a floor of 10 000. + > **Calibration note.** Pre-7.30.2 used 100 KB per result instead of 25 KB, which produced caps that were 4× too tight for typical workloads (an 8 KB / result reality). 7.30.2 recalibrated to match observed entity sizes; existing `limit: 10_000` safety patterns now pass silently on any reasonably-sized box. ## What happens when you exceed the cap diff --git a/src/aggregation/AggregationIndex.ts b/src/aggregation/AggregationIndex.ts index ecd16228..d74d732b 100644 --- a/src/aggregation/AggregationIndex.ts +++ b/src/aggregation/AggregationIndex.ts @@ -327,6 +327,22 @@ export class AggregationIndex { /** Track aggregates with stale MIN/MAX (need lazy recompute) */ private staleMinMax = new Map>() + /** Resolves when init() has finished loading persisted definitions/state. */ + private initPromise: Promise | null = null + + /** True once init() has settled (success or failure). */ + private initDone = false + + /** + * Aggregates registered by the app before init() finished loading persisted + * state, awaiting reconciliation: init() adopts the persisted state when the + * definition hash matches; anything left unadopted when init settles resolves + * to a backfill. Deciding backfill eagerly at define time was the boot-order + * bug that wiped valid persisted state on every restart — the synchronous + * defineAggregate() always beats the async init(). + */ + private pendingAdopt = new Set() + constructor(storage: StorageAdapter, nativeProvider?: AggregationProvider) { this.storage = storage this.nativeProvider = nativeProvider @@ -336,19 +352,87 @@ export class AggregationIndex { /** * Initialize: load persisted definitions and state, detect changes, rebuild stale. + * + * Idempotent — repeated calls return the same promise. Definitions registered + * *before* this completes (the normal boot order: `defineAggregate()` is + * synchronous and always beats this async load) are reconciled rather than + * clobbered: the app's definition wins, and its persisted state is adopted + * when the definition hash matches — backfill happens only on a real change. */ - async init(): Promise { + init(): Promise { + if (!this.initPromise) { + this.initPromise = this.loadPersisted().finally(() => { + this.resolvePendingAdoptToBackfill() + this.initDone = true + }) + } + return this.initPromise + } + + /** + * Await the persisted-state load (if one was started) and settle every + * pending adoption decision. After this resolves, `getPendingBackfills()` + * is authoritative: a name is listed iff it genuinely needs a rescan. + * Query paths must await this before consulting backfill state. + */ + async ready(): Promise { + if (this.initPromise) { + try { + await this.initPromise + } catch { + // The owner already surfaced the load failure loudly; backfill covers. + } + } + this.resolvePendingAdoptToBackfill() + } + + /** + * Any definition still awaiting state adoption has no persisted state to + * adopt (or init never ran / failed) — it must backfill. + */ + private resolvePendingAdoptToBackfill(): void { + for (const name of this.pendingAdopt) this.needsBackfill.add(name) + this.pendingAdopt.clear() + } + + private async loadPersisted(): Promise { // Load persisted definitions const savedDefs = await this.storage.getMetadata(DEFINITIONS_KEY) if (savedDefs && typeof savedDefs === 'object' && savedDefs.definitions) { const defs = savedDefs.definitions as Array for (const def of defs) { - this.definitions.set(def.name, def) - const currentHash = hashDefinition(def) const savedHash = def._hash || '' - // Load persisted state + if (this.definitions.has(def.name)) { + // The app re-registered this aggregate before the load finished. + // The app's definition wins — never clobber it with the persisted + // copy. Adopt the persisted state when the definition is unchanged + // AND no write has landed for it yet (a landed write would be lost + // by adoption; the hook flips such names to backfill). + 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) { + const groupMap = new Map() + for (const group of stateData.groups as AggregateGroupState[]) { + groupMap.set(serializeGroupKey(group.groupKey), group) + } + this.states.set(def.name, groupMap) + this.pendingAdopt.delete(def.name) + this.needsBackfill.delete(def.name) + } + // No/invalid persisted state: stays in pendingAdopt and resolves + // to backfill when init settles. + } + continue + } + + // Not registered this session — restore definition + state from + // persistence. + this.definitions.set(def.name, def) + const currentHash = hashDefinition(def) + const stateData = await this.storage.getMetadata(`${STATE_KEY_PREFIX}${def.name}__`) if (stateData && stateData.groups && savedHash === currentHash) { // Definition unchanged — load state @@ -358,6 +442,7 @@ export class AggregationIndex { groupMap.set(serialized, group) } this.states.set(def.name, groupMap) + this.needsBackfill.delete(def.name) } else { // Definition changed or no saved state — start fresh and backfill from // existing entities (the owner drains needsBackfill on first query). @@ -452,10 +537,19 @@ export class AggregationIndex { this.definitions.set(def.name, def) this.definitionHashes.set(def.name, newHash) + // First sight this session, before init() settled: defer the backfill + // decision — init() adopts the persisted state on hash match, and anything + // left unadopted resolves to backfill. Deciding eagerly here wiped valid + // persisted state on every restart. + if (!this.states.has(def.name) && !this.initDone) { + this.states.set(def.name, new Map()) + this.pendingAdopt.add(def.name) + } // Reset state if definition changed or doesn't exist yet, and flag it for // backfill so already-stored entities are counted (write-time hooks only see // future writes). The owner drains this on the next query via getPendingBackfills(). - if (!this.states.has(def.name) || (oldHash && oldHash !== newHash)) { + else if (!this.states.has(def.name) || (oldHash && oldHash !== newHash)) { + this.pendingAdopt.delete(def.name) this.states.set(def.name, new Map()) this.needsBackfill.add(def.name) } @@ -476,6 +570,8 @@ export class AggregationIndex { this.definitionHashes.delete(name) this.states.delete(name) this.staleMinMax.delete(name) + this.pendingAdopt.delete(name) + this.needsBackfill.delete(name) // Notify native provider if (this.nativeProvider?.removeAggregate) { @@ -545,6 +641,20 @@ export class AggregationIndex { // ============= Write-Time Hooks ============= + /** + * A write is landing for an aggregate whose persisted-state adoption is still + * pending — adopting after this write would lose its contribution. Settle the + * decision now: an exact rescan instead of adoption. The window is the few + * milliseconds between a boot-time defineAggregate() and init() completing, + * so this rarely fires; when it does, correctness wins over the walk. + */ + private resolveAdoptOnWrite(name: string): void { + if (this.pendingAdopt.has(name)) { + this.pendingAdopt.delete(name) + this.needsBackfill.add(name) + } + } + /** * Called when an entity is added. Updates all matching aggregates. */ @@ -553,6 +663,7 @@ export class AggregationIndex { for (const [name, def] of this.definitions) { if (!matchesSource(entity, def.source)) continue + this.resolveAdoptOnWrite(name) if (this.nativeProvider) { const results = this.nativeProvider.incrementalUpdate(name, def, entity, 'add') @@ -579,6 +690,10 @@ export class AggregationIndex { const oldMatches = matchesSource(oldEntity, def.source) const newMatches = matchesSource(newEntity, def.source) + if (oldMatches || newMatches) { + this.resolveAdoptOnWrite(name) + } + if (this.nativeProvider && (oldMatches || newMatches)) { const results = this.nativeProvider.incrementalUpdate(name, def, newEntity, 'update', oldEntity) this.applyNativeResults(name, results) @@ -605,6 +720,7 @@ export class AggregationIndex { for (const [name, def] of this.definitions) { if (!matchesSource(entity, def.source)) continue + this.resolveAdoptOnWrite(name) if (this.nativeProvider) { const results = this.nativeProvider.incrementalUpdate(name, def, entity, 'delete') diff --git a/src/brainy.ts b/src/brainy.ts index 383b1195..12fda014 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -598,6 +598,7 @@ export class Brainy implements BrainyInterface { private _hub?: IntegrationHub // Integration Hub for external tools private _pendingMigrationRunner?: MigrationRunner // Deferred migration runner for large datasets private _aggregationIndex?: AggregationIndex // Incremental aggregation engine + private _aggregationBackfillFlight: Promise | null = null // Single-flight backfill walk private _materializer?: AggregateMaterializer // Debounced materialization of aggregate results /** * Fields registered via `brain.trackField()` — drives optional value validation on @@ -5783,6 +5784,10 @@ export class Brainy implements BrainyInterface { ): Promise { await this.ensureInitialized() this.ensureAggregationIndex() + // Persisted definitions load asynchronously — wait for them before deciding + // the aggregate doesn't exist (an app that relies on persisted definitions + // without re-defining at boot would otherwise race a spurious throw here). + await this._aggregationIndex!.ready() if (!this._aggregationIndex!.hasAggregate(name)) { throw new Error(`Aggregate '${name}' is not defined. Call defineAggregate() first.`) } @@ -15810,9 +15815,42 @@ export class Brainy implements BrainyInterface { */ private async backfillAggregateIfNeeded(name: string): Promise { const index = this._aggregationIndex - if (!index || !index.getPendingBackfills().includes(name)) return + if (!index) return - index.beginBackfill(name) + // Persisted-state adoption happens inside ready(); after it resolves the + // pending-backfill set is authoritative (an unchanged definition with valid + // persisted state is NOT listed — no walk at all on a clean reopen). + await index.ready() + + // Single-flight: concurrent queries share ONE walk instead of each wiping + // the others' partial state and starting their own (the stampede that kept + // a busy store from ever converging). The loop covers the rare case where + // 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)) { + if (!this._aggregationBackfillFlight) { + this._aggregationBackfillFlight = this.runAggregationBackfillWalk() + .finally(() => { + this._aggregationBackfillFlight = null + }) + } + await this._aggregationBackfillFlight + } + } + + /** + * One store walk fills EVERY aggregate currently pending backfill — M pending + * aggregates cost one enumeration, not M. Only reached when an aggregate + * genuinely needs a rescan (new definition over a populated store, changed + * definition, or failed state load); a clean reopen adopts persisted state + * and never walks. + */ + private async runAggregationBackfillWalk(): Promise { + const index = this._aggregationIndex! + const names = index.getPendingBackfills() + if (names.length === 0) return + + for (const n of names) index.beginBackfill(n) const PAGE = 500 let offset = 0 @@ -15822,7 +15860,10 @@ export class Brainy implements BrainyInterface { pagination: cursor ? { limit: PAGE, cursor } : { limit: PAGE, offset } }) for (const noun of page.items) { - index.backfillEntity(name, noun as unknown as Record) + const record = noun as unknown as Record + for (const n of names) { + index.backfillEntity(n, record) + } } if (!page.hasMore || page.items.length === 0) break if (page.nextCursor) { @@ -15832,7 +15873,7 @@ export class Brainy implements BrainyInterface { } } - index.finishBackfill(name) + for (const n of names) index.finishBackfill(n) } /** diff --git a/src/storage/baseStorage.ts b/src/storage/baseStorage.ts index 73b8e859..7c4c7c45 100644 --- a/src/storage/baseStorage.ts +++ b/src/storage/baseStorage.ts @@ -377,7 +377,14 @@ export abstract class BaseStorage extends BaseStorageAdapter { id.startsWith('statistics_') || id === 'statistics' || id.startsWith('__chunk__') || // Metadata index chunks (roaring bitmap data) - id.startsWith('__sparse_index__') // Metadata sparse indices (zone maps + bloom filters) + id.startsWith('__sparse_index__') || // Metadata sparse indices (zone maps + bloom filters) + id.startsWith('__aggregation_') || // Aggregation engine definitions + state (routing is + // identical to the unknown-key fallback these keys hit + // before being listed here — this only kills the + // per-boot "Unknown key format" warning) + isSingletonSystemKey(id) // Known singletons (e.g. brainy:entityIdMapper) hit the + // same warn-then-route fallback without this — the + // routing below already handles them identically if (isSystemKey) { if (isSingletonSystemKey(id)) { diff --git a/src/utils/paramValidation.ts b/src/utils/paramValidation.ts index 57a9a1c0..ca439524 100644 --- a/src/utils/paramValidation.ts +++ b/src/utils/paramValidation.ts @@ -209,8 +209,10 @@ export interface ValidationConfigOptions { } /** - * Auto-configured limits based on system resources - * These adapt to available memory and observed performance + * Auto-configured limits based on system resources. + * Derived from memory (explicit overrides > reserved memory > container limit > + * free memory). Query timing is recorded for diagnostics only — it never + * changes the cap (see `recordQuery`). */ export class ValidationConfig { private static instance: ValidationConfig | null = null @@ -323,24 +325,23 @@ export class ValidationConfig { } /** - * Learn from actual usage to adjust limits + * Record query timing for diagnostics. Telemetry ONLY — never mutates the cap. + * + * `maxLimit` is a MEMORY-protection bound; query duration says nothing about + * memory-per-result, so duration must never drive it. An earlier version + * "learned" here: while the lifetime-average query time exceeded 1s it shrank + * `maxLimit` by 20% per recorded query down to a floor of 1000 — below the + * documented `MIN_AUTO_QUERY_LIMIT` (10 000) and with no recovery once slow + * samples poisoned the cumulative average. On a production host a burst of + * slow aggregate queries silently strangled every consumer's `find()` to + * 1000 while the error message blamed "available free memory" — exactly the + * silent throttling this module's own contract forbids. The cap now comes + * from its construction-time basis (or explicit overrides) alone. */ recordQuery(duration: number, resultCount: number) { + void resultCount this.queryCount++ this.avgQueryTime = (this.avgQueryTime * (this.queryCount - 1) + duration) / this.queryCount - - // Only auto-adjust if not using explicit overrides - if (this.limitBasis !== 'override') { - // If queries are consistently fast with large results, increase limits - if (this.avgQueryTime < 100 && resultCount > this.maxLimit * 0.8) { - this.maxLimit = Math.min(this.maxLimit * 1.5, 100000) - } - - // If queries are slow, reduce limits - if (this.avgQueryTime > 1000) { - this.maxLimit = Math.max(this.maxLimit * 0.8, 1000) - } - } } } diff --git a/tests/integration/aggregation-state-persistence.test.ts b/tests/integration/aggregation-state-persistence.test.ts new file mode 100644 index 00000000..4ec3d22d --- /dev/null +++ b/tests/integration/aggregation-state-persistence.test.ts @@ -0,0 +1,235 @@ +/** + * @module tests/integration/aggregation-state-persistence + * @description The boot-order contract for the aggregation engine. Five laws: + * (1) STATE ADOPTION — a reopen with an unchanged defineAggregate() adopts the + * persisted state and performs NO store walk. (The pre-fix behavior: the + * synchronous define always beat the async init, flagged a backfill, and + * the first query wiped the just-loaded state and re-walked the whole + * store — every restart, forever.) + * (2) SINGLE-FLIGHT + BATCH — concurrent cold queries across multiple pending + * aggregates share exactly ONE store walk; a query never wipes another's + * partial progress and M pending aggregates cost one enumeration, not M. + * (3) CHANGED DEFINITION — a real definition change still backfills, exactly. + * (4) PERSISTED-ONLY DEFINITIONS — an app that does not re-define at boot can + * query a persisted aggregate without racing a spurious "not defined". + * (5) QUIET KEYS — the engine's persistence keys (__aggregation_*) are + * recognized system keys: no "Unknown key format" warning at boot. + */ +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest' +import * as fs from 'node:fs' +import * as os from 'node:os' +import * as path from 'node:path' +import { Brainy } from '../../src/index.js' +import { NounType } from '../../src/types/graphTypes.js' +import type { AggregateDefinition } from '../../src/types/brainy.types.js' +import { prodLog } from '../../src/utils/logger.js' + +const SPENDING: AggregateDefinition = { + name: 'spending', + source: { type: NounType.Event, where: { domain: 'financial' } }, + groupBy: ['category'], + metrics: { + total: { op: 'sum', field: 'amount' }, + count: { op: 'count' } + } +} + +/** Same name, different metrics — a REAL definition change (hash differs). */ +const SPENDING_CHANGED: AggregateDefinition = { + ...SPENDING, + metrics: { + total: { op: 'sum', field: 'amount' }, + count: { op: 'count' }, + average: { op: 'avg', field: 'amount' } + } +} + +describe('aggregation state persistence — boot-order contract', () => { + let dir: string + + beforeEach(() => { + dir = fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-agg-persist-')) + }) + + afterEach(() => { + fs.rmSync(dir, { recursive: true, force: true }) + }) + + const open = async (): Promise => { + const b: any = new Brainy({ + requireSubtype: false, + storage: { type: 'filesystem', path: dir }, + silent: true + }) + await b.init() + return b + } + + /** Count store walks by intercepting the storage adapter's getNouns. */ + const countWalks = (brain: any): { count: () => number } => { + const storage = brain.storage + const orig = storage.getNouns.bind(storage) + let calls = 0 + storage.getNouns = async (opts: unknown) => { + calls++ + return orig(opts) + } + return { count: () => calls } + } + + const seed = async (brain: any): Promise => { + for (let i = 0; i < 12; i++) { + await brain.add({ + data: `tx ${i}`, + type: NounType.Event, + metadata: { + domain: 'financial', + category: i % 2 === 0 ? 'food' : 'transport', + amount: 10 + i + } + }) + } + } + + it('adopts persisted state on reopen with an unchanged definition — zero walks', async () => { + const brain1 = await open() + brain1.defineAggregate(SPENDING) + await seed(brain1) + const before = await brain1.queryAggregate('spending') + expect(before.length).toBe(2) + await brain1.close() + + const brain2 = await open() + brain2.defineAggregate(SPENDING) // the standard declarative boot pattern + await brain2.getNounCount() // settle init paths before counting walks + const walks = countWalks(brain2) + + const after = await brain2.queryAggregate('spending') + + expect(walks.count()).toBe(0) + const key = (r: any) => r.groupKey.category + expect( + after.map((r: any) => [key(r), r.metrics.total, r.metrics.count]).sort() + ).toEqual( + before.map((r: any) => [key(r), r.metrics.total, r.metrics.count]).sort() + ) + await brain2.close() + }) + + it('adopted state keeps accumulating: post-reopen writes land on top of it', async () => { + const brain1 = await open() + brain1.defineAggregate(SPENDING) + await seed(brain1) + await brain1.queryAggregate('spending') + await brain1.close() + + const brain2 = await open() + brain2.defineAggregate(SPENDING) + // A write BEFORE the first query: if it lands before state adoption the + // engine must choose an exact rescan over adoption — either way the + // result must include all 13 entities. + await brain2.add({ + data: 'late tx', + type: NounType.Event, + metadata: { domain: 'financial', category: 'food', amount: 100 } + }) + const rows = await brain2.queryAggregate('spending') + const food = rows.find((r: any) => r.groupKey.category === 'food') + expect(food.metrics.count).toBe(7) // 6 seeded + 1 late + await brain2.close() + }) + + it('a changed definition still backfills — exactly once, with correct results', async () => { + const brain1 = await open() + brain1.defineAggregate(SPENDING) + await seed(brain1) + await brain1.queryAggregate('spending') + await brain1.close() + + const brain2 = await open() + brain2.defineAggregate(SPENDING_CHANGED) + await brain2.getNounCount() + const walks = countWalks(brain2) + + const rows = await brain2.queryAggregate('spending') + + expect(walks.count()).toBe(1) // 12 entities = one page = one getNouns call + const food = rows.find((r: any) => r.groupKey.category === 'food') + expect(food.metrics.count).toBe(6) + expect(food.metrics.average).toBeCloseTo(food.metrics.total / 6) + await brain2.close() + }) + + it('concurrent cold queries across two aggregates share exactly ONE walk', async () => { + const brain = await open() + brain.defineAggregate(SPENDING) + brain.defineAggregate({ + ...SPENDING, + name: 'by_category_count', + metrics: { count: { op: 'count' } } + }) + await seed(brain) + await brain.getNounCount() + const walks = countWalks(brain) + + const results = await Promise.all([ + brain.queryAggregate('spending'), + brain.queryAggregate('by_category_count'), + brain.queryAggregate('spending'), + brain.queryAggregate('by_category_count'), + brain.queryAggregate('spending'), + brain.queryAggregate('by_category_count') + ]) + + expect(walks.count()).toBe(1) // 12 entities = one page; one walk fills both + for (const rows of results) { + const total = rows.reduce((s: number, r: any) => s + r.metrics.count, 0) + expect(total).toBe(12) + } + + // Warm re-query: converged, no further walks. + await brain.queryAggregate('spending') + expect(walks.count()).toBe(1) + await brain.close() + }) + + it('persisted-only definitions are queryable without re-defining at boot', async () => { + const brain1 = await open() + brain1.defineAggregate(SPENDING) + await seed(brain1) + await brain1.queryAggregate('spending') + await brain1.close() + + const brain2 = await open() + // NO defineAggregate — the app relies on the persisted definition. + const walks = countWalks(brain2) + const rows = await brain2.queryAggregate('spending') + expect(walks.count()).toBe(0) // persisted state adopted here too + expect(rows.length).toBe(2) + await brain2.close() + }) + + it('aggregation persistence keys never log "Unknown key format"', async () => { + const warnSpy = vi.spyOn(prodLog, 'warn') + const brain1 = await open() + brain1.defineAggregate(SPENDING) + await seed(brain1) + await brain1.queryAggregate('spending') + await brain1.close() + + const brain2 = await open() + brain2.defineAggregate(SPENDING) + await brain2.queryAggregate('spending') + await brain2.close() + + const offenders = warnSpy.mock.calls + .map(args => String(args[0])) + .filter( + msg => + msg.includes('Unknown key format') && + (msg.includes('__aggregation_') || msg.includes('brainy:entityIdMapper')) + ) + expect(offenders).toEqual([]) + warnSpy.mockRestore() + }) +}) diff --git a/tests/unit/aggregation/AggregationIndex.test.ts b/tests/unit/aggregation/AggregationIndex.test.ts index 05f4a62a..a0c10e62 100644 --- a/tests/unit/aggregation/AggregationIndex.test.ts +++ b/tests/unit/aggregation/AggregationIndex.test.ts @@ -675,4 +675,104 @@ describe('AggregationIndex', () => { await reloaded.close() }) }) + + // ============= Boot-order reconciliation ============= + // + // The production boot pattern: defineAggregate() is synchronous and always + // beats the async init() that loads persisted state. The old code flagged a + // backfill at define time and init never cleared it — so the loaded state + // was wiped and the whole store re-walked on EVERY restart. + + describe('boot-order reconciliation (define-before-init)', () => { + const DEF: AggregateDefinition = { + name: 'boot_agg', + source: { type: NounType.Event }, + groupBy: ['category'], + metrics: { count: { op: 'count' } } + } + + const entity = (id: string, category: string): Record => ({ + id, + noun: NounType.Event, + metadata: { category }, + createdAt: Date.now(), + updatedAt: Date.now() + }) + + /** Simulate the previous session: define, contribute, flush. */ + async function seedAndFlush(store: MemoryStorage): Promise { + const first = new AggregationIndex(store) + await first.init() + first.defineAggregate(DEF) + first.onEntityAdded('e1', entity('e1', 'food')) + first.onEntityAdded('e2', entity('e2', 'food')) + first.onEntityAdded('e3', entity('e3', 'transport')) + await first.flush() + } + + it('adopts persisted state when define beats init with an unchanged definition', async () => { + const store = new MemoryStorage() + await store.init() + await seedAndFlush(store) + + const second = new AggregationIndex(store) + second.defineAggregate(DEF) // synchronous define FIRST — the real boot order + await second.init() + await second.ready() + + expect(second.getPendingBackfills()).toEqual([]) + const rows = second.queryAggregate({ name: 'boot_agg' }) + const food = rows.find(r => r.groupKey.category === 'food')! + expect(food.metrics.count).toBe(2) + }) + + it('a write landing before adoption forces an exact rescan instead', async () => { + const store = new MemoryStorage() + await store.init() + await seedAndFlush(store) + + const second = new AggregationIndex(store) + second.defineAggregate(DEF) + second.onEntityAdded('e4', entity('e4', 'food')) // lands before init settles + await second.init() + await second.ready() + + // Adoption would lose e4's contribution — the engine must rescan. + expect(second.getPendingBackfills()).toEqual(['boot_agg']) + }) + + it('init never clobbers a changed app definition registered before it', async () => { + const store = new MemoryStorage() + await store.init() + await seedAndFlush(store) + + const CHANGED: AggregateDefinition = { + ...DEF, + metrics: { count: { op: 'count' }, total: { op: 'sum', field: 'amount' } } + } + const second = new AggregationIndex(store) + second.defineAggregate(CHANGED) + await second.init() + await second.ready() + + const def = second.getDefinitions().find(d => d.name === 'boot_agg')! + expect(Object.keys(def.metrics).sort()).toEqual(['count', 'total']) + expect(second.getPendingBackfills()).toEqual(['boot_agg']) + }) + + it('init alone restores persisted definitions with adopted state, no backfill', async () => { + const store = new MemoryStorage() + await store.init() + await seedAndFlush(store) + + const second = new AggregationIndex(store) + await second.init() + await second.ready() + + expect(second.hasAggregate('boot_agg')).toBe(true) + expect(second.getPendingBackfills()).toEqual([]) + const rows = second.queryAggregate({ name: 'boot_agg' }) + expect(rows.reduce((s, r) => s + (r.metrics.count as number), 0)).toBe(3) + }) + }) }) diff --git a/tests/unit/utils/paramValidation.test.ts b/tests/unit/utils/paramValidation.test.ts index e7113506..4dc83554 100644 --- a/tests/unit/utils/paramValidation.test.ts +++ b/tests/unit/utils/paramValidation.test.ts @@ -270,27 +270,24 @@ describe('Zero-Config Parameter Validation', () => { expect(config.availableMemory).toBeGreaterThan(0) }) - it('should adapt limits based on query performance', () => { - const initialConfig = getValidationConfig() - const initialLimit = initialConfig.maxLimit - - // Simulate fast queries with large results + it('never mutates the cap from query timing (telemetry only)', () => { + const initialLimit = getValidationConfig().maxLimit + + // Fast queries with large results: no silent growth. for (let i = 0; i < 10; i++) { recordQueryPerformance(50, initialLimit * 0.9) } - - const updatedConfig = getValidationConfig() - // Limit might increase if performance is good - expect(updatedConfig.maxLimit).toBeGreaterThanOrEqual(initialLimit) - - // Simulate slow queries - for (let i = 0; i < 10; i++) { - recordQueryPerformance(2000, 100) + expect(getValidationConfig().maxLimit).toBe(initialLimit) + + // A burst of catastrophically slow queries must not strangle the cap. + // The removed "learning" ratchet shrank it 20% per recorded query down + // to a floor of 1000 — below the documented MIN_AUTO_QUERY_LIMIT — and + // the error message blamed "available free memory" (a production + // incident: every find({ limit: 5000 }) failed on an idle 23GB-free box). + for (let i = 0; i < 50; i++) { + recordQueryPerformance(90_000, 100) } - - const finalConfig = getValidationConfig() - // Limit should decrease if performance is poor - expect(finalConfig.maxLimit).toBeLessThanOrEqual(updatedConfig.maxLimit) + expect(getValidationConfig().maxLimit).toBe(initialLimit) }) }) }) \ No newline at end of file