feat(memory-b): port C3 centroid staleness and async rebuild scheduling

This commit is contained in:
2026-11-01 12:32:16 +08:00
parent d57c9dc798
commit d73d47c081
2 changed files with 546 additions and 0 deletions

View File

@ -1,3 +1,11 @@
function safeParseJson(value) {
try { return JSON.parse(value ?? '{}') } catch { return null }
}
function defaultCentroidSchedule(fn, delayMs) {
setTimeout(fn, delayMs)
}
/**
* SQLite storage for block memory: the three tables required by the MVP slice.
*
@ -299,6 +307,8 @@ export function blockId(text) {
export class BlockStore {
/** @type {DatabaseSync | null} */
#db = null
/** @type {Map<string, number>} */
#centroidRebuiltAt = new Map()
/** @type {string} */
#path
@ -531,6 +541,185 @@ export class BlockStore {
return 0.6
}
/** Relative change in the vectored-block count that marks a centroid stale.
* Ported from `centroidRebuildRatio` (internal/memory/centroid_store.go:55).
* The Go comment pins the rationale: below 10% the shift does not move the
* principal direction, and being more sensitive than 20% makes similarity
* drift with every write — a non-reproducible recall cannot serve regression. */
static get CENTROID_REBUILD_RATIO() {
return 0.2
}
/** Minimum interval between two scheduled rebuilds of the same space.
* Ported from `centroidRebuildWait` (internal/memory/centroid_store.go:324). */
static get CENTROID_REBUILD_WAIT_MS() {
return 10 * 60 * 1000
}
/** Delay before the scheduled rebuild runs, so the current read settles.
* Ported from `time.Sleep(2 * time.Second)` (internal/memory/centroid_store.go:338). */
static get CENTROID_REBUILD_DELAY_MS() {
return 2000
}
/**
* Number of blocks in `fingerprint`'s space that currently carry a vector.
*
* This is `countVectoredBlocksLocked` (internal/memory/centroid_store.go:178):
* the DENOMINATOR side of the staleness ratio. It is deliberately NOT called
* from `blockStates` (this file's pre-step hot path) — it only feeds
* {@link centroidStatus}.
*
* @param {string} fingerprint
* @returns {number}
*/
countVectored(fingerprint) {
return this.#handle()
.prepare(
`SELECT COUNT(*) AS n FROM memory_blocks
WHERE fingerprint = ? AND vector != '' AND vector != 'null'`,
)
.get(String(fingerprint ?? '')).n
}
/**
* Centroid staleness FACTS for one space (C3).
*
* Ports the two Go call sites that answer different questions:
*
* · recall path (internal/memory/centroid_store.go:179-204): a centroid
* built from zero samples is simply UNUSABLE — no staleness judgement.
* · status path (internal/memory/centroid_store.go:425-433): the same row
* IS reported stale when blocks now carry vectors.
*
* They are separate fields (`available` / `stale`) rather than one merged flag
* because merging them makes the report self-contradictory.
*
* ★ Why the rebuild is scheduled and never run inline (Go :196-204): Go holds
* `g.mu.RLock` for this whole function and `RebuildCentroid`/`SaveCentroid`
* need the write lock, so an inline rebuild self-deadlocks (their test hit a
* 180 s timeout). The JS equivalent is not a lock but a transaction: node:sqlite
* refuses a nested `BEGIN IMMEDIATE` ("cannot start a transaction within a
* transaction") and throws SQLITE_BUSY when a statement is still iterating. So
* this method only REGISTERS a rebuild and returns "no centering this round".
*
* ★ C3 gives facts only. Distinguishing WHICH reason disabled centering (no
* centroid / dimension mismatch / stale / isotropic) is C5 and is NOT covered
* here.
*
* @param {string} fingerprint
* @param {{queryDim?: number, now?: () => number,
* schedule?: (fn: () => void, delayMs: number) => void}} [options]
* `now`/`schedule` are injectable so the throttle and the delay are testable
* against the real branch logic instead of a stubbed global timer.
* @returns {{available: boolean, stale: boolean|null, dim: number|null,
* vectorCount: number|null, currentBlocks: number, delta: number|null,
* rebuildScheduled: boolean}}
* @throws when the stored centroid metadata is corrupt (`dim <= 0`,
* internal/memory/centroid_store.go:165-167)
*/
centroidStatus(fingerprint, options = {}) {
const { queryDim, now = Date.now, schedule = defaultCentroidSchedule } = options
const currentBlocks = this.countVectored(fingerprint)
const row = this.#handle()
.prepare('SELECT dim, vector_count, stats FROM graph_centroid WHERE fingerprint = ?')
.get(String(fingerprint ?? ''))
if (row === undefined) {
// :153-155 — no row at all is a normal state (Saved=false), not an error.
return {
available: false, stale: null, dim: null, vectorCount: null,
currentBlocks, delta: null, rebuildScheduled: false,
}
}
const dim = Number(row.dim)
if (dim <= 0) {
// :165-167 — corrupt metadata is an ERROR, not a silent "no centering".
throw new Error(`dsh-memory: centroid dim is ${dim} (corrupted?)`)
}
const vectorCount = Number(row.vector_count)
if (Number.isInteger(queryDim) && queryDim > 0 && dim !== queryDim) {
// :168-170 — wrong width for this query: unusable, not stale.
return {
available: false, stale: null, dim, vectorCount, currentBlocks,
delta: null, rebuildScheduled: false,
}
}
// :186-189 — (current − built) / built, sign discarded.
const raw = vectorCount > 0 ? (currentBlocks - vectorCount) / vectorCount : null
const delta = raw === null ? null : Math.abs(raw)
if (vectorCount <= 0) {
// :179-185 (recall) vs :431-432 (status): NOT available either way, but the
// status side still calls it stale when blocks exist now.
return {
available: false,
stale: currentBlocks > 0,
dim, vectorCount, currentBlocks, delta, rebuildScheduled: false,
}
}
// :190 — STRICTLY greater; exactly 20% is not stale.
if (delta > BlockStore.CENTROID_REBUILD_RATIO) {
const rebuildScheduled = this.#scheduleCentroidRebuild(fingerprint, { now, schedule })
return {
available: false, stale: true, dim, vectorCount, currentBlocks, delta, rebuildScheduled,
}
}
const stats = safeParseJson(row.stats)
if (stats?.anomalous === false || !this.loadCentroid(fingerprint)?.isAnomalous) {
// :205-208 — built as isotropic: centering is not enabled for this space.
return {
available: false, stale: false, dim, vectorCount, currentBlocks, delta, rebuildScheduled: false,
}
}
return {
available: true, stale: false, dim, vectorCount, currentBlocks, delta, rebuildScheduled: false,
}
}
/**
* Recall-facing projection of C3 facts. Go `LoadCentroid` returns
* `(nil,false,nil)` for every unusable center, including a zero-sample center
* (`internal/memory/centroid_store.go:179-185`); it does NOT expose the
* operator-only stale bit from `CentroidStatusOf:425-433`.
* @param {string} fingerprint
* @param {{queryDim?: number, now?: () => number,
* schedule?: (fn: () => void, delayMs: number) => void}} [options]
*/
centroidRecallState(fingerprint, options = {}) {
const status = this.centroidStatus(fingerprint, options)
return { ...status, available: status.available, stale: null }
}
/**
* Port of `scheduleCentroidRebuild` (internal/memory/centroid_store.go:327-347).
*
* @param {string} fingerprint
* @param {{now: () => number, schedule: (fn: () => void, delayMs: number) => void}} clock
* @returns {boolean} whether a rebuild was registered
*/
#scheduleCentroidRebuild(fingerprint, { now, schedule }) {
const last = this.#centroidRebuiltAt.get(fingerprint)
if (last !== undefined && now() - last < BlockStore.CENTROID_REBUILD_WAIT_MS) {
return false // :329-332 — inside the throttle window
}
this.#centroidRebuiltAt.set(fingerprint, now())
schedule(() => {
try {
this.rebuildCentroid(fingerprint)
} catch {
// :340-343 — a failed rebuild must not turn into a recall failure; the
// next status call sees the still-stale centroid and schedules again.
}
}, BlockStore.CENTROID_REBUILD_DELAY_MS)
return true
}
/**
* Rebuild the centroid for one vector space FROM ITS BLOCKS.
*

View File

@ -3527,3 +3527,360 @@ for (const [name, replacements, withCentroids, criterion] of [
})
})
}
// ── C3: centroid staleness, async rebuild scheduling, status facts ────────
//
// Ported from Go `internal/memory/centroid_store.go`:
// :55 centroidRebuildRatio = 0.20
// :178 curVectored = countVectoredBlocks()
// :186-189 delta = (curVectored - vecCount) / vecCount, sign discarded
// :190 delta > ratio (STRICTLY greater)
// :165-167 dim <= 0 => error (corrupted metadata)
// :179-185 recall path: vecCount <= 0 => unusable, NOT stale
// :425-433 status path: vecCount == 0 && currentBlocks > 0 => stale
// :327-347 schedule once per 10 min, run 2 s later, never inline
//
// ★ C3 gives FACTS only. Which reason disabled centering (none / dim mismatch /
// stale / isotropic) is C5 and is NOT covered here.
/** A store with `n` vectored blocks in space 'fp-c3'. */
function c3Store(n) {
const store = new BlockStore({ path: ':memory:' }).open()
for (let i = 0; i < n; i += 1) {
store.putBlock({ text: `C3 block ${i}`, vector: [1, 0, 0], fingerprint: 'fp-c3' })
}
return store
}
/** Save a centroid row claiming `vectorCount` samples. */
function c3Centroid(store, vectorCount, { dim = 3, anomalous = true } = {}) {
store.saveCentroid({
fingerprint: 'fp-c3',
dim,
vectorCount,
mean: Array(dim).fill(0),
isAnomalous: anomalous,
})
}
/** A manually-advanced clock plus a scheduler that records instead of running. */
function c3Clock(start = 0) {
const state = { t: start, scheduled: [], ran: [] }
return {
state,
now: () => state.t,
schedule: (fn, delayMs) => { state.scheduled.push({ fn, delayMs, at: state.t }) },
advance(ms) { state.t += ms },
/** Run everything registered so far, as the timer would. */
fireAll() { for (const { fn } of state.scheduled.splice(0)) { fn(); state.ran.push(true) } },
}
}
test('C3-a/b: the 10-minute throttle registers once inside the window and again outside', () => {
const store = c3Store(13) // 13 vs 10 samples => delta 0.30 => stale
try {
c3Centroid(store, 10)
const clock = c3Clock()
const first = store.centroidStatus('fp-c3', { now: clock.now, schedule: clock.schedule })
assert.equal(first.rebuildScheduled, true, 'first stale observation must register a rebuild')
clock.advance(9 * 60 * 1000 + 59 * 1000) // 9m59s: inside the 10-minute window
const second = store.centroidStatus('fp-c3', { now: clock.now, schedule: clock.schedule })
assert.equal(second.rebuildScheduled, false, 'inside the window nothing is registered again')
assert.equal(clock.state.scheduled.length, 1, 'exactly one rebuild was scheduled so far')
clock.advance(10 * 60 * 1000 - (9 * 60 + 59) * 1000 + 1000) // past 10m01s
const third = store.centroidStatus('fp-c3', { now: clock.now, schedule: clock.schedule })
assert.equal(third.rebuildScheduled, true, 'outside the window a new rebuild is registered')
assert.equal(clock.state.scheduled.length, 2, 'two registrations total')
} finally {
store.close()
}
})
test('C3-c: the scheduled rebuild carries the Go 2-second delay and does not run inline', () => {
const store = c3Store(13)
try {
c3Centroid(store, 10)
const clock = c3Clock()
const status = store.centroidStatus('fp-c3', { now: clock.now, schedule: clock.schedule })
assert.equal(status.rebuildScheduled, true)
assert.equal(clock.state.scheduled.length, 1)
const [entry] = clock.state.scheduled
assert.equal(entry.delayMs, BlockStore.CENTROID_REBUILD_DELAY_MS, 'delay is the Go 2 s constant')
assert.equal(entry.delayMs, 2000, 'and that constant is exactly 2000 ms')
// Nothing ran yet: registration is not execution.
assert.equal(clock.state.ran.length, 0, 'the rebuild must not run during the status call')
} finally {
store.close()
}
})
test('C3-d: the registration reaches a path that really runs, and clears staleness', () => {
const store = c3Store(13)
try {
c3Centroid(store, 10)
const clock = c3Clock()
let rebuildCalls = 0
const realRebuild = store.rebuildCentroid.bind(store)
store.rebuildCentroid = (fp) => { rebuildCalls += 1; return realRebuild(fp) }
const first = store.centroidStatus('fp-c3', { now: clock.now, schedule: clock.schedule })
assert.equal(first.stale, true, 'precondition: the space starts stale (13 vs 10 samples)')
clock.fireAll() // the timer fires
assert.equal(rebuildCalls, 1, 'the registered callback rebuilt exactly once (not merely recorded)')
const after = store.centroidStatus('fp-c3', {
now: () => clock.state.t + 11 * 60 * 1000, schedule: clock.schedule,
})
assert.equal(after.stale, false,
'after the rebuild the space is no longer stale — this is what the 1391→1877 incident broke')
} finally {
store.close()
}
})
test('C3-e: recall-path truth — a centroid built from zero samples is unusable, not stale', () => {
const store = c3Store(5)
try {
c3Centroid(store, 0)
// The recall projection is `LoadCentroid`'s answer (go :179-185): unusable,
// and with NO staleness judgement at all.
const status = store.centroidRecallState('fp-c3')
assert.equal(status.available, false, 'built from 0 samples: no centering')
assert.equal(status.stale, null, 'the recall path carries no stale bit (go :179-185)')
assert.equal(status.vectorCount, 0, 'the sample count is reported as-is')
} finally {
store.close()
}
})
test('C3-f: status-path truth — zero samples with blocks present IS stale', () => {
const store = c3Store(5)
try {
c3Centroid(store, 0)
const status = store.centroidStatus('fp-c3')
assert.equal(status.stale, true, 'vecCount == 0 && currentBlocks > 0 => stale (go :431-432)')
assert.equal(status.currentBlocks, 5)
assert.equal(status.delta, null, 'delta is undefined when the denominator is 0')
} finally {
store.close()
}
})
test('C3-g: the ratio is STRICTLY greater than 0.20', () => {
const store = c3Store(1201) // 1201 vs 1000 samples => delta 0.201
try {
c3Centroid(store, 1000)
const justOver = store.centroidStatus('fp-c3', { schedule: () => {}, now: () => 0 })
assert.equal(justOver.delta, 0.201, 'precondition: delta is 0.201')
assert.equal(justOver.stale, true, '0.201 > 0.20 => stale')
} finally {
store.close()
}
const exact = c3Store(12) // 12 vs 10 => delta exactly 0.20
try {
c3Centroid(exact, 10)
const at = exact.centroidStatus('fp-c3', { schedule: () => {}, now: () => 0 })
assert.equal(at.delta, 0.2, 'precondition: delta is exactly 0.20')
assert.equal(at.stale, false, 'exactly 20% is NOT stale — the comparison is strictly greater')
} finally {
exact.close()
}
})
test('C3-h: delta uses (current − built) / built, not the inverse', () => {
const store = c3Store(13)
try {
c3Centroid(store, 10)
const status = store.centroidStatus('fp-c3', { schedule: () => {}, now: () => 0 })
assert.equal(status.currentBlocks, 13)
assert.equal(status.vectorCount, 10)
// (13-10)/10 = 0.3. The inverted form would give (10-13)/13 ≈ 0.2308 and the
// swapped-denominator form would give 0 — neither equals 0.3.
assert.equal(status.delta, 0.3, 'delta must be (curVectored - vecCount) / vecCount')
assert.equal(status.stale, true, '0.3 > 0.2 => stale')
} finally {
store.close()
}
})
test('C3-i: a stale hit registers the rebuild and never rebuilds inline', () => {
const store = c3Store(13)
try {
c3Centroid(store, 10)
let syncCalls = 0
const realRebuild = store.rebuildCentroid.bind(store)
store.rebuildCentroid = (fp) => { syncCalls += 1; return realRebuild(fp) }
const clock = c3Clock()
const status = store.centroidStatus('fp-c3', { now: clock.now, schedule: clock.schedule })
assert.equal(status.stale, true)
assert.equal(status.available, false, 'a stale centroid is not used for centering this round')
assert.equal(syncCalls, 0, 'rebuilding inline would nest a transaction on the caller’s statement')
assert.equal(clock.state.scheduled.length, 1, 'it is registered instead')
} finally {
store.close()
}
})
test('C3-j: corrupt centroid metadata is an error, not a silent "no centering"', () => {
const store = c3Store(5)
try {
store.raw().prepare(
'INSERT INTO graph_centroid (fingerprint, dim, vector_count, vector, stats, is_anomalous, built_at) VALUES (?,?,?,?,?,?,?)',
).run('fp-c3', 0, 5, '[]', '{}', 1, Date.now())
assert.throws(() => store.centroidStatus('fp-c3'), /centroid dim is 0 \(corrupted\?\)/,
'a stored dim <= 0 is corrupt and must surface (go :165-167)')
} finally {
store.close()
}
})
test('C3-i†: replacing registration with inline rebuild bites the unchanged async criterion', async (t) => {
await withMutatedCopy(
'store.js',
[[
` const rebuildScheduled = this.#scheduleCentroidRebuild(fingerprint, { now, schedule })`,
` this.rebuildCentroid(fingerprint) // MUTATED: rebuild inline\n const rebuildScheduled = true`,
]],
async ({ BlockStore: MutantStore }) => {
const store = new MutantStore({ path: ':memory:' }).open()
try {
for (let i = 0; i < 13; i += 1) {
store.putBlock({ text: `C3 mutant ${i}`, vector: [1, 0, 0], fingerprint: 'fp-c3' })
}
store.saveCentroid({ fingerprint: 'fp-c3', dim: 3, vectorCount: 10, mean: [0, 0, 0], isAnomalous: true })
let syncCalls = 0
const rebuild = store.rebuildCentroid.bind(store)
store.rebuildCentroid = (fp) => { syncCalls += 1; return rebuild(fp) }
// The unchanged criterion: a stale observation must NOT rebuild inline.
// (It also must not lose the block, which is the second half.)
assert.throws(() => {
const status = store.centroidStatus('fp-c3', { now: () => 0, schedule: () => {} })
assert.equal(syncCalls, 0, 'stale handling must register, never rebuild inline')
assert.equal(status.rebuildScheduled, true, 'and the rebuild must still be registered')
}, (err) => {
assert.equal(err.code, 'ERR_ASSERTION')
t.diagnostic(`C3-i† RAW RED >>> ${err.message}`)
return true
}, 'inline rebuild must break the unchanged async criterion')
} finally { store.close() }
},
)
})
// C3 mutation guards. Each mutation changes lib/store.js and then runs the
// unchanged positive predicate; only ERR_ASSERTION is accepted as a bite.
async function c3Mutant(replacements, body) {
await withMutatedCopy('store.js', replacements, async ({ BlockStore: MutantStore }) => {
await body(MutantStore)
})
}
function c3Seed(store, current = 13, built = 10) {
for (let i = 0; i < current; i += 1) {
store.putBlock({ text: `C3 mutation ${i}`, vector: [1, 0, 0], fingerprint: 'fp-c3' })
}
store.saveCentroid({ fingerprint: 'fp-c3', dim: 3, vectorCount: built, mean: [0, 0, 0], isAnomalous: true })
}
test('C3† ratio boundary: >= instead of > bites the exact-0.20 criterion', async (t) => {
await c3Mutant([[
'if (delta > BlockStore.CENTROID_REBUILD_RATIO) {',
'if (delta >= BlockStore.CENTROID_REBUILD_RATIO) { // MUTATED: inclusive boundary',
]], async (MutantStore) => {
const store = new MutantStore({ path: ':memory:' }).open()
try {
c3Seed(store, 12, 10)
assert.throws(() => {
assert.equal(store.centroidStatus('fp-c3', { schedule: () => {}, now: () => 0 }).stale, false,
'exactly 0.20 must not be stale')
}, (err) => { assert.equal(err.code, 'ERR_ASSERTION'); t.diagnostic(`C3† boundary RAW RED >>> ${err.message}`); return true })
} finally { store.close() }
})
})
test('C3† direction: using currentBlocks as the denominator bites delta 0.30', async (t) => {
await c3Mutant([[
'(currentBlocks - vectorCount) / vectorCount',
'(currentBlocks - vectorCount) / currentBlocks', // MUTATED: wrong denominator
]], async (MutantStore) => {
const store = new MutantStore({ path: ':memory:' }).open()
try {
c3Seed(store, 13, 10)
assert.throws(() => {
assert.equal(store.centroidStatus('fp-c3', { schedule: () => {}, now: () => 0 }).delta, 0.3,
'delta must use built vectorCount as denominator')
}, (err) => { assert.equal(err.code, 'ERR_ASSERTION'); t.diagnostic(`C3† direction RAW RED >>> ${err.message}`); return true })
} finally { store.close() }
})
})
test('C3† throttle: deleting the 10-minute guard bites the window criterion', async (t) => {
await c3Mutant([[
'if (last !== undefined && now() - last < BlockStore.CENTROID_REBUILD_WAIT_MS) {',
'if (false) { // MUTATED: no throttle',
]], async (MutantStore) => {
const store = new MutantStore({ path: ':memory:' }).open()
try {
c3Seed(store)
const clock = c3Clock()
store.centroidStatus('fp-c3', { now: clock.now, schedule: clock.schedule })
clock.advance(1)
store.centroidStatus('fp-c3', { now: clock.now, schedule: clock.schedule })
assert.throws(() => {
assert.equal(clock.state.scheduled.length, 1, 'one registration within the throttle window')
}, (err) => { assert.equal(err.code, 'ERR_ASSERTION'); t.diagnostic(`C3† throttle RAW RED >>> ${err.message}`); return true })
} finally { store.close() }
})
})
test('C3† delay: changing 2000 to 0 bites the real-delay criterion', async (t) => {
await c3Mutant([[
'return 2000',
'return 0 // MUTATED: no two-second settling delay',
]], async (MutantStore) => {
const store = new MutantStore({ path: ':memory:' }).open()
try {
c3Seed(store)
const clock = c3Clock()
store.centroidStatus('fp-c3', { now: clock.now, schedule: clock.schedule })
assert.throws(() => {
assert.equal(clock.state.scheduled[0].delayMs, 2000, 'scheduled delay must be 2000 ms')
}, (err) => { assert.equal(err.code, 'ERR_ASSERTION'); t.diagnostic(`C3† delay RAW RED >>> ${err.message}`); return true })
} finally { store.close() }
})
})
test('C3† execution: dropping the rebuild call leaves stale and bites the live-path criterion', async (t) => {
await c3Mutant([[
' this.rebuildCentroid(fingerprint)',
' // MUTATED: registered callback no longer executes the rebuild',
]], async (MutantStore) => {
const store = new MutantStore({ path: ':memory:' }).open()
try {
c3Seed(store)
const clock = c3Clock()
store.centroidStatus('fp-c3', { now: clock.now, schedule: clock.schedule })
clock.fireAll()
assert.throws(() => {
assert.equal(store.centroidStatus('fp-c3', { now: () => 11 * 60 * 1000, schedule: clock.schedule }).stale, false,
'a registered rebuild must execute and clear stale')
}, (err) => { assert.equal(err.code, 'ERR_ASSERTION'); t.diagnostic(`C3† execution RAW RED >>> ${err.message}`); return true })
} finally { store.close() }
})
})
test('C3-k: countVectored is NOT wired into the blockStates hot path', async () => {
const { readFileSync } = await import('node:fs')
const source = readFileSync(new URL('../lib/store.js', import.meta.url), 'utf8')
const start = source.indexOf(' blockStates(activeFingerprint, activeDimension) {')
assert.notEqual(start, -1)
const end = source.indexOf('\n }\n\n /**', start)
assert.notEqual(end, -1)
const body = source.slice(start, end)
assert.equal(body.includes('countVectored'), false,
'blockStates runs on the pre-step hot path; the C3 counting query must stay out of it')
})