diff --git a/Cargo.lock b/Cargo.lock index 0e0fae5..09185be 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2500,6 +2500,7 @@ dependencies = [ "aws-config", "aws-sdk-dynamodb", "axum", + "nxs-server", "nxs-sync", "nxs-test-support", "postgres", diff --git a/changes/r3jf-relay-seq-in-transaction.md b/changes/r3jf-relay-seq-in-transaction.md new file mode 100644 index 0000000..c55e194 --- /dev/null +++ b/changes/r3jf-relay-seq-in-transaction.md @@ -0,0 +1,12 @@ +--- +type: fixed +facade: none +--- +[en] +On a DynamoDB relay, a change made in one window could be permanently missing in another when two pushes to the same stream overlapped: a reader could be handed the later push's position before the earlier push was written, and it never asked below that position again. The relay now takes an op's position in the same write that stores the op, and that write also checks that the op one position below is already stored, so a reader is only ever handed a position once every lower one is written, as on the SQLite and Postgres relays. A re-pushed op no longer leaves a gap in the positions either. **Upgrade step for DynamoDB operators: grant the relay's role `dynamodb:ConditionCheckItem` on the op table before you deploy this version** — without it every push fails with an access-denied error (a 500) while reads keep working. The wire protocol and the relay table are unchanged; no data migration is needed. Ops a replica already skipped before the upgrade are not fetched again by that replica. Also for operators: appends to one stream now take turns, so heavy concurrent writing to a single stream is slower than before, and a push now stops after 20 seconds with a 503 (set `NXF_RELAY_PUSH_DEADLINE_SECS` to change it), which the client simply pushes again; while older relays still write the same table they answer more pushes with a 500. +[de] +Auf einem DynamoDB-Relay konnte eine Änderung aus einem Fenster in einem anderen dauerhaft fehlen, wenn sich zwei Pushes in denselben Stream überschnitten: Ein Leser konnte die Position des späteren Pushs erhalten, bevor der frühere geschrieben war, und fragte danach nie mehr unterhalb dieser Position. Das Relay vergibt die Position eines Ops jetzt im selben Schreibvorgang, der das Op speichert, und dieser prüft zugleich, dass das Op eine Position darunter schon gespeichert ist — ein Leser erhält eine Position also erst, wenn jede niedrigere geschrieben ist, wie bei den SQLite- und Postgres-Relays. Ein erneut gepushtes Op hinterlässt auch keine Lücke in den Positionen mehr. **Update-Schritt für den Betrieb mit DynamoDB: Gewähre der Rolle des Relays `dynamodb:ConditionCheckItem` auf der Op-Tabelle, bevor du diese Version ausrollst** — ohne das schlägt jeder Push mit einem Zugriffsfehler fehl (ein 500), während Lesen weiter funktioniert. Wire-Protokoll und Relay-Tabelle bleiben unverändert; eine Datenmigration ist nicht nötig. Ops, die ein Replikat vor dem Update bereits übersprungen hat, holt dieses Replikat nicht nachträglich. Ebenfalls für den Betrieb: Appends in einen Stream kommen jetzt nacheinander dran, daher ist starkes gleichzeitiges Schreiben in einen einzelnen Stream langsamer als bisher, und ein Push hört jetzt nach 20 Sekunden mit einem 503 auf (änderbar über `NXF_RELAY_PUSH_DEADLINE_SECS`), den der Client einfach wiederholt; solange ältere Relays noch in dieselbe Tabelle schreiben, antworten diese öfter mit einem 500. +[migration.en] +Manual, DynamoDB relays only, before deploying: add `dynamodb:ConditionCheckItem` on the op table to the relay's IAM role (the op table now needs `Query`, `PutItem`, `UpdateItem`, `ConditionCheckItem`). Automatic: everything else — no table or data change. SQLite and Postgres relays need nothing. +[migration.de] +Von Hand, nur DynamoDB-Relays, vor dem Ausrollen: `dynamodb:ConditionCheckItem` auf der Op-Tabelle zur IAM-Rolle des Relays hinzufügen (die Op-Tabelle braucht jetzt `Query`, `PutItem`, `UpdateItem`, `ConditionCheckItem`). Automatisch: alles andere — keine Tabellen- oder Datenänderung. SQLite- und Postgres-Relays brauchen nichts. diff --git a/crates/cli/docs/guide/de/running-a-relay.md b/crates/cli/docs/guide/de/running-a-relay.md index c8029b5..1e15ec5 100644 --- a/crates/cli/docs/guide/de/running-a-relay.md +++ b/crates/cli/docs/guide/de/running-a-relay.md @@ -327,6 +327,19 @@ Du brauchst `cmake` und eine C-Toolchain auf der Baumaschine, und du musst die b selbst anlegen — der Relay legt oder wandert sie nie. Ihre nötige Form steht oben in `crates/server/src/store_ddb.rs`. +Die IAM-Rolle des Relays braucht auf der **Op-Tabelle** `dynamodb:Query`, `dynamodb:PutItem`, +`dynamodb:UpdateItem` und **`dynamodb:ConditionCheckItem`**, auf der **Registry-Tabelle** +`dynamodb:GetItem`, `dynamodb:PutItem` und `dynamodb:Query`. `ConditionCheckItem` ist neu im ersten +Release nach v0.210.0: Jeder Schreibvorgang prüft jetzt auch das Op eine Position darunter, damit +ein Leser nie eine Position erhält, bevor jede niedrigere geschrieben ist. **Gewähre es, bevor du +diese Version ausrollst** — ohne es lehnt der Relay jeden Push mit `AccessDeniedException` ab (für +den Client ein 500), während Lesen weiter funktioniert. + +Ein Push darf höchstens 20 Sekunden schreiben; danach hört der Relay auf und antwortet 503, und der +Client pusht beim nächsten Durchlauf erneut. So bleibt ein Push auf einem stark umkämpften Stream +innerhalb der ~30 s, die API Gateway oder CloudFront vor einem serverlosen Relay erlauben. Hinter +einem kürzeren Timeout setzt du `NXF_RELAY_PUSH_DEADLINE_SECS` darunter. + Zeigst du eine **ausgelieferte** Binary auf `NXF_RELAY_BACKEND=dynamodb`, startet sie nicht und sagt dir genau das: der Backend ist eine Bau-Zeit-Fähigkeit, und dieser Bau trägt sie nicht. diff --git a/crates/cli/docs/guide/en/running-a-relay.md b/crates/cli/docs/guide/en/running-a-relay.md index 7bb2bd2..fed582e 100644 --- a/crates/cli/docs/guide/en/running-a-relay.md +++ b/crates/cli/docs/guide/en/running-a-relay.md @@ -301,6 +301,19 @@ You need `cmake` and a C toolchain on the build host, and you must create the tw yourself — the relay never creates or migrates them. Their required shape is documented at the top of `crates/server/src/store_ddb.rs`. +The relay's IAM role needs, on the **op table**, `dynamodb:Query`, `dynamodb:PutItem`, +`dynamodb:UpdateItem` and **`dynamodb:ConditionCheckItem`**, and on the **registry table** +`dynamodb:GetItem`, `dynamodb:PutItem` and `dynamodb:Query`. `ConditionCheckItem` is new in the +first release after v0.210.0: each write now also checks the op one position below it, so that a +reader is never handed a position before every lower one is written. **Grant it before you deploy +that version** — a relay without it refuses every push with `AccessDeniedException` (a 500 to the +client) while reads keep working. + +A push may write for at most 20 seconds; past that the relay stops and answers 503, and the client +pushes again on its next pass. That keeps a push on a heavily contended stream inside the ~30 s +that API Gateway or CloudFront in front of a serverless relay allow. Behind a shorter timeout, set +`NXF_RELAY_PUSH_DEADLINE_SECS` below it. + If you point a **released** binary at `NXF_RELAY_BACKEND=dynamodb`, it will refuse to start and tell you exactly this: the backend is a build-time feature, and this build does not carry it. diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index bfe1e54..8ba4687 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -37,6 +37,11 @@ postgres = [ # TransactionConflict sleeps between attempts (see store_ddb::claim_with_retry). `sha2` derives # the range key of an op's claim item, which makes append idempotent (store_ddb::claim_seq). dynamodb = ["dep:aws-sdk-dynamodb", "dep:aws-config", "dep:sha2", "tokio/time"] +# Test seams on the production types (`DynamoDbOpStore::with_before_write`, 6j6v.r3jf). Off in every +# shipped build: only this package's own dev-dependency on itself (below) turns it on, so the +# integration tests can hold an append between choosing its seq and sending its write while the +# relay binary and every embedder never carry the hook. +test-seams = [] [dependencies] # Wire types ONLY — no `engine` feature, so core's fold code never enters this graph. @@ -103,3 +108,6 @@ nxs-test-support = { path = "../test-support" } # (`nxf`/`nxs` go through the gated `nxs-test-support::cargo_bin` instead; they belong to another # package, which is exactly the case that needs the freshness gate.) assert_cmd = "2" +# This package itself, with its test seams on (see the `test-seams` feature). A dev-dependency, so +# the feature is unified into the library only when tests are built. +nxs-server = { path = ".", features = ["test-seams"] } diff --git a/crates/server/src/app.rs b/crates/server/src/app.rs index c2b3242..15cd3f3 100644 --- a/crates/server/src/app.rs +++ b/crates/server/src/app.rs @@ -4,6 +4,7 @@ //! the stores — they fold nothing, derive nothing, and reject no well-formed op (spec §2/§3). use std::sync::Arc; +use std::time::{Duration, Instant}; use axum::extract::{DefaultBodyLimit, Path, Query, State}; use axum::http::StatusCode; @@ -51,6 +52,8 @@ pub struct AppState { pub ops: SharedStore, pub registry: SharedRegistry, pub clock: Clock, + /// How long one push may write ([`PUSH_DEADLINE`] unless the relay was built with another). + pub push_deadline: Duration, } /// Default page size when a client omits `limit` — bounds the response without forcing @@ -78,6 +81,23 @@ const MAX_PRESENCE_BYTES: usize = 4 * 1024; /// body would pass [`MAX_PUSH_BYTES`] (6j6v.3gq0), so a page of large ops trips neither axis. pub const MAX_PUSH_OPS: usize = 1000; +/// How long one push may write before the relay stops it and answers 503 (`6j6v.r3jf`). The +/// deployed relay sits behind API Gateway and CloudFront (E4 continuous-sync spec §9a.3), whose +/// own request timeouts are about 30 s (API Gateway's integration limit is 29-30 s, CloudFront's +/// default origin response timeout 30 s). Past that the client has already got a 504 and will +/// re-push while the handler would still be writing, which only adds a writer to a stream that +/// is already slow. 20 s leaves a third of that margin and is still about twice what an +/// uncontended push of a full client page (500 ops, `SYNC_PAGE_LIMIT`) takes on DynamoDB, the +/// slowest backend (one transaction of ~10-20 ms per op). A relay behind a shorter timeout sets +/// `NXF_RELAY_PUSH_DEADLINE_SECS` (read at boot, `src/main.rs`) below it. +/// +/// It is checked between ops by every backend and, on DynamoDB, also between an append's rounds +/// and against the round in flight ([`OpStore::append_batch`]). The answer is 503 Service +/// Unavailable: the client treats every push status but 413 the same way — the pass fails, +/// nothing is marked pushed, and the next pass re-pushes the page, which idempotence by +/// `op_id` makes safe — so the status only tells an operator which failures were the deadline. +pub const PUSH_DEADLINE: Duration = Duration::from_secs(20); + /// Build the relay router over the op log + registry, on the system clock. Loopback, no auth (a /// later slice). The body limit guards the push route; register/read/presence bodies are tiny. pub fn app(ops: SharedStore, registry: SharedRegistry) -> Router { @@ -86,6 +106,17 @@ pub fn app(ops: SharedStore, registry: SharedRegistry) -> Router { /// [`app`] with the clock presence is stamped with named explicitly — what a test turns by hand. pub fn app_with_clock(ops: SharedStore, registry: SharedRegistry, clock: Clock) -> Router { + app_with_push_deadline(ops, registry, clock, PUSH_DEADLINE) +} + +/// [`app_with_clock`] with the push deadline named explicitly (see [`PUSH_DEADLINE`]) — what +/// `src/main.rs` builds when `NXF_RELAY_PUSH_DEADLINE_SECS` is set, and what a test shrinks. +pub fn app_with_push_deadline( + ops: SharedStore, + registry: SharedRegistry, + clock: Clock, + push_deadline: Duration, +) -> Router { Router::new() .route("/streams/:id/ops", post(append).get(read_since)) .route("/streams/:id/register", post(register)) @@ -100,6 +131,7 @@ pub fn app_with_clock(ops: SharedStore, registry: SharedRegistry, clock: Clock) ops, registry, clock, + push_deadline, }) } @@ -114,12 +146,21 @@ async fn append( return Err(StatusCode::PAYLOAD_TOO_LARGE); } let stream = StreamId(id); - for op in &req.ops { - state - .ops - .append(&stream, op) - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - } + let deadline = Instant::now() + state.push_deadline; + state + .ops + .append_batch(&stream, &req.ops, deadline) + .map_err(|e| { + // A failure once the deadline has passed is the deadline, whichever check caught it + // (between ops, between rounds, or the round in flight cut short): 503, and the + // reason in the operator's log. Anything earlier is a store fault, as before. + if Instant::now() >= deadline { + eprintln!("nxf-relay: {e}"); + StatusCode::SERVICE_UNAVAILABLE + } else { + StatusCode::INTERNAL_SERVER_ERROR + } + })?; Ok(Json(PushResponse { appended: req.ops.len(), })) diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index 7164107..7cc6605 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -10,7 +10,9 @@ use std::sync::Arc; -use nxs_server::app::{app, SharedRegistry, SharedStore}; +use nxs_server::app::{ + app_with_push_deadline, system_clock, SharedRegistry, SharedStore, PUSH_DEADLINE, +}; use nxs_server::registry::SqlitePrefixRegistry; use nxs_server::store::SqliteOpStore; @@ -160,6 +162,20 @@ fn sqlite_backends() -> (SharedStore, SharedRegistry) { (ops, registry) } +/// The push deadline from `NXF_RELAY_PUSH_DEADLINE_SECS` (whole seconds, at least 1), or +/// [`PUSH_DEADLINE`] when it is unset. Set it below the timeout of whatever fronts the relay; see +/// `PUSH_DEADLINE` for why the default is 20 s. A value that is not a positive whole number stops +/// the boot rather than silently running with the default. +fn push_deadline(raw: Option) -> std::time::Duration { + match raw { + None => PUSH_DEADLINE, + Some(raw) => match raw.trim().parse::() { + Ok(secs) if secs > 0 => std::time::Duration::from_secs(secs), + _ => panic!("NXF_RELAY_PUSH_DEADLINE_SECS must be a whole number of seconds above 0, got {raw:?}"), + }, + } +} + #[tokio::main] async fn main() { let addr = std::env::var("NXF_RELAY_ADDR").unwrap_or_else(|_| "127.0.0.1:8787".to_string()); @@ -168,7 +184,9 @@ async fn main() { .await .expect("bind relay address"); eprintln!("nxf-relay listening on http://{addr}"); - axum::serve(listener, app(ops, registry)) + let push_deadline = push_deadline(std::env::var("NXF_RELAY_PUSH_DEADLINE_SECS").ok()); + let router = app_with_push_deadline(ops, registry, system_clock(), push_deadline); + axum::serve(listener, router) .await .expect("relay server error"); } @@ -177,6 +195,19 @@ async fn main() { mod tests { use super::*; + #[test] + fn the_push_deadline_defaults_and_takes_whole_seconds() { + assert_eq!(push_deadline(None), PUSH_DEADLINE); + assert_eq!( + push_deadline(Some(" 8 ".into())), + std::time::Duration::from_secs(8) + ); + for bad in ["0", "-1", "2.5", "soon"] { + let caught = std::panic::catch_unwind(|| push_deadline(Some(bad.into()))); + assert!(caught.is_err(), "{bad:?} must stop the boot"); + } + } + /// The default build (this test's own features are the crate's defaults) must never /// recommend a feature it already carries — the regression 6j6v.0hjs describes. #[test] diff --git a/crates/server/src/store.rs b/crates/server/src/store.rs index 28bd99e..ea89ef9 100644 --- a/crates/server/src/store.rs +++ b/crates/server/src/store.rs @@ -69,8 +69,19 @@ pub(crate) fn decode_extra( }) } +/// The error [`OpStore::append_batch`] stops with once its deadline has passed: how far the push +/// got, so an operator reading the log can tell a slow stream from a stuck one. +pub(crate) fn past_deadline(stream: &StreamId, done: usize, of: usize) -> StoreError { + StoreError(format!( + "push to stream {:?} stopped at its deadline after {done} of {of} ops; the client \ + re-pushes, and the ops already stored are turned back by op id", + stream.as_str() + )) +} + /// The durable op log behind the relay. Two operations, each with explicit write -/// semantics — the atomic, per-stream monotone sequence is the real backend risk (§5). +/// semantics — the atomic, per-stream monotone sequence is the real backend risk (§5) — plus +/// [`OpStore::append_batch`], the push handler's way in, which a backend need not implement. pub trait OpStore { /// Append one opaque op and assign it the next monotone sequence **for its stream**. /// Atomic under concurrent writers (the contract the durable backends must honour). @@ -79,6 +90,32 @@ pub trait OpStore { /// so the result is never "the position the stream has reached". fn append(&self, stream: &StreamId, op: &WireOp) -> StoreResult; + /// Append one push's ops in order, stopping at the first failure — and once `deadline` has + /// passed (`6j6v.r3jf`): the relay's push handler sets it below the front door's own timeout, + /// so a push that cannot finish in time stops writing and answers, instead of writing on + /// after the client has given up on it and re-pushed (one more writer on a contended + /// stream). The ops appended before the stop stay stored; the client re-pushes the whole + /// page, and idempotence by `op_id` turns those back without growing the log. + /// + /// The default checks the deadline between ops and suits a backend whose single append is + /// local and short (SQLite, Postgres). A backend whose append can retry over the network + /// overrides it to check between its rounds too, and may carry state from one op of the push + /// to the next (DynamoDB carries the seq it wrote as the counter value to expect). + fn append_batch( + &self, + stream: &StreamId, + ops: &[WireOp], + deadline: std::time::Instant, + ) -> StoreResult<()> { + for (done, op) in ops.iter().enumerate() { + if std::time::Instant::now() >= deadline { + return Err(past_deadline(stream, done, ops.len())); + } + self.append(stream, op)?; + } + Ok(()) + } + /// One page of ops with `seq > cursor`, ordered by seq, at most `limit`. Never returns the /// whole stream at once. /// diff --git a/crates/server/src/store_ddb.rs b/crates/server/src/store_ddb.rs index 2c22b95..86bfa35 100644 --- a/crates/server/src/store_ddb.rs +++ b/crates/server/src/store_ddb.rs @@ -19,7 +19,7 @@ //! //! | `seq` | item | attributes | //! |--------|------------------------------|----------------------------------------------------------| -//! | `0` | the per-stream counter | `next_seq` (N) — the highest seq handed out | +//! | `0` | the per-stream counter | `next_seq` (N) — the highest seq taken | //! | `>= 1` | one op | one attribute per [`WireOp`] field (see below) | //! | `< 0` | one op's claim (`6j6v.tm4k`) | `op_id` (S), `op_seq` (N) — the seq that op is stored at | //! @@ -68,8 +68,20 @@ //! envelope attributes, so item-to-`WireOp` conversion fails on them. A reader must constrain //! the query to `seq > 0`, as [`DynamoDbOpStore::read_since`] does. The same holds for a //! consumer of a DynamoDB stream on this table: an item with `seq <= 0` is bookkeeping, never -//! an op. Sequence numbers are unique and monotone but **not** dense: see the allocation -//! section below. +//! an op. Sequence numbers are unique and monotone, and a seq is visible only once every +//! lower seq this build wrote is (see the allocation section below, with what that rests on). +//! +//! # IAM — what the relay's role needs +//! +//! **Upgrade step, required since `6j6v.r3jf`: grant `dynamodb:ConditionCheckItem` on the op table +//! BEFORE deploying a relay of this build.** Every append past a stream's first op now carries a +//! `ConditionCheck` in its transaction (below), and DynamoDB authorises each action inside a `TransactWriteItems` +//! as its own IAM action; a role without `ConditionCheckItem` gets every push refused with +//! `AccessDeniedException`, surfaced as a 500, while reads keep working. +//! +//! - **Op table:** `dynamodb:Query`, `dynamodb:PutItem`, `dynamodb:UpdateItem`, +//! `dynamodb:ConditionCheckItem`. +//! - **Registry table:** `dynamodb:GetItem`, `dynamodb:PutItem`, `dynamodb:Query`. //! //! # Design notes //! @@ -85,25 +97,107 @@ //! from inside an axum handler; see `block_on`'s doc for why that specific shape (channel //! handoff, not a second `Runtime::block_on`) is required. //! -//! **Sequence allocation.** `append` allocates a stream's next seq with a single -//! `UpdateItem` `ADD` on the counter item (`seq = 0`, attribute `next_seq`) — atomic and -//! read-free, so two concurrent allocations can never observe the same value (an earlier -//! read-then-`PutItem` shape did exactly that, racily). The op itself then goes in with a -//! put conditioned on `attribute_not_exists(stream_id)` at that `(stream_id, seq)`: -//! belt and braces, not redundant defenses of the same thing. The counter (the belt) -//! already guarantees *uniqueness* — no other allocation call can ever be handed this same -//! value again. The condition (the braces) guarantees *safety of the write*: if a seq were -//! ever computed twice anyway (a bug in the allocation path, a replayed request), the -//! second put fails loudly instead of silently overwriting the first writer's op. On -//! `ConditionalCheckFailedException` the whole allocate-then-put is retried from scratch, -//! bounded by [`APPEND_ATTEMPTS`]. +//! **Sequence allocation: a seq is taken by the write that commits it** (`6j6v.r3jf`). Each +//! append round knows a counter value `c` (read from the counter item, `seq = 0`, attribute +//! `next_seq`, absent counting as 0 — or learned without a read, below) and sends ONE +//! `TransactWriteItems` that +//! +//! 1. sets `next_seq` to `s = c + 1`, conditioned on `next_seq` still being `c`; +//! 2. puts the op's claim (below), conditioned on its key being free; +//! 3. for `s > 1`, checks the op at `s - 1` with a `ConditionCheck` — `attribute_exists`; +//! 4. puts the op at `s`, conditioned on that key being free. +//! +//! A round that finds the counter moved has lost to a concurrent append and retries from the +//! counter the failed check hands back (`ReturnValuesOnConditionCheckFailure = ALL_OLD`), after a +//! short random pause ([`append_with_retry`]). +//! +//! Why not an `UpdateItem` `ADD` ahead of the write, as before: an `ADD` hands out the seq before +//! the op exists. Writer A got 10, writer B got 11, B committed first, and a reader at 9 was +//! handed `[11]` and stored cursor 11 — so A, committed at 10 a moment later, sat below every +//! cursor that reader would send again, missing for good, and a fold watermark moved past it too. +//! +//! *What AWS documents, and nothing more:* `TransactWriteItems` is **serializable** against other +//! transactions (and against single-item writes) — they behave as if they ran one at a time, and a +//! transaction that touches an item another transaction is still writing is cancelled with +//! `TransactionConflict`. A `Query` is only **read-committed** against a transaction: it never sees +//! uncommitted data, but a transaction's items need not all become visible to it at the same +//! instant, so a `Query` can see one item of a committing transaction in its new state and another +//! still in its old one. Each per-item reason in a cancellation is reported as that item's own +//! outcome (`None` when its condition held). +//! +//! *What the design builds on those two facts:* +//! +//! - **The counter condition orders the transactions.** The transaction that takes `s` expects +//! the counter at `c = s - 1`, which only the transaction that took `c` can have set. By +//! serializability, `T_s` runs after `T_c`. That alone does NOT order their *visibility* to a +//! `Query`: `T_c` can show its counter before its op, `T_s` can read that counter, write op `s` +//! — a different item — and commit while op `c` is still invisible, and a reader in that window +//! is handed `[s]`. (Review of PR #59, Integrity #1.) +//! - **The `ConditionCheck` on op `s - 1` closes that window.** It puts op `c`'s item into +//! `T_s`, so `T_s` and `T_c` now share an item: while `T_c` is still writing op `c`, `T_s` is +//! cancelled with `TransactionConflict` (a lost round, retried); once `T_c` is done, op `c` +//! exists and `T_s` may commit. So op `s` cannot become visible before op `s - 1` has been +//! written — transaction against transaction, the case AWS documents — and by induction a +//! strongly consistent `read_since` that returns seq `n` has every op below `n` this build +//! wrote. That is the order SQLite (a mutex) and Postgres (an advisory lock held to commit) +//! give, and what a reader storing its cursor unconditionally, and a snapshot at P (spec +//! `streams-without-lifetime-growth` §4.1), need. +//! - **When op `s - 1` is legitimately absent, the check flips, it is never dropped.** A failed +//! `attribute_exists` with the counter condition holding means: the counter stands at `c`, no +//! transaction is writing op `c` (that would have been a conflict), and op `c` is not there. No +//! transaction of this build can still write it — the only one that could is the one that moved +//! the counter to `c`, and it writes counter and op together. So `c` is a seq an earlier build's +//! `ADD` burned (or one an `ADD`-era relay still holds, see the rolling upgrade). The next round, +//! for the SAME `s` only, checks op `c` with `attribute_not_exists` instead: still an item of the +//! transaction, so a writer of op `c` still conflicts, and should op `c` appear meanwhile the check +//! fails and the round after goes back to `attribute_exists`. Seq 1 has no predecessor and no +//! check. A floor step (below) takes `s` one above an occupied seq, so its check finds that op. A +//! legacy gap therefore costs one extra round, once, and never wedges the stream: the round that +//! steps over it moves the counter to an op that exists. +//! - **The op's own condition stays as the brace.** A seq occupied by an op the counter does not +//! account for (a stream written before the counter existed, an item put there out of band) is +//! stepped over, never overwritten: the round finds the highest occupied seq above the counter +//! with ONE reverse `Query` (`Limit 1`) and takes the seq after it, so a run of any length costs +//! one extra round, not one per seq. Seqs are `i64`; a counter at `i64::MAX` is a loud error. +//! +//! If a future change deletes ops (truncation below a snapshot, spec slice 2), it must keep the op +//! the counter points at, or the next append takes one extra round over the gap — correct, but it +//! is the kind of gap the flip above treats as legacy. +//! +//! **What it costs.** Uncontended: one transaction per op (four items: counter, claim, the +//! predecessor check, op) plus one strongly consistent `Query` of the counter for the first op of +//! a push — the rest of the push carries the seq it wrote forward as the counter to expect +//! ([`OpStore::append_batch`]), so a push of `n` ops costs `n` transactions and one read. +//! DynamoDB bills a transaction at twice the plain write per item; the `ConditionCheck` is a read +//! of op `s - 1` billed at transactional read rates, sized by that op. Appends to ONE stream +//! serialise on the counter item, as on the other two backends, and appends to different streams +//! never contend. +//! +//! Under contention a lost round is a whole cancelled transaction, and **cancelled transactions +//! consume capacity like committed ones**, so `N` writers racing for one stream cost up to about +//! `N` times the capacity per op stored. Two transactions can also both be cancelled in one round, +//! so contention is not "one winner per round". The per-stream ceiling is about one append per +//! counter round-trip — on the order of 30-60 appends per second per stream on real AWS by +//! latency arithmetic; **unmeasured** (filed as a chore, see the PR). [`CONTENTION_ROUNDS`] bounds +//! the rounds one append loses in a row, [`APPEND_ATTEMPTS`] the rounds that fail for any other +//! reason, and the push deadline ([`OpStore::append_batch`], `app::PUSH_DEADLINE`) bounds the +//! wall clock of a whole push; each ends in a loud `StoreError`, and the client re-pushes, which +//! the claims make safe. +//! +//! **Density.** A seq is taken only together with its op, so a stream written by this build has +//! no gaps: neither a re-push nor a lost round nor the SDK's retry of a write whose response was +//! lost consumes one (the retry finds the claim taken). Streams written before it carry the +//! gaps the `ADD` left. Nothing downstream depends on density — `read_since` is `seq > cursor` +//! and the client follows the returned cursor — so read it as a property, not a contract. //! -//! `UpdateItem` + `ADD` is *not* idempotent, though, so **gaps in the sequence are normal in -//! production**: any retry of that call — including one the SDK performs internally after a -//! throttle or a lost response — burns a seq that no op will ever occupy. Nothing downstream -//! depends on density (`read_since` is `seq > cursor` and the client follows the returned -//! cursor), but do not read the acceptance suite's contiguity assertion as a contract: it -//! holds only because DynamoDB Local never throttles. +//! **Rolling upgrade.** First grant `dynamodb:ConditionCheckItem` (above). A relay of an earlier +//! build still allocates with `ADD` and can still expose a seq ahead of an uncommitted one while it +//! serves; mixed with this build nothing is overwritten (both condition the op's put), the counter +//! never moves backwards (this build's `SET` is conditioned on the value it read, and an `ADD` +//! landing in between fails that condition), and the ordering guarantee holds once every relay +//! writing a table runs this build. Meanwhile the old relays answer more pushes with a 500: their +//! `ADD` on the counter is itself cancelled with `TransactionConflict` whenever one of this build's +//! transactions holds the counter, and the old build does not retry that. //! //! **Per-item size ceiling.** DynamoDB rejects any item over **400 KB**, so a single op //! whose `value` exceeds roughly that is a hard `StoreError` here where SQLite and Postgres @@ -123,36 +217,39 @@ //! //! What it costs and where it differs from the other two backends: //! -//! * **The re-push still burns a seq.** The seq is allocated before the transaction finds the -//! claim taken, so a duplicate leaves a gap where SQLite leaves none — and every round a writer -//! loses to a concurrent transaction burns one more, which is why such a round first reads the -//! claim and then backs off before it allocates again. Gaps are already normal here (above), -//! and nothing reads density. +//! * **The claim rides the append's transaction.** It is an item of the same +//! `TransactWriteItems` that moves the counter and puts the op, so a re-push is turned back by +//! its taken claim before the counter moves: it consumes no seq, as on SQLite and Postgres. +//! A taken claim outranks a moved counter, so a re-push that also lost the counter race still +//! answers with its stored seq. At most four items per transaction, far below DynamoDB's 100. //! * **Ops stored before the claim items existed have none.** Re-pushing such an op stores it //! once more, with a claim this time; every later re-push is then absorbed. //! * **The claim key is a 63-bit hash, so two op ids can share it.** The claim records the op //! id it was written for, so a shared key is detected rather than mistaken for a duplicate: //! the second op is then stored WITHOUT a claim, exactly as before this change, and only it //! loses the dedupe. Nothing is ever lost or overwritten by a collision. -//! * **No new table, index or IAM action.** The claim lives in the op table under the same key -//! schema, it is written with `PutItem` inside the transaction and read with `Query`, the two -//! actions the op table's role already has. Every op write now costs a transaction, which -//! DynamoDB bills at twice the plain write, for two items. +//! * **No new table or index.** The claim lives in the op table under the same key schema, it is +//! written with `PutItem` inside the transaction and read with `Query`. (The transaction's +//! `ConditionCheck`, `6j6v.r3jf`, is what needs the one new IAM action — see "IAM" above.) use std::collections::HashMap; use std::time::{Duration, Instant}; use aws_sdk_dynamodb::error::ProvideErrorMetadata; -use aws_sdk_dynamodb::operation::put_item::PutItemError; use aws_sdk_dynamodb::operation::transact_write_items::TransactWriteItemsError; -use aws_sdk_dynamodb::types::{AttributeValue, Put, ReturnValue, TransactWriteItem}; +use aws_sdk_dynamodb::types::{ + AttributeValue, ConditionCheck, Put, ReturnValuesOnConditionCheckFailure, TransactWriteItem, + Update, +}; use aws_sdk_dynamodb::Client; use nxs_sync::protocol::{Cursor, MachineHello, RegisterOutcome, StreamId}; use nxs_sync::wire::WireOp; use crate::presence::{MachineRecord, MachinesPage, PresenceStore}; use crate::registry::{candidate_prefix, PrefixRegistry, REASSIGN_ATTEMPTS}; -use crate::store::{decode_extra, encode_extra, page_limit, OpStore, StoreError, StoreResult}; +use crate::store::{ + decode_extra, encode_extra, page_limit, past_deadline, OpStore, StoreError, StoreResult, +}; /// The DynamoDB SDK is async; `OpStore`/`PrefixRegistry` are sync (they were shaped for the /// sync SQLite/Postgres drivers). Bridging on a DEDICATED runtime — never the caller's — is @@ -334,111 +431,445 @@ pub fn claim_seq(op_id: &str) -> i64 { -((u64::from_be_bytes(head) >> 1) as i64) - 1 } -/// Bound on allocate-then-put retries after a `ConditionalCheckFailedException`. A named, -/// finite bound rather than an unbounded loop: exhausting it means something is -/// structurally wrong (e.g. a bug that keeps recomputing the same seq), and that must -/// surface as a loud `StoreError`, never spin the caller's thread forever. +/// Bound on the append rounds that end WITHOUT another append having committed in between: a +/// taken seq, a predecessor check that did not hold, a throttled transaction, a claim key found +/// taken by another op id or read back absent. Counted over the whole append. Exhausting it means +/// something is structurally wrong or the table is saturated, and that must surface as a loud +/// `StoreError`, never spin the caller's thread forever. const APPEND_ATTEMPTS: u32 = 32; -/// Outcome of one conditional-put attempt at an allocated seq. `Collided` means the -/// caller should allocate a fresh seq and try again — never overwrite what is already -/// there. `Existing` means the op was already stored, at the seq it carries: the append is -/// done, and that seq is its answer. `Conflicted` means the transaction lost to a concurrent one -/// or to capacity: wait ([`claim_backoff`]), then allocate afresh. `PartialEq`/`Debug` exist for -/// [`classify_put_error`]'s unit tests. +/// Bound on the rounds one append loses IN A ROW to a concurrent append of the same stream — the +/// counter moved under it, or DynamoDB cancelled it with a `TransactionConflict`. Any round that +/// ends otherwise resets the count. Separate from [`APPEND_ATTEMPTS`], and larger, because a lost +/// round usually means another writer committed: the stream is making progress, only this writer +/// is losing the race for the next seq. +/// +/// A backstop, not a promise about fairness: two transactions can also cancel each other in the +/// same round, so "one writer wins every round" does not hold on real AWS. What bounds the wall +/// clock of a push is its deadline ([`OpStore::append_batch`]). Without one (a bare +/// [`OpStore::append`]), one append's ceiling is about `CONTENTION_ROUNDS` x (one transaction, +/// ~10-30 ms on AWS, plus at most 16 ms of jitter) per run of losses — 5-12 s — plus at most +/// `APPEND_ATTEMPTS - 1` stalled rounds, each of whose SDK calls [`OPERATION_TIMEOUT`] caps. +const CONTENTION_ROUNDS: u32 = 256; + +/// What one append round reads and writes: the counter value it expects to find (`0` when the +/// stream has no counter yet) and the seq it moves the counter to and puts the op at, in ONE +/// transaction. `PartialEq`/`Debug` exist for the unit tests. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct Slot { + expected: i64, + seq: i64, +} + +impl Slot { + /// The slot after `counter`: one above it, or one above `floor` when that is higher. `floor` + /// is the highest seq an earlier round found occupied although the counter had not reached + /// it — a stream written before the counter existed, or an item put there out of band. + /// Stepping over such a seq is safe: it already holds a committed op. A seq past `i64::MAX` + /// is a loud error, never a wrap into the negative range the claims live in. + fn after(counter: i64, floor: i64) -> StoreResult { + let seq = counter.max(floor).checked_add(1).ok_or_else(|| { + StoreError(format!( + "append: the next seq would pass i64::MAX (counter {counter}, highest occupied \ + seq {floor})" + )) + })?; + Ok(Slot { + expected: counter, + seq, + }) + } + + /// The seq the predecessor check looks at: the one below, `None` for a stream's first seq. + fn predecessor(self) -> Option { + if self.seq > 1 { + Some(self.seq - 1) + } else { + None + } + } +} + +/// Where a round's counter value comes from. Only [`Counter::Unknown`] costs a read. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Counter { + /// Not known: read the counter item first. + Unknown, + /// Handed back by the previous round's failed counter check (`ALL_OLD`). + Hint(i64), + /// The value the previous round's counter condition held at, in a round that failed on + /// something else (a taken seq, a predecessor check). + Known(i64), + /// The seq the previous op of the same push wrote ([`OpStore::append_batch`]). + Carried(i64), +} + +impl Counter { + /// The name a test seam reports it by. + fn source(self) -> &'static str { + match self { + Counter::Unknown => "read", + Counter::Hint(_) => "hint", + Counter::Known(_) => "known", + Counter::Carried(_) => "carried", + } + } +} + +/// How a round checks the op below the seq it takes (module doc, "Sequence allocation"). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Predecessor { + /// Seq 1: there is nothing below. + None, + /// `attribute_exists` — every round but the one below. + Exists, + /// `attribute_not_exists` — the round right after an `Exists` check at the same seq found the + /// op below absent: a legacy gap. Still an item of the transaction, so a writer of it still + /// conflicts. + Absent, +} + +/// What the next round of [`append_with_retry`] starts from: the counter value to expect, the +/// highest seq found occupied above the counter, whether the op still goes in with its claim, and +/// the one seq whose predecessor a round just found absent (`gap`, cleared by every other outcome). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct Round { + counter: Counter, + floor: i64, + claimed: bool, + gap: Option, +} + +impl Round { + fn first(counter: Counter) -> Round { + Round { + counter, + floor: 0, + claimed: true, + gap: None, + } + } + + /// The predecessor check for `slot`: none for seq 1, `Absent` for the seq a previous round + /// found the op below missing at, `Exists` otherwise. + fn predecessor_check(&self, slot: Slot) -> Predecessor { + match slot.predecessor() { + None => Predecessor::None, + Some(_) if self.gap == Some(slot.seq) => Predecessor::Absent, + Some(_) => Predecessor::Exists, + } + } +} + +/// Outcome of one append round. `PartialEq`/`Debug` exist for the unit tests. #[derive(Debug, PartialEq, Eq)] enum PutOutcome { - Written, - Collided, + /// The op is stored at this seq, and the counter stands at it. + Written(i64), + /// The op was already stored, at this seq: the append is done, and that is its answer. Existing(i64), - Conflicted, + /// Another append of this stream committed first (the counter moved), or is committing (a + /// `TransactionConflict`). Carries the counter value the failed check saw, when DynamoDB + /// handed it back, so the next round needs no read. + Lost(Option), + /// The counter held at `counter`, but the seq is occupied by an op the counter does not + /// account for; `top` is the highest occupied seq above the counter. + SeqTaken { counter: i64, top: i64 }, + /// The counter held at `counter`, but the op below `seq` was not as the check expected: + /// `absent` under an `Exists` check (a legacy gap: flip the check for this seq), present under + /// an `Absent` one (it appeared meanwhile: check for it again). + Predecessor { + counter: i64, + seq: i64, + absent: bool, + }, + /// Throttled or over capacity: wait ([`claim_backoff`]), then try again. + Throttled, + /// The claim key belongs to another op id: store this op without a claim (module doc). + ClaimForeign, + /// The claim was reported taken and read back absent: try again. + ClaimVanished, } -/// Classify a failed `PutItem` on the op table: a `ConditionalCheckFailedException` is the -/// ONLY reason to retry (something already occupies this exact `(stream_id, seq)`, so the -/// caller must allocate a fresh seq — never overwrite it); every other failure, from -/// throttling to a missing table to an over-400 KB item, is a real fault that must surface. -/// -/// Pulled out as a pure function over the typed error — the same move -/// [`classify_cancellation`] makes for the registry — because this branch is *unreachable* -/// against a live DynamoDB: `allocate_seq`'s atomic counter never hands two callers the same -/// value, so no acceptance test can provoke a collision. A pure function can be handed one -/// directly. -/// -/// Matching the SDK's typed variant, never a substring of its `Display`, is what keeps a -/// renamed or reworded service message from silently turning a hard error into a retry. -fn classify_put_error(err: PutItemError) -> StoreResult { - match err { - PutItemError::ConditionalCheckFailedException(_) => Ok(PutOutcome::Collided), - other => Err(ddb_err(other)), +/// How an append settled: written by this call, or found already stored. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Settled { + Written(i64), + Existing(i64), +} + +impl Settled { + fn seq(self) -> i64 { + match self { + Settled::Written(seq) | Settled::Existing(seq) => seq, + } } } -/// Drive one allocate-then-put cycle until it lands, at most [`APPEND_ATTEMPTS`] times. +/// The error an append stops with at the push deadline. `app.rs` answers it 503 by the clock, not +/// by this text; [`OpStore::append_batch`] replaces it with how far the push got. +fn deadline_passed(stream_id: &str) -> StoreError { + StoreError(format!( + "append to stream {stream_id:?} stopped at the push deadline" + )) +} + +/// Drive append rounds until one settles, within [`APPEND_ATTEMPTS`], [`CONTENTION_ROUNDS`] and, +/// when given, `deadline` — checked before every round and raced against the round in flight. A +/// round cut short may still commit on DynamoDB's side; the client's re-push then finds its claim. /// -/// Extracted from `append` and made generic over the cycle for the same reason -/// [`classify_put_error`] is a free function: against a live DynamoDB the loop provably runs -/// exactly once (the atomic counter guarantees a unique seq), so neither the retry branch nor -/// the exhaustion error can be reached by an acceptance test. A stand-in cycle that collides -/// on demand reaches both. -async fn append_with_retry(stream_id: &str, mut cycle: F) -> StoreResult +/// Generic over the round so the retry policy — which outcome retries with what, after how long, +/// within which budget — is unit-tested without a table: DynamoDB Local serialises transactions, +/// so most of these branches never occur against it. +async fn append_with_retry( + stream_id: &str, + start: Counter, + deadline: Option, + mut round: F, +) -> StoreResult where - F: FnMut() -> Fut, - Fut: std::future::Future>, + F: FnMut(Round) -> Fut, + Fut: std::future::Future>, { - for attempt in 0..APPEND_ATTEMPTS { - let (seq, outcome) = cycle().await?; + let mut next = Round::first(start); + let (mut stalls, mut lost) = (0u32, 0u32); + loop { + let outcome = match deadline { + None => round(next).await?, + Some(deadline) => { + if Instant::now() >= deadline { + return Err(deadline_passed(stream_id)); + } + let until = tokio::time::Instant::from_std(deadline); + match tokio::time::timeout_at(until, round(next)).await { + Ok(outcome) => outcome?, + Err(_) => return Err(deadline_passed(stream_id)), + } + } + }; + next.gap = None; match outcome { - PutOutcome::Written => return Ok(Cursor(seq)), - PutOutcome::Existing(stored) => return Ok(Cursor(stored)), - PutOutcome::Collided => continue, - // Writers re-pushing the same op all contend on its one claim key; retrying in - // lockstep would burn a seq per round for each of them. Doubling, as the registry. - PutOutcome::Conflicted => tokio::time::sleep(claim_backoff(attempt)).await, + PutOutcome::Written(seq) => return Ok(Settled::Written(seq)), + PutOutcome::Existing(seq) => return Ok(Settled::Existing(seq)), + PutOutcome::Lost(seen) => { + lost += 1; + if lost >= CONTENTION_ROUNDS { + return Err(StoreError(format!( + "append: lost {CONTENTION_ROUNDS} rounds in a row to concurrent appends \ + of stream {stream_id:?}; the stream is under more write contention than \ + one append waits out, retry the push" + ))); + } + next.counter = seen.map_or(Counter::Unknown, Counter::Hint); + tokio::time::sleep(contention_jitter(lost)).await; + } + stalled => { + lost = 0; + stalls += 1; + if stalls >= APPEND_ATTEMPTS { + return Err(StoreError(format!( + "append: exhausted {APPEND_ATTEMPTS} attempts writing an op to stream \ + {stream_id:?} without a successful conditional write (last: {stalled:?})" + ))); + } + next.counter = Counter::Unknown; + match stalled { + PutOutcome::SeqTaken { counter, top } => { + next.counter = Counter::Known(counter); + next.floor = next.floor.max(top); + } + PutOutcome::Predecessor { + counter, + seq, + absent, + } => { + next.counter = Counter::Known(counter); + if absent { + next.gap = Some(seq); + } + } + PutOutcome::ClaimForeign => next.claimed = false, + PutOutcome::Throttled => { + let base = claim_backoff(stalls - 1); + tokio::time::sleep(base + jitter(base)).await; + } + _ => {} + } + } } } - Err(StoreError(format!( - "append: exhausted {APPEND_ATTEMPTS} attempts allocating a seq for stream \ - {stream_id:?} without a successful conditional put" - ))) } -/// Why an append's `TransactWriteItems` (index 0 = the claim, index 1 = the op — see -/// [`DynamoDbOpStore::try_append_op`]) was cancelled, when the cancellation is an outcome -/// rather than a fault. -#[derive(Debug, PartialEq, Eq)] +/// A uniformly random duration in `[0, cap]`. Seeded from `RandomState`, whose keys `std` draws +/// from the OS once per thread and steps per instance — enough to de-synchronise writers, which +/// is all a backoff needs, without a `rand` dependency. +fn jitter(cap: Duration) -> Duration { + use std::hash::BuildHasher; + let cap_us = cap.as_micros().max(1) as u64; + let draw = std::collections::hash_map::RandomState::new().hash_one(Instant::now()); + Duration::from_micros(draw % (cap_us + 1)) +} + +/// The pause before the next round after losing one to a concurrent append: full jitter up to +/// 1, 2, 4, 8, then 16 ms. Small and capped, unlike a throttle backoff, because a lost round means +/// the stream moved on — waiting longer would only hand the next seq to whoever did not wait — +/// and random, so the losers do not all come back for the same seq at the same instant. +fn contention_jitter(lost: u32) -> Duration { + jitter(Duration::from_millis(1u64 << lost.min(4))) +} + +/// Which item of an append's `TransactWriteItems` sits at which index — the order +/// [`DynamoDbOpStore::try_append_op`] adds them in: the counter, then the claim when the op +/// carries one, then the predecessor check when the seq has one, last the op. DynamoDB reports +/// cancellation reasons positionally, so this is the one place the order is decoded. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct TxLayout { + claimed: bool, + checked: bool, +} + +impl TxLayout { + const COUNTER: usize = 0; + + fn claim(self) -> Option { + self.claimed.then_some(1) + } + + fn predecessor(self) -> Option { + self.checked.then_some(1 + usize::from(self.claimed)) + } + + fn op(self) -> usize { + 1 + usize::from(self.claimed) + usize::from(self.checked) + } +} + +/// Why an append's `TransactWriteItems` was cancelled, when the cancellation is an outcome rather +/// than a fault. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] enum AppendCancellation { /// The claim's key is taken: this op — or one whose id shares the key — is stored. Claimed, - /// The op's seq is taken: allocate a fresh one. + /// The counter is not at the value this round expected: another append committed first. + CounterMoved, + /// The counter held but the op's seq is occupied: step over it. SeqTaken, - /// Lost to concurrency or capacity: try again, with a fresh seq. - Retryable, + /// The counter held but the predecessor check did not. + PredecessorFailed, + /// A concurrent transaction on one of the items: another append is committing. + Conflict, + /// Throttled or over capacity. + Throttled, } -/// Classify a cancelled append from its positional per-item codes. Pure for the same reason -/// [`classify_cancellation`] is: DynamoDB Local serialises transactions, so the transient -/// codes never reach an acceptance test. A taken claim outranks everything: it is definitive, -/// and it means the op is already stored, so the seq question no longer matters. +/// Classify a cancelled append from its positional per-item codes, laid out as `layout` says. +/// Pure for the same reason [`classify_cancellation`] is: DynamoDB Local serialises transactions, +/// so the transient codes never reach an acceptance test. +/// +/// A taken claim outranks everything: it is definitive, and it means the op is already stored, so +/// the seq question no longer matters. A moved counter outranks the rest: the next round starts +/// from the new counter and so picks another seq and another predecessor anyway. A taken seq +/// outranks a failed predecessor check: the next round steps above the occupied seqs, whose top +/// one is then the predecessor, and it exists. fn classify_append_cancellation( - claim_code: Option<&str>, - op_code: Option<&str>, + layout: TxLayout, + codes: &[Option<&str>], ) -> Result { - if claim_code == Some("ConditionalCheckFailed") { + const CCF: Option<&str> = Some("ConditionalCheckFailed"); + let at = |index: Option| index.and_then(|i| codes.get(i).copied().flatten()); + let counter = at(Some(TxLayout::COUNTER)); + let claim = at(layout.claim()); + let predecessor = at(layout.predecessor()); + let op = at(Some(layout.op())); + if claim == CCF { return Ok(AppendCancellation::Claimed); } - if op_code == Some("ConditionalCheckFailed") { + if counter == CCF { + return Ok(AppendCancellation::CounterMoved); + } + if op == CCF { return Ok(AppendCancellation::SeqTaken); } - if is_retryable_code(claim_code) || is_retryable_code(op_code) { - return Ok(AppendCancellation::Retryable); + if predecessor == CCF { + return Ok(AppendCancellation::PredecessorFailed); + } + let codes = [counter, claim, predecessor, op]; + if codes.contains(&Some("TransactionConflict")) { + return Ok(AppendCancellation::Conflict); + } + if codes.into_iter().any(is_retryable_code) { + return Ok(AppendCancellation::Throttled); } Err(format!( - "append transaction cancelled for a reason other than a taken claim or seq \ - (claim item code: {claim_code:?}, op item code: {op_code:?})" + "append transaction cancelled for a reason other than a moved counter, a taken claim or \ + seq, or a predecessor check (counter item code: {counter:?}, claim item code: {claim:?}, \ + predecessor check code: {predecessor:?}, op item code: {op:?})" )) } +/// What a classified cancellation asks for next: an outcome right away, or one more read first. +#[derive(Debug, PartialEq, Eq)] +enum Settle { + Outcome(PutOutcome), + /// A taken claim: read it, then [`settle_claim`]. + ReadClaim, + /// A taken seq: find the highest occupied seq above the counter. + FindTop, +} + +/// Map a cancellation to the next step. `hint` is the counter the failed counter check handed +/// back ([`counter_hint`]). Pure, so the mapping is unit-tested; the two reads it may ask for are +/// [`DynamoDbOpStore::try_append_op`]'s. +fn settle_cancellation( + cancellation: AppendCancellation, + slot: Slot, + check: Predecessor, + hint: Option, +) -> Settle { + match cancellation { + AppendCancellation::Claimed => Settle::ReadClaim, + AppendCancellation::SeqTaken => Settle::FindTop, + AppendCancellation::CounterMoved => Settle::Outcome(PutOutcome::Lost(hint)), + AppendCancellation::PredecessorFailed => Settle::Outcome(PutOutcome::Predecessor { + counter: slot.expected, + seq: slot.seq, + absent: check == Predecessor::Exists, + }), + // Another transaction holds the counter, the predecessor (the append below this one, + // still writing its op) or the claim (a racing re-push of this very op): it is + // committing. The next round reads the counter and, should that transaction have been + // this op's own, finds the claim taken. + AppendCancellation::Conflict => Settle::Outcome(PutOutcome::Lost(None)), + AppendCancellation::Throttled => Settle::Outcome(PutOutcome::Throttled), + } +} + +/// What the claim read back after a taken claim means: the op's existing seq when it was written +/// for this op id, a foreign claim (store without one) when for another, and — read back absent — +/// a race with something that deletes claims (nothing does today): try again. +fn settle_claim(found: Option<(String, i64)>, op_id: &str) -> PutOutcome { + match found { + Some((claimed_for, stored)) if claimed_for == op_id => PutOutcome::Existing(stored), + Some(_) => PutOutcome::ClaimForeign, + None => PutOutcome::ClaimVanished, + } +} + +/// The counter item's value: `next_seq`, or 0 when the item or the attribute is absent (a stream +/// nothing has been appended to by a counter-aware build). +fn counter_value(item: Option<&HashMap>) -> StoreResult { + match item { + Some(item) if item.contains_key("next_seq") => parse_n(item, "next_seq"), + _ => Ok(0), + } +} + +/// The counter a failed counter check handed back (`ALL_OLD`): `None` when DynamoDB returned no +/// item — the next round then reads it — otherwise its value as [`counter_value`] reads it. +fn counter_hint(item: Option<&HashMap>) -> StoreResult> { + item.map(|item| counter_value(Some(item))).transpose() +} + fn item_to_wire(item: &HashMap) -> StoreResult { Ok(WireOp { envelope_version: parse_n(item, "envelope_version")?, @@ -481,6 +912,39 @@ fn item_to_wire(item: &HashMap) -> StoreResult { pub struct DynamoDbOpStore { client: Client, table: String, + /// Always `None` outside a build with the `test-seams` feature, which alone can set it. + before_write: Option, +} + +/// The test seam's hook, see `DynamoDbOpStore::with_before_write`. +type BeforeWrite = std::sync::Arc; + +#[cfg(feature = "test-seams")] +pub use seam::WriteAttempt; +#[cfg(not(feature = "test-seams"))] +use seam::WriteAttempt; + +/// The test seam's argument. Public only through the `test-seams` feature (the re-export above): +/// in a shipped build it is private to this file. +mod seam { + /// One append attempt, as the test seam sees it right before the attempt sends its write. + /// Without the feature nothing can install a hook, so nothing reads the fields. + #[derive(Debug, Clone)] + #[cfg_attr(not(feature = "test-seams"), allow(dead_code))] + pub struct WriteAttempt { + pub op_id: String, + /// The seq the attempt takes. + pub seq: i64, + /// The counter value the attempt's condition expects. + pub expected: i64, + /// The predecessor check: `None` for seq 1, `Some(true)` for `attribute_exists` on + /// `seq - 1`, `Some(false)` for `attribute_not_exists` (the flip over a legacy gap). + pub predecessor: Option, + /// Where `expected` came from: `"read"`, `"hint"` (handed back by a failed counter check), + /// `"known"` (the previous round's counter condition held) or `"carried"` (the seq the + /// previous op of the same push wrote). + pub counter_from: &'static str, + } } impl DynamoDbOpStore { @@ -492,36 +956,119 @@ impl DynamoDbOpStore { ) -> DynamoDbOpStore { let table = table.into(); let client = block_on(build_client(endpoint, region)); - DynamoDbOpStore { client, table } - } - - /// Atomically allocate the next seq for `stream_id`. A single `UpdateItem` `ADD` is an - /// uninterruptible read-modify-write inside DynamoDB itself — unlike an earlier - /// `GetItem` + `PutItem`, there is no window between reading and writing the counter - /// for a second allocation to observe the same value. `ADD` on a missing `next_seq` - /// attribute starts from 0, so a stream's first append needs no separate - /// initialisation step, and `ReturnValues::UpdatedNew` hands back the post-increment - /// value directly, so no follow-up `GetItem` is needed to learn what this call got. - async fn allocate_seq(client: &Client, table: &str, stream_id: &str) -> StoreResult { - let updated = client - .update_item() + DynamoDbOpStore { + client, + table, + before_write: None, + } + } + + /// Test seam (`6j6v.r3jf`), only in a build with the `test-seams` feature — which only this + /// package's own tests turn on, so the relay binary and every embedder never carry it. `hook` + /// is called on every append attempt with what the attempt is about to write, after it has + /// decided its seq and before it sends the write. A test blocks in it to hold one append + /// between those two points while another append commits, or acts as an older relay in that + /// window. It runs on the bridge runtime inside `block_in_place`, so blocking in it is + /// allowed. + #[cfg(feature = "test-seams")] + 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 + } + + /// Read the stream's counter: the highest seq written, `0` when the stream has none yet. A + /// strongly consistent `Query` on the counter's exact key rather than a `GetItem`, for the + /// reason [`Self::read_claim`] gives: the op table's role needs no action it did not have. + async fn read_counter(client: &Client, table: &str, stream_id: &str) -> StoreResult { + let resp = client + .query() .table_name(table) - .key("stream_id", AttributeValue::S(stream_id.to_string())) - .key("seq", AttributeValue::N(COUNTER_SEQ.to_string())) - .update_expression("ADD next_seq :one") - .expression_attribute_values(":one", AttributeValue::N("1".to_string())) - .return_values(ReturnValue::UpdatedNew) + .key_condition_expression("stream_id = :sid AND seq = :counter") + .expression_attribute_values(":sid", AttributeValue::S(stream_id.to_string())) + .expression_attribute_values(":counter", AttributeValue::N(COUNTER_SEQ.to_string())) + .consistent_read(true) .send() .await .map_err(ddb_err)?; + counter_value(resp.items().first()) + } - let attrs = updated.attributes().ok_or_else(|| { - StoreError( - "UpdateItem on the counter item returned no attributes for ReturnValues::UpdatedNew" - .into(), - ) - })?; - parse_n::(attrs, "next_seq") + /// The highest seq occupied above `counter`, or `None` when nothing is. ONE strongly + /// consistent reverse `Query` with `Limit 1`, so a run of occupied seqs above the counter — + /// of any length — costs one read, not one round per seq (review of PR #59, Integrity #4). + async fn read_top_seq( + client: &Client, + table: &str, + stream_id: &str, + counter: i64, + ) -> StoreResult> { + let resp = client + .query() + .table_name(table) + .key_condition_expression("stream_id = :sid AND seq > :counter") + .expression_attribute_values(":sid", AttributeValue::S(stream_id.to_string())) + .expression_attribute_values(":counter", AttributeValue::N(counter.to_string())) + .projection_expression("#seq") + .expression_attribute_names("#seq", "seq") + .scan_index_forward(false) + .limit(1) + .consistent_read(true) + .send() + .await + .map_err(ddb_err)?; + resp.items() + .first() + .map(|item| parse_n(item, "seq")) + .transpose() + } + + /// The counter's half of an append transaction: move `next_seq` to `slot.seq`, on the + /// condition that it still stands at `slot.expected` (or, for `0`, that the stream has no + /// counter yet). This condition orders the append transactions of a stream; the predecessor + /// check ([`Self::predecessor_check`]) orders what a reader sees. On a failed check DynamoDB + /// hands back the counter as it found it, so the next round needs no read. + fn counter_update(table: &str, stream_id: &str, slot: Slot) -> StoreResult { + let update = Update::builder() + .table_name(table) + .key("stream_id", AttributeValue::S(stream_id.to_string())) + .key("seq", AttributeValue::N(COUNTER_SEQ.to_string())) + .update_expression("SET next_seq = :seq") + .expression_attribute_values(":seq", AttributeValue::N(slot.seq.to_string())) + .expression_attribute_values(":expected", AttributeValue::N(slot.expected.to_string())) + .return_values_on_condition_check_failure(ReturnValuesOnConditionCheckFailure::AllOld); + let update = if slot.expected == 0 { + update.condition_expression("attribute_not_exists(next_seq) OR next_seq = :expected") + } else { + update.condition_expression("next_seq = :expected") + }; + update.build().map_err(|e| StoreError(e.to_string())) + } + + /// The predecessor check of an append transaction (module doc, "Sequence allocation"): a + /// `ConditionCheck` on the op at `below` — `attribute_exists` normally, `attribute_not_exists` + /// for the flip over a legacy gap. Either way the item is part of the transaction, so a + /// transaction still writing it cancels this one with `TransactionConflict`. Needs the + /// `dynamodb:ConditionCheckItem` IAM action. + fn predecessor_check( + table: &str, + stream_id: &str, + below: i64, + check: Predecessor, + ) -> StoreResult { + let condition = match check { + Predecessor::Absent => "attribute_not_exists(stream_id)", + Predecessor::Exists | Predecessor::None => "attribute_exists(stream_id)", + }; + ConditionCheck::builder() + .table_name(table) + .key("stream_id", AttributeValue::S(stream_id.to_string())) + .key("seq", AttributeValue::N(below.to_string())) + .condition_expression(condition) + .build() + .map_err(|e| StoreError(e.to_string())) } /// The op's item at `seq`: one attribute per [`WireOp`] field (see the module doc). @@ -559,78 +1106,65 @@ impl DynamoDbOpStore { item } - /// Put the op at `seq` WITHOUT a claim, conditioned on nothing already occupying this - /// exact `(stream_id, seq)`. On a composite key, `attribute_not_exists(stream_id)` is - /// evaluated against the item addressed by the request's own key — so it reads as "no - /// item at this stream_id+seq pair", not "stream_id is absent from the table" (which - /// would be true of every other seq in the same stream). A collision here means - /// `allocate_seq` must be called again for a fresh value — see `append`'s retry loop - /// for why both this condition and the atomic counter exist. + /// One append round: ONE `TransactWriteItems` laid out as [`TxLayout`] says — the counter + /// moved to `slot.seq` ([`Self::counter_update`]), the op's claim at [`claim_seq`] when + /// `claimed`, the predecessor check per `check` ([`Self::predecessor_check`]), and the op at + /// `slot.seq`. The claim and the op are each conditioned on their key being free — + /// `attribute_not_exists(stream_id)`, which on a composite key reads "no item at this exact + /// key", not "the stream is absent". So a seq is taken by exactly the transaction that also + /// writes its op, only after the transaction that wrote the seq below, and not before that op + /// is written (`6j6v.r3jf`). /// - /// Only reached when the op's claim key belongs to a different op id (see the module - /// doc): such an op is stored as every op was before the claims existed. - async fn try_put_op( - client: &Client, - table: &str, - stream_id: &str, - seq: i64, - op: WireOp, - ) -> StoreResult { - let result = client - .put_item() - .table_name(table) - .set_item(Some(Self::op_item(stream_id, seq, op))) - .condition_expression("attribute_not_exists(stream_id)") - .send() - .await; - - match result { - Ok(_) => Ok(PutOutcome::Written), - Err(err) => classify_put_error(err.into_service_error()), - } - } - - /// Put the op at `seq` together with its claim, in one transaction: index 0 the claim at - /// [`claim_seq`], index 1 the op, each conditioned on its own key being free (the same - /// `attribute_not_exists(stream_id)` reading as [`Self::try_put_op`]). A taken claim means - /// the op is already stored, so this answers with the seq the claim recorded — unless the - /// claim was written for a different op id, in which case the op goes in unclaimed. + /// A taken claim means the op is already stored, so this answers with the seq the claim + /// recorded — unless the claim was written for a different op id, in which case the next + /// round stores the op without one. async fn try_append_op( client: &Client, table: &str, stream_id: &str, - seq: i64, - op: WireOp, + slot: Slot, + check: Predecessor, + op: &WireOp, + claimed: bool, ) -> StoreResult { let claim = claim_seq(&op.op_id); - let op_id = op.op_id.clone(); - // `build()` can only fail on a required field left unset, and every one is set here; - // mapped rather than unwrapped for the reason `try_claim` gives. - let claim_put = Put::builder() - .table_name(table) - .item("stream_id", AttributeValue::S(stream_id.to_string())) - .item("seq", AttributeValue::N(claim.to_string())) - .item("op_id", AttributeValue::S(op_id.clone())) - .item("op_seq", AttributeValue::N(seq.to_string())) - .condition_expression("attribute_not_exists(stream_id)") - .build() - .map_err(|e| StoreError(e.to_string()))?; + let layout = TxLayout { + claimed, + checked: check != Predecessor::None, + }; + let mut tx = client.transact_write_items().transact_items( + TransactWriteItem::builder() + .update(Self::counter_update(table, stream_id, slot)?) + .build(), + ); + if claimed { + // `build()` can only fail on a required field left unset, and every one is set here; + // mapped rather than unwrapped for the reason `try_claim` gives. + let claim_put = Put::builder() + .table_name(table) + .item("stream_id", AttributeValue::S(stream_id.to_string())) + .item("seq", AttributeValue::N(claim.to_string())) + .item("op_id", AttributeValue::S(op.op_id.clone())) + .item("op_seq", AttributeValue::N(slot.seq.to_string())) + .condition_expression("attribute_not_exists(stream_id)") + .build() + .map_err(|e| StoreError(e.to_string()))?; + tx = tx.transact_items(TransactWriteItem::builder().put(claim_put).build()); + } + if let (true, Some(below)) = (layout.checked, slot.predecessor()) { + let check = Self::predecessor_check(table, stream_id, below, check)?; + tx = tx.transact_items(TransactWriteItem::builder().condition_check(check).build()); + } let op_put = Put::builder() .table_name(table) - .set_item(Some(Self::op_item(stream_id, seq, op.clone()))) + .set_item(Some(Self::op_item(stream_id, slot.seq, op.clone()))) .condition_expression("attribute_not_exists(stream_id)") .build() .map_err(|e| StoreError(e.to_string()))?; + tx = tx.transact_items(TransactWriteItem::builder().put(op_put).build()); - let result = client - .transact_write_items() - .transact_items(TransactWriteItem::builder().put(claim_put).build()) - .transact_items(TransactWriteItem::builder().put(op_put).build()) - .send() - .await; - - let reasons = match result { - Ok(_) => return Ok(PutOutcome::Written), + let reasons = match tx.send().await { + Ok(_) => return Ok(PutOutcome::Written(slot.seq)), Err(err) => match err.into_service_error() { TransactWriteItemsError::TransactionCanceledException(e) => { e.cancellation_reasons().to_vec() @@ -638,39 +1172,9 @@ impl DynamoDbOpStore { other => return Err(ddb_err(other)), }, }; - let claim_code = reasons.first().and_then(|r| r.code()); - let op_code = reasons.get(1).and_then(|r| r.code()); - match classify_append_cancellation(claim_code, op_code) { - Ok(AppendCancellation::Claimed) => { - match Self::read_claim(client, table, stream_id, claim).await? { - Some((claimed_for, stored)) if claimed_for == op_id => { - Ok(PutOutcome::Existing(stored)) - } - // The key is another op id's: store this op without a claim. - Some((claimed_for, _)) => { - eprintln!( - "nxf-relay: op {op_id:?} shares its claim key {claim} with op \ - {claimed_for:?} in stream {stream_id:?}; stored without a claim, \ - so a re-push of it is stored again" - ); - Self::try_put_op(client, table, stream_id, seq, op).await - } - // Reported taken, read back absent: nothing deletes a claim today, so this is - // a race with something that does. Try again rather than fail the push. - None => Ok(PutOutcome::Collided), - } - } - Ok(AppendCancellation::SeqTaken) => Ok(PutOutcome::Collided), - // The transaction that won may have been this op's own re-push: if its claim now - // stands, that is the answer, and no further seq is burnt. - Ok(AppendCancellation::Retryable) => { - match Self::read_claim(client, table, stream_id, claim).await? { - Some((claimed_for, stored)) if claimed_for == op_id => { - Ok(PutOutcome::Existing(stored)) - } - _ => Ok(PutOutcome::Conflicted), - } - } + let codes: Vec> = reasons.iter().map(|r| r.code()).collect(); + let cancellation = match classify_append_cancellation(layout, &codes) { + Ok(cancellation) => cancellation, Err(msg) => { // The service's own sentences (an item past 400 KB, say) are carried along: // the codes alone do not say which validation failed. @@ -681,7 +1185,33 @@ impl DynamoDbOpStore { .join("; "); let message = format!("append: {msg}: {detail} (stream {stream_id:?})"); eprintln!("nxf-relay: dynamodb error: {message}"); - Err(StoreError(message)) + return Err(StoreError(message)); + } + }; + let hint = counter_hint(reasons.first().and_then(|r| r.item()))?; + match settle_cancellation(cancellation, slot, check, hint) { + Settle::Outcome(outcome) => Ok(outcome), + Settle::ReadClaim => { + let found = Self::read_claim(client, table, stream_id, claim).await?; + let outcome = settle_claim(found.clone(), &op.op_id); + if let (PutOutcome::ClaimForeign, Some((claimed_for, _))) = (&outcome, found) { + eprintln!( + "nxf-relay: op {:?} shares its claim key {claim} with op \ + {claimed_for:?} in stream {stream_id:?}; stored without a claim, \ + so a re-push of it is stored again", + op.op_id + ); + } + Ok(outcome) + } + Settle::FindTop => { + // The counter held (its condition passed), so it stands at `slot.expected`, and + // the next round needs no read of it. + let top = Self::read_top_seq(client, table, stream_id, slot.expected).await?; + Ok(PutOutcome::SeqTaken { + counter: slot.expected, + top: top.unwrap_or(slot.seq).max(slot.seq), + }) } } } @@ -710,6 +1240,44 @@ impl DynamoDbOpStore { None => Ok(None), } } + + /// Append one op: rounds of [`Self::try_append_op`] under [`append_with_retry`], starting + /// from `start` (what the caller already knows of the counter) and stopping at `deadline`. + /// Uncontended it runs once: one transaction, plus one read when `start` is unknown. + async fn append_one( + client: &Client, + table: &str, + stream_id: &str, + op: &WireOp, + start: Counter, + deadline: Option, + before_write: Option<&BeforeWrite>, + ) -> StoreResult { + append_with_retry(stream_id, start, deadline, move |round| async move { + let counter = match round.counter { + Counter::Unknown => Self::read_counter(client, table, stream_id).await?, + Counter::Hint(c) | Counter::Known(c) | Counter::Carried(c) => c, + }; + let slot = Slot::after(counter, round.floor)?; + let check = round.predecessor_check(slot); + if let Some(hook) = before_write { + let attempt = WriteAttempt { + op_id: op.op_id.clone(), + seq: slot.seq, + expected: slot.expected, + predecessor: match check { + Predecessor::None => None, + Predecessor::Exists => Some(true), + Predecessor::Absent => Some(false), + }, + counter_from: round.counter.source(), + }; + tokio::task::block_in_place(|| hook(&attempt)); + } + Self::try_append_op(client, table, stream_id, slot, check, op, round.claimed).await + }) + .await + } } impl OpStore for DynamoDbOpStore { @@ -718,26 +1286,66 @@ 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 - // the overwhelmingly common case this cycle runs exactly once. The conditional - // put in `try_append_op` is what makes a seq collision — however it arose — a - // retried allocation instead of a silently overwritten op, and its claim is what - // makes a re-pushed op answer with the seq it already has. - append_with_retry(&stream_id, || { - // Re-borrowing inside the closure (rather than capturing it) keeps the - // returned future borrowing the enclosing scope, not the closure itself — - // 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(); - async move { - let seq = Self::allocate_seq(client, table, stream_id).await?; - let outcome = Self::try_append_op(client, table, stream_id, seq, op).await?; - Ok((seq, outcome)) + let settled = Self::append_one( + &client, + &table, + &stream_id, + &op, + Counter::Unknown, + None, + before_write.as_ref(), + ) + .await?; + Ok(Cursor(settled.seq())) + }) + } + + /// The push path: [`OpStore::append_batch`]'s contract, with the deadline also checked before + /// every round and raced against the round in flight, and the seq each written op took + /// carried to the next op as the counter to expect — so an uncontended push of `n` ops costs + /// one read and `n` transactions. A wrong guess (another writer appended in between) costs one + /// lost round, which hands back the real counter. + fn append_batch( + &self, + stream: &StreamId, + ops: &[WireOp], + deadline: Instant, + ) -> StoreResult<()> { + let client = self.client.clone(); + let table = self.table.clone(); + let stream = stream.clone(); + let ops = ops.to_vec(); + let before_write = self.before_write.clone(); + block_on(async move { + let stream_id = stream.as_str(); + let mut start = Counter::Unknown; + for (done, op) in ops.iter().enumerate() { + if Instant::now() >= deadline { + return Err(past_deadline(&stream, done, ops.len())); } - }) - .await + let settled = Self::append_one( + &client, + &table, + stream_id, + op, + start, + Some(deadline), + before_write.as_ref(), + ) + .await; + match settled { + Ok(Settled::Written(seq)) => start = Counter::Carried(seq), + // An op already stored says nothing new about the counter. + Ok(Settled::Existing(_)) => {} + Err(_) if Instant::now() >= deadline => { + return Err(past_deadline(&stream, done, ops.len())); + } + Err(e) => return Err(e), + } + } + Ok(()) }) } @@ -883,7 +1491,7 @@ fn is_retryable_code(code: Option<&str>) -> bool { /// self-race must never be treated as an ordinary collision. Only once neither condition /// failed does a transient code ([`is_retryable_code`]) make this a [`ClaimOutcome::Retryable`]. /// Everything left over — a `ValidationError`, or no recognised code at all — is a hard -/// error, matching `try_put_op`'s stricter pattern above: `register` must surface a real +/// error, matching `classify_append_cancellation`'s stricter pattern above: `register` must surface a real /// fault rather than misread it as "try another candidate" or spin on it. fn classify_cancellation( prefix_item_code: Option<&str>, @@ -999,7 +1607,7 @@ impl DynamoDbPrefixRegistry { /// would make `owned_prefix` miss it on the replica's very next (idempotent) register. /// Each `Put` is conditioned on `attribute_not_exists(stream_id)`, evaluated against the /// item at THAT put's own `(stream_id, sk)` key — "no item here yet", not "stream_id is - /// absent from the table" (mirrors `try_put_op` in the op store above). + /// absent from the table" (mirrors `try_append_op` in the op store above). /// /// A cancelled transaction MIGHT mean someone else won a race on one of the two keys — or /// it might be transient contention (a conflict, a throttle), or a genuine fault (a @@ -1377,74 +1985,215 @@ mod tests { .block_on(fut) } - // ----- classify_put_error + append_with_retry --------------------------------------- + // ----- Slot + append_with_retry ------------------------------------------------------ // - // Both are unreachable against a live DynamoDB: `allocate_seq`'s atomic counter never - // hands two callers the same seq, so no acceptance test can produce a conditional-check - // collision on the op table. These drive the branches directly. + // DynamoDB Local serialises transactions, so a throttle, a `TransactionConflict` or a + // contention budget running out never happen against it. These drive the policy directly. #[test] - fn a_conditional_check_failure_on_the_op_put_is_a_collision_not_an_error() { - let err = PutItemError::ConditionalCheckFailedException( - aws_sdk_dynamodb::types::error::ConditionalCheckFailedException::builder().build(), + fn a_slot_takes_the_seq_after_the_counter_and_steps_over_an_occupied_one() { + let slot = |counter, floor| Slot::after(counter, floor).unwrap(); + assert_eq!( + slot(0, 0), + Slot { + expected: 0, + seq: 1 + }, + "a stream without a counter starts at 1" ); assert_eq!( - classify_put_error(err).expect("a condition failure is an outcome, not an error"), - PutOutcome::Collided + slot(9, 0), + Slot { + expected: 9, + seq: 10 + } + ); + assert_eq!( + slot(0, 1), + Slot { + expected: 0, + seq: 2 + }, + "an op stored before the counter existed is stepped over, the counter still expected" + ); + assert_eq!( + slot(12, 3), + Slot { + expected: 12, + seq: 13 + }, + "a floor below the counter changes nothing" + ); + assert_eq!(slot(0, 0).predecessor(), None, "seq 1 has nothing below"); + assert_eq!(slot(9, 0).predecessor(), Some(9)); + assert_eq!( + slot(0, 40).predecessor(), + Some(40), + "a floor step checks the floor" ); } #[test] - fn any_other_put_failure_is_a_hard_error_not_a_collision() { - // A missing table must never be mistaken for "that seq is taken" — that would spin - // the append loop 32 times and then report the wrong cause. - let err = PutItemError::ResourceNotFoundException( - aws_sdk_dynamodb::types::error::ResourceNotFoundException::builder() - .message("Requested resource not found") - .build(), + fn a_seq_past_i64_max_is_an_error_not_a_wrap_into_the_claim_range() { + let err = Slot::after(i64::MAX, 0).expect_err("no seq after i64::MAX"); + assert!(err.to_string().contains("i64::MAX"), "{err}"); + assert!(Slot::after(0, i64::MAX).is_err(), "nor after a floor there"); + assert_eq!(Slot::after(i64::MAX - 1, 0).unwrap().seq, i64::MAX); + } + + #[test] + fn the_predecessor_check_flips_only_for_the_seq_found_over_a_gap() { + let slot = Slot::after(9, 0).unwrap(); + let mut round = Round::first(Counter::Unknown); + assert_eq!(round.predecessor_check(slot), Predecessor::Exists); + round.gap = Some(10); + assert_eq!(round.predecessor_check(slot), Predecessor::Absent); + round.gap = Some(11); + assert_eq!( + round.predecessor_check(slot), + Predecessor::Exists, + "a gap found at another seq does not relax this one" ); - let outcome = classify_put_error(err); - assert!(outcome.is_err(), "expected a hard error, got {outcome:?}"); + assert_eq!( + round.predecessor_check(Slot::after(0, 0).unwrap()), + Predecessor::None + ); + } + + /// The rounds a stand-in was asked for, shared with the test. + type Rounds = std::rc::Rc>>; + + /// What a stand-in round answers. + type Answer = std::future::Ready>; + + /// A stand-in round that records each round it is asked for and answers from `script`. + fn scripted(script: Vec) -> (Rounds, impl FnMut(Round) -> Answer) { + let rounds: Rounds = std::rc::Rc::default(); + let seen = std::rc::Rc::clone(&rounds); + let mut script = script.into_iter(); + let round = move |r: Round| { + seen.borrow_mut().push(r); + std::future::ready(Ok(script.next().expect("the script covers every round"))) + }; + (rounds, round) + } + + /// [`append_with_retry`] from an unknown counter and without a deadline. + fn retry(round: F) -> StoreResult + where + F: FnMut(Round) -> Fut, + Fut: std::future::Future>, + { + drive(append_with_retry("s", Counter::Unknown, None, round)) } #[test] - fn append_retries_a_collided_put_with_a_freshly_allocated_seq() { - // The invariant the retry exists for: a collision must NOT re-put the same seq (that - // would livelock), it must take the next allocation. The stand-in hands out 1, 2, 3 - // and only lets the third land. - let calls = std::cell::Cell::new(0i64); - let cursor = drive(append_with_retry("s", || { - let seq = calls.get() + 1; - calls.set(seq); - async move { - let outcome = if seq < 3 { - PutOutcome::Collided - } else { - PutOutcome::Written - }; - Ok((seq, outcome)) - } - })) - .expect("the third attempt lands"); - assert_eq!(cursor, Cursor(3), "the cursor is the seq that was written"); - assert_eq!(calls.get(), 3, "exactly three allocate-then-put cycles ran"); + fn a_round_lost_to_a_concurrent_append_retries_from_the_counter_it_saw() { + // The race this backend now settles inside the transaction (6j6v.r3jf): the loser does not + // keep a seq, it learns where the counter went and takes the seq after that. + let (rounds, round) = scripted(vec![ + PutOutcome::Lost(Some(10)), + PutOutcome::Lost(None), + PutOutcome::Written(12), + ]); + assert_eq!(retry(round).unwrap(), Settled::Written(12)); + let counters: Vec = rounds.borrow().iter().map(|r| r.counter).collect(); + assert_eq!( + counters, + [Counter::Unknown, Counter::Hint(10), Counter::Unknown], + "read first; then the counter the failed check saw; read again when none came back" + ); } #[test] - fn append_gives_up_loudly_after_append_attempts_collisions() { - // Never an unbounded spin: a cycle that always collides is structurally broken, and - // the caller must be told so rather than have its thread held forever. - let calls = std::cell::Cell::new(0u32); - let err = drive(append_with_retry("s", || { - calls.set(calls.get() + 1); - async { Ok((1, PutOutcome::Collided)) } - })) - .expect_err("an always-colliding cycle must not loop forever"); + fn a_push_carries_the_counter_it_was_given_into_the_first_round() { + let (rounds, round) = scripted(vec![PutOutcome::Written(8)]); + drive(append_with_retry("s", Counter::Carried(7), None, round)).unwrap(); + assert_eq!(rounds.borrow()[0].counter, Counter::Carried(7), "no read"); + } + + #[test] + fn a_taken_seq_raises_the_floor_to_the_top_and_keeps_the_counter_known() { + let (rounds, round) = scripted(vec![ + PutOutcome::SeqTaken { + counter: 0, + top: 40, + }, + PutOutcome::ClaimForeign, + PutOutcome::Written(41), + ]); + assert_eq!(retry(round).unwrap(), Settled::Written(41)); + let rounds = rounds.borrow(); assert_eq!( - calls.get(), - APPEND_ATTEMPTS, - "the bound is APPEND_ATTEMPTS, not more" + rounds[1], + Round { + counter: Counter::Known(0), + floor: 40, + claimed: true, + gap: None, + }, + "one step to the top occupied seq, and no re-read of a counter whose condition held" ); + assert_eq!( + rounds[2], + Round { + counter: Counter::Unknown, + floor: 40, + claimed: false, + gap: None, + }, + "an op whose claim key is another op id's goes in unclaimed, as before the claims" + ); + } + + #[test] + fn a_missing_predecessor_flips_the_check_for_that_seq_and_only_for_the_next_round() { + let (rounds, round) = scripted(vec![ + PutOutcome::Predecessor { + counter: 9, + seq: 10, + absent: true, + }, + PutOutcome::Lost(Some(10)), + PutOutcome::Predecessor { + counter: 10, + seq: 11, + absent: false, + }, + PutOutcome::Written(11), + ]); + assert_eq!(retry(round).unwrap(), Settled::Written(11)); + let rounds = rounds.borrow(); + assert_eq!( + (rounds[1].counter, rounds[1].gap), + (Counter::Known(9), Some(10)), + "the same slot again, its predecessor checked for absence" + ); + assert_eq!( + rounds[2].gap, None, + "a lost round moves to another slot, and the check is attribute_exists again" + ); + assert_eq!( + (rounds[3].counter, rounds[3].gap), + (Counter::Known(10), None), + "a predecessor that appeared is checked for again, not stepped over" + ); + } + + #[test] + fn append_gives_up_loudly_after_append_attempts_stalled_rounds() { + // Never an unbounded spin: rounds that keep failing with no other writer committing are + // structurally broken or saturated, and the caller must be told so. + let (rounds, round) = scripted( + (1..=i64::from(APPEND_ATTEMPTS)) + .map(|seq| PutOutcome::SeqTaken { + counter: 0, + top: seq, + }) + .collect(), + ); + let err = retry(round).expect_err("a stall must not loop forever"); + assert_eq!(rounds.borrow().len() as u32, APPEND_ATTEMPTS); assert!( err.to_string().contains(&APPEND_ATTEMPTS.to_string()), "the error names the bound it hit: {err}" @@ -1452,55 +2201,148 @@ mod tests { } #[test] - fn append_propagates_a_hard_error_from_the_cycle_immediately() { - // A real fault must not be retried 32 times: it is not a collision. + fn contention_has_its_own_larger_budget_and_a_clear_error_when_it_runs_out() { + // Lost rounds are other writers' commits, so they do not eat the stall budget: more of + // them than APPEND_ATTEMPTS still land... + let mut script: Vec = (0..APPEND_ATTEMPTS * 2) + .map(|_| PutOutcome::Lost(None)) + .collect(); + script.push(PutOutcome::Written(99)); + let (_, round) = scripted(script); + assert_eq!(retry(round).unwrap(), Settled::Written(99)); + + // ...but they are bounded too, and the error says what happened. + let (rounds, round) = scripted( + (0..CONTENTION_ROUNDS) + .map(|_| PutOutcome::Lost(None)) + .collect(), + ); + let err = retry(round).expect_err("contention is bounded"); + assert_eq!(rounds.borrow().len() as u32, CONTENTION_ROUNDS); + let err = err.to_string(); + assert!( + err.contains("concurrent appends") && err.contains(&CONTENTION_ROUNDS.to_string()), + "{err}" + ); + } + + #[test] + fn the_contention_budget_counts_lost_rounds_in_a_row() { + // CONTENTION_ROUNDS - 1 losses, one round that ends otherwise, then as many losses again: + // twice the budget in total, never the budget in a row, so the append still lands. + let losses = || (1..CONTENTION_ROUNDS).map(|_| PutOutcome::Lost(None)); + let script: Vec = losses() + .chain([PutOutcome::Predecessor { + counter: 3, + seq: 4, + absent: true, + }]) + .chain(losses()) + .chain([PutOutcome::Written(500)]) + .collect(); + let (_, round) = scripted(script); + assert_eq!( + retry(round).expect("a round that is not lost resets the count"), + Settled::Written(500) + ); + } + + #[test] + fn append_propagates_a_hard_error_from_the_round_immediately() { let calls = std::cell::Cell::new(0u32); - let err = drive(append_with_retry("s", || { + let err = retry(|_| { calls.set(calls.get() + 1); async { Err(StoreError("table gone".into())) } - })) + }) .expect_err("a hard error surfaces"); - assert_eq!(calls.get(), 1, "no retry on a non-collision failure"); + assert_eq!(calls.get(), 1, "no retry on a fault"); assert!(err.to_string().contains("table gone")); } #[test] - fn a_conflicted_round_backs_off_and_then_allocates_afresh() { - // A lost transaction is retried after a pause, not in lockstep — and still with a fresh - // seq: the stand-in hands out 1, 2, 3 and lands the third. - let calls = std::cell::Cell::new(0i64); - let started = std::time::Instant::now(); - let cursor = drive(append_with_retry("s", || { - let seq = calls.get() + 1; - calls.set(seq); - async move { - let outcome = if seq < 3 { - PutOutcome::Conflicted - } else { - PutOutcome::Written - }; - Ok((seq, outcome)) - } - })) - .expect("the third attempt lands"); - assert_eq!(cursor, Cursor(3)); + fn a_throttled_round_backs_off_before_the_next() { + let (_, round) = scripted(vec![ + PutOutcome::Throttled, + PutOutcome::Throttled, + PutOutcome::Written(3), + ]); + let started = Instant::now(); + assert_eq!(retry(round).unwrap(), Settled::Written(3)); assert!( started.elapsed() >= claim_backoff(0) + claim_backoff(1), - "both conflicted rounds waited their backoff" + "both throttled rounds waited at least their backoff" ); } #[test] fn append_answers_an_already_stored_op_with_its_existing_seq() { - // The allocated seq (7) is burned; the answer is the seq the op already has (3). - let cursor = drive(append_with_retry("s", || async { - Ok((7, PutOutcome::Existing(3))) - })) - .expect("an already stored op is an answer, not an error"); - assert_eq!(cursor, Cursor(3)); + let (_, round) = scripted(vec![PutOutcome::Existing(3)]); + assert_eq!(retry(round).unwrap(), Settled::Existing(3)); + } + + #[test] + fn a_passed_deadline_stops_before_the_next_round() { + let (rounds, round) = scripted(vec![PutOutcome::Lost(None)]); + let deadline = Instant::now(); + let err = drive(append_with_retry( + "s", + Counter::Unknown, + Some(deadline), + round, + )) + .expect_err("nothing is sent once the deadline has passed"); + assert!(rounds.borrow().is_empty()); + assert!(err.to_string().contains("deadline"), "{err}"); } - // ----- claim_seq + classify_append_cancellation -------------------------------------- + #[test] + fn the_deadline_cuts_a_round_in_flight_short() { + // A round that would take ten seconds (a stalled endpoint, say) is not waited out. + let started = Instant::now(); + let deadline = started + Duration::from_millis(50); + let err = drive(append_with_retry( + "s", + Counter::Unknown, + Some(deadline), + |_| async { + tokio::time::sleep(Duration::from_secs(10)).await; + Ok(PutOutcome::Written(1)) + }, + )) + .expect_err("the round is abandoned at the deadline"); + assert!(err.to_string().contains("deadline"), "{err}"); + assert!( + started.elapsed() < Duration::from_secs(2), + "stopped at the deadline, not after the round: {:?}", + started.elapsed() + ); + } + + #[test] + fn the_jitter_stays_inside_its_cap() { + for lost in 0..40 { + assert!(contention_jitter(lost) <= Duration::from_millis(16)); + } + assert!(jitter(Duration::ZERO) <= Duration::from_micros(1)); + assert!(jitter(Duration::from_millis(5)) <= Duration::from_millis(5)); + } + + #[test] + fn the_jitter_actually_spreads_its_draws() { + // A jitter that is always zero (or always one value) would pass the cap test above and + // bring every loser back for the same seq at the same instant. 200 draws over 16 ms: the + // chance that a uniform draw lands below 4 ms every time is 0.25^200. + let cap = Duration::from_millis(16); + let draws: Vec = (0..200).map(|_| jitter(cap)).collect(); + let distinct: std::collections::BTreeSet = draws.iter().copied().collect(); + assert!(distinct.len() > 20, "{} distinct draws", distinct.len()); + assert!( + draws.iter().any(|d| *d >= cap / 4), + "no draw reached a quarter of the cap" + ); + } + + // ----- claim_seq + the append transaction's cancellations ----------------------------- #[test] fn a_claim_key_lies_below_the_counter_and_is_stable_per_op_id() { @@ -1522,37 +2364,231 @@ mod tests { assert_eq!(claim_seq("o1"), -1_272_668_462_371_106_663); } + const ALL: TxLayout = TxLayout { + claimed: true, + checked: true, + }; + #[test] - fn a_taken_claim_outranks_a_taken_seq_and_a_conflict() { + fn the_layout_puts_each_item_at_its_index() { + assert_eq!( + (ALL.claim(), ALL.predecessor(), ALL.op()), + (Some(1), Some(2), 3) + ); + let unclaimed = TxLayout { + claimed: false, + checked: true, + }; + assert_eq!( + (unclaimed.claim(), unclaimed.predecessor(), unclaimed.op()), + (None, Some(1), 2) + ); + let first_seq = TxLayout { + claimed: true, + checked: false, + }; + assert_eq!( + (first_seq.claim(), first_seq.predecessor(), first_seq.op()), + (Some(1), None, 2) + ); + let bare = TxLayout { + claimed: false, + checked: false, + }; + assert_eq!( + (bare.claim(), bare.predecessor(), bare.op()), + (None, None, 1) + ); + } + + #[test] + fn a_taken_claim_outranks_a_moved_counter_which_outranks_a_taken_seq_and_the_predecessor() { let ccf = Some("ConditionalCheckFailed"); + let none = Some("None"); + let conflict = Some("TransactionConflict"); + let classify = classify_append_cancellation; assert_eq!( - classify_append_cancellation(ccf, ccf), - Ok(AppendCancellation::Claimed) + classify(ALL, &[ccf, ccf, ccf, ccf]), + Ok(AppendCancellation::Claimed), + "a re-push that lost the counter race still answers with its stored seq" ); assert_eq!( - classify_append_cancellation(ccf, Some("TransactionConflict")), + classify(ALL, &[conflict, ccf, none, none]), Ok(AppendCancellation::Claimed) ); assert_eq!( - classify_append_cancellation(Some("None"), ccf), + classify(ALL, &[ccf, none, ccf, ccf]), + Ok(AppendCancellation::CounterMoved) + ); + assert_eq!( + classify(ALL, &[none, none, ccf, ccf]), + Ok(AppendCancellation::SeqTaken), + "a taken seq outranks a failed predecessor check" + ); + assert_eq!( + classify(ALL, &[none, none, ccf, none]), + Ok(AppendCancellation::PredecessorFailed) + ); + assert_eq!( + classify(ALL, &[none, none, conflict, none]), + Ok(AppendCancellation::Conflict), + "the append below still writing its op: contention, retried" + ); + assert_eq!( + classify(ALL, &[none, Some("ThrottlingError"), none, conflict]), + Ok(AppendCancellation::Conflict), + "a conflict means another append is committing: contention, not a stall" + ); + assert_eq!( + classify( + ALL, + &[none, none, none, Some("ProvisionedThroughputExceeded")] + ), + Ok(AppendCancellation::Throttled) + ); + } + + #[test] + fn the_codes_are_read_at_the_layouts_indexes() { + let ccf = Some("ConditionalCheckFailed"); + let none = Some("None"); + // Unclaimed: index 1 is the predecessor check, not a claim. + let unclaimed = TxLayout { + claimed: false, + checked: true, + }; + assert_eq!( + classify_append_cancellation(unclaimed, &[none, ccf, none]), + Ok(AppendCancellation::PredecessorFailed) + ); + assert_eq!( + classify_append_cancellation(unclaimed, &[none, none, ccf]), Ok(AppendCancellation::SeqTaken) ); + // Seq 1, claimed: index 1 is the claim, index 2 the op. + let first_seq = TxLayout { + claimed: true, + checked: false, + }; + assert_eq!( + classify_append_cancellation(first_seq, &[none, ccf, none]), + Ok(AppendCancellation::Claimed) + ); assert_eq!( - classify_append_cancellation(Some("TransactionConflict"), Some("None")), - Ok(AppendCancellation::Retryable) + classify_append_cancellation(first_seq, &[none, none, ccf]), + Ok(AppendCancellation::SeqTaken) ); + let bare = TxLayout { + claimed: false, + checked: false, + }; assert_eq!( - classify_append_cancellation(None, Some("ThrottlingError")), - Ok(AppendCancellation::Retryable) + classify_append_cancellation(bare, &[ccf, None]), + Ok(AppendCancellation::CounterMoved), + "an unclaimed round" ); } #[test] fn any_other_append_cancellation_is_a_hard_error() { - let err = classify_append_cancellation(Some("None"), Some("ValidationError")) + let none = Some("None"); + let err = classify_append_cancellation(ALL, &[none, none, none, Some("ValidationError")]) .expect_err("a validation failure is a fault, never a retry"); assert!(err.contains("ValidationError"), "{err}"); - assert!(classify_append_cancellation(None, None).is_err()); + assert!(classify_append_cancellation(ALL, &[]).is_err()); + } + + #[test] + fn each_cancellation_maps_to_its_next_step() { + let slot = Slot::after(9, 0).unwrap(); + let settle = |c, check, hint| settle_cancellation(c, slot, check, hint); + let exists = Predecessor::Exists; + assert_eq!( + settle(AppendCancellation::CounterMoved, exists, Some(12)), + Settle::Outcome(PutOutcome::Lost(Some(12))), + "a moved counter carries the counter the failed check handed back" + ); + assert_eq!( + settle(AppendCancellation::Conflict, exists, Some(12)), + Settle::Outcome(PutOutcome::Lost(None)), + "a conflict says nothing about the counter: read it" + ); + assert_eq!( + settle(AppendCancellation::PredecessorFailed, exists, None), + Settle::Outcome(PutOutcome::Predecessor { + counter: 9, + seq: 10, + absent: true + }), + "attribute_exists failed: the op below is absent" + ); + assert_eq!( + settle( + AppendCancellation::PredecessorFailed, + Predecessor::Absent, + None + ), + Settle::Outcome(PutOutcome::Predecessor { + counter: 9, + seq: 10, + absent: false + }), + "attribute_not_exists failed: the op below appeared" + ); + assert_eq!( + settle(AppendCancellation::Throttled, exists, None), + Settle::Outcome(PutOutcome::Throttled) + ); + assert_eq!( + settle(AppendCancellation::Claimed, exists, None), + Settle::ReadClaim + ); + assert_eq!( + settle(AppendCancellation::SeqTaken, exists, None), + Settle::FindTop + ); + } + + #[test] + fn a_claim_read_back_settles_as_existing_foreign_or_vanished() { + assert_eq!( + settle_claim(Some(("o1".into(), 7)), "o1"), + PutOutcome::Existing(7) + ); + assert_eq!( + settle_claim(Some(("other".into(), 7)), "o1"), + PutOutcome::ClaimForeign + ); + assert_eq!(settle_claim(None, "o1"), PutOutcome::ClaimVanished); + } + + #[test] + fn the_counter_reads_as_zero_when_absent_and_the_hint_only_when_handed_back() { + let with = |n: &str| { + HashMap::from([ + ("stream_id".to_string(), AttributeValue::S("s".into())), + ("next_seq".to_string(), AttributeValue::N(n.into())), + ]) + }; + let without = HashMap::from([("stream_id".to_string(), AttributeValue::S("s".into()))]); + assert_eq!(counter_value(Some(&with("12"))).unwrap(), 12); + assert_eq!(counter_value(Some(&without)).unwrap(), 0); + assert_eq!(counter_value(None).unwrap(), 0); + assert_eq!(counter_hint(Some(&with("12"))).unwrap(), Some(12)); + assert_eq!( + counter_hint(Some(&without)).unwrap(), + Some(0), + "an item without the attribute is a stream without a counter" + ); + assert_eq!( + counter_hint(None).unwrap(), + None, + "nothing handed back: the next round reads" + ); + assert!( + counter_hint(Some(&with("twelve"))).is_err(), + "a malformed counter is loud, never a silent 0" + ); } // ----- retry_claim ------------------------------------------------------------------- diff --git a/crates/server/tests/dynamodb_local.rs b/crates/server/tests/dynamodb_local.rs index 85825b4..310cf43 100644 --- a/crates/server/tests/dynamodb_local.rs +++ b/crates/server/tests/dynamodb_local.rs @@ -35,7 +35,7 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Once}; use std::thread; -use std::time::Duration; +use std::time::{Duration, Instant}; use aws_sdk_dynamodb::types::{ AttributeDefinition, AttributeValue, BillingMode, KeySchemaElement, KeyType, @@ -47,7 +47,7 @@ use nxs_server::app::app; use nxs_server::presence::PresenceStore; use nxs_server::registry::{PrefixRegistry, SqlitePrefixRegistry}; use nxs_server::store::OpStore; -use nxs_server::store_ddb::{DynamoDbOpStore, DynamoDbPrefixRegistry}; +use nxs_server::store_ddb::{DynamoDbOpStore, DynamoDbPrefixRegistry, WriteAttempt}; use nxs_sync::protocol::{ Cursor, MachineHello, MachinesResponse, PullResponse, PushResponse, RegisterOutcome, StreamId, }; @@ -295,50 +295,216 @@ fn append_read_since_round_trips_a_batch_in_order() { #[test] fn dynamodb_appends_idempotently_by_op_id() { - // 6j6v.tm4k: the contract SQLite and Postgres keep (spec §4.4), run from the same script. The - // one documented difference: a re-push burns the seq it allocated, so the stream is not dense. + // 6j6v.tm4k: the contract SQLite and Postgres keep (spec §4.4), run from the same script. + // Since 6j6v.r3jf the seq is taken inside the transaction the claim turns back, so a re-push + // consumes no seq here either: dense, like the other two. let Some(ep) = endpoint("dynamodb_appends_idempotently_by_op_id") else { return; }; - idempotence::assert_idempotent_append(&opstore(&ep), wire, false); + idempotence::assert_idempotent_append(&opstore(&ep), wire, true); } /// Put an item straight into the op table, as the store writes an op — without its claim. What a /// stream written before the claims existed holds. fn put_unclaimed_op(ep: &str, table: &str, seq: i64, op_id: &str) { - let client = admin_client(ep); - admin_rt().block_on(async { - client - .put_item() - .table_name(table) - .item("stream_id", AttributeValue::S("s".into())) - .item("seq", AttributeValue::N(seq.to_string())) - .item( - "envelope_version", - AttributeValue::N(ENVELOPE_VERSION.to_string()), - ) - .item("op_id", AttributeValue::S(op_id.into())) - .item("lamport", AttributeValue::N("1".into())) - .item("site", AttributeValue::N("1".into())) - .item("domain", AttributeValue::S("task".into())) - .item("target_kind", AttributeValue::S("item".into())) - .item("target_id", AttributeValue::S("ab12.0001".into())) - .item("field", AttributeValue::S("title".into())) - .item("op_type", AttributeValue::S("set".into())) - .item("value", AttributeValue::S("t".into())) - .item("author", AttributeValue::S("a".into())) - .item("wall_clock", AttributeValue::S(String::new())) - .send() - .await - .expect("plant an unclaimed op"); + assert!( + put_op(ep, table, seq, op_id, false), + "plant an unclaimed op" + ); +} + +/// Put an op item at `seq` of stream "s", without a claim; with `if_free`, conditioned on the key +/// being free, as every build's op put is — how an older relay writes the op it allocated with +/// `ADD`. Whether the put landed. Runs off any runtime, so a test seam may call it. +fn put_op(ep: &str, table: &str, seq: i64, op_id: &str, if_free: bool) -> bool { + let (ep, table, op_id) = (ep.to_string(), table.to_string(), op_id.to_string()); + off_runtime(move || { + let client = admin_client(&ep); + admin_rt().block_on(async { + let put = client + .put_item() + .table_name(&table) + .item("stream_id", AttributeValue::S("s".into())) + .item("seq", AttributeValue::N(seq.to_string())) + .item( + "envelope_version", + AttributeValue::N(ENVELOPE_VERSION.to_string()), + ) + .item("op_id", AttributeValue::S(op_id)) + .item("lamport", AttributeValue::N("1".into())) + .item("site", AttributeValue::N("1".into())) + .item("domain", AttributeValue::S("task".into())) + .item("target_kind", AttributeValue::S("item".into())) + .item("target_id", AttributeValue::S("ab12.0001".into())) + .item("field", AttributeValue::S("title".into())) + .item("op_type", AttributeValue::S("set".into())) + .item("value", AttributeValue::S("t".into())) + .item("author", AttributeValue::S("a".into())) + .item("wall_clock", AttributeValue::S(String::new())); + let put = if if_free { + put.condition_expression("attribute_not_exists(stream_id)") + } else { + put + }; + match put.send().await { + Ok(_) => true, + Err(e) if if_free => { + eprintln!("conditional put at {seq} refused: {e:?}"); + false + } + Err(e) => panic!("put an op at {seq}: {e:?}"), + } + }) + }) +} + +/// Run `f` on a plain thread of its own. The store's test seam runs on the store's bridge runtime, +/// where driving the harness's admin runtime directly would panic (a runtime inside a runtime). +fn off_runtime(f: impl FnOnce() -> T + Send) -> T { + thread::scope(|scope| scope.spawn(f).join().expect("the admin call panicked")) +} + +/// The stream "s"'s counter (`next_seq` of the item at seq 0), `None` when there is none. +fn counter_of(ep: &str, table: &str) -> Option { + let (ep, table) = (ep.to_string(), table.to_string()); + off_runtime(move || { + let client = admin_client(&ep); + admin_rt().block_on(async { + let resp = client + .query() + .table_name(&table) + .key_condition_expression("stream_id = :s AND seq = :zero") + .expression_attribute_values(":s", AttributeValue::S("s".into())) + .expression_attribute_values(":zero", AttributeValue::N("0".into())) + .consistent_read(true) + .send() + .await + .expect("read the counter"); + resp.items() + .first() + .and_then(|item| item.get("next_seq")) + .map(|n| n.as_n().unwrap().parse().unwrap()) + }) + }) +} + +/// Is an item stored at `seq` of stream "s"? +fn occupied(ep: &str, table: &str, seq: i64) -> bool { + let (ep, table) = (ep.to_string(), table.to_string()); + off_runtime(move || { + let client = admin_client(&ep); + admin_rt().block_on(async { + let resp = client + .query() + .table_name(&table) + .key_condition_expression("stream_id = :s AND seq = :seq") + .expression_attribute_values(":s", AttributeValue::S("s".into())) + .expression_attribute_values(":seq", AttributeValue::N(seq.to_string())) + .consistent_read(true) + .send() + .await + .expect("read an item"); + !resp.items().is_empty() + }) + }) +} + +/// What an older relay does to allocate a seq: `UpdateItem ADD next_seq :1` on the counter, with +/// no condition. Returns the seq it got. +fn old_relay_allocates(ep: &str, table: &str) -> i64 { + let (ep, table) = (ep.to_string(), table.to_string()); + off_runtime(move || { + let client = admin_client(&ep); + admin_rt().block_on(async { + let resp = client + .update_item() + .table_name(&table) + .key("stream_id", AttributeValue::S("s".into())) + .key("seq", AttributeValue::N("0".into())) + .update_expression("ADD next_seq :one") + .expression_attribute_values(":one", AttributeValue::N("1".into())) + .return_values(aws_sdk_dynamodb::types::ReturnValue::UpdatedNew) + .send() + .await + .expect("ADD on the counter"); + resp.attributes().unwrap()["next_seq"] + .as_n() + .unwrap() + .parse() + .unwrap() + }) + }) +} + +/// Set the stream "s"'s counter to `n`, as a transaction that took seq `n` shows it before its op, +/// or an earlier build's `ADD` left it over a seq whose op never came. +fn set_counter(ep: &str, table: &str, n: i64) { + let (ep, table) = (ep.to_string(), table.to_string()); + off_runtime(move || { + let client = admin_client(&ep); + admin_rt().block_on(async { + client + .put_item() + .table_name(&table) + .item("stream_id", AttributeValue::S("s".into())) + .item("seq", AttributeValue::N("0".into())) + .item("next_seq", AttributeValue::N(n.to_string())) + .send() + .await + .expect("set the counter"); + }) + }) +} + +/// An ops table of its own and a store on it, plus the table's name for the admin calls above. +fn opstore_and_table(ep: &str) -> (DynamoDbOpStore, String) { + ensure_fake_credentials(); + let table = unique_ops_table(ep); + let store = DynamoDbOpStore::new(table.clone(), Some(ep.to_string()), None); + (store, table) +} + +/// Every append attempt the store's test seam saw, in order. +type Attempts = Arc>>; + +/// A store that records every attempt into the returned list, and calls `also` on each. +fn recording( + store: DynamoDbOpStore, + also: impl Fn(&WriteAttempt) + Send + Sync + 'static, +) -> (DynamoDbOpStore, Attempts) { + let attempts: Attempts = Arc::default(); + let seen = Arc::clone(&attempts); + let store = store.with_before_write(move |attempt| { + seen.lock().unwrap().push(attempt.clone()); + also(attempt); }); + (store, attempts) +} + +/// `(expected, seq, counter_from, predecessor)` of one attempt. +type Seen = (i64, i64, &'static str, Option); + +/// What each attempt for `op_id` was about to write. +fn attempts_of(attempts: &Attempts, op_id: &str) -> Vec { + attempts + .lock() + .unwrap() + .iter() + .filter(|a| a.op_id == op_id) + .map(|a| (a.expected, a.seq, a.counter_from, a.predecessor)) + .collect() +} + +fn ids(ops: &[WireOp]) -> Vec<&str> { + ops.iter().map(|o| o.op_id.as_str()).collect() } #[test] fn an_op_stored_before_the_claims_is_stored_once_more_and_then_absorbed() { // The documented trade-off for streams written before 6j6v.tm4k. The legacy op sits at seq 1 - // with no counter, so the first allocation (1) also collides on the op's own seq and must be - // retried at a fresh one — the taken-seq path, for real. + // with no counter, so the first round (counter 0, seq 1) finds seq 1 occupied: the taken-seq + // path, for real. The round after steps above the highest occupied seq, still expecting the + // counter at 0, and checks that seq's op (the legacy one) as its predecessor. let Some(ep) = endpoint("an_op_stored_before_the_claims_is_stored_once_more_and_then_absorbed") else { return; @@ -350,20 +516,23 @@ fn an_op_stored_before_the_claims_is_stored_once_more_and_then_absorbed() { let s = StreamId("s".into()); let again = store.append(&s, &wire("legacy")).unwrap(); - assert!( - again > Cursor(1), - "stored once more, at a fresh seq: {again:?}" + assert_eq!( + again, + Cursor(2), + "stored once more, at the seq right above the occupied one" ); assert_eq!( store.append(&s, &wire("legacy")).unwrap(), again, "from then on its claim absorbs every re-push" ); - let (ops, _) = store.read_since(&s, Cursor::BEGINNING, 10).unwrap(); assert_eq!( - ops.iter().map(|o| o.op_id.as_str()).collect::>(), - ["legacy", "legacy"] + store.append(&s, &wire("fresh")).unwrap(), + Cursor(3), + "the counter moved to the stepped-over seq, so the next op follows it densely" ); + let (ops, _) = store.read_since(&s, Cursor::BEGINNING, 10).unwrap(); + assert_eq!(ids(&ops), ["legacy", "legacy", "fresh"]); } #[test] @@ -520,9 +689,8 @@ fn an_op_over_dynamodbs_400kb_item_ceiling_fails_loudly() { // The failure is confined to the oversized op: the stream still works afterwards, and the // failed transaction left no claim behind that would answer a retry with a seq nothing - // occupies. (The seq it consumed is gone — allocate_seq runs before the put — which is - // exactly the "gaps are normal in production" property the module doc documents.) - store.append(&s, &wire("after")).unwrap(); + // occupies — nor a moved counter: the seq it tried is taken by the next op (6j6v.r3jf). + assert_eq!(store.append(&s, &wire("after")).unwrap(), Cursor(1)); store.append(&s, &wire("huge")).unwrap(); let (ops, _) = store.read_since(&s, Cursor::BEGINNING, 10).unwrap(); assert_eq!( @@ -789,14 +957,10 @@ fn concurrent_appends_get_unique_contiguous_sequences() { .collect(); all.sort_unstable(); let expected: Vec = (1..=THREADS * PER).collect(); - // CONTIGUITY IS NOT A BACKEND GUARANTEE — do not read this assertion as a contract. The - // backend guarantees uniqueness and monotonicity; density is a happy accident of the test - // environment. `allocate_seq`'s `UpdateItem` + `ADD` is not idempotent, so any retry of - // that call — including one the SDK performs internally after a throttle or a lost - // response — burns a seq that no op will ever occupy. Gaps are therefore normal against - // real AWS. This holds here only because DynamoDB Local never throttles, and asserting - // the stronger property is deliberate: it is the sharpest available detector of the - // duplicate/overwrite bug this test exists for. + // Contiguity is a property of this build, not a contract (module doc, "Density"): since + // 6j6v.r3jf a seq is taken only by the transaction that writes its op, so neither a lost + // round nor a retry consumes one. Asserting it is deliberate: it is the sharpest available + // detector of the duplicate/overwrite bug this test exists for. assert_eq!( all, expected, "sequences are unique and contiguous under contention" @@ -842,7 +1006,7 @@ fn concurrent_appends_get_unique_contiguous_sequences() { #[test] fn concurrent_appends_to_different_streams_keep_independent_sequences() { // Independent per-stream state: hammering two DIFFERENT streams concurrently must not - // let one stream's allocations bleed into the other's — each still reaches a + // let one stream's seqs bleed into the other's — each still reaches a // contiguous, gap-free 1..=N of its own. let Some(ep) = endpoint("concurrent_appends_to_different_streams_keep_independent_sequences") else { @@ -899,6 +1063,483 @@ fn concurrent_appends_to_different_streams_keep_independent_sequences() { ); } +// ----- seq order is visibility order (6j6v.r3jf) -------------------------------------- +// +// DynamoDB Local serialises transactions, so nothing here can hold a transaction INSIDE its +// commit, and no test here shows visibility order on real AWS. The test seam holds an append +// after it has chosen its seq and before it sends its write; the predecessor check's +// transaction-against-transaction argument (module doc) is what covers the commit itself. + +/// A test-seam hook, boxed so it can be handed around by name. +type Hook = Box; + +/// One-shot gate for the test seam: the first call that `pick`s takes it, says so on the returned +/// receiver, and blocks until the returned sender fires; every other call runs straight through. +fn gate( + pick: impl Fn(&WriteAttempt) -> bool + Send + Sync + 'static, +) -> ( + Hook, + std::sync::mpsc::Receiver<()>, + std::sync::mpsc::Sender<()>, +) { + let (held_tx, held_rx) = std::sync::mpsc::channel::<()>(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let gate = std::sync::Mutex::new(Some((held_tx, release_rx))); + let hook = move |attempt: &WriteAttempt| { + if !pick(attempt) { + return; + } + let taken = gate.lock().unwrap().take(); + if let Some((held, release)) = taken { + held.send(()).unwrap(); + release.recv().unwrap(); + } + }; + (Box::new(hook), held_rx, release_tx) +} + +#[test] +fn a_reader_is_never_handed_a_seq_above_an_append_still_in_flight() { + // 6j6v.r3jf, the reproduction. Append A decides its seq and is held right there — AFTER + // choosing its seq and BEFORE sending its write (not while DynamoDB commits it, which no test + // against DynamoDB Local can do); append B then runs start to finish; a reader pulls between + // the two, exactly as the sync engine does, storing the cursor it is handed; then A is let go. + // Before the fix A held seq 10 from an `ADD`, B committed 11, the reader stored cursor 11 — + // and A, committed at 10 afterwards, was below every cursor that reader would ever send + // again: missing for good. Now a seq is taken only by the write that commits it, so B lands + // at 10, A is turned back by the counter and lands at 11, and the reader gets both. + let Some(ep) = endpoint("a_reader_is_never_handed_a_seq_above_an_append_still_in_flight") + else { + return; + }; + let (hook, held_rx, release_tx) = gate(|a| a.op_id == "a"); + let (store, attempts) = recording(opstore(&ep), hook); + let store = Arc::new(store); + let s = StreamId("s".into()); + for i in 1..=9 { + store.append(&s, &wire(&format!("o{i}"))).unwrap(); + } + let reader_cursor = Cursor(9); + + let writer_a = { + let (store, s) = (Arc::clone(&store), s.clone()); + thread::spawn(move || store.append(&s, &wire("a")).unwrap()) + }; + held_rx + .recv_timeout(Duration::from_secs(30)) + .expect("A reaches the point between choosing its seq and writing"); + let b = store.append(&s, &wire("b")).unwrap(); + + // The reader polls while A is still in flight and keeps the cursor it is handed. + let (first, reader_cursor) = store.read_since(&s, reader_cursor, 100).unwrap(); + assert_eq!(ids(&first), ["b"], "only the committed append is visible"); + assert_eq!(reader_cursor, b, "the cursor is B's seq"); + + release_tx.send(()).unwrap(); + let a = writer_a.join().unwrap(); + + let (second, _) = store.read_since(&s, reader_cursor, 100).unwrap(); + assert_eq!( + ids(&second), + ["a"], + "the reader's next pull, from the cursor it stored, still gets A: nothing was skipped" + ); + assert_eq!( + (b, a), + (Cursor(10), Cursor(11)), + "seq order is commit order, with no seq burned" + ); + assert_eq!( + attempts_of(&attempts, "a"), + [(9, 10, "read", Some(true)), (10, 11, "hint", Some(true))], + "A's second round takes the counter the failed check handed back (ALL_OLD), with no \ + read, and checks B's op below it" + ); +} + +#[test] +fn an_append_does_not_commit_over_a_missing_op_below_it_and_crosses_a_legacy_gap_once() { + // The predecessor check (review of PR #59, Integrity #1). The counter stands at 4 but op 4 is + // not there: on real AWS that is either the window in which the transaction that took 4 shows + // its counter but not yet its op — where the check makes the next append conflict and retry — + // or a seq an earlier build's `ADD` burned. Either way the round that checks op 4 must not + // commit; this pins that, and that the next round, for the same seq only, flips the check to + // `attribute_not_exists` and steps over the gap — once, without wedging the stream. + let Some(ep) = endpoint( + "an_append_does_not_commit_over_a_missing_op_below_it_and_crosses_a_legacy_gap_once", + ) else { + return; + }; + let (store, table) = opstore_and_table(&ep); + for seq in 1..=3 { + put_unclaimed_op(&ep, &table, seq, &format!("o{seq}")); + } + set_counter(&ep, &table, 4); + let probe = { + let (ep, table) = (ep.clone(), table.clone()); + move |attempt: &WriteAttempt| { + if attempt.op_id == "x" && attempt.predecessor == Some(false) { + // Between the checked round and the flipped one: the checked round wrote nothing. + assert!(!occupied(&ep, &table, 5), "the checked round stored no op"); + assert_eq!(counter_of(&ep, &table), Some(4), "nor moved the counter"); + } + } + }; + let (store, attempts) = recording(store, probe); + let s = StreamId("s".into()); + + assert_eq!(store.append(&s, &wire("x")).unwrap(), Cursor(5)); + assert_eq!( + attempts_of(&attempts, "x"), + [(4, 5, "read", Some(true)), (4, 5, "known", Some(false))], + "op 4 missing: no commit; then the same slot with the check flipped, counter not re-read" + ); + assert_eq!(store.append(&s, &wire("y")).unwrap(), Cursor(6)); + assert_eq!( + attempts_of(&attempts, "y"), + [(5, 6, "read", Some(true))], + "past the gap the check finds the op below again: one round" + ); + let (ops, _) = store.read_since(&s, Cursor::BEGINNING, 10).unwrap(); + assert_eq!(ids(&ops), ["o1", "o2", "o3", "x", "y"]); +} + +#[test] +fn an_older_relay_allocating_with_add_meanwhile_overwrites_nothing_and_never_moves_the_counter_back( +) { + // Rolling upgrade (review of PR #59, Test Quality #4), deterministic: while this build's append + // sits between choosing its seq and writing, an older relay allocates twice with a raw + // `UpdateItem ADD` and writes its ops only afterwards. Pins the counter condition: were the + // counter set unconditionally, the new append would commit at 4, move the counter back from 5 + // to 4, and the older relay's own write at 4 would be refused. + let Some(ep) = endpoint( + "an_older_relay_allocating_with_add_meanwhile_overwrites_nothing_and_never_moves_the_counter_back", + ) else { + return; + }; + let (store, table) = opstore_and_table(&ep); + let old_seqs: Arc>> = Arc::default(); + let in_hook = { + let (ep, table, old_seqs) = (ep.clone(), table.clone(), Arc::clone(&old_seqs)); + let done = std::sync::atomic::AtomicBool::new(false); + move |attempt: &WriteAttempt| { + if attempt.op_id == "new" && !done.swap(true, Ordering::SeqCst) { + let mut seqs = old_seqs.lock().unwrap(); + seqs.push(old_relay_allocates(&ep, &table)); + seqs.push(old_relay_allocates(&ep, &table)); + } + } + }; + let (store, attempts) = recording(store, in_hook); + let s = StreamId("s".into()); + for i in 1..=3 { + store.append(&s, &wire(&format!("o{i}"))).unwrap(); + } + + let new = store.append(&s, &wire("new")).unwrap(); + let old = old_seqs.lock().unwrap().clone(); + assert_eq!(old, [4, 5], "the older relay allocated 4 and 5 by ADD"); + assert_eq!(new, Cursor(6), "this build steps above both allocations"); + assert_eq!( + attempts_of(&attempts, "new"), + [ + (3, 4, "read", Some(true)), + (5, 6, "hint", Some(true)), + (5, 6, "known", Some(false)), + ], + "turned back by the moved counter; op 5 not there yet, so the check flips once" + ); + assert_eq!( + counter_of(&ep, &table), + Some(6), + "the counter never went below 5" + ); + + // The older relay now writes the ops it allocated, as it does: conditioned on the key. + assert!( + put_op(&ep, &table, 4, "old-a", true), + "seq 4 is still free for it" + ); + assert!(put_op(&ep, &table, 5, "old-b", true), "and so is seq 5"); + assert_eq!(counter_of(&ep, &table), Some(6)); + + let (ops, end) = store.read_since(&s, Cursor::BEGINNING, 100).unwrap(); + assert_eq!( + ids(&ops), + ["o1", "o2", "o3", "old-a", "old-b", "new"], + "nothing overwritten, every op readable in seq order" + ); + assert_eq!(end, Cursor(6)); + assert_eq!( + store.append(&s, &wire("after")).unwrap(), + Cursor(7), + "and the stream goes on" + ); +} + +#[test] +fn a_re_push_that_lost_the_counter_race_answers_its_own_seq_and_burns_none() { + // Review of PR #59, Test Quality #5, deterministic: the re-push of o1 chooses seq 2 and is held + // there; o2 commits at 2; the re-push's transaction then fails on BOTH the moved counter and + // its taken claim, and the claim must win: o1's own seq, in that one round, and the next fresh + // op right after o2. + let Some(ep) = + endpoint("a_re_push_that_lost_the_counter_race_answers_its_own_seq_and_burns_none") + else { + return; + }; + let (hook, held_rx, release_tx) = gate(|a| a.op_id == "o1" && a.expected == 1); + let (store, attempts) = recording(opstore(&ep), hook); + let store = Arc::new(store); + let s = StreamId("s".into()); + assert_eq!(store.append(&s, &wire("o1")).unwrap(), Cursor(1)); + + let re_push = { + let (store, s) = (Arc::clone(&store), s.clone()); + thread::spawn(move || store.append(&s, &wire("o1")).unwrap()) + }; + held_rx + .recv_timeout(Duration::from_secs(30)) + .expect("the re-push reaches its write"); + assert_eq!(store.append(&s, &wire("o2")).unwrap(), Cursor(2)); + release_tx.send(()).unwrap(); + + assert_eq!(re_push.join().unwrap(), Cursor(1), "o1's own seq"); + assert_eq!( + attempts_of(&attempts, "o1"), + [(0, 1, "read", None), (1, 2, "read", Some(true))], + "the re-push settled in its one round: the taken claim outranked the moved counter" + ); + assert_eq!( + store.append(&s, &wire("o3")).unwrap(), + Cursor(3), + "no seq burned: the next op lands right after o2" + ); + let (ops, _) = store.read_since(&s, Cursor::BEGINNING, 10).unwrap(); + assert_eq!(ids(&ops), ["o1", "o2", "o3"]); +} + +#[test] +fn a_run_of_occupied_seqs_above_the_counter_is_stepped_over_in_one_round() { + // Review of PR #59, Integrity #4: forty ops with no counter (out-of-band writes, or a stream + // older than the counter). Stepping one seq per round would exhaust APPEND_ATTEMPTS (32) and + // wedge the stream; one reverse Query finds the top and the next round lands above it. + let Some(ep) = + endpoint("a_run_of_occupied_seqs_above_the_counter_is_stepped_over_in_one_round") + else { + return; + }; + let (store, table) = opstore_and_table(&ep); + for seq in 1..=40 { + put_unclaimed_op(&ep, &table, seq, &format!("legacy-{seq}")); + } + let (store, attempts) = recording(store, |_| {}); + let s = StreamId("s".into()); + + assert_eq!(store.append(&s, &wire("x")).unwrap(), Cursor(41)); + assert_eq!( + attempts_of(&attempts, "x"), + [(0, 1, "read", None), (0, 41, "known", Some(true))], + "one taken seq, then straight above the top one, the counter not re-read" + ); + assert_eq!(store.append(&s, &wire("y")).unwrap(), Cursor(42)); +} + +#[test] +fn a_push_carries_the_seq_it_wrote_forward_and_reads_the_counter_once() { + // Review of PR #59, Integrity #3: within one push the seq an op was written at is the counter + // the next op expects, so a push of n ops costs n transactions and one read. + let Some(ep) = endpoint("a_push_carries_the_seq_it_wrote_forward_and_reads_the_counter_once") + else { + return; + }; + let (store, attempts) = recording(opstore(&ep), |_| {}); + let s = StreamId("s".into()); + store.append(&s, &wire("o1")).unwrap(); + + let push = [wire("o2"), wire("o1"), wire("o3"), wire("o4")]; + store + .append_batch(&s, &push, Instant::now() + Duration::from_secs(60)) + .unwrap(); + let from: Vec<(String, &str)> = attempts + .lock() + .unwrap() + .iter() + .skip(1) + .map(|a| (a.op_id.clone(), a.counter_from)) + .collect(); + assert_eq!( + from, + [ + ("o2".to_string(), "read"), + ("o1".to_string(), "carried"), + ("o3".to_string(), "carried"), + ("o4".to_string(), "carried"), + ], + "read once; an already stored op (o1) carries the same expectation on" + ); + let (ops, end) = store.read_since(&s, Cursor::BEGINNING, 10).unwrap(); + assert_eq!(ids(&ops), ["o1", "o2", "o3", "o4"]); + assert_eq!(end, Cursor(4)); +} + +#[test] +fn a_push_stops_at_its_deadline_and_keeps_what_it_stored() { + // Review of PR #59, Integrity #2: the deadline is checked between ops and against the round in + // flight. Here the second op's round is held past it. + let Some(ep) = endpoint("a_push_stops_at_its_deadline_and_keeps_what_it_stored") else { + return; + }; + // The held op sleeps until just past the deadline the push runs under — whatever that is — + // so the test does not depend on how fast DynamoDB Local serves the op before it. + let push_deadline: Arc>> = Arc::default(); + let store = { + let push_deadline = Arc::clone(&push_deadline); + opstore(&ep).with_before_write(move |attempt| { + let deadline = *push_deadline.lock().unwrap(); + if let (Some(deadline), "b") = (deadline, attempt.op_id.as_str()) { + thread::sleep(deadline.saturating_duration_since(Instant::now())); + thread::sleep(Duration::from_millis(50)); + } + }) + }; + let s = StreamId("s".into()); + + let err = store + .append_batch(&s, &[wire("a")], Instant::now()) + .expect_err("a deadline already passed stores nothing"); + assert!(err.to_string().contains("after 0 of 1"), "{err}"); + let (ops, _) = store.read_since(&s, Cursor::BEGINNING, 10).unwrap(); + assert!(ops.is_empty()); + + let push = [wire("a"), wire("b"), wire("c")]; + let deadline = Instant::now() + Duration::from_secs(3); + *push_deadline.lock().unwrap() = Some(deadline); + let err = store + .append_batch(&s, &push, deadline) + .expect_err("the held op runs past the deadline"); + *push_deadline.lock().unwrap() = None; + assert!(err.to_string().contains("after 1 of 3"), "{err}"); + let (ops, _) = store.read_since(&s, Cursor::BEGINNING, 10).unwrap(); + assert_eq!( + ids(&ops)[0], + "a", + "what was stored before the deadline stays" + ); + assert!( + !ids(&ops).contains(&"c"), + "nothing after the deadline is sent" + ); + + // The client's re-push of the whole page then lands every op exactly once. + store + .append_batch(&s, &push, Instant::now() + Duration::from_secs(60)) + .unwrap(); + let (ops, end) = store.read_since(&s, Cursor::BEGINNING, 10).unwrap(); + assert_eq!(ids(&ops), ["a", "b", "c"]); + assert_eq!(end, Cursor(3)); +} + +#[test] +fn soak_concurrent_writers_and_pullers_see_every_op_once_in_gap_free_seq_order() { + // A SOAK test, not a proof: nothing here forces the interleaving the reproduction test above + // forces, so it catches an ordering bug only when the scheduler happens to produce it (and + // DynamoDB Local serialises transactions besides). It stays as a regression net over the + // whole path — many writers, pullers following the stream. + // + // N writers push in parallel while pullers follow the stream the way the sync engine does — + // `since` is the cursor of the previous answer, stored unconditionally. Every puller must end + // up with every op exactly once, and the one-op-per-page puller sees the seqs themselves: + // strictly increasing by one, no seq skipped. + let Some(ep) = + endpoint("soak_concurrent_writers_and_pullers_see_every_op_once_in_gap_free_seq_order") + else { + return; + }; + let store = Arc::new(opstore(&ep)); + let s = StreamId("s".into()); + const WRITERS: usize = 6; + const PER: usize = 30; + const TOTAL: usize = WRITERS * PER; + let done = Arc::new(std::sync::atomic::AtomicBool::new(false)); + + let pullers: Vec<_> = [1usize, 7, 64] + .into_iter() + .map(|page| { + let (store, s, done) = (Arc::clone(&store), s.clone(), Arc::clone(&done)); + thread::spawn(move || { + let mut cursor = Cursor::BEGINNING; + let mut seen: Vec<(String, Cursor)> = Vec::new(); + loop { + // Read the flag BEFORE the pull: once it is set, a pull that comes back empty + // has seen everything every writer committed. + let finished = done.load(Ordering::SeqCst); + let (ops, next) = store.read_since(&s, cursor, page).unwrap(); + let empty = ops.is_empty(); + seen.extend(ops.into_iter().map(|o| (o.op_id, next))); + cursor = next.max(cursor); + if empty && finished { + return (page, seen); + } + } + }) + }) + .collect(); + + let writers: Vec<_> = (0..WRITERS) + .map(|w| { + let (store, s) = (Arc::clone(&store), s.clone()); + thread::spawn(move || { + (0..PER) + .map(|i| store.append(&s, &wire(&format!("w{w}-{i}"))).unwrap().0) + .collect::>() + }) + }) + .collect(); + let mut answered: Vec = writers + .into_iter() + .flat_map(|h| h.join().unwrap()) + .collect(); + done.store(true, Ordering::SeqCst); + answered.sort_unstable(); + assert_eq!( + answered, + (1..=TOTAL as i64).collect::>(), + "every append is answered with its own seq, none burned" + ); + + let mut expected: Vec = (0..WRITERS) + .flat_map(|w| (0..PER).map(move |i| format!("w{w}-{i}"))) + .collect(); + expected.sort(); + for puller in pullers { + let (page, seen) = puller.join().unwrap(); + let mut ids: Vec<&str> = seen.iter().map(|(id, _)| id.as_str()).collect(); + ids.sort_unstable(); + let missing: Vec<&String> = expected + .iter() + .filter(|id| ids.binary_search(&id.as_str()).is_err()) + .collect(); + assert!( + missing.is_empty(), + "the page-{page} puller never received {missing:?}" + ); + assert_eq!( + ids.len(), + TOTAL, + "the page-{page} puller received an op more than once" + ); + if page == 1 { + let seqs: Vec = seen.iter().map(|(_, at)| at.0).collect(); + assert_eq!( + seqs, + (1..=TOTAL as i64).collect::>(), + "one op per page: the seqs arrive strictly increasing, none skipped" + ); + } + } +} + // ----- DynamoDB PrefixRegistry — claim-if-free --------------------------------------- #[test] diff --git a/crates/server/tests/relay_http.rs b/crates/server/tests/relay_http.rs index 335f88e..59ac5f1 100644 --- a/crates/server/tests/relay_http.rs +++ b/crates/server/tests/relay_http.rs @@ -4,7 +4,7 @@ use std::sync::atomic::{AtomicI64, Ordering}; use std::sync::Arc; -use nxs_server::app::{app, app_with_clock, MAX_PUSH_OPS}; +use nxs_server::app::{app, app_with_clock, app_with_push_deadline, system_clock, MAX_PUSH_OPS}; use nxs_server::registry::PrefixRegistry; use nxs_server::registry::SqlitePrefixRegistry; use nxs_server::store::SqliteOpStore; @@ -510,3 +510,94 @@ fn a_registry_written_before_presence_still_builds_the_relay_which_then_answers_ ); assert_eq!(push(&base, "s1", &[wire("o1", "set")]).appended, 1); } + +// ----- the push deadline (6j6v.r3jf) ---------------------------------------------------------- + +/// A store whose every append takes `delay` — a stand-in for a contended DynamoDB stream — and +/// that fails outright for an op id starting with "broken". +struct SlowStore { + inner: SqliteOpStore, + delay: std::time::Duration, +} + +impl nxs_server::store::OpStore for SlowStore { + fn append(&self, stream: &StreamId, op: &WireOp) -> StoreResult { + if op.op_id.starts_with("broken") { + return Err(nxs_server::store::StoreError("broken on purpose".into())); + } + std::thread::sleep(self.delay); + self.inner.append(stream, op) + } + + fn read_since( + &self, + stream: &StreamId, + cursor: nxs_sync::protocol::Cursor, + limit: usize, + ) -> StoreResult<(Vec, nxs_sync::protocol::Cursor)> { + self.inner.read_since(stream, cursor, limit) + } +} + +fn spawn_slow_relay(delay: std::time::Duration, deadline: std::time::Duration) -> String { + let ops = Arc::new(SlowStore { + inner: SqliteOpStore::open_in_memory().unwrap(), + delay, + }); + let registry = Arc::new(SqlitePrefixRegistry::open_in_memory().unwrap()); + let router = app_with_push_deadline(ops, registry, system_clock(), deadline); + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + listener.set_nonblocking(true).unwrap(); + std::thread::spawn(move || { + let rt = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .unwrap(); + rt.block_on(async move { + let listener = tokio::net::TcpListener::from_std(listener).unwrap(); + axum::serve(listener, router).await.unwrap(); + }); + }); + format!("http://{addr}") +} + +fn push_status(base: &str, stream: &str, ops: &[WireOp]) -> u16 { + let body = serde_json::json!({ "ops": ops }); + match ureq::post(&format!("{base}/streams/{stream}/ops")).send_json(body) { + Ok(resp) => resp.status(), + Err(ureq::Error::Status(status, _)) => status, + Err(e) => panic!("transport error: {e}"), + } +} + +#[test] +fn a_push_past_its_deadline_stops_writing_and_answers_503() { + // Review of PR #59, Integrity #2: a push that cannot finish before the front door gives up + // stops writing and answers, instead of writing on while the client re-pushes. 503, which the + // client handles like any failed push (every status but 413): nothing is marked pushed, and + // the next pass re-pushes the page, which idempotence by op id makes safe. + let base = spawn_slow_relay( + std::time::Duration::from_millis(100), + std::time::Duration::from_millis(250), + ); + let ops: Vec = (0..10).map(|i| wire(&format!("o{i}"), "set")).collect(); + let started = std::time::Instant::now(); + assert_eq!(push_status(&base, "s1", &ops), 503); + assert!( + started.elapsed() < std::time::Duration::from_millis(900), + "stopped near the deadline, not after all ten ops: {:?}", + started.elapsed() + ); + let stored = pull(&base, "s1", 0, 100).ops; + assert!( + !stored.is_empty() && stored.len() < ops.len(), + "what was appended before the deadline stays, the rest was never sent: {}", + stored.len() + ); + + // A store fault before the deadline is still a 500. + let failing = [wire("broken-1", "set")]; + assert_eq!(push_status(&base, "s2", &failing), 500); +} diff --git a/crates/server/tests/support/idempotence.rs b/crates/server/tests/support/idempotence.rs index 99cdb4f..e2687ea 100644 --- a/crates/server/tests/support/idempotence.rs +++ b/crates/server/tests/support/idempotence.rs @@ -1,7 +1,7 @@ //! The idempotent-append contract every `OpStore` keeps (spec §4.4, 6j6v.tm4k), as ONE script the -//! SQLite, Postgres and DynamoDB suites each run against their backend. The backends may differ in -//! one documented way only: SQLite and Postgres consume no seq for a re-push (`dense`), DynamoDB -//! burns the one it allocated. +//! SQLite, Postgres and DynamoDB suites each run against their backend. `dense` asserts that a +//! re-push consumes no seq. All three backends keep it today; DynamoDB does since 6j6v.r3jf moved +//! its seq into the write transaction (before, it burned the seq it had allocated). use nxs_server::store::OpStore; use nxs_sync::protocol::{Cursor, StreamId}; @@ -39,4 +39,18 @@ pub fn assert_idempotent_append(store: &dyn OpStore, wire: impl Fn(&str) -> Wire "each op is stored once" ); assert_eq!(end, o2); + + // The re-pushes after o2 consumed no seq either: the next fresh op follows o2 directly. + let o3 = store.append(&s, &wire("o3")).unwrap(); + assert!( + o3 > o2, + "a fresh op after the re-pushes still advances the stream" + ); + if dense { + assert_eq!( + o3, + Cursor(o2.0 + 1), + "this backend consumed no seq for the later re-pushes" + ); + } } diff --git a/docs/specs/E4-sync-protocol-server.md b/docs/specs/E4-sync-protocol-server.md index 01d63b0..08b69a4 100644 --- a/docs/specs/E4-sync-protocol-server.md +++ b/docs/specs/E4-sync-protocol-server.md @@ -148,9 +148,11 @@ push/pull therefore converges without double-counting. The relay keeps the same union on its side: every backend appends idempotently by `op_id`, so a re-pushed op answers with the cursor it already has and the log does not grow. SQLite and Postgres do it with `UNIQUE (stream_id, op_id)`; DynamoDB, which has no secondary unique constraint, with a -claim item per op written in the same transaction (`6j6v.tm4k`, contract in `store_ddb.rs`). There -a re-push burns the seq it allocated, so the sequence gaps where SQLite's does not; nothing reads -density. +claim item per op written in the same transaction (`6j6v.tm4k`, contract in `store_ddb.rs`). Since +`6j6v.r3jf` that transaction also takes the seq, conditioned on the stream's counter, and checks +the op one seq below (a `ConditionCheck`, which needs `dynamodb:ConditionCheckItem` on the op table), +so on every backend a seq becomes visible only after every lower one, and a re-push consumes none; +nothing reads density. ### 4.4.1 Opaque chunks (added 2026-10-08, `6j6v.c0kn`) @@ -252,7 +254,9 @@ trait PrefixRegistry { - **DynamoDB:** PK = `stream_id`, SK = `seq`. `read_since` = range query `SK > cursor`. The **main risk is the atomic monotonic sequence per stream** (atomic counter item / conditional put), **not** the 400 KB item limit — the trait already handles that via `limit` + - `next cursor` pagination. + `next cursor` pagination. **Impl note (`6j6v.r3jf`):** the seq is taken by the transaction that + writes the op — counter `SET` conditioned on the value read, plus a `ConditionCheck` on the op + one seq below — so commit visibility follows seq order; contract and IAM in `store_ddb.rs`. Slice 1: a single SQLite impl carries **both** traits and proves convergence. The durable backends (§10) are split **per trait**, because `PrefixRegistry` only comes into being with diff --git a/docs/specs/streams-without-lifetime-growth.md b/docs/specs/streams-without-lifetime-growth.md index 767d482..8e51a73 100644 --- a/docs/specs/streams-without-lifetime-growth.md +++ b/docs/specs/streams-without-lifetime-growth.md @@ -114,7 +114,7 @@ The chat message reducer's "first write wins by ARRIVAL", recorded on `47jy.d28m | # | Where | What decides | Two-op example | Who hits it | Snapshot at P + tail ≠ full replay? | |---|---|---|---|---|---| -| 1 | DynamoDB relay: seq is allocated by `ADD` (`crates/server/src/store_ddb.rs:505-525`) before the op is written (`:716-742`). `read_since` returns `seq > since` and sets `next` to the last seq seen (`:780,793-795`). The client stores that cursor unconditionally (`crates/sync/src/engine.rs:486`). fold-ddb advances to the batch's max seq (`crates/fold-ddb/src/fold.rs:336-338`), and `advance` never moves back (`:343`). | Commit order vs. seq order of concurrent appends | Writer A gets seq 10, writer B gets 11, and B commits first. A reader at `since=9` gets [11] and stores cursor 11. Then A@10 commits. That reader never asks below 11 again, so A is **missing for good** on it. A replica that pulled later has A. | **Honest replicas**, whenever two pushes to one stream overlap: a service and a manual pass, two devices, the hosted writer and a device. SQLite and Postgres relays are immune: a mutex (`crates/server/src/store.rs:206-212`) and an advisory lock held to commit (`crates/server/src/store_pg.rs:334,352-356`). | **Yes, and it propagates.** A snapshot at P from a replica that skipped op 10 claims P and omits 10. Every replica started from it misses 10 too. The anchor check only compares the op AT P (`crates/sync/src/snapshot.rs:422-440`). r3jf rated live reading harmless. That holds for the fold, which folds a late op when it arrives through the DynamoDB stream, but **not for pullers**. **[inferred from code, not reproduced]** | +| 1 | DynamoDB relay, before #59 (`r3jf`): the seq was allocated by an `UpdateItem ADD` on the counter item before the op was written by its own transaction (`allocate_seq` and `try_append_op` in `crates/server/src/store_ddb.rs` at `130c048`). **Fixed by #59, see §4.1.** `read_since` returns `seq > since` and sets `next` to the last seq seen. The client stores that cursor unconditionally (`crates/sync/src/engine.rs:486`). fold-ddb advances to the batch's max seq (`crates/fold-ddb/src/fold.rs:336-338`), and `advance` never moves back (`:343`). | Commit order vs. seq order of concurrent appends | Writer A gets seq 10, writer B gets 11, and B commits first. A reader at `since=9` gets [11] and stores cursor 11. Then A@10 commits. That reader never asks below 11 again, so A is **missing for good** on it. A replica that pulled later has A. | **Honest replicas**, whenever two pushes to one stream overlap: a service and a manual pass, two devices, the hosted writer and a device. SQLite and Postgres relays are immune: a mutex (`crates/server/src/store.rs:206-212`) and an advisory lock held to commit (`crates/server/src/store_pg.rs:334,352-356`). | **Yes, and it propagates.** A snapshot at P from a replica that skipped op 10 claims P and omits 10. Every replica started from it misses 10 too. The anchor check only compares the op AT P (`crates/sync/src/snapshot.rs:422-440`). r3jf rated live reading harmless. That holds for the fold, which folds a late op when it arrives through the DynamoDB stream, but **not for pullers**. **[reproduced: #59's test `a_reader_is_never_handed_a_seq_above_an_append_still_in_flight`, backported onto the `ADD` allocation, fails with the second pull empty]** | | 2 | Rows holding several registers whose version columns default to 0: `items` (`crates/core/src/schema.rs:69-73`), and chat `channels`, `profiles`, `threads`. The fresh-row INSERT writes unconditionally (`change.rs:246-272`; DynamoDB `crates/fold-ddb/src/write.rs:147-156`). `lamport_in_bound` bounds only the top (`crates/foundation/src/model.rs:100-102`). | Which op creates the row | A sets `title=x` at (−1, 5). B sets `status=open` at (5, 2). In order AB the row is created with `title=x`. In order BA, B creates the row with `title_v=0`, and A needs (−1,5) > (0,0), so it is refused: `title` stays NULL. | Hostile or malformed ops only: honest clocks start at 1 (`crates/foundation/src/store.rs:226-231`). | Yes: the snapshot freezes its source's arrival order. A local full refold folds in `(lamport, site)` order (`store.rs:573-578`), so it can disagree even with the replica's own incremental fold. | | 3 | Two different ops with the same `(lamport, site)`. Locally the unique coordinate index makes the second one `Collided` and it is never stored (`store.rs:499-531`). The server folds both, and an equal version keeps the first (`fold.rs:38-44`). | Which copy arrives first | A and B are both at (7,3) and both set `title`. Order AB keeps A, order BA keeps B. | Logs written before fc5p only: four pairs in this board's own log (`hehx` point 4). | Yes: every snapshot lineage inherits its source's choice. | | 4 | Two copies of one `op_id`: `INSERT OR IGNORE` keeps the first copy's content, signature and verdict (`store.rs:496-531`). | Which copy arrives first | A stripped copy arrives first and is stored `unsigned`. The signed copy arrives later and is ignored. `acting_ops` then differs between replicas (`crates/foundation/src/trust.rs:123-127`), and so do chat's `held` and `answers` (`crates/chat/src/store.rs:344-363,417-438`). | A stripping or hostile transport. | State: no (same content). Actions: yes. A snapshot carries its source's `key_id`/`sig`; the importer re-verifies them but has no other copy (`crates/foundation/src/image.rs:468-470`). | @@ -200,29 +200,78 @@ every platform reducer's views. fold-ddb's `platform_reducers()` already does ### 4.1 Precondition: relay seq order equals visibility order (r3jf) — lands separately **Owner, 2026-10-12: fixed immediately, as its own small PR with a reproduction test, before slice -1.** Slice 1 assumes it is merged and released; this section records what that PR does so the -spec stays readable on its own. The reproduction test comes first: two concurrent appends on -dynamodb-local with the second one's write delayed between allocation and `TransactWriteItems`, -and a reader polling between them, which today stores a cursor past the missing seq. - - -A snapshot at P must contain every op with seq ≤ P. The design makes the DynamoDB relay assign the -seq inside the write, as the other two backends do. - -- **Fix.** `try_append_op` reads the counter. Its `TransactWriteItems` then carries three things: - - `Update counter SET next_seq = :n` with the condition `next_seq = :n-1`; - - the claim (as today); - - the op. - - Concurrent appends to a stream serialize on the counter item. A loser retries with a fresh read. - Commit order becomes seq order, as with Postgres' advisory lock. -- **Cost.** One strongly consistent `GetItem` more per append. Contention retries on a hot stream - are bounded by `APPEND_ATTEMPTS = 32` (`store_ddb.rs:341`). DynamoDB caps a transaction at 100 - items; this one has 3. **[unverified]:** throughput under contention. Measure it in slice 1 with - dynamodb-local. -- **What is rejected.** A "visibility horizon" (serving only seqs older than N seconds) is rejected: - gaps are normal on this backend (`store_ddb.rs:101-106,126-130`), so a hole cannot be told from a - burned seq. +1.** Slice 1 assumes it is merged and released; this section records what that PR (#59) built, so +the spec stays readable on its own. The reproduction test came first: two appends on +dynamodb-local, the first one held by a test seam after it has chosen its seq and before it sends +its write, the second one committing meanwhile, and a reader polling between them — which on the +`ADD` allocation stores a cursor past the missing seq. + +A snapshot at P must contain every op with seq ≤ P. The DynamoDB relay now assigns the seq inside +the write, as the other two backends do. + +- **Fix.** Each append round knows a counter value `c` — read with one strongly consistent `Query` + of the counter item for the first op of a push, otherwise learned without a read (handed back by + a failed counter check through `ReturnValuesOnConditionCheckFailure = ALL_OLD`, or carried from + the op the same push wrote before). Its `TransactWriteItems` then carries up to four items: + - `Update counter SET next_seq = :seq` (`s = c + 1`) with the condition `next_seq = :expected` + (`c`; for `c = 0`, also "no counter yet"); + - the claim (as before); + - for `s > 1`, a `ConditionCheck attribute_exists` on the op at `s − 1` — the **predecessor + check**; + - the op, conditioned on its key being free. + + Concurrent appends to a stream serialize on the counter item; a loser retries from the counter + the failed check handed back, after a jitter of at most 16 ms. +- **Why the predecessor check (review of #59, Integrity #1).** AWS documents transactions as + serializable against each other, but a `Query` only as read-committed against a transaction: a + transaction's items need not become visible to a reader at the same instant. The counter + condition alone orders the transactions, not what a reader sees — the transaction that took `c` + can show its counter before its op, the next one can write op `c + 1` and commit, and a reader in + that window is handed `[c + 1]`. The predecessor check makes op `c`'s item part of the next + transaction, so that transaction conflicts (`TransactionConflict`, retried as a lost round) while + op `c` is still being written and can commit only once it exists. Commit visibility then follows + seq order — transaction against transaction, the case AWS documents — as with Postgres' advisory + lock. +- **Legacy gaps.** When op `s − 1` is legitimately absent (a seq an `ADD`-era relay burned), a failed + `attribute_exists` with the counter condition holding proves it: no transaction of this build can + still write that seq. The next round, for the same `s` only, checks `attribute_not_exists` + instead — still an item of the transaction, so a writer of it still conflicts — and steps over + the gap. One extra round per legacy gap; the stream never wedges. Seq 1 has no check; a floor + step (below) checks the occupied seq it steps over. +- **Occupied seqs above the counter** (a stream older than the counter, out-of-band writes) are + found with one reverse `Query` (`Limit 1`), and the next round takes the seq above the top one — + one extra round for a run of any length, not one per seq. Seqs past `i64::MAX` are a loud error. +- **IAM — changed.** The predecessor check needs **`dynamodb:ConditionCheckItem`** on the op table. + It must be granted **before** a relay of this build is deployed; without it every push past a + stream's first op fails with `AccessDeniedException` (a 500). The op table's role now needs + `Query`, `PutItem`, `UpdateItem` and `ConditionCheckItem`. Wire protocol and table shape are + unchanged. +- **Cost.** Uncontended, one transaction per op plus one counter read per push. A transaction is + billed at twice the plain write per item; the predecessor check is a transactional read of op + `s − 1`. Under contention a lost round is a cancelled transaction, and cancelled transactions + consume capacity like committed ones, so `N` writers racing for one stream cost up to about `N` + times the capacity per stored op; two transactions can also cancel each other, so there is not + always a winner per round. The per-stream ceiling is about one append per counter round-trip — + 30-60 appends/s by latency arithmetic. **[unverified]:** throughput and capacity under contention + on real AWS; not measurable on dynamodb-local, which serialises transactions (chore filed with + #59). Bounds: 256 rounds lost in a row (`CONTENTION_ROUNDS`), 32 rounds failing otherwise + (`APPEND_ATTEMPTS`), and a per-push deadline of 20 s (`PUSH_DEADLINE`, configurable as + `NXF_RELAY_PUSH_DEADLINE_SECS`), below the ~30 s of API Gateway and CloudFront in front of the + hosted relay; checked between ops, between rounds and against the round in flight, answered 503, + which the client re-pushes like any failed push. +- **Density.** A stream written by this build has no gaps: a seq is taken only together with its + op, so neither a re-push, a lost round nor an SDK retry consumes one. Streams written before + carry the `ADD`'s gaps. Nothing downstream depends on density. +- **What is rejected.** A "visibility horizon" (serving only seqs older than N seconds): a time + bound is a guess about commit latency, not an ordering, and the old streams' burned seqs make a + hole indistinguishable from a gap anyway. Also rejected (review of #59): bounding `read_since` + by the counter and stopping a page at an unexpected gap — it would put the ordering into every + reader instead of the one writer path, and legacy gaps would stall it. +- **Rolling upgrade.** Grant `ConditionCheckItem` first. Mixed with an `ADD`-era relay nothing is + overwritten and the counter never moves backwards; the ordering guarantee holds once every relay + writing a table runs the new build. Meanwhile the old relays answer more pushes with a 500 (their + `ADD` is cancelled by a `TransactionConflict` whenever a new relay's transaction holds the + counter). - **What it does not repair.** Ops already skipped by today's replicas stay skipped. The writer check in §4.3 (count and digest ≤ P) detects such a replica before it publishes.