ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-12 21:53:00 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(slack): keep previously joined threads active beyond 24 hours (#121708)

Browse Source 
* fix(slack): keep participated threads active beyond 24 hours

* fix(plugin-state): reject transient TTLs after durable migration

* fix(slack): keep channel entrypoint on narrow plugin API

* chore(sdk): refresh sharded durable-state API baselines

* fix(slack): preserve existing thread expiration without migration

* fix(slack): preserve hydrated legacy thread expiration

---------

Co-authored-by: kevinlin-openai <kevinlin@openai.com>
This commit is contained in:
Kevin Lin
2026-08-11 11:31:07 -07:00
committed by GitHub
parent bd6ff17d4d
commit 2d0c30fd65
5 changed files with 548 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extensions/slack/src/monitor/message-handler/prepare.test.ts
+150 -69
View File
@@ -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<typeof createInboundSlackTestContext>[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<SlackMessageEvent>;
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<typeof createInboundSlackTestContext>[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. */
extensions/slack/src/sent-thread-cache.test.ts
+198 -12
View File
@@ -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<string, { repliedAt: number }>();
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<string, { repliedAt: number }>();
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 () => {
extensions/slack/src/sent-thread-cache.ts
+2 -4
View File
@@ -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<SlackThreadParticipationRecord>({
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<SlackThreadParticipation
"thread-participation-state",
"Slack persistent thread participation state failed",
),
// Restoring participation must not extend its original mention-bypass window.
readTimestamp: ({ repliedAt }) => repliedAt,
},
});
src/plugin-sdk/dedupe-runtime.test.ts
+171 -3
View File
@@ -20,16 +20,19 @@ function createCache(params?: {
openStore?: () => ReturnType<typeof createMemoryStore>["store"] | undefined;
logError?: (error: unknown) => void;
readTimestamp?: (record: Record) => number | undefined;
ttlMs?: number;
maxSize?: number;
persistentMaxEntries?: number;
}) {
const backing = createMemoryStore();
const cache = createPersistentDedupeCache<Record>({
// 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(() => {
src/plugin-sdk/dedupe-runtime.ts
+27 -4
View File
@@ -10,6 +10,11 @@ type PersistentDedupeStore<TRecord> = {
lookup(key: string): Promise<TRecord | undefined>;
};
type PersistentDedupeEntryMetadata = {
key: string;
expiresAt?: number;
};
/** Dual-layer presence cache: process-memory dedupe plus best-effort persistent state. */
export type PersistentDedupeCache<TRecord> = {
/** Memory-only presence check without refreshing recency. */
@@ -67,7 +72,7 @@ export function createPersistentDedupeCache<TRecord>(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<TRecord>(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<PersistentDedupeEntryMetadata[] | undefined>;
};
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;
},