From b44dd81dc9eae4bac83899aa895d0ebd24e677ec Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 2 Nov 2026 03:37:58 +0800 Subject: [PATCH 1/4] perf(console): keep directory serving and skip no-op mirror writes --- tests/kv_directory_validation.mjs | 80 +++++++++++++++++++++ web/strategy-switch-console/kv-directory.md | 2 +- web/strategy-switch-console/kv_directory.js | 46 +++++++++--- 3 files changed, 118 insertions(+), 10 deletions(-) diff --git a/tests/kv_directory_validation.mjs b/tests/kv_directory_validation.mjs index 9d0c3a33..f4a6f011 100644 --- a/tests/kv_directory_validation.mjs +++ b/tests/kv_directory_validation.mjs @@ -74,6 +74,86 @@ assert.equal(pages, 2); assert.equal((await paged.command({ action: "read", prefix: "control_plane_source:", limit: 100 })).keys.length, 3, "a publisher between migration pages is retained"); +// A reconciliation re-scan of a complete directory keeps serving the last +// completed snapshot while it pages, and skips KV mirror writes on pages +// where nothing mirror-visible changed. +const reconcileValues = new Map(), reconcileState = new Map(); +let reconcileLists = 0, reconcileMirrorPuts = 0, reconcilePaged = false; +const reconcileTime = { value: 1_800_000_000_000 }; +const reconcile = new KvDirectory({ + async get(key) { return structuredClone(reconcileState.get(key)); }, + async put(key, value) { reconcileState.set(key, structuredClone(value)); }, +}, { + async get(key) { return reconcileValues.get(key) ?? null; }, + async put(key, value) { reconcileMirrorPuts++; reconcileValues.set(key, value); }, + async list(options) { + reconcileLists++; + if (!reconcilePaged) return { keys: [{ name: "control_plane_source:known" }, + { name: "control_plane_source:out-of-band" }], list_complete: true }; + if (!options.cursor) return { keys: [{ name: "control_plane_source:late" }], + list_complete: false, cursor: "page-2" }; + if (options.cursor === "page-2") return { keys: [{ name: "control_plane_source:known" }], + list_complete: false, cursor: "page-3" }; + return { keys: [], list_complete: true }; + }, +}, {}, () => reconcileTime.value); +for (let i = 0; i < 3; i++) await reconcile.command({ action: "put", key: `control_plane_source:known-${i}`, value: "{}" }); +assert.equal((await reconcile.command({ action: "maintain" })).complete, true); +const migrationScannedAt = (await reconcile.command({ action: "status" })).data_as_of; +const reconcileMirrorPutsAfterMigration = reconcileMirrorPuts; +reconcilePaged = true; +reconcileTime.value += 21_601_000; // past the configured reconcile interval +assert.equal((await reconcile.command({ action: "maintain" })).ok, true, "page 1 of a re-scan succeeds"); +assert.equal((await reconcile.command({ action: "read", prefix: "control_plane_source:", limit: 100 })).ok, true, + "reads keep working between re-scan pages"); +assert.equal((await reconcile.command({ action: "read", prefix: "control_plane_source:", limit: 100 })).keys.length, 6, + "the served snapshot includes keys discovered by the page in flight"); +assert.equal(reconcileMirrorPuts, reconcileMirrorPutsAfterMigration + 1, + "a page that discovers new keys mirrors them immediately"); +const listsAfterFirstPage = reconcileLists; +assert.deepEqual(await reconcile.command({ action: "maintain" }), + { ok: true, complete: true, count: 6, scanned_at: migrationScannedAt, cached: true }, + "a second invocation inside the page delay reuses the in-flight snapshot"); +assert.equal(reconcileLists, listsAfterFirstPage, "the page-delay window must not list again"); +reconcileTime.value += 61_000; +await reconcile.command({ action: "maintain" }); +assert.equal(reconcileMirrorPuts, reconcileMirrorPutsAfterMigration + 1, + "a page that adds no mirror-visible change must not rewrite the mirror"); +reconcileTime.value += 61_000; +assert.equal((await reconcile.command({ action: "maintain" })).ok, true); +assert.equal((await reconcile.command({ action: "status" })).complete, true); +assert.equal((await reconcile.command({ action: "status" })).data_as_of, reconcileTime.value, + "the re-scan stamps scanned_at only when the last page completes"); +assert.equal(reconcileMirrorPuts, reconcileMirrorPutsAfterMigration + 2, + "the completion page mirrors the new scanned_at exactly once"); + +// A failed re-scan keeps the loud failure contract: the directory reports +// the error, refuses reads and stays on the daily quota cooldown. +const quotaReconcileValues = new Map(), quotaReconcileState = new Map(); +let quotaReconcileLists = 0, quotaReconcileFailing = false; +const quotaReconcileTime = { value: 1_800_000_000_000 }; +const quotaReconcile = new KvDirectory({ + async get(key) { return structuredClone(quotaReconcileState.get(key)); }, + async put(key, value) { quotaReconcileState.set(key, structuredClone(value)); }, +}, { + async get(key) { return quotaReconcileValues.get(key) ?? null; }, + async put(key, value) { quotaReconcileValues.set(key, value); }, + async list() { quotaReconcileLists++; if (quotaReconcileFailing) throw Error("KV list() limit exceeded for the day."); + return { keys: [...quotaReconcileValues.keys()].map(name => ({ name })), list_complete: true }; }, +}, {}, () => quotaReconcileTime.value); +await quotaReconcile.command({ action: "put", key: "research_promotion_ticket:seeded", value: "{}" }); +assert.equal((await quotaReconcile.command({ action: "maintain" })).complete, true); +quotaReconcileTime.value += 21_601_000; quotaReconcileFailing = true; +assert.equal((await quotaReconcile.command({ action: "maintain" })).error, "kv_list_daily_limit_exceeded"); +assert.equal((await quotaReconcile.command({ action: "read", prefix: "research_promotion_ticket:", limit: 100 })).ok, false, + "a failed re-scan still refuses reads until the cooldown opens"); +const quotaReconcileStatus = await quotaReconcile.command({ action: "status" }); +assert.equal(quotaReconcileStatus.complete, false); +assert.equal(quotaReconcileStatus.last_error.code, "kv_list_daily_limit_exceeded"); +const quotaReconcileListsAfterFailure = quotaReconcileLists; +for (let i = 0; i < 100; i++) await quotaReconcile.command({ action: "maintain" }); +assert.equal(quotaReconcileLists, quotaReconcileListsAfterFailure, "the daily quota cooldown spans repeated cron ticks"); + const redacted = new KvDirectory({ async get() { return { keys: ["control_plane_source:private-subject"], complete: false, cursor: "private-cursor", retry_at: 0, error: "private-provider-error", last_error: { code: "private-token-detail", at: time }, owner: "private-owner" }; }, diff --git a/web/strategy-switch-console/kv-directory.md b/web/strategy-switch-console/kv-directory.md index cf5f8380..4a44c780 100644 --- a/web/strategy-switch-console/kv-directory.md +++ b/web/strategy-switch-console/kv-directory.md @@ -18,7 +18,7 @@ The existing `STRATEGY_SWITCH_RUNTIME_INSTANCES` binding also hosts a separate o All seven source queues and daily history read this maintained manifest. Source/daily publishers register each key durably before writing its existing KV record, under one serialized object queue. New keys update the mirror after the record write; a failed mirror write leaves a persisted dirty marker for retry. Records, freshness, binding checks and decision authority are still read through their original paths. A listed record that cannot be read remains a source error. An incomplete manifest never means an empty queue; unknown daily history is `null`. -Existing keys require a complete migration scan. A 0% Worker override does not select the Durable Object version. If the private receiver returns the exact legacy actor-required rejection, reads use the KV mirror and the explicit authenticated admin operation can perform one bounded migration page directly. No background or read-path fallback scans are allowed; a real object outage does not enable this fallback. The new object adopts that mirror on activation. During this bootstrap, invoke admin maintenance once and inspect its result; the KV mirror alone does not provide cross-region serialization against simultaneous operator requests. Only the existing five-minute cron or an authenticated same-origin admin POST to `/api/admin/kv-directory` can run maintenance. Ordinary reads and acceptance never trigger scans. Each maintenance invocation reads at most one unfiltered KV page (1,000 keys), retaining its cursor. Pending migrations remain unavailable. Concurrent publishers are retained between pages. After success, periodic reconciliation covers legacy/out-of-band writers; records written through the current ingress are registered immediately. An admin POST requests a fresh scan, with a minimum 60-second successful-scan interval; its response includes `scanned_at` and whether a recent scan was reused. During a mixed-version candidate test, reconcile after the last legacy writer before relying on directory coverage. No index can infer keys that were never discovered or registered. +Existing keys require a complete migration scan. A 0% Worker override does not select the Durable Object version. If the private receiver returns the exact legacy actor-required rejection, reads use the KV mirror and the explicit authenticated admin operation can perform one bounded migration page directly. No background or read-path fallback scans are allowed; a real object outage does not enable this fallback. The new object adopts that mirror on activation. During this bootstrap, invoke admin maintenance once and inspect its result; the KV mirror alone does not provide cross-region serialization against simultaneous operator requests. Only the existing five-minute cron or an authenticated same-origin admin POST to `/api/admin/kv-directory` can run maintenance. Ordinary reads and acceptance never trigger scans. Each maintenance invocation reads at most one unfiltered KV page (1,000 keys), retaining its cursor. Pending migrations remain unavailable. A reconciliation re-scan of an already complete directory keeps serving the last completed snapshot while it pages; only the initial migration or a failed scan leaves the directory unavailable. Concurrent publishers are retained between pages. After success, periodic reconciliation covers legacy/out-of-band writers; records written through the current ingress are registered immediately. An admin POST requests a fresh scan, with a minimum 60-second successful-scan interval; its response includes `scanned_at` and whether a recent scan was reused. During a mixed-version candidate test, reconcile after the last legacy writer before relying on directory coverage. No index can infer keys that were never discovered or registered. A daily list-quota failure is persisted until the next UTC reset plus 60 seconds. Other migration errors use bounded exponential backoff. The old production version can still consume quota until promotion; installing a candidate cannot restore already-exhausted quota. Never seed a supposedly complete index from a few observed source IDs. If migration cannot complete, retain the explicit source outage. Apply the approved deployment criterion: no worse than the same-request baseline and zero page errors. A new unknown, truncated or mismatched directory is a regression and blocks promotion; an independently confirmed identical quota outage in both versions may remain visible under that criterion. diff --git a/web/strategy-switch-console/kv_directory.js b/web/strategy-switch-console/kv_directory.js index 3e0ed52c..dfe6839a 100644 --- a/web/strategy-switch-console/kv_directory.js +++ b/web/strategy-switch-console/kv_directory.js @@ -51,6 +51,22 @@ async function mirrorManifest(store, state) { await store.put(MANIFEST_KEY, JSON.stringify({ ...state, mirror_dirty: false, schema: SCHEMA })); } +// The mirror only carries what legacy receivers and a fresh object adoption +// actually read; cursor and failures are private pacing state. +function mirrorVisible(state) { + return { keys: state.keys, complete: state.complete, scanned_at: state.scanned_at, + error: state.error, last_error: state.last_error }; +} + +// Keys only ever grow and stay sorted, so an equal length means an equal list. +function mirrorStateChanged(before, state) { + return before.keys.length !== state.keys.length + || before.complete !== state.complete + || before.scanned_at !== state.scanned_at + || before.error !== state.error + || JSON.stringify(before.last_error) !== JSON.stringify(state.last_error); +} + function manifestRead(state, prefix, limit) { if (!state?.complete) { const error = new Error(state?.error || "kv_directory_uninitialized"); @@ -107,14 +123,22 @@ export class KvDirectory { } if (command.action === "maintain") { const interval = bounded(this.env.KV_DIRECTORY_RECONCILE_SECONDS, 21600, 3600, 86400) * 1000; - if (state.complete && ((!command.refresh && this.now() < state.scanned_at + interval) + // A re-scan of a complete directory pages behind the existing snapshot: + // keys are union-only, so serving the last completed list is never less + // safe than failing the read while the pages run. + if (state.complete && !state.cursor && ((!command.refresh && this.now() < state.scanned_at + interval) || this.now() < state.scanned_at + 60000)) return { ok: true, complete: true, count: state.keys.length, scanned_at: state.scanned_at, cached: true }; - if (state.retry_at > this.now()) return this.unavailable(state); - state.complete = false; + if (state.retry_at > this.now()) { + if (!state.complete) return this.unavailable(state); + // A scan is already in flight; reuse the snapshot it started from. + return { ok: true, complete: true, count: state.keys.length, scanned_at: state.scanned_at, cached: true }; + } + const completeBeforeScan = state.complete; + const mirrorBefore = mirrorVisible(state); try { - // Migration only: at most one page per maintenance invocation, shared - // across every prefix and every Worker isolate. Reads never call list. + // One page per maintenance invocation, shared across every prefix and + // every Worker isolate. Reads never call list. const listing = await this.store.list({ limit: 1000, ...(state.cursor ? { cursor: state.cursor } : {}) }); if (!Array.isArray(listing?.keys) || typeof listing.list_complete !== "boolean" || (!listing.list_complete && !listing.cursor)) throw Error("invalid_listing"); @@ -123,13 +147,17 @@ export class KvDirectory { const maximum = bounded(this.env.KV_DIRECTORY_MAX_KEYS, 10000, 100, 100000); if (keys.size > maximum || new TextEncoder().encode(JSON.stringify([...keys])).byteLength + 1024 > 120000) throw Error("directory_too_large"); state.keys = [...keys].sort(); - state.complete = listing.list_complete; - if (state.complete) state.scanned_at = this.now(); + state.complete = listing.list_complete || completeBeforeScan; + if (listing.list_complete) state.scanned_at = this.now(); state.cursor = listing.list_complete ? null : listing.cursor; state.failures = 0; state.error = state.complete ? null : "kv_directory_migration_pending"; - state.retry_at = state.complete ? 0 : this.now() + bounded(this.env.KV_DIRECTORY_PAGE_DELAY_SECONDS, 60, 30, 3600) * 1000; + state.retry_at = listing.list_complete ? 0 + : this.now() + bounded(this.env.KV_DIRECTORY_PAGE_DELAY_SECONDS, 60, 30, 3600) * 1000; } catch (error) { + // A failed scan keeps the loud contract: the directory reports the + // error and refuses reads until the retry window opens. + state.complete = false; state.failures += 1; const quota = error.message === "KV list() limit exceeded for the day."; state.error = quota ? "kv_list_daily_limit_exceeded" : "kv_directory_migration_failed"; @@ -139,7 +167,7 @@ export class KvDirectory { : this.now() + Math.min(3600000, delay * 1000 * 2 ** Math.min(state.failures - 1, 4)); } await this.storage.put(STATE_KEY, state); - await mirrorManifest(this.store, state); + if (!state.complete || mirrorStateChanged(mirrorBefore, state)) await mirrorManifest(this.store, state); return state.complete ? { ok: true, complete: true, count: state.keys.length, scanned_at: state.scanned_at } : this.unavailable(state); } if (command.action !== "read" || !indexedKey(command.prefix)) throw Error("kv_directory_invalid_request"); From 0d5d68f49e4e9aa12de0c249c689e862f93ba39f Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 2 Nov 2026 03:37:58 +0800 Subject: [PATCH 2/4] perf(console): clone current-strategy signature without JSON round-trip --- web/strategy-switch-console/worker.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/web/strategy-switch-console/worker.js b/web/strategy-switch-console/worker.js index 8c00c8cf..b5d877e7 100644 --- a/web/strategy-switch-console/worker.js +++ b/web/strategy-switch-console/worker.js @@ -6424,7 +6424,7 @@ let _memRefreshing = false; // prevent concurrent background refreshes // byte equality can never hold; the comparison drops those stamps and keeps the // stored value untouched for readers. function currentStrategiesSignature(payload) { - const clone = JSON.parse(JSON.stringify(payload ?? null)); + const clone = structuredClone(payload ?? null); const visit = (node) => { if (Array.isArray(node)) { node.forEach(visit); From 4fe6b9e7f986399ed62a4ae0a1f2a869608a667a Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 2 Nov 2026 04:02:59 +0800 Subject: [PATCH 3/4] perf(kv): bound directory scans and serialize cache writes --- tests/helpers/directory_worker.mjs | 11 +++- tests/kv_directory_validation.mjs | 57 +++++++++++++++++++- web/strategy-switch-console/kv-directory.md | 8 +-- web/strategy-switch-console/kv_directory.js | 60 +++++++++++++++++++-- web/strategy-switch-console/worker.js | 27 ++++------ 5 files changed, 135 insertions(+), 28 deletions(-) diff --git a/tests/helpers/directory_worker.mjs b/tests/helpers/directory_worker.mjs index a2e4ceaa..9d0e9bab 100644 --- a/tests/helpers/directory_worker.mjs +++ b/tests/helpers/directory_worker.mjs @@ -1,11 +1,20 @@ // Existing business fixtures model discovery directly. Real manifest lifecycle, // persistence and concurrency are exercised by kv_directory_validation.mjs. import worker, { __test as originalTest } from "../../web/strategy-switch-console/worker.js"; -import { installDirectoryTestAdapter } from "../../web/strategy-switch-console/kv_directory.js"; +import { KvDirectory, installDirectoryTestAdapter } from "../../web/strategy-switch-console/kv_directory.js"; +const cacheWriters = new WeakMap(); function prepare(env) { installDirectoryTestAdapter(env, async command => { const store = env.STRATEGY_SWITCH_CONFIG; + if (command.action === "cache") { + if (!cacheWriters.has(store)) { + const values = new Map(); + cacheWriters.set(store, new KvDirectory({ async get(k) { return values.get(k); }, + async put(k, v) { values.set(k, structuredClone(v)); } }, store)); + } + return cacheWriters.get(store).command(command); + } if (command.action === "put") { await store.put(command.key, command.value); return { ok: true }; diff --git a/tests/kv_directory_validation.mjs b/tests/kv_directory_validation.mjs index f4a6f011..1143a226 100644 --- a/tests/kv_directory_validation.mjs +++ b/tests/kv_directory_validation.mjs @@ -102,7 +102,7 @@ assert.equal((await reconcile.command({ action: "maintain" })).complete, true); const migrationScannedAt = (await reconcile.command({ action: "status" })).data_as_of; const reconcileMirrorPutsAfterMigration = reconcileMirrorPuts; reconcilePaged = true; -reconcileTime.value += 21_601_000; // past the configured reconcile interval +reconcileTime.value += 86_401_000; // past the configured reconcile interval assert.equal((await reconcile.command({ action: "maintain" })).ok, true, "page 1 of a re-scan succeeds"); assert.equal((await reconcile.command({ action: "read", prefix: "control_plane_source:", limit: 100 })).ok, true, "reads keep working between re-scan pages"); @@ -110,6 +110,8 @@ assert.equal((await reconcile.command({ action: "read", prefix: "control_plane_s "the served snapshot includes keys discovered by the page in flight"); assert.equal(reconcileMirrorPuts, reconcileMirrorPutsAfterMigration + 1, "a page that discovers new keys mirrors them immediately"); +assert.equal((await reconcile.command({ action: "status" })).in_progress, true, + "a completed snapshot with a pending reconciliation cursor is still in progress"); const listsAfterFirstPage = reconcileLists; assert.deepEqual(await reconcile.command({ action: "maintain" }), { ok: true, complete: true, count: 6, scanned_at: migrationScannedAt, cached: true }, @@ -127,6 +129,32 @@ assert.equal((await reconcile.command({ action: "status" })).data_as_of, reconci assert.equal(reconcileMirrorPuts, reconcileMirrorPutsAfterMigration + 2, "the completion page mirrors the new scanned_at exactly once"); +// One synthetic day of five-minute cron ticks spends one List and mirror write. +const dailyState = new Map(); let dailyLists = 0, dailyPuts = 0; +let dailyNow = Date.parse("2026-11-01T00:02:00Z"); +const dailyStorage = { async get(k) { return structuredClone(dailyState.get(k)); }, + async put(k, v) { dailyState.set(k, structuredClone(v)); } }; +const dailyKv = { async get() { return null; }, async put() { dailyPuts++; }, + async list() { dailyLists++; return { keys: [], list_complete: true }; } }; +for (let tick = 0; tick < 288; tick++) { + await new KvDirectory(dailyStorage, dailyKv, {}, () => dailyNow).command({ action: "maintain" }); + dailyNow += 300_000; +} +assert.equal(dailyLists, 1, "default reconciliation cuts daily lists from four to one"); +assert.equal(dailyPuts, 1, "default reconciliation cuts daily mirror writes from four to one"); +// Explicit refreshes and object restarts share a persisted 100-attempt UTC budget. +for (let tick = 0; tick < 120; tick++) { + dailyNow = Date.parse("2026-11-01T01:00:00Z") + tick * 61_000; + await new KvDirectory(dailyStorage, dailyKv, {}, () => dailyNow).command({ action: "maintain", refresh: true }); +} +assert.equal(dailyLists, 100, "manual refresh cannot exhaust the account's 1000-list allowance"); +const budgetStatus = await new KvDirectory(dailyStorage, dailyKv, {}, () => dailyNow).command({ action: "status" }); +assert.equal(budgetStatus.last_error.code, "kv_directory_list_budget_exceeded"); +assert.equal(budgetStatus.retry_at, Date.parse("2026-11-02T00:01:00Z")); +dailyNow = budgetStatus.retry_at; +assert.equal((await new KvDirectory(dailyStorage, dailyKv, {}, () => dailyNow).command({ action: "maintain" })).ok, true); +assert.equal(dailyLists, 101, "the UTC reset opens a new budget"); + // A failed re-scan keeps the loud failure contract: the directory reports // the error, refuses reads and stays on the daily quota cooldown. const quotaReconcileValues = new Map(), quotaReconcileState = new Map(); @@ -143,7 +171,7 @@ const quotaReconcile = new KvDirectory({ }, {}, () => quotaReconcileTime.value); await quotaReconcile.command({ action: "put", key: "research_promotion_ticket:seeded", value: "{}" }); assert.equal((await quotaReconcile.command({ action: "maintain" })).complete, true); -quotaReconcileTime.value += 21_601_000; quotaReconcileFailing = true; +quotaReconcileTime.value += 86_401_000; quotaReconcileFailing = true; assert.equal((await quotaReconcile.command({ action: "maintain" })).error, "kv_list_daily_limit_exceeded"); assert.equal((await quotaReconcile.command({ action: "read", prefix: "research_promotion_ticket:", limit: 100 })).ok, false, "a failed re-scan still refuses reads until the cooldown opens"); @@ -237,3 +265,28 @@ assert.equal((await worker.fetch(new Request("https://fixture.invalid/api/admin/ const source = await readFile(new URL("../web/strategy-switch-console/worker.js", import.meta.url), "utf8"); assert.doesNotMatch(source, /store\.list\(/, "every production read path must use the manifest"); console.log("kv_directory_validation: PASS (100 cold reads/100 cooldown ticks/100 warm reads)"); + +// Cold isolates and object restarts share the cache writer's semantic comparison. +const cacheMetadata = new Map(), cacheValues = new Map(); let cachePuts = 0; +const cacheStorage = { async get(k) { return structuredClone(cacheMetadata.get(k)); }, + async put(k, v) { cacheMetadata.set(k, structuredClone(v)); } }; +const cacheKv = { async get(k) { return cacheValues.get(k) ?? null; }, + async put(k, v) { cachePuts++; cacheValues.set(k, v); } }; +const cachePayload = stamp => ({ account: { execution_evidence: { observed_at: stamp }, balance: 10 } }); +let cacheDirectory = new KvDirectory(cacheStorage, cacheKv); +await Promise.all(Array.from({ length: 100 }, (_, i) => cacheDirectory.command({ + action: "cache", payload: cachePayload(String(i)), ts: 1_800_000_000_000 + i }))); +assert.equal(cachePuts, 1, "100 concurrent cold refreshes write once instead of 100 times"); +cacheDirectory = new KvDirectory(cacheStorage, cacheKv); +await cacheDirectory.command({ action: "cache", payload: cachePayload("restart"), ts: 1_800_000_300_000 }); +assert.equal(cachePuts, 1, "cache write dedup survives object restart"); +await cacheDirectory.command({ action: "cache", payload: cachePayload("bump"), ts: 1_800_000_900_000 }); +assert.equal(cachePuts, 2, "the existing 15-minute keep-alive ceiling is preserved"); +await cacheDirectory.command({ action: "cache", payload: { changed: true }, ts: 1_800_000_900_001 }); +assert.equal(cachePuts, 3, "real content changes still write immediately"); +cacheKv.put = async () => { throw Error("synthetic_store_unavailable"); }; +await assert.rejects(cacheDirectory.command({ action: "cache", payload: { changed: false }, ts: 1_800_000_900_002 }), /synthetic_store_unavailable/); +cacheKv.put = async (k, v) => { cachePuts++; cacheValues.set(k, v); }; +await cacheDirectory.command({ action: "cache", payload: { changed: false }, ts: 1_800_000_900_003 }); +assert.equal(cachePuts, 4, "a failed write does not advance the successful-write signature"); +console.log("kv_operation_budget: PASS (288 ticks: 1 List/1 mirror put; 100 cold refreshes: 1 cache put)"); diff --git a/web/strategy-switch-console/kv-directory.md b/web/strategy-switch-console/kv-directory.md index 4a44c780..4dd85cfe 100644 --- a/web/strategy-switch-console/kv-directory.md +++ b/web/strategy-switch-console/kv-directory.md @@ -16,17 +16,19 @@ action is serialized with existing commands and performs only exact reads. The existing `STRATEGY_SWITCH_RUNTIME_INSTANCES` binding also hosts a separate object named `console-kv-directory-v1`. Its manifest stores discovery keys only, with a maintained KV mirror at `console_kv_directory:v1`. It has no decision, trading, account, IAM or secret authority, and requires no D1 database, new binding, migration or dependency. +The same serialized object also coordinates writes to the non-authoritative `current_strategies_cache`. Its successful-write signature and timestamp survive restarts, so concurrent cold isolates cannot each pay for the same unchanged cache value. The semantic comparison, immediate writes on real changes and 15-minute keep-alive ceiling are preserved; no observations are re-stamped on reads. Older receivers use the existing KV comparison until upgraded. + All seven source queues and daily history read this maintained manifest. Source/daily publishers register each key durably before writing its existing KV record, under one serialized object queue. New keys update the mirror after the record write; a failed mirror write leaves a persisted dirty marker for retry. Records, freshness, binding checks and decision authority are still read through their original paths. A listed record that cannot be read remains a source error. An incomplete manifest never means an empty queue; unknown daily history is `null`. -Existing keys require a complete migration scan. A 0% Worker override does not select the Durable Object version. If the private receiver returns the exact legacy actor-required rejection, reads use the KV mirror and the explicit authenticated admin operation can perform one bounded migration page directly. No background or read-path fallback scans are allowed; a real object outage does not enable this fallback. The new object adopts that mirror on activation. During this bootstrap, invoke admin maintenance once and inspect its result; the KV mirror alone does not provide cross-region serialization against simultaneous operator requests. Only the existing five-minute cron or an authenticated same-origin admin POST to `/api/admin/kv-directory` can run maintenance. Ordinary reads and acceptance never trigger scans. Each maintenance invocation reads at most one unfiltered KV page (1,000 keys), retaining its cursor. Pending migrations remain unavailable. A reconciliation re-scan of an already complete directory keeps serving the last completed snapshot while it pages; only the initial migration or a failed scan leaves the directory unavailable. Concurrent publishers are retained between pages. After success, periodic reconciliation covers legacy/out-of-band writers; records written through the current ingress are registered immediately. An admin POST requests a fresh scan, with a minimum 60-second successful-scan interval; its response includes `scanned_at` and whether a recent scan was reused. During a mixed-version candidate test, reconcile after the last legacy writer before relying on directory coverage. No index can infer keys that were never discovered or registered. +Existing keys require a complete migration scan. A 0% Worker override does not select the Durable Object version. If the private receiver returns the exact legacy actor-required rejection, reads use the KV mirror and the explicit authenticated admin operation can perform one bounded migration page directly. No background or read-path fallback scans are allowed; a real object outage does not enable this fallback. The new object adopts that mirror on activation. During this bootstrap, invoke admin maintenance once and inspect its result; the KV mirror alone does not provide cross-region serialization against simultaneous operator requests. Only the existing five-minute cron or an authenticated same-origin admin POST to `/api/admin/kv-directory` can run maintenance. Ordinary reads and acceptance never trigger scans. Each maintenance invocation reads at most one unfiltered KV page (1,000 keys), retaining its cursor. Pending migrations remain unavailable. A reconciliation re-scan of an already complete directory keeps serving the last completed snapshot while it pages; only the initial migration or a failed scan leaves the directory unavailable. Concurrent publishers are retained between pages. After success, daily reconciliation covers legacy/out-of-band writers (up to 24 hours of discovery delay by default); records written through the current ingress are registered immediately. An admin POST requests a fresh scan, with a minimum 60-second successful-scan interval; its response includes `scanned_at` and whether a recent scan was reused. During a mixed-version candidate test, reconcile after the last legacy writer before relying on directory coverage. No index can infer keys that were never discovered or registered. -A daily list-quota failure is persisted until the next UTC reset plus 60 seconds. Other migration errors use bounded exponential backoff. The old production version can still consume quota until promotion; installing a candidate cannot restore already-exhausted quota. Never seed a supposedly complete index from a few observed source IDs. If migration cannot complete, retain the explicit source outage. Apply the approved deployment criterion: no worse than the same-request baseline and zero page errors. A new unknown, truncated or mismatched directory is a regression and blocks promotion; an independently confirmed identical quota outage in both versions may remain visible under that criterion. +The serialized directory persists a budget of 100 List attempts per UTC day, including admin refreshes and failed calls. Exhaustion reports `kv_directory_list_budget_exceeded` and keeps the directory unavailable until the next UTC reset plus 60 seconds; it does not claim empty or current data. The legacy receiver fallback cannot guarantee a global budget across concurrent requests, so use it only for the documented single-operator bootstrap. A daily list-quota failure is persisted until the next UTC reset plus 60 seconds. Other migration errors use bounded exponential backoff. The old production version can still consume quota until promotion; installing a candidate cannot restore already-exhausted quota. Never seed a supposedly complete index from a few observed source IDs. If migration cannot complete, retain the explicit source outage. Apply the approved deployment criterion: no worse than the same-request baseline and zero page errors. A new unknown, truncated or mismatched directory is a regression and blocks promotion; an independently confirmed identical quota outage in both versions may remain visible under that criterion. Configuration uses optional Worker vars with conservative defaults; no production settings need to be changed to use the defaults: | Variable | Default | Allowed range | Purpose | | --- | --- | --- | --- | -| `KV_DIRECTORY_RECONCILE_SECONDS` | 21600 | 3600–86400 | Minimum interval between completed migration/reconciliation scans | +| `KV_DIRECTORY_RECONCILE_SECONDS` | 86400 | 3600–86400 | Minimum interval between completed migration/reconciliation scans | | `KV_DIRECTORY_PAGE_DELAY_SECONDS` | 60 | 30–3600 | Minimum delay between migration pages | | `KV_DIRECTORY_RETRY_SECONDS` | 300 | 60–3600 | Initial transient-error backoff; exponential delay capped at one hour | | `KV_DIRECTORY_MAX_KEYS` | 10000 | 100–100000 | Registry safety bound; the manifest also has a conservative 120 KB object-value limit | diff --git a/web/strategy-switch-console/kv_directory.js b/web/strategy-switch-console/kv_directory.js index dfe6839a..7189b285 100644 --- a/web/strategy-switch-console/kv_directory.js +++ b/web/strategy-switch-console/kv_directory.js @@ -1,6 +1,8 @@ // One existing Durable Object owns the discovery manifest. KV remains the // record store; the manifest contains names only and grants no authority. export const DIRECTORY_PATH = "/kv-directory"; +const CACHE_WRITE_KEY = "console-current-strategies-write-v1"; +const CACHE_BUMP_MS = 900_000; const STATE_KEY = "console-kv-directory-v1"; export const MANIFEST_KEY = "console_kv_directory:v1"; const SCHEMA = "qsl_console_kv_directory.v1"; @@ -9,7 +11,7 @@ const PREFIXES = ["control_plane_source:", "adaptive_selection_source:", "runtime_target_lifecycle_source:", "research_task_source:", "research_promotion_ticket:", "runtime_daily:"]; const STATUS_ERRORS = new Set(["kv_directory_uninitialized", "kv_directory_unavailable", - "kv_directory_migration_pending", "kv_directory_migration_failed", "kv_list_daily_limit_exceeded"]); + "kv_directory_migration_pending", "kv_directory_migration_failed", "kv_list_daily_limit_exceeded", "kv_directory_list_budget_exceeded"]); // Whitelist every public field. Never spread stored manifests or provider errors. function publicStatus(state) { @@ -22,11 +24,28 @@ function publicStatus(state) { at: timestamp(lastError.at) } : state?.error && state.error !== "kv_directory_migration_pending" ? { code: STATUS_ERRORS.has(state.error) ? state.error : "kv_directory_migration_failed", at: null } : null, retry_at: timestamp(state?.retry_at), - in_progress: state ? !state.complete && !!state.cursor : null, + in_progress: state ? !!state.cursor : null, data_as_of: timestamp(state?.scanned_at), unavailable_reason: state?.complete ? null : STATUS_ERRORS.has(state?.error) ? state.error : "kv_directory_uninitialized" }; } +export function currentStrategiesSignature(payload) { + const clone = structuredClone(payload ?? null); + const visit = (node) => { + if (Array.isArray(node)) { + node.forEach(visit); + return; + } + if (!node || typeof node !== "object") return; + for (const [key, value] of Object.entries(node)) { + if (key === "execution_evidence" && value && typeof value === "object") delete value.observed_at; + else visit(value); + } + }; + visit(clone); + return JSON.stringify(clone); +} + export function indexedKey(key) { return typeof key === "string" && PREFIXES.some(prefix => key.startsWith(prefix)); } @@ -96,6 +115,29 @@ export class KvDirectory { } async run(command) { + if (command.action === "cache") { + if (!Number.isFinite(command.ts) || command.ts <= 0 || !command.payload || typeof command.payload !== "object") + throw Error("kv_directory_invalid_request"); + const digest = async payload => Array.from(new Uint8Array(await crypto.subtle.digest("SHA-256", + new TextEncoder().encode(currentStrategiesSignature(payload)))), b => b.toString(16).padStart(2, "0")).join(""); + const signature = await digest(command.payload); + let previous = await this.storage.get(CACHE_WRITE_KEY); + if (!previous) { + const text = await this.store.get("current_strategies_cache"); + try { + const cached = JSON.parse(text); + if (cached?.data && Number.isFinite(cached.ts)) + previous = { signature: await digest(cached.data), ts: cached.ts }; + } catch { /* A malformed cache is repairable; failed KV reads are not. */ } + } + if (previous?.signature === signature && command.ts >= previous.ts && command.ts - previous.ts < CACHE_BUMP_MS) { + await this.storage.put(CACHE_WRITE_KEY, previous); + return { ok: true, cached: true }; + } + await this.store.put("current_strategies_cache", JSON.stringify({ ts: command.ts, data: command.payload }, null, 2)); + await this.storage.put(CACHE_WRITE_KEY, { signature, ts: command.ts }); + return { ok: true, cached: false }; + } const stored = await this.storage.get(STATE_KEY) || await sharedManifest(this.store); if (command.action === "status") return publicStatus(stored); const state = stored || { @@ -122,7 +164,7 @@ export class KvDirectory { return { ok: true }; } if (command.action === "maintain") { - const interval = bounded(this.env.KV_DIRECTORY_RECONCILE_SECONDS, 21600, 3600, 86400) * 1000; + const interval = bounded(this.env.KV_DIRECTORY_RECONCILE_SECONDS, 86400, 3600, 86400) * 1000; // A re-scan of a complete directory pages behind the existing snapshot: // keys are union-only, so serving the last completed list is never less // safe than failing the read while the pages run. @@ -139,6 +181,13 @@ export class KvDirectory { try { // One page per maintenance invocation, shared across every prefix and // every Worker isolate. Reads never call list. + // Persist attempts before provider I/O; restarts and explicit refreshes + // share the budget. Leave room for every other namespace on Free. + const day = Math.floor(this.now() / 86400000); + if (state.list_day !== day) { state.list_day = day; state.list_attempts = 0; } + if (state.list_attempts >= 100) throw Error("kv_directory_list_budget_exceeded"); + state.list_attempts = (state.list_attempts || 0) + 1; + await this.storage.put(STATE_KEY, state); const listing = await this.store.list({ limit: 1000, ...(state.cursor ? { cursor: state.cursor } : {}) }); if (!Array.isArray(listing?.keys) || typeof listing.list_complete !== "boolean" || (!listing.list_complete && !listing.cursor)) throw Error("invalid_listing"); @@ -160,10 +209,11 @@ export class KvDirectory { state.complete = false; state.failures += 1; const quota = error.message === "KV list() limit exceeded for the day."; - state.error = quota ? "kv_list_daily_limit_exceeded" : "kv_directory_migration_failed"; + const budget = error.message === "kv_directory_list_budget_exceeded"; + state.error = quota ? "kv_list_daily_limit_exceeded" : budget ? error.message : "kv_directory_migration_failed"; state.last_error = { code: state.error, at: this.now() }; const delay = bounded(this.env.KV_DIRECTORY_RETRY_SECONDS, 300, 60, 3600); - state.retry_at = quota ? (Math.floor(this.now() / 86400000) + 1) * 86400000 + 60000 + state.retry_at = quota || budget ? (Math.floor(this.now() / 86400000) + 1) * 86400000 + 60000 : this.now() + Math.min(3600000, delay * 1000 * 2 ** Math.min(state.failures - 1, 4)); } await this.storage.put(STATE_KEY, state); diff --git a/web/strategy-switch-console/worker.js b/web/strategy-switch-console/worker.js index b5d877e7..983eff26 100644 --- a/web/strategy-switch-console/worker.js +++ b/web/strategy-switch-console/worker.js @@ -4,7 +4,7 @@ import { runtimeReadbackProjection } from "./application/console_api/runtime_rea import { initialEnrollmentBindingsMatch, initialEnrollmentDigest } from "./application/console_api/initial_enrollment.js"; import { readNotificationLog } from "./application/console_api/notification_receipts.js"; import { readRuntimeControlReceipt } from "./application/console_api/runtime_control_receipts.js"; -import { KvDirectory, DIRECTORY_PATH, directoryCommand, directoryList, directoryStatus, indexedKey, maintainDirectory } from "./kv_directory.js"; +import { KvDirectory, currentStrategiesSignature, DIRECTORY_PATH, directoryCommand, directoryList, directoryStatus, indexedKey, maintainDirectory } from "./kv_directory.js"; import { sourceListFailure } from "./application/console_api/read_failures.js"; import { strategyTargetExecution, accountExecutionEvidence } from "./application/console_api/strategy_execution.js"; import { performanceResponse } from "./application/console_api/performance_routes.js"; @@ -6423,22 +6423,6 @@ let _memRefreshing = false; // prevent concurrent background refreshes // stamped with new Date() (see accountExecutionEvidence callers), so whole-payload // byte equality can never hold; the comparison drops those stamps and keeps the // stored value untouched for readers. -function currentStrategiesSignature(payload) { - const clone = structuredClone(payload ?? null); - const visit = (node) => { - if (Array.isArray(node)) { - node.forEach(visit); - return; - } - if (!node || typeof node !== "object") return; - for (const [key, value] of Object.entries(node)) { - if (key === "execution_evidence" && value && typeof value === "object") delete value.observed_at; - else visit(value); - } - }; - visit(clone); - return JSON.stringify(clone); -} // readConfigJson reports a damaged value as `STRATEGY_SWITCH_CONFIG. must be // valid JSON`. Only that shape is a cache miss; a KV read that failed for any other @@ -6459,6 +6443,15 @@ async function readCurrentStrategiesCacheEntry(env) { } async function writeCurrentStrategiesCache(env, payload, ts) { + if (env.STRATEGY_SWITCH_RUNTIME_INSTANCES) { + try { + await directoryCommand(env, { action: "cache", payload, ts }); + return; + } catch (error) { + // Receiver-first compatibility only; a real storage failure remains loud. + if (error.message !== "kv_directory_legacy_receiver" && error.message !== "kv_directory_invalid_request") throw error; + } + } const signature = currentStrategiesSignature(payload); // A damaged stored value must not wedge the cache; this is the only path that can // overwrite it, so the comparison reads through the same repair-aware helper. From 91aef12bc428e2af68f8a57640620a840f51dd7a Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 2 Nov 2026 04:03:33 +0800 Subject: [PATCH 4/4] test(kv): cover serialized cache writes through Worker bridge --- tests/kv_directory_validation.mjs | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/tests/kv_directory_validation.mjs b/tests/kv_directory_validation.mjs index 1143a226..e694cda9 100644 --- a/tests/kv_directory_validation.mjs +++ b/tests/kv_directory_validation.mjs @@ -290,3 +290,29 @@ cacheKv.put = async (k, v) => { cachePuts++; cacheValues.set(k, v); }; await cacheDirectory.command({ action: "cache", payload: { changed: false }, ts: 1_800_000_900_003 }); assert.equal(cachePuts, 4, "a failed write does not advance the successful-write signature"); console.log("kv_operation_budget: PASS (288 ticks: 1 List/1 mirror put; 100 cold refreshes: 1 cache put)"); + +// Exercise the production Worker -> private directory command bridge as well. +const bridgeState = new Map(); let bridgeWrites = 0; +const bridgeKv = { async get() { return null; }, async put() { bridgeWrites++; } }; +const bridgeDirectory = new KvDirectory({ async get(k) { return bridgeState.get(k); }, + async put(k, v) { bridgeState.set(k, structuredClone(v)); } }, bridgeKv); +const bridgeEnv = { STRATEGY_SWITCH_CONFIG: bridgeKv, STRATEGY_SWITCH_RUNTIME_INSTANCES: { + idFromName: name => name, get: name => ({ async fetch(_url, init) { + assert.equal(name, "console-kv-directory-v1"); + const result = await bridgeDirectory.command(JSON.parse(init.body)); + return Response.json(result); + } }), +} }; +await Promise.all(Array.from({ length: 20 }, (_, i) => __test.writeCurrentStrategiesCache( + bridgeEnv, cachePayload(String(i)), 1_800_000_000_000 + i))); +assert.equal(bridgeWrites, 1, "production Worker cache writes actually use the serialized owner"); +const legacyCache = new Map(); let legacyCacheWrites = 0; +const legacyCacheEnv = { STRATEGY_SWITCH_CONFIG: { + async get(k) { return legacyCache.get(k) ?? null; }, + async put(k, v) { legacyCacheWrites++; legacyCache.set(k, v); }, +}, STRATEGY_SWITCH_RUNTIME_INSTANCES: { idFromName: x => x, get: () => ({ async fetch() { + return Response.json({ error: "runtime_instance_actor_required" }, { status: 400 }); +} }) } }; +await __test.writeCurrentStrategiesCache(legacyCacheEnv, cachePayload("old receiver"), 1_800_000_000_000); +await __test.writeCurrentStrategiesCache(legacyCacheEnv, cachePayload("retry"), 1_800_000_000_001); +assert.equal(legacyCacheWrites, 1, "older private receivers retain the existing KV comparison");