feat(dedupe): add a DynamoDB backend, tested but not yet wired

[EricAndrechek]

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

Addresses #648 (a Reserve-on-error flake; closes when this stack lands on main).

What changes

dedupe.Dynamo (internal/dedupe/dynamodb.go) implements the Deduplicator contract from #625 on DynamoDB. Every tenant's keys live in one shared table. pk is a String holding the escaped key tenant/table/id (AppendKey-encoded), and is the only key; there is no sort key. Tenant(id) *Managed is the Factory, mirroring Embedded.Tenant. The client is shared, so opening a tenant's store is free.

Call	DynamoDB request	Notes
Reserve	conditional PutItem{pk, st=1, ex, tk} per key, parallel up to ReserveConcurrency (64)	condition attribute_not_exists(pk) OR ex <= :now. On a failed condition, ReturnValuesOnConditionCheckFailure: ALL_OLD returns the live item, so st says Duplicate or InFlight with no extra read. dynamodb-local 3.3.1 supports this (measured: the Duplicate cases pass).
Commit	unconditional BatchWriteItem of {pk, st=2, tk, ex?}, 25 per call, in parallel, every chunk attempted even if another fails (the records are already published)	retries UnprocessedItems with jittered backoff (doubling from 25 ms up to 200 ms, 8 rounds), then fails with ErrUnavailable. A key handed over twice is written once, because BatchWriteItem refuses duplicates. Retention 0 writes no ex.
Release	DeleteItem conditional on tk = :tk AND st = :pending, every claim attempted even if another fails	a failed condition counts as success: the claim lapsed, was re-claimed, or was committed.

• Expiry is the native TTL attribute ex, in epoch seconds, rounded up, so a lease or retention never ends early. The item is live while now < ex, which is why the condition is ex <= :now rather than the design's ex < now. The two differ only in the boundary second. Rounding up plus the half-open interval lets the suite's whole-second sleeps work without an injected clock. Correctness never depends on TTL having deleted anything. This is pinned by planting past-ex items directly and checking they are claimable.
• Reserve on error stops sending puts after the first failure and releases, by token, every put it sent that may have landed. That covers a put that errored or timed out, not only those that answered Claimed. The token is generated client-side, and a conditional delete of an item the token does not hold is a no-op. A sent put runs under the caller's own context rather than the group's, so a sibling's failure cancels only puts not yet sent, and a sent put's release waits for it to answer first — though a put cut off by the caller's own cancellation or deadline (not a sibling's failure) can still land after its release and hold the key until the lease ends. This makes "all-or-nothing on error" hold even for a put whose outcome is unknown.
• A retried put keeps its own claim. When the SDK retries a conditional PutItem whose first attempt was applied (a 500 or a connection reset after the write), the returned old item is pending and carries this put's own token, so Reserve answers Claimed instead of InFlight — which would otherwise lock the id for the whole lease and give the producer a spurious 503.
• Error mapping. ProvisionedThroughputExceeded, Throttling, RequestLimitExceeded (the SDK's throttle codes), InternalServerError, any server fault, ReplicatedWriteConflictException, deadline exceeded, and connection errors all wrap dedupe.ErrUnavailable. fix(ingest): windowed reserve/publish/commit; 503 when dedupe down #629 turns that into 503 + Retry-After. ResourceNotFound, AccessDenied and validation errors stay plain errors (500), because they are configuration bugs. The sentinel is fix(dedupe)!: reserve/commit ids, windowed ingest, retention, dynamodb #625's existing ErrUnavailable; no new sentinel was needed.
• Circuit breaker (design §7): five unavailable PutItems within a second short-circuit Reserve for a second (ErrUnavailable, no request sent). Only Reserve's puts feed it. A Release or Commit answer says nothing about whether a new claim would get through, and the conditional deletes of a failed Reserve's cleanup would otherwise reset the count on every failure. A put cancelled because a sibling failed does not touch the breaker either, and is counted as outcome canceled; a put cancelled by the caller itself (not a sibling) leaves the breaker's count unchanged too.
• Client. NewDynamo(ctx, cfg, extra...) uses config.LoadDefaultConfig, so EKS Pod Identity and IRSA need no WaveHouse code. The Region override is optional. Endpoint sets BaseEndpoint for dynamodb-local. The retryer is standard or adaptive with MaxAttempts (default 3) and full-jitter backoff, uniform over [0, min(25ms·2^attempt, ceiling)] with ceiling = Timeout/(2·(MaxAttempts−1)) — a call's retries wait at most half its Timeout (62.5 ms per retry, 125 ms total at the defaults). A throttled call ends as a max-attempts error mapped to ErrUnavailable, not DeadlineExceeded. Each call gets a Timeout deadline (default 250 ms, SDK retries included). extra passes config.LoadOptions through, which is how tests inject static credentials and a fault-injecting HTTP client.
• Connection pool: the HTTP client keeps at least max(10, ReserveConcurrency) idle connections per host, so a warm Reserve of ReserveConcurrency keys does not re-dial for each one (measured: a 64-key Reserve opened 46–54 new connections before this fix, 0 after).
• Check(ctx) is the boot check, for feat(app): choose the DynamoDB dedupe backend at boot #635 to call. It refuses a table whose key schema is not pk (HASH, S) alone, and warns (without refusing) when TTL is not enabled on ex. CreateTable(ctx) is the dev-only helper. It returns ErrCreateTableNeedsEndpoint unless Endpoint is set, creates pk S on-demand, waits for the table, and enables TTL on ex. An existing table is left alone.
• Metrics: wavehouse_dedupe_dynamodb_requests_total{op,outcome} (outcome ok / condition_failed / unavailable / canceled / error), wavehouse_dedupe_dynamodb_request_duration_seconds{op}, wavehouse_dedupe_dynamodb_unprocessed_items_total, wavehouse_dedupe_dynamodb_short_circuits_total.

Dependencies. github.com/aws/aws-sdk-go-v2 core, service/dynamodb and config, plus what those require: credentials, sts/sso/ssooidc/signin, imds, smithy-go. The design allows config because the default credential chain is the requirement. Nothing else was added, and no existing module version moved (go.mod diff is additions only).

Docs. deployment.md gains "A shared dedupe table on DynamoDB", which covers the required attributes, an example Terraform table (on-demand, pk S, TTL on ex, deletion protection, SSE) and the least-privilege IAM policy (PutItem, DeleteItem, BatchWriteItem, DescribeTable, DescribeTimeToLive; no Scan, no CreateTable). The example tags the table with neutral placeholder values. architecture.md, AGENTS.md and CHANGELOG.md also describe the backend as built but not selectable.

Not wired: deliberately left to later PRs

• feat(app): choose the DynamoDB dedupe backend at boot #635: dedupe.backend: dynamodb, the dedupe.dynamodb.* boot keys (table, region, endpoint, timeout, max_attempts, retry_mode, create_table), reserve_concurrency, the wireDedupe case, calling Check at boot, and refusing create_table without endpoint at config load. Until then no deployment can select this backend; every deployment keeps Pebble.
• fix(ingest): windowed reserve/publish/commit; 503 when dedupe down #629: 503 + Retry-After for ErrUnavailable, and the windowed ingest that makes a network backend viable for batches. Today's per-record sequential ingest would call DynamoDB serially.
• feat(dedupe): retention per tenant and table, and an expiry sweep #633: per-tenant/table retention. Ingest passes 0 today, so a committed item carries no ex and is kept forever. TTL reclaims only lapsed claims until then, and the deployment page says so.
• Real-table benchmark. internal/dedupe/dynamodb_bench_test.go (build tag dynamobench, never in CI) measures Reserve+Commit for 1 key and for a 256-key window against any table named by DEDUPE_BENCH_TABLE. It has not been run against a real AWS table, because this PR creates no AWS resources. Run against dynamodb-local only (1 key ≈ 4 ms, 256 keys ≈ 140 ms). Those figures say nothing about the service, so design §5.3's latency figures remain inferred until someone runs it against a real table.
• Creating a production table is out of scope for this repo.

Test evidence

• Integration (tests/integration/dedupe_dynamodb_test.go, against amazon/dynamodb-local:3.3.1, pinned, in-memory, no persistent volume, started beside ClickHouse in setup_test.go):
  • dedupetest.Run covers all 15 conformance cases. Factory and Peer are two separate clients over one table, with a fresh table per case and no Advance, so the suite sleeps. FailNextReserve is implemented as an HTTP-level fault on the (n+1)th PutItem.
  • 32 separate clients race one id: exactly one Claimed, 31 InFlight, then 32 Duplicate after the commit.
  • Throttling: ThrottlingException, ProvisionedThroughputExceededException and RequestLimitExceeded are injected as real DynamoDB JSON error responses. Each maps to ErrUnavailable after the SDK's retry.
  • An unreachable endpoint gives ErrUnavailable, and the breaker opens after five failures.
  • A missing table is not ErrUnavailable.
  • Check refuses a table keyed on a binary pk. CreateTable is idempotent and enables TTL on ex.
  • TTL/expiry: a pending item's ex is the lease end, a commit's ex is the retention end, retention 0 writes no ex, and items planted with a past ex (both pending and committed) are claimable.
  • CreateTable is refused without an endpoint.
  • The conformance suite (28 cases, including the Reserve-race and retried-put cases) passes against amazon/dynamodb-local:3.3.1, 10/10 runs.
• Mutation check: weakening the Reserve condition failed 11 of the conformance cases plus the 32-client test.
• Unit (internal/dedupe/dynamodb_test.go, fake API, for the paths dynamodb-local cannot produce):
  • the error classification table;
  • ALL_OLD state read into Duplicate / InFlight;
  • a failed Reserve releasing every possibly-landed put by its exact token;
  • Commit's unprocessed-item retries (3 chunks, each retried once) and dedupe of a repeated key;
  • Commit giving up with ErrUnavailable;
  • Commit attempting every chunk, and Release every claim, when one fails. The fakes honour cancellation, and both tests fail against the first commit's cancel-on-first-error code (measured, 5/5);
  • a multi-key Reserve with one throttled put: only sent puts are released, and the breaker still trips after five such Reserves;
  • Release treating a failed condition as done;
  • breaker trip, cool-down and reset;
  • Check schema and TTL paths;
  • config defaults, retry modes and CreateTable refusal;
  • expiresAt rounding.
    internal/dedupe unit coverage is 93.1%.
• .testcoverage.yml: internal/dedupe/dynamodb.go is excluded from the e2e suite gate only. The e2e binary runs Pebble dedupe, and with the file included e2e measured 58.6%, under its 60% floor. The precedent is the internal/settings e2e exclusion. Unit and integration still cover the file, and the merged total counts it.
• make ci passed locally: unit 93.3%, integration 50.9%, e2e 61.3%, Go total 94.6%. One run hit an unrelated flaky test in internal/mq (TestEmbeddedNATS_PacesTheRetriesOfAQueueThatCannotOpen, a mkdir race); it passed on re-run.

🤖 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: e5cedf13-db52-40db-adce-74a679dd1c13

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