diff --git a/.env.example b/.env.example index 338bcf5cdf..bc15daead4 100644 --- a/.env.example +++ b/.env.example @@ -858,6 +858,12 @@ PROVIDER_LIMITS_SYNC_SPACING_MS=1500 # (>= 3 retrievals = never compressed). 1 disables the ramp (binary skip at the threshold only). # Used by: open-sse/services/compression/engines/ccr/index.ts. Default: 2. #COMPRESSION_CCR_RETRIEVAL_RAMP_FACTOR=2 +# CCR durable block store (#9061). The in-memory store loses blocks to LRU eviction, the TTL, a +# restart, or a retrieve landing on another instance, while the model is told it can retrieve them +# verbatim. Set to false to keep blocks in memory only, at the cost of that promise. Blocks over +# 512KB and cloud runtimes are memory-only regardless. +# Used by: open-sse/services/compression/engines/ccr/index.ts. Default: true. +#COMPRESSION_CCR_DURABLE_STORE=true # T08/H5 — usage-observed prefix freeze (OPT-IN, default off). When enabled, a system prompt seen # >= THRESHOLD times is treated as a stable cacheable prefix and preserved from compression even # for providers the static cache-aware heuristic does not recognize (freeze = preserve, never diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index a36f60b7be..221db58669 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/docs/reference/ENVIRONMENT.md @@ -454,6 +454,7 @@ detection above). | `COMPRESSION_PIPELINE_BREAKER_THRESHOLD` | `3` | `open-sse/services/compression/pipelineEngineBreaker.ts` | Consecutive cross-request failures before an engine's breaker opens. | | `COMPRESSION_PIPELINE_BREAKER_COOLDOWN_MS` | `30000` | `open-sse/services/compression/pipelineEngineBreaker.ts` | Milliseconds an opened engine stays skipped before a half-open probe. | | `COMPRESSION_CCR_RETRIEVAL_RAMP_FACTOR` | `2` | `open-sse/services/compression/engines/ccr/index.ts` | T08/H8 CCR retrieval-feedback ramp: each prior retrieval of a stored block raises its effective `minChars` linearly (frequently-retrieved content compresses less; `>=3` retrievals = never compressed). `1` disables the ramp (binary skip at the threshold only). | +| `COMPRESSION_CCR_DURABLE_STORE` | `true` | `open-sse/services/compression/engines/ccr/index.ts` | CCR durable block store (#9061). Backs the in-memory store with SQLite so a block survives LRU eviction, the TTL, a restart, or a retrieve landing on another instance. Set `false` to keep blocks in memory only. Blocks over 512KB and cloud runtimes stay memory-only regardless. | | `COMPRESSION_PREFIX_FREEZE_ENABLED` | `false` | `open-sse/services/compression/prefixFreeze.ts` | T08/H5 usage-observed prefix freeze master switch. **Opt-in (default off)** — when on, a system prompt observed `>=` the threshold is treated as a stable cacheable prefix and preserved from compression even for providers the static cache heuristic misses (freeze only *preserves*, never mutates). | | `COMPRESSION_PREFIX_FREEZE_THRESHOLD` | `3` | `open-sse/services/compression/prefixFreeze.ts` | Observations of a system prompt before it is treated as a frozen stable prefix. | | `OMNIROUTE_BOOTSTRAPPED` | `false` | `src/app/(dashboard)/dashboard/page.tsx` | Set `true` by bootstrap script after initial setup. Controls setup wizard visibility. | diff --git a/open-sse/services/compression/engines/ccr/index.ts b/open-sse/services/compression/engines/ccr/index.ts index e52755129f..e854666182 100644 --- a/open-sse/services/compression/engines/ccr/index.ts +++ b/open-sse/services/compression/engines/ccr/index.ts @@ -37,6 +37,13 @@ */ import crypto from "node:crypto"; +import { + deleteAllCcrBlocks, + deleteCcrBlockRow, + loadCcrBlock, + persistCcrBlock, + touchCcrBlock, +} from "../../../../../src/lib/db/ccrBlocks.ts"; import { createCompressionStats } from "../../stats.ts"; import { queryBlock, type CcrQuery } from "./ccrQuery.ts"; import { injectCcrProtocolInstruction } from "./protocolInstruction.ts"; @@ -156,6 +163,150 @@ function buildStoreKey(hash: string, principalId?: string): string { return `${principalId ?? ANON} ${hash}`; } +// ─── durable second tier (#9061) ────────────────────────────────────────────── + +/** + * The map above is the hot cache. It loses entries to cross-principal LRU eviction, to + * the TTL, to restarts, and to a retrieve landing on another instance, while + * `fidelityGateStep` waives fidelity checks for sampling engines on the grounds that + * their drop is "CCR-recoverable", and the protocol instruction promises the model a + * verbatim block. These helpers put the block on disk so that promise survives. + * + * Every one of them is best-effort: a store without a usable database (compression + * preview, unit tests, a read-only volume) degrades to today's in-memory behaviour + * rather than failing the request. + * + * Three guards keep this from changing what the deployment stores at rest more than it + * has to. They follow the call-log artifact path, which faced the same question: + * + * 1. Blocks over `MAX_DURABLE_BLOCK_BYTES` stay memory-only. `MAX_CCR_BLOCK_BYTES` is + * 2 MB, and 5,000 of those would be 10 GB of prompt text in SQLite. 512 KB is the + * ceiling #1647 already set on call artifacts for this exact reason. + * 2. No durable tier on a cloud runtime, which has no local disk to write to. + * 3. `COMPRESSION_CCR_DURABLE_STORE=false` turns it off. The content is prompt text, and an + * operator who does not want that on disk needs a switch that is not a rebuild. + * + * The switch defaults to on because the model is already told, by the CCR protocol + * instruction, that it can retrieve the block verbatim. Leaving it off by default would + * keep that promise hollow for everyone who never reads this file. + */ +const MAX_DURABLE_BLOCK_BYTES = 512 * 1024; + +/** Matches the detection call-log artifacts use (`callLogArtifacts.ts`). */ +const isCloudRuntime = typeof globalThis.caches === "object" && globalThis.caches !== null; + +function durableTierEnabled(): boolean { + return !isCloudRuntime && process.env.COMPRESSION_CCR_DURABLE_STORE !== "false"; +} + +const loggedDurableErrors = new Set(); + +function warnDurableError(operation: string, error: unknown): void { + if (process.env.NODE_ENV === "test") return; + if (loggedDurableErrors.has(operation)) return; + if (loggedDurableErrors.size >= 20) { + const first = loggedDurableErrors.values().next().value; + if (first !== undefined) loggedDurableErrors.delete(first); + } + loggedDurableErrors.add(operation); + const message = error instanceof Error ? error.message : String(error); + console.warn(`[ccr] durable ${operation} failed: ${message}`); +} + +/** + * Writes are deferred off the request path. Measured on this repo, a synchronous + * `persistCcrBlock` costs 0.032 ms at the 600-char minimum block but 1.77 ms at 500 KB, + * past the 1 ms this engine declares in its metadata, and roughly 7x the sha256 it + * already pays over the same bytes. The block is in the map before the defer runs, so an + * in-process retrieve never waits for the disk; only a crash inside that tick loses the + * durable copy, and the client's next request re-stores it under the same hash. + * + * Persist and delete share this queue so they cannot reorder: `setImmediate` is FIFO, and + * a delete that overtook its own persist would resurrect the block it just removed. + */ +const MAX_PENDING_DURABLE_WRITES = 1_000; +let pendingDurableWrites = 0; +let droppedDurableWrites = 0; + +function deferDurable(operation: string, work: () => void, droppable = false): void { + // Backpressure. A burst faster than SQLite drains would otherwise queue without bound + // and hold every block's content live in the closure. Dropping a persist is safe: the + // block is still in the map, and the client re-stores it under the same hash on its + // next request. Deletes are never dropped, or a deleted block would come back. + if (droppable && pendingDurableWrites >= MAX_PENDING_DURABLE_WRITES) { + droppedDurableWrites++; + return; + } + pendingDurableWrites++; + setImmediate(() => { + pendingDurableWrites--; + try { + work(); + } catch (error) { + warnDurableError(operation, error); + } + }); +} + +function persistEntry(entry: CcrEntry): void { + if (!durableTierEnabled()) return; + if (entry.bytes > MAX_DURABLE_BLOCK_BYTES) return; + const snapshot = { ...entry }; + deferDurable("persist", () => persistCcrBlock(snapshot), true); +} + +function forgetEntry(hash: string, principalId: string): void { + if (!durableTierEnabled()) return; + deferDurable("delete", () => deleteCcrBlockRow(principalId, hash)); +} + +/** + * Read a block the map no longer holds and put it back in the map, so the LRU/byte + * accounting keeps working from there. Returns null when there is no durable row, which + * is also what a missing database looks like. + */ +function rehydrateEntry(hash: string, principalId: string, now: number): CcrEntry | null { + if (!durableTierEnabled()) return null; + let row: ReturnType; + try { + row = loadCcrBlock(principalId, hash, now); + } catch (error) { + warnDurableError("load", error); + return null; + } + if (!row) return null; + + const entry: CcrEntry = { + hash: row.hash, + principalId: row.principalId, + content: row.content, + bytes: row.bytes, + chars: row.chars, + lines: row.lines, + contentType: row.contentType, + source: row.source as CcrEntrySource, + createdAt: row.createdAt, + lastAccessedAt: now, + expiresAt: row.expiresAt, + }; + + // Re-admit through the same budgets a fresh store would face. If the block no longer + // fits, it stays on disk and is served straight from the row instead of being cached. + if (enforcePrincipalBudget(entry.principalId, entry.bytes) && enforceGlobalBudget(entry.bytes)) { + const key = buildStoreKey(hash, principalId === ANON ? undefined : principalId); + ccrStore.set(key, entry); + ccrTotalBytes += entry.bytes; + principalBytesMap.set(entry.principalId, principalBytes(entry.principalId) + entry.bytes); + } + + try { + touchCcrBlock(entry.principalId, hash, now); + } catch (error) { + warnDurableError("touch", error); + } + return entry; +} + function readLifecycleCounters(principalId: string): CcrLifecycleCounters { return ( lifecycleByPrincipal.get(principalId) ?? { @@ -203,7 +354,12 @@ function removeEntry(key: string, reason?: "expired" | "capacity"): boolean { if (remainingPrincipalBytes === 0) principalBytesMap.delete(entry.principalId); else principalBytesMap.set(entry.principalId, remainingPrincipalBytes); const counters = mutableLifecycleCounters(entry.principalId); - if (reason === "expired") counters.expiredEvictions++; + if (reason === "expired") { + counters.expiredEvictions++; + // Expiry is the one eviction that means the block is finished. Capacity eviction is + // not: that block stays on disk, which is the point of the durable tier (#9061). + forgetEntry(entry.hash, entry.principalId); + } if (reason === "capacity") counters.capacityEvictions++; return true; } @@ -350,6 +506,7 @@ export function tryStoreBlock( ccrStore.set(key, entry); ccrTotalBytes += bytes; principalBytesMap.set(owner, principalBytes(owner) + bytes); + persistEntry(entry); return { stored: true, hash, metadata: publicMetadata(entry) }; } @@ -372,7 +529,12 @@ export function storeBlock( export function retrieveBlock(hash: string, principalId?: string, now = Date.now()): string | null { const key = buildStoreKey(hash, principalId); const entry = getActiveEntry(key, now); - if (!entry) return null; + if (!entry) { + // Miss in the hot cache is not proof the block is gone (#9061): LRU eviction, the TTL + // sweep, a restart, or another instance all land here while the row is still on disk. + const restored = rehydrateEntry(hash, principalId ?? ANON, now); + return restored ? restored.content : null; + } entry.lastAccessedAt = now; ccrStore.delete(key); ccrStore.set(key, entry); @@ -435,6 +597,14 @@ export function resetCcrStore(): void { principalBytesMap.clear(); ccrTotalBytes = 0; lifecycleByPrincipal.clear(); + // Through the same queue as persist/delete, so a reset cannot overtake a write it was + // meant to clear. + deferDurable("reset", deleteAllCcrBlocks); +} + +/** Resolves once the deferred durable writes queued so far have run. */ +export function flushCcrDurableWrites(): Promise { + return new Promise((resolve) => setImmediate(resolve)); } export function inspectCcrBlock( @@ -469,7 +639,11 @@ export function listCcrBlocks( } export function deleteCcrBlock(hash: string, principalId?: string, _now = Date.now()): boolean { - return removeEntry(buildStoreKey(hash, principalId)); + const removedFromCache = removeEntry(buildStoreKey(hash, principalId)); + // An explicit delete must reach the durable tier too, otherwise the next retrieve + // rehydrates the block the caller just deleted. + forgetEntry(hash, principalId ?? ANON); + return removedFromCache; } export function getCcrStoreStats(principalId?: string, now = Date.now()): CcrStoreStats { diff --git a/src/lib/db/ccrBlocks.ts b/src/lib/db/ccrBlocks.ts new file mode 100644 index 0000000000..56d49c131a --- /dev/null +++ b/src/lib/db/ccrBlocks.ts @@ -0,0 +1,143 @@ +/** + * Durable second tier for the CCR block store (#9061). + * + * The CCR engine's in-process `Map` stays the hot cache; this module is what makes its + * promise survive an eviction, a TTL sweep, a restart, or a retrieve that lands on + * another instance. Every function here is best-effort: the caller treats a throw as + * "not durable this time", never as a request failure. + */ +import { getDbInstance } from "./core"; + +export interface CcrBlockRow { + principalId: string; + hash: string; + content: string; + bytes: number; + chars: number; + lines: number; + contentType: string; + source: string; + createdAt: number; + lastAccessedAt: number; + expiresAt: number; +} + +interface CcrBlockDbRow { + content: string; + bytes: number; + chars: number; + lines: number; + content_type: string; + source: string; + created_at: number; + last_accessed_at: number; + expires_at: number; +} + +/** + * Writes older than this are not worth a prune scan on every call. + * ponytail: counter-based throttle, swap for a scheduled sweep if the table ever grows + * fast enough that 200 writes of drift matters. + */ +const PRUNE_EVERY_N_WRITES = 200; +let writesSincePrune = 0; + +export function persistCcrBlock(row: CcrBlockRow): void { + const db = getDbInstance(); + db.prepare( + `INSERT OR REPLACE INTO ccr_blocks ( + principal_id, hash, content, bytes, chars, lines, + content_type, source, created_at, last_accessed_at, expires_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` + ).run( + row.principalId, + row.hash, + row.content, + row.bytes, + row.chars, + row.lines, + row.contentType, + row.source, + row.createdAt, + row.lastAccessedAt, + row.expiresAt + ); + + if (++writesSincePrune >= PRUNE_EVERY_N_WRITES) { + writesSincePrune = 0; + pruneExpiredCcrBlocks(Date.now()); + } +} + +/** Returns the block only while it is unexpired; an expired row is deleted and read as a miss. */ +export function loadCcrBlock(principalId: string, hash: string, now: number): CcrBlockRow | null { + const db = getDbInstance(); + const row = db + .prepare( + `SELECT content, bytes, chars, lines, content_type, source, + created_at, last_accessed_at, expires_at + FROM ccr_blocks WHERE principal_id = ? AND hash = ?` + ) + .get(principalId, hash) as CcrBlockDbRow | undefined; + + if (!row) return null; + if (row.expires_at <= now) { + deleteCcrBlockRow(principalId, hash); + return null; + } + + return { + principalId, + hash, + content: row.content, + bytes: row.bytes, + chars: row.chars, + lines: row.lines, + contentType: row.content_type, + source: row.source, + createdAt: row.created_at, + lastAccessedAt: row.last_accessed_at, + expiresAt: row.expires_at, + }; +} + +export function touchCcrBlock(principalId: string, hash: string, lastAccessedAt: number): void { + getDbInstance() + .prepare(`UPDATE ccr_blocks SET last_accessed_at = ? WHERE principal_id = ? AND hash = ?`) + .run(lastAccessedAt, principalId, hash); +} + +export function deleteCcrBlockRow(principalId: string, hash: string): void { + getDbInstance() + .prepare(`DELETE FROM ccr_blocks WHERE principal_id = ? AND hash = ?`) + .run(principalId, hash); +} + +/** + * Drops every durable block. Mirrors the engine's `resetCcrStore()` so that "reset" keeps + * meaning reset once the store has a second tier. + */ +export function deleteAllCcrBlocks(): void { + getDbInstance().prepare(`DELETE FROM ccr_blocks`).run(); +} + +/** Drops every block whose TTL has passed. Returns how many rows went. */ +export function pruneExpiredCcrBlocks(now: number): number { + const result = getDbInstance().prepare(`DELETE FROM ccr_blocks WHERE expires_at <= ?`).run(now); + return result.changes ?? 0; +} + +export function countCcrBlocks(principalId?: string): number { + const db = getDbInstance(); + const row = ( + principalId === undefined + ? db.prepare(`SELECT COUNT(*) AS n FROM ccr_blocks`).get() + : db.prepare(`SELECT COUNT(*) AS n FROM ccr_blocks WHERE principal_id = ?`).get(principalId) + ) as { n: number } | undefined; + return row?.n ?? 0; +} + +/** Test seam: the write throttle is module state and has to be resettable between tests. */ +export function resetCcrBlockPruneCounter(): void { + writesSincePrune = 0; +} diff --git a/src/lib/db/cleanup.ts b/src/lib/db/cleanup.ts index fc866800b3..20d0358ce4 100644 --- a/src/lib/db/cleanup.ts +++ b/src/lib/db/cleanup.ts @@ -370,6 +370,29 @@ export async function cleanupCompressionRunTelemetry(): Promise { return result; } +/** + * Clean up expired CCR blocks (#9061). + * + * Unlike the tables above, these rows carry their own expiry: the engine writes + * `expires_at` from the block's TTL, so this needs no retention-days setting of its own. + * It is the same sweep the engine does opportunistically, run on the operator's schedule + * so the table cannot sit on rows nobody will read again. + */ +export async function cleanupCcrBlocks(): Promise { + const result: CleanupResult = { deleted: 0, errors: 0 }; + + try { + const { pruneExpiredCcrBlocks } = await import("./ccrBlocks"); + result.deleted = pruneExpiredCcrBlocks(Date.now()); + console.log(`[Cleanup] Deleted ${result.deleted} expired ccr_blocks`); + } catch (err: unknown) { + console.error("[Cleanup] Error cleaning ccr_blocks:", err); + result.errors++; + } + + return result; +} + /** * Run all cleanup functions if auto-cleanup is enabled. */ @@ -401,6 +424,7 @@ export async function runAutoCleanup(): Promise<{ xpAuditLog: await cleanupXpAuditLog(), compressionRunTelemetry: await cleanupCompressionRunTelemetry(), proxyLogs: await cleanupProxyLogs(), + ccrBlocks: await cleanupCcrBlocks(), }; const totalDeleted = Object.values(results).reduce((sum, r) => sum + r.deleted, 0); diff --git a/src/lib/db/migrations/134_ccr_blocks.sql b/src/lib/db/migrations/134_ccr_blocks.sql new file mode 100644 index 0000000000..a1dd4b87cb --- /dev/null +++ b/src/lib/db/migrations/134_ccr_blocks.sql @@ -0,0 +1,31 @@ +-- CCR durable block store (#9061). +-- +-- The CCR engine keeps blocks in a process-local Map. That map loses entries to +-- cross-principal LRU eviction (5,000 entries / 64 MB global cap), to the 24h TTL, to +-- process restarts, and to multi-instance deployments where the MCP retrieve call lands +-- on an instance that never ran the compression. Meanwhile `fidelityGateStep.ts` waives +-- fidelity verification for sampling engines *because* their drop is "CCR-recoverable", +-- and the model is told by the CCR protocol instruction that it can retrieve the block +-- verbatim. +-- +-- This table is the second tier behind that Map: the Map stays the hot cache and keeps +-- its LRU/byte-budget behaviour untouched, and a block evicted from it is still readable +-- here. Scoping matches the in-memory store's compound key (principal + content hash). + +CREATE TABLE IF NOT EXISTS ccr_blocks ( + principal_id TEXT NOT NULL, + hash TEXT NOT NULL, + content TEXT NOT NULL, + bytes INTEGER NOT NULL, + chars INTEGER NOT NULL, + lines INTEGER NOT NULL, + content_type TEXT NOT NULL DEFAULT 'text/plain', + source TEXT NOT NULL DEFAULT 'compression', + created_at INTEGER NOT NULL, + last_accessed_at INTEGER NOT NULL, + expires_at INTEGER NOT NULL, + PRIMARY KEY (principal_id, hash) +); + +-- Pruning scans by expiry. +CREATE INDEX IF NOT EXISTS idx_ccr_blocks_expires ON ccr_blocks(expires_at); diff --git a/src/lib/localDb.ts b/src/lib/localDb.ts index 5ec050fbf3..1da3925b95 100755 --- a/src/lib/localDb.ts +++ b/src/lib/localDb.ts @@ -89,6 +89,7 @@ export { reorderCombos, deleteCombo, } from "./db/combos"; +export * from "./db/ccrBlocks"; export * from "./db/compressionCacheStats"; export * from "./db/compressionCombos"; export * from "./db/compressionContextBudget"; diff --git a/tests/unit/ccr-durable-store-9061.test.ts b/tests/unit/ccr-durable-store-9061.test.ts new file mode 100644 index 0000000000..dec112749e --- /dev/null +++ b/tests/unit/ccr-durable-store-9061.test.ts @@ -0,0 +1,164 @@ +/** + * #9061. The CCR store's in-process Map loses blocks to LRU eviction, the TTL, a + * restart, or a retrieve landing on another instance, while `fidelityGateStep` waives + * fidelity checks for sampling engines because their drop is "CCR-recoverable" and the + * protocol instruction promises the model a verbatim block. + * + * The regression test is the restart: a fresh module instance has an empty Map, so a + * block stored before it must come back from the durable tier or not at all. + */ +import { describe, it, before, beforeEach, after } from "node:test"; +import assert from "node:assert/strict"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +const tmpDir = mkdtempSync(join(tmpdir(), "omniroute-ccr-9061-")); +process.env.DATA_DIR = tmpDir; + +const core = await import("../../src/lib/db/core.ts"); +core.resetDbInstance(); + +const { + persistCcrBlock, + loadCcrBlock, + deleteCcrBlockRow, + deleteAllCcrBlocks, + pruneExpiredCcrBlocks, + countCcrBlocks, +} = await import("../../src/lib/db/ccrBlocks.ts"); + +const ccrPath = "../../open-sse/services/compression/engines/ccr/index.ts"; +const ccr = await import(ccrPath); + +const HOUR = 60 * 60 * 1000; + +function row(overrides: Record = {}) { + const now = 1_000_000; + return { + principalId: "principal-a", + hash: "aaaaaaaaaaaaaaaaaaaaaaaa", + content: "the verbatim block", + bytes: 18, + chars: 18, + lines: 1, + contentType: "text/plain", + source: "compression", + createdAt: now, + lastAccessedAt: now, + expiresAt: now + HOUR, + ...overrides, + }; +} + +describe("ccrBlocks durable store (#9061)", () => { + beforeEach(() => { + deleteAllCcrBlocks(); + }); + + after(() => { + core.resetDbInstance(); + }); + + it("round-trips a block", () => { + persistCcrBlock(row()); + const loaded = loadCcrBlock("principal-a", "aaaaaaaaaaaaaaaaaaaaaaaa", 1_000_000); + assert.equal(loaded?.content, "the verbatim block"); + assert.equal(loaded?.contentType, "text/plain"); + assert.equal(loaded?.source, "compression"); + }); + + it("scopes by principal, so another cannot read the block", () => { + persistCcrBlock(row()); + assert.equal(loadCcrBlock("principal-b", "aaaaaaaaaaaaaaaaaaaaaaaa", 1_000_000), null); + }); + + it("reads an expired block as a miss and drops the row", () => { + persistCcrBlock(row({ expiresAt: 1_000_000 })); + assert.equal(loadCcrBlock("principal-a", "aaaaaaaaaaaaaaaaaaaaaaaa", 1_000_001), null); + assert.equal(countCcrBlocks(), 0); + }); + + it("prunes only what has expired", () => { + persistCcrBlock(row({ hash: "a".repeat(24), expiresAt: 500 })); + persistCcrBlock(row({ hash: "b".repeat(24), expiresAt: 9_000_000 })); + assert.equal(pruneExpiredCcrBlocks(1_000_000), 1); + assert.equal(countCcrBlocks(), 1); + }); + + it("deletes a single block", () => { + persistCcrBlock(row()); + deleteCcrBlockRow("principal-a", "aaaaaaaaaaaaaaaaaaaaaaaa"); + assert.equal(countCcrBlocks(), 0); + }); +}); + +describe("CCR engine survives losing its in-memory map (#9061)", () => { + before(() => { + ccr.resetCcrStore(); + }); + + after(() => { + ccr.resetCcrStore(); + core.resetDbInstance(); + }); + + it("retrieves a block from a fresh module instance, the way a restart sees it", async () => { + const text = "x".repeat(2_000); + const stored = ccr.tryStoreBlock(text, "principal-a"); + assert.equal(stored.stored, true); + await ccr.flushCcrDurableWrites(); + + // A fresh instance of the engine module has its own empty Map, the same state the + // process has after a restart, and the same state another instance starts in. + const restarted = await import(`${ccrPath}?restart=9061`); + assert.equal( + restarted.retrieveBlock(stored.hash, "principal-a"), + text, + "block must come back from the durable tier after the in-memory map is gone" + ); + }); + + it("keeps the principal boundary across the restart", async () => { + const text = "y".repeat(2_000); + const stored = ccr.tryStoreBlock(text, "principal-a"); + await ccr.flushCcrDurableWrites(); + const restarted = await import(`${ccrPath}?restart=9061-scope`); + assert.equal(restarted.retrieveBlock(stored.hash, "principal-b"), null); + }); + + it("keeps an oversized block memory-only", async () => { + // 512KB is the ceiling call artifacts already use (#1647). Above it the block still + // works from the map, it just never reaches disk. + const text = "b".repeat(600 * 1024); + const stored = ccr.tryStoreBlock(text, "principal-a"); + assert.equal(stored.stored, true); + await ccr.flushCcrDurableWrites(); + + assert.equal(ccr.retrieveBlock(stored.hash, "principal-a"), text, "map still serves it"); + assert.equal(loadCcrBlock("principal-a", stored.hash, Date.now()), null, "but disk has no row"); + }); + + it("writes nothing when COMPRESSION_CCR_DURABLE_STORE is false", async () => { + const previous = process.env.COMPRESSION_CCR_DURABLE_STORE; + process.env.COMPRESSION_CCR_DURABLE_STORE = "false"; + try { + const stored = ccr.tryStoreBlock("c".repeat(2_000), "principal-a"); + await ccr.flushCcrDurableWrites(); + assert.equal(loadCcrBlock("principal-a", stored.hash, Date.now()), null); + } finally { + if (previous === undefined) delete process.env.COMPRESSION_CCR_DURABLE_STORE; + else process.env.COMPRESSION_CCR_DURABLE_STORE = previous; + } + }); + + it("does not resurrect a block that was explicitly deleted", async () => { + const text = "z".repeat(2_000); + const stored = ccr.tryStoreBlock(text, "principal-a"); + ccr.deleteCcrBlock(stored.hash, "principal-a"); + // Persist and delete share one FIFO queue; one flush covers both. + await ccr.flushCcrDurableWrites(); + const restarted = await import(`${ccrPath}?restart=9061-deleted`); + assert.equal(restarted.retrieveBlock(stored.hash, "principal-a"), null); + }); +});