fix(ingest): windowed reserve/publish/commit; 503 when dedupe down

[EricAndrechek]

Fixes #384. Part of #613.

Part of the remote-dedupe design. It is stacked on #625 (feat/dedupe-reserve).

What changes

Ingest runs in windows (internal/api/ingest.go). processRecord is split. prepareRecord does validate → permissions → canonicalize → resolve the dedupe id → encode. ingestWindow then drives each window of up to 256 records through three phases:

1. Reserve: one Reserve for the window's keyed records.
2. Publish: publish in record order. A Claimed record is published under mq.WithIdempotencyKey(dedupe.IdempotencyKey(tenant, key)).
3. Commit: one Commit for the claims that were published.

The single-object path is a window of one. On Pebble, each Commit is one fsync, so a batch now costs one fsync per window instead of one per record.

How a failure settles:

Where	What happens
Reserve → dedupe.ErrUnavailable (store not open; later a remote backend throttled or unreachable)	503 {"error":"dedupe store unavailable"}, Retry-After: 5. Nothing in the window is published. Earlier windows stay published and committed.
Reserve → any other error	500 dedupe failed, unchanged
Reserve → an InFlight key	The window's claims are released. 503 with the lease as Retry-After, unchanged.
Reserve → ErrDisabled	The window is published un-deduped and counted, unchanged
Publish fails at record k with mq.ErrQueueFull (definite: the broker refused it)	Records < k are committed and records ≥ k released. 503 with Retry-After: 30.
Publish fails at k with mq.ErrUnavailable (uncertain: it may have been stored)	Records < k are committed. k's claim is left to lapse with the lease, and records > k are released. 503 with Retry-After equal to the lease in whole seconds when k held a claim (a sooner retry only gets the in-flight 503), else 5.
Publish fails at k with anything else (uncertain: it may have been stored)	Records < k are committed. k's claim is left to lapse with the lease, and records > k are released. 500 publish failed, no Retry-After.
Commit fails after the publishes	The records still succeed. The failure is logged and counted per record.
A mid-body read error, or a prepare abort	The open window is dropped unpublished

The uncertain-publish rule. The earlier design released the claim on every publish failure. Now only a definite failure releases it, so a retry of an event that JetStream actually stored cannot publish a second copy:

• A retry inside the lease answers the in-flight 503.
• A retry after the lease is claimed again and republished under the same Nats-Msg-Id. JetStream drops that copy and reports success.

This relies on 2×lease + 1s ≤ duplicate window (the in-flight 503 sends the full lease as Retry-After, so an obeying client can republish up to about twice the lease after the reserve; a network backend may round expiry up by 1 s). The embedded ingest stream now sets Duplicates: mq.EmbeddedDuplicateWindow (2 min) explicitly. Existing on-disk streams get it through the CreateOrUpdateStream that SetMaxBytes already does, and a test pins that. TestIngest_DedupeLeaseFitsTheDuplicateWindow pins dedupe.DefaultLease (30 s) against the invariant.

mq (internal/mq/mq.go +14, embedded.go +11): WithIdempotencyKey(key) PublishOpt sets Nats-Msg-Id inside internal/mq, so nothing outside it names a NATS header. Every Broker implementation is conformance-tested to honour it (IdempotencyKeyDropsARepeat). dedupe (key.go): IdempotencyKey(tenant, key) is hex(sha256(stored key))[:32], which is the design's formula over the layout #625 owns.

503 sentinel. #625 already defines dedupe.ErrUnavailable in managed.go ("wrapped by a backend's error when a retry later can succeed"). This PR maps it; no new sentinel. The DynamoDB backend must wrap its throttling and timeout errors in it.

Deviations from / additions to the design

• mq.ErrQueueFull is the only definite publish failure. The design also counts "a failure before anything was sent (marshal, invalid topic)" as definite. Marshal now happens in prepare, before Reserve. An invalid topic cannot be told apart from other errors without an mq change, and treating it as uncertain only delays a retry by one lease. mq.ErrUnavailable ("cannot be reached or does not answer in time") is its own uncertain case: a timeout may have stored the event, so it gets its own 503 row above rather than the generic 500.
• A mid-body read error or a prepare abort drops the open window unpublished. Before, the records before it were published. The whole request fails either way, and publishing less before a failure means fewer duplicates on retry. TestIngest_JSONArray_SyntaxError_Fatal now asserts nothing was published. The api.md at-least-once caution says so.
• The windows run sequentially: reserve, then publish, then commit, then the next window. The design text mentions pipelining reserve with publish. That only pays off on a network backend, and it can wait for a benchmark once the DynamoDB backend is wired in.
• MockDeduplicator now answers Duplicate for a key repeated in one Reserve call, as Managed does. It also gained ErrAfter, the Reserves/Commits counters and Committed(k).
• A Reserve that fails because the request's own context ended logs at Debug; a real backend failure stays ERROR. TestIngest_Dedup_ReserveError_LogLevel pins both by exact level and message.
• Measured: each deduped publish holds about 110 B of heap in JetStream's duplicate map for the 2-minute window (200k publishes: 153 vs 44 B/msg) — about 130 MB at 10k deduped events/s.

Deliberately left to later PRs

• The DynamoDB backend (feat(dedupe): add a DynamoDB backend, tested but not yet wired #628) must wrap retryable errors in dedupe.ErrUnavailable to get the 503.
• Retention (feat(dedupe): retention per tenant and table, and an expiry sweep #633): ingestWindow passes 0 (forever), as fix(dedupe)!: reserve/commit ids, windowed ingest, retention, dynamodb #625 did.
• The dedupe.lease boot key (feat(app): choose the DynamoDB dedupe backend at boot #635): boot must refuse or warn when the lease exceeds the embedded window (mq.EmbeddedDuplicateWindow), including a test with a non-default lease once it's wired. For an external broker, it must warn when the lease exceeds the operator's duplicate_window. Today the lease is always the 30 s default.

Test evidence

• New in internal/api/ingest_window_test.go:

  • Windows of 1/255/256/257/600 records: one Reserve and one Commit per window.
  • A publish failure at k ∈ {1, 100, 256, 257, 590} (refused) and {100, 400} (uncertain). Each case checks what was committed, what was released, and that later windows were never reserved. After a refusal, a whole-batch retry publishes every record exactly once.
  • dedupe.ErrUnavailable → 503 + Retry-After: 5 in these cases: store not open, backend throttled, second window throttled (the first window stays published), and single object.
  • In-flight releases the window.
  • Result order across windows, for NDJSON and JSON-array bodies, covering rejects, repeats inside and across windows, and records with no id.
  • The bug(ingest): dedupe marks the event id before the NATS publish — a failed publish + client retry permanently drops the event #384 scenario end to end over the real embedded broker and real Pebble: refused publish → retry → exactly one event in the queue, and a third try is a duplicate.
  • Uncertain publish + retry, also end to end: the event is stored but the publish reports failure. The retry is in flight until the lease lapses, then republished and dropped by the queue: 3 events for 3 records, then all duplicates. Mutation-checked: with the idempotency key removed, the test counts 4 events.

• internal/mq: a repeated key is dropped as a success, and a pre-existing stream gains the window on its next budget apply. The stream config asserts Duplicates.

• internal/dedupe/key_test.go: IdempotencyKey is stable, 32 hex characters, and distinct across tenant, table, id, and a hashed long id.

• Fsyncs on Pebble (measured under heavy load, go test -bench BenchmarkIngest_DedupBatchOnPebble -benchtime 20x ./internal/api/), for a 1,000-record NDJSON batch:

  Window	Time per batch	Syncs per batch
  1 (per record)	5.71 s	1,000
  256	23.8 ms	4

  TestIngest_Windows_OneSyncPerWindowOnPebble pins 4 Commits for 1,000 records.

• make ci passes; every coverage gate passed; Go total coverage 94.5%; internal/api took 5.2 s under -race.

🤖 Generated with Claude Code

https://claude.ai/code/session_01G6Cz4H5k1spJAPk5CZeGds

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Advanced

Run ID: f2f3833d-74ba-4619-bf1c-2ae8ae602c26

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}