fix: aggregation state adoption on reopen + single-flight backfill + query-cap ratchet removal
- AggregationIndex: defineAggregate before init no longer forces a backfill. init reconciles instead of clobbering: the app definition wins, persisted state is adopted on hash match, a write landing pre-adoption forces an exact rescan, and a successful state load clears the backfill flag. New ready() settles every adoption decision before query paths consult backfill state. - brainy: backfills are single-flight and batched. Concurrent queries share one store walk and every pending aggregate fills from that same walk; the old behavior let each concurrent query wipe the others' partial state and start its own full walk, so a store under steady aggregate traffic never converged. queryAggregate also waits for persisted definitions before deciding an aggregate does not exist. - paramValidation: recordQuery is telemetry-only. The duration-based cap ratchet (x0.8 per recorded query while lifetime-average exceeded 1s, floored at 1000 - below the documented 10000 auto floor, reported under a stale basis label) is removed; the cap comes from its construction-time basis or explicit overrides alone. - storage: __aggregation_* and singleton system keys (brainy:entityIdMapper) are recognized before the unknown-key warning fires; routing is unchanged. - docs: find-limits cap-immutability note, aggregation reopen/adoption semantics, RELEASES.md 8.5.1 entry.
This commit is contained in:
parent
593bb8b0f9
commit
da55be7520
10 changed files with 584 additions and 44 deletions
33
RELEASES.md
33
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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -327,6 +327,22 @@ export class AggregationIndex {
|
|||
/** Track aggregates with stale MIN/MAX (need lazy recompute) */
|
||||
private staleMinMax = new Map<string, Set<string>>()
|
||||
|
||||
/** Resolves when init() has finished loading persisted definitions/state. */
|
||||
private initPromise: Promise<void> | 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<string>()
|
||||
|
||||
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<void> {
|
||||
init(): Promise<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
// Load persisted definitions
|
||||
const savedDefs = await this.storage.getMetadata(DEFINITIONS_KEY)
|
||||
if (savedDefs && typeof savedDefs === 'object' && savedDefs.definitions) {
|
||||
const defs = savedDefs.definitions as Array<AggregateDefinition & { _hash?: string }>
|
||||
|
||||
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<string, AggregateGroupState>()
|
||||
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')
|
||||
|
|
|
|||
|
|
@ -598,6 +598,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
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<void> | 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<T = any> implements BrainyInterface<T> {
|
|||
): Promise<AggregateResult[]> {
|
||||
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<T = any> implements BrainyInterface<T> {
|
|||
*/
|
||||
private async backfillAggregateIfNeeded(name: string): Promise<void> {
|
||||
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<void> {
|
||||
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<T = any> implements BrainyInterface<T> {
|
|||
pagination: cursor ? { limit: PAGE, cursor } : { limit: PAGE, offset }
|
||||
})
|
||||
for (const noun of page.items) {
|
||||
index.backfillEntity(name, noun as unknown as Record<string, unknown>)
|
||||
const record = noun as unknown as Record<string, unknown>
|
||||
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<T = any> implements BrainyInterface<T> {
|
|||
}
|
||||
}
|
||||
|
||||
index.finishBackfill(name)
|
||||
for (const n of names) index.finishBackfill(n)
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -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)) {
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
235
tests/integration/aggregation-state-persistence.test.ts
Normal file
235
tests/integration/aggregation-state-persistence.test.ts
Normal file
|
|
@ -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<any> => {
|
||||
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<void> => {
|
||||
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()
|
||||
})
|
||||
})
|
||||
|
|
@ -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<string, unknown> => ({
|
||||
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<void> {
|
||||
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)
|
||||
})
|
||||
})
|
||||
})
|
||||
|
|
|
|||
|
|
@ -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
|
||||
it('never mutates the cap from query timing (telemetry only)', () => {
|
||||
const initialLimit = getValidationConfig().maxLimit
|
||||
|
||||
// Simulate fast queries with large results
|
||||
// Fast queries with large results: no silent growth.
|
||||
for (let i = 0; i < 10; i++) {
|
||||
recordQueryPerformance(50, initialLimit * 0.9)
|
||||
}
|
||||
expect(getValidationConfig().maxLimit).toBe(initialLimit)
|
||||
|
||||
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)
|
||||
// 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)
|
||||
})
|
||||
})
|
||||
})
|
||||
Loading…
Add table
Add a link
Reference in a new issue