diff --git a/extensions/slack/src/monitor/message-handler/prepare.test.ts b/extensions/slack/src/monitor/message-handler/prepare.test.ts index f751c5d7d883..1b0ded3dac00 100644 --- a/extensions/slack/src/monitor/message-handler/prepare.test.ts +++ b/extensions/slack/src/monitor/message-handler/prepare.test.ts @@ -4906,110 +4906,191 @@ describe("slack implicit mention policy", () => { storeFixture.setup(); }); + beforeEach(() => { + clearSlackThreadParticipationCache(); + }); + afterAll(() => { storeFixture.cleanup(); }); - function createCtxWithImplicitMentions(implicitMentions?: { - replyToBot?: boolean; - threadParticipation?: boolean; - }) { + function createCtxWithImplicitMentions( + implicitMentions?: { + replyToBot?: boolean; + threadParticipation?: boolean; + }, + options?: Pick< + Parameters[0], + "channelsConfig" | "groupPolicy" + >, + ) { const ctx = createInboundSlackTestContext({ cfg: { - channels: { slack: { enabled: true, implicitMentions } }, + channels: { + slack: { + enabled: true, + implicitMentions, + ...(options?.channelsConfig ? { channels: options.channelsConfig } : {}), + ...(options?.groupPolicy ? { groupPolicy: options.groupPolicy } : {}), + }, + }, session: {}, } as OpenClawConfig, + ...options, }); ctx.resolveUserName = async () => ({ name: "Alice" }); return ctx; } - it("drops a reply to the bot when replyToBot is disabled", async () => { - const ctx = createCtxWithImplicitMentions({ replyToBot: false }); + async function prepareThreadMessage(params: { + ctx: SlackMonitorContext; + message?: Partial; + eventScope?: SlackEventScope; + }) { const { storePath } = storeFixture.makeTmpStorePath(); vi.spyOn( await import("openclaw/plugin-sdk/session-store-runtime"), "resolveStorePath", ).mockReturnValue(storePath); - const account = createSlackTestAccount(); - const message: SlackMessageEvent = { - type: "message", - channel: "C123", - channel_type: "channel", - user: "U1", - text: "hello", - ts: "1700000001.000001", - thread_ts: "1700000000.000000", - parent_user_id: "B1", // bot is thread parent - }; - const result = await prepareSlackMessage({ + return await prepareSlackMessage({ + ctx: params.ctx, + account: createSlackTestAccount(), + message: { + type: "message", + channel: "C123", + channel_type: "channel", + user: "U1", + text: "hello", + ts: "1700000001.000001", + thread_ts: "1700000000.000000", + parent_user_id: "U2", + ...params.message, + }, + opts: { + source: "message", + ...(params.eventScope ? { eventScope: params.eventScope } : {}), + }, + }); + } + + it("drops a reply to the bot when replyToBot is disabled", async () => { + const ctx = createCtxWithImplicitMentions({ replyToBot: false }); + const result = await prepareThreadMessage({ ctx, - account, - message, - opts: { source: "message" }, + message: { parent_user_id: "B1" }, }); expect(result).toBeNull(); }); it("allows an explicit mention when all implicit thread signals are disabled", async () => { - const ctx = createCtxWithImplicitMentions({ - replyToBot: false, - threadParticipation: false, - }); - const { storePath } = storeFixture.makeTmpStorePath(); - vi.spyOn( - await import("openclaw/plugin-sdk/session-store-runtime"), - "resolveStorePath", - ).mockReturnValue(storePath); - const account = createSlackTestAccount(); - const message: SlackMessageEvent = { - type: "message", - channel: "C123", - channel_type: "channel", - user: "U1", - text: "<@B1> hello", - ts: "1700000001.000002", - thread_ts: "1700000000.000000", - parent_user_id: "B1", - }; - const result = await prepareSlackMessage({ + const ctx = createCtxWithImplicitMentions( + { replyToBot: false, threadParticipation: false }, + { channelsConfig: { C123: { requireMention: true } } }, + ); + const result = await prepareThreadMessage({ ctx, - account, - message, - opts: { source: "message" }, + message: { text: "<@B1> hello", parent_user_id: "B1" }, }); - if (!result) { - throw new Error("expected Slack thread reply message"); - } + expect(result?.ctxPayload.MentionSource).toBe("explicit_bot"); }); it("controls persisted thread participation independently from replies to the bot", async () => { const threadTs = "1700000000.000000"; recordSlackThreadParticipation("default", "C123", threadTs); const ctx = createCtxWithImplicitMentions({ threadParticipation: false }); - const { storePath } = storeFixture.makeTmpStorePath(); - vi.spyOn( - await import("openclaw/plugin-sdk/session-store-runtime"), - "resolveStorePath", - ).mockReturnValue(storePath); - const account = createSlackTestAccount(); - const message: SlackMessageEvent = { - type: "message", - channel: "C123", - channel_type: "channel", - user: "U1", - text: "hello", - ts: "1700000001.000003", - thread_ts: threadTs, - parent_user_id: "U2", - }; - const result = await prepareSlackMessage({ + const result = await prepareThreadMessage({ ctx, - account, - message, - opts: { source: "message" }, + message: { thread_ts: threadTs }, }); expect(result).toBeNull(); }); + + it("accepts an unmentioned reply more than 24 hours after joining a required-mention thread", async () => { + const threadTs = "1700000000.000000"; + const initialNow = 1_700_000_000_000; + const nowSpy = vi.spyOn(Date, "now").mockReturnValue(initialNow); + + try { + recordSlackThreadParticipation("default", "C123", threadTs); + nowSpy.mockReturnValue(initialNow + 25 * 60 * 60 * 1000); + + const ctx = createCtxWithImplicitMentions(undefined, { + channelsConfig: { C123: { requireMention: true } }, + }); + const result = await prepareThreadMessage({ ctx, message: { thread_ts: threadTs } }); + + expect(result?.ctxPayload.MentionSource).toBe("implicit_thread"); + expect(result?.ctxPayload.ImplicitMentionKinds).toEqual(["bot_thread_participant"]); + } finally { + nowSpy.mockRestore(); + } + }); + + it("continues requiring a mention in an unrelated thread the bot never joined", async () => { + recordSlackThreadParticipation("default", "C123", "1700000000.000999"); + const ctx = createCtxWithImplicitMentions(undefined, { + channelsConfig: { C123: { requireMention: true } }, + }); + + expect(await prepareThreadMessage({ ctx })).toBeNull(); + }); + + it("preserves explicit channel settings that do not require mentions", async () => { + const ctx = createCtxWithImplicitMentions(undefined, { + channelsConfig: { C123: { requireMention: false } }, + }); + + const result = await prepareThreadMessage({ ctx }); + + expect(result?.ctxPayload.MentionSource).toBe("none"); + expect(result?.ctxPayload.ImplicitMentionKinds).toBeUndefined(); + }); + + const unauthorizedThreadCases: Array<{ + authorization: string; + options: Pick< + Parameters[0], + "channelsConfig" | "groupPolicy" + >; + }> = [ + { + authorization: "channel", + options: { + channelsConfig: { C_ALLOWED: { enabled: true, requireMention: true } }, + groupPolicy: "allowlist" as const, + }, + }, + { + authorization: "sender", + options: { + channelsConfig: { C123: { requireMention: true, users: ["U_ALLOWED"] } }, + }, + }, + ]; + + it.each(unauthorizedThreadCases)( + "rejects a joined thread when $authorization authorization fails", + async ({ options }) => { + recordSlackThreadParticipation("default", "C123", "1700000000.000000"); + const ctx = createCtxWithImplicitMentions(undefined, options); + + expect(await prepareThreadMessage({ ctx })).toBeNull(); + }, + ); + + it("does not accept participation recorded in a different enterprise workspace", async () => { + recordSlackThreadParticipation("default", "C123", "1700000000.000000", { + teamId: "T_OTHER", + }); + const ctx = createCtxWithImplicitMentions(undefined, { + channelsConfig: { C123: { requireMention: true } }, + }); + const eventScope = { + teamId: "T1", + client: {} as SlackEventScope["client"], + } satisfies SlackEventScope; + + expect(await prepareThreadMessage({ ctx, eventScope })).toBeNull(); + }); }); /* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/extensions/slack/src/sent-thread-cache.test.ts b/extensions/slack/src/sent-thread-cache.test.ts index 1291ce71fd37..808972bcc599 100644 --- a/extensions/slack/src/sent-thread-cache.test.ts +++ b/extensions/slack/src/sent-thread-cache.test.ts @@ -1,5 +1,11 @@ // Slack tests cover sent thread cache plugin behavior. +import type { OpenKeyedStoreOptions } from "openclaw/plugin-sdk/plugin-state-runtime"; +import { + createPluginStateKeyedStoreForTests, + resetPluginStateStoreForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; import { importFreshModule } from "openclaw/plugin-sdk/test-fixtures"; +import { withOpenClawTestState } from "openclaw/plugin-sdk/test-state"; import { afterEach, describe, expect, it, vi } from "vitest"; import { setSlackRuntime } from "./runtime.js"; import { @@ -13,6 +19,7 @@ describe("slack sent-thread-cache", () => { afterEach(() => { clearSlackThreadParticipationCache(); setSlackRuntime(null as never); + resetPluginStateStoreForTests(); vi.restoreAllMocks(); }); @@ -37,6 +44,14 @@ describe("slack sent-thread-cache", () => { expect(hasSlackThreadParticipation("A1", "C123", "1700000000.000001")).toBe(true); }); + it("scopes participation by enterprise workspace without matching unscoped threads", () => { + recordSlackThreadParticipation("A1", "C123", "1700000000.000001", { teamId: "T1" }); + + expect(hasSlackThreadParticipation("A1", "C123", "1700000000.000001", "T1")).toBe(true); + expect(hasSlackThreadParticipation("A1", "C123", "1700000000.000001", "T2")).toBe(false); + expect(hasSlackThreadParticipation("A1", "C123", "1700000000.000001")).toBe(false); + }); + it("ignores empty accountId, channelId, or threadTs", () => { recordSlackThreadParticipation("", "C123", "1700000000.000001"); recordSlackThreadParticipation("A1", "", "1700000000.000001"); @@ -77,11 +92,10 @@ describe("slack sent-thread-cache", () => { } }); - it("expired entries return false and are cleaned up on read", () => { + it("retains thread participation more than 24 hours after the bot replied", () => { recordSlackThreadParticipation("A1", "C123", "1700000000.000001"); - // Advance time past the 24-hour TTL vi.spyOn(Date, "now").mockReturnValue(Date.now() + 25 * 60 * 60 * 1000); - expect(hasSlackThreadParticipation("A1", "C123", "1700000000.000001")).toBe(false); + expect(hasSlackThreadParticipation("A1", "C123", "1700000000.000001")).toBe(true); }); it("enforces maximum entries by evicting oldest fresh entries", () => { @@ -93,14 +107,14 @@ describe("slack sent-thread-cache", () => { expect(hasSlackThreadParticipation("A1", "C123", "1700000000.005000")).toBe(true); }); - it("restores persistent thread participation without extending its original expiry", async () => { + it("restores persistent thread participation more than 24 hours after the bot replied", async () => { const repliedAt = 1_711_406_400_000; - const ttlMs = 24 * 60 * 60 * 1000; const now = vi.spyOn(Date, "now").mockReturnValue(repliedAt); - const register = vi.fn().mockResolvedValue(undefined); - const lookup = vi - .fn() - .mockImplementation(async () => (Date.now() < repliedAt + ttlMs ? { repliedAt } : undefined)); + const persistedRecords = new Map(); + const register = vi.fn(async (key: string, value: { repliedAt: number }) => { + persistedRecords.set(key, value); + }); + const lookup = vi.fn(async (key: string) => persistedRecords.get(key)); const openKeyedStore = vi.fn(() => ({ register, lookup, @@ -120,8 +134,12 @@ describe("slack sent-thread-cache", () => { expect(register).toHaveBeenCalledWith("A1:C123:1700000000.000002", { repliedAt, }); + expect(openKeyedStore).toHaveBeenCalledWith({ + namespace: "slack.thread-participation", + maxEntries: 1000, + }); - now.mockReturnValue(repliedAt + ttlMs - 1000); + now.mockReturnValue(repliedAt + 25 * 60 * 60 * 1000); clearSlackThreadParticipationCache(); await expect( hasSlackThreadParticipationWithPersistence({ @@ -143,15 +161,183 @@ describe("slack sent-thread-cache", () => { ).resolves.toBe(true); expect(lookup).not.toHaveBeenCalled(); - now.mockReturnValue(repliedAt + ttlMs + 1000); + now.mockReturnValue(repliedAt + 365 * 24 * 60 * 60 * 1000); await expect( hasSlackThreadParticipationWithPersistence({ accountId: "A1", channelId: "C123", threadTs: "1700000000.000002", }), + ).resolves.toBe(true); + expect(lookup).not.toHaveBeenCalled(); + }); + + it("preserves hydrated legacy expiration while new participation survives restart", async () => { + await withOpenClawTestState( + { label: "slack-thread-participation", layout: "state-only", applyEnv: false }, + async (state) => { + resetPluginStateStoreForTests(); + const repliedAt = 1_711_406_400_000; + const now = vi.spyOn(Date, "now").mockReturnValue(repliedAt); + const legacyThreadTs = "1700000000.000004"; + const legacyKey = `A1:T1:C123:${legacyThreadTs}`; + const legacyStore = createPluginStateKeyedStoreForTests<{ repliedAt: number }>("slack", { + namespace: "slack.thread-participation", + maxEntries: 1000, + defaultTtlMs: 24 * 60 * 60 * 1000, + env: state.env, + }); + await legacyStore.register(legacyKey, { repliedAt }); + const legacyEntry = (await legacyStore.entries()).find((entry) => entry.key === legacyKey); + expect(legacyEntry?.expiresAt).toBe(repliedAt + 24 * 60 * 60 * 1000); + resetPluginStateStoreForTests(); + + const openKeyedStore = vi.fn((options: OpenKeyedStoreOptions) => + createPluginStateKeyedStoreForTests<{ repliedAt: number }>("slack", { + ...options, + env: state.env, + }), + ); + setSlackRuntime({ + state: { openKeyedStore }, + logging: { getChildLogger: () => ({ warn: vi.fn() }) }, + } as never); + + now.mockReturnValue(repliedAt + 23 * 60 * 60 * 1000); + await expect( + hasSlackThreadParticipationWithPersistence({ + accountId: "A1", + teamId: "T1", + channelId: "C123", + threadTs: legacyThreadTs, + }), + ).resolves.toBe(true); + expect(hasSlackThreadParticipation("A1", "C123", legacyThreadTs, "T1")).toBe(false); + expect(openKeyedStore).toHaveBeenCalledExactlyOnceWith({ + namespace: "slack.thread-participation", + maxEntries: 1000, + }); + const store = createPluginStateKeyedStoreForTests<{ repliedAt: number }>("slack", { + namespace: "slack.thread-participation", + maxEntries: 1000, + env: state.env, + }); + const preservedLegacyEntry = (await store.entries()).find( + (candidate) => candidate.key === legacyKey, + ); + expect(preservedLegacyEntry?.expiresAt).toBe(legacyEntry?.expiresAt); + + now.mockReturnValue(repliedAt + 25 * 60 * 60 * 1000); + await expect( + hasSlackThreadParticipationWithPersistence({ + accountId: "A1", + teamId: "T1", + channelId: "C123", + threadTs: legacyThreadTs, + }), + ).resolves.toBe(false); + expect(hasSlackThreadParticipation("A1", "C123", legacyThreadTs, "T1")).toBe(false); + expect(await store.lookup(legacyKey)).toBeUndefined(); + recordSlackThreadParticipation("A1", "C123", "1700000000.000003", { teamId: "T1" }); + await vi.waitFor(async () => { + await expect(store.lookup("A1:T1:C123:1700000000.000003")).resolves.toEqual({ + repliedAt: repliedAt + 25 * 60 * 60 * 1000, + }); + }); + const entry = (await store.entries()).find( + (candidate) => candidate.key === "A1:T1:C123:1700000000.000003", + ); + expect(entry).toBeDefined(); + expect(entry).not.toHaveProperty("expiresAt"); + + now.mockReturnValue(repliedAt + 50 * 60 * 60 * 1000); + resetPluginStateStoreForTests(); + clearSlackThreadParticipationCache(); + + await expect( + hasSlackThreadParticipationWithPersistence({ + accountId: "A1", + teamId: "T1", + channelId: "C123", + threadTs: legacyThreadTs, + }), + ).resolves.toBe(false); + await expect( + hasSlackThreadParticipationWithPersistence({ + accountId: "A1", + teamId: "T1", + channelId: "C123", + threadTs: "1700000000.000003", + }), + ).resolves.toBe(true); + expect(hasSlackThreadParticipation("A1", "C123", "1700000000.000003", "T1")).toBe(true); + expect(openKeyedStore).toHaveBeenCalledTimes(2); + + for (const probe of [ + { accountId: "A2", teamId: "T1", channelId: "C123", threadTs: "1700000000.000003" }, + { accountId: "A1", teamId: "T2", channelId: "C123", threadTs: "1700000000.000003" }, + { accountId: "A1", channelId: "C123", threadTs: "1700000000.000003" }, + { accountId: "A1", teamId: "T1", channelId: "C456", threadTs: "1700000000.000003" }, + { accountId: "A1", teamId: "T1", channelId: "C123", threadTs: "1700000000.000005" }, + ]) { + await expect(hasSlackThreadParticipationWithPersistence(probe)).resolves.toBe(false); + } + }, + ); + }); + + it("bounds persistent participation to 1,000 entries and evicts the oldest", async () => { + const persistedRecords = new Map(); + const register = vi.fn(async (key: string, value: { repliedAt: number }) => { + persistedRecords.delete(key); + persistedRecords.set(key, value); + if (persistedRecords.size > 1000) { + const oldestKey = persistedRecords.keys().next().value; + if (oldestKey !== undefined) { + persistedRecords.delete(oldestKey); + } + } + }); + const lookup = vi.fn(async (key: string) => persistedRecords.get(key)); + const openKeyedStore = vi.fn(() => ({ register, lookup })); + setSlackRuntime({ + state: { openKeyedStore }, + logging: { getChildLogger: () => ({ warn: vi.fn() }) }, + } as never); + + for (let i = 0; i < 1001; i += 1) { + recordSlackThreadParticipation("A1", "C123", `1700000000.${String(i).padStart(6, "0")}`); + } + + await vi.waitFor(() => expect(register).toHaveBeenCalledTimes(1001)); + expect(openKeyedStore).toHaveBeenCalledWith({ + namespace: "slack.thread-participation", + maxEntries: 1000, + }); + expect(persistedRecords.size).toBe(1000); + clearSlackThreadParticipationCache(); + + await expect( + hasSlackThreadParticipationWithPersistence({ + accountId: "A1", + channelId: "C123", + threadTs: "1700000000.000000", + }), ).resolves.toBe(false); - expect(lookup).toHaveBeenCalledWith("A1:C123:1700000000.000002"); + await expect( + hasSlackThreadParticipationWithPersistence({ + accountId: "A1", + channelId: "C123", + threadTs: "1700000000.000001", + }), + ).resolves.toBe(true); + await expect( + hasSlackThreadParticipationWithPersistence({ + accountId: "A1", + channelId: "C123", + threadTs: "1700000000.001000", + }), + ).resolves.toBe(true); }); it("falls back to in-memory thread participation when persistent state cannot open", async () => { diff --git a/extensions/slack/src/sent-thread-cache.ts b/extensions/slack/src/sent-thread-cache.ts index a06c27ea609d..96da39293459 100644 --- a/extensions/slack/src/sent-thread-cache.ts +++ b/extensions/slack/src/sent-thread-cache.ts @@ -8,7 +8,6 @@ import { getOptionalSlackRuntime } from "./runtime.js"; * Used to auto-respond in threads without requiring @mention after the first reply. */ -const TTL_MS = 24 * 60 * 60 * 1000; // 24 hours const MAX_ENTRIES = 5000; const PERSISTENT_MAX_ENTRIES = 1000; const PERSISTENT_NAMESPACE = "slack.thread-participation"; @@ -25,7 +24,8 @@ type SlackThreadParticipationRecord = { const SLACK_THREAD_PARTICIPATION_KEY = Symbol.for("openclaw.slackThreadParticipation"); const threadParticipation = createPersistentDedupeCache({ globalKey: SLACK_THREAD_PARTICIPATION_KEY, - ttlMs: TTL_MS, + // Participation remains valid until bounded oldest-entry eviction removes it. + ttlMs: 0, maxSize: MAX_ENTRIES, persistent: { namespace: PERSISTENT_NAMESPACE, @@ -37,8 +37,6 @@ const threadParticipation = createPersistentDedupeCache repliedAt, }, }); diff --git a/src/plugin-sdk/dedupe-runtime.test.ts b/src/plugin-sdk/dedupe-runtime.test.ts index 449b5b7bab2d..6bdac230e0e5 100644 --- a/src/plugin-sdk/dedupe-runtime.test.ts +++ b/src/plugin-sdk/dedupe-runtime.test.ts @@ -20,16 +20,19 @@ function createCache(params?: { openStore?: () => ReturnType["store"] | undefined; logError?: (error: unknown) => void; readTimestamp?: (record: Record) => number | undefined; + ttlMs?: number; + maxSize?: number; + persistentMaxEntries?: number; }) { const backing = createMemoryStore(); const cache = createPersistentDedupeCache({ // Plain Symbol() is unique per cache, so parallel tests never share memory layers. globalKey: Symbol("test.persistent-dedupe"), - ttlMs: 60_000, - maxSize: 100, + ttlMs: params?.ttlMs ?? 60_000, + maxSize: params?.maxSize ?? 100, persistent: { namespace: "test.persistent-dedupe", - maxEntries: 100, + maxEntries: params?.persistentMaxEntries ?? 100, openStore: params?.openStore ?? (() => backing.store), logError: params?.logError, readTimestamp: params?.readTimestamp, @@ -52,6 +55,31 @@ describe("createPersistentDedupeCache", () => { expect(backing.store.lookup).not.toHaveBeenCalled(); }); + it.each([0, -1])("omits the persistent TTL when ttlMs is %i", async (ttlMs) => { + const openStore = vi.fn(() => createMemoryStore().store); + const { cache } = createCache({ ttlMs, openStore }); + + await cache.register("non-expiring", { at: 1 }); + + expect(openStore).toHaveBeenCalledWith({ + namespace: "test.persistent-dedupe", + maxEntries: 100, + }); + }); + + it("forwards positive persistent TTLs without changing capacity", async () => { + const openStore = vi.fn(() => createMemoryStore().store); + const { cache } = createCache({ ttlMs: 60_000, persistentMaxEntries: 3, openStore }); + + await cache.register("expiring", { at: 1 }); + + expect(openStore).toHaveBeenCalledWith({ + namespace: "test.persistent-dedupe", + maxEntries: 3, + defaultTtlMs: 60_000, + }); + }); + it("falls back to persistence and re-primes memory on a hit", async () => { const { cache, backing } = createCache(); backing.entries.set("k2", { at: 42 }); @@ -71,6 +99,146 @@ describe("createPersistentDedupeCache", () => { expect(cache.peek("k3")).toBe(false); }); + it("keeps legacy persistent expirations authoritative in non-expiring memory", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000_000); + const backing = createMemoryStore(); + const record = { at: Date.now() }; + const expiresAt = Date.now() + 60_000; + backing.entries.set("legacy", record); + const store = { + ...backing.store, + lookup: vi.fn(async (key: string) => + Date.now() < expiresAt ? backing.entries.get(key) : undefined, + ), + entries: vi.fn(async () => [ + { key: "legacy", value: record, createdAt: record.at, expiresAt }, + ]), + }; + const { cache } = createCache({ ttlMs: 0, openStore: () => store }); + + expect(await cache.lookup("legacy")).toBe(true); + expect(cache.peek("legacy")).toBe(false); + + vi.setSystemTime(expiresAt); + expect(await cache.lookup("legacy")).toBe(false); + expect(cache.peek("legacy")).toBe(false); + expect(store.lookup).toHaveBeenCalledTimes(2); + }); + + it("re-primes non-expiring memory only for durable persistent entries", async () => { + const backing = createMemoryStore(); + const record = { at: 1_000_000 }; + backing.entries.set("durable", record); + const store = { + ...backing.store, + entries: vi.fn(async () => [{ key: "durable", value: record, createdAt: record.at }]), + }; + const { cache } = createCache({ ttlMs: 0, openStore: () => store }); + + expect(await cache.lookup("durable")).toBe(true); + expect(cache.peek("durable")).toBe(true); + expect(await cache.lookup("durable")).toBe(true); + expect(store.lookup).toHaveBeenCalledOnce(); + }); + + it("rejects a persistent hit that disappears before its expiration metadata is read", async () => { + const backing = createMemoryStore(); + backing.entries.set("missing", { at: 1_000_000 }); + const store = { + ...backing.store, + entries: vi.fn(async () => []), + }; + const { cache } = createCache({ ttlMs: 0, openStore: () => store }); + + expect(await cache.lookup("missing")).toBe(false); + expect(cache.peek("missing")).toBe(false); + }); + + it("re-primes non-expiring memory when entry metadata is unavailable", async () => { + const backing = createMemoryStore(); + backing.entries.set("durable", { at: 1_000_000 }); + const store = { + ...backing.store, + entries: vi.fn(async () => undefined), + }; + const { cache } = createCache({ ttlMs: 0, openStore: () => store }); + + expect(await cache.lookup("durable")).toBe(true); + expect(cache.peek("durable")).toBe(true); + }); + + it("disables persistence after an expiration metadata read fails", async () => { + const logError = vi.fn(); + const backing = createMemoryStore(); + backing.entries.set("legacy", { at: 1_000_000 }); + const error = new Error("metadata read failed"); + const store = { + ...backing.store, + entries: vi.fn(async () => { + throw error; + }), + }; + const { cache } = createCache({ ttlMs: 0, openStore: () => store, logError }); + + expect(await cache.lookup("legacy")).toBe(false); + expect(cache.peek("legacy")).toBe(false); + expect(logError).toHaveBeenCalledExactlyOnceWith(error); + await cache.register("later", { at: 1_000_001 }); + expect(store.register).not.toHaveBeenCalled(); + expect(store.entries).toHaveBeenCalledOnce(); + }); + + it("does not inspect expiration metadata for positive-TTL caches", async () => { + const backing = createMemoryStore(); + backing.entries.set("expiring", { at: 1_000_000 }); + const store = { + ...backing.store, + entries: vi.fn(async () => []), + }; + const { cache } = createCache({ ttlMs: 60_000, openStore: () => store }); + + expect(await cache.lookup("expiring")).toBe(true); + expect(cache.peek("expiring")).toBe(true); + expect(store.entries).not.toHaveBeenCalled(); + }); + + it("retains non-expiring entries past 24 hours and restores them after restart", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000_000); + const { cache, backing } = createCache({ ttlMs: 0, readTimestamp: (record) => record.at }); + + await cache.register("durable", { at: Date.now() }, { at: Date.now() }); + vi.setSystemTime(1_000_000 + 25 * 60 * 60 * 1000); + expect(cache.peek("durable")).toBe(true); + + const { cache: restarted } = createCache({ + ttlMs: 0, + openStore: () => backing.store, + readTimestamp: (record) => record.at, + }); + expect(restarted.peek("durable")).toBe(false); + expect(await restarted.lookup("durable")).toBe(true); + expect(restarted.peek("durable")).toBe(true); + + restarted.clearForTest(); + expect(restarted.peek("durable")).toBe(false); + expect(await restarted.lookup("durable")).toBe(true); + expect(backing.store.lookup).toHaveBeenCalledTimes(2); + }); + + it("deterministically evicts the oldest non-expiring memory entry at capacity", async () => { + const { cache } = createCache({ ttlMs: 0, maxSize: 2, openStore: () => undefined }); + + await cache.register("oldest", { at: 1 }, { at: 1 }); + await cache.register("retained", { at: 2 }, { at: 2 }); + await cache.register("newest", { at: 3 }, { at: 3 }); + + expect(cache.peek("oldest")).toBe(false); + expect(cache.peek("retained")).toBe(true); + expect(cache.peek("newest")).toBe(true); + }); + it("disables persistence after an open failure and never rejects", async () => { const logError = vi.fn(); const openStore = vi.fn(() => { diff --git a/src/plugin-sdk/dedupe-runtime.ts b/src/plugin-sdk/dedupe-runtime.ts index a4993ae87111..2687751ee67a 100644 --- a/src/plugin-sdk/dedupe-runtime.ts +++ b/src/plugin-sdk/dedupe-runtime.ts @@ -10,6 +10,11 @@ type PersistentDedupeStore = { lookup(key: string): Promise; }; +type PersistentDedupeEntryMetadata = { + key: string; + expiresAt?: number; +}; + /** Dual-layer presence cache: process-memory dedupe plus best-effort persistent state. */ export type PersistentDedupeCache = { /** Memory-only presence check without refreshing recency. */ @@ -67,7 +72,7 @@ export function createPersistentDedupeCache(params: { persistentStore = params.persistent.openStore({ namespace: params.persistent.namespace, maxEntries: params.persistent.maxEntries, - defaultTtlMs: params.ttlMs, + ...(params.ttlMs > 0 ? { defaultTtlMs: params.ttlMs } : {}), }); return persistentStore; } catch (error) { @@ -89,13 +94,31 @@ export function createPersistentDedupeCache(params: { let record: TRecord | undefined; try { record = await store.lookup(key); + if (record === undefined) { + return false; + } + if (params.ttlMs <= 0) { + const metadataStore = store as { + entries?: () => Promise; + }; + if (typeof metadataStore.entries === "function") { + const entries = await metadataStore.entries(); + if (Array.isArray(entries)) { + const entry = entries.find((candidate) => candidate.key === key); + if (!entry) { + return false; + } + if (entry.expiresAt !== undefined) { + // Legacy TTL-bound entries must never become permanent memory hits. + return entry.expiresAt > Date.now(); + } + } + } + } } catch (error) { disablePersistentStore(error); return false; } - if (record === undefined) { - return false; - } memory.check(key, params.persistent.readTimestamp?.(record)); return true; },