5f5b08bfda
Root cause: recordGrowEvent returned the existing row on dedupe hits, and
routeGrowEventToUserActor was called unconditionally — so every duplicate
ingest re-enqueued the same event to the actor queue, saturating it.
Fix at the persistence boundary:
- recordGrowEventWithResult returns {event, inserted} using atomic
onConflictDoNothing().returning() (row present = fresh insert, no row =
dedupe hit). No clock heuristics, no xmax probe.
- Conflict lookup tries dedupeKey first, then id PK fallback, covering both
constraint types.
- shouldRouteGrowEvent(event, inserted) gates routing to only newly-inserted,
user-resolved, pending events. Dedupe hits never re-enqueue.
- recordGrowEvent kept as thin .event wrapper — zero changes to ~10 callers.
Fix the actor run loop (mirrors workflow-run-actor.ts):
- tryStep with catch:['timeout','exhausted'] replaces step+rethrow. Per-event
failure is marked terminal in a separate record-failure step, loop advances.
- isProcessableStatus guard allows pending/processing/processed — only skips
failed/unresolved. This ensures mission reducers (actor-only) still run on
events that got synchronous projections.
- Status guard + claim moved inside tryStep's run for atomic retry unit.
- onError handler added to workflow().
Ingress callers migrated (3 paths):
- redis-consumer recordAndRoute: shouldRouteGrowEvent gate + always mark
processed after projections.
- routes/events.ts ingest: shouldRouteGrowEvent gate + always mark processed.
- v1/events/events-routes.ts /track: inserted flag for dedupe skip.
Tests: 3 new (event-dedupe-routing, user-event-actor-advance,
event-ingress-route-integration with injectable enqueue proving 5x duplicate
ingest enqueues exactly once). All existing touched-scope tests green.
Co-authored-by: independent review (2 rounds)
106 lines
4.4 KiB
TypeScript
106 lines
4.4 KiB
TypeScript
import assert from "node:assert/strict";
|
|
import { shouldRouteGrowEvent } from "../src/events/record-grow-event.js";
|
|
import type { GrowEventRow } from "../src/db/schema.js";
|
|
|
|
/**
|
|
* Regression: duplicate/idempotent ingest must not enqueue the same event twice.
|
|
*
|
|
* Before the fix, recordGrowEvent returned the existing row on a dedupe hit,
|
|
* and routeGrowEventToUserActor was called unconditionally — so every duplicate
|
|
* ingest re-enqueued the event to the actor queue, causing saturation.
|
|
*
|
|
* After the fix, routing is gated by shouldRouteGrowEvent(event, inserted):
|
|
* only newly inserted, user-resolved, pending events are routed.
|
|
*/
|
|
|
|
// ── 1. Freshly inserted, user-resolved, pending → route ──────────────────────
|
|
{
|
|
const event = { userId: "user_a", processingStatus: "pending" } as Pick<GrowEventRow, "userId" | "processingStatus">;
|
|
assert.equal(
|
|
shouldRouteGrowEvent(event, true),
|
|
true,
|
|
"newly inserted pending event with a userId must route",
|
|
);
|
|
}
|
|
|
|
// ── 2. Dedupe hit (inserted=false) → DO NOT route ────────────────────────────
|
|
// This is the core regression: a duplicate ingest returns the existing row
|
|
// with inserted=false. The gate must refuse to re-enqueue.
|
|
{
|
|
const event = { userId: "user_a", processingStatus: "processed" } as Pick<GrowEventRow, "userId" | "processingStatus">;
|
|
assert.equal(
|
|
shouldRouteGrowEvent(event, false),
|
|
false,
|
|
"dedupe hit (inserted=false) must NOT route even if status looks processable",
|
|
);
|
|
}
|
|
|
|
// ── 3. Freshly inserted but already processed → DO NOT route ─────────────────
|
|
// Edge case: the row was just inserted but somehow already processed (e.g. the
|
|
// caller did in-process projection synchronously). No re-enqueue.
|
|
{
|
|
const event = { userId: "user_a", processingStatus: "processed" } as Pick<GrowEventRow, "userId" | "processingStatus">;
|
|
assert.equal(
|
|
shouldRouteGrowEvent(event, true),
|
|
false,
|
|
"inserted but already-processed event must NOT route",
|
|
);
|
|
}
|
|
|
|
// ── 4. Freshly inserted but unresolved (no userId) → DO NOT route ────────────
|
|
{
|
|
const event = { userId: null, processingStatus: "unresolved" } as Pick<GrowEventRow, "userId" | "processingStatus">;
|
|
assert.equal(
|
|
shouldRouteGrowEvent(event, true),
|
|
false,
|
|
"unresolved event (no userId) must NOT route",
|
|
);
|
|
}
|
|
|
|
// ── 5. Freshly inserted, user-resolved, but failed → DO NOT route ────────────
|
|
// A terminal-failed event should not be re-enqueued by a duplicate ingest.
|
|
{
|
|
const event = { userId: "user_a", processingStatus: "failed" } as Pick<GrowEventRow, "userId" | "processingStatus">;
|
|
assert.equal(
|
|
shouldRouteGrowEvent(event, true),
|
|
false,
|
|
"failed event must NOT route on replay",
|
|
);
|
|
}
|
|
|
|
// ── 6. Dedupe hit on a pending event → DO NOT route ──────────────────────────
|
|
// Critical: even if the existing row is still pending (not yet processed), a
|
|
// dedupe hit means another caller already inserted it and likely already routed
|
|
// it. Re-routing would create a duplicate queue message.
|
|
{
|
|
const event = { userId: "user_a", processingStatus: "pending" } as Pick<GrowEventRow, "userId" | "processingStatus">;
|
|
assert.equal(
|
|
shouldRouteGrowEvent(event, false),
|
|
false,
|
|
"dedupe hit on a pending event must NOT re-route (already enqueued by first caller)",
|
|
);
|
|
}
|
|
|
|
// ── 7. Idempotent double-ingest simulation ───────────────────────────────────
|
|
// Simulate two calls to recordGrowEventWithResult for the same event:
|
|
// first returns inserted=true, second returns inserted=false.
|
|
// The first routes, the second does not — proving no double-enqueue.
|
|
{
|
|
const event = { userId: "user_a", processingStatus: "pending" } as Pick<GrowEventRow, "userId" | "processingStatus">;
|
|
|
|
// First ingest: newly inserted
|
|
const firstRoute = shouldRouteGrowEvent(event, true);
|
|
// Second ingest: dedupe hit
|
|
const secondRoute = shouldRouteGrowEvent(event, false);
|
|
|
|
assert.equal(firstRoute, true, "first ingest must route");
|
|
assert.equal(secondRoute, false, "second (dedupe) ingest must NOT route");
|
|
assert.ok(
|
|
firstRoute && !secondRoute,
|
|
"duplicate ingest must not enqueue twice",
|
|
);
|
|
}
|
|
|
|
console.log("event-dedupe-routing tests passed");
|
|
process.exit(0);
|