fix(dedupe)!: reserve/commit ids, windowed ingest, retention, dynamodb

[EricAndrechek]

Part of #613. This PR carries the whole remote-dedupe stack: the reserve/commit contract, windowed ingest, retention, and the DynamoDB backend with its boot wiring. #629, #633, #628 and #635 were reviewed on their own and merged into this branch, and #667's test fix came in with them.

Summary

• Reserve, Commit, Release (fixes bug(dedupe): CheckAndMark is not atomic — concurrent same-id requests both pass #390, Per-table dedupe id_field (+ fix cross-table dedupe keyspace collision) #222, bug(ingest): explicit null id_field slips past require_id and dedupes every row onto "<nil>" #370). Deduplicator.CheckAndMark is replaced by a two-phase contract that every backend implements:
  • Reserve(ctx, keys, lease) claims each key atomically and answers Claimed, Duplicate or InFlight for it. It is all-or-nothing on error.
  • Commit(ctx, claims, retention) marks the claims as seen. Release(ctx, claims) gives them back, matched by token.
  • A claim that is neither committed nor released lapses after its lease, so a request that dies mid-publish never strands an id.
  • Keys are scoped per tenant and table in the readable keyenc format <tenant>/<escaped table>/<escaped id>, for example acme/clicks/evt%2D123. An id whose escaped form is over 1,024 bytes is stored as #<sha256 hex>. The same id in two tables is now two ids (Per-table dedupe id_field (+ fix cross-table dedupe keyspace collision) #222).
  • Two concurrent requests with one id now publish once (bug(dedupe): CheckAndMark is not atomic — concurrent same-id requests both pass #390). An explicit null id counts as missing (bug(ingest): explicit null id_field slips past require_id and dedupes every row onto "<nil>" #370).
  • Pebble keeps pending claims in memory in 64 locked shards beside the instance, and writes each Commit in one batch with one fsync.
  • The conformance suite internal/dedupe/dedupetest runs every backend through the same cases.
• Windowed ingest (fixes bug(ingest): dedupe marks the event id before the NATS publish — a failed publish + client retry permanently drops the event #384). Ingest runs in windows of up to 256 records. Each window makes one Reserve, publishes in record order, then makes one Commit.
  • A deduped record is published under a Nats-Msg-Id idempotency key derived from its tenant, table and id. The embedded ingest stream sets its duplicate window to 2 minutes explicitly.
  • Only a definite publish failure (the queue refused it) releases the claim. After an uncertain failure, the claim lapses with the lease, and a retry is dropped by the stream's duplicate window. The event is stored once and never lost.
  • A dedupe store that cannot answer (dedupe.ErrUnavailable, now "dedupe store unavailable") answers 503 {"error":"dedupe store unavailable"} with Retry-After: 5. It used to answer 500 dedupe failed.
  • On Pebble, a 1,000-record batch now costs 4 fsyncs instead of 1,000.
• Per-table retention (fixes Optional TTL/size bound on the dedupe store + durability docs #220). dedupe.retention in config.json, with a per-table override in dedupe.tables.<table>.retention, sets how long a committed id stays a duplicate. The default is "0", which keeps ids forever, so an existing config.json needs no change.
  • A finite retention below 2 minutes (the queue's duplicate window) is refused, not clamped.
  • A background sweep on the Pebble instance deletes expired ids and the old-format keys. It runs a minute after open and then hourly, and it never deletes a key that was committed again after the sweep read it. wavehouse_dedupe_swept_keys_total{reason} counts what it deletes.
• DynamoDB backend, selected by dedupe.backend: dynamodb. Every tenant and every process share one table, so an id ingested through one pod is a duplicate through every other.
  • Reserve is a conditional PutItem per key. Commit is BatchWriteItem with retries. Release is a conditional DeleteItem. Expiry is the native TTL attribute ex, and correctness never waits on TTL.
  • Throttling, timeouts and connection failures wrap ErrUnavailable and answer 503. A circuit breaker short-circuits Reserve for a second after five unavailable claims in a row.
  • New boot keys: dedupe.lease (the lease is now configurable, 30 s by default, at most 59 s with the embedded queue), dedupe.reserve_concurrency, and the dedupe.dynamodb.* block. create_table is refused unless endpoint is set, so WaveHouse never creates a table in AWS.
  • Boot rule: boot checks the table whether or not any tenant has dedupe on. A misconfigured table (missing, the wrong key schema, access denied) refuses boot only with a flat settings directory whose tenant has dedupe on. In every other case, including transient failures, nested directories, and no tenant with dedupe on yet, the process boots, and every tenant with dedupe on fails closed with the 503. The check is retried in the background and again right after every reload.
• Caller-cancel fix (addresses flake: Dynamo failed Reserve can leave a cancelled put's claim behind #648). When a caller cancels mid-Reserve, the puts not yet sent are skipped. A put already sent runs to its answer before it is released. Only its own call deadline can cut it off, and then it holds its key at most until the lease ends, as a crashed request's claim does.
• Test teardown (absorbs fix(test): remove the broker store after late consumer writes land #667). Tests that start the embedded broker no longer fail in t.TempDir cleanup when the broker's consumer-state flusher writes after Close. The new internal/testutil/storedir retries the removal.

Behaviour and compatibility notes

• Old-format dedupe keys are swept, not migrated. An id seen before the upgrade is accepted once more after it. The retention sweep deletes the old keys on its first pass. Nothing released depends on them.
• A dedupe backend that cannot answer returns 503 + Retry-After: 5 where it used to return 500. The SDK already retries a 503.
• A mid-body read error or a prepare failure now drops the open window unpublished. Before, the records ahead of it were published.
• The in-flight 503 sends the lease as Retry-After. That is 30 s by default, as before.

Known follow-ups

• ingest: row-by-row isolation silently drops identical rows on a deduplicating table #660: row-by-row isolation silently drops identical rows on a deduplicating table.
• bug(mq): a durable's last ack can land after Close, or never if the process exits #665: a durable's last ack can land after Close, or never if the process exits.
• chore(config): pin the lease cap's window to mq.EmbeddedDuplicateWindow once #629 and #635 land #668: this PR does its three items: config.embeddedDuplicateWindow is pinned to mq.EmbeddedDuplicateWindow by a test, the reserve_concurrency wording is updated, and the lease rule is stated once. Close it by hand after this lands.
• dedupe: cross-region dedupe on DynamoDB MRSC needs a lapsed-claim sweeper #651: cross-region dedupe on DynamoDB MRSC needs a sweeper for lapsed claims.
• ingest: accept events durably while the dedupe backend is down #652: accept events durably while the dedupe backend is down.

Tests

• Conformance: dedupetest.Run runs against Pebble twice (on an injected clock and on the real clock) and against amazon/dynamodb-local:3.3.1. It covers claim, duplicate and in-flight, release then re-claim, lease lapse, retention expiry, a 64-way concurrent Reserve, a Reserve racing a Commit, key isolation per tenant and table, hashed long ids, and all-or-nothing on a mid-call failure.
• Ingest (internal/api/ingest_window_test.go, ingest_retention_test.go):
  • window boundaries and publish failures at chosen records;
  • the 503 for an unavailable store;
  • 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 broker and Pebble;
  • the uncertain-publish retry, mutation-checked against a missing idempotency key;
  • retention reaching Commit and changing on reload, including mid-window.
• Pebble sweep (internal/dedupe/sweep_test.go): chunk boundaries, expired and old-format keys, and a Commit racing a chunk. Each case is mutation-checked.
• DynamoDB: unit tests against a fake API cover error classification, Reserve cleanup, a Reserve cancelled by its caller leaving nothing claimed, a retried put keeping its own claim, Commit retries, the breaker, and Check. Integration tests against dynamodb-local cover the conformance suite, 32 clients racing one id, throttling, an unreachable endpoint, TTL and expiry, and two app.New instances sharing seen ids through one table.
• Boot wiring (internal/app/dedupe_dynamodb_test.go, internal/config/backends_test.go): the table check in both directory shapes, the background retry, reloads that make no table call, and every new config key and refusal.
• Pinning tests: the retention floor is at least the queue's duplicate window, and config.embeddedDuplicateWindow equals mq.EmbeddedDuplicateWindow.
• make ci passes on the merged stack: every coverage gate passed. Unit 94.0%, integration 52.4%, e2e 60.2% (60% floor), Go total 95.0%.

Fixes #390. Fixes #222. Fixes #370. Fixes #384. Fixes #220. Closes #442. Closes #648. Part of #613.

🤖 Generated with Claude Code

https://claude.ai/code/session_01FyrXjhR7iDg33paioLQHFq

[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: 4d68f443-9186-4299-9c5b-83020f13d64e

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.}