@@ -6,10 +6,20 @@ import {
66readMemoryHostEvents,
77resolveMemoryHostEventLogPath,
88} from "./memory-host-events.js";
9+import { createClaimableDedupe, createPersistentDedupe } from "./persistent-dedupe.js";
910import { createPluginSdkTestHarness } from "./test-helpers.js";
10111112const { createTempDir } = createPluginSdkTestHarness();
121314+function createDedupe(root: string, overrides?: { ttlMs?: number }) {
15+return createPersistentDedupe({
16+ttlMs: overrides?.ttlMs ?? 24 * 60 * 60 * 1000,
17+memoryMaxSize: 100,
18+fileMaxEntries: 1000,
19+resolveFilePath: (namespace) => path.join(root, `${namespace}.json`),
20+});
21+}
22+1323describe("memory host event journal helpers", () => {
1424it("appends and reads typed workspace events", async () => {
1525const workspaceDir = await createTempDir("memory-host-events-");
@@ -53,3 +63,133 @@ describe("memory host event journal helpers", () => {
5363expect(tail[0]?.type).toBe("memory.dream.completed");
5464});
5565});
66+67+describe("createPersistentDedupe", () => {
68+it("deduplicates keys, persists across instances, warms up, and checks recent keys", async () => {
69+const root = await createTempDir("openclaw-dedupe-");
70+const first = createDedupe(root);
71+expect(await first.checkAndRecord("m1", { namespace: "a" })).toBe(true);
72+expect(await first.checkAndRecord("m1", { namespace: "a" })).toBe(false);
73+expect(await first.checkAndRecord("m2", { namespace: "a" })).toBe(true);
74+75+const second = createDedupe(root);
76+expect(await second.hasRecent("m1", { namespace: "a" })).toBe(true);
77+expect(await second.hasRecent("missing", { namespace: "a" })).toBe(false);
78+expect(await second.warmup("a")).toBe(2);
79+expect(await second.checkAndRecord("m1", { namespace: "a" })).toBe(false);
80+expect(await second.checkAndRecord("m2", { namespace: "a" })).toBe(false);
81+expect(await second.checkAndRecord("m3", { namespace: "a" })).toBe(true);
82+expect(await second.checkAndRecord("m1", { namespace: "b" })).toBe(true);
83+84+const raceDedupe = createDedupe(root, { ttlMs: 10_000 });
85+const [raceFirst, raceSecond] = await Promise.all([
86+raceDedupe.checkAndRecord("race-key", { namespace: "feishu" }),
87+raceDedupe.checkAndRecord("race-key", { namespace: "feishu" }),
88+]);
89+expect(raceFirst).toBe(true);
90+expect(raceSecond).toBe(false);
91+});
92+93+it("falls back to memory-only behavior on disk errors", async () => {
94+const dedupe = createPersistentDedupe({
95+ttlMs: 10_000,
96+memoryMaxSize: 100,
97+fileMaxEntries: 1000,
98+resolveFilePath: () => path.join("/dev/null", "dedupe.json"),
99+});
100+101+expect(await dedupe.checkAndRecord("memory-only", { namespace: "x" })).toBe(true);
102+expect(await dedupe.checkAndRecord("memory-only", { namespace: "x" })).toBe(false);
103+});
104+105+it("warms empty namespaces and skips expired disk entries", async () => {
106+const root = await createTempDir("openclaw-dedupe-");
107+const emptyReader = createDedupe(root, { ttlMs: 10_000 });
108+expect(await emptyReader.warmup("nonexistent")).toBe(0);
109+110+const writer = createDedupe(root, { ttlMs: 1000 });
111+const oldNow = Date.now() - 2000;
112+expect(await writer.checkAndRecord("old-msg", { namespace: "acct", now: oldNow })).toBe(true);
113+expect(await writer.checkAndRecord("new-msg", { namespace: "acct" })).toBe(true);
114+115+const reader = createDedupe(root, { ttlMs: 1000 });
116+expect(await reader.warmup("acct")).toBe(1);
117+expect(await reader.checkAndRecord("old-msg", { namespace: "acct" })).toBe(true);
118+expect(await reader.checkAndRecord("new-msg", { namespace: "acct" })).toBe(false);
119+});
120+});
121+122+describe("createClaimableDedupe", () => {
123+it("mirrors in-flight duplicates, serializes races, and records on commit", async () => {
124+const dedupe = createClaimableDedupe({
125+ttlMs: 10_000,
126+memoryMaxSize: 100,
127+});
128+129+await expect(dedupe.claim("line:evt-1")).resolves.toEqual({ kind: "claimed" });
130+const duplicate = await dedupe.claim("line:evt-1");
131+expect(duplicate.kind).toBe("inflight");
132+133+const commit = dedupe.commit("line:evt-1");
134+await expect(commit).resolves.toBe(true);
135+if (duplicate.kind === "inflight") {
136+await expect(duplicate.pending).resolves.toBe(true);
137+}
138+await expect(dedupe.claim("line:evt-1")).resolves.toEqual({ kind: "duplicate" });
139+140+const claims = await Promise.all([dedupe.claim("line:race-1"), dedupe.claim("line:race-1")]);
141+expect(claims.filter((claim) => claim.kind === "claimed")).toHaveLength(1);
142+expect(claims.filter((claim) => claim.kind === "inflight")).toHaveLength(1);
143+144+const waitingClaim = claims.find((claim) => claim.kind === "inflight");
145+await expect(dedupe.commit("line:race-1")).resolves.toBe(true);
146+if (waitingClaim?.kind === "inflight") {
147+await expect(waitingClaim.pending).resolves.toBe(true);
148+}
149+await expect(dedupe.claim("line:race-1")).resolves.toEqual({ kind: "duplicate" });
150+});
151+152+it("rejects waiting duplicates when the active claim releases with an error", async () => {
153+const dedupe = createClaimableDedupe({
154+ttlMs: 10_000,
155+memoryMaxSize: 100,
156+});
157+158+await expect(dedupe.claim("line:evt-2")).resolves.toEqual({ kind: "claimed" });
159+const duplicate = await dedupe.claim("line:evt-2");
160+expect(duplicate.kind).toBe("inflight");
161+162+const failure = new Error("transient failure");
163+dedupe.release("line:evt-2", { error: failure });
164+if (duplicate.kind === "inflight") {
165+await expect(duplicate.pending).rejects.toThrow("transient failure");
166+}
167+await expect(dedupe.claim("line:evt-2")).resolves.toEqual({ kind: "claimed" });
168+});
169+170+it("supports persistent-backed recent checks and warmup", async () => {
171+const root = await createTempDir("openclaw-claimable-dedupe-");
172+const writer = createClaimableDedupe({
173+ttlMs: 10_000,
174+memoryMaxSize: 100,
175+fileMaxEntries: 1000,
176+resolveFilePath: (namespace) => path.join(root, `${namespace}.json`),
177+});
178+179+await expect(writer.claim("m1", { namespace: "acct" })).resolves.toEqual({ kind: "claimed" });
180+await expect(writer.commit("m1", { namespace: "acct" })).resolves.toBe(true);
181+182+const reader = createClaimableDedupe({
183+ttlMs: 10_000,
184+memoryMaxSize: 100,
185+fileMaxEntries: 1000,
186+resolveFilePath: (namespace) => path.join(root, `${namespace}.json`),
187+});
188+189+expect(await reader.hasRecent("m1", { namespace: "acct" })).toBe(true);
190+expect(await reader.warmup("acct")).toBe(1);
191+await expect(reader.claim("m1", { namespace: "acct" })).resolves.toEqual({
192+kind: "duplicate",
193+});
194+});
195+});
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。