From c2bf87a7b37de9e10995d9d85e801247f38d43c5 Mon Sep 17 00:00:00 2001 From: Choti Wongbussakorn <126886556+Chewji9875@users.noreply.github.com> Date: Sat, 29 Aug 2026 22:43:30 +0700 Subject: [PATCH] fix(state): reclaim orphaned index shards and add index diagnostics/healing - Implement generation tracking via generations:registry in KV store - Purge obsolete generation shards upon manifest publish and during startup sweep - Enforce fail-closed manifest validation, FIFO save queue, and 60s in-flight grace period - Add category 'index' to mem::diagnose and mem::heal with audit trail logging - Add comprehensive unit tests covering corrupt manifests, crash recovery, and GC Closes #1115 --- src/functions/diagnostics.ts | 356 ++++++++++++++++++++++++++- src/mcp/tools-registry.ts | 2 +- src/state/index-persistence.ts | 272 +++++++++++++++++++-- test/diagnostics.test.ts | 313 +++++++++++++++++++++++- test/index-persistence.test.ts | 428 +++++++++++++++++++++++++++++++++ 5 files changed, 1347 insertions(+), 24 deletions(-) diff --git a/src/functions/diagnostics.ts b/src/functions/diagnostics.ts index cc982883f..1aac9bedc 100644 --- a/src/functions/diagnostics.ts +++ b/src/functions/diagnostics.ts @@ -10,7 +10,6 @@ import type { Insight, Lease, Lesson, - Checkpoint, Crystal, ProceduralMemory, SemanticMemory, @@ -38,6 +37,7 @@ const ALL_CATEGORIES = [ "crystals", "insights", "mesh", + "index", ]; const TWENTY_FOUR_HOURS_MS = 24 * 60 * 60 * 1000; @@ -620,6 +620,200 @@ export function registerDiagnosticsFunction(sdk: ISdk, kv: StateKV): void { } } + if (categories.includes("index")) { + const [bm25Settled, vectorSettled, registrySettled] = + await Promise.allSettled([ + kv.get<{ + v: number; + generation?: string; + shards?: unknown[]; + chars?: number; + }>(KV.bm25Index, "data:manifest"), + kv.get<{ + v: number; + generation?: string; + shards?: unknown[]; + chars?: number; + }>(KV.bm25Index, "vectors:manifest"), + kv.get<{ + v: number; + generations?: Record< + string, + { + type: "bm25" | "vector"; + createdAt: string; + shardScopes: string[]; + } + >; + }>(KV.bm25Index, "generations:registry"), + ]); + + let bm25Eligible = false; + let bm25Manifest: { + v: number; + generation?: string; + shards?: unknown[]; + chars?: number; + } | null = null; + if (bm25Settled.status === "fulfilled") { + const m = bm25Settled.value; + if (m === null || m === undefined) { + bm25Eligible = true; + bm25Manifest = null; + checks.push({ + name: "index-manifest-bm25", + category: "index", + status: "warn", + message: + "BM25 index manifest not found (index not yet persisted or empty)", + fixable: false, + }); + } else if (m && m.v === 1 && Array.isArray(m.shards)) { + bm25Eligible = true; + bm25Manifest = m; + checks.push({ + name: "index-manifest-bm25", + category: "index", + status: "pass", + message: `BM25 index manifest is valid (${m.shards.length} shards)`, + fixable: false, + }); + } else { + checks.push({ + name: "index-manifest-bm25", + category: "index", + status: "fail", + message: "BM25 index manifest is corrupt", + fixable: false, + }); + } + } else { + checks.push({ + name: "index-manifest-bm25", + category: "index", + status: "fail", + message: "BM25 index manifest read failed", + fixable: false, + }); + } + + let vectorEligible = false; + let vectorManifest: { + v: number; + generation?: string; + shards?: unknown[]; + chars?: number; + } | null = null; + if (vectorSettled.status === "fulfilled") { + const m = vectorSettled.value; + if (m === null || m === undefined) { + vectorEligible = true; + vectorManifest = null; + checks.push({ + name: "index-manifest-vectors", + category: "index", + status: "warn", + message: + "Vector index manifest not found (index not yet persisted or empty)", + fixable: false, + }); + } else if (m && m.v === 1 && Array.isArray(m.shards)) { + vectorEligible = true; + vectorManifest = m; + checks.push({ + name: "index-manifest-vectors", + category: "index", + status: "pass", + message: `Vector index manifest is valid (${m.shards.length} shards)`, + fixable: false, + }); + } else { + checks.push({ + name: "index-manifest-vectors", + category: "index", + status: "fail", + message: "Vector index manifest is corrupt", + fixable: false, + }); + } + } else { + checks.push({ + name: "index-manifest-vectors", + category: "index", + status: "fail", + message: "Vector index manifest read failed", + fixable: false, + }); + } + + const registry = + registrySettled.status === "fulfilled" ? registrySettled.value : null; + const activeBm25Gen = + bm25Eligible && + bm25Manifest && + typeof bm25Manifest.generation === "string" + ? bm25Manifest.generation + : null; + const activeVectorGen = + vectorEligible && + vectorManifest && + typeof vectorManifest.generation === "string" + ? vectorManifest.generation + : null; + + const INDEX_GRACE_PERIOD_MS = 60_000; + let orphanGenCount = 0; + let orphanShardCount = 0; + + if ( + registry && + registry.v === 1 && + registry.generations && + typeof registry.generations === "object" + ) { + for (const [genId, genInfo] of Object.entries(registry.generations)) { + if (genInfo.type === "bm25") { + if (!bm25Eligible) continue; + if (activeBm25Gen && genId === activeBm25Gen) continue; + } else if (genInfo.type === "vector") { + if (!vectorEligible) continue; + if (activeVectorGen && genId === activeVectorGen) continue; + } else { + continue; + } + + const createdAtMs = Date.parse(genInfo.createdAt); + if ( + Number.isNaN(createdAtMs) || + now - createdAtMs > INDEX_GRACE_PERIOD_MS + ) { + orphanGenCount++; + if (Array.isArray(genInfo.shardScopes)) { + orphanShardCount += genInfo.shardScopes.length; + } + } + } + } + + if (orphanGenCount > 0) { + checks.push({ + name: "index-orphan-shards", + category: "index", + status: "fail", + message: `Found ${orphanGenCount} orphan generations (${orphanShardCount} shards) in index registry`, + fixable: true, + }); + } else { + checks.push({ + name: "index-orphan-shards", + category: "index", + status: "pass", + message: "Index shard generations are clean (no orphan shards)", + fixable: false, + }); + } + } + const summary = { pass: checks.filter((c) => c.status === "pass").length, warn: checks.filter((c) => c.status === "warn").length, @@ -1057,6 +1251,166 @@ export function registerDiagnosticsFunction(sdk: ISdk, kv: StateKV): void { } } + if (categories.includes("index")) { + const [bm25Settled, vectorSettled, registrySettled] = + await Promise.allSettled([ + kv.get<{ + v: number; + generation?: string; + shards?: unknown[]; + chars?: number; + }>(KV.bm25Index, "data:manifest"), + kv.get<{ + v: number; + generation?: string; + shards?: unknown[]; + chars?: number; + }>(KV.bm25Index, "vectors:manifest"), + kv.get<{ + v: number; + generations?: Record< + string, + { + type: "bm25" | "vector"; + createdAt: string; + shardScopes: string[]; + } + >; + }>(KV.bm25Index, "generations:registry"), + ]); + + let bm25Eligible = false; + let bm25Manifest: { + v: number; + generation?: string; + shards?: unknown[]; + chars?: number; + } | null = null; + if (bm25Settled.status === "fulfilled") { + const m = bm25Settled.value; + if (m === null || m === undefined) { + bm25Eligible = true; + bm25Manifest = null; + } else if (m && m.v === 1 && Array.isArray(m.shards)) { + bm25Eligible = true; + bm25Manifest = m; + } + } + + let vectorEligible = false; + let vectorManifest: { + v: number; + generation?: string; + shards?: unknown[]; + chars?: number; + } | null = null; + if (vectorSettled.status === "fulfilled") { + const m = vectorSettled.value; + if (m === null || m === undefined) { + vectorEligible = true; + vectorManifest = null; + } else if (m && m.v === 1 && Array.isArray(m.shards)) { + vectorEligible = true; + vectorManifest = m; + } + } + + const registry = + registrySettled.status === "fulfilled" ? registrySettled.value : null; + const activeBm25Gen = + bm25Eligible && + bm25Manifest && + typeof bm25Manifest.generation === "string" + ? bm25Manifest.generation + : null; + const activeVectorGen = + vectorEligible && + vectorManifest && + typeof vectorManifest.generation === "string" + ? vectorManifest.generation + : null; + + const INDEX_GRACE_PERIOD_MS = 60_000; + const orphanGens: Array<{ + id: string; + type: "bm25" | "vector"; + createdAt: string; + shardScopes: string[]; + }> = []; + let orphanShardCount = 0; + + if ( + registry && + registry.v === 1 && + registry.generations && + typeof registry.generations === "object" + ) { + for (const [genId, genInfo] of Object.entries(registry.generations)) { + if (genInfo.type === "bm25") { + if (!bm25Eligible) continue; + if (activeBm25Gen && genId === activeBm25Gen) continue; + } else if (genInfo.type === "vector") { + if (!vectorEligible) continue; + if (activeVectorGen && genId === activeVectorGen) continue; + } else { + continue; + } + + const createdAtMs = Date.parse(genInfo.createdAt); + if ( + Number.isNaN(createdAtMs) || + now - createdAtMs > INDEX_GRACE_PERIOD_MS + ) { + const scopes = Array.isArray(genInfo.shardScopes) + ? genInfo.shardScopes + : []; + orphanGens.push({ + id: genId, + type: genInfo.type, + createdAt: genInfo.createdAt, + shardScopes: scopes, + }); + orphanShardCount += scopes.length; + } + } + } + + if (orphanGens.length > 0) { + if (dryRun) { + details.push( + `[dry-run] Would delete ${orphanShardCount} orphan shards across ${orphanGens.length} unreferenced generations`, + ); + fixed++; + } else { + const deletePromises: Promise[] = []; + for (const gen of orphanGens) { + for (const scope of gen.shardScopes) { + deletePromises.push(kv.delete(scope, "data")); + } + } + await Promise.allSettled(deletePromises); + + for (const gen of orphanGens) { + delete registry!.generations![gen.id]; + } + await kv.set(KV.bm25Index, "generations:registry", registry); + + for (const gen of orphanGens) { + await recordAudit(kv, "heal", "mem::heal", [gen.id], { + entityType: "index_shard", + reason: "orphan-shard-gc", + action: "delete", + }); + } + + details.push( + `Deleted ${orphanShardCount} orphan shards from ${orphanGens.length} unreferenced generations`, + ); + fixed++; + } + } + } + return { success: true, fixed, skipped, details }; }, ); diff --git a/src/mcp/tools-registry.ts b/src/mcp/tools-registry.ts index 1225b4ce7..a49fef976 100644 --- a/src/mcp/tools-registry.ts +++ b/src/mcp/tools-registry.ts @@ -672,7 +672,7 @@ export const V051_TOOLS: McpToolDef[] = [ { name: "memory_diagnose", description: - "Run health checks across all subsystems (actions, leases, sentinels, sketches, signals, sessions, memories, mesh). Identifies stuck, orphaned, and inconsistent state.", + "Run health checks across all subsystems (actions, leases, sentinels, sketches, signals, sessions, memories, mesh, index). Identifies stuck, orphaned, and inconsistent state.", inputSchema: { type: "object", properties: { diff --git a/src/state/index-persistence.ts b/src/state/index-persistence.ts index 6df0e2fda..4afad6487 100644 --- a/src/state/index-persistence.ts +++ b/src/state/index-persistence.ts @@ -16,6 +16,20 @@ const VECTOR_MANIFEST_KEY = "vectors:manifest"; const VECTOR_SHARD_SCOPE_PREFIX = `${KV.bm25Index}:vectors:`; const INDEX_SHARD_KEY = "data"; const DEFAULT_INDEX_SHARD_CHARS = 2_000_000; +const GENERATIONS_REGISTRY_KEY = "generations:registry"; +const SWEEP_GRACE_PERIOD_MS = 60_000; + +type GenerationRegistry = { + v: 1; + generations: Record< + string, + { + type: "bm25" | "vector"; + createdAt: string; + shardScopes: string[]; + } + >; +}; type IndexShardManifest = { v: 1; @@ -24,9 +38,10 @@ type IndexShardManifest = { chars: number; }; -type IndexPersistenceOptions = { +export type IndexPersistenceOptions = { shardChars?: number; createGeneration?: () => string; + sweepGracePeriodMs?: number; }; function shardChars(options: IndexPersistenceOptions): number { @@ -60,6 +75,7 @@ function isValidShardDescriptor( candidate.scope.length > 0 && typeof candidate.key === "string" && candidate.key.length > 0 && + typeof candidate.chars === "number" && Number.isInteger(candidate.chars) && candidate.chars >= 0 ); @@ -68,6 +84,7 @@ function isValidShardDescriptor( export class IndexPersistence { private timer: ReturnType | null = null; private lastFailureLogAt = 0; + private saveQueue: Promise = Promise.resolve(); constructor( private kv: StateKV, @@ -88,18 +105,24 @@ export class IndexPersistence { } async save(): Promise { - if (this.timer) { - clearTimeout(this.timer); - this.timer = null; - } - try { - await this.saveBm25Index(this.bm25.serialize()); - if (this.vector) { - await this.saveVectorIndex(this.vector.serialize()); + const run = async () => { + if (this.timer) { + clearTimeout(this.timer); + this.timer = null; } - } catch (err) { - this.logFailure(err); - } + try { + await this.saveBm25Index(this.bm25.serialize()); + if (this.vector) { + await this.saveVectorIndex(this.vector.serialize()); + } + } catch (err) { + this.logFailure(err); + } + }; + + const next = this.saveQueue.then(run, run); + this.saveQueue = next; + await next; } async load(): Promise<{ @@ -119,9 +142,144 @@ export class IndexPersistence { vector = VectorIndex.deserialize(vecData); } + this.sweepOrphanShards().catch((err) => { + logger.warn("index persistence: orphan shard sweep failed during load", { + message: errorMessage(err), + }); + }); + return { bm25, vector }; } + async sweepOrphanShards(): Promise<{ + deletedShards: number; + purgedGenerations: number; + }> { + let registry: GenerationRegistry; + try { + registry = await this.getRegistry(); + } catch (err) { + logger.warn( + "index persistence: failed to read generation registry during orphan sweep, skipping GC", + { message: errorMessage(err) }, + ); + return { deletedShards: 0, purgedGenerations: 0 }; + } + + let bm25Eligible = false; + let activeBm25Gen: string | null = null; + try { + const m = await this.kv.get( + KV.bm25Index, + BM25_MANIFEST_KEY, + ); + if (m === null || m === undefined) { + bm25Eligible = true; + activeBm25Gen = null; + } else if (m && m.v === 1 && Array.isArray(m.shards)) { + bm25Eligible = true; + activeBm25Gen = typeof m.generation === "string" ? m.generation : null; + } else { + logger.warn( + "index persistence: BM25 manifest corrupt during orphan sweep, skipping BM25 GC", + ); + } + } catch (err) { + logger.warn( + "index persistence: BM25 manifest read failed during orphan sweep, skipping BM25 GC", + { message: errorMessage(err) }, + ); + } + + let vectorEligible = false; + let activeVectorGen: string | null = null; + try { + const m = await this.kv.get( + KV.bm25Index, + VECTOR_MANIFEST_KEY, + ); + if (m === null || m === undefined) { + vectorEligible = true; + activeVectorGen = null; + } else if (m && m.v === 1 && Array.isArray(m.shards)) { + vectorEligible = true; + activeVectorGen = typeof m.generation === "string" ? m.generation : null; + } else { + logger.warn( + "index persistence: Vector manifest corrupt during orphan sweep, skipping Vector GC", + ); + } + } catch (err) { + logger.warn( + "index persistence: Vector manifest read failed during orphan sweep, skipping Vector GC", + { message: errorMessage(err) }, + ); + } + + const now = Date.now(); + const gracePeriod = + this.options.sweepGracePeriodMs ?? SWEEP_GRACE_PERIOD_MS; + const orphanGenerations: string[] = []; + const orphanShards: IndexShardManifest["shards"] = []; + + for (const [genId, genInfo] of Object.entries(registry.generations)) { + if (genInfo.type === "bm25") { + if (!bm25Eligible) continue; + if (activeBm25Gen && genId === activeBm25Gen) continue; + } else if (genInfo.type === "vector") { + if (!vectorEligible) continue; + if (activeVectorGen && genId === activeVectorGen) continue; + } else { + continue; + } + + const createdAtMs = Date.parse(genInfo.createdAt); + if (!Number.isNaN(createdAtMs) && now - createdAtMs < gracePeriod) { + continue; + } + + orphanGenerations.push(genId); + if (Array.isArray(genInfo.shardScopes)) { + for (const scope of genInfo.shardScopes) { + orphanShards.push({ scope, key: INDEX_SHARD_KEY, chars: 0 }); + } + } + } + + if (orphanShards.length > 0) { + await this.deleteShards(orphanShards, "orphan_shard_gc"); + } + + if (orphanGenerations.length > 0) { + for (const genId of orphanGenerations) { + delete registry.generations[genId]; + } + await this.saveRegistry(registry).catch(() => {}); + } + + const stats = { + deletedShards: orphanShards.length, + purgedGenerations: orphanGenerations.length, + }; + + if (stats.purgedGenerations > 0) { + logger.info("index persistence: orphan shard sweep completed", { + purgedGenerations: stats.purgedGenerations, + deletedShards: stats.deletedShards, + }); + await this.auditIndexPersistence( + "orphan_shard_gc", + [statePath(KV.bm25Index, GENERATIONS_REGISTRY_KEY)], + { + deletedShards: stats.deletedShards, + purgedGenerations: stats.purgedGenerations, + }, + ); + } + + return stats; + } + stop(): void { if (this.timer) { clearTimeout(this.timer); @@ -129,6 +287,34 @@ export class IndexPersistence { } } + private async getRegistry(): Promise { + const reg = await this.kv.get( + KV.bm25Index, + GENERATIONS_REGISTRY_KEY, + ); + if ( + reg && + reg.v === 1 && + reg.generations && + typeof reg.generations === "object" && + !Array.isArray(reg.generations) + ) { + return reg; + } + if (reg === null || reg === undefined) { + return { v: 1, generations: {} }; + } + throw new Error("Invalid generations registry format in KV store"); + } + + private async saveRegistry(registry: GenerationRegistry): Promise { + await this.kv.set( + KV.bm25Index, + GENERATIONS_REGISTRY_KEY, + registry, + ); + } + private logFailure(err: unknown): void { const now = Date.now(); // Throttle: persistence failures under load arrive in bursts @@ -154,6 +340,7 @@ export class IndexPersistence { BM25_MANIFEST_KEY, BM25_KEY, BM25_SHARD_SCOPE_PREFIX, + "bm25", ); } @@ -163,6 +350,7 @@ export class IndexPersistence { VECTOR_MANIFEST_KEY, VECTOR_KEY, VECTOR_SHARD_SCOPE_PREFIX, + "vector", ); } @@ -171,6 +359,7 @@ export class IndexPersistence { manifestKey: string, legacyKey: string, scopePrefix: string, + type: "bm25" | "vector", ): Promise { const previous = await this.kv .get(KV.bm25Index, manifestKey) @@ -192,6 +381,14 @@ export class IndexPersistence { chunks.push(chunk); } + const registry = await this.getRegistry(); + registry.generations[generation] = { + type, + createdAt: new Date().toISOString(), + shardScopes: shards.map((s) => s.scope), + }; + await this.saveRegistry(registry); + const writeResults = await Promise.allSettled( shards.map(async (shard, index) => { const chunk = chunks[index] ?? ""; @@ -212,6 +409,11 @@ export class IndexPersistence { ); if (failedWrite) { await this.deleteShards(shards, "shard_write_rollback"); + const curReg = await this.getRegistry().catch(() => null); + if (curReg && curReg.generations[generation]) { + delete curReg.generations[generation]; + await this.saveRegistry(curReg).catch(() => {}); + } throw failedWrite.reason; } @@ -250,17 +452,53 @@ export class IndexPersistence { }); } else { await this.deleteShards(shards, "manifest_publish_rollback"); + const curReg = await this.getRegistry().catch(() => null); + if (curReg && curReg.generations[generation]) { + delete curReg.generations[generation]; + await this.saveRegistry(curReg).catch(() => {}); + } + throw err; } - throw err; } await this.deleteKey(KV.bm25Index, legacyKey, "legacy_cleanup"); + + const activeRegistry = await this.getRegistry(); + const obsoleteGenerations: string[] = []; + const obsoleteShards: IndexShardManifest["shards"] = []; + + for (const [genId, genInfo] of Object.entries(activeRegistry.generations)) { + if (genInfo.type === type && genId !== generation) { + obsoleteGenerations.push(genId); + if (Array.isArray(genInfo.shardScopes)) { + for (const scope of genInfo.shardScopes) { + obsoleteShards.push({ scope, key: INDEX_SHARD_KEY, chars: 0 }); + } + } + } + } + + if (obsoleteShards.length > 0) { + await this.deleteShards(obsoleteShards, "previous_generation_cleanup"); + } + + if (obsoleteGenerations.length > 0) { + for (const genId of obsoleteGenerations) { + delete activeRegistry.generations[genId]; + } + await this.saveRegistry(activeRegistry).catch(() => {}); + } + if (previous?.v === 1 && Array.isArray(previous.shards)) { const currentShardIds = new Set( shards.map((shard) => `${shard.scope}\0${shard.key}`), ); + const obsoleteShardIds = new Set( + obsoleteShards.map((shard) => `${shard.scope}\0${shard.key}`), + ); for (const shard of previous.shards) { - if (currentShardIds.has(`${shard.scope}\0${shard.key}`)) continue; + const id = `${shard.scope}\0${shard.key}`; + if (currentShardIds.has(id) || obsoleteShardIds.has(id)) continue; await this.deleteShards([shard], "previous_generation_cleanup"); } } @@ -306,9 +544,9 @@ export class IndexPersistence { shards: IndexShardManifest["shards"], reason: string, ): Promise { - for (const shard of shards) { - await this.deleteKey(shard.scope, shard.key, reason); - } + await Promise.allSettled( + shards.map((shard) => this.deleteKey(shard.scope, shard.key, reason)), + ); } private async isManifestPublished( diff --git a/test/diagnostics.test.ts b/test/diagnostics.test.ts index 1e168767a..ff612c28a 100644 --- a/test/diagnostics.test.ts +++ b/test/diagnostics.test.ts @@ -195,15 +195,15 @@ describe("Diagnostics Functions", () => { }; expect(result.success).toBe(true); - // 15 = 8 original (actions, leases, sentinels, sketches, signals, + // 16 = 8 original (actions, leases, sentinels, sketches, signals, // sessions, memories, mesh) + 6 added in #lesson-visibility // (lessons, summaries, semantic, procedural, crystals, insights) + - // 1 added in #memory-project-scope (memory-project-coverage). - expect(result.summary.pass).toBe(15); - expect(result.summary.warn).toBe(0); + // 1 added in #memory-project-scope (memory-project-coverage) + + // 1 added in #index-persistence (index-orphan-shards). + expect(result.summary.pass).toBe(16); + expect(result.summary.warn).toBe(2); expect(result.summary.fail).toBe(0); expect(result.summary.fixable).toBe(0); - expect(result.checks.every((c) => c.status === "pass")).toBe(true); }); it("active action with no lease produces warn", async () => { @@ -863,6 +863,309 @@ describe("Diagnostics Functions", () => { expect(result.checks.find((c) => c.name === "insight-bad-confidence:ins_inf")?.status).toBe("warn"); expect(result.checks.find((c) => c.name === "semantic-bad-confidence:sem_nan")?.status).toBe("warn"); }); + + it("index category: passes with valid manifests and clean registry", async () => { + await kv.set(KV.bm25Index, "data:manifest", { + v: 1, + generation: "gen_bm25_1", + shards: [{ scope: "mem:index:bm25:bm25:gen_bm25_1:00000", key: "data", chars: 50 }], + chars: 50, + }); + await kv.set(KV.bm25Index, "vectors:manifest", { + v: 1, + generation: "gen_vec_1", + shards: [{ scope: "mem:index:bm25:vectors:gen_vec_1:00000", key: "data", chars: 50 }], + chars: 50, + }); + await kv.set(KV.bm25Index, "generations:registry", { + v: 1, + generations: { + gen_bm25_1: { + type: "bm25", + createdAt: new Date(Date.now() - 10_000).toISOString(), + shardScopes: ["mem:index:bm25:bm25:gen_bm25_1:00000"], + }, + gen_vec_1: { + type: "vector", + createdAt: new Date(Date.now() - 10_000).toISOString(), + shardScopes: ["mem:index:bm25:vectors:gen_vec_1:00000"], + }, + }, + }); + + const result = (await sdk.trigger("mem::diagnose", { + categories: ["index"], + })) as { checks: DiagnosticCheck[]; success: boolean }; + + expect(result.success).toBe(true); + expect(result.checks.find((c) => c.name === "index-manifest-bm25")?.status).toBe("pass"); + expect(result.checks.find((c) => c.name === "index-manifest-vectors")?.status).toBe("pass"); + expect(result.checks.find((c) => c.name === "index-orphan-shards")?.status).toBe("pass"); + }); + + it("index category: warns on missing manifests", async () => { + const result = (await sdk.trigger("mem::diagnose", { + categories: ["index"], + })) as { checks: DiagnosticCheck[] }; + + expect(result.checks.find((c) => c.name === "index-manifest-bm25")?.status).toBe("warn"); + expect(result.checks.find((c) => c.name === "index-manifest-vectors")?.status).toBe("warn"); + expect(result.checks.find((c) => c.name === "index-orphan-shards")?.status).toBe("pass"); + }); + + it("index category: fails on corrupt manifests", async () => { + await kv.set(KV.bm25Index, "data:manifest", { v: 2, invalid: true }); + await kv.set(KV.bm25Index, "vectors:manifest", { v: 1, shards: "not-array" }); + + const result = (await sdk.trigger("mem::diagnose", { + categories: ["index"], + })) as { checks: DiagnosticCheck[] }; + + expect(result.checks.find((c) => c.name === "index-manifest-bm25")?.status).toBe("fail"); + expect(result.checks.find((c) => c.name === "index-manifest-vectors")?.status).toBe("fail"); + }); + + it("index category: fails with fixable flag on orphan generations", async () => { + await kv.set(KV.bm25Index, "data:manifest", { + v: 1, + generation: "gen_bm25_active", + shards: [{ scope: "mem:index:bm25:bm25:gen_bm25_active:00000", key: "data", chars: 50 }], + chars: 50, + }); + await kv.set(KV.bm25Index, "generations:registry", { + v: 1, + generations: { + gen_bm25_active: { + type: "bm25", + createdAt: new Date(Date.now() - 120_000).toISOString(), + shardScopes: ["mem:index:bm25:bm25:gen_bm25_active:00000"], + }, + gen_bm25_orphan1: { + type: "bm25", + createdAt: new Date(Date.now() - 120_000).toISOString(), + shardScopes: ["mem:index:bm25:bm25:gen_bm25_orphan1:00000"], + }, + gen_vec_orphan2: { + type: "vector", + createdAt: new Date(Date.now() - 120_000).toISOString(), + shardScopes: [ + "mem:index:bm25:vectors:gen_vec_orphan2:00000", + "mem:index:bm25:vectors:gen_vec_orphan2:00001", + ], + }, + }, + }); + + const result = (await sdk.trigger("mem::diagnose", { + categories: ["index"], + })) as { checks: DiagnosticCheck[] }; + + const orphanCheck = result.checks.find((c) => c.name === "index-orphan-shards"); + expect(orphanCheck?.status).toBe("fail"); + expect(orphanCheck?.fixable).toBe(true); + expect(orphanCheck?.message).toContain("2 orphan generations"); + expect(orphanCheck?.message).toContain("3 shards"); + }); + + it("index category: ignores in-flight generation younger than 60s grace period", async () => { + await kv.set(KV.bm25Index, "data:manifest", { + v: 1, + generation: "gen_bm25_active", + shards: [{ scope: "mem:index:bm25:bm25:gen_bm25_active:00000", key: "data", chars: 50 }], + chars: 50, + }); + await kv.set(KV.bm25Index, "generations:registry", { + v: 1, + generations: { + gen_bm25_active: { + type: "bm25", + createdAt: new Date(Date.now() - 120_000).toISOString(), + shardScopes: ["mem:index:bm25:bm25:gen_bm25_active:00000"], + }, + gen_inflight: { + type: "bm25", + createdAt: new Date(Date.now() - 10_000).toISOString(), + shardScopes: ["mem:index:bm25:bm25:gen_inflight:00000"], + }, + }, + }); + + const result = (await sdk.trigger("mem::diagnose", { + categories: ["index"], + })) as { checks: DiagnosticCheck[] }; + + expect(result.checks.find((c) => c.name === "index-orphan-shards")?.status).toBe("pass"); + }); + + it("index category: heal dry-run reports orphan shards without modifying storage", async () => { + await kv.set(KV.bm25Index, "data:manifest", { + v: 1, + generation: "gen_active", + shards: [{ scope: "mem:index:bm25:bm25:gen_active:00000", key: "data", chars: 50 }], + chars: 50, + }); + await kv.set("mem:index:bm25:bm25:gen_orphan:00000", "data", "orphan-data"); + await kv.set(KV.bm25Index, "generations:registry", { + v: 1, + generations: { + gen_active: { + type: "bm25", + createdAt: new Date(Date.now() - 120_000).toISOString(), + shardScopes: ["mem:index:bm25:bm25:gen_active:00000"], + }, + gen_orphan: { + type: "bm25", + createdAt: new Date(Date.now() - 120_000).toISOString(), + shardScopes: ["mem:index:bm25:bm25:gen_orphan:00000"], + }, + }, + }); + + const result = (await sdk.trigger("mem::heal", { + categories: ["index"], + dryRun: true, + })) as { success: boolean; fixed: number; details: string[] }; + + expect(result.success).toBe(true); + expect(result.fixed).toBe(1); + expect(result.details[0]).toBe("[dry-run] Would delete 1 orphan shards across 1 unreferenced generations"); + + // Storage remains intact + const orphanData = await kv.get("mem:index:bm25:bm25:gen_orphan:00000", "data"); + expect(orphanData).toBe("orphan-data"); + const registry = await kv.get<{ generations: Record }>(KV.bm25Index, "generations:registry"); + expect(registry?.generations["gen_orphan"]).toBeDefined(); + }); + + it("index category: heal live deletes orphan shards, updates registry, and writes audit logs", async () => { + await kv.set(KV.bm25Index, "data:manifest", { + v: 1, + generation: "gen_active", + shards: [{ scope: "mem:index:bm25:bm25:gen_active:00000", key: "data", chars: 50 }], + chars: 50, + }); + await kv.set("mem:index:bm25:bm25:gen_active:00000", "data", "active-data"); + await kv.set("mem:index:bm25:bm25:gen_orphan1:00000", "data", "orphan-data-1"); + await kv.set("mem:index:bm25:vectors:gen_orphan2:00000", "data", "orphan-data-2"); + await kv.set("mem:index:bm25:vectors:gen_orphan2:00001", "data", "orphan-data-3"); + + await kv.set(KV.bm25Index, "generations:registry", { + v: 1, + generations: { + gen_active: { + type: "bm25", + createdAt: new Date(Date.now() - 120_000).toISOString(), + shardScopes: ["mem:index:bm25:bm25:gen_active:00000"], + }, + gen_orphan1: { + type: "bm25", + createdAt: new Date(Date.now() - 120_000).toISOString(), + shardScopes: ["mem:index:bm25:bm25:gen_orphan1:00000"], + }, + gen_orphan2: { + type: "vector", + createdAt: new Date(Date.now() - 120_000).toISOString(), + shardScopes: [ + "mem:index:bm25:vectors:gen_orphan2:00000", + "mem:index:bm25:vectors:gen_orphan2:00001", + ], + }, + }, + }); + + const result = (await sdk.trigger("mem::heal", { + categories: ["index"], + dryRun: false, + })) as { success: boolean; fixed: number; details: string[] }; + + expect(result.success).toBe(true); + expect(result.fixed).toBe(1); + expect(result.details[0]).toBe("Deleted 3 orphan shards from 2 unreferenced generations"); + + // Orphan shards deleted + expect(await kv.get("mem:index:bm25:bm25:gen_orphan1:00000", "data")).toBeNull(); + expect(await kv.get("mem:index:bm25:vectors:gen_orphan2:00000", "data")).toBeNull(); + expect(await kv.get("mem:index:bm25:vectors:gen_orphan2:00001", "data")).toBeNull(); + + // Active shard preserved + expect(await kv.get("mem:index:bm25:bm25:gen_active:00000", "data")).toBe("active-data"); + + // Registry pruned + const registry = await kv.get<{ generations: Record }>(KV.bm25Index, "generations:registry"); + expect(registry?.generations["gen_orphan1"]).toBeUndefined(); + expect(registry?.generations["gen_orphan2"]).toBeUndefined(); + expect(registry?.generations["gen_active"]).toBeDefined(); + + // Audit entries created + const audits = await kv.list<{ targetIds: string[]; details: { reason: string } }>(KV.audit); + const shardAudits = audits.filter((a) => a.details?.reason === "orphan-shard-gc"); + expect(shardAudits.length).toBe(2); + expect(shardAudits.map((a) => a.targetIds[0])).toEqual(expect.arrayContaining(["gen_orphan1", "gen_orphan2"])); + }); + + it("index category: heal fails closed and does not delete shards when data:manifest is corrupt or throws", async () => { + const failingKv = { + ...kv, + get: vi.fn(async (scope: string, key: string): Promise => { + if (scope === KV.bm25Index && key === "data:manifest") { + throw new Error("backend read failure"); + } + return kv.get(scope, key); + }), + }; + + await kv.set("mem:index:bm25:bm25:gen_orphan1:00000", "data", "orphan-data-1"); + await kv.set(KV.bm25Index, "generations:registry", { + v: 1, + generations: { + gen_orphan1: { + type: "bm25", + createdAt: new Date(Date.now() - 120_000).toISOString(), + shardScopes: ["mem:index:bm25:bm25:gen_orphan1:00000"], + }, + }, + }); + + const localSdk = mockSdk(); + registerDiagnosticsFunction(localSdk as never, failingKv as never); + + const result = (await localSdk.trigger("mem::heal", { + categories: ["index"], + dryRun: false, + })) as { success: boolean; fixed: number; details: string[] }; + + expect(result.success).toBe(true); + expect(result.fixed).toBe(0); + expect(result.details.length).toBe(0); + + // Storage was NOT deleted + expect(await kv.get("mem:index:bm25:bm25:gen_orphan1:00000", "data")).toBe("orphan-data-1"); + const registry = await kv.get<{ generations: Record }>(KV.bm25Index, "generations:registry"); + expect(registry?.generations["gen_orphan1"]).toBeDefined(); + }); + + it("index category: diagnose does not report false-positive orphan failure when manifest is corrupt", async () => { + await kv.set(KV.bm25Index, "data:manifest", { v: 999, invalid: true }); + await kv.set(KV.bm25Index, "generations:registry", { + v: 1, + generations: { + gen_bm25_1: { + type: "bm25", + createdAt: new Date(Date.now() - 120_000).toISOString(), + shardScopes: ["mem:index:bm25:bm25:gen_bm25_1:00000"], + }, + }, + }); + + const result = (await sdk.trigger("mem::diagnose", { + categories: ["index"], + })) as { checks: DiagnosticCheck[] }; + + // Manifest check fails because manifest is corrupt + expect(result.checks.find((c) => c.name === "index-manifest-bm25")?.status).toBe("fail"); + // But orphan check is NOT reported as false-positive fixable fail because BM25 is not eligible + expect(result.checks.find((c) => c.name === "index-orphan-shards")?.status).toBe("pass"); + }); }); }); }); diff --git a/test/index-persistence.test.ts b/test/index-persistence.test.ts index 929791657..357340da2 100644 --- a/test/index-persistence.test.ts +++ b/test/index-persistence.test.ts @@ -790,4 +790,432 @@ describe("IndexPersistence", () => { await expect(persistence.load()).resolves.toBeDefined(); }); + + it("cleans up orphan shards from multiple older crashed generations on next save", async () => { + const REGISTRY_KEY = "generations:registry"; + // Simulate two older crashed generations left in registry and KV + await kv.set("mem:index:bm25:bm25:gen_crash1:00000", "data", "crash1-shard0"); + await kv.set("mem:index:bm25:bm25:gen_crash1:00001", "data", "crash1-shard1"); + await kv.set("mem:index:bm25:bm25:gen_crash2:00000", "data", "crash2-shard0"); + + await kv.set(BM25_SCOPE, REGISTRY_KEY, { + v: 1, + generations: { + gen_crash1: { + type: "bm25", + createdAt: new Date(Date.now() - 120_000).toISOString(), + shardScopes: [ + "mem:index:bm25:bm25:gen_crash1:00000", + "mem:index:bm25:bm25:gen_crash1:00001", + ], + }, + gen_crash2: { + type: "bm25", + createdAt: new Date(Date.now() - 100_000).toISOString(), + shardScopes: ["mem:index:bm25:bm25:gen_crash2:00000"], + }, + }, + }); + + const bm25 = makeBm25("obs_1", "healthy active index"); + const persistence = new IndexPersistence(kv as never, bm25, null, { + shardChars: 80, + createGeneration: () => "gen_active", + }); + + await persistence.save(); + + // Shards from crashed generations must be deleted + await expect( + kv.get("mem:index:bm25:bm25:gen_crash1:00000", "data"), + ).resolves.toBeNull(); + await expect( + kv.get("mem:index:bm25:bm25:gen_crash1:00001", "data"), + ).resolves.toBeNull(); + await expect( + kv.get("mem:index:bm25:bm25:gen_crash2:00000", "data"), + ).resolves.toBeNull(); + + // Active shards must exist + const manifest = await getBm25Manifest(kv); + expect(manifest.generation).toBe("gen_active"); + await expect( + kv.get(manifest.shards[0].scope, manifest.shards[0].key), + ).resolves.toEqual(expect.any(String)); + + // Registry must now only contain gen_active + const registry = await kv.get<{ + v: 1; + generations: Record; + }>(BM25_SCOPE, REGISTRY_KEY); + expect(registry).not.toBeNull(); + expect(Object.keys(registry!.generations)).toEqual(["gen_active"]); + }); + + it("cleans up orphan shards from multiple older crashed generations on sweepOrphanShards()", async () => { + const REGISTRY_KEY = "generations:registry"; + + // Setup active BM25 generation + const activeBm25 = makeBm25("obs_bm25", "active bm25"); + const activeVector = makeVector("obs_vec"); + + const p = new IndexPersistence(kv as never, activeBm25, activeVector, { + shardChars: 80, + createGeneration: () => "gen_active", + }); + await p.save(); + + // Inject orphan BM25 and vector generations into registry and KV (older than 60s grace period) + await kv.set("mem:index:bm25:bm25:gen_orphan_bm25:00000", "data", "orphan-bm25-shard"); + await kv.set("mem:index:bm25:vectors:gen_orphan_vec:00000", "data", "orphan-vec-shard"); + + const registry = await kv.get<{ + v: 1; + generations: Record< + string, + { type: "bm25" | "vector"; createdAt: string; shardScopes: string[] } + >; + }>(BM25_SCOPE, REGISTRY_KEY); + expect(registry).not.toBeNull(); + + registry!.generations["gen_orphan_bm25"] = { + type: "bm25", + createdAt: new Date(Date.now() - 120_000).toISOString(), + shardScopes: ["mem:index:bm25:bm25:gen_orphan_bm25:00000"], + }; + registry!.generations["gen_orphan_vec"] = { + type: "vector", + createdAt: new Date(Date.now() - 120_000).toISOString(), + shardScopes: ["mem:index:bm25:vectors:gen_orphan_vec:00000"], + }; + await kv.set(BM25_SCOPE, REGISTRY_KEY, registry); + + const stats = await p.sweepOrphanShards(); + expect(stats.purgedGenerations).toBe(2); + expect(stats.deletedShards).toBe(2); + + // Orphan shards must be deleted + await expect( + kv.get("mem:index:bm25:bm25:gen_orphan_bm25:00000", "data"), + ).resolves.toBeNull(); + await expect( + kv.get("mem:index:bm25:vectors:gen_orphan_vec:00000", "data"), + ).resolves.toBeNull(); + + // Active shards must remain intact + const bm25Manifest = await kv.get( + BM25_SCOPE, + BM25_MANIFEST_KEY, + ); + expect(bm25Manifest?.generation).toBe("gen_active"); + await expect( + kv.get(bm25Manifest!.shards[0].scope, "data"), + ).resolves.toEqual(expect.any(String)); + + const vectorManifest = await kv.get( + BM25_SCOPE, + VECTOR_MANIFEST_KEY, + ); + expect(vectorManifest?.generation).toBe("gen_active"); + await expect( + kv.get(vectorManifest!.shards[0].scope, "data"), + ).resolves.toEqual(expect.any(String)); + + // Registry must now only contain active generations + const updatedRegistry = await kv.get<{ + v: 1; + generations: Record; + }>(BM25_SCOPE, REGISTRY_KEY); + expect(Object.keys(updatedRegistry!.generations).sort()).toEqual(["gen_active"]); + }); + + it("recovers from crash before publishing manifest by sweeping uncommitted shards", async () => { + const REGISTRY_KEY = "generations:registry"; + + // Initial committed generation + const initialBm25 = makeBm25("obs_init", "initial active"); + const persistence = new IndexPersistence(kv as never, initialBm25, null, { + shardChars: 80, + createGeneration: () => "gen_initial", + }); + await persistence.save(); + + // Crash simulation: save wrote shards and updated registry for gen_crash, but died before manifest write + await kv.set("mem:index:bm25:bm25:gen_crash:00000", "data", "crash-shard-data"); + const registry = await kv.get<{ + v: 1; + generations: Record< + string, + { type: "bm25" | "vector"; createdAt: string; shardScopes: string[] } + >; + }>(BM25_SCOPE, REGISTRY_KEY); + registry!.generations["gen_crash"] = { + type: "bm25", + createdAt: new Date(Date.now() - 120_000).toISOString(), + shardScopes: ["mem:index:bm25:bm25:gen_crash:00000"], + }; + await kv.set(BM25_SCOPE, REGISTRY_KEY, registry); + + // Startup or load runs sweepOrphanShards + const stats = await persistence.sweepOrphanShards(); + expect(stats.purgedGenerations).toBe(1); + expect(stats.deletedShards).toBe(1); + + // Uncommitted crashed shards purged + await expect( + kv.get("mem:index:bm25:bm25:gen_crash:00000", "data"), + ).resolves.toBeNull(); + + // Initial generation still healthy and loadable + const loaded = await persistence.load(); + expect(loaded.bm25!.search("initial").length).toBe(1); + }); + + it("rolls back registry entry and shards when shard write fails", async () => { + const REGISTRY_KEY = "generations:registry"; + + const failingKv = { + ...kv, + set: vi.fn(async (scope: string, key: string, data: T): Promise => { + if (scope.includes(":gen_fail:")) { + throw new Error("shard write disk full"); + } + return kv.set(scope, key, data); + }), + }; + + const bm25 = makeBm25("obs_1", "fail shard write"); + const persistence = new IndexPersistence(failingKv as never, bm25, null, { + shardChars: 80, + createGeneration: () => "gen_fail", + }); + + await persistence.save(); + + // Registry must not retain gen_fail + const registry = await kv.get<{ + v: 1; + generations: Record; + }>(BM25_SCOPE, REGISTRY_KEY); + expect(registry?.generations["gen_fail"]).toBeUndefined(); + await expect( + kv.get("mem:index:bm25:bm25:gen_fail:00000", "data"), + ).resolves.toBeNull(); + }); + + it("rolls back registry entry and shards when manifest publish fails", async () => { + const REGISTRY_KEY = "generations:registry"; + + const failingKv = { + ...kv, + set: vi.fn(async (scope: string, key: string, data: T): Promise => { + if (scope === BM25_SCOPE && key === BM25_MANIFEST_KEY) { + throw new Error("manifest publish failed"); + } + return kv.set(scope, key, data); + }), + }; + + const bm25 = makeBm25("obs_1", "fail manifest publish"); + const persistence = new IndexPersistence(failingKv as never, bm25, null, { + shardChars: 80, + createGeneration: () => "gen_fail_manifest", + }); + + await persistence.save(); + + // Registry must not retain gen_fail_manifest + const registry = await kv.get<{ + v: 1; + generations: Record; + }>(BM25_SCOPE, REGISTRY_KEY); + expect(registry?.generations["gen_fail_manifest"]).toBeUndefined(); + await expect( + kv.get("mem:index:bm25:bm25:gen_fail_manifest:00000", "data"), + ).resolves.toBeNull(); + }); + + it("load() triggers non-blocking orphan shard sweep", async () => { + const REGISTRY_KEY = "generations:registry"; + + // Setup active generation + const activeBm25 = makeBm25("obs_1", "active bm25"); + const persistence = new IndexPersistence(kv as never, activeBm25, null, { + shardChars: 80, + createGeneration: () => "gen_active", + }); + await persistence.save(); + + // Inject orphan generation (older than 60s) + await kv.set("mem:index:bm25:bm25:gen_orphan:00000", "data", "orphan-data"); + const registry = await kv.get<{ + v: 1; + generations: Record< + string, + { type: "bm25" | "vector"; createdAt: string; shardScopes: string[] } + >; + }>(BM25_SCOPE, REGISTRY_KEY); + registry!.generations["gen_orphan"] = { + type: "bm25", + createdAt: new Date(Date.now() - 120_000).toISOString(), + shardScopes: ["mem:index:bm25:bm25:gen_orphan:00000"], + }; + await kv.set(BM25_SCOPE, REGISTRY_KEY, registry); + + const loaded = await persistence.load(); + expect(loaded.bm25).not.toBeNull(); + + // Flush async sweep + await vi.runAllTimersAsync(); + await Promise.resolve(); + + await expect( + kv.get("mem:index:bm25:bm25:gen_orphan:00000", "data"), + ).resolves.toBeNull(); + }); + + it("fails closed on manifest read error during sweepOrphanShards without deleting shards", async () => { + const REGISTRY_KEY = "generations:registry"; + + // Write a registered generation and its shard + await kv.set("mem:index:bm25:bm25:gen_test:00000", "data", "protected-data"); + await kv.set(BM25_SCOPE, REGISTRY_KEY, { + v: 1, + generations: { + gen_test: { + type: "bm25", + createdAt: new Date(Date.now() - 120_000).toISOString(), + shardScopes: ["mem:index:bm25:bm25:gen_test:00000"], + }, + }, + }); + + const errorKv = { + ...kv, + get: vi.fn(async (scope: string, key: string) => { + if (scope === BM25_SCOPE && key === BM25_MANIFEST_KEY) { + throw new Error("backend manifest read error"); + } + return kv.get(scope, key); + }), + }; + + const persistence = new IndexPersistence( + errorKv as never, + new SearchIndex(), + null, + ); + + const stats = await persistence.sweepOrphanShards(); + expect(stats.purgedGenerations).toBe(0); + expect(stats.deletedShards).toBe(0); + + // Shards and registry must remain untouched + await expect( + kv.get("mem:index:bm25:bm25:gen_test:00000", "data"), + ).resolves.toBe("protected-data"); + + const reg = await kv.get<{ v: 1; generations: Record }>( + BM25_SCOPE, + REGISTRY_KEY, + ); + expect(reg?.generations["gen_test"]).toBeDefined(); + }); + + it("preserves uncommitted generation younger than 60s grace period during sweep", async () => { + const REGISTRY_KEY = "generations:registry"; + + // Generation created 10 seconds ago (in-flight write) + await kv.set("mem:index:bm25:bm25:gen_inflight:00000", "data", "in-flight-shard"); + await kv.set(BM25_SCOPE, REGISTRY_KEY, { + v: 1, + generations: { + gen_inflight: { + type: "bm25", + createdAt: new Date(Date.now() - 10_000).toISOString(), + shardScopes: ["mem:index:bm25:bm25:gen_inflight:00000"], + }, + }, + }); + + const persistence = new IndexPersistence( + kv as never, + new SearchIndex(), + null, + ); + + const stats = await persistence.sweepOrphanShards(); + expect(stats.purgedGenerations).toBe(0); + expect(stats.deletedShards).toBe(0); + + // In-flight shard and registry entry must be preserved + await expect( + kv.get("mem:index:bm25:bm25:gen_inflight:00000", "data"), + ).resolves.toBe("in-flight-shard"); + + const reg = await kv.get<{ v: 1; generations: Record }>( + BM25_SCOPE, + REGISTRY_KEY, + ); + expect(reg?.generations["gen_inflight"]).toBeDefined(); + }); + + it("serializes concurrent save calls cleanly", async () => { + const executionOrder: string[] = []; + let activeSaves = 0; + let maxConcurrentSaves = 0; + + const serializeKv = { + ...kv, + set: vi.fn(async (scope: string, key: string, data: T): Promise => { + if (key === BM25_MANIFEST_KEY) { + activeSaves++; + if (activeSaves > maxConcurrentSaves) { + maxConcurrentSaves = activeSaves; + } + // Yield execution across microtasks to ensure concurrency contention is tested + await Promise.resolve(); + await Promise.resolve(); + executionOrder.push(`manifest_${(data as TestIndexShardManifest).generation}`); + activeSaves--; + } + return kv.set(scope, key, data); + }), + }; + + let genCounter = 0; + const bm25 = makeBm25("obs_concurrent", "concurrent test"); + const persistence = new IndexPersistence(serializeKv as never, bm25, null, { + shardChars: 80, + createGeneration: () => `gen_${++genCounter}`, + }); + + // Launch multiple saves concurrently + await Promise.all([ + persistence.save(), + persistence.save(), + persistence.save(), + ]); + + expect(maxConcurrentSaves).toBe(1); + expect(executionOrder).toEqual(["manifest_gen_1", "manifest_gen_2", "manifest_gen_3"]); + }); + + it("fails closed when generation registry is corrupted or throws on get", async () => { + const REGISTRY_KEY = "generations:registry"; + + // Set corrupted registry (not valid object) + await kv.set(BM25_SCOPE, REGISTRY_KEY, "invalid-registry-string"); + + const bm25 = makeBm25("obs_corrupt", "corrupt reg test"); + const persistence = new IndexPersistence(kv as never, bm25, null); + + // Save should catch failure and not throw unhandled exception + await persistence.save(); + + // sweepOrphanShards should return 0/0 and not delete anything + const stats = await persistence.sweepOrphanShards(); + expect(stats.purgedGenerations).toBe(0); + expect(stats.deletedShards).toBe(0); + }); });