Compare commits
No commits in common. "main" and "v8.8.1" have entirely different histories.
28 changed files with 87 additions and 1907 deletions
13
CHANGELOG.md
13
CHANGELOG.md
|
|
@ -2,19 +2,6 @@
|
|||
|
||||
All notable changes to this project will be documented in this file. See [standard-version](https://github.com/conventional-changelog/standard-version) for commit guidelines.
|
||||
|
||||
### [8.9.0](https://github.com/soulcraftlabs/brainy/compare/v8.8.2...v8.9.0) (2026-07-19)
|
||||
|
||||
- docs: measured performance envelopes v1 (per-op p50/p95 at 1k and 10k, pure-JS floor) (5cabd78)
|
||||
- fix: release drains in-flight writer-lock heartbeat — no phantom lock after unlink (70e4bc8)
|
||||
- feat: flush() never compacts — history maintenance moves to close() with bounded passes (300d9f2)
|
||||
|
||||
|
||||
### [8.8.2](https://github.com/soulcraftlabs/brainy/compare/v8.8.1...v8.8.2) (2026-07-19)
|
||||
|
||||
- fix: one field-resolution law across aggregation hooks, source.where, removeMany, and find() spellings (945d92d)
|
||||
- chore: push public docs to the soulcraft.com ingest door on release (42037d0)
|
||||
|
||||
|
||||
### [8.8.1](https://github.com/soulcraftlabs/brainy/compare/v8.8.0...v8.8.1) (2026-07-18)
|
||||
|
||||
- fix: O(1) adaptive retention accounting + historyStats fleet audit (6207e48)
|
||||
|
|
|
|||
71
RELEASES.md
71
RELEASES.md
|
|
@ -8,79 +8,8 @@ Full auto-generated changelog: `CHANGELOG.md` · Releases: https://github.com/so
|
|||
- Debugging data, query, or storage behaviour
|
||||
- A new Brainy feature is available that you want to adopt
|
||||
|
||||
## Removed APIs — 7.x → 8.x (the complete ledger)
|
||||
|
||||
Every public API removed at the 8.0 major, with its sanctioned replacement. If your code
|
||||
still calls a left-column name on 8.x it throws (or the config key is rejected) — the
|
||||
replacement is always a one-line change. (Standing contract from 8.9.0 forward: removals
|
||||
happen only at majors, after ≥1 minor of loud runtime deprecation naming the replacement.)
|
||||
|
||||
| Removed (7.x) | Replacement (8.x) |
|
||||
|---|---|
|
||||
| `brain.search(query, k)` | `find({ query })` — semantic; `find({ query, searchMode })` for hybrid |
|
||||
| `brain.getRelations({...})` | `related(id, opts)` for adjacency; `find({ connected: {...} })` for scoped traversal |
|
||||
| `brain.neural()` clustering | `find({ vector })` + aggregation `GROUP BY` |
|
||||
| `Db.search()` | `db.find({ vector })` |
|
||||
| Pre-8.0 storage path aliases (`directory`, `basePath`, …) | one `storage.path` key (old aliases throw) |
|
||||
| Reserved keys inside `metadata` bags (silently remapped in 7.x) | top-level params (`subtype`, `visibility`, `confidence`, `weight`, …) — reserved-in-bag throws |
|
||||
| 7.x COW branches layout (`branches/main/`) | generational MVCC (`asOf()`, `now()`, `db.persist(path)`) — on-disk migration is automatic at first 8.x open |
|
||||
|
||||
The fork/snapshot family (`brain.snapshot()`, `createSnapshot()`, `restoreSnapshot()`)
|
||||
is sometimes cited as a 7.x removal — those methods never existed on 7.x; the 8.0 Db API
|
||||
(`asOf`/`persist`/`restore({confirm})`) is their first real implementation.
|
||||
|
||||
---
|
||||
|
||||
## v8.9.0 — 2026-07-19 (flush is durability-only: history maintenance moves to close())
|
||||
|
||||
The write path stops paying maintenance costs — the last structural piece of the
|
||||
flush-storm class (a production deployment measured single writes blocked 25–191s behind
|
||||
history reclaim running inline on flush under memory pressure):
|
||||
|
||||
- **`flush()` never compacts history.** It persists the current window's deltas and
|
||||
nothing else — its cost no longer depends on history backlog or retention mode, in any
|
||||
configuration. **`close()` is the auto-compaction site** (time-bounded per pass, ~5s;
|
||||
an early stop is a consistent prefix and the next pass resumes).
|
||||
- **`compactHistory()` gains `timeBudgetMs`** — bound your own maintenance windows; the
|
||||
same resumable-prefix guarantee applies.
|
||||
- **The documented trade**: a long-lived writer that never closes accumulates history
|
||||
until its next explicit `compactHistory()`. Predictable writes, explicit maintenance.
|
||||
If you run bounded retention on an always-on service, schedule a periodic
|
||||
`compactHistory({ ...caps, timeBudgetMs })` in your maintenance window.
|
||||
- **New public doc: `docs/performance-envelopes.md`** — measured per-op envelopes
|
||||
(p50/p95 at stated scales, hardware, and backend, with the measuring script cited).
|
||||
Refresh rule going forward: any release touching a measured path re-runs that op's
|
||||
benchmark and updates the envelope in the same release.
|
||||
- **New in this file: the Removed APIs 7.x→8.x table** (top of this document) — every
|
||||
removal with its sanctioned replacement, one place, per the engine-currency contract.
|
||||
Standing from here: removals only at majors, after ≥1 minor of loud runtime deprecation.
|
||||
|
||||
## v8.8.2 — 2026-07-19 (one field-resolution law: reserved-field aggregates stop drifting)
|
||||
|
||||
Four fixes from a consumer conformance audit, all rooted in the same disease — two field-resolution
|
||||
regimes where there must be one:
|
||||
|
||||
- **Aggregates grouped by a RESERVED field (`subtype`, `visibility`, …) now decrement on
|
||||
delete.** The delete/update hooks fed the aggregation engine a partial entity view (type,
|
||||
service, data, metadata only), so a reserved-field `groupBy` resolved to a nonexistent group
|
||||
on the way DOWN — counts drifted upward forever after any delete, and updates that moved an
|
||||
entity between reserved-field groups double-counted it. The hooks now pass the full-fidelity
|
||||
entity view (every reserved field top-level, the same shape the add path uses). If your
|
||||
deployment derives stats from reserved-field aggregates, re-define those aggregates once
|
||||
after upgrading (a changed definition triggers one rescan) or run them fresh — the drifted
|
||||
persisted counts do not self-heal retroactively.
|
||||
- **Aggregation `source.where` on reserved fields now filters** instead of silently matching
|
||||
nothing: the matcher resolves fields through the same resolver `groupBy` uses (top-level
|
||||
standard fields + custom metadata), so `where: { subtype: 'note' }` means what it says.
|
||||
- **`removeMany()` refuses empty/invalid selectors loudly.** A bare array passed positionally
|
||||
(`removeMany([id])` instead of `removeMany({ ids: [id] })`), an empty params object, or
|
||||
`ids: []` used to resolve successfully having deleted nothing. All three now throw.
|
||||
- **`find()` accepts both where-key spellings.** Metadata is flattened at index time
|
||||
(`metadata.entry.title` indexes as `entry.title`); a `metadata.`-prefixed where key now
|
||||
falls back to its flattened spelling when the prefixed one isn't indexed — the
|
||||
"unindexed field(s), returning []" confusion for storage-shaped spellings is gone. (A
|
||||
literal nested custom key named `metadata` still wins when indexed as spelled.)
|
||||
|
||||
## v8.8.1 — 2026-07-18 (flush no longer walks the whole generation history + the import dedup off-switch is now honest)
|
||||
|
||||
### The flush-storm fix (production incident, reported by a long-running deployment)
|
||||
|
|
|
|||
|
|
@ -344,12 +344,8 @@ For per-entity write coordination (rather than whole-store history), the
|
|||
## Keeping history bounded
|
||||
|
||||
Under Model-B every write is a generation, so history can grow quickly —
|
||||
Brainy auto-compacts at `close()` (time-bounded per pass) under the
|
||||
**`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:
|
||||
Brainy auto-compacts on every `flush()`/`close()` under the **`retention`**
|
||||
knob (configured on the constructor):
|
||||
|
||||
```typescript
|
||||
// Zero-config: ADAPTIVE — keep as much history as free disk/RAM allows,
|
||||
|
|
@ -363,13 +359,10 @@ new Brainy({ retention: 'all' })
|
|||
new Brainy({ retention: { maxGenerations: 1000, maxAge: 7 * 86_400_000, maxBytes: 512 * 1024 ** 2 } })
|
||||
```
|
||||
|
||||
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):
|
||||
Reclaim manually at any time (the same caps):
|
||||
|
||||
```typescript
|
||||
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
|
||||
|
|
|
|||
|
|
@ -1,83 +0,0 @@
|
|||
---
|
||||
title: Performance Envelopes
|
||||
slug: guides/performance-envelopes
|
||||
public: true
|
||||
category: guides
|
||||
template: guide
|
||||
order: 40
|
||||
description: Measured per-operation latency envelopes at stated scales — what to expect, on what hardware, and exactly how each number was produced.
|
||||
next:
|
||||
- guides/find-limits
|
||||
---
|
||||
|
||||
# Performance Envelopes
|
||||
|
||||
Every number on this page is **measured, never projected** — produced by the script
|
||||
cited at the bottom, against the built package (the artifact you install), on the stated
|
||||
hardware. Each entry says what was measured, at what scale, on which storage backend.
|
||||
When a release touches a measured path, that operation is re-measured and this page
|
||||
updates in the same release.
|
||||
|
||||
Two scopes to keep straight:
|
||||
|
||||
- **These envelopes are the pure-JS engine** (no native accelerator registered) on
|
||||
filesystem storage. This is the floor every deployment gets from `npm install` alone.
|
||||
- **Accelerated deployments** (the optional native provider) publish their own numbers —
|
||||
this page never claims them.
|
||||
|
||||
## Read operations
|
||||
|
||||
Reads are where the architecture pays off: after the write path has done its indexing
|
||||
work, queries answer from purpose-built indexes without scanning.
|
||||
|
||||
| Operation | 1,000 entities | 10,000 entities | Notes |
|
||||
|---|---|---|---|
|
||||
| `get(id)` (warm) | p50 < 0.1ms | p50 < 0.1ms | served from cache/metadata index |
|
||||
| `find` (metadata: indexed equality + range, limit 100) | p50 1.0ms · p95 1.8ms | p50 7.0ms · p95 8.9ms | column-store bitmap paths |
|
||||
| `related(id)` (per-node adjacency) | p50 < 0.1ms · p95 0.2ms | p50 < 0.1ms | LSM adjacency index — O(degree), scale-independent |
|
||||
| `find` (semantic: embed + HNSW, 1k docs) | p50 178ms · p95 393ms | — | dominated by WASM query embedding (measured on a machine under concurrent load — treat the p95 as an upper bound); the vector search itself is single-digit ms |
|
||||
|
||||
## Write operations
|
||||
|
||||
Under Model-B **every write is its own durable generation** — a single-op `add` pays
|
||||
serialization, before-image staging, and fsync before it acks. That durability is priced
|
||||
into the write path visibly, by design:
|
||||
|
||||
| Operation | 1,000 entities | 10,000 entities | Notes |
|
||||
|---|---|---|---|
|
||||
| `add` (single-op) | p50 167ms · p95 171ms | p50 165ms · p95 172ms | full durable generation per write — flat across scale |
|
||||
| `addMany` (bulk) | ~163ms/entity | ~187ms/entity | **currently per-item commits** — see the honest note below |
|
||||
| `relateMany` | ~0.8ms/edge | ~0.9ms/edge | edges batch efficiently today |
|
||||
| `flush` (steady-state, 1 pending write) | p50 8ms · p95 10ms | p50 45ms · p95 52ms | durability-only since 8.9.0 — cost no longer depends on history backlog or retention mode |
|
||||
|
||||
**The honest note on bulk writes:** `addMany` today commits each item as its own
|
||||
generation (the same durability as single-op `add`, serialized by the single-writer
|
||||
lock), so bulk-load cost is N × single-op cost. Batched chunk commits (one generation
|
||||
and one fsync window per chunk, as `removeMany` already does) are designed into the
|
||||
unified-commit work on the current roadmap. Until that ships, size bulk imports
|
||||
accordingly — 10k entities is minutes, not seconds, on filesystem storage.
|
||||
|
||||
## Open / close
|
||||
|
||||
| Operation | 1,000 entities | 10,000 entities | Notes |
|
||||
|---|---|---|---|
|
||||
| `open` (empty store) | ~560ms | ~190ms | includes embedder initialization |
|
||||
| `open` (warm, populated, clean shutdown) | 763ms | 4.9s | pure-JS vector index load dominates and grows with entity count; the native accelerator exists precisely to remove this |
|
||||
| `close` | bounded | bounded | auto-compaction pass is time-bounded (~5s max) since 8.9.0 |
|
||||
|
||||
A store that was NOT cleanly closed pays index rebuilds on top of the warm-open
|
||||
number (tens of seconds at 10k) — clean shutdown is worth engineering for.
|
||||
|
||||
## How these were produced
|
||||
|
||||
- **Hardware**: Intel Core i9-14900HX (32 threads), 62GB RAM, NVMe, Linux, Node v22.
|
||||
- **Backend**: `storage: { type: 'filesystem' }`, pure JS (no native providers).
|
||||
- **Embeddings**: deterministic stub for non-semantic ops (isolates engine cost);
|
||||
the real WASM embedder for the semantic row (that's what you'll run).
|
||||
- **Method**: p50/p95 over 50–200 samples per op against the built `dist/`;
|
||||
the measuring script ships in the repo history and re-runs per release.
|
||||
|
||||
Numbers on different hardware will differ; the *shape* (sub-2ms indexed reads,
|
||||
~160ms embedding-bound semantic queries, durability-priced writes) is the envelope
|
||||
you should hold your deployment against. If your measurements diverge from these
|
||||
shapes by an order of magnitude, something is wrong — file it.
|
||||
4
package-lock.json
generated
4
package-lock.json
generated
|
|
@ -1,12 +1,12 @@
|
|||
{
|
||||
"name": "@soulcraft/brainy",
|
||||
"version": "8.9.0",
|
||||
"version": "8.8.1",
|
||||
"lockfileVersion": 3,
|
||||
"requires": true,
|
||||
"packages": {
|
||||
"": {
|
||||
"name": "@soulcraft/brainy",
|
||||
"version": "8.9.0",
|
||||
"version": "8.8.1",
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"@msgpack/msgpack": "^3.1.2",
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
{
|
||||
"name": "@soulcraft/brainy",
|
||||
"version": "8.9.0",
|
||||
"version": "8.8.1",
|
||||
"description": "Universal Knowledge Protocol™ - World's first Triple Intelligence database unifying vector, graph, and document search in one API. Stage 3 CANONICAL: 42 nouns × 127 verbs covering 96-97% of all human knowledge.",
|
||||
"main": "dist/index.js",
|
||||
"module": "dist/index.js",
|
||||
|
|
|
|||
|
|
@ -1,116 +0,0 @@
|
|||
#!/usr/bin/env node
|
||||
/**
|
||||
* @module scripts/push-docs
|
||||
* @description Push this repo's PUBLIC docs to the soulcraft.com docs ingest
|
||||
* door after an npm publish (VENUE-DOCS-RELEASE-PUSH — retires the old
|
||||
* build-time docs sync).
|
||||
*
|
||||
* Contract (mirrors the reference implementation on the serving side):
|
||||
* POST {base}/api/docs/ingest
|
||||
* headers: x-service-secret: $DOCS_INGEST_SECRET, Content-Type: application/json
|
||||
* body: { docs: [{ slug, title, markdown, nav: { order, section } }] }
|
||||
* batches of 10, idempotent per slug.
|
||||
*
|
||||
* A doc is public iff its frontmatter has `public: true` AND a `slug`. The
|
||||
* frontmatter is stripped; `category` → nav.section, `order` → nav.order.
|
||||
*
|
||||
* Deliberately NOT pushed: the combined /docs landing index. It spans BOTH
|
||||
* engine corpora (this repo's and the native accelerator's), so a per-repo
|
||||
* push would clobber the union — the index is authored on the serving side.
|
||||
*
|
||||
* Env: DOCS_INGEST_SECRET (required), DOCS_INGEST_BASE (default
|
||||
* https://soulcraft.com). Exits 0 with a LOUD warning when the secret is
|
||||
* absent (the npm publish has already happened; the serving side runs its
|
||||
* interim sync on request) and exits 1 when a push actually fails — the docs
|
||||
* site would silently trail npm otherwise, and that must be visible.
|
||||
*/
|
||||
import * as fs from 'node:fs'
|
||||
import * as path from 'node:path'
|
||||
|
||||
const BASE = (process.env.DOCS_INGEST_BASE || 'https://soulcraft.com').replace(/\/+$/, '')
|
||||
const SECRET = process.env.DOCS_INGEST_SECRET
|
||||
const DOCS_DIR = path.join(path.dirname(new URL(import.meta.url).pathname), '..', 'docs')
|
||||
const BATCH = 10
|
||||
|
||||
if (!SECRET) {
|
||||
console.warn(
|
||||
'⚠️ DOCS PUSH SKIPPED: DOCS_INGEST_SECRET is not set.\n' +
|
||||
' soulcraft.com/docs now TRAILS this npm release until docs are pushed.\n' +
|
||||
' Either export DOCS_INGEST_SECRET and re-run `node scripts/push-docs.js`,\n' +
|
||||
' or ping venue on VENUE-DOCS-RELEASE-PUSH for the interim sync.'
|
||||
)
|
||||
process.exit(0)
|
||||
}
|
||||
|
||||
/** Minimal frontmatter split — returns [meta, body] or [null, raw]. */
|
||||
function parseFrontmatter(raw) {
|
||||
const m = raw.match(/^---\n([\s\S]*?)\n---\n([\s\S]*)$/)
|
||||
if (!m) return [null, raw]
|
||||
const meta = {}
|
||||
for (const line of m[1].split('\n')) {
|
||||
const kv = line.match(/^(\w[\w-]*):\s*(.*)$/)
|
||||
if (kv) meta[kv[1]] = kv[2].trim().replace(/^["']|["']$/g, '')
|
||||
}
|
||||
return [meta, m[2]]
|
||||
}
|
||||
|
||||
const docs = []
|
||||
;(function walk(dir) {
|
||||
for (const entry of fs.readdirSync(dir, { withFileTypes: true })) {
|
||||
const full = path.join(dir, entry.name)
|
||||
if (entry.isDirectory()) walk(full)
|
||||
else if (entry.name.endsWith('.md')) {
|
||||
const [meta, body] = parseFrontmatter(fs.readFileSync(full, 'utf-8'))
|
||||
if (!meta || meta.public !== 'true' || !meta.slug) continue
|
||||
docs.push({
|
||||
slug: meta.slug,
|
||||
title: meta.title || meta.slug,
|
||||
markdown: body.trim(),
|
||||
nav: {
|
||||
order: Number.parseInt(meta.order || '99', 10) || 99,
|
||||
section: meta.category || 'guides'
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
})(DOCS_DIR)
|
||||
|
||||
if (docs.length === 0) {
|
||||
console.error('❌ DOCS PUSH FAILED: zero public docs collected — refusing to push an empty corpus.')
|
||||
process.exit(1)
|
||||
}
|
||||
docs.sort((a, b) => a.slug.localeCompare(b.slug))
|
||||
console.log(`Pushing ${docs.length} public docs to ${BASE}/api/docs/ingest …`)
|
||||
|
||||
let failed = false
|
||||
for (let i = 0; i < docs.length; i += BATCH) {
|
||||
const batch = docs.slice(i, i + BATCH)
|
||||
try {
|
||||
const res = await fetch(`${BASE}/api/docs/ingest`, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'x-service-secret': SECRET,
|
||||
'Content-Type': 'application/json',
|
||||
'User-Agent': 'brainy-docs-push/1.0'
|
||||
},
|
||||
body: JSON.stringify({ docs: batch }),
|
||||
signal: AbortSignal.timeout(120_000)
|
||||
})
|
||||
if (!res.ok) {
|
||||
throw new Error(`HTTP ${res.status}: ${(await res.text()).slice(0, 300)}`)
|
||||
}
|
||||
console.log(` batch ${i / BATCH + 1}: ${batch.map((d) => d.slug).join(', ')} → ok`)
|
||||
} catch (err) {
|
||||
failed = true
|
||||
console.error(` batch ${i / BATCH + 1} FAILED: ${err instanceof Error ? err.message : err}`)
|
||||
}
|
||||
}
|
||||
|
||||
if (failed) {
|
||||
console.error(
|
||||
'❌ DOCS PUSH INCOMPLETE — soulcraft.com/docs may trail npm. ' +
|
||||
'Re-run `node scripts/push-docs.js` or ping venue on VENUE-DOCS-RELEASE-PUSH.'
|
||||
)
|
||||
process.exit(1)
|
||||
}
|
||||
console.log('✅ Docs pushed.')
|
||||
|
|
@ -196,18 +196,6 @@ else
|
|||
fi
|
||||
echo -e "${GREEN}✅ GitHub release created${NC}\n"
|
||||
|
||||
# Step 12: Push public docs to the soulcraft.com docs ingest door
|
||||
# (VENUE-DOCS-RELEASE-PUSH). Skips with a loud warning when
|
||||
# DOCS_INGEST_SECRET is unset; fails loudly (without undoing the publish —
|
||||
# that already happened) when a push errors, so the docs site never
|
||||
# silently trails npm.
|
||||
echo -e "${BLUE}1️⃣2️⃣ Pushing public docs to soulcraft.com/docs...${NC}"
|
||||
if node scripts/push-docs.js; then
|
||||
echo -e "${GREEN}✅ Docs push step done${NC}\n"
|
||||
else
|
||||
echo -e "${RED}❌ Docs push FAILED — soulcraft.com/docs trails npm until re-run or interim sync${NC}\n"
|
||||
fi
|
||||
|
||||
echo -e "${GREEN}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
|
||||
echo -e "${GREEN}🎉 Release ${NEW_VERSION} complete!${NC}"
|
||||
echo -e "${GREEN}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
|
||||
|
|
|
|||
|
|
@ -88,18 +88,10 @@ function matchesSource(entity: Record<string, unknown>, source: AggregateDefinit
|
|||
if (entity.service !== source.service) return false
|
||||
}
|
||||
|
||||
// Where filter — resolve each filtered field through resolveEntityField,
|
||||
// the SAME single source of truth groupBy uses (top-level standard fields
|
||||
// + custom metadata). Matching only the metadata sub-object made
|
||||
// where:{subtype}/{visibility}/… a silent no-op: reserved fields never
|
||||
// live in the custom bag, so those filters could never match anything.
|
||||
// Metadata where filter — match against the entity's metadata sub-object
|
||||
if (source.where && Object.keys(source.where).length > 0) {
|
||||
const e = entity as unknown as HNSWNounWithMetadata
|
||||
const resolved: Record<string, unknown> = {}
|
||||
for (const key of Object.keys(source.where)) {
|
||||
resolved[key] = resolveEntityField(e, key)
|
||||
}
|
||||
if (!matchesMetadataFilter(resolved, source.where)) return false
|
||||
const metadata = (entity.metadata ?? entity) as Record<string, unknown>
|
||||
if (!matchesMetadataFilter(metadata, source.where)) return false
|
||||
}
|
||||
|
||||
return true
|
||||
|
|
|
|||
203
src/brainy.ts
203
src/brainy.ts
|
|
@ -396,15 +396,6 @@ export type IndexFamily = 'vector' | 'metadata' | 'graph'
|
|||
*/
|
||||
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
|
||||
* REAL IMPLEMENTATION - No stubs, no mocks
|
||||
|
|
@ -1892,29 +1883,6 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Build the AGGREGATION view of an entity from a stored flat
|
||||
* metadata record — EVERY reserved field mapped to its top-level entity
|
||||
* name (stored `noun` → `type`), custom metadata in `metadata`. This must
|
||||
* mirror the add-path `entityForIndexing` shape exactly: the aggregation
|
||||
* engine resolves groupBy/where fields via `resolveEntityField`
|
||||
* (top-level standard fields + custom metadata), so a view that drops a
|
||||
* reserved field makes every aggregate grouped by that field decrement a
|
||||
* group that does not exist — counts then drift upward forever after
|
||||
* deletes (SELF-AGGREGATE-DELETE-DRIFT). Do not hand-roll subsets of this.
|
||||
* @param record - The stored flat metadata record (before-image or pre-delete read).
|
||||
* @returns The full-fidelity entity view for aggregation hooks.
|
||||
*/
|
||||
private entityForAggFromRawRecord(record: Record<string, unknown>): Record<string, unknown> {
|
||||
const { reserved, custom } = splitNounMetadataRecord(record)
|
||||
const { noun, ...rest } = reserved
|
||||
return {
|
||||
type: noun,
|
||||
...rest,
|
||||
metadata: custom
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Add an entity (noun) to the brain. Embeds `data` into a vector and
|
||||
* indexes the entity across all three intelligences — vector similarity, graph
|
||||
|
|
@ -3157,16 +3125,15 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
]
|
||||
: undefined)
|
||||
|
||||
// Aggregation hook (outside transaction — derived data). `existing` is
|
||||
// the full get() view — every reserved field top-level — and must be
|
||||
// passed whole: a subset view makes the old-side decrement miss any
|
||||
// reserved-field group (update would then double-count it).
|
||||
// Aggregation hook (outside transaction — derived data)
|
||||
if (this._aggregationIndex) {
|
||||
this._aggregationIndex.onEntityUpdated(
|
||||
params.id,
|
||||
entityForIndexing,
|
||||
existing as unknown as Record<string, unknown>
|
||||
)
|
||||
const oldEntityForAgg = {
|
||||
type: existing.type,
|
||||
service: existing.service,
|
||||
data: existing.data,
|
||||
metadata: existing.metadata
|
||||
}
|
||||
this._aggregationIndex.onEntityUpdated(params.id, entityForIndexing, oldEntityForAgg)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -3278,15 +3245,19 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
]
|
||||
: undefined)
|
||||
|
||||
// Aggregation hook (outside transaction — derived data). The view must
|
||||
// carry EVERY reserved field top-level (not a subset): a groupBy on
|
||||
// subtype/visibility/etc. otherwise decrements a nonexistent group and
|
||||
// the real count never comes down.
|
||||
// Aggregation hook (outside transaction — derived data)
|
||||
if (this._aggregationIndex && metadata) {
|
||||
this._aggregationIndex.onEntityDeleted(
|
||||
id,
|
||||
this.entityForAggFromRawRecord(metadata as Record<string, unknown>)
|
||||
)
|
||||
// Reconstruct entity-like object from stored metadata via the
|
||||
// canonical reserved/custom split (the hand-rolled destructure here
|
||||
// missed subtype/_rev, leaking them into the aggregation view).
|
||||
const { reserved, custom } = splitNounMetadataRecord(metadata)
|
||||
const entityForAgg = {
|
||||
type: reserved.noun,
|
||||
service: reserved.service,
|
||||
data: reserved.data,
|
||||
metadata: custom
|
||||
}
|
||||
this._aggregationIndex.onEntityDeleted(id, entityForAgg)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -6914,30 +6885,6 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
this.assertWritable('removeMany')
|
||||
await this.ensureInitialized()
|
||||
|
||||
// Loud selector validation: a call with no usable selector used to
|
||||
// resolve successfully having deleted NOTHING (total: 0) — the classic
|
||||
// silent no-op being a bare array passed positionally
|
||||
// (removeMany([id]) instead of removeMany({ ids: [id] })). The caller
|
||||
// believes the delete happened; every count derived afterwards is "wrong"
|
||||
// while the engine was never even asked. Refuse instead.
|
||||
if (Array.isArray(params)) {
|
||||
throw new Error(
|
||||
`removeMany() takes a params object, not a bare array — use removeMany({ ids: [...] })`
|
||||
)
|
||||
}
|
||||
if (!params || (!params.ids && !params.type && !params.where)) {
|
||||
throw new Error(
|
||||
`removeMany() requires a selector: { ids } and/or { type, where }. ` +
|
||||
`An empty selector would silently delete nothing — refusing.`
|
||||
)
|
||||
}
|
||||
if (params.ids && params.ids.length === 0) {
|
||||
throw new Error(
|
||||
`removeMany() received ids: [] — an empty id list deletes nothing. ` +
|
||||
`Pass the ids to delete, or omit ids and select by { type, where }.`
|
||||
)
|
||||
}
|
||||
|
||||
// Determine what to delete
|
||||
let idsToDelete: string[] = []
|
||||
|
||||
|
|
@ -8265,27 +8212,6 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
return this.generationStore.compact(options)
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Repack cold generation history into sealed segments —
|
||||
* re-representation, never deletion: every record and delta stays readable
|
||||
* (`asOf()` unchanged); the physical file count drops by orders of
|
||||
* magnitude. Runs automatically (time-bounded) at `close()`; call this for
|
||||
* explicit maintenance windows on long-lived writers. The ONLY history
|
||||
* transform permitted under the archival profile (`retention: 'all'`).
|
||||
* @param options - `timeBudgetMs` bounds the pass (early stop = consistent
|
||||
* prefix, next pass resumes); `batchGenerations` sizes each fold.
|
||||
* @returns Folded generation count and segments created.
|
||||
*/
|
||||
async repackHistory(options?: {
|
||||
timeBudgetMs?: number
|
||||
batchGenerations?: number
|
||||
}): Promise<{ foldedGenerations: number; segmentsCreated: number }> {
|
||||
this.assertWritable('repackHistory')
|
||||
await this.ensureInitialized()
|
||||
await this.generationStore.flushPendingSingleOps()
|
||||
return this.generationStore.repackHistory(options)
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Read-only generational-history footprint for fleet audits:
|
||||
* generation count, total on-disk bytes, generation/timestamp range, the
|
||||
|
|
@ -8313,24 +8239,6 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @description A deterministic content digest of the generation log through
|
||||
* `g` (D8 — gate-to-generation provenance): identical history produces the
|
||||
* identical digest on any machine; divergence produces a different one.
|
||||
* Release gates and suite verdicts pin `{generation, digest}` and verify
|
||||
* both at execution time instead of pinning a git commit. O(segments +
|
||||
* live-tier window), never O(all generations). Throws `RangeError` out of
|
||||
* range and `GenerationCompactedError` below the horizon — a gate can
|
||||
* never silently pin reclaimed history.
|
||||
* @example
|
||||
* const gate = { generation: brain.generation(), digest: await brain.generationDigest(brain.generation()) }
|
||||
*/
|
||||
async generationDigest(g: number): Promise<string> {
|
||||
await this.ensureInitialized()
|
||||
await this.generationStore.flushPendingSingleOps()
|
||||
return this.generationStore.generationDigest(g)
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Drive the adaptive retention byte budget at runtime — the
|
||||
* settable input a machine-level coordinator (e.g. cor's `ResourceManager`,
|
||||
|
|
@ -8410,24 +8318,19 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
|
||||
/**
|
||||
* @description Run history compaction under the resolved `retention` policy
|
||||
* when `autoCompact` is on (the default). Invoked from `close()` ONLY
|
||||
* (8.9.0) — flush() is durability work and never pays maintenance costs; a
|
||||
* 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.
|
||||
* when `autoCompact` is on (the default). Invoked from `flush()` and
|
||||
* `close()` so generational record-sets cannot accumulate unbounded across a
|
||||
* long-lived writer's lifetime.
|
||||
*
|
||||
* - `'all'` → returns without reclaiming (index compaction for speed still
|
||||
* runs elsewhere; history is decoupled and kept).
|
||||
* - `'adaptive'` → reclaim oldest-unpinned history down to the byte budget.
|
||||
* - `'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.
|
||||
* Pinned generations are never reclaimed ({@link GenerationStore.compact}).
|
||||
* Failures are logged and swallowed — compaction is housekeeping and must
|
||||
* never fail a clean shutdown.
|
||||
* never fail a flush or a clean shutdown.
|
||||
*/
|
||||
private async autoCompactHistory(): Promise<void> {
|
||||
// Nothing to compact on a read-only instance or before init wired up the
|
||||
|
|
@ -8443,16 +8346,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
if (policy.mode === 'adaptive') {
|
||||
const budget = this.adaptiveHistoryBudgetBytes(policy.budgetBytes)
|
||||
if (budget === Infinity) return // no pressure signal → keep everything this pass
|
||||
await this.generationStore.compact({
|
||||
maxBytes: budget,
|
||||
timeBudgetMs: CLOSE_COMPACTION_BUDGET_MS
|
||||
})
|
||||
await this.generationStore.compact({ maxBytes: budget })
|
||||
} else {
|
||||
await this.generationStore.compact({
|
||||
maxGenerations: policy.maxGenerations,
|
||||
maxAge: policy.maxAge,
|
||||
maxBytes: policy.maxBytes,
|
||||
timeBudgetMs: CLOSE_COMPACTION_BUDGET_MS
|
||||
maxBytes: policy.maxBytes
|
||||
})
|
||||
}
|
||||
} catch (error) {
|
||||
|
|
@ -9489,9 +9388,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
)
|
||||
plan.touchedNouns.push(params.id)
|
||||
|
||||
// The full planGetEntity view, passed whole — a subset view makes the
|
||||
// old-side decrement miss reserved-field groups (double-count on update).
|
||||
const oldEntityForAgg = existing as unknown as Record<string, unknown>
|
||||
const oldEntityForAgg = {
|
||||
type: existing.type,
|
||||
service: existing.service,
|
||||
data: existing.data,
|
||||
metadata: existing.metadata
|
||||
}
|
||||
plan.postCommit.push(() => {
|
||||
if (this._aggregationIndex) {
|
||||
this._aggregationIndex.onEntityUpdated(params.id, entityForIndexing, oldEntityForAgg)
|
||||
|
|
@ -9612,9 +9514,14 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
}
|
||||
|
||||
if (metadata) {
|
||||
// Mirror of remove()'s aggregation hook — the FULL reserved view, so
|
||||
// reserved-field groupBy decrements find their group.
|
||||
const entityForAgg = this.entityForAggFromRawRecord(metadata as Record<string, unknown>)
|
||||
// Canonical reserved/custom split — mirror of remove()'s aggregation hook.
|
||||
const { reserved, custom } = splitNounMetadataRecord(metadata)
|
||||
const entityForAgg = {
|
||||
type: reserved.noun,
|
||||
service: reserved.service,
|
||||
data: reserved.data,
|
||||
metadata: custom
|
||||
}
|
||||
plan.postCommit.push(() => {
|
||||
if (this._aggregationIndex) {
|
||||
this._aggregationIndex.onEntityDeleted(id, entityForAgg)
|
||||
|
|
@ -10597,14 +10504,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
this.generationStore.persistCounterNow()
|
||||
])
|
||||
|
||||
// NOTE (8.9.0): flush() no longer compacts history. Flush is DURABILITY
|
||||
// work — it must cost what this window's deltas cost, never what the
|
||||
// history backlog costs. Auto-compaction (a MAINTENANCE concern) runs at
|
||||
// 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.
|
||||
// 6. Auto-compact generational history per config.retention (default on).
|
||||
// Runs after the flush so durable state is in place; respects live
|
||||
// Db pins and an explicit autoCompact: false.
|
||||
await this.autoCompactHistory()
|
||||
|
||||
// 6. Stamp the entity tree: which source generation the canonical tree
|
||||
// 7. Stamp the entity tree: which source generation the canonical tree
|
||||
// reflects + the rollup invariants that verify it whole (the counters
|
||||
// persisted in step 1). Written at flush boundaries — the tree tracks
|
||||
// every commit by construction, so the stamp is a durable checkpoint,
|
||||
|
|
@ -16231,21 +16136,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
await this.generationStore.flushPendingSingleOps()
|
||||
}
|
||||
|
||||
// Phase 0b: REPACK cold history into sealed segments (D1+D3 —
|
||||
// re-representation, never deletion; the only history transform under the
|
||||
// archival profile), then auto-compact per config.retention. Repack runs
|
||||
// FIRST so bounded-retention reclaim can drop whole segments. Both are
|
||||
// time-bounded maintenance passes (8.9.0 law: flush() never pays these);
|
||||
// both are housekeeping — failures warn, never fail a clean shutdown.
|
||||
if (!this.isReadOnly && this.generationStore) {
|
||||
try {
|
||||
await this.generationStore.repackHistory({ timeBudgetMs: 5_000 })
|
||||
} catch (error) {
|
||||
console.warn(
|
||||
`History repacking failed (non-fatal): ${error instanceof Error ? error.message : String(error)}`
|
||||
)
|
||||
}
|
||||
}
|
||||
// Phase 0b: Auto-compact generational history per config.retention (default
|
||||
// on) BEFORE the generation store closes below. Respects live Db pins and
|
||||
// an explicit autoCompact: false; no-op on read-only instances.
|
||||
await this.autoCompactHistory()
|
||||
|
||||
// Phase 1: Flush ALL components in parallel to persist buffered data
|
||||
|
|
|
|||
|
|
@ -102,26 +102,12 @@ export interface FactScanBatch {
|
|||
segmentId: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Liveness bound on a scan's FIRST batch (Stage-2 co-freeze, D1 contract):
|
||||
* `batches()` must yield its first batch — or fail loudly — within this many
|
||||
* ms of the first pull. A backlogged or damaged store may be SLOW, but it may
|
||||
* never be SILENT: a consumer awaiting the first batch is otherwise
|
||||
* indistinguishable from a wedge (the exact failure shape a production heal
|
||||
* hit against a generations-backlogged brain).
|
||||
*/
|
||||
export const SCANFACTS_FIRST_BATCH_MS = 10_000
|
||||
|
||||
/** The telemetry a scan OPEN returns (frozen shape). */
|
||||
export interface FactScanHandle {
|
||||
headGeneration: number
|
||||
segmentCount: number
|
||||
approxFactCount: number
|
||||
/**
|
||||
* Ordered batches; a detected gap aborts LOUDLY, never a silent skip.
|
||||
* Liveness contract: the FIRST batch resolves or rejects within
|
||||
* {@link SCANFACTS_FIRST_BATCH_MS} of the first pull — never a silent hang.
|
||||
*/
|
||||
/** Ordered batches; a detected gap aborts LOUDLY, never a silent skip. */
|
||||
batches: () => AsyncGenerator<FactScanBatch>
|
||||
/** Close telemetry — the invariant cross-check, valid after iteration ends. */
|
||||
summary: () => { factsYielded: number; segmentsRead: number }
|
||||
|
|
@ -454,8 +440,6 @@ export class FactLog {
|
|||
toGeneration?: number
|
||||
kinds?: Array<'noun' | 'verb'>
|
||||
batchSize?: number
|
||||
/** Test override for the first-batch liveness bound (default {@link SCANFACTS_FIRST_BATCH_MS}). */
|
||||
firstBatchTimeoutMs?: number
|
||||
}): FactScanHandle {
|
||||
const from = options?.fromGeneration ?? 1
|
||||
const to = options?.toGeneration ?? this.head
|
||||
|
|
@ -530,43 +514,11 @@ export class FactLog {
|
|||
}
|
||||
}
|
||||
|
||||
// Liveness wrapper: the FIRST pull races the contract deadline. Only the
|
||||
// first — the bound is time-to-first-batch (proof the producer is alive),
|
||||
// not per-batch pacing; and it runs only while a pull is actually pending,
|
||||
// so consumer think-time between pulls never counts against the producer.
|
||||
const firstBatchTimeoutMs = options?.firstBatchTimeoutMs ?? SCANFACTS_FIRST_BATCH_MS
|
||||
async function* batchesWithLiveness(this: void): AsyncGenerator<FactScanBatch> {
|
||||
const inner = batches()
|
||||
let timer: NodeJS.Timeout | undefined
|
||||
try {
|
||||
const deadline = new Promise<never>((_, reject) => {
|
||||
timer = setTimeout(
|
||||
() =>
|
||||
reject(
|
||||
new Error(
|
||||
`fact log: scanFacts produced no first batch within ${firstBatchTimeoutMs}ms ` +
|
||||
`(liveness contract) — the store is wedged or unreadably slow; aborting scan LOUDLY ` +
|
||||
`instead of hanging the consumer.`
|
||||
)
|
||||
),
|
||||
firstBatchTimeoutMs
|
||||
)
|
||||
timer.unref?.()
|
||||
})
|
||||
const first = await Promise.race([inner.next(), deadline])
|
||||
if (first.done) return
|
||||
yield first.value
|
||||
} finally {
|
||||
clearTimeout(timer)
|
||||
}
|
||||
yield* inner
|
||||
}
|
||||
|
||||
return {
|
||||
headGeneration: this.head,
|
||||
segmentCount: segments.length + (tailSnapshot.length > 0 ? 1 : 0),
|
||||
approxFactCount,
|
||||
batches: batchesWithLiveness,
|
||||
batches,
|
||||
summary: () => ({ factsYielded, segmentsRead })
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,459 +0,0 @@
|
|||
/**
|
||||
* @module db/generationSegments
|
||||
* @description The generation-segment store — Stage-2 D1+D3+repacking's file
|
||||
* format (co-frozen 2026-07-19; design: the d1-d3-repacking spec).
|
||||
*
|
||||
* Packs CONSECUTIVE cold generations' record-sets (before-images + delta)
|
||||
* into append-once segment files with derived sidecar indexes, so history
|
||||
* scales in SEGMENTS (tens) instead of FILES-PER-GENERATION (hundreds of
|
||||
* thousands), and cold-open reads ONE manifest instead of listing the
|
||||
* backlog. Layout under `_generations/segments/`:
|
||||
*
|
||||
* - `seg-<firstGen, zero-padded 20>.bgs` — magic "BGS1", then one frame per
|
||||
* generation: `u32 payloadLen | u32 crc32c | msgpack payload`. Payload is
|
||||
* POSITIONAL: `[generation, timestamp, delta, records[], flags]` with
|
||||
* records `[kindByte, id, record]`. `flags` reserves encoding evolution
|
||||
* (bit 0 = compressed payload — v1 always 0; a future writer upgrade,
|
||||
* never a format break). Sealed segments are IMMUTABLE — the fact log's
|
||||
* own law, generalized.
|
||||
* - `seg-<firstGen>.idx` — DERIVED sidecar (msgpack): per-generation frame
|
||||
* offsets (point reads = one ranged read, never a listing) + per-id
|
||||
* generation postings (per-id chain rebuilds read only what they need).
|
||||
* Corrupt/missing → rebuilt from its segment in one sequential read,
|
||||
* loudly.
|
||||
* - `manifest.json` — the segment catalogue + `compactedBelow` (D3's
|
||||
* horizon marker). Cold-open reads THIS; the packed backlog is never
|
||||
* listed.
|
||||
*
|
||||
* D3 semantics carried here: bounded-retention reclaim drops WHOLE segments
|
||||
* at boundaries (O(1) per segment, no rewrite); under the archival profile
|
||||
* (`retention: 'all'`) nothing here is ever dropped — folding is the only
|
||||
* transform (re-representation, never deletion).
|
||||
*/
|
||||
|
||||
import { encode as msgpackEncode, decode as msgpackDecode } from '@msgpack/msgpack'
|
||||
import { crc32c } from '../utils/crc32c.js'
|
||||
import type { FactLogStorage } from './factLog.js'
|
||||
import { prodLog } from '../utils/logger.js'
|
||||
|
||||
/** Directory for segment files + manifest, under the generations prefix. */
|
||||
export const SEGMENTS_PREFIX = '_generations/segments'
|
||||
|
||||
/** Target sealed-segment size (co-freeze proposal; tunable on evidence). */
|
||||
export const SEGMENT_TARGET_BYTES = 64 * 1024 * 1024
|
||||
|
||||
const MAGIC = new TextEncoder().encode('BGS1')
|
||||
const FRAME_PREFIX_BYTES = 8 // u32 payloadLen + u32 crc32c
|
||||
const MANIFEST_PATH = `${SEGMENTS_PREFIX}/manifest.json`
|
||||
|
||||
/** One generation's fold input — exactly what the live tier holds for it. */
|
||||
export interface FoldGeneration {
|
||||
generation: number
|
||||
timestamp: number
|
||||
/** The tx.json delta object, carried verbatim. */
|
||||
delta: unknown
|
||||
/** The before-image record-set (empty for record-less generations). */
|
||||
records: Array<{ kind: 'noun' | 'verb'; id: string; record: unknown }>
|
||||
}
|
||||
|
||||
/** Manifest entry for one sealed segment. */
|
||||
export interface SegmentMeta {
|
||||
file: string
|
||||
firstGeneration: number
|
||||
lastGeneration: number
|
||||
frames: number
|
||||
bytes: number
|
||||
/** crc32c of the full segment byte stream — the digest chain's link. */
|
||||
checksum: number
|
||||
}
|
||||
|
||||
interface SegmentManifest {
|
||||
version: 1
|
||||
compactedBelow: number
|
||||
segments: SegmentMeta[]
|
||||
}
|
||||
|
||||
interface SidecarIndex {
|
||||
version: 1
|
||||
/** [generation, frameOffset, frameLen] ascending by generation. */
|
||||
generations: Array<[number, number, number]>
|
||||
/** `${kindByte}:${id}` → ascending generations holding a record for it. */
|
||||
ids: Record<string, number[]>
|
||||
}
|
||||
|
||||
const segmentFileName = (firstGeneration: number): string =>
|
||||
`seg-${String(firstGeneration).padStart(20, '0')}.bgs`
|
||||
const sidecarFileName = (firstGeneration: number): string =>
|
||||
`seg-${String(firstGeneration).padStart(20, '0')}.idx`
|
||||
|
||||
/**
|
||||
* The generation-segment store. Owns the packed tier ONLY — the live
|
||||
* per-generation tier and the routing between tiers belong to
|
||||
* `GenerationStore`. All mutating entry points here are called under the
|
||||
* generation store's commit mutex.
|
||||
*/
|
||||
export class GenerationSegmentStore {
|
||||
private readonly storage: FactLogStorage
|
||||
private manifest: SegmentManifest = { version: 1, compactedBelow: 0, segments: [] }
|
||||
/** Sidecar cache — segments are immutable, so entries never invalidate. */
|
||||
private readonly sidecars = new Map<string, SidecarIndex>()
|
||||
|
||||
constructor(storage: FactLogStorage) {
|
||||
this.storage = storage
|
||||
}
|
||||
|
||||
/** Load the manifest (ONE read — never a directory listing). */
|
||||
async open(): Promise<void> {
|
||||
const raw = (await this.storage.readRawObject(MANIFEST_PATH)) as SegmentManifest | null
|
||||
if (raw) {
|
||||
if (raw.version !== 1) {
|
||||
throw new Error(
|
||||
`[GenerationSegments] manifest version ${String(raw.version)} is newer than this ` +
|
||||
`engine understands — refusing to serve partial history. Upgrade the engine.`
|
||||
)
|
||||
}
|
||||
this.manifest = raw
|
||||
}
|
||||
}
|
||||
|
||||
/** The packed tier's catalogue (ascending, immutable snapshot). */
|
||||
segments(): readonly SegmentMeta[] {
|
||||
return this.manifest.segments
|
||||
}
|
||||
|
||||
/** D3's horizon marker: generations below this were reclaimed (bounded profiles only). */
|
||||
compactedBelow(): number {
|
||||
return this.manifest.compactedBelow
|
||||
}
|
||||
|
||||
/** The covering sealed segment for `gen`, or null if it lives outside the packed tier. */
|
||||
private coveringSegment(gen: number): SegmentMeta | null {
|
||||
// Manifest is ascending and ranges never overlap — binary search.
|
||||
const segs = this.manifest.segments
|
||||
let lo = 0
|
||||
let hi = segs.length - 1
|
||||
while (lo <= hi) {
|
||||
const mid = (lo + hi) >> 1
|
||||
const s = segs[mid]
|
||||
if (gen < s.firstGeneration) hi = mid - 1
|
||||
else if (gen > s.lastGeneration) lo = mid + 1
|
||||
else return s
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
/** True when `gen` is packed (readable from this tier). */
|
||||
hasGeneration(gen: number): boolean {
|
||||
return this.coveringSegment(gen) !== null
|
||||
}
|
||||
|
||||
/**
|
||||
* Fold consecutive generations into ONE new sealed segment + sidecar and
|
||||
* append it to the manifest atomically. Caller guarantees: `gens` is
|
||||
* ascending, contiguous with the packed tier (first = last packed + 1 when
|
||||
* segments exist), and already durable in the live tier. Crash between the
|
||||
* segment write and the caller's live-tier delete leaves a DUPLICATE
|
||||
* representation — resolved live-tier-wins by the reader; never a gap.
|
||||
*/
|
||||
async fold(gens: FoldGeneration[]): Promise<SegmentMeta> {
|
||||
if (gens.length === 0) {
|
||||
throw new Error('[GenerationSegments] fold() requires at least one generation')
|
||||
}
|
||||
for (let i = 1; i < gens.length; i++) {
|
||||
if (gens[i].generation <= gens[i - 1].generation) {
|
||||
throw new Error('[GenerationSegments] fold() input must be strictly ascending')
|
||||
}
|
||||
}
|
||||
const last = this.manifest.segments[this.manifest.segments.length - 1]
|
||||
if (last && gens[0].generation <= last.lastGeneration) {
|
||||
throw new Error(
|
||||
`[GenerationSegments] fold() overlaps the packed tier: ${gens[0].generation} ≤ ` +
|
||||
`sealed ${last.lastGeneration} — segments are immutable, never rewritten`
|
||||
)
|
||||
}
|
||||
|
||||
const first = gens[0].generation
|
||||
const file = segmentFileName(first)
|
||||
const sidecar: SidecarIndex = { version: 1, generations: [], ids: {} }
|
||||
|
||||
// Encode all frames, tracking offsets for the sidecar.
|
||||
const parts: Uint8Array[] = [MAGIC]
|
||||
let offset = MAGIC.length
|
||||
for (const g of gens) {
|
||||
const payload = msgpackEncode([
|
||||
g.generation,
|
||||
g.timestamp,
|
||||
g.delta,
|
||||
g.records.map((r) => [r.kind === 'noun' ? 0 : 1, r.id, r.record]),
|
||||
0 // flags: v1 = uncompressed
|
||||
])
|
||||
const frame = new Uint8Array(FRAME_PREFIX_BYTES + payload.length)
|
||||
const view = new DataView(frame.buffer)
|
||||
view.setUint32(0, payload.length, true)
|
||||
view.setUint32(4, crc32c(payload), true)
|
||||
frame.set(payload, FRAME_PREFIX_BYTES)
|
||||
sidecar.generations.push([g.generation, offset, frame.length])
|
||||
for (const r of g.records) {
|
||||
const key = `${r.kind === 'noun' ? 0 : 1}:${r.id}`
|
||||
;(sidecar.ids[key] ??= []).push(g.generation)
|
||||
}
|
||||
parts.push(frame)
|
||||
offset += frame.length
|
||||
}
|
||||
const total = parts.reduce((n, p) => n + p.length, 0)
|
||||
const bytes = new Uint8Array(total)
|
||||
let at = 0
|
||||
for (const p of parts) {
|
||||
bytes.set(p, at)
|
||||
at += p.length
|
||||
}
|
||||
|
||||
const meta: SegmentMeta = {
|
||||
file,
|
||||
firstGeneration: first,
|
||||
lastGeneration: gens[gens.length - 1].generation,
|
||||
frames: gens.length,
|
||||
bytes: total,
|
||||
checksum: crc32c(bytes)
|
||||
}
|
||||
|
||||
// Durability order: segment + sidecar fsync'd BEFORE the manifest names
|
||||
// them (a crash before the manifest = invisible orphan files, harmless);
|
||||
// manifest last, atomically.
|
||||
const segPath = `${SEGMENTS_PREFIX}/${file}`
|
||||
const idxPath = `${SEGMENTS_PREFIX}/${sidecarFileName(first)}`
|
||||
await this.storage.writeRawBytes(segPath, bytes)
|
||||
await this.storage.writeRawBytes(idxPath, msgpackEncode(sidecar))
|
||||
await this.storage.syncRawObjects([segPath, idxPath])
|
||||
const next: SegmentManifest = {
|
||||
...this.manifest,
|
||||
segments: [...this.manifest.segments, meta]
|
||||
}
|
||||
await this.storage.writeRawObject(MANIFEST_PATH, next)
|
||||
await this.storage.syncRawObjects([MANIFEST_PATH])
|
||||
this.manifest = next
|
||||
this.sidecars.set(file, sidecar)
|
||||
return meta
|
||||
}
|
||||
|
||||
/** Load (or rebuild, loudly) a segment's sidecar. */
|
||||
private async sidecarFor(meta: SegmentMeta): Promise<SidecarIndex> {
|
||||
const cached = this.sidecars.get(meta.file)
|
||||
if (cached) return cached
|
||||
const idxPath = `${SEGMENTS_PREFIX}/${sidecarFileName(meta.firstGeneration)}`
|
||||
const raw = await this.storage.readRawBytes(idxPath)
|
||||
if (raw) {
|
||||
try {
|
||||
const idx = msgpackDecode(raw) as SidecarIndex
|
||||
if (idx.version === 1) {
|
||||
this.sidecars.set(meta.file, idx)
|
||||
return idx
|
||||
}
|
||||
} catch {
|
||||
// fall through to rebuild
|
||||
}
|
||||
}
|
||||
// Sidecars are DERIVED: rebuild from the segment, loudly — never serve
|
||||
// wrong offsets silently.
|
||||
prodLog.warn(
|
||||
`[GenerationSegments] sidecar for ${meta.file} missing or unreadable — rebuilding from the segment`
|
||||
)
|
||||
const rebuilt = await this.rebuildSidecar(meta)
|
||||
await this.storage.writeRawBytes(idxPath, msgpackEncode(rebuilt))
|
||||
this.sidecars.set(meta.file, rebuilt)
|
||||
return rebuilt
|
||||
}
|
||||
|
||||
/** One sequential read of the segment → a fresh sidecar. Verifies every frame CRC. */
|
||||
private async rebuildSidecar(meta: SegmentMeta): Promise<SidecarIndex> {
|
||||
const frames = await this.readAllFrames(meta)
|
||||
const idx: SidecarIndex = { version: 1, generations: [], ids: {} }
|
||||
for (const f of frames) {
|
||||
idx.generations.push([f.generation, f.offset, f.frameLen])
|
||||
for (const r of f.records) {
|
||||
const key = `${r.kind === 'noun' ? 0 : 1}:${r.id}`
|
||||
;(idx.ids[key] ??= []).push(f.generation)
|
||||
}
|
||||
}
|
||||
return idx
|
||||
}
|
||||
|
||||
private decodeFrame(
|
||||
payload: Uint8Array
|
||||
): { generation: number; timestamp: number; delta: unknown; records: FoldGeneration['records'] } {
|
||||
const [generation, timestamp, delta, rawRecords] = msgpackDecode(payload) as [
|
||||
number,
|
||||
number,
|
||||
unknown,
|
||||
Array<[number, string, unknown]>,
|
||||
number
|
||||
]
|
||||
return {
|
||||
generation,
|
||||
timestamp,
|
||||
delta,
|
||||
records: rawRecords.map(([kindByte, id, record]) => ({
|
||||
kind: kindByte === 0 ? ('noun' as const) : ('verb' as const),
|
||||
id,
|
||||
record
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
private async readAllFrames(meta: SegmentMeta): Promise<
|
||||
Array<ReturnType<GenerationSegmentStore['decodeFrame']> & { offset: number; frameLen: number }>
|
||||
> {
|
||||
const bytes = await this.storage.readRawBytes(`${SEGMENTS_PREFIX}/${meta.file}`)
|
||||
if (!bytes) {
|
||||
throw new Error(
|
||||
`[GenerationSegments] sealed segment ${meta.file} is MISSING — packed history is damaged; ` +
|
||||
`refusing to continue silently`
|
||||
)
|
||||
}
|
||||
const out: Array<ReturnType<GenerationSegmentStore['decodeFrame']> & { offset: number; frameLen: number }> = []
|
||||
let at = MAGIC.length
|
||||
const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength)
|
||||
while (at + FRAME_PREFIX_BYTES <= bytes.length) {
|
||||
const payloadLen = view.getUint32(at, true)
|
||||
const crc = view.getUint32(at + 4, true)
|
||||
const payload = bytes.subarray(at + FRAME_PREFIX_BYTES, at + FRAME_PREFIX_BYTES + payloadLen)
|
||||
if (payload.length !== payloadLen || crc32c(payload) !== crc) {
|
||||
throw new Error(
|
||||
`[GenerationSegments] frame CRC mismatch in ${meta.file} at offset ${at} — ` +
|
||||
`packed history is damaged; refusing to serve it`
|
||||
)
|
||||
}
|
||||
out.push({ ...this.decodeFrame(payload), offset: at, frameLen: FRAME_PREFIX_BYTES + payloadLen })
|
||||
at += FRAME_PREFIX_BYTES + payloadLen
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
/** Read one packed generation's frame via its sidecar offset (one ranged read). */
|
||||
private async readFrame(
|
||||
gen: number
|
||||
): Promise<ReturnType<GenerationSegmentStore['decodeFrame']> | null> {
|
||||
const meta = this.coveringSegment(gen)
|
||||
if (!meta) return null
|
||||
const idx = await this.sidecarFor(meta)
|
||||
// generations ascending → binary search.
|
||||
const gens = idx.generations
|
||||
let lo = 0
|
||||
let hi = gens.length - 1
|
||||
while (lo <= hi) {
|
||||
const mid = (lo + hi) >> 1
|
||||
if (gens[mid][0] < gen) lo = mid + 1
|
||||
else if (gens[mid][0] > gen) hi = mid - 1
|
||||
else {
|
||||
const [, offset, frameLen] = gens[mid]
|
||||
const bytes = await this.storage.readRawBytes(`${SEGMENTS_PREFIX}/${meta.file}`)
|
||||
if (!bytes) {
|
||||
throw new Error(`[GenerationSegments] sealed segment ${meta.file} is MISSING`)
|
||||
}
|
||||
const frame = bytes.subarray(offset, offset + frameLen)
|
||||
const view = new DataView(frame.buffer, frame.byteOffset, frame.byteLength)
|
||||
const payloadLen = view.getUint32(0, true)
|
||||
const crc = view.getUint32(4, true)
|
||||
const payload = frame.subarray(FRAME_PREFIX_BYTES, FRAME_PREFIX_BYTES + payloadLen)
|
||||
if (payload.length !== payloadLen || crc32c(payload) !== crc) {
|
||||
throw new Error(
|
||||
`[GenerationSegments] frame CRC mismatch for generation ${gen} in ${meta.file} — ` +
|
||||
`packed history is damaged; refusing to serve it`
|
||||
)
|
||||
}
|
||||
return this.decodeFrame(payload)
|
||||
}
|
||||
}
|
||||
// In the covering range but not present: the packed tier is dense by
|
||||
// construction (fold packs every generation it is handed, including
|
||||
// record-less ones) — absence inside a sealed range is damage.
|
||||
throw new Error(
|
||||
`[GenerationSegments] generation ${gen} is inside sealed segment ${meta.file}'s declared ` +
|
||||
`range but has no frame — packed history is damaged`
|
||||
)
|
||||
}
|
||||
|
||||
/** The packed tier's delta for `gen` (null = not packed). */
|
||||
async readDelta(gen: number): Promise<{ delta: unknown; timestamp: number } | null> {
|
||||
const frame = await this.readFrame(gen)
|
||||
return frame ? { delta: frame.delta, timestamp: frame.timestamp } : null
|
||||
}
|
||||
|
||||
/** The packed tier's full record-set for `gen` (null = not packed). */
|
||||
async readRecords(gen: number): Promise<FoldGeneration['records'] | null> {
|
||||
const frame = await this.readFrame(gen)
|
||||
return frame ? frame.records : null
|
||||
}
|
||||
|
||||
/** One packed before-image (null = not packed OR no record for the id in that generation). */
|
||||
async readRecord(gen: number, kind: 'noun' | 'verb', id: string): Promise<unknown | null> {
|
||||
const frame = await this.readFrame(gen)
|
||||
if (!frame) return null
|
||||
const hit = frame.records.find((r) => r.kind === kind && r.id === id)
|
||||
return hit ? hit.record : null
|
||||
}
|
||||
|
||||
/**
|
||||
* D3 reclaim: drop WHOLE segments whose lastGeneration < `belowGeneration`
|
||||
* and bump `compactedBelow`. Partial segments are never dropped — the
|
||||
* boundary waits. NEVER called under the archival profile (the caller
|
||||
* enforces retention semantics; this method only executes boundary drops).
|
||||
*/
|
||||
async dropSegmentsBelow(belowGeneration: number): Promise<{ dropped: number; compactedBelow: number }> {
|
||||
const keep: SegmentMeta[] = []
|
||||
const drop: SegmentMeta[] = []
|
||||
for (const s of this.manifest.segments) {
|
||||
;(s.lastGeneration < belowGeneration ? drop : keep).push(s)
|
||||
}
|
||||
if (drop.length === 0) {
|
||||
return { dropped: 0, compactedBelow: this.manifest.compactedBelow }
|
||||
}
|
||||
const compactedBelow = Math.max(
|
||||
this.manifest.compactedBelow,
|
||||
drop[drop.length - 1].lastGeneration + 1
|
||||
)
|
||||
// Manifest first (the drop is authoritative once named), then bytes —
|
||||
// a crash between leaves orphan segment files invisible to the manifest,
|
||||
// harmless and re-collectable.
|
||||
const next: SegmentManifest = { ...this.manifest, compactedBelow, segments: keep }
|
||||
await this.storage.writeRawObject(MANIFEST_PATH, next)
|
||||
await this.storage.syncRawObjects([MANIFEST_PATH])
|
||||
this.manifest = next
|
||||
for (const s of drop) {
|
||||
await this.storage.deleteRawObject(`${SEGMENTS_PREFIX}/${s.file}`)
|
||||
await this.storage.deleteRawObject(`${SEGMENTS_PREFIX}/${sidecarFileName(s.firstGeneration)}`)
|
||||
this.sidecars.delete(s.file)
|
||||
}
|
||||
return { dropped: drop.length, compactedBelow }
|
||||
}
|
||||
|
||||
/**
|
||||
* D8 rider — the packed portion of `generationDigest(g)`: a deterministic
|
||||
* crc32c chain over sealed-segment checksums fully below `g`, plus the
|
||||
* frame CRC of `g`'s own frame when `g` is mid-segment. O(segments), not
|
||||
* O(generations); identical history ⇒ identical digest on any machine.
|
||||
* The live-tier portion is composed by the caller.
|
||||
*/
|
||||
async digestThroughPacked(g: number): Promise<number | null> {
|
||||
let digest = 0
|
||||
let covered = false
|
||||
for (const s of this.manifest.segments) {
|
||||
if (s.lastGeneration <= g) {
|
||||
digest = crc32c(new TextEncoder().encode(`${digest}:${s.checksum}`))
|
||||
if (s.lastGeneration === g) covered = true
|
||||
} else if (s.firstGeneration <= g) {
|
||||
// g is mid-segment: chain the partial prefix via g's frame CRC.
|
||||
const frame = await this.readFrame(g)
|
||||
if (frame === null) return null
|
||||
const idx = await this.sidecarFor(s)
|
||||
const upTo = idx.generations.filter(([gen]) => gen <= g)
|
||||
for (const [gen, offset, frameLen] of upTo) {
|
||||
digest = crc32c(new TextEncoder().encode(`${digest}:${gen}:${offset}:${frameLen}`))
|
||||
}
|
||||
covered = true
|
||||
break
|
||||
}
|
||||
}
|
||||
return covered || this.manifest.segments.length > 0 ? digest : null
|
||||
}
|
||||
}
|
||||
|
|
@ -46,8 +46,6 @@ import type {
|
|||
TxLogEntry
|
||||
} from './types.js'
|
||||
import { FactLog, storageSupportsFactLog, type CommitFact, type FactOp } from './factLog.js'
|
||||
import { GenerationSegmentStore, type FoldGeneration } from './generationSegments.js'
|
||||
import { crc32c } from '../utils/crc32c.js'
|
||||
|
||||
/**
|
||||
* The byte-identical before-images of every id a commit touches, read UNDER
|
||||
|
|
@ -268,21 +266,6 @@ export class GenerationStore {
|
|||
*/
|
||||
private historyBytesTotal: number | null = null
|
||||
|
||||
/**
|
||||
* The packed tier (D1+D3): sealed segments holding folded cold
|
||||
* generations. Null until {@link open} wires it (and on storage adapters
|
||||
* without raw-byte primitives — the live tier then carries everything,
|
||||
* exactly as before the packed tier existed).
|
||||
*/
|
||||
private segments: GenerationSegmentStore | null = null
|
||||
|
||||
/**
|
||||
* Live-tier window: generations newer than `committed - REPACK_LIVE_WINDOW`
|
||||
* are never folded — the hot tail stays in the per-generation layout the
|
||||
* write path owns. Matches the resident chain window's scale.
|
||||
*/
|
||||
static readonly REPACK_LIVE_WINDOW = 1024
|
||||
|
||||
/**
|
||||
* Model-B per-write group-commit — the in-memory PENDING tier.
|
||||
*
|
||||
|
|
@ -450,33 +433,6 @@ export class GenerationStore {
|
|||
this.factLog = null
|
||||
}
|
||||
|
||||
// PACKED TIER (D1+D3): same capability gate as the fact log. Opening
|
||||
// reads ONE manifest — never a listing of the packed backlog — and seeds
|
||||
// committedRanges with the sealed ranges so packed generations resolve
|
||||
// exactly like live ones.
|
||||
if (storageSupportsFactLog(this.storage)) {
|
||||
this.segments = new GenerationSegmentStore(this.storage)
|
||||
await this.segments.open()
|
||||
const packedRanges = this.segments
|
||||
.segments()
|
||||
.map((s): [number, number] => [s.firstGeneration, Math.min(s.lastGeneration, this.committed)])
|
||||
.filter(([lo, hi]) => lo <= hi)
|
||||
if (packedRanges.length > 0) {
|
||||
// Merge packed (older) + live (newer) interval sets — both ascending;
|
||||
// coalesce adjacency so range arithmetic stays interval-exact.
|
||||
const merged: Array<[number, number]> = []
|
||||
for (const r of [...packedRanges, ...this.committedRanges].sort((a, b) => a[0] - b[0])) {
|
||||
const last = merged[merged.length - 1]
|
||||
if (last && r[0] <= last[1] + 1) last[1] = Math.max(last[1], r[1])
|
||||
else merged.push([r[0], r[1]])
|
||||
}
|
||||
this.committedRanges = merged
|
||||
}
|
||||
this.horizonGen = Math.max(this.horizonGen, this.segments.compactedBelow() - 1)
|
||||
} else {
|
||||
this.segments = null
|
||||
}
|
||||
|
||||
// Hook single-op write batches so generation() is always meaningful.
|
||||
// Suppressed while a transact batch executes (the batch is ONE generation).
|
||||
if (!options?.readOnly) {
|
||||
|
|
@ -544,51 +500,6 @@ export class GenerationStore {
|
|||
* deltas (cache-bounded reads).
|
||||
* @returns Counts, bytes, generation range, and the compaction horizon.
|
||||
*/
|
||||
/**
|
||||
* @description D8 (gate-to-generation provenance): a deterministic content
|
||||
* digest of the generation log THROUGH `g` — identical history ⇒ identical
|
||||
* digest on any machine; any divergence (different records, different
|
||||
* order, reclaimed range) ⇒ different digest. Composed from the packed
|
||||
* tier's sealed-segment checksum chain (O(segments)) plus the live tier's
|
||||
* per-generation delta digests (O(live window at most)). Release gates pin
|
||||
* {generation, digest} and verify both at execution time.
|
||||
* @param g - The generation to digest through (≤ committed).
|
||||
* @returns A hex digest string, stable across reopen and repacking states
|
||||
* ONLY for fully-packed prefixes — repacking changes representation, so
|
||||
* the composed digest is defined over CONTENT: live-tier gens hash their
|
||||
* delta + record ids, packed gens hash via frame CRCs. A gate should pin
|
||||
* after a repack pass for long-term stability, or re-pin on repack.
|
||||
*/
|
||||
async generationDigest(g: number): Promise<string> {
|
||||
if (!Number.isInteger(g) || g < 1 || g > this.committed) {
|
||||
throw new RangeError(
|
||||
`generationDigest(): generation ${g} is out of range [1, ${this.committed}]`
|
||||
)
|
||||
}
|
||||
if (g <= this.horizonGen) {
|
||||
throw new GenerationCompactedError(g, this.horizonGen)
|
||||
}
|
||||
let digest = 0
|
||||
const enc = new TextEncoder()
|
||||
if (this.segments) {
|
||||
const packed = await this.segments.digestThroughPacked(g)
|
||||
if (packed !== null) digest = packed
|
||||
}
|
||||
// Live-tier composition: every committed gen ≤ g not covered by a sealed
|
||||
// segment hashes its delta content in ascending order.
|
||||
for (const gen of this.committedGensAsc()) {
|
||||
if (gen > g) break
|
||||
if (this.segments?.hasGeneration(gen)) continue
|
||||
const delta = await this.getDelta(gen)
|
||||
digest = crc32c(
|
||||
enc.encode(
|
||||
`${digest}:${gen}:${delta.timestamp}:${[...delta.nouns].sort().join(',')}:${[...delta.verbs].sort().join(',')}`
|
||||
)
|
||||
)
|
||||
}
|
||||
return digest.toString(16).padStart(8, '0')
|
||||
}
|
||||
|
||||
async historyStats(): Promise<{
|
||||
generations: number
|
||||
bytes: number
|
||||
|
|
@ -627,17 +538,14 @@ export class GenerationStore {
|
|||
try {
|
||||
paths = await this.storage.listRawObjects(`${GENERATIONS_PREFIX}/${gen}/prev`)
|
||||
} catch {
|
||||
paths = []
|
||||
return []
|
||||
}
|
||||
const records: GenerationRecord[] = []
|
||||
for (const p of paths) {
|
||||
const record = (await this.storage.readRawObject(p)) as GenerationRecord | null
|
||||
if (record) records.push(record)
|
||||
}
|
||||
if (records.length > 0) return records
|
||||
// Two-tier: folded generations serve their record-set from the segment.
|
||||
const packed = await this.segments?.readRecords(gen)
|
||||
return packed ? (packed.map((r) => r.record) as GenerationRecord[]) : []
|
||||
return records
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -1875,15 +1783,9 @@ export class GenerationStore {
|
|||
if (pending) {
|
||||
return (kind === 'noun' ? pending.nouns : pending.verbs).get(id) ?? null
|
||||
}
|
||||
const live = (await this.storage.readRawObject(
|
||||
return (await this.storage.readRawObject(
|
||||
`${GENERATIONS_PREFIX}/${gen}/prev/${id}.json`
|
||||
)) as GenerationRecord | null
|
||||
if (live) return live
|
||||
// Two-tier: the packed tier serves folded generations (live-tier-wins).
|
||||
if (this.segments?.hasGeneration(gen)) {
|
||||
return (await this.segments.readRecord(gen, kind, id)) as GenerationRecord | null
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -2230,21 +2132,6 @@ export class GenerationStore {
|
|||
`${GENERATIONS_PREFIX}/${gen}/tx.json`
|
||||
)) as GenerationDelta | null
|
||||
if (delta === null) {
|
||||
// Two-tier read (D1+D3): not in the live tier → the packed tier.
|
||||
// Live-tier-wins ordering (a crash mid-fold leaves a duplicate, never
|
||||
// a gap), so the segment lookup runs only after the live miss.
|
||||
const packed = await this.segments?.readDelta(gen)
|
||||
if (packed) {
|
||||
const d = packed.delta as GenerationDelta
|
||||
const entry = {
|
||||
nouns: new Set(d.nouns),
|
||||
verbs: new Set(d.verbs),
|
||||
timestamp: packed.timestamp,
|
||||
bytes: d.bytes ?? 0
|
||||
}
|
||||
this.setDelta(gen, entry)
|
||||
return entry
|
||||
}
|
||||
throw new Error(
|
||||
`Generation delta missing: ${GENERATIONS_PREFIX}/${gen}/tx.json ` +
|
||||
`(store corrupted or records removed outside compactHistory())`
|
||||
|
|
@ -2326,94 +2213,6 @@ export class GenerationStore {
|
|||
* @param options - Retention caps (see {@link CompactHistoryOptions}).
|
||||
* @returns Count of removed record-sets and the new horizon.
|
||||
*/
|
||||
/**
|
||||
* @description The REPACKER (D1+D3+repacking): fold cold live-tier
|
||||
* generations into sealed segments — re-representation, never deletion.
|
||||
* Every record and delta stays readable (asOf/chains unchanged); the
|
||||
* per-generation directories are deleted only AFTER their segment is
|
||||
* durable (crash between = duplicate representation, resolved
|
||||
* live-tier-wins by every reader; never a gap). This is the transform that
|
||||
* takes a 70k-file history to tens of segment files, and the ONLY history
|
||||
* transform permitted under the archival profile.
|
||||
*
|
||||
* Folds oldest-first, contiguous from the packed boundary, in batches, and
|
||||
* stops at the live window ({@link GenerationStore.REPACK_LIVE_WINDOW})
|
||||
* or when `timeBudgetMs` is spent — an early stop is a consistent prefix;
|
||||
* the next pass resumes.
|
||||
*/
|
||||
async repackHistory(options?: { timeBudgetMs?: number; batchGenerations?: number }): Promise<{
|
||||
foldedGenerations: number
|
||||
segmentsCreated: number
|
||||
}> {
|
||||
if (!this.segments) return { foldedGenerations: 0, segmentsCreated: 0 }
|
||||
const segments = this.segments
|
||||
return this.withMutex(async () => {
|
||||
const deadline =
|
||||
options?.timeBudgetMs !== undefined ? Date.now() + options.timeBudgetMs : undefined
|
||||
const batchSize = options?.batchGenerations ?? 512
|
||||
const coldCeiling = this.committed - GenerationStore.REPACK_LIVE_WINDOW
|
||||
const packedThrough =
|
||||
segments.segments().length > 0
|
||||
? segments.segments()[segments.segments().length - 1].lastGeneration
|
||||
: 0
|
||||
|
||||
// Cold, unpacked, committed generations — ascending, contiguous scan.
|
||||
const eligible: number[] = []
|
||||
for (const gen of this.committedGensAsc()) {
|
||||
if (gen > coldCeiling) break
|
||||
if (gen <= packedThrough) continue // already packed (dup fold barred)
|
||||
if (this.pendingBuffer.has(gen)) continue // un-flushed = live by definition
|
||||
eligible.push(gen)
|
||||
}
|
||||
|
||||
let folded = 0
|
||||
let segmentsCreated = 0
|
||||
for (let i = 0; i < eligible.length; i += batchSize) {
|
||||
if (deadline !== undefined && Date.now() >= deadline) break
|
||||
const batch = eligible.slice(i, i + batchSize)
|
||||
const foldInput: FoldGeneration[] = []
|
||||
for (const gen of batch) {
|
||||
const delta = (await this.storage.readRawObject(
|
||||
`${GENERATIONS_PREFIX}/${gen}/tx.json`
|
||||
)) as GenerationDelta | null
|
||||
if (delta === null) {
|
||||
// Already folded by a prior crashed pass whose dirs were removed,
|
||||
// or damage — getDelta's two-tier read decides which, loudly,
|
||||
// when someone asks. Skip; never fold a generation we cannot read.
|
||||
continue
|
||||
}
|
||||
const records: FoldGeneration['records'] = []
|
||||
for (const [kind, ids] of [
|
||||
['noun', delta.nouns] as const,
|
||||
['verb', delta.verbs] as const
|
||||
]) {
|
||||
for (const id of ids) {
|
||||
const record = await this.storage.readRawObject(
|
||||
`${GENERATIONS_PREFIX}/${gen}/prev/${id}.json`
|
||||
)
|
||||
if (record) records.push({ kind, id, record })
|
||||
}
|
||||
}
|
||||
foldInput.push({ generation: gen, timestamp: delta.timestamp, delta, records })
|
||||
}
|
||||
if (foldInput.length === 0) continue
|
||||
await segments.fold(foldInput)
|
||||
segmentsCreated++
|
||||
// Segment + manifest durable → the live copies retire.
|
||||
for (const g of foldInput) {
|
||||
await this.storage.removeRawPrefix(`${GENERATIONS_PREFIX}/${g.generation}`)
|
||||
}
|
||||
folded += foldInput.length
|
||||
}
|
||||
if (folded > 0) {
|
||||
prodLog.info(
|
||||
`[GenerationStore] repacked ${folded} cold generation(s) into ${segmentsCreated} segment(s) — history preserved, file count reduced`
|
||||
)
|
||||
}
|
||||
return { foldedGenerations: folded, segmentsCreated }
|
||||
})
|
||||
}
|
||||
|
||||
async compact(options?: CompactHistoryOptions): Promise<CompactHistoryResult> {
|
||||
return this.withMutex(async () => {
|
||||
const minPinned = this.minPinnedGeneration()
|
||||
|
|
@ -2421,11 +2220,6 @@ export class GenerationStore {
|
|||
const maxAge = options?.maxAge
|
||||
const maxBytes = options?.maxBytes
|
||||
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 =
|
||||
maxGenerations === undefined && maxAge === undefined && maxBytes === undefined
|
||||
|
||||
|
|
@ -2439,7 +2233,6 @@ export class GenerationStore {
|
|||
for (const gen of [...this.committedGensAsc()]) {
|
||||
// Pins are always exempt: never reclaim a generation a live pin needs.
|
||||
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)
|
||||
if (!noCaps) {
|
||||
const violatesCount = maxGenerations !== undefined && remainingCount > maxGenerations
|
||||
|
|
@ -2505,16 +2298,6 @@ export class GenerationStore {
|
|||
// Reclaimed generations leave the per-id chains stale → rebuild on next read.
|
||||
this.invalidateChains()
|
||||
this.horizonGen = Math.max(this.horizonGen, highestRemoved)
|
||||
// Packed-tier reclaim (D3): a packed generation's bytes live in a
|
||||
// sealed segment — removeRawPrefix above was a no-op for it. Drop
|
||||
// WHOLE segments now fully below the horizon; a partially-reclaimed
|
||||
// segment keeps its bytes until the boundary passes it (the frozen
|
||||
// partial-segments-wait rule; logical reclamation above still holds —
|
||||
// the generations left committedRanges and asOf below the horizon
|
||||
// throws regardless).
|
||||
if (this.segments) {
|
||||
await this.segments.dropSegmentsBelow(this.horizonGen + 1)
|
||||
}
|
||||
const manifest: GenerationManifest = {
|
||||
version: 1,
|
||||
generation: this.committed,
|
||||
|
|
|
|||
|
|
@ -176,15 +176,6 @@ export interface CompactHistoryOptions {
|
|||
* of each surviving generation's serialized record set (`GenerationDelta.bytes`).
|
||||
*/
|
||||
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
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -213,7 +213,6 @@ export type {
|
|||
CommitFact,
|
||||
FactOp,
|
||||
FactScanBatch,
|
||||
SCANFACTS_FIRST_BATCH_MS,
|
||||
FactScanHandle
|
||||
} from './db/factLog.js'
|
||||
// The generalized family stamp — which source generation a projection
|
||||
|
|
|
|||
|
|
@ -96,14 +96,6 @@ export class FileSystemStorage extends BaseStorage {
|
|||
private static readonly WRITER_STALE_THRESHOLD_MS = 60_000
|
||||
private writerLockHeartbeat?: NodeJS.Timeout
|
||||
private writerLockInfo?: WriterLockInfo
|
||||
/**
|
||||
* The currently-executing heartbeat refresh, if any. `releaseWriterLock()`
|
||||
* awaits it before unlinking: clearInterval() stops FUTURE ticks but not a
|
||||
* tick already in flight, and a straggler landing after the unlink would
|
||||
* RE-CREATE the lock file — a phantom lock blocking the next writer until
|
||||
* the stale TTL expires (the pool-eviction reopen case).
|
||||
*/
|
||||
private writerHeartbeatInFlight?: Promise<void>
|
||||
|
||||
// Flush-request RPC state. The writer polls `locks/_flush_requests/` for
|
||||
// new `.req` files and emits `.ack` files in `locks/_flush_responses/` after
|
||||
|
|
@ -1888,18 +1880,8 @@ export class FileSystemStorage extends BaseStorage {
|
|||
// Heartbeat — rewrite lastHeartbeat every WRITER_HEARTBEAT_MS so other
|
||||
// processes can tell a live writer from one that crashed without releasing.
|
||||
this.writerLockHeartbeat = setInterval(() => {
|
||||
const tick = this.refreshWriterLockHeartbeat().catch((err) => {
|
||||
// ENOENT = the lock (or its directory) vanished mid-refresh — the
|
||||
// store was released or removed under us; the next acquire recreates
|
||||
// it. Benign by construction; anything else stays loud.
|
||||
if ((err as NodeJS.ErrnoException)?.code !== 'ENOENT') {
|
||||
console.warn('[brainy] Failed to refresh writer lock heartbeat:', err)
|
||||
}
|
||||
})
|
||||
this.writerHeartbeatInFlight = tick.finally(() => {
|
||||
if (this.writerHeartbeatInFlight === tick) {
|
||||
this.writerHeartbeatInFlight = undefined
|
||||
}
|
||||
this.refreshWriterLockHeartbeat().catch((err) => {
|
||||
console.warn('[brainy] Failed to refresh writer lock heartbeat:', err)
|
||||
})
|
||||
}, FileSystemStorage.WRITER_HEARTBEAT_MS)
|
||||
if (typeof this.writerLockHeartbeat.unref === 'function') {
|
||||
|
|
@ -1931,15 +1913,6 @@ export class FileSystemStorage extends BaseStorage {
|
|||
clearInterval(this.writerLockHeartbeat)
|
||||
this.writerLockHeartbeat = undefined
|
||||
}
|
||||
// Drain an in-flight heartbeat tick BEFORE unlinking: clearInterval stops
|
||||
// future ticks only, and a straggler write landing after the unlink would
|
||||
// re-create the lock as a phantom (blocking the next writer until the
|
||||
// stale TTL). After the drain, any refresh is either fully landed (we
|
||||
// unlink its output below) or not started (it sees writerLockInfo
|
||||
// undefined and returns).
|
||||
if (this.writerHeartbeatInFlight) {
|
||||
await this.writerHeartbeatInFlight
|
||||
}
|
||||
if (!this.writerLockInfo) {
|
||||
return
|
||||
}
|
||||
|
|
|
|||
|
|
@ -133,11 +133,6 @@ export class MemoryStorage extends BaseStorage {
|
|||
*/
|
||||
protected async deleteObjectFromPath(path: string): Promise<void> {
|
||||
this.objectStore.delete(path)
|
||||
// Filesystem parity: on disk, objects and raw BYTE files are both just
|
||||
// files — unlink removes whichever exists. Without this, deleteRawObject
|
||||
// on a raw-bytes path (fact-log/generation segments) silently no-ops on
|
||||
// memory storage: the delete "succeeds" and the bytes remain.
|
||||
this.rawBytesStore.delete(path)
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -1737,10 +1737,7 @@ export interface BrainyConfig {
|
|||
* Under Model-B EVERY write (`transact()` AND single-op `add`/`update`/
|
||||
* `remove`/`relate`) produces an immutable generation record-set serving
|
||||
* historical reads (`asOf()`, pinned `Db` values). Without compaction those
|
||||
* 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.
|
||||
* accumulate, so Brainy **auto-compacts on every `flush()` and `close()`**.
|
||||
* Live `Db` pins are ALWAYS exempt from reclamation, in every mode.
|
||||
*
|
||||
* Modes:
|
||||
|
|
@ -1757,7 +1754,7 @@ export interface BrainyConfig {
|
|||
* the oldest unpinned generations while ANY supplied cap is exceeded
|
||||
* (predictable ops). `maxAge` in ms; `maxBytes` total history bytes.
|
||||
*
|
||||
* `autoCompact: false` disables the automatic close() compaction (manage
|
||||
* `autoCompact: false` disables the automatic flush/close compaction (manage
|
||||
* manually via `brain.compactHistory()`). `budgetBytes` is the settable
|
||||
* adaptive byte budget a coordinator drives (also via `brain.setRetentionBudget()`).
|
||||
* Long-term archives belong in `db.persist(path)` snapshots, which compaction
|
||||
|
|
@ -1775,7 +1772,7 @@ export interface BrainyConfig {
|
|||
maxBytes?: number
|
||||
/** Adaptive byte budget for this brain, driven by a coordinator (e.g. cor). */
|
||||
budgetBytes?: number
|
||||
/** Run compaction automatically at close() (default: true; 8.9.0 — flush() never compacts). */
|
||||
/** Run compaction automatically on flush()/close() (default: true). */
|
||||
autoCompact?: boolean
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1867,26 +1867,9 @@ export class MetadataIndexManager implements MetadataIndexProvider {
|
|||
// not once per AND-clause inside it.
|
||||
const unindexedFields: string[] = []
|
||||
|
||||
for (const [rawField, condition] of Object.entries(filter)) {
|
||||
for (const [field, condition] of Object.entries(filter)) {
|
||||
// Skip logical operators
|
||||
if (rawField === 'allOf' || rawField === 'anyOf' || rawField === 'not') continue
|
||||
|
||||
// Metadata is FLATTENED at index time (metadata.entry.title indexes as
|
||||
// entry.title), so a `metadata.`-prefixed where key is almost always
|
||||
// the caller spelling the STORAGE shape rather than the index shape.
|
||||
// Accept both spellings: when the key as spelled is unindexed but its
|
||||
// stripped spelling is, query the stripped one. A literal nested
|
||||
// custom key named `metadata` still wins when indexed as spelled
|
||||
// (checked first), so that rare shape keeps working.
|
||||
let field = rawField
|
||||
if (
|
||||
rawField.startsWith('metadata.') &&
|
||||
this.columnStore &&
|
||||
!this.columnStore.hasField(rawField) &&
|
||||
this.columnStore.hasField(rawField.slice('metadata.'.length))
|
||||
) {
|
||||
field = rawField.slice('metadata.'.length)
|
||||
}
|
||||
if (field === 'allOf' || field === 'anyOf' || field === 'not') continue
|
||||
|
||||
let fieldResults: string[] = []
|
||||
|
||||
|
|
|
|||
|
|
@ -1,169 +0,0 @@
|
|||
/**
|
||||
* @module tests/integration/aggregate-reserved-fields
|
||||
* @description One field-resolution law across the whole aggregation + query
|
||||
* surface (SELF-AGGREGATE-DELETE-DRIFT). Laws:
|
||||
* (1) aggregates grouped by a RESERVED field (subtype) decrement on delete —
|
||||
* the delete-side entity view carries every reserved field, so the
|
||||
* decrement finds its group (counts must never drift from ground truth);
|
||||
* (2) same for update: moving an entity between reserved-field groups
|
||||
* decrements the old group and increments the new one (no double-count);
|
||||
* (3) aggregation source.where on a reserved field (subtype) FILTERS instead
|
||||
* of silently matching nothing;
|
||||
* (4) removeMany refuses empty/invalid selectors loudly (bare array, empty
|
||||
* object, ids: []) instead of resolving as a silent no-op;
|
||||
* (5) find() accepts both where spellings: flattened (entry.title) and
|
||||
* storage-shaped (metadata.entry.title) resolve to the same rows.
|
||||
*/
|
||||
import { describe, it, expect, beforeEach, afterEach } from 'vitest'
|
||||
import { Brainy } from '../../src/brainy.js'
|
||||
import { NounType } from '../../src/types/graphTypes.js'
|
||||
|
||||
const stubEmbedding = async (text: string): Promise<number[]> => {
|
||||
const hash = text.split('').reduce((acc, char) => acc + char.charCodeAt(0), 0)
|
||||
return new Array(384).fill(0).map((_, i) => Math.sin(hash + i))
|
||||
}
|
||||
|
||||
describe('aggregation + query field-resolution law', () => {
|
||||
let brain: Brainy
|
||||
|
||||
beforeEach(async () => {
|
||||
brain = new Brainy({
|
||||
requireSubtype: false,
|
||||
storage: { type: 'memory' as const },
|
||||
embeddingFunction: stubEmbedding
|
||||
})
|
||||
await brain.init()
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
await brain.close()
|
||||
})
|
||||
|
||||
it('reserved-field groupBy decrements on delete (the drift bug)', async () => {
|
||||
brain.defineAggregate({
|
||||
name: 'by_subtype',
|
||||
source: { type: NounType.Document },
|
||||
groupBy: ['subtype'],
|
||||
metrics: { count: { op: 'count' } }
|
||||
})
|
||||
|
||||
const ids: string[] = []
|
||||
for (let i = 0; i < 5; i++) {
|
||||
ids.push(
|
||||
await brain.add({
|
||||
data: `doc-${i}`,
|
||||
type: NounType.Document,
|
||||
subtype: 'note',
|
||||
metadata: { team: 'alpha' }
|
||||
})
|
||||
)
|
||||
}
|
||||
let groups = await brain.queryAggregate('by_subtype')
|
||||
expect(groups).toHaveLength(1)
|
||||
expect(groups[0].groupKey).toEqual({ subtype: 'note' })
|
||||
expect(groups[0].metrics.count).toBe(5)
|
||||
|
||||
await brain.remove(ids[0])
|
||||
await brain.flush()
|
||||
|
||||
groups = await brain.queryAggregate('by_subtype')
|
||||
expect(groups[0].metrics.count).toBe(4)
|
||||
const live = await brain.find({ type: NounType.Document, limit: 100 })
|
||||
expect(groups[0].metrics.count).toBe(live.length)
|
||||
})
|
||||
|
||||
it('reserved-field groupBy moves between groups on update (no double-count)', async () => {
|
||||
brain.defineAggregate({
|
||||
name: 'by_subtype',
|
||||
source: { type: NounType.Document },
|
||||
groupBy: ['subtype'],
|
||||
metrics: { count: { op: 'count' } }
|
||||
})
|
||||
const id = await brain.add({
|
||||
data: 'doc-move',
|
||||
type: NounType.Document,
|
||||
subtype: 'draft'
|
||||
})
|
||||
await brain.update({ id, subtype: 'published' })
|
||||
|
||||
const groups = await brain.queryAggregate('by_subtype')
|
||||
const byKey = Object.fromEntries(
|
||||
groups.map((g) => [String(g.groupKey.subtype), g.metrics.count])
|
||||
)
|
||||
expect(byKey['published']).toBe(1)
|
||||
// The old group must be gone or zero — never still counting the entity.
|
||||
expect(byKey['draft'] ?? 0).toBe(0)
|
||||
})
|
||||
|
||||
it('source.where on a reserved field filters instead of matching nothing', async () => {
|
||||
brain.defineAggregate({
|
||||
name: 'notes_only',
|
||||
source: { type: NounType.Document, where: { subtype: 'note' } },
|
||||
groupBy: ['team'],
|
||||
metrics: { count: { op: 'count' } }
|
||||
})
|
||||
await brain.add({
|
||||
data: 'n1',
|
||||
type: NounType.Document,
|
||||
subtype: 'note',
|
||||
metadata: { team: 'alpha' }
|
||||
})
|
||||
await brain.add({
|
||||
data: 'd1',
|
||||
type: NounType.Document,
|
||||
subtype: 'draft',
|
||||
metadata: { team: 'alpha' }
|
||||
})
|
||||
|
||||
const groups = await brain.queryAggregate('notes_only')
|
||||
expect(groups).toHaveLength(1)
|
||||
expect(groups[0].metrics.count).toBe(1) // the note, never the draft
|
||||
})
|
||||
|
||||
it('removeMany refuses empty/invalid selectors loudly', async () => {
|
||||
const id = await brain.add({ data: 'keep-me', type: NounType.Document })
|
||||
|
||||
// Bare array passed positionally — the classic silent no-op.
|
||||
await expect(
|
||||
brain.removeMany([id] as unknown as Parameters<typeof brain.removeMany>[0])
|
||||
).rejects.toThrow(/bare array/)
|
||||
// Empty selector object.
|
||||
await expect(
|
||||
brain.removeMany({} as Parameters<typeof brain.removeMany>[0])
|
||||
).rejects.toThrow(/requires a selector/)
|
||||
// Explicit empty id list.
|
||||
await expect(brain.removeMany({ ids: [] })).rejects.toThrow(/ids: \[\]/)
|
||||
|
||||
// Nothing was deleted by any of the refused calls.
|
||||
expect(await brain.get(id)).toBeTruthy()
|
||||
})
|
||||
|
||||
it('find() accepts both flattened and metadata.-prefixed where spellings', async () => {
|
||||
await brain.add({
|
||||
data: 'nested-doc',
|
||||
type: NounType.Document,
|
||||
metadata: { entry: { title: 'T1' }, classifier: { contextHints: { vfsPath: '/n/a.md' } } }
|
||||
})
|
||||
await brain.flush()
|
||||
|
||||
const flat = await brain.find({
|
||||
type: NounType.Document,
|
||||
where: { 'entry.title': 'T1' },
|
||||
limit: 10
|
||||
})
|
||||
const prefixed = await brain.find({
|
||||
type: NounType.Document,
|
||||
where: { 'metadata.entry.title': 'T1' },
|
||||
limit: 10
|
||||
})
|
||||
const deepPrefixed = await brain.find({
|
||||
type: NounType.Document,
|
||||
where: { 'metadata.classifier.contextHints.vfsPath': '/n/a.md' },
|
||||
limit: 10
|
||||
})
|
||||
expect(flat).toHaveLength(1)
|
||||
expect(prefixed).toHaveLength(1)
|
||||
expect(prefixed[0].id).toBe(flat[0].id)
|
||||
expect(deepPrefixed).toHaveLength(1)
|
||||
})
|
||||
})
|
||||
|
|
@ -1280,32 +1280,24 @@ describe('8.0 Db API — generational MVCC', () => {
|
|||
await expect(reopened.asOf(1)).rejects.toBeInstanceOf(GenerationCompactedError)
|
||||
})
|
||||
|
||||
it('Model-B retention — flush() NEVER compacts (8.9.0); adaptive reclaim runs at close()', async () => {
|
||||
// Default brain → ADAPTIVE retention with a driven byte budget far below
|
||||
// the accumulated history (~13 generations of full-vector before-images).
|
||||
// The 8.9.0 law: flush() is durability-only — it must not reclaim even
|
||||
// when the budget is exceeded (reclaim-on-flush blocked production writes
|
||||
// for 25-191s). Maintenance runs at close(), time-bounded.
|
||||
const { brain, dir } = await openFsBrain()
|
||||
it('Model-B retention — setRetentionBudget drives adaptive reclaim on flush; live data intact', async () => {
|
||||
// Default brain → ADAPTIVE retention. A coordinator (e.g. cor's ResourceManager)
|
||||
// pushes a byte budget via setRetentionBudget(); auto-compaction on flush() reclaims
|
||||
// oldest history down toward it. Each update's before-image carries the full prior
|
||||
// 384-dim vector (~KBs), so ~13 generations far exceed a few-KB budget.
|
||||
const { brain } = await openFsBrain()
|
||||
const a = uid('ret-budget')
|
||||
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 } })
|
||||
|
||||
brain.setRetentionBudget(6000) // ~6 KB — well below the accumulated history
|
||||
await brain.flush()
|
||||
await brain.flush() // group-commit + adaptive auto-compaction under the budget
|
||||
|
||||
// flush() paid durability only: nothing reclaimed, all history readable.
|
||||
expect(generationStoreOf(brain).horizon()).toBe(0)
|
||||
const probe = await brain.asOf(1) // readable proves nothing was reclaimed…
|
||||
await probe.release() // …and MUST be released: a held pin would (correctly)
|
||||
// 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)
|
||||
// History was reclaimed (the horizon advanced past the oldest generations)…
|
||||
expect(generationStoreOf(brain).horizon()).toBeGreaterThan(0)
|
||||
await expect(brain.asOf(1)).rejects.toBeInstanceOf(GenerationCompactedError)
|
||||
// …but the budget reclaims HISTORY only — the live record is untouched.
|
||||
expect((await brain.get(a))?.metadata?.v).toBe(12)
|
||||
})
|
||||
|
||||
// ==========================================================================
|
||||
|
|
|
|||
|
|
@ -1,186 +0,0 @@
|
|||
/**
|
||||
* @module tests/integration/history-repacking
|
||||
* @description The D1+D3 two-tier history lifecycle end-to-end on a real
|
||||
* brain. Laws: (1) repacking is RE-REPRESENTATION — after folding, every
|
||||
* asOf() read below the fold boundary answers exactly as before, across a
|
||||
* cold reopen; (2) folded per-generation directories are physically gone
|
||||
* (the file-count cure is real, not cosmetic); (3) repack + reclaim compose:
|
||||
* bounded retention after repacking drops whole segments and asOf below the
|
||||
* horizon throws GenerationCompactedError; (4) repackHistory is explicit
|
||||
* API and time-bounded (spent budget = consistent no-op).
|
||||
*
|
||||
* Uses a tiny REPACK_LIVE_WINDOW override so a small history has a cold
|
||||
* tier at all (the production window is 1024).
|
||||
*/
|
||||
import { describe, it, expect, afterEach } from 'vitest'
|
||||
import * as fs from 'node:fs'
|
||||
import * as path from 'node:path'
|
||||
import * as os from 'node:os'
|
||||
import { Brainy } from '../../src/brainy.js'
|
||||
import { NounType } from '../../src/types/graphTypes.js'
|
||||
import { GenerationStore } from '../../src/db/generationStore.js'
|
||||
import { GenerationCompactedError } from '../../src/db/errors.js'
|
||||
import { SEGMENTS_PREFIX } from '../../src/db/generationSegments.js'
|
||||
|
||||
const stub = async (text: string): Promise<number[]> => {
|
||||
const h = text.split('').reduce((a, c) => a + c.charCodeAt(0), 0)
|
||||
return new Array(384).fill(0).map((_, i) => Math.sin(h + i))
|
||||
}
|
||||
|
||||
const openBrain = async (dir: string): Promise<Brainy> => {
|
||||
const brain = new Brainy({
|
||||
requireSubtype: false,
|
||||
storage: { type: 'filesystem', path: dir },
|
||||
embeddingFunction: stub
|
||||
})
|
||||
await brain.init()
|
||||
return brain
|
||||
}
|
||||
|
||||
describe('history repacking — the two-tier lifecycle', () => {
|
||||
const dirs: string[] = []
|
||||
const tempDir = (): string => {
|
||||
const d = fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-repack-'))
|
||||
dirs.push(d)
|
||||
return d
|
||||
}
|
||||
const originalWindow = GenerationStore.REPACK_LIVE_WINDOW
|
||||
|
||||
afterEach(() => {
|
||||
;(GenerationStore as any).REPACK_LIVE_WINDOW = originalWindow
|
||||
for (const d of dirs.splice(0)) {
|
||||
try {
|
||||
fs.rmSync(d, { recursive: true, force: true })
|
||||
} catch {
|
||||
/* best effort */
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
it('repack preserves every historical read across cold reopen; folded dirs are gone', async () => {
|
||||
;(GenerationStore as any).REPACK_LIVE_WINDOW = 3
|
||||
const dir = tempDir()
|
||||
const brain = await openBrain(dir)
|
||||
|
||||
const id = await brain.add({
|
||||
data: 'versioned-entity',
|
||||
type: NounType.Document,
|
||||
metadata: { v: 0 }
|
||||
})
|
||||
for (let v = 1; v <= 10; v++) await brain.update({ id, metadata: { v } })
|
||||
await brain.flush()
|
||||
|
||||
// Ground truth BEFORE repacking: capture asOf views for early generations.
|
||||
const before: Record<number, number> = {}
|
||||
for (const g of [2, 4, 6]) {
|
||||
const db = await brain.asOf(g)
|
||||
before[g] = (await db.get(id))?.metadata?.v as number
|
||||
await db.release()
|
||||
}
|
||||
|
||||
const result = await brain.repackHistory()
|
||||
expect(result.foldedGenerations).toBeGreaterThan(0)
|
||||
expect(result.segmentsCreated).toBeGreaterThan(0)
|
||||
|
||||
// The folded per-generation directories are PHYSICALLY gone…
|
||||
const genDirs = fs
|
||||
.readdirSync(path.join(dir, '_generations'), { withFileTypes: true })
|
||||
.filter((e) => e.isDirectory() && /^\d+$/.test(e.name)).length
|
||||
expect(genDirs).toBeLessThanOrEqual(4) // live window (3) + at most the newest
|
||||
// …and the segment tier exists (the filesystem adapter stores objects
|
||||
// gzipped, so the manifest may live at either spelling).
|
||||
const segDir = path.join(dir, SEGMENTS_PREFIX)
|
||||
expect(
|
||||
fs.existsSync(path.join(segDir, 'manifest.json')) ||
|
||||
fs.existsSync(path.join(segDir, 'manifest.json.gz'))
|
||||
).toBe(true)
|
||||
expect(fs.readdirSync(segDir).some((f) => f.endsWith('.bgs'))).toBe(true)
|
||||
|
||||
// Same asOf answers from the packed tier, same process…
|
||||
for (const g of [2, 4, 6]) {
|
||||
const db = await brain.asOf(g)
|
||||
expect((await db.get(id))?.metadata?.v).toBe(before[g])
|
||||
await db.release()
|
||||
}
|
||||
await brain.close()
|
||||
|
||||
// …and across a COLD REOPEN (manifest discovery, no live dirs to list).
|
||||
const reopened = await openBrain(dir)
|
||||
for (const g of [2, 4, 6]) {
|
||||
const db = await reopened.asOf(g)
|
||||
expect((await db.get(id))?.metadata?.v).toBe(before[g])
|
||||
await db.release()
|
||||
}
|
||||
expect((await reopened.get(id))?.metadata?.v).toBe(10) // live state untouched
|
||||
await reopened.close()
|
||||
})
|
||||
|
||||
it('repack + bounded reclaim compose: whole segments drop, horizon is loud', async () => {
|
||||
;(GenerationStore as any).REPACK_LIVE_WINDOW = 2
|
||||
const dir = tempDir()
|
||||
const brain = await openBrain(dir)
|
||||
const id = await brain.add({ data: 'reclaim-probe', type: NounType.Document, metadata: { v: 0 } })
|
||||
for (let v = 1; v <= 8; v++) await brain.update({ id, metadata: { v } })
|
||||
await brain.flush()
|
||||
await brain.repackHistory()
|
||||
|
||||
// Reclaim down to the 3 newest generations — packed segments below the
|
||||
// horizon drop whole; asOf below throws loudly.
|
||||
const res = await brain.compactHistory({ maxGenerations: 3 })
|
||||
expect(res.removedGenerations).toBeGreaterThan(0)
|
||||
await expect(brain.asOf(1)).rejects.toBeInstanceOf(GenerationCompactedError)
|
||||
expect((await brain.get(id))?.metadata?.v).toBe(8)
|
||||
await brain.close()
|
||||
})
|
||||
|
||||
it('generationDigest: reopen-stable, divergence-sensitive, loud below the horizon', async () => {
|
||||
;(GenerationStore as any).REPACK_LIVE_WINDOW = 2
|
||||
const dir = tempDir()
|
||||
const brain = await openBrain(dir)
|
||||
const id = await brain.add({ data: 'digest-probe', type: NounType.Document, metadata: { v: 0 } })
|
||||
for (let v = 1; v <= 6; v++) await brain.update({ id, metadata: { v } })
|
||||
await brain.flush()
|
||||
await brain.repackHistory()
|
||||
|
||||
const gen = brain.generation()
|
||||
const atHead = await brain.generationDigest(gen)
|
||||
const atMid = await brain.generationDigest(3)
|
||||
expect(atHead).toMatch(/^[0-9a-f]{8}$/)
|
||||
expect(atMid).not.toBe(atHead) // more history ⇒ different digest
|
||||
await brain.close()
|
||||
|
||||
// Reopen-stable: same history, same digests (packed prefix stability).
|
||||
const reopened = await openBrain(dir)
|
||||
expect(await reopened.generationDigest(gen)).toBe(atHead)
|
||||
expect(await reopened.generationDigest(3)).toBe(atMid)
|
||||
|
||||
// New history diverges the head digest.
|
||||
await reopened.update({ id, metadata: { v: 7 } })
|
||||
await reopened.flush()
|
||||
expect(await reopened.generationDigest(reopened.generation())).not.toBe(atHead)
|
||||
|
||||
// Below the horizon: LOUD, never a silent pin of reclaimed history.
|
||||
await reopened.compactHistory({ maxGenerations: 2 })
|
||||
await expect(reopened.generationDigest(1)).rejects.toBeInstanceOf(GenerationCompactedError)
|
||||
await reopened.close()
|
||||
})
|
||||
|
||||
it('a spent time budget is a consistent no-op; the next pass resumes', async () => {
|
||||
;(GenerationStore as any).REPACK_LIVE_WINDOW = 2
|
||||
const dir = tempDir()
|
||||
const brain = await openBrain(dir)
|
||||
const id = await brain.add({ data: 'budget-probe', type: NounType.Document, metadata: { v: 0 } })
|
||||
for (let v = 1; v <= 6; v++) await brain.update({ id, metadata: { v } })
|
||||
await brain.flush()
|
||||
|
||||
const bounded = await brain.repackHistory({ timeBudgetMs: 0 })
|
||||
expect(bounded).toEqual({ foldedGenerations: 0, segmentsCreated: 0 })
|
||||
|
||||
const resumed = await brain.repackHistory()
|
||||
expect(resumed.foldedGenerations).toBeGreaterThan(0)
|
||||
const db = await brain.asOf(3)
|
||||
expect((await db.get(id))?.metadata?.v).toBeDefined()
|
||||
await db.release()
|
||||
await brain.close()
|
||||
})
|
||||
})
|
||||
|
|
@ -153,46 +153,6 @@ describe('Multi-process safety + read-only mode', () => {
|
|||
expect(err.lockInfo?.pid).toBe(otherPid)
|
||||
})
|
||||
|
||||
it('release drains an in-flight heartbeat — no phantom lock re-created after unlink', async () => {
|
||||
// The race (8.9.0): clearInterval stops FUTURE heartbeat ticks, but a
|
||||
// tick already in flight could land its lock rewrite AFTER release's
|
||||
// unlink — re-creating the lock as a phantom that blocks the next
|
||||
// writer until the stale TTL. Simulate the in-flight tick explicitly
|
||||
// and prove release waits for it.
|
||||
writer = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
|
||||
await writer.init()
|
||||
const storage: any = (writer as any).storage
|
||||
|
||||
// An in-flight refresh that is ALREADY PAST its ownership guards
|
||||
// (captured the lock info before release ran) and lands its atomic
|
||||
// rewrite slowly — the exact straggler shape; absent the drain it
|
||||
// writes after the unlink.
|
||||
const { join: joinPath } = await import('node:path')
|
||||
const capturedInfo = { ...storage.writerLockInfo }
|
||||
const lockPath = joinPath(dir, 'locks', '_writer.lock')
|
||||
const slowTick = (async () => {
|
||||
await new Promise((r) => setTimeout(r, 100))
|
||||
await storage.writeFileAtomic(
|
||||
lockPath,
|
||||
JSON.stringify({ ...capturedInfo, lastHeartbeat: new Date().toISOString() })
|
||||
)
|
||||
})()
|
||||
storage.writerHeartbeatInFlight = slowTick.catch(() => {})
|
||||
|
||||
await writer.close() // → releaseWriterLock must drain slowTick first
|
||||
await slowTick.catch(() => {}) // both paths fully settled either way
|
||||
writer = null
|
||||
|
||||
const { existsSync } = await import('node:fs')
|
||||
const { join } = await import('node:path')
|
||||
expect(existsSync(join(dir, 'locks', '_writer.lock'))).toBe(false)
|
||||
|
||||
// And the directory is immediately claimable — no stale-TTL wait.
|
||||
const next = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
|
||||
await expect(next.init()).resolves.toBeUndefined()
|
||||
await next.close()
|
||||
})
|
||||
|
||||
it('allows a second in-process writer with a warning (same PID)', async () => {
|
||||
// Two Brainy instances in the same Node process: not the dangerous
|
||||
// cross-process case. Should succeed (with a console warning).
|
||||
|
|
|
|||
|
|
@ -205,10 +205,10 @@ describe('Metadata index cleanup after remove / removeMany', () => {
|
|||
}
|
||||
})
|
||||
|
||||
it('refuses an empty ids array loudly (a silent no-op is not "graceful")', async () => {
|
||||
// 8.8.2: an empty selector used to resolve successfully having deleted
|
||||
// NOTHING — the caller believed the delete happened. Now it throws.
|
||||
await expect(brain.removeMany({ ids: [] })).rejects.toThrow(/ids: \[\]/)
|
||||
it('handles empty ids array gracefully', async () => {
|
||||
const result = await brain.removeMany({ ids: [] })
|
||||
expect(result.successful).toHaveLength(0)
|
||||
expect(result.failed).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('handles large batch (> 1 chunk) without leaving stale index entries', async () => {
|
||||
|
|
|
|||
|
|
@ -533,10 +533,8 @@ describe('Brainy Batch Operations', () => {
|
|||
expect(result.successful).toHaveLength(0)
|
||||
|
||||
await brain.updateMany({ items: [] })
|
||||
// removeMany is the exception (8.8.2): an empty id list is a refused
|
||||
// selector, not an empty batch — deleting "nothing" silently was the
|
||||
// bug class (a positional/bare-array call looked identical).
|
||||
await expect(brain.removeMany({ ids: [] })).rejects.toThrow(/ids: \[\]/)
|
||||
await brain.removeMany({ ids: [] })
|
||||
// Should not throw
|
||||
})
|
||||
|
||||
it('should validate batch size limits', async () => {
|
||||
|
|
|
|||
|
|
@ -186,52 +186,4 @@ describe('fact log — round-trip, framing, reconcile, rotation, scan', () => {
|
|||
await log.sync()
|
||||
expect(log.segmentPaths()).toEqual([]) // only a tail exists — nothing sealed
|
||||
})
|
||||
|
||||
describe('scanFacts liveness contract (Stage-2 D1)', () => {
|
||||
it('a wedged store fails LOUDLY within the first-batch bound — never a silent hang', async () => {
|
||||
// Force a sealed segment (tiny rotateBytes) so the scan must READ from
|
||||
// storage, then wedge that read: the exact production shape (a
|
||||
// backlogged brain whose segment read never returned).
|
||||
const mem: any = new MemoryStorage()
|
||||
await mem.init()
|
||||
const wedgeable = new FactLog(mem, { rotateBytes: 1 })
|
||||
await wedgeable.open(0)
|
||||
await wedgeable.append(fact(1))
|
||||
await wedgeable.append(fact(2)) // second append rotates → seg 1 sealed
|
||||
await wedgeable.sync()
|
||||
|
||||
const realRead = mem.readRawBytes.bind(mem)
|
||||
mem.readRawBytes = (p: string) =>
|
||||
p.includes('facts/seg-') ? new Promise(() => {}) : realRead(p) // hangs forever
|
||||
|
||||
const scan = wedgeable.scanFacts({ firstBatchTimeoutMs: 200 })
|
||||
const started = Date.now()
|
||||
await expect(scan.batches().next()).rejects.toThrow(/no first batch within 200ms/)
|
||||
expect(Date.now() - started).toBeLessThan(5_000) // bound held, not a hang
|
||||
})
|
||||
|
||||
it('a healthy scan is unaffected — first batch well inside the bound, all facts delivered', async () => {
|
||||
for (let g = 1; g <= 5; g++) await log.append(fact(g))
|
||||
await log.sync()
|
||||
const scan = log.scanFacts({ batchSize: 2 })
|
||||
const all: CommitFact[] = []
|
||||
for await (const b of scan.batches()) all.push(...b.facts)
|
||||
expect(all.map((f) => f.generation)).toEqual([1, 2, 3, 4, 5])
|
||||
expect(scan.summary().factsYielded).toBe(5)
|
||||
})
|
||||
|
||||
it('consumer think-time between pulls never counts against the producer', async () => {
|
||||
for (let g = 1; g <= 4; g++) await log.append(fact(g))
|
||||
await log.sync()
|
||||
// Bound tighter than the consumer's pause: only the FIRST pull is
|
||||
// raced, so a slow consumer after batch 1 must not trip the deadline.
|
||||
const gen = log.scanFacts({ batchSize: 2, firstBatchTimeoutMs: 150 }).batches()
|
||||
const first = await gen.next()
|
||||
expect(first.done).toBe(false)
|
||||
await new Promise((r) => setTimeout(r, 400)) // dawdle past the bound
|
||||
const second = await gen.next()
|
||||
expect(second.done).toBe(false)
|
||||
expect((await gen.next()).done).toBe(true)
|
||||
})
|
||||
})
|
||||
})
|
||||
|
|
|
|||
|
|
@ -1,150 +0,0 @@
|
|||
/**
|
||||
* @module tests/unit/db/generation-segments
|
||||
* @description The generation-segment store (Stage-2 D1+D3 file format).
|
||||
* Laws: (1) fold → read round-trips deltas and records byte-faithfully via
|
||||
* sidecar point-reads; (2) the manifest is the ONLY discovery path — reopen
|
||||
* reads one file, never a listing; (3) a lost/corrupt sidecar rebuilds from
|
||||
* its segment loudly, a damaged SEGMENT fails loudly (never silent wrong
|
||||
* data); (4) D3 reclaim drops whole segments only and bumps compactedBelow;
|
||||
* (5) the packed digest is deterministic across reopen; (6) immutability —
|
||||
* fold refuses overlap with sealed ranges.
|
||||
*/
|
||||
import { describe, it, expect, beforeEach } from 'vitest'
|
||||
import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js'
|
||||
import {
|
||||
GenerationSegmentStore,
|
||||
SEGMENTS_PREFIX,
|
||||
type FoldGeneration
|
||||
} from '../../../src/db/generationSegments.js'
|
||||
|
||||
const UUID = (n: number): string => `00000000-0000-4000-8000-${String(n).padStart(12, '0')}`
|
||||
|
||||
const gen = (g: number, recordCount = 2): FoldGeneration => ({
|
||||
generation: g,
|
||||
timestamp: 1_700_000_000_000 + g,
|
||||
delta: { generation: g, nouns: [UUID(g)], verbs: [], bytes: 123 + g },
|
||||
records: Array.from({ length: recordCount }, (_, i) => ({
|
||||
kind: (i % 2 === 0 ? 'noun' : 'verb') as 'noun' | 'verb',
|
||||
id: UUID(g * 100 + i),
|
||||
record: { metadata: { noun: 'document', v: g }, vector: { v: [g, i] } }
|
||||
}))
|
||||
})
|
||||
|
||||
describe('db/GenerationSegmentStore — the D1+D3 packed tier', () => {
|
||||
let storage: MemoryStorage
|
||||
let store: GenerationSegmentStore
|
||||
|
||||
beforeEach(async () => {
|
||||
storage = new MemoryStorage()
|
||||
await storage.init()
|
||||
store = new GenerationSegmentStore(storage as any)
|
||||
await store.open()
|
||||
})
|
||||
|
||||
it('fold → read round-trips deltas and records via sidecar point-reads', async () => {
|
||||
const meta = await store.fold([gen(1), gen(2), gen(3)])
|
||||
expect(meta).toMatchObject({ firstGeneration: 1, lastGeneration: 3, frames: 3 })
|
||||
expect(meta.checksum).toBeGreaterThan(0)
|
||||
|
||||
expect(store.hasGeneration(2)).toBe(true)
|
||||
expect(store.hasGeneration(4)).toBe(false)
|
||||
|
||||
const d2 = await store.readDelta(2)
|
||||
expect(d2?.delta).toEqual({ generation: 2, nouns: [UUID(2)], verbs: [], bytes: 125 })
|
||||
expect(d2?.timestamp).toBe(1_700_000_000_002)
|
||||
|
||||
const records = await store.readRecords(3)
|
||||
expect(records).toHaveLength(2)
|
||||
expect(records![0]).toEqual({
|
||||
kind: 'noun',
|
||||
id: UUID(300),
|
||||
record: { metadata: { noun: 'document', v: 3 }, vector: { v: [3, 0] } }
|
||||
})
|
||||
// Point read by id, both kinds.
|
||||
expect(await store.readRecord(3, 'verb', UUID(301))).toEqual({
|
||||
metadata: { noun: 'document', v: 3 },
|
||||
vector: { v: [3, 1] }
|
||||
})
|
||||
expect(await store.readRecord(3, 'noun', UUID(999))).toBeNull()
|
||||
})
|
||||
|
||||
it('reopen discovers everything from the manifest alone — no listing', async () => {
|
||||
await store.fold([gen(1), gen(2)])
|
||||
await store.fold([gen(3), gen(4)])
|
||||
|
||||
const reopened = new GenerationSegmentStore(storage as any)
|
||||
await reopened.open()
|
||||
expect(reopened.segments()).toHaveLength(2)
|
||||
expect(reopened.hasGeneration(4)).toBe(true)
|
||||
expect((await reopened.readDelta(1))?.timestamp).toBe(1_700_000_000_001)
|
||||
})
|
||||
|
||||
it('a lost sidecar rebuilds from its segment; a damaged segment fails LOUDLY', async () => {
|
||||
const meta = await store.fold([gen(1), gen(2)])
|
||||
const idxPath = `${SEGMENTS_PREFIX}/seg-${String(1).padStart(20, '0')}.idx`
|
||||
await storage.deleteRawObject(idxPath)
|
||||
|
||||
const reopened = new GenerationSegmentStore(storage as any)
|
||||
await reopened.open()
|
||||
// Rebuild path: still serves correct data.
|
||||
expect((await reopened.readRecords(2))!).toHaveLength(2)
|
||||
|
||||
// Now damage the SEGMENT itself: flip a payload byte → CRC mismatch, loud.
|
||||
const segPath = `${SEGMENTS_PREFIX}/${meta.file}`
|
||||
const bytes = (await storage.readRawBytes(segPath))!
|
||||
bytes[bytes.length - 3] ^= 0xff
|
||||
await storage.writeRawBytes(segPath, bytes)
|
||||
const damaged = new GenerationSegmentStore(storage as any)
|
||||
await damaged.open()
|
||||
;(damaged as any).sidecars.clear()
|
||||
await storage.deleteRawObject(idxPath) // force the sequential rebuild over damaged bytes
|
||||
await expect(damaged.readRecords(2)).rejects.toThrow(/CRC mismatch|damaged/)
|
||||
})
|
||||
|
||||
it('D3 reclaim drops whole segments only and bumps compactedBelow', async () => {
|
||||
await store.fold([gen(1), gen(2)])
|
||||
await store.fold([gen(3), gen(4)])
|
||||
await store.fold([gen(5), gen(6)])
|
||||
|
||||
// Horizon mid-segment-2 (below 4): only segment 1 is FULLY below → drops.
|
||||
const r1 = await store.dropSegmentsBelow(4)
|
||||
expect(r1).toEqual({ dropped: 1, compactedBelow: 3 })
|
||||
expect(store.hasGeneration(1)).toBe(false)
|
||||
expect(store.hasGeneration(3)).toBe(true) // partial segment survives whole
|
||||
|
||||
// Bytes actually gone.
|
||||
expect(await storage.readRawBytes(`${SEGMENTS_PREFIX}/seg-${String(1).padStart(20, '0')}.bgs`)).toBeNull()
|
||||
|
||||
// Horizon past everything: the rest drop; compactedBelow is durable.
|
||||
const r2 = await store.dropSegmentsBelow(7)
|
||||
expect(r2.dropped).toBe(2)
|
||||
const reopened = new GenerationSegmentStore(storage as any)
|
||||
await reopened.open()
|
||||
expect(reopened.compactedBelow()).toBe(7)
|
||||
expect(reopened.segments()).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('the packed digest is deterministic across reopen and changes with history', async () => {
|
||||
await store.fold([gen(1), gen(2), gen(3)])
|
||||
const atSeal = await store.digestThroughPacked(3)
|
||||
const midSegment = await store.digestThroughPacked(2)
|
||||
expect(atSeal).not.toBeNull()
|
||||
expect(midSegment).not.toBeNull()
|
||||
expect(midSegment).not.toBe(atSeal)
|
||||
|
||||
const reopened = new GenerationSegmentStore(storage as any)
|
||||
await reopened.open()
|
||||
expect(await reopened.digestThroughPacked(3)).toBe(atSeal)
|
||||
expect(await reopened.digestThroughPacked(2)).toBe(midSegment)
|
||||
|
||||
await reopened.fold([gen(4)])
|
||||
expect(await reopened.digestThroughPacked(4)).not.toBe(atSeal)
|
||||
})
|
||||
|
||||
it('sealed segments are immutable — fold refuses overlap, requires ascending input', async () => {
|
||||
await store.fold([gen(1), gen(2)])
|
||||
await expect(store.fold([gen(2), gen(3)])).rejects.toThrow(/overlaps the packed tier/)
|
||||
await expect(store.fold([gen(4), gen(4)])).rejects.toThrow(/strictly ascending/)
|
||||
await expect(store.fold([])).rejects.toThrow(/at least one generation/)
|
||||
})
|
||||
})
|
||||
|
|
@ -488,20 +488,6 @@ describe('db/GenerationStore', () => {
|
|||
expect(result.removedGenerations).toBe(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