fix(ingest): event de-duplication idempotency, file rotation deduplication, and bounded eventLog

[isasadiq]

Closes #2905

Summary

Resolves the check-then-act race, rotation amnesia, process-then-record inconsistency, and O(file) ingest bottleneck in POST /api/v1/events by introducing an atomic SQLite-backed EventIdLedger, multi-file rotation-aware deduplication in FileRawEventStore, and bounded memory ring-buffer in EventProcessor.eventLog.

---

Claim Key Decision & Justification

We evaluated using the event id (event.id) versus the blockchain natural key (tx_hash, event_index). We chose event.id as the primary ledger claim key for the following reasons:

1. HTTP Ingestion Route Semantics: The API endpoint POST /api/v1/events returns per-item response entries keyed by id ({ id, status: "processed" | "duplicate" | "failed" | "rejected" }).
2. Deterministic Fallback: In routes/v1/index.ts, getStableEventId(event) derives ${event.txHash}:${event.eventIndex} whenever an explicit event.id is not supplied, naturally collapsing to the chain-guaranteed uniqueness key while supporting explicit IDs and unit test fixtures (claim("e1")).
3. Storage Alignment: SQLite's primary key index processed_events(id TEXT PRIMARY KEY) provides O(1) indexed lookups and atomic INSERT ... ON CONFLICT DO NOTHING.

---

Claim / Process / Complete State Machine

               [ Incoming Event ]
                       │
                       ▼
             eventIdLedger.claim(id)
            ┌──────────┴──────────┐
      [ changes === 1 ]     [ changes === 0 ]
            │                     │
            ▼                     ▼
        "claimed"            "duplicate" ──► Return { status: "duplicate" }
            │
            ▼
    eventProcessor.processEvent(event)
    ┌───────┴───────┐
[ Success ]     [ Throws ]
    │               │
    ▼               ▼
eventIdLedger    eventIdLedger.release(id)
.complete(id)       │
    │               ▼
    │           Return { status: "failed" } (allows clean retry)
    ▼
rawEventStore.storeEvents([event])
    ┌───────┴───────┐
[ Success ]     [ Throws ]
    │               │
    ▼               ▼
Return {        Return { status: "failed" }
status:         (Claim remains "completed" so retry
"processed" }    will NOT re-run side effects)

---

How the Four Failure Modes are Closed

1. Race Condition (Check-then-act):

   • Previous: hasEventId(id) then processEvent(event) then storeEvents([event]) with awaits in between allowed concurrent callers with identical events to both evaluate hasEventId as false and execute duplicate side effects.
   • Fixed: EventIdLedger.claim(id) executes an atomic SQLite INSERT INTO processed_events (id, status, claimed_at) VALUES (?, 'claimed', ?) ON CONFLICT(id) DO NOTHING. Exactly one concurrent caller receives "claimed" (changes === 1), and all concurrent or subsequent callers receive "duplicate".

2. Process-then-record Inconsistency:

   • Previous: If processEvent succeeded but storeEvents threw, the claim wasn't recorded. A subsequent indexer retry would re-execute side effects.
   • Fixed: Once processEvent succeeds, eventIdLedger.complete(id) transitions the event status to 'completed' BEFORE disk archive write. If storeEvents throws, the HTTP route reports failure, but subsequent retries observe "duplicate" and bypass re-execution of side effects. If processEvent throws, eventIdLedger.release(id) removes the claim so the failed event can be retried cleanly.

3. Rotation Amnesia:

   • Previous: FileRawEventStore.hasEventId() only read the active events.jsonl file. When files rotated to events-<timestamp>.jsonl, past IDs were forgotten.
   • Fixed: Ingest duplicate checking is decoupled from file scans to SQLite O(1) index. In addition, FileRawEventStore.hasEventId() now scans across all rotated generation files (events*.jsonl) with an in-memory seenIds cache, ensuring duplicate detection persists across file rotations.

4. O(file) per Event Scan:

   • Previous: hasEventId() read and JSON-parsed the entire active archive file on each event (up to 100 MB per event).
   • Fixed: Hot path duplicate checks run against the indexed processed_events SQLite table in O(1) time. Archive writes maintain a persistent FileHandle and check size before writing, removing redundant disk stats and file system re-reads.

5. Unbounded Memory (eventProcessor.eventLog):

   • Bounded eventLog to a configurable ring buffer (default 10,000 entries) via setMaxLogEntries() and automatic slicing, preventing unbounded heap growth while maintaining cursor-based pagination in listEvents().

---

Test Suite Output & Performance Benchmark

 RUN  v2.1.9 quicklendx-protocol/backend

 ✓ src/tests/ingestion-event-idempotency.test.ts (9 tests) 179ms
 ✓ src/tests/event-ingest-dedup-2905.test.ts (6 tests) 875ms
 ✓ src/tests/perf/ingest-idempotency-perf.test.ts (1 test) 3216ms

stdout | src/tests/perf/ingest-idempotency-perf.test.ts > Performance: Ingestion Event Idempotency with 200,000 pre-seeded IDs
Pre-seeding 200,000 event IDs into processed_events...
Pre-seeded 200,000 events in 2970.47ms

stdout | src/tests/perf/ingest-idempotency-perf.test.ts > Performance: Ingestion Event Idempotency with 200,000 pre-seeded IDs > AC 6: with 200,000 pre-seeded ids, ingesting a 100-event batch takes less than 250 ms

📊 Ingestion Performance:
   100-event batch ingested in 125.30ms (budget: 250ms)

 Test Files  3 passed (3)
      Tests  16 passed (16)
   Duration  6.95s

[drips-wave]

@isasadiq Great news! 🎉 Based on an automated assessment of this PR, the linked Wave issue(s) no longer count against your application limits.

You can now already apply to more issues while waiting for a review of this PR. Keep up the great work! 🚀

Learn more about application limits