outbox.test.ts · Source

outbox.test.tspackages/collaboration/src/outbox.test.tsraw ← src16,801 B← Back to the app
1/**
2 * TASK-0095 R1 — the outbox worker (the S12 gate's named unit test):
3 *   - per-resource FIFO ordering (a later mutation to the same resource never
4 *     replays before an earlier one),
5 *   - parallel resource chains (different resources interleave, bounded
6 *     concurrency — the pool caps the in-flight chains),
7 *   - the failed-list transitions (4xx → failed; 5xx / network → backoff
8 *     retry; 2xx incl. an idempotent hit → done),
9 *   - the 24 h GC (pending only — failed items need an explicit Discard),
10 *   - the 500-item cap (oldest evicted, surfaced via onEvicted),
11 *   - retry = a FRESH key (the old key is dead),
12 *   - the backoff curve (1 s · 2^attempts, capped at 30 s),
13 *   - drain re-entrancy + the clear-on-logout rule.
14 *
15 * Deterministic: an injected clock + a fake `send` (no real timers, no network
16 * — AGENTS rule 25). The concurrency test flushes microtasks (not wall clock)
17 * to observe the pool's in-flight peak.
18 */
19import { describe, expect, it } from "vitest";
20import {
21  OUTBOX_BACKOFF_MAX_MS,
22  OUTBOX_GC_MS,
23  OUTBOX_MAX_ITEMS,
24  MemoryOutboxStore,
25  OutboxQueue,
26  backoffForAttempt,
27  generateIdempotencyKey,
28  resourceKey,
29} from "./outbox.js";
30import type { OutboxItem, OutboxResult } from "./types.js";
31 
32// --- fakes -------------------------------------------------------------------
33 
34/** Flush a handful of microtask turns (lets the drain pool pull chains). */
35async function flushMicrotasks(turns = 8): Promise<void> {
36  for (let i = 0; i < turns; i += 1) await Promise.resolve();
37}
38 
39/** A fake `send`: records call order + in-flight concurrency, optional hold. */
40function fakeSend() {
41  const records: string[] = [];
42  let active = 0;
43  let maxActive = 0;
44  let hold: Promise<void> | null = null;
45  const responses = new Map<string, OutboxResult>();
46  const send = async (item: OutboxItem): Promise<OutboxResult> => {
47    active += 1;
48    maxActive = Math.max(maxActive, active);
49    records.push(item.key);
50    try {
51      if (hold) await hold;
52      return responses.get(item.key) ?? { ok: true, status: 200 };
53    } finally {
54      active -= 1;
55    }
56  };
57  return {
58    send,
59    order: () => records,
60    maxActive: () => maxActive,
61    onKey: (key: string, result: OutboxResult) => responses.set(key, result),
62    hold: (p: Promise<void>) => {
63      hold = p;
64    },
65  };
66}
67 
68/** The queue's send options (a default send keeps the tests' focus on the
69 *  state machine; most tests inject their own via `fakeSend`). */
70type QueueOverrides = {
71  send?: (item: OutboxItem) => Promise<OutboxResult>;
72  maxConcurrentResources?: number;
73  onCountChange?: (info: { pending: number; failed: number }) => void;
74  onEvicted?: (item: OutboxItem) => void;
75};
76 
77/** A queue over an in-memory store with a manually-advanced clock. */
78function makeQueue(over: QueueOverrides = {}) {
79  let nowMs = 1_000_000;
80  const store = new MemoryOutboxStore();
81  const send =
82    over.send ?? (async (): Promise<OutboxResult> => ({ ok: true, status: 200 }));
83  const queue = new OutboxQueue({
84    store,
85    send,
86    now: () => nowMs,
87    ...(over.maxConcurrentResources !== undefined
88      ? { maxConcurrentResources: over.maxConcurrentResources }
89      : {}),
90    ...(over.onCountChange ? { onCountChange: over.onCountChange } : {}),
91    ...(over.onEvicted ? { onEvicted: over.onEvicted } : {}),
92  });
93  return {
94    queue,
95    store,
96    advance: (ms: number) => {
97      nowMs += ms;
98    },
99  };
100}
101 
102/** A fresh pending item with a controlled field set (bypasses enqueue). */
103function mkItem(over: Partial<OutboxItem> = {}): OutboxItem {
104  return {
105    key: over.key ?? generateIdempotencyKey(),
106    method: over.method ?? "POST",
107    path: over.path ?? "/api/v1/things",
108    queuedAt: over.queuedAt ?? 0,
109    resource: over.resource ?? { type: "thing", id: "t1" },
110    attempts: over.attempts ?? 0,
111    ...(over.body !== undefined ? { body: over.body } : {}),
112    ...(over.nextAttemptAt !== undefined ? { nextAttemptAt: over.nextAttemptAt } : {}),
113    ...(over.lastError !== undefined ? { lastError: over.lastError } : {}),
114  };
115}
116 
117/** Push items straight into the queue's pending/failed lists (test seam). */
118function push(queue: OutboxQueue, list: "pending" | "failed", ...items: OutboxItem[]): void {
119  const target = queue as unknown as { pending: OutboxItem[]; failed: OutboxItem[] };
120  for (const i of items) target[list].push(i);
121}
122 
123// --- pure helpers ---------------------------------------------------------------
124 
125describe("pure helpers", () => {
126  it("generateIdempotencyKey: key_ + 32 hex, unique", () => {
127    const a = generateIdempotencyKey();
128    const b = generateIdempotencyKey();
129    expect(a).toMatch(/^key_[0-9a-f]{32}$/);
130    expect(b).toMatch(/^key_[0-9a-f]{32}$/);
131    expect(a).not.toBe(b);
132  });
133 
134  it("resourceKey scopes the ordering chain per (type, id)", () => {
135    expect(resourceKey({ type: "page", id: "p1" })).toBe("page:p1");
136    expect(resourceKey({ type: "page", id: "p1" })).not.toBe(
137      resourceKey({ type: "page", id: "p2" }),
138    );
139  });
140 
141  it("backoffForAttempt: 1 s · 2^attempts, capped at 30 s", () => {
142    expect(backoffForAttempt(0)).toBe(1_000);
143    expect(backoffForAttempt(1)).toBe(2_000);
144    expect(backoffForAttempt(2)).toBe(4_000);
145    expect(backoffForAttempt(4)).toBe(16_000);
146    expect(backoffForAttempt(5)).toBe(OUTBOX_BACKOFF_MAX_MS); // 32 s → capped
147    expect(backoffForAttempt(-3)).toBe(1_000); // negative → clamped
148  });
149});
150 
151// --- enqueue + cap ---------------------------------------------------------------
152 
153describe("enqueue", () => {
154  it("creates a pending item with a fresh key + fires onCountChange", () => {
155    const counts: { pending: number; failed: number }[] = [];
156    const { queue } = makeQueue({ onCountChange: (c) => counts.push(c) });
157    const item = queue.enqueue("PATCH", "/api/v1/pages/p1", { title: "x" }, {
158      type: "page",
159      id: "p1",
160    });
161    expect(item.key).toMatch(/^key_[0-9a-f]{32}$/);
162    expect(item.attempts).toBe(0);
163    expect(queue.count()).toBe(1);
164    expect(queue.failedCount()).toBe(0);
165    expect(counts.at(-1)).toEqual({ pending: 1, failed: 0 });
166  });
167 
168  it("persists to the store on enqueue", () => {
169    const { queue, store } = makeQueue();
170    queue.enqueue("POST", "/api/v1/things", undefined, { type: "thing", id: "t1" });
171    expect(store.items).toHaveLength(1);
172  });
173 
174  it("evicts the OLDEST pending item at the 500-cap (onEvicted)", () => {
175    const evicted: OutboxItem[] = [];
176    const { queue } = makeQueue({ onEvicted: (i) => evicted.push(i) });
177    for (let i = 0; i < OUTBOX_MAX_ITEMS; i += 1) {
178      queue.enqueue("POST", "/api/v1/things", undefined, { type: "thing", id: `t${i}` });
179    }
180    expect(queue.count()).toBe(OUTBOX_MAX_ITEMS);
181    queue.enqueue("POST", "/api/v1/things", undefined, { type: "thing", id: "new" });
182    expect(queue.count()).toBe(OUTBOX_MAX_ITEMS); // capped, not +1
183    expect(evicted).toHaveLength(1);
184    expect(evicted[0]?.resource.id).toBe("t0"); // the oldest is evicted
185  });
186});
187 
188// --- drain: ordering + concurrency -------------------------------------------------
189 
190describe("drain — ordering + concurrency", () => {
191  it("replays a resource chain strictly FIFO (a later mutation never precedes an earlier one)", async () => {
192    const fake = fakeSend();
193    const { queue } = makeQueue({ send: fake.send });
194    // Two mutations to the SAME resource (queued 1 before queued 2) + one to a
195    // different resource (may interleave anywhere).
196    const a1 = mkItem({ resource: { type: "page", id: "p1" }, queuedAt: 10 });
197    const a2 = mkItem({ resource: { type: "page", id: "p1" }, queuedAt: 20 });
198    const b1 = mkItem({ resource: { type: "record", id: "r1" }, queuedAt: 15 });
199    push(queue, "pending", a1, a2, b1);
200 
201    const res = await queue.drain();
202    expect(res.sent).toBe(3);
203    expect(queue.count()).toBe(0);
204 
205    // Within the p1 chain: a1 strictly before a2. The record may be anywhere.
206    const order = fake.order();
207    expect(order.indexOf(a1.key)).toBeGreaterThanOrEqual(0);
208    expect(order.indexOf(a1.key)).toBeLessThan(order.indexOf(a2.key));
209    expect(order).toContain(b1.key);
210  });
211 
212  it("bounds parallel resource chains by maxConcurrentResources", async () => {
213    const fake = fakeSend();
214    let resolveGate: () => void = () => undefined;
215    const gate = new Promise<void>((r) => {
216      resolveGate = r;
217    });
218    fake.hold(gate);
219    const { queue } = makeQueue({
220      send: fake.send,
221      maxConcurrentResources: 2,
222    });
223    // 3 distinct resources (3 chains) with a held send → the pool (cap 2)
224    // in-flights exactly 2 chains; the 3rd waits for a free worker.
225    push(
226      queue,
227      "pending",
228      mkItem({ resource: { type: "page", id: "p1" } }),
229      mkItem({ resource: { type: "record", id: "r1" } }),
230      mkItem({ resource: { type: "comment", id: "c1" } }),
231    );
232 
233    const drainP = queue.drain();
234    await flushMicrotasks(); // let the pool pull the chains into their held send
235    expect(fake.maxActive(), "the pool caps the in-flight chains at the concurrency").toBe(2);
236 
237    resolveGate();
238    const res = await drainP;
239    expect(res.sent).toBe(3);
240    expect(queue.count()).toBe(0);
241    expect(fake.maxActive(), "the peak never exceeded the cap").toBe(2);
242  });
243});
244 
245// --- drain: failure classification ---------------------------------------------------
246 
247describe("drain — failure classification", () => {
248  it("4xx → the failed list (a user decision: Retry/Discard)", async () => {
249    const fake = fakeSend();
250    const { queue } = makeQueue({ send: fake.send });
251    const item = mkItem({ resource: { type: "page", id: "p1" } });
252    fake.onKey(item.key, { ok: false, status: 403, error: "Permission revoked" });
253    push(queue, "pending", item);
254 
255    const res = await queue.drain();
256    expect(res.failed).toBe(1);
257    expect(res.sent).toBe(0);
258    expect(queue.count()).toBe(0); // left pending
259    expect(queue.failedCount()).toBe(1);
260    const failed = queue.failedItems[0];
261    expect(failed?.key).toBe(item.key);
262    expect(failed?.lastError).toBe("Permission revoked");
263  });
264 
265  it("5xx → stays pending with attempts+1 + a backoff window (not re-sent early)", async () => {
266    const fake = fakeSend();
267    const { queue, advance } = makeQueue({ send: fake.send });
268    const item = mkItem({ resource: { type: "page", id: "p1" } });
269    fake.onKey(item.key, { ok: false, status: 503, error: "upstream down" });
270    push(queue, "pending", item);
271 
272    const res = await queue.drain();
273    expect(res.retrying).toBe(1);
274    expect(queue.count()).toBe(1);
275    expect(queue.failedCount()).toBe(0);
276    expect(queue.pendingItems[0]?.attempts).toBe(1);
277    expect(queue.pendingItems[0]?.nextAttemptAt).toBeGreaterThan(0);
278 
279    // A drain BEFORE the backoff window closes does NOT re-send.
280    const res2 = await queue.drain();
281    expect(res2.sent).toBe(0);
282    expect(res2.retrying).toBe(1);
283    expect(fake.order()).toHaveLength(1); // only the first send happened
284 
285    // Advance past the backoff → the item re-sends (same key — safe: the
286    // server never executed, so the middleware checks before run).
287    advance(OUTBOX_BACKOFF_MAX_MS + 5_000);
288    fake.onKey(item.key, { ok: true, status: 200 });
289    const res3 = await queue.drain();
290    expect(res3.sent).toBe(1);
291    expect(queue.count()).toBe(0);
292    expect(fake.order()).toHaveLength(2);
293  });
294 
295  it("network failure (send throws) → retryable, like a 5xx", async () => {
296    let calls = 0;
297    const { queue, advance } = makeQueue({
298      send: async () => {
299        calls += 1;
300        if (calls === 1) throw new Error("socket closed");
301        return { ok: true, status: 200 };
302      },
303    });
304    push(queue, "pending", mkItem());
305    const res = await queue.drain();
306    expect(res.retrying).toBe(1);
307    expect(queue.pendingItems[0]?.lastError).toBe("socket closed");
308    advance(OUTBOX_BACKOFF_MAX_MS + 5_000);
309    const res2 = await queue.drain();
310    expect(res2.sent).toBe(1);
311  });
312 
313  it("2xx with idempotentHit → done, removed (the server replayed, no re-run)", async () => {
314    const fake = fakeSend();
315    const { queue } = makeQueue({ send: fake.send });
316    const item = mkItem();
317    fake.onKey(item.key, { ok: true, status: 200, idempotentHit: true });
318    push(queue, "pending", item);
319    const res = await queue.drain();
320    expect(res.sent).toBe(1);
321    expect(queue.count()).toBe(0);
322  });
323 
324  it("is re-entrancy safe (a second concurrent drain does not re-send)", async () => {
325    const fake = fakeSend();
326    const { queue } = makeQueue({ send: fake.send });
327    push(queue, "pending", mkItem());
328    const [r1, r2] = await Promise.all([queue.drain(), queue.drain()]);
329    expect(r1.sent + r2.sent).toBe(1); // sent exactly once
330    expect(fake.order()).toHaveLength(1);
331    expect(queue.count()).toBe(0);
332  });
333});
334 
335// --- GC ------------------------------------------------------------------------------
336 
337describe("GC (24 h)", () => {
338  it("drops pending items queued > 24 h ago (on gc() + before each enqueue)", () => {
339    const { queue } = makeQueue();
340    const fresh = mkItem({ queuedAt: 1_000_000 }); // now
341    const stale = mkItem({ queuedAt: 1_000_000 - (OUTBOX_GC_MS + 1) }); // 24 h + 1 ms
342    push(queue, "pending", stale, fresh);
343    queue.gc();
344    expect(queue.pendingItems.map((i) => i.key)).toEqual([fresh.key]);
345 
346    // enqueue GCs first (a day-old queue makes room).
347    push(queue, "pending", mkItem({ queuedAt: 1_000_000 - (OUTBOX_GC_MS + 2) }));
348    const added = queue.enqueue("POST", "/api/v1/x", undefined, { type: "x", id: "x" });
349    expect(queue.pendingItems.map((i) => i.key)).toEqual([fresh.key, added.key]);
350  });
351 
352  it("does NOT auto-GC failed items (they need an explicit Discard)", () => {
353    const { queue } = makeQueue();
354    const oldFailed = mkItem({
355      queuedAt: 1_000_000 - OUTBOX_GC_MS * 2,
356      lastError: "Permission revoked",
357    });
358    push(queue, "failed", oldFailed);
359    queue.gc();
360    expect(queue.failedCount()).toBe(1);
361    expect(queue.failedItems[0]?.key).toBe(oldFailed.key);
362  });
363});
364 
365// --- retry / discard / clear ----------------------------------------------------------
366 
367describe("retry / discard / clear", () => {
368  it("retry: a failed item re-enters pending with a FRESH key (the old key is dead)", async () => {
369    const fake = fakeSend();
370    const { queue } = makeQueue({ send: fake.send });
371    const item = mkItem();
372    push(queue, "failed", {
373      ...item,
374      lastError: "Permission revoked",
375      attempts: 2,
376      nextAttemptAt: 9_999_999,
377    });
378    const failedItem = queue.failedItems[0];
379    if (failedItem === undefined) throw new Error("expected one failed item");
380    const fresh = queue.retry(failedItem);
381    expect(fresh.key).not.toBe(item.key); // fresh key
382    expect(fresh.attempts).toBe(0); // reset
383    expect(fresh.lastError).toBeUndefined();
384    expect(fresh.nextAttemptAt).toBeUndefined();
385    expect(queue.failedCount()).toBe(0);
386    expect(queue.count()).toBe(1);
387    // A drain sends the FRESH key (the middleware sees a new key → executes).
388    const res = await queue.drain();
389    expect(res.sent).toBe(1);
390    expect(fake.order()).toEqual([fresh.key]);
391  });
392 
393  it("retry of an unknown key is a no-op", () => {
394    const { queue } = makeQueue();
395    const ghost = mkItem();
396    expect(queue.retry(ghost)).toBe(ghost);
397    expect(queue.count()).toBe(0);
398  });
399 
400  it("discard: the user's explicit choice removes it (failed + pending, idempotent)", () => {
401    const { queue } = makeQueue();
402    const f = mkItem({ lastError: "x" });
403    const p = mkItem();
404    push(queue, "failed", f);
405    push(queue, "pending", p);
406    queue.discard(f);
407    queue.discard(p);
408    expect(queue.failedCount()).toBe(0);
409    expect(queue.count()).toBe(0);
410    queue.discard(f); // a double discard is safe
411  });
412 
413  it("clear: logout abandons the outbox (pending + failed)", async () => {
414    const { queue } = makeQueue();
415    push(queue, "pending", mkItem());
416    push(queue, "failed", mkItem({ lastError: "x" }));
417    await queue.clear();
418    expect(queue.count()).toBe(0);
419    expect(queue.failedCount()).toBe(0);
420  });
421});
422 
423// --- init (store load) ------------------------------------------------------------------
424 
425describe("init (persistence round-trip)", () => {
426  it("loads persisted items, splitting pending vs failed by lastError", async () => {
427    const store = new MemoryOutboxStore();
428    store.items = [mkItem({ queuedAt: 1 }), mkItem({ queuedAt: 2, lastError: "403" })];
429    const q = new OutboxQueue({
430      store,
431      send: async () => ({ ok: true, status: 200 }),
432      now: () => 1_000_000,
433    });
434    await q.init();
435    expect(q.count()).toBe(1);
436    expect(q.failedCount()).toBe(1);
437    expect(q.failedItems[0]?.lastError).toBe("403");
438  });
439});
440