From 505d7b9f52d981dfc262bbbe60b004d03d0314a9 Mon Sep 17 00:00:00 2001 From: Benjamin Badejo Date: Thu, 23 Jul 2026 03:42:28 +0300 Subject: [PATCH] fix(state): preserve doctor import TTL --- extensions/matrix/doctor-contract-api.test.ts | 41 +++++++++++++++++++ .../monitor/inbound-dedupe-migration.ts | 1 + src/infra/state-migrations.doctor.ts | 2 +- src/plugin-state/plugin-state-store.ts | 10 ++++- src/plugins/doctor-contract-registry.ts | 4 +- 5 files changed, 53 insertions(+), 5 deletions(-) diff --git a/extensions/matrix/doctor-contract-api.test.ts b/extensions/matrix/doctor-contract-api.test.ts index f34b636a41ad..f2e89aa2bb13 100644 --- a/extensions/matrix/doctor-contract-api.test.ts +++ b/extensions/matrix/doctor-contract-api.test.ts @@ -1006,4 +1006,45 @@ describe("matrix doctor contract state migrations", () => { ].toSorted(), ); }); + + it("preserves a legacy inbound dedupe marker's remaining TTL", async () => { + const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-matrix-doctor-")); + tempDirs.push(stateDir); + const env = { OPENCLAW_STATE_DIR: stateDir }; + const io = { context: createContext(), env }; + const now = 2_000_000_000_000; + const remainingTtlMs = 1_000; + const roomId = "!room:example.org"; + const eventId = "$near-expiry"; + const key = `ops\0${roomId}\0${eventId}`; + const markerTs = now - MATRIX_INBOUND_DEDUPE_TTL_MS + remainingTtlMs; + const storedEntry = createPersistentDedupeImportEntry({ key, seenAt: markerTs }); + const nowSpy = vi.spyOn(Date, "now").mockReturnValue(now); + + await expect( + importNewestInboundDedupeMarkers({ + io, + now, + markers: [ + { + accountId: "ops", + roomId, + eventId, + ts: markerTs, + }, + ], + }), + ).resolves.toEqual({ imported: 1, total: 1 }); + + const store = createPluginStateKeyedStoreForTests("matrix", { + namespace: resolveMatrixInboundDedupeStateNamespace(), + maxEntries: 20_000, + defaultTtlMs: MATRIX_INBOUND_DEDUPE_TTL_MS, + env, + }); + nowSpy.mockReturnValue(now + remainingTtlMs - 1); + await expect(store.lookup(storedEntry.key)).resolves.toEqual(storedEntry.value); + nowSpy.mockReturnValue(now + remainingTtlMs + 1); + await expect(store.lookup(storedEntry.key)).resolves.toBeUndefined(); + }); }); diff --git a/extensions/matrix/src/matrix/monitor/inbound-dedupe-migration.ts b/extensions/matrix/src/matrix/monitor/inbound-dedupe-migration.ts index e1f33e2eb772..e599366a5a48 100644 --- a/extensions/matrix/src/matrix/monitor/inbound-dedupe-migration.ts +++ b/extensions/matrix/src/matrix/monitor/inbound-dedupe-migration.ts @@ -449,6 +449,7 @@ export async function importNewestInboundDedupeMarkers(params: { key: entry.key, value: entry.value, createdAt: marker.ts, + ...(entry.ttlMs != null ? { ttlMs: entry.ttlMs } : {}), })), ); const importedKeys = new Set((await store.entries()).map((entry) => entry.key)); diff --git a/src/infra/state-migrations.doctor.ts b/src/infra/state-migrations.doctor.ts index 96314d12ccfc..298eb8e65f6b 100644 --- a/src/infra/state-migrations.doctor.ts +++ b/src/infra/state-migrations.doctor.ts @@ -285,7 +285,7 @@ function createPluginDoctorStateMigrationContext( }, importPluginStateEntries( options: OpenKeyedStoreOptions, - entries: readonly { key: string; value: unknown; createdAt: number }[], + entries: readonly { key: string; value: unknown; createdAt: number; ttlMs?: number }[], ) { importPluginStateEntriesForDoctor(pluginId, { ...options, env: options.env ?? env }, entries); }, diff --git a/src/plugin-state/plugin-state-store.ts b/src/plugin-state/plugin-state-store.ts index 3cf3ed93fef8..15e892ee9d83 100644 --- a/src/plugin-state/plugin-state-store.ts +++ b/src/plugin-state/plugin-state-store.ts @@ -69,6 +69,7 @@ type PluginStateImportEntry = { key: string; value: unknown; createdAt: number; + ttlMs?: number; }; const namespaceOptionSignatures = new Map(); @@ -555,7 +556,7 @@ export function registerPluginStateSyncSequencedJournalEntry(params: { }); } -/** Doctor-only import that preserves source age for retention ordering. */ +/** Doctor-only import that preserves source age and remaining retention. */ export function importPluginStateEntriesForDoctor( pluginId: string, options: OpenKeyedStoreOptions, @@ -575,7 +576,12 @@ export function importPluginStateEntriesForDoctor( if (!Number.isSafeInteger(entry.createdAt)) { throw invalidInput("plugin state import createdAt must be a safe integer", "register"); } - const prepared = prepareRegisterParams(entry.key, entry.value, defaultTtlMs); + const prepared = prepareRegisterParams( + entry.key, + entry.value, + defaultTtlMs, + entry.ttlMs != null ? { ttlMs: entry.ttlMs } : undefined, + ); pluginStateRegister({ pluginId, namespace, diff --git a/src/plugins/doctor-contract-registry.ts b/src/plugins/doctor-contract-registry.ts index 4a0780593c86..7cf6181ea227 100644 --- a/src/plugins/doctor-contract-registry.ts +++ b/src/plugins/doctor-contract-registry.ts @@ -59,10 +59,10 @@ export type PluginDoctorStateMigrationDetection = { export type PluginDoctorStateMigrationContext = { openPluginStateKeyedStore: (options: OpenKeyedStoreOptions) => PluginStateKeyedStore; - /** Doctor-only batch import preserving source age for retention ordering. */ + /** Doctor-only batch import preserving source age and remaining retention. */ importPluginStateEntries?: ( options: OpenKeyedStoreOptions, - entries: readonly { key: string; value: unknown; createdAt: number }[], + entries: readonly { key: string; value: unknown; createdAt: number; ttlMs?: number }[], ) => void; /** Plugin-wide live-row capacity for import preflight. Older test hosts may omit it. */ getPluginStateCapacity?: () => { liveEntries: number; maxEntries: number };