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
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue