soulcraft/brainy
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases 335 Packages Wiki Activity Actions

fix: aggregation state adoption on reopen + single-flight backfill + query-cap ratchet removal

Browse source 
- 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:
David Snelling 2026-07-17 09:20:09 -07:00
parent 593bb8b0f9
commit da55be7520
10 changed files with 584 additions and 44 deletions
Download patch file Download diff file Expand all files Collapse all files

126
src/aggregation/AggregationIndex.ts
View file

@ -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')

49
src/brainy.ts
View file

@ -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)
}
/**

9
src/storage/baseStorage.ts
View file

@ -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)) {

33
src/utils/paramValidation.ts
View file

@ -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)
}
}
}
}