feat: flush() never compacts — history maintenance moves to close() with bounded passes
flush() is durability work: it must cost what the current window's deltas cost, never what the history backlog costs. Under adaptive retention the byte budget derives from free memory, so bulk-load pressure shrank the budget exactly at peak write volume and flush paid actual reclaim inline — a production deployment measured single writes blocked 25-191s behind reclaim-on-flush. - flush() no longer calls autoCompactHistory(); close() is THE auto-compaction site (already ran there; now alone). - Every auto pass is time-bounded (CLOSE_COMPACTION_BUDGET_MS = 5s): reclamation is oldest-first, so an early stop is a consistent prefix and the next pass resumes. Explicit compactHistory() gains an optional timeBudgetMs for caller-chosen maintenance windows. - Documented trade stated where operators read: a long-lived writer that never closes accumulates history until its next explicit compactHistory() — predictable writes, explicit maintenance. Pins: flush-never-reclaims + close-reclaims-durably (db-mvcc), bounded pass stops-then-resumes as a consistent prefix (generationStore unit).
This commit is contained in:
parent
a16567d626
commit
300d9f2a16
7 changed files with 100 additions and 31 deletions
|
|
@ -344,8 +344,12 @@ For per-entity write coordination (rather than whole-store history), the
|
||||||
## Keeping history bounded
|
## Keeping history bounded
|
||||||
|
|
||||||
Under Model-B every write is a generation, so history can grow quickly —
|
Under Model-B every write is a generation, so history can grow quickly —
|
||||||
Brainy auto-compacts on every `flush()`/`close()` under the **`retention`**
|
Brainy auto-compacts at `close()` (time-bounded per pass) under the
|
||||||
knob (configured on the constructor):
|
**`retention`** knob (configured on the constructor). Since 8.9.0, `flush()`
|
||||||
|
never compacts: flushing is durability work and costs only what the current
|
||||||
|
window's writes cost, regardless of history backlog. A long-lived writer that
|
||||||
|
never closes keeps its history until its next explicit `compactHistory()` —
|
||||||
|
schedule one in your maintenance window if you run bounded retention:
|
||||||
|
|
||||||
```typescript
|
```typescript
|
||||||
// Zero-config: ADAPTIVE — keep as much history as free disk/RAM allows,
|
// Zero-config: ADAPTIVE — keep as much history as free disk/RAM allows,
|
||||||
|
|
@ -359,10 +363,13 @@ new Brainy({ retention: 'all' })
|
||||||
new Brainy({ retention: { maxGenerations: 1000, maxAge: 7 * 86_400_000, maxBytes: 512 * 1024 ** 2 } })
|
new Brainy({ retention: { maxGenerations: 1000, maxAge: 7 * 86_400_000, maxBytes: 512 * 1024 ** 2 } })
|
||||||
```
|
```
|
||||||
|
|
||||||
Reclaim manually at any time (the same caps):
|
Reclaim manually at any time (the same caps, plus an optional per-pass time
|
||||||
|
budget for maintenance windows — an early stop is a consistent prefix and the
|
||||||
|
next pass resumes):
|
||||||
|
|
||||||
```typescript
|
```typescript
|
||||||
await brain.compactHistory({ maxGenerations: 100, maxAge: 7 * 24 * 60 * 60 * 1000 })
|
await brain.compactHistory({ maxGenerations: 100, maxAge: 7 * 24 * 60 * 60 * 1000 })
|
||||||
|
await brain.compactHistory({ maxBytes: 512 * 1024 ** 2, timeBudgetMs: 10_000 })
|
||||||
```
|
```
|
||||||
|
|
||||||
Compaction never breaks a pinned read — record-sets are reclaimed only when
|
Compaction never breaks a pinned read — record-sets are reclaimed only when
|
||||||
|
|
|
||||||
|
|
@ -396,6 +396,15 @@ export type IndexFamily = 'vector' | 'metadata' | 'graph'
|
||||||
*/
|
*/
|
||||||
const AGGREGATION_BACKFILL_RETRY_COOLDOWN_MS = 30_000
|
const AGGREGATION_BACKFILL_RETRY_COOLDOWN_MS = 30_000
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Time budget for the auto-compaction pass at close() (8.9.0). Bounds how long
|
||||||
|
* a clean shutdown spends reclaiming history backlog — an early stop is a
|
||||||
|
* consistent prefix and the next close/explicit pass resumes. Explicit
|
||||||
|
* `compactHistory()` calls are unbounded unless the caller passes their own
|
||||||
|
* `timeBudgetMs` (maintenance windows choose their own budgets).
|
||||||
|
*/
|
||||||
|
const CLOSE_COMPACTION_BUDGET_MS = 5_000
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* The main Brainy class - Clean, Beautiful, Powerful
|
* The main Brainy class - Clean, Beautiful, Powerful
|
||||||
* REAL IMPLEMENTATION - No stubs, no mocks
|
* REAL IMPLEMENTATION - No stubs, no mocks
|
||||||
|
|
@ -8362,19 +8371,24 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @description Run history compaction under the resolved `retention` policy
|
* @description Run history compaction under the resolved `retention` policy
|
||||||
* when `autoCompact` is on (the default). Invoked from `flush()` and
|
* when `autoCompact` is on (the default). Invoked from `close()` ONLY
|
||||||
* `close()` so generational record-sets cannot accumulate unbounded across a
|
* (8.9.0) — flush() is durability work and never pays maintenance costs; a
|
||||||
* long-lived writer's lifetime.
|
* production deployment measured reclaim-on-flush blocking single writes
|
||||||
|
* for 25-191s under memory pressure. Long-lived writers that never close
|
||||||
|
* accumulate history until their next explicit `compactHistory()` — the
|
||||||
|
* documented trade: predictable writes, explicit maintenance.
|
||||||
*
|
*
|
||||||
* - `'all'` → returns without reclaiming (index compaction for speed still
|
* - `'all'` → returns without reclaiming (index compaction for speed still
|
||||||
* runs elsewhere; history is decoupled and kept).
|
* runs elsewhere; history is decoupled and kept).
|
||||||
* - `'adaptive'` → reclaim oldest-unpinned history down to the byte budget.
|
* - `'adaptive'` → reclaim oldest-unpinned history down to the byte budget.
|
||||||
* - `'explicit'` → apply the supplied `maxGenerations`/`maxAge`/`maxBytes` caps.
|
* - `'explicit'` → apply the supplied `maxGenerations`/`maxAge`/`maxBytes` caps.
|
||||||
*
|
*
|
||||||
|
* Every auto pass is TIME-BOUNDED ({@link CLOSE_COMPACTION_BUDGET_MS}) so a
|
||||||
|
* large backlog can never stall a clean shutdown; the next pass resumes.
|
||||||
* Read-only instances and an explicit `autoCompact: false` skip silently.
|
* Read-only instances and an explicit `autoCompact: false` skip silently.
|
||||||
* Pinned generations are never reclaimed ({@link GenerationStore.compact}).
|
* Pinned generations are never reclaimed ({@link GenerationStore.compact}).
|
||||||
* Failures are logged and swallowed — compaction is housekeeping and must
|
* Failures are logged and swallowed — compaction is housekeeping and must
|
||||||
* never fail a flush or a clean shutdown.
|
* never fail a clean shutdown.
|
||||||
*/
|
*/
|
||||||
private async autoCompactHistory(): Promise<void> {
|
private async autoCompactHistory(): Promise<void> {
|
||||||
// Nothing to compact on a read-only instance or before init wired up the
|
// Nothing to compact on a read-only instance or before init wired up the
|
||||||
|
|
@ -8390,12 +8404,16 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
if (policy.mode === 'adaptive') {
|
if (policy.mode === 'adaptive') {
|
||||||
const budget = this.adaptiveHistoryBudgetBytes(policy.budgetBytes)
|
const budget = this.adaptiveHistoryBudgetBytes(policy.budgetBytes)
|
||||||
if (budget === Infinity) return // no pressure signal → keep everything this pass
|
if (budget === Infinity) return // no pressure signal → keep everything this pass
|
||||||
await this.generationStore.compact({ maxBytes: budget })
|
await this.generationStore.compact({
|
||||||
|
maxBytes: budget,
|
||||||
|
timeBudgetMs: CLOSE_COMPACTION_BUDGET_MS
|
||||||
|
})
|
||||||
} else {
|
} else {
|
||||||
await this.generationStore.compact({
|
await this.generationStore.compact({
|
||||||
maxGenerations: policy.maxGenerations,
|
maxGenerations: policy.maxGenerations,
|
||||||
maxAge: policy.maxAge,
|
maxAge: policy.maxAge,
|
||||||
maxBytes: policy.maxBytes
|
maxBytes: policy.maxBytes,
|
||||||
|
timeBudgetMs: CLOSE_COMPACTION_BUDGET_MS
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
|
|
@ -10540,12 +10558,14 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
this.generationStore.persistCounterNow()
|
this.generationStore.persistCounterNow()
|
||||||
])
|
])
|
||||||
|
|
||||||
// 6. Auto-compact generational history per config.retention (default on).
|
// NOTE (8.9.0): flush() no longer compacts history. Flush is DURABILITY
|
||||||
// Runs after the flush so durable state is in place; respects live
|
// work — it must cost what this window's deltas cost, never what the
|
||||||
// Db pins and an explicit autoCompact: false.
|
// history backlog costs. Auto-compaction (a MAINTENANCE concern) runs at
|
||||||
await this.autoCompactHistory()
|
// close() and via explicit compactHistory(); a production deployment
|
||||||
|
// measured reclaim-on-flush stalling single writes for 25-191s under
|
||||||
|
// memory pressure, which is exactly the class this separation ends.
|
||||||
|
|
||||||
// 7. Stamp the entity tree: which source generation the canonical tree
|
// 6. Stamp the entity tree: which source generation the canonical tree
|
||||||
// reflects + the rollup invariants that verify it whole (the counters
|
// reflects + the rollup invariants that verify it whole (the counters
|
||||||
// persisted in step 1). Written at flush boundaries — the tree tracks
|
// persisted in step 1). Written at flush boundaries — the tree tracks
|
||||||
// every commit by construction, so the stamp is a durable checkpoint,
|
// every commit by construction, so the stamp is a durable checkpoint,
|
||||||
|
|
@ -16173,8 +16193,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
}
|
}
|
||||||
|
|
||||||
// Phase 0b: Auto-compact generational history per config.retention (default
|
// Phase 0b: Auto-compact generational history per config.retention (default
|
||||||
// on) BEFORE the generation store closes below. Respects live Db pins and
|
// on) BEFORE the generation store closes below. This is THE auto-compaction
|
||||||
// an explicit autoCompact: false; no-op on read-only instances.
|
// site (8.9.0 — flush() never compacts): time-bounded per pass, respects
|
||||||
|
// live Db pins and an explicit autoCompact: false; no-op on read-only
|
||||||
|
// instances.
|
||||||
await this.autoCompactHistory()
|
await this.autoCompactHistory()
|
||||||
|
|
||||||
// Phase 1: Flush ALL components in parallel to persist buffered data
|
// Phase 1: Flush ALL components in parallel to persist buffered data
|
||||||
|
|
|
||||||
|
|
@ -2220,6 +2220,11 @@ export class GenerationStore {
|
||||||
const maxAge = options?.maxAge
|
const maxAge = options?.maxAge
|
||||||
const maxBytes = options?.maxBytes
|
const maxBytes = options?.maxBytes
|
||||||
const ageCutoff = maxAge !== undefined ? Date.now() - maxAge : undefined
|
const ageCutoff = maxAge !== undefined ? Date.now() - maxAge : undefined
|
||||||
|
// Bounded maintenance pass (8.9.0): stop reclaiming once the budget is
|
||||||
|
// spent. Safe mid-loop — reclamation is oldest-first, so an early stop
|
||||||
|
// leaves a consistent contiguous prefix and the next pass resumes.
|
||||||
|
const deadline =
|
||||||
|
options?.timeBudgetMs !== undefined ? Date.now() + options.timeBudgetMs : undefined
|
||||||
const noCaps =
|
const noCaps =
|
||||||
maxGenerations === undefined && maxAge === undefined && maxBytes === undefined
|
maxGenerations === undefined && maxAge === undefined && maxBytes === undefined
|
||||||
|
|
||||||
|
|
@ -2233,6 +2238,7 @@ export class GenerationStore {
|
||||||
for (const gen of [...this.committedGensAsc()]) {
|
for (const gen of [...this.committedGensAsc()]) {
|
||||||
// Pins are always exempt: never reclaim a generation a live pin needs.
|
// Pins are always exempt: never reclaim a generation a live pin needs.
|
||||||
if (gen > minPinned) break // committedGensAsc ascending → nothing newer is eligible either
|
if (gen > minPinned) break // committedGensAsc ascending → nothing newer is eligible either
|
||||||
|
if (deadline !== undefined && Date.now() >= deadline) break // budget spent — resume next pass
|
||||||
const delta = await this.getDelta(gen)
|
const delta = await this.getDelta(gen)
|
||||||
if (!noCaps) {
|
if (!noCaps) {
|
||||||
const violatesCount = maxGenerations !== undefined && remainingCount > maxGenerations
|
const violatesCount = maxGenerations !== undefined && remainingCount > maxGenerations
|
||||||
|
|
|
||||||
|
|
@ -176,6 +176,15 @@ export interface CompactHistoryOptions {
|
||||||
* of each surviving generation's serialized record set (`GenerationDelta.bytes`).
|
* of each surviving generation's serialized record set (`GenerationDelta.bytes`).
|
||||||
*/
|
*/
|
||||||
maxBytes?: number
|
maxBytes?: number
|
||||||
|
/**
|
||||||
|
* Stop reclaiming after this many milliseconds even if caps are still
|
||||||
|
* exceeded (8.9.0). Compaction is maintenance — a bounded pass keeps
|
||||||
|
* `close()` (and any explicit maintenance window) from stalling on a large
|
||||||
|
* backlog; the next pass resumes where this one stopped (reclamation is
|
||||||
|
* oldest-first, so an early stop is always a consistent prefix). Unset =
|
||||||
|
* run to completion.
|
||||||
|
*/
|
||||||
|
timeBudgetMs?: number
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
|
||||||
|
|
@ -1737,7 +1737,10 @@ export interface BrainyConfig {
|
||||||
* Under Model-B EVERY write (`transact()` AND single-op `add`/`update`/
|
* Under Model-B EVERY write (`transact()` AND single-op `add`/`update`/
|
||||||
* `remove`/`relate`) produces an immutable generation record-set serving
|
* `remove`/`relate`) produces an immutable generation record-set serving
|
||||||
* historical reads (`asOf()`, pinned `Db` values). Without compaction those
|
* historical reads (`asOf()`, pinned `Db` values). Without compaction those
|
||||||
* accumulate, so Brainy **auto-compacts on every `flush()` and `close()`**.
|
* accumulate, so Brainy **auto-compacts at `close()`** (time-bounded per
|
||||||
|
* pass; 8.9.0 removed compaction from `flush()` — flush is durability work
|
||||||
|
* and never pays maintenance costs). A long-lived writer that never closes
|
||||||
|
* accumulates history until its next explicit `compactHistory()` call.
|
||||||
* Live `Db` pins are ALWAYS exempt from reclamation, in every mode.
|
* Live `Db` pins are ALWAYS exempt from reclamation, in every mode.
|
||||||
*
|
*
|
||||||
* Modes:
|
* Modes:
|
||||||
|
|
@ -1754,7 +1757,7 @@ export interface BrainyConfig {
|
||||||
* the oldest unpinned generations while ANY supplied cap is exceeded
|
* the oldest unpinned generations while ANY supplied cap is exceeded
|
||||||
* (predictable ops). `maxAge` in ms; `maxBytes` total history bytes.
|
* (predictable ops). `maxAge` in ms; `maxBytes` total history bytes.
|
||||||
*
|
*
|
||||||
* `autoCompact: false` disables the automatic flush/close compaction (manage
|
* `autoCompact: false` disables the automatic close() compaction (manage
|
||||||
* manually via `brain.compactHistory()`). `budgetBytes` is the settable
|
* manually via `brain.compactHistory()`). `budgetBytes` is the settable
|
||||||
* adaptive byte budget a coordinator drives (also via `brain.setRetentionBudget()`).
|
* adaptive byte budget a coordinator drives (also via `brain.setRetentionBudget()`).
|
||||||
* Long-term archives belong in `db.persist(path)` snapshots, which compaction
|
* Long-term archives belong in `db.persist(path)` snapshots, which compaction
|
||||||
|
|
@ -1772,7 +1775,7 @@ export interface BrainyConfig {
|
||||||
maxBytes?: number
|
maxBytes?: number
|
||||||
/** Adaptive byte budget for this brain, driven by a coordinator (e.g. cor). */
|
/** Adaptive byte budget for this brain, driven by a coordinator (e.g. cor). */
|
||||||
budgetBytes?: number
|
budgetBytes?: number
|
||||||
/** Run compaction automatically on flush()/close() (default: true). */
|
/** Run compaction automatically at close() (default: true; 8.9.0 — flush() never compacts). */
|
||||||
autoCompact?: boolean
|
autoCompact?: boolean
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1280,24 +1280,32 @@ describe('8.0 Db API — generational MVCC', () => {
|
||||||
await expect(reopened.asOf(1)).rejects.toBeInstanceOf(GenerationCompactedError)
|
await expect(reopened.asOf(1)).rejects.toBeInstanceOf(GenerationCompactedError)
|
||||||
})
|
})
|
||||||
|
|
||||||
it('Model-B retention — setRetentionBudget drives adaptive reclaim on flush; live data intact', async () => {
|
it('Model-B retention — flush() NEVER compacts (8.9.0); adaptive reclaim runs at close()', async () => {
|
||||||
// Default brain → ADAPTIVE retention. A coordinator (e.g. cor's ResourceManager)
|
// Default brain → ADAPTIVE retention with a driven byte budget far below
|
||||||
// pushes a byte budget via setRetentionBudget(); auto-compaction on flush() reclaims
|
// the accumulated history (~13 generations of full-vector before-images).
|
||||||
// oldest history down toward it. Each update's before-image carries the full prior
|
// The 8.9.0 law: flush() is durability-only — it must not reclaim even
|
||||||
// 384-dim vector (~KBs), so ~13 generations far exceed a few-KB budget.
|
// when the budget is exceeded (reclaim-on-flush blocked production writes
|
||||||
const { brain } = await openFsBrain()
|
// for 25-191s). Maintenance runs at close(), time-bounded.
|
||||||
|
const { brain, dir } = await openFsBrain()
|
||||||
const a = uid('ret-budget')
|
const a = uid('ret-budget')
|
||||||
await brain.add({ id: a, type: NounType.Document, data: 'v0', vector: vec(1), metadata: { v: 0 } })
|
await brain.add({ id: a, type: NounType.Document, data: 'v0', vector: vec(1), metadata: { v: 0 } })
|
||||||
for (let v = 1; v <= 12; v++) await brain.update({ id: a, metadata: { v } })
|
for (let v = 1; v <= 12; v++) await brain.update({ id: a, metadata: { v } })
|
||||||
|
|
||||||
brain.setRetentionBudget(6000) // ~6 KB — well below the accumulated history
|
brain.setRetentionBudget(6000) // ~6 KB — well below the accumulated history
|
||||||
await brain.flush() // group-commit + adaptive auto-compaction under the budget
|
await brain.flush()
|
||||||
|
|
||||||
// History was reclaimed (the horizon advanced past the oldest generations)…
|
// flush() paid durability only: nothing reclaimed, all history readable.
|
||||||
expect(generationStoreOf(brain).horizon()).toBeGreaterThan(0)
|
expect(generationStoreOf(brain).horizon()).toBe(0)
|
||||||
await expect(brain.asOf(1)).rejects.toBeInstanceOf(GenerationCompactedError)
|
const probe = await brain.asOf(1) // readable proves nothing was reclaimed…
|
||||||
// …but the budget reclaims HISTORY only — the live record is untouched.
|
await probe.release() // …and MUST be released: a held pin would (correctly)
|
||||||
expect((await brain.get(a))?.metadata?.v).toBe(12)
|
// protect every newer generation through the close() compaction below.
|
||||||
|
await brain.close() // ← THE auto-compaction site now
|
||||||
|
|
||||||
|
// close() reclaimed under the budget; live record intact; horizon durable.
|
||||||
|
const { brain: reopened } = await openFsBrain(dir)
|
||||||
|
expect(generationStoreOf(reopened).horizon()).toBeGreaterThan(0)
|
||||||
|
await expect(reopened.asOf(1)).rejects.toBeInstanceOf(GenerationCompactedError)
|
||||||
|
expect((await reopened.get(a))?.metadata?.v).toBe(12)
|
||||||
})
|
})
|
||||||
|
|
||||||
// ==========================================================================
|
// ==========================================================================
|
||||||
|
|
|
||||||
|
|
@ -488,6 +488,20 @@ describe('db/GenerationStore', () => {
|
||||||
expect(result.removedGenerations).toBe(2)
|
expect(result.removedGenerations).toBe(2)
|
||||||
store.release(2)
|
store.release(2)
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it('timeBudgetMs bounds a pass; the next pass resumes the same prefix', async () => {
|
||||||
|
await manyGens(4)
|
||||||
|
// A spent budget (0ms) stops before reclaiming anything — an early stop
|
||||||
|
// is a consistent prefix, never a partial generation.
|
||||||
|
const bounded = await store.compact({ timeBudgetMs: 0 })
|
||||||
|
expect(bounded.removedGenerations).toBe(0)
|
||||||
|
expect(bounded.horizon).toBe(0)
|
||||||
|
// The next (unbounded) pass picks up exactly where the bounded one
|
||||||
|
// stopped and completes the same work.
|
||||||
|
const resumed = await store.compact()
|
||||||
|
expect(resumed.removedGenerations).toBe(4)
|
||||||
|
expect(resumed.horizon).toBe(4)
|
||||||
|
})
|
||||||
})
|
})
|
||||||
|
|
||||||
// ==========================================================================
|
// ==========================================================================
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue