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

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.

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.
 

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.
 

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"] }

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(),
     }))

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<String>) -> std::time::Duration {
+    match raw {
+        None => PUSH_DEADLINE,
+        Some(raw) => match raw.trim().parse::<u64>() {
+            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]

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<Cursor>;
 
+    /// 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.
     ///