@@ -0,0 +1,176 @@
1+import { describe, expect, it, vi } from "vitest";
2+import {
3+createTelegramUpdateTracker,
4+type TelegramUpdateTrackerState,
5+} from "./bot-update-tracker.js";
6+import type { TelegramUpdateKeyContext } from "./bot-updates.js";
7+8+const updateCtx = (updateId: number): TelegramUpdateKeyContext => ({
9+update: { update_id: updateId },
10+});
11+12+async function flushTrackerMicrotasks() {
13+await Promise.resolve();
14+await Promise.resolve();
15+}
16+17+function deferred() {
18+let resolve!: () => void;
19+const promise = new Promise<void>((resolvePromise) => {
20+resolve = resolvePromise;
21+});
22+return { promise, resolve };
23+}
24+25+describe("createTelegramUpdateTracker", () => {
26+it("persists accepted offsets before earlier pending updates complete", async () => {
27+const onAcceptedUpdateId = vi.fn();
28+const tracker = createTelegramUpdateTracker({
29+initialUpdateId: 100,
30+ onAcceptedUpdateId,
31+});
32+33+const update101 = tracker.beginUpdate(updateCtx(101));
34+if (!update101.accepted) {
35+throw new Error("expected update 101 to be accepted");
36+}
37+await flushTrackerMicrotasks();
38+expect(onAcceptedUpdateId).toHaveBeenCalledWith(101);
39+40+const update102 = tracker.beginUpdate(updateCtx(102));
41+if (!update102.accepted) {
42+throw new Error("expected update 102 to be accepted");
43+}
44+tracker.finishUpdate(update102.update, { completed: true });
45+await flushTrackerMicrotasks();
46+47+expect(onAcceptedUpdateId.mock.calls.map((call) => Number(call[0]))).toEqual([101, 102]);
48+expect(tracker.getState()).toMatchObject({
49+highestAcceptedUpdateId: 102,
50+highestPersistedAcceptedUpdateId: 102,
51+highestCompletedUpdateId: 102,
52+safeCompletedUpdateId: 100,
53+pendingUpdateIds: [101],
54+failedUpdateIds: [],
55+} satisfies Partial<TelegramUpdateTrackerState>);
56+57+tracker.finishUpdate(update101.update, { completed: true });
58+expect(tracker.getState()).toMatchObject({
59+highestCompletedUpdateId: 102,
60+safeCompletedUpdateId: 102,
61+pendingUpdateIds: [],
62+} satisfies Partial<TelegramUpdateTrackerState>);
63+});
64+65+it("skips restart replays once the accepted offset is restored", async () => {
66+const onAcceptedUpdateId = vi.fn();
67+const firstProcess = createTelegramUpdateTracker({
68+initialUpdateId: 100,
69+ onAcceptedUpdateId,
70+});
71+72+const accepted = firstProcess.beginUpdate(updateCtx(101));
73+expect(accepted.accepted).toBe(true);
74+await flushTrackerMicrotasks();
75+76+const restartedProcess = createTelegramUpdateTracker({
77+initialUpdateId: Number(onAcceptedUpdateId.mock.calls.at(-1)?.[0]),
78+});
79+80+expect(restartedProcess.beginUpdate(updateCtx(101))).toEqual({
81+accepted: false,
82+reason: "accepted-watermark",
83+});
84+});
85+86+it("serializes and coalesces accepted offset persistence", async () => {
87+const firstWrite = deferred();
88+const secondWrite = deferred();
89+const writes: number[] = [];
90+const onAcceptedUpdateId = vi.fn((updateId: number) => {
91+writes.push(updateId);
92+if (updateId === 101) {
93+return firstWrite.promise;
94+}
95+return secondWrite.promise;
96+});
97+const tracker = createTelegramUpdateTracker({
98+initialUpdateId: 100,
99+ onAcceptedUpdateId,
100+});
101+102+const update101 = tracker.beginUpdate(updateCtx(101));
103+const update102 = tracker.beginUpdate(updateCtx(102));
104+const update103 = tracker.beginUpdate(updateCtx(103));
105+expect(update101.accepted).toBe(true);
106+expect(update102.accepted).toBe(true);
107+expect(update103.accepted).toBe(true);
108+109+await flushTrackerMicrotasks();
110+expect(writes).toEqual([101]);
111+expect(tracker.getState()).toMatchObject({
112+highestAcceptedUpdateId: 103,
113+highestPersistedAcceptedUpdateId: 100,
114+} satisfies Partial<TelegramUpdateTrackerState>);
115+116+firstWrite.resolve();
117+await flushTrackerMicrotasks();
118+expect(writes).toEqual([101, 103]);
119+expect(onAcceptedUpdateId).not.toHaveBeenCalledWith(102);
120+121+secondWrite.resolve();
122+await flushTrackerMicrotasks();
123+expect(tracker.getState()).toMatchObject({
124+highestPersistedAcceptedUpdateId: 103,
125+} satisfies Partial<TelegramUpdateTrackerState>);
126+});
127+128+it("keeps failed accepted updates retryable in the same process", () => {
129+const tracker = createTelegramUpdateTracker({ initialUpdateId: 200 });
130+const first = tracker.beginUpdate(updateCtx(201));
131+if (!first.accepted) {
132+throw new Error("expected first update to be accepted");
133+}
134+tracker.finishUpdate(first.update, { completed: false });
135+136+expect(tracker.getState()).toMatchObject({
137+highestAcceptedUpdateId: 201,
138+highestCompletedUpdateId: 200,
139+safeCompletedUpdateId: 200,
140+failedUpdateIds: [201],
141+} satisfies Partial<TelegramUpdateTrackerState>);
142+143+const retry = tracker.beginUpdate(updateCtx(201));
144+if (!retry.accepted) {
145+throw new Error("expected failed update retry to be accepted");
146+}
147+tracker.finishUpdate(retry.update, { completed: true });
148+149+expect(tracker.getState()).toMatchObject({
150+highestAcceptedUpdateId: 201,
151+highestCompletedUpdateId: 201,
152+safeCompletedUpdateId: 201,
153+failedUpdateIds: [],
154+} satisfies Partial<TelegramUpdateTrackerState>);
155+expect(tracker.beginUpdate(updateCtx(201))).toEqual({
156+accepted: false,
157+reason: "accepted-watermark",
158+});
159+});
160+161+it("dedupes handler dispatch separately from the accepted watermark", () => {
162+const onSkip = vi.fn();
163+const tracker = createTelegramUpdateTracker({ initialUpdateId: 300, onSkip });
164+const accepted = tracker.beginUpdate(updateCtx(301));
165+if (!accepted.accepted) {
166+throw new Error("expected update to be accepted");
167+}
168+169+expect(tracker.shouldSkipHandlerDispatch(updateCtx(301))).toBe(false);
170+expect(tracker.shouldSkipHandlerDispatch(updateCtx(301))).toBe(true);
171+expect(onSkip).toHaveBeenCalledWith("update:301");
172+173+tracker.finishUpdate(accepted.update, { completed: true });
174+expect(tracker.shouldSkipHandlerDispatch(updateCtx(301))).toBe(true);
175+});
176+});
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。