@@ -2,15 +2,28 @@ import { mkdtempSync, rmSync } from "node:fs";
22import { tmpdir } from "node:os";
33import path from "node:path";
44import type { Message } from "grammy/types";
5-import { afterEach, describe, expect, it } from "vitest";
65import {
6+createPluginStateKeyedStoreForTests,
7+createPluginStateSyncKeyedStoreForTests,
8+resetPluginStateStoreForTests,
9+} from "openclaw/plugin-sdk/plugin-state-test-runtime";
10+import { afterEach, beforeEach, describe, expect, it } from "vitest";
11+import {
12+TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES,
13+TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE,
714buildTelegramMessageDispatchReplayKey,
815claimTelegramMessageDispatchReplay,
916commitTelegramMessageDispatchReplay,
1017createTelegramMessageDispatchReplayGuard,
1118releaseTelegramMessageDispatchReplay,
19+setTelegramMessageDispatchDedupeStoreForTest,
1220} from "./message-dispatch-dedupe.js";
132122+type MessageDispatchDedupeStore = NonNullable<
23+Parameters<typeof setTelegramMessageDispatchDedupeStoreForTest>[0]
24+>;
25+type SyncMessageDispatchDedupeStore = Extract<MessageDispatchDedupeStore, { entries(): unknown[] }>;
26+1427const tempDirs: string[] = [];
15281629function createStorePath(): string {
@@ -27,7 +40,19 @@ function message(params?: { chatId?: number; messageId?: number }): Message {
2740} as Message;
2841}
294243+beforeEach(async () => {
44+resetPluginStateStoreForTests({ closeDatabase: false });
45+const store = createPluginStateKeyedStoreForTests("telegram", {
46+namespace: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE,
47+maxEntries: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES,
48+}) as NonNullable<Parameters<typeof setTelegramMessageDispatchDedupeStoreForTest>[0]>;
49+await store.clear();
50+setTelegramMessageDispatchDedupeStoreForTest(store);
51+});
52+3053afterEach(() => {
54+setTelegramMessageDispatchDedupeStoreForTest(undefined);
55+resetPluginStateStoreForTests();
3156for (const dir of tempDirs.splice(0)) {
3257rmSync(dir, { recursive: true, force: true });
3358}
@@ -73,6 +98,161 @@ describe("Telegram message dispatch replay guard", () => {
7398).resolves.toEqual({ kind: "duplicate" });
7499});
75100101+it("preserves concurrent commits that share dedupe buckets", async () => {
102+const storePath = createStorePath();
103+const writer = createTelegramMessageDispatchReplayGuard({ storePath });
104+const keys = Array.from({ length: 400 }, (_, index) =>
105+JSON.stringify(["message", "1234", index + 1]),
106+);
107+108+await commitTelegramMessageDispatchReplay({
109+guard: writer,
110+accountId: "default",
111+ keys,
112+});
113+114+const reader = createTelegramMessageDispatchReplayGuard({ storePath });
115+await expect(reader.warmup("default")).resolves.toBe(keys.length);
116+});
117+118+it("falls back to same-process replay protection when plugin-state cannot open", async () => {
119+setTelegramMessageDispatchDedupeStoreForTest(undefined);
120+const errors: unknown[] = [];
121+const storePath = createStorePath();
122+const guard = createTelegramMessageDispatchReplayGuard({
123+ storePath,
124+onDiskError: (error) => errors.push(error),
125+});
126+const first = await claimTelegramMessageDispatchReplay({
127+ guard,
128+accountId: "default",
129+msg: message(),
130+});
131+if (first.kind !== "claimed") {
132+throw new Error("expected initial claim");
133+}
134+135+await expect(guard.commit(first.key, { namespace: "default" })).resolves.toBe(false);
136+137+await expect(
138+claimTelegramMessageDispatchReplay({
139+ guard,
140+accountId: "default",
141+msg: message(),
142+}),
143+).resolves.toEqual({ kind: "duplicate" });
144+await expect(guard.hasRecent(first.key, { namespace: "default" })).resolves.toBe(true);
145+expect(errors.length).toBeGreaterThan(0);
146+});
147+148+it("keeps same-process replay protection when plugin-state commit fails", async () => {
149+const failingStore = createPluginStateKeyedStoreForTests("telegram", {
150+namespace: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE,
151+maxEntries: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES,
152+}) as NonNullable<Parameters<typeof setTelegramMessageDispatchDedupeStoreForTest>[0]>;
153+setTelegramMessageDispatchDedupeStoreForTest({
154+ ...failingStore,
155+async register() {
156+throw new Error("state write failed");
157+},
158+});
159+const storePath = createStorePath();
160+const guard = createTelegramMessageDispatchReplayGuard({ storePath });
161+const first = await claimTelegramMessageDispatchReplay({
162+ guard,
163+accountId: "default",
164+msg: message(),
165+});
166+if (first.kind !== "claimed") {
167+throw new Error("expected initial claim");
168+}
169+170+await expect(guard.commit(first.key, { namespace: "default" })).resolves.toBe(false);
171+172+await expect(
173+claimTelegramMessageDispatchReplay({
174+ guard,
175+accountId: "default",
176+msg: message(),
177+}),
178+).resolves.toEqual({ kind: "duplicate" });
179+await expect(guard.hasRecent(first.key, { namespace: "default" })).resolves.toBe(true);
180+await expect(guard.warmup("default")).resolves.toBe(1);
181+});
182+183+it("keeps same-process replay protection when lookup fails after a successful commit", async () => {
184+const backingStore = createPluginStateSyncKeyedStoreForTests("telegram", {
185+namespace: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE,
186+maxEntries: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES,
187+}) as SyncMessageDispatchDedupeStore;
188+let failLookup = false;
189+setTelegramMessageDispatchDedupeStoreForTest({
190+ ...backingStore,
191+lookup(key) {
192+if (failLookup) {
193+throw new Error("state read failed");
194+}
195+return backingStore.lookup(key);
196+},
197+});
198+const storePath = createStorePath();
199+const guard = createTelegramMessageDispatchReplayGuard({ storePath });
200+const first = await claimTelegramMessageDispatchReplay({
201+ guard,
202+accountId: "default",
203+msg: message(),
204+});
205+if (first.kind !== "claimed") {
206+throw new Error("expected initial claim");
207+}
208+await expect(guard.commit(first.key, { namespace: "default" })).resolves.toBe(true);
209+210+failLookup = true;
211+212+await expect(
213+claimTelegramMessageDispatchReplay({
214+ guard,
215+accountId: "default",
216+msg: message(),
217+}),
218+).resolves.toEqual({ kind: "duplicate" });
219+});
220+221+it("keeps replay histories isolated by session store path", async () => {
222+const firstStorePath = createStorePath();
223+const secondStorePath = createStorePath();
224+const firstGuard = createTelegramMessageDispatchReplayGuard({
225+storePath: firstStorePath,
226+});
227+const first = await claimTelegramMessageDispatchReplay({
228+guard: firstGuard,
229+accountId: "default",
230+msg: message(),
231+});
232+if (first.kind !== "claimed") {
233+throw new Error("expected initial claim");
234+}
235+await commitTelegramMessageDispatchReplay({
236+guard: firstGuard,
237+accountId: "default",
238+keys: [first.key],
239+});
240+241+const secondGuard = createTelegramMessageDispatchReplayGuard({
242+storePath: secondStorePath,
243+});
244+await expect(
245+claimTelegramMessageDispatchReplay({
246+guard: secondGuard,
247+accountId: "default",
248+msg: message(),
249+}),
250+).resolves.toEqual({
251+kind: "claimed",
252+key: first.key,
253+});
254+});
255+76256it("keeps accounts isolated and releases retryable pre-dispatch claims", async () => {
77257const storePath = createStorePath();
78258const guard = createTelegramMessageDispatchReplayGuard({ storePath });
@@ -112,4 +292,34 @@ describe("Telegram message dispatch replay guard", () => {
112292key: first.key,
113293});
114294});
295+296+it("lets an in-flight duplicate retry after the first claim is released", async () => {
297+const storePath = createStorePath();
298+const guard = createTelegramMessageDispatchReplayGuard({ storePath });
299+const first = await claimTelegramMessageDispatchReplay({
300+ guard,
301+accountId: "default",
302+msg: message(),
303+});
304+if (first.kind !== "claimed") {
305+throw new Error("expected initial claim");
306+}
307+308+const duplicate = claimTelegramMessageDispatchReplay({
309+ guard,
310+accountId: "default",
311+msg: message(),
312+});
313+releaseTelegramMessageDispatchReplay({
314+ guard,
315+accountId: "default",
316+keys: [first.key],
317+error: new Error("retry"),
318+});
319+320+await expect(duplicate).resolves.toEqual({
321+kind: "claimed",
322+key: first.key,
323+});
324+});
115325});
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。