DynamoDB relay: take the seq inside the write transaction (nxf 6j6v.r3jf)

[cabcookie]

Problem (r3jf, P1)

The DynamoDB relay allocated a stream's seq with an UpdateItem ADD before the TransactWriteItems that stores the op and its tm4k claim. When two appends overlap (A gets 10, B gets 11, B commits first), a puller at cursor 9 gets [11], stores cursor 11 (crates/sync/src/engine.rs), and never asks below 11 again. A, committed at 10 a moment later, is missing for good on that replica, and the fold watermark (fold_batch) can move past an unfolded op. SQLite and Postgres were never affected (a mutex, and an advisory lock held to commit). Spec: docs/specs/streams-without-lifetime-growth.md §2 item 1 and §4.1 (PR #58, merged into this branch).

Fix

Each append round knows a counter value c (one strongly consistent Query of seq = 0 for the first op of a push; afterwards learned without a read) and sends one TransactWriteItems:

1. Update counter: SET next_seq = :seq (s = c + 1), condition next_seq = :expected (or attribute_not_exists(next_seq) OR next_seq = :expected for 0), ReturnValuesOnConditionCheckFailure = ALL_OLD
2. Put claim at claim_seq(op_id), attribute_not_exists (unchanged tm4k item)
3. for s > 1: ConditionCheck attribute_exists(stream_id) on the op at s − 1 (the predecessor check, new after review)
4. Put op at s, attribute_not_exists

What AWS documents vs. what the design builds on (now spelled out in the module doc). AWS documents transactions as serializable against each other and a Query only as read-committed against a transaction — a transaction's items need not become visible at the same instant. The counter condition alone orders the transactions but not visibility: T_n can show its counter before its op, T_{n+1} writes op n+1 (a different item) and commits, and a reader in that window gets [n+1] (review, Integrity #1). The predecessor check puts op n's item into T_{n+1}: while T_n is still writing op n, T_{n+1} is cancelled with TransactionConflict (handled as a lost round); once op n exists, T_{n+1} may commit. That is transaction against transaction — the case AWS documents.

When the check is relaxed, and why that is safe. The check never disappears from the transaction for s > 1; it flips:

• A failed attribute_exists while the counter condition held (counter reason None) means: the counter stands at c, no transaction is writing op c (that would be a conflict), and op c is absent. No transaction of this build can still write op c — only the one that moved the counter to c could, and it writes counter and op atomically. So c is a seq an ADD-era relay burned (or still holds, rolling upgrade).
• The next round, for that same s only (with the counter now known, no read), checks attribute_not_exists on op c instead. The item is still in the transaction, so any writer of it still conflicts; if op c appears meanwhile, that check fails and the round after goes back to attribute_exists. Any other outcome (a lost round → a new slot) clears the flip.
• Seq 1 has no predecessor and no check. A floor step lands on top + 1 where top is an occupied seq, so its check finds an op.
• Net effect: a legacy gap costs one extra round, once; the round that steps over it moves the counter to an op that exists, so the stream never wedges.

Also after review:

• Per-push deadline PUSH_DEADLINE = 20 s (configurable as NXF_RELAY_PUSH_DEADLINE_SECS, read in main.rs), below the ~30 s of API Gateway / CloudFront in front of the hosted relay (E4 §9a.3) and about 2× an uncontended 500-op page. Checked between ops (every backend, OpStore::append_batch default method), between rounds and against the round in flight (DynamoDB races the round with timeout_at). A failure after the deadline answers 503. The client treats every push status except 413 the same way (pass fails, nothing marked pushed, next pass re-pushes), and the claims make the re-push safe. A round cut short may still commit; the re-push then finds its claim.
• Carry-forward: within a push the seq an op was written at is the next op's expected counter → n transactions and one read per n-op push. A wrong guess costs one lost round, which hands back the real counter.
• Occupied seqs above the counter: SeqTaken now finds the top occupied seq with one reverse Query (Limit 1) and keeps the counter known (no re-read); a run of any length costs one extra round (Integrity memory: the release runbook names the npm approval step (nxf 6j6v.c7dd) #4). checked_add makes a counter at i64::MAX a loud error instead of a wrap into the claim range.
• Contention budget now counts lost rounds in a row (reset by any other outcome); its doc no longer claims a winner per round.
• with_before_write is behind a test-seams cargo feature, enabled only by the package's dev-dependency on itself; the shipped binary and embedders never carry it.

Contracts

• Wire protocol: unchanged.
• Relay table: unchanged (counter item seq = 0 / next_seq, claim and op items; no new attribute, index or table).
• IAM: CHANGED — +dynamodb:ConditionCheckItem on the op table, REQUIRED before deploy. DynamoDB authorises each action inside TransactWriteItems separately; without it every push past a stream's first op fails with AccessDeniedException (a 500) while reads keep working. The op table's role needs Query, PutItem, UpdateItem, ConditionCheckItem; the registry table GetItem, PutItem, Query. Documented in the store_ddb.rs module doc ("IAM"), the relay guide (en/de), spec §4.1, E4, and the changelog fragment's migration note.
• New behaviour: push deadline → 503 (all backends).
• Rolling upgrade: grant ConditionCheckItem first. Mixed with an ADD-era relay nothing is overwritten and the counter never moves backwards (pinned by a test); the ordering guarantee holds once every relay writing a table runs this build. Meanwhile old relays answer more pushes with a 500 (their ADD is cancelled by TransactionConflict whenever a new relay's transaction holds the counter).

Tests

DynamoDB Local (deterministic, via the test seam):

• a_reader_is_never_handed_a_seq_above_an_append_still_in_flight — the reproduction; A is held after choosing its seq and before sending its write (not during the commit, which no test against DynamoDB Local can do). Now also asserts A's second round took the ALL_OLD hint and checked B's op.
• an_append_does_not_commit_over_a_missing_op_below_it_and_crosses_a_legacy_gap_once — counter at 4, op 4 absent: the checked round stores nothing and leaves the counter (asserted between rounds), the flipped round lands at 5, the next op at 6 in one round.
• an_older_relay_allocating_with_add_meanwhile_overwrites_nothing_and_never_moves_the_counter_back — raw UpdateItem ADD ×2 inside the hook; new op lands at 6, counter never below 5, old relay's conditional puts at 4 and 5 succeed, stream readable in order.
• a_re_push_that_lost_the_counter_race_answers_its_own_seq_and_burns_none — held re-push of o1, o2 commits, re-push answers 1 in one round, o3 lands at 3.
• a_run_of_occupied_seqs_above_the_counter_is_stepped_over_in_one_round (40 legacy ops → 41 in two rounds), a_push_carries_the_seq_it_wrote_forward_and_reads_the_counter_once, a_push_stops_at_its_deadline_and_keeps_what_it_stored.
• Legacy-op test pins the floor end to end (== Cursor(2), then fresh == Cursor(3)); dense in support/idempotence.rs adds a final fresh append == o2 + 1 (all three backends).
• soak_concurrent_writers_and_pullers_see_every_op_once_in_gap_free_seq_order — renamed and documented as a soak test: it catches an ordering bug only when the scheduler produces it.

Unit (no endpoint): layout → reason index mapping (with/without claim, with/without check), ranking incl. the predecessor check, cancellation → next step (Conflict → Lost(None), hint carried, flip direction), claim read-back (Existing/ClaimForeign/ClaimVanished), ALL_OLD hint parsing (3 branches + malformed), flip only for its seq, in-a-row budget, deadline before a round and cutting a round in flight, overflow, jitter spread (fails for an always-zero jitter). relay_http: push past its deadline → 503 with partial storage; a store fault before it → 500. main.rs: deadline env parsing.

Mutation check (run locally): dropping the predecessor check fails 2 tests (an_append_does_not_commit…, the mixed-relay test); an unconditional counter SET fails 2 (mixed-relay, reproduction).

Run locally against DynamoDB Local 3.3.1: store_ddb unit tests 40/40 (73/73 lib), dynamodb_local 37/37 (four full runs), idempotent_append 1/1, relay_http 17/17, nxf-relay bin 5/5; cargo check -p nxs-server --features dynamodb --tests clean; cargo fmt --all --check clean; cargo xtask changelog check origin/main ok. Clippy not run locally (CI's dynamodb-local job runs it with -D warnings; the type_complexity in scripted() is fixed with type aliases and the diff was scanned for other clippy-prone patterns).

Reproducing the red run on main's code

The seam does not exist on main, so the red state is a backport. It was re-run during mitigation and fails as claimed:

1. git worktree add --detach /tmp/r3jf-red 130c048 (v0.210.0, main before this PR).
2. Apply the patch below to crates/server/src/store_ddb.rs — the hook goes right after allocate_seq and before try_append_op in OpStore::append.
3. Append to crates/server/tests/dynamodb_local.rs, from this branch's copy: the helpers Attempts, recording, Seen, attempts_of, ids, Hook, gate, and the test a_reader_is_never_handed_a_seq_above_an_append_still_in_flight; import WriteAttempt from nxs_server::store_ddb.
4. NXF_RELAY_DDB_ENDPOINT=http://127.0.0.1:8000 cargo test -p nxs-server --features dynamodb --test dynamodb_local a_reader_is_never_handed
5. Result: assertion left == right failed: the reader's next pull, from the cursor it stored, still gets A: nothing was skipped — left: [] right: ["a"].

Backport patch (store_ddb.rs at 130c048)

diff --git a/crates/server/src/store_ddb.rs b/crates/server/src/store_ddb.rs
index 2c22b95..1cd974a 100644
--- a/crates/server/src/store_ddb.rs
+++ b/crates/server/src/store_ddb.rs
@@ -481,9 +481,25 @@ fn item_to_wire(item: &HashMap<String, AttributeValue>) -> StoreResult<WireOp> {
 pub struct DynamoDbOpStore {
     client: Client,
     table: String,
+    before_write: Option<std::sync::Arc<dyn Fn(&WriteAttempt) + Send + Sync>>,
+}
+
+/// BACKPORT SHIM (r3jf red run): the test seam of #59, on the ADD allocation.
+#[derive(Debug, Clone)]
+pub struct WriteAttempt {
+    pub op_id: String,
+    pub seq: i64,
+    pub expected: i64,
+    pub predecessor: Option<bool>,
+    pub counter_from: &'static str,
 }
 
 impl DynamoDbOpStore {
+    pub fn with_before_write(mut self, hook: impl Fn(&WriteAttempt) + Send + Sync + 'static) -> Self {
+        self.before_write = Some(std::sync::Arc::new(hook));
+        self
+    }
+
     /// See [`build_client`] for the constructor-argument rationale.
     pub fn new(
         table: impl Into<String>,
@@ -492,7 +508,7 @@ impl DynamoDbOpStore {
     ) -> DynamoDbOpStore {
         let table = table.into();
         let client = block_on(build_client(endpoint, region));
-        DynamoDbOpStore { client, table }
+        DynamoDbOpStore { client, table, before_write: None }
     }
 
     /// Atomically allocate the next seq for `stream_id`. A single `UpdateItem` `ADD` is an
@@ -718,6 +734,7 @@ impl OpStore for DynamoDbOpStore {
         let table = self.table.clone();
         let stream_id = stream.as_str().to_string();
         let op = op.clone();
+        let before_write = self.before_write.clone();
         block_on(async move {
             // Belt and braces (see the module doc for the full reasoning): `allocate_seq`
             // alone already guarantees no two calls are ever handed the same value, so in
@@ -731,8 +748,13 @@ impl OpStore for DynamoDbOpStore {
                 // the shape `FnMut() -> impl Future` needs to type-check.
                 let (client, table, stream_id) = (&client, table.as_str(), stream_id.as_str());
                 let op = op.clone();
+                let before_write = before_write.clone();
                 async move {
                     let seq = Self::allocate_seq(client, table, stream_id).await?;
+                    if let Some(hook) = &before_write {
+                        let attempt = WriteAttempt { op_id: op.op_id.clone(), seq, expected: seq - 1, predecessor: None, counter_from: "add" };
+                        tokio::task::block_in_place(|| hook(&attempt));
+                    }
                     let outcome = Self::try_append_op(client, table, stream_id, seq, op).await?;
                     Ok((seq, outcome))
                 }

Performance and cost

Uncontended: one transaction per op (up to four items: counter, claim, predecessor check, op) plus one counter Query per push. A transaction is billed at twice the plain write per item; the predecessor check is a transactional read of op s − 1. Contended on one stream: appends take turns on the counter, like on the other two backends. Cancelled transactions consume capacity like committed ones, so N writers racing for one stream cost up to about N× capacity per stored op; the per-stream ceiling is about one append per counter round-trip (≈30–60 appends/s by latency arithmetic). Unmeasured on real AWS — dynamodb-local serialises transactions and we had no AWS access; filed as 6j6v.wv9e. Appends to different streams never contend.

Follow-ups filed

• 6j6v.wv9e (P3 chore): measure contended throughput and capacity per stored op on real AWS.
• 6j6v.ynav (P3 chore): fold-ddb folded_through relies on DynamoDB Streams record order across items — verify or make the watermark gap-aware.
• Note on 6j6v.5vxp: truncation must keep the op the counter points at.

🤖 Generated with Claude Code

https://claude.ai/code/session_016q5uMUkZJQW4fckJoA3FME

[cabcookie]

Mitigation of the review (.nxf/review/2026-10-12T03-48-26Z.md) — head 82d97b2

Finding (source · severity)	Disposition	Notes
CI red: clippy type_complexity in scripted() (Code #1 · High)	✅ Fixed	Rounds/Answer type aliases; diff scanned for other clippy-prone patterns (redundant closure, lazy then, complex test-helper returns → aliases/boxed hook). Clippy not run locally; CI decides.
No gate evidence on this head (Test #1 · High)	✅ Fixed (pending CI)	Full DynamoDB Local suite 37/37 locally on 3.3.1 (4 runs); CI runs amazon/dynamodb-local:latest on 82d97b2.
Visibility relies on undocumented ordering (Integrity #1 · High)	✅ Fixed	Owner decision: ConditionCheck attribute_exists on op s−1 (s>1); on a proven legacy gap the next round, same seq only, checks attribute_not_exists (item stays in the tx). Module doc separates AWS-documented facts from design. IAM +dynamodb:ConditionCheckItem, required before deploy.
Module doc states premise as fact (Code #7 · Info)	✅ Fixed	Rewritten ("What AWS documents" / "What the design builds on").
Spec §4.1 / §2 row 1 / test comment stale (Code #2 · Medium)	✅ Fixed	§4.1 rewritten to what is built (Query, :expected, carry, predecessor check, IAM, budgets, deadline, rejected alternatives); row 1 cites symbols at 130c048 and is marked reproduced; dynamodb_local.rs legacy-test comment fixed; E4 updated.
Contention budget cumulative but called "in a row"; per-append ceiling undocumented (Code #3 · Medium)	✅ Fixed	lost resets on any non-Lost outcome (test the_contention_budget_counts_lost_rounds_in_a_row); ceiling documented on CONTENTION_ROUNDS; "one winner per round" claim removed.
No bound on push time (Integrity #2 · Medium)	✅ Fixed	PUSH_DEADLINE 20 s (NXF_RELAY_PUSH_DEADLINE_SECS), checked between ops/rounds and against the round in flight → 503.
Contention unmeasured on AWS, cost understated (Integrity #3 · Medium)	⏭️ Filed 6j6v.wv9e + partly fixed	No AWS access here. Cost model (cancelled tx consume capacity, ~N× per op, ~30–60/s ceiling) in module doc, spec, PR. Carry-forward implemented (one read per push).
"Failed on old code" not repeatable (Test #2 · Medium)	✅ Fixed	Backport re-run during mitigation (fails left: [] right: ["a"]); exact recipe + patch in the PR body.
Concurrency test catches bug only by chance (Test #3 · Medium)	✅ Fixed	Renamed soak_…, comment says soak, not proof.
Rolling upgrade untested; counter condition unpinned (Test #4 · Medium)	✅ Fixed	an_older_relay_allocating_with_add_meanwhile_… (raw ADD ×2 in the hook). Mutation: unconditional SET fails it.
Re-push that lost the counter race only unit-tested (Test #5 · Medium)	✅ Fixed	a_re_push_that_lost_the_counter_race_… (held re-push, density asserted).
dense proves only first re-push (Test #6 · Medium)	✅ Fixed	Final fresh o3 == o2 + 1 in support/idempotence.rs (all backends).
Floor steps one seq per stall, wedges at 32+; needless re-read (Code #4 / Integrity #4 · Low)	✅ Fixed	One reverse Query Limit 1 for the top occupied seq; counter carried as Known; 40-op test; checked_add.
Test seam on production type (Code #5 / Test #10 · Low)	✅ Fixed	test-seams feature via self dev-dependency; WriteAttempt private without it.
Duplicated counter decoding; brittle reason index (Code #6 · Low)	✅ Fixed	counter_value/counter_hint; TxLayout.
ALL_OLD hint never asserted (Test #7 · Low)	✅ Fixed	Unit test (3 branches + malformed); reproduction asserts counter_from == "hint".
Floor weakly checked end to end (Test #8 · Low)	✅ Fixed	== Cursor(2), then fresh == Cursor(3).
Jitter test passes for zero jitter; reason→outcome mapping untested (Test #9 · Low)	✅ Fixed	the_jitter_actually_spreads_its_draws; settle_cancellation/settle_claim/layout tests.
Test name overclaims (Test #11 · Info)	✅ Fixed	Doc comment: held before sending, not during commit; section note on what DynamoDB Local cannot show.
Rolling upgrade: old relays 500 more (Integrity #5 · Low)	✅ Fixed	Sentence in module doc, spec, changelog; plus the IAM step first.
tm4k idempotence holds (Integrity #6 · Info)	— Context	Unchanged; claim still outranks counter and the new predecessor reason.
IAM unchanged claim (Integrity #7 · Info)	✅ Updated	No longer true: +ConditionCheckItem, documented everywhere it is listed.
fold-ddb Streams order across items (Integrity #1, consequence)	⏭️ Filed 6j6v.ynav	Out of scope for the relay write path.

Gates (local): store_ddb unit 40/40 · lib 73/73 · dynamodb_local 37/37 (DynamoDB Local 3.3.1) · idempotent_append 1/1 · relay_http 17/17 · bin 5/5 · cargo check --features dynamodb --tests clean · cargo fmt --check clean · changelog check origin/main ok. Clippy: CI only.