diff --git a/Cargo.lock b/Cargo.lock index 56bf80e..200b6ad 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -51,6 +51,8 @@ dependencies = [ "axum", "base64", "chrono", + "ciborium", + "cid", "dotenvy", "futures", "p256", @@ -60,6 +62,7 @@ dependencies = [ "serde_json", "sqlx", "tokio", + "tokio-tungstenite", "tower", "tower-http", "tracing", @@ -287,6 +290,7 @@ checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f" dependencies = [ "async-trait", "axum-core", + "base64", "bytes", "futures-util", "http", @@ -305,8 +309,10 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", + "sha1", "sync_wrapper", "tokio", + "tokio-tungstenite", "tower", "tower-layer", "tower-service", @@ -1885,6 +1891,7 @@ dependencies = [ "ciborium", "cid", "dotenvy", + "futures", "hex", "k256", "p256", @@ -1895,6 +1902,7 @@ dependencies = [ "sha2", "sqlx", "tokio", + "tokio-tungstenite", "tower", "tower-http", "tracing", diff --git a/crates/pds-server/Cargo.toml b/crates/pds-server/Cargo.toml index f56da4f..3696e99 100644 --- a/crates/pds-server/Cargo.toml +++ b/crates/pds-server/Cargo.toml @@ -16,7 +16,7 @@ path = "src/main.rs" [dependencies] tokio = { workspace = true } dotenvy = { workspace = true } -axum = { workspace = true } +axum = { workspace = true, features = ["ws"] } tower = { workspace = true } tower-http = { workspace = true } serde = { workspace = true } @@ -39,6 +39,7 @@ hex = { workspace = true } rand = { workspace = true } uuid = { workspace = true } bytes = { workspace = true } +futures = { workspace = true } cid = { workspace = true } k256 = { workspace = true } p256 = { workspace = true } @@ -56,3 +57,6 @@ sha2 = { workspace = true } hex = { workspace = true } sqlx = { workspace = true } at-crypto = { workspace = true } +futures = { workspace = true } +tokio-tungstenite = { workspace = true } +chrono = { workspace = true } diff --git a/crates/pds-server/src/dag_cbor.rs b/crates/pds-server/src/dag_cbor.rs new file mode 100644 index 0000000..fd43cd9 --- /dev/null +++ b/crates/pds-server/src/dag_cbor.rs @@ -0,0 +1,557 @@ +//! A small, self-contained DAG-CBOR encoder + decoder. +//! +//! Why this exists +//! +//! `ciborium` (the CBOR crate the rest of the workspace uses) speaks plain +//! CBOR through serde. It has no notion of an IPLD *CID link*, which DAG-CBOR +//! encodes as the IANA-registered tag `42` wrapping a byte string whose first +//! byte is the multibase-identity prefix `0x00` followed by the binary CID. +//! Serde has no representation for a CBOR tag, so `ciborium` silently encodes +//! `cid::Cid` as a newtype struct instead — which is *not* DAG-CBOR and which +//! no atproto consumer can read. +//! +//! `car.rs` already hand-rolled the handful of primitives needed for the CAR +//! v1 header (`{version, roots: []}`). The firehose frames need +//! exactly the same primitives plus a couple more (arrays of maps, nullable +//! links, byte strings, i64). Rather than write the encoder twice, both +//! callers now go through this module. +//! +//! ## What "correct DAG-CBOR" means here +//! +//! * Map keys are text strings, sorted in the DAG-CBOR canonical order: +//! shorter keys first, then bytewise-ascending within a length. This is the +//! ordering `@ipld/dag-cbor` (and therefore the reference atproto +//! implementation) emits, so a frame produced here is byte-identical to one +//! produced by a Typescript PDS for the same logical value. +//! * Integers use the shortest possible head. Byte and text strings likewise. +//! * A CID link is `tag(42) || bytes(0x00 || )`. The leading +//! `0x00` is the multibase identity prefix mandated by the DAG-CBOR spec for +//! binary CIDs; forgetting it is the single most common interop bug, so the +//! decoder asserts on it too. +//! * Floats are deliberately *not* supported. DAG-CBOR permits them but +//! nothing in this codebase emits one, and accepting them would mean +//! deciding on a canonical float encoding we would never exercise. +//! +//! ## Scope +//! +//! This is not a general CBOR library. It handles definite-length items only +//! (DAG-CBOR forbids indefinite lengths anyway) and rejects everything it does +//! not understand rather than guessing. It is used for the *envelope* of +//! things — CAR headers and firehose frames — never for repo blocks, which +//! are produced by `at_repo`/`at_crypto` with their own (see +//! `crate::firehose`) conventions. + +use anyhow::{anyhow, bail, Result}; +use cid::Cid; + +/// The IPLD CID-link tag. See . +pub const CID_TAG: u64 = 42; + +/// A decoded (or to-be-encoded) DAG-CBOR value. +/// +/// `Link` is kept distinct from `Bytes` so a round-trip through +/// [`decode`] / [`encode`] preserves the tag rather than flattening a link +/// into an anonymous byte string. +#[derive(Debug, Clone, PartialEq)] +pub enum Value { + Null, + Bool(bool), + /// Signed integer. CBOR major types 0 (non-negative) and 1 (negative). + Int(i64), + Bytes(Vec), + Text(String), + Array(Vec), + /// Map with text keys. Insertion order is irrelevant — [`encode`] sorts + /// into the canonical DAG-CBOR order, and [`decode`] returns keys in the + /// order they appeared on the wire. + Map(Vec<(String, Value)>), + /// An IPLD CID link — `tag(42)` wrapping the identity-prefixed CID bytes. + Link(Cid), +} + +impl Value { + /// Convenience: build a `Map` from an iterator of pairs. + pub fn map(pairs: I) -> Value + where + I: IntoIterator, + K: Into, + { + Value::Map(pairs.into_iter().map(|(k, v)| (k.into(), v)).collect()) + } + + /// Convenience: a text value from anything string-ish. + pub fn text(s: impl Into) -> Value { + Value::Text(s.into()) + } + + /// Look up a key in a `Map`. Returns `None` for a non-map or a + /// missing key. Used by the tests and by frame consumers. + pub fn get(&self, key: &str) -> Option<&Value> { + match self { + Value::Map(entries) => entries.iter().find(|(k, _)| k == key).map(|(_, v)| v), + _ => None, + } + } + + pub fn as_i64(&self) -> Option { + match self { + Value::Int(i) => Some(*i), + _ => None, + } + } + + pub fn as_str(&self) -> Option<&str> { + match self { + Value::Text(s) => Some(s.as_str()), + _ => None, + } + } + + pub fn as_bool(&self) -> Option { + match self { + Value::Bool(b) => Some(*b), + _ => None, + } + } + + pub fn as_bytes(&self) -> Option<&[u8]> { + match self { + Value::Bytes(b) => Some(b.as_slice()), + _ => None, + } + } + + pub fn as_array(&self) -> Option<&[Value]> { + match self { + Value::Array(a) => Some(a.as_slice()), + _ => None, + } + } + + pub fn as_link(&self) -> Option<&Cid> { + match self { + Value::Link(c) => Some(c), + _ => None, + } + } + + pub fn is_null(&self) -> bool { + matches!(self, Value::Null) + } +} + +// -- encoding --------------------------------------------------------------- + +/// Write a CBOR head: the 3-bit major type plus the argument, using the +/// shortest encoding that fits. +/// +/// Public because `car.rs` builds its length-prefixed sections around the same +/// primitive and there is no reason to have two copies. +pub fn write_head(out: &mut Vec, major: u8, n: u64) { + let m = (major & 0x07) << 5; + if n < 24 { + out.push(m | n as u8); + } else if n <= u8::MAX as u64 { + out.push(m | 24); + out.push(n as u8); + } else if n <= u16::MAX as u64 { + out.push(m | 25); + out.extend_from_slice(&(n as u16).to_be_bytes()); + } else if n <= u32::MAX as u64 { + out.push(m | 26); + out.extend_from_slice(&(n as u32).to_be_bytes()); + } else { + out.push(m | 27); + out.extend_from_slice(&n.to_be_bytes()); + } +} + +/// Append a CBOR text string (major type 3). +pub fn write_text(out: &mut Vec, s: &str) { + write_head(out, 3, s.len() as u64); + out.extend_from_slice(s.as_bytes()); +} + +/// Append a CBOR byte string (major type 2). +pub fn write_bytes(out: &mut Vec, b: &[u8]) { + write_head(out, 2, b.len() as u64); + out.extend_from_slice(b); +} + +/// Append a CID as a DAG-CBOR link: `tag(42) || bytes(0x00 || cid)`. +/// +/// The `0x00` is the multibase identity prefix. Binary CIDs inside DAG-CBOR +/// always carry it; the textual form (`bafy…`) never does. +pub fn write_link(out: &mut Vec, cid: &Cid) { + write_head(out, 6, CID_TAG); + let raw = cid.to_bytes(); + let mut prefixed = Vec::with_capacity(raw.len() + 1); + prefixed.push(0x00); + prefixed.extend_from_slice(&raw); + write_bytes(out, &prefixed); +} + +/// DAG-CBOR canonical map-key order: shorter keys sort first; equal-length +/// keys sort bytewise ascending. +/// +/// This is RFC 7049's "canonical CBOR" rule, which DAG-CBOR inherited and +/// which `@ipld/dag-cbor` implements. (RFC 8949 later switched the *core* +/// deterministic profile to plain bytewise ordering, but DAG-CBOR did not +/// follow — using 8949's rule here would produce frames that differ from the +/// reference implementation's for keys like `"op"` vs `"t"`.) +fn canonical_key_cmp(a: &str, b: &str) -> std::cmp::Ordering { + a.len() + .cmp(&b.len()) + .then_with(|| a.as_bytes().cmp(b.as_bytes())) +} + +/// Encode a value into `out`. +pub fn encode_into(out: &mut Vec, value: &Value) { + match value { + Value::Null => out.push(0xF6), + Value::Bool(false) => out.push(0xF4), + Value::Bool(true) => out.push(0xF5), + Value::Int(i) => { + if *i >= 0 { + write_head(out, 0, *i as u64); + } else { + // CBOR major type 1 stores -1-n, so n = -(i+1). Computed on + // i64 via `i128` to stay correct at `i64::MIN`, where + // `-(i + 1)` would overflow. + let n = (-((*i as i128) + 1)) as u64; + write_head(out, 1, n); + } + } + Value::Bytes(b) => write_bytes(out, b), + Value::Text(s) => write_text(out, s), + Value::Array(items) => { + write_head(out, 4, items.len() as u64); + for item in items { + encode_into(out, item); + } + } + Value::Map(entries) => { + let mut sorted: Vec<&(String, Value)> = entries.iter().collect(); + sorted.sort_by(|a, b| canonical_key_cmp(&a.0, &b.0)); + write_head(out, 5, sorted.len() as u64); + for (k, v) in sorted { + write_text(out, k); + encode_into(out, v); + } + } + Value::Link(cid) => write_link(out, cid), + } +} + +/// Encode a value to a fresh `Vec`. +/// +/// `#[allow(dead_code)]`: the server itself always appends into an existing +/// buffer via [`encode_into`] (a frame is two values in one allocation), so +/// this convenience wrapper is exercised only by the tests that assert on +/// exact byte sequences. It is kept because the decoder half needs a matching +/// encoder half to be testable at all. +#[allow(dead_code)] +pub fn encode(value: &Value) -> Vec { + let mut out = Vec::new(); + encode_into(&mut out, value); + out +} + +// -- decoding --------------------------------------------------------------- + +// -- Why a decoder lives in a server that only encodes ---------------------- +// +// The PDS never parses a firehose frame in production — it writes them. The +// decoder exists so the frame *contract* can be tested from the outside: a +// unit test that only checks "the encoder produced these bytes" locks in +// whatever the encoder happens to do, including its bugs. Decoding the bytes +// back and asserting on the structure is what actually verifies that a tag-42 +// link is a link and not a byte string, that the two frame halves are +// separable, and that a `null` `since` is `null` rather than absent. +// +// The integration test uses it for the same reason from the client side, and +// `crates/appview` builds its consumer against the same shape. + +/// Decode exactly one value, requiring it to consume the whole input. +#[allow(dead_code)] +pub fn decode(bytes: &[u8]) -> Result { + let (v, used) = decode_one(bytes)?; + if used != bytes.len() { + bail!( + "trailing bytes after DAG-CBOR value: consumed {used} of {}", + bytes.len() + ); + } + Ok(v) +} + +/// Decode one value from the front of `bytes`, returning it along with the +/// number of bytes consumed. +/// +/// This is the entry point the firehose frame reader needs: a frame is two +/// concatenated DAG-CBOR values (header then body) with no length prefix +/// between them, so the only way to find the body is to decode the header and +/// see where it ended. +#[allow(dead_code)] +pub fn decode_one(bytes: &[u8]) -> Result<(Value, usize)> { + decode_at(bytes, 0).map(|(v, end)| (v, end)) +} + +#[allow(dead_code)] +fn decode_at(bytes: &[u8], offset: usize) -> Result<(Value, usize)> { + let (major, arg, mut p) = read_head(bytes, offset)?; + match major { + 0 => { + let i = i64::try_from(arg).map_err(|_| anyhow!("CBOR uint {arg} exceeds i64"))?; + Ok((Value::Int(i), p)) + } + 1 => { + let v = -(arg as i128) - 1; + let i = i64::try_from(v).map_err(|_| anyhow!("CBOR nint {v} exceeds i64"))?; + Ok((Value::Int(i), p)) + } + 2 => { + let end = p + arg as usize; + if end > bytes.len() { + bail!("CBOR byte string exceeds input"); + } + Ok((Value::Bytes(bytes[p..end].to_vec()), end)) + } + 3 => { + let end = p + arg as usize; + if end > bytes.len() { + bail!("CBOR text string exceeds input"); + } + let s = std::str::from_utf8(&bytes[p..end]) + .map_err(|e| anyhow!("invalid UTF-8 in CBOR text: {e}"))?; + Ok((Value::Text(s.to_string()), end)) + } + 4 => { + let mut items = Vec::with_capacity(arg.min(1024) as usize); + for _ in 0..arg { + let (v, next) = decode_at(bytes, p)?; + items.push(v); + p = next; + } + Ok((Value::Array(items), p)) + } + 5 => { + let mut entries = Vec::with_capacity(arg.min(1024) as usize); + for _ in 0..arg { + let (k, next) = decode_at(bytes, p)?; + p = next; + let key = match k { + Value::Text(s) => s, + other => bail!("DAG-CBOR map keys must be text, got {other:?}"), + }; + let (v, next) = decode_at(bytes, p)?; + p = next; + entries.push((key, v)); + } + Ok((Value::Map(entries), p)) + } + 6 => { + if arg != CID_TAG { + bail!("unsupported CBOR tag {arg}; DAG-CBOR allows only 42"); + } + let (inner, end) = decode_at(bytes, p)?; + let raw = match inner { + Value::Bytes(b) => b, + other => bail!("CBOR tag 42 must wrap a byte string, got {other:?}"), + }; + let stripped = raw + .split_first() + .filter(|(first, _)| **first == 0x00) + .map(|(_, rest)| rest) + .ok_or_else(|| { + anyhow!("CID link missing the 0x00 multibase identity prefix") + })?; + let cid = Cid::read_bytes(stripped) + .map_err(|e| anyhow!("invalid CID inside tag 42: {e}"))?; + Ok((Value::Link(cid), end)) + } + 7 => match arg { + 20 => Ok((Value::Bool(false), p)), + 21 => Ok((Value::Bool(true), p)), + 22 => Ok((Value::Null, p)), + // 23 is `undefined`, 25/26/27 are floats. DAG-CBOR forbids + // `undefined`; floats are out of scope (see the module header). + other => bail!("unsupported CBOR simple/float value {other}"), + }, + other => bail!("unsupported CBOR major type {other}"), + } +} + +/// Read a CBOR head at `offset`, returning `(major, argument, next_offset)`. +/// +/// Public so `car.rs` can drive its own (deliberately non-conformant, see +/// there) header parser off the same primitive instead of keeping a second +/// copy. +pub fn read_head(bytes: &[u8], offset: usize) -> Result<(u8, u64, usize)> { + let first = *bytes + .get(offset) + .ok_or_else(|| anyhow!("CBOR read past end of input at {offset}"))?; + let major = first >> 5; + let low = first & 0x1f; + let (arg, extra) = match low { + 0..=23 => (low as u64, 0usize), + 24 => (read_uint(bytes, offset + 1, 1)?, 1), + 25 => (read_uint(bytes, offset + 1, 2)?, 2), + 26 => (read_uint(bytes, offset + 1, 4)?, 4), + 27 => (read_uint(bytes, offset + 1, 8)?, 8), + // 28..=30 are reserved; 31 is the indefinite-length marker, which + // DAG-CBOR forbids outright. + other => bail!("invalid or indefinite CBOR head 0x{other:02x}"), + }; + Ok((major, arg, offset + 1 + extra)) +} + +fn read_uint(bytes: &[u8], offset: usize, width: usize) -> Result { + if offset + width > bytes.len() { + bail!("truncated CBOR integer of width {width}"); + } + let mut n: u64 = 0; + for b in &bytes[offset..offset + width] { + n = (n << 8) | *b as u64; + } + Ok(n) +} + +#[cfg(test)] +mod tests { + use super::*; + use at_crypto::cid::cid_for_cbor; + + fn round_trip(v: Value) { + let bytes = encode(&v); + let back = decode(&bytes).expect("decode"); + // Maps come back in canonical (encoded) order, so compare the + // re-encoding rather than the structure for map-bearing values. + assert_eq!(encode(&back), bytes, "re-encode must be stable"); + } + + #[test] + fn scalars_round_trip() { + for v in [ + Value::Null, + Value::Bool(true), + Value::Bool(false), + Value::Int(0), + Value::Int(23), + Value::Int(24), + Value::Int(255), + Value::Int(256), + Value::Int(65_535), + Value::Int(65_536), + Value::Int(i64::MAX), + Value::Int(-1), + Value::Int(-24), + Value::Int(-1000), + Value::Int(i64::MIN), + Value::Text("hello".into()), + Value::Text(String::new()), + Value::Bytes(vec![1, 2, 3]), + Value::Bytes(Vec::new()), + ] { + let bytes = encode(&v); + assert_eq!(decode(&bytes).unwrap(), v, "round trip of {v:?}"); + } + } + + #[test] + fn int_heads_are_shortest_form() { + assert_eq!(encode(&Value::Int(1)), vec![0x01]); + assert_eq!(encode(&Value::Int(24)), vec![0x18, 24]); + assert_eq!(encode(&Value::Int(-1)), vec![0x20]); + assert_eq!(encode(&Value::Int(-25)), vec![0x38, 24]); + } + + #[test] + fn cid_link_carries_tag_42_and_identity_prefix() { + let cid = cid_for_cbor(b"a block").unwrap(); + let bytes = encode(&Value::Link(cid)); + // 0xD8 0x2A == tag(42) in two-byte form. + assert_eq!(&bytes[0..2], &[0xD8, 0x2A]); + // Then a byte string whose first content byte is the 0x00 prefix. + let (_major, len, p) = read_head(&bytes, 2).unwrap(); + assert_eq!(len as usize, cid.to_bytes().len() + 1); + assert_eq!(bytes[p], 0x00); + assert_eq!(decode(&bytes).unwrap(), Value::Link(cid)); + } + + #[test] + fn link_without_identity_prefix_is_rejected() { + let cid = cid_for_cbor(b"x").unwrap(); + let mut bytes = Vec::new(); + write_head(&mut bytes, 6, CID_TAG); + // Deliberately omit the leading 0x00. + write_bytes(&mut bytes, &cid.to_bytes()); + let e = decode(&bytes).unwrap_err().to_string(); + assert!(e.contains("identity prefix"), "got: {e}"); + } + + #[test] + fn map_keys_are_sorted_length_first() { + // The firehose header is exactly this shape, and the reference + // implementation emits `t` before `op` because it is shorter. + let v = Value::map([("op", Value::Int(1)), ("t", Value::text("#commit"))]); + let bytes = encode(&v); + assert_eq!(bytes[0], 0xA2, "map(2)"); + assert_eq!(bytes[1], 0x61, "text(1)"); + assert_eq!(bytes[2], b't'); + // …and `op` follows after the "#commit" value. + let decoded = decode(&bytes).unwrap(); + match &decoded { + Value::Map(entries) => { + assert_eq!(entries[0].0, "t"); + assert_eq!(entries[1].0, "op"); + } + other => panic!("expected map, got {other:?}"), + } + } + + #[test] + fn nested_structures_round_trip() { + let cid = cid_for_cbor(b"nested").unwrap(); + round_trip(Value::map([ + ("seq", Value::Int(42)), + ("commit", Value::Link(cid)), + ( + "ops", + Value::Array(vec![Value::map([ + ("action", Value::text("create")), + ("path", Value::text("app.twi.post/3l")), + ("cid", Value::Link(cid)), + ])]), + ), + ("blobs", Value::Array(vec![])), + ("since", Value::Null), + ])); + } + + #[test] + fn decode_one_stops_at_the_value_boundary() { + // Two concatenated values — exactly how a firehose frame is laid out. + let mut buf = encode(&Value::map([("op", Value::Int(1))])); + let header_len = buf.len(); + buf.extend_from_slice(&encode(&Value::map([("seq", Value::Int(7))]))); + let (header, used) = decode_one(&buf).unwrap(); + assert_eq!(used, header_len); + assert_eq!(header.get("op").and_then(Value::as_i64), Some(1)); + let body = decode(&buf[used..]).unwrap(); + assert_eq!(body.get("seq").and_then(Value::as_i64), Some(7)); + } + + #[test] + fn indefinite_length_is_rejected() { + // 0x9F == array(*) — legal CBOR, illegal DAG-CBOR. + assert!(decode(&[0x9F, 0x01, 0xFF]).is_err()); + } + + #[test] + fn trailing_bytes_are_rejected() { + let mut bytes = encode(&Value::Int(1)); + bytes.push(0x01); + assert!(decode(&bytes).is_err()); + } +} diff --git a/crates/pds-server/src/firehose.rs b/crates/pds-server/src/firehose.rs new file mode 100644 index 0000000..d4f03dd --- /dev/null +++ b/crates/pds-server/src/firehose.rs @@ -0,0 +1,946 @@ +//! `com.atproto.sync.subscribeRepos` — the event log, the frame codec and the +//! in-process broadcast channel. +//! +//! # What this is for +//! +//! Before this module the PDS emitted no firehose. A locally created record +//! reached the AppView through exactly one channel: the best-effort HTTP push +//! in [`crate::appview_push`], a detached `tokio::spawn` whose failure branch +//! logs "jetstream will replay". For records that only exist on this PDS +//! there *is* no Jetstream to replay them, so a lost push meant the post was +//! never indexed and nothing would ever notice. The firehose replaces that +//! hope with a durable, ordered log: every repo write appends one row inside +//! the same transaction as the commit, and a consumer can ask for everything +//! after a cursor at any later time. +//! +//! # Frame format +//! +//! One WebSocket **binary** message is two DAG-CBOR values written back to +//! back with nothing between them — a header, then a body. There is no length +//! prefix; the reader decodes the header and continues the body at the offset +//! where the header ended (that is what [`crate::dag_cbor::decode_one`] is +//! for). +//! +//! ```text +//! regular: {"op": 1, "t": "#commit"} {"seq": …, "repo": …, …} +//! {"op": 1, "t": "#info"} {"name": …, "message": …} +//! error: {"op": -1} {"error": "", "message": ""} +//! ``` +//! +//! The `#commit` body carries: +//! +//! | field | type | +//! |-----------|---------------------------------------------------------| +//! | `seq` | int — the cursor value for this event | +//! | `rebase` | bool — always `false` (we never rebase a repo) | +//! | `tooBig` | bool — always `false` (see the size note below) | +//! | `repo` | text — the DID | +//! | `commit` | **CID link (tag 42)** — the new commit block | +//! | `rev` | text — the new commit's revision | +//! | `since` | text or null — the previous commit's revision | +//! | `blocks` | byte string — a CAR v1 file, commit block as root | +//! | `ops` | array of `{action, path, cid}`; `cid` is a link or null | +//! | `blobs` | array — always empty (blob refs live inside the record) | +//! | `time` | text — RFC 3339, when the event was appended | +//! +//! Map keys are emitted in DAG-CBOR canonical order (shortest first, then +//! bytewise), so the bytes match what a reference atproto implementation +//! would produce for the same logical value. +//! +//! # Deliberate deviation from the atproto spec +//! +//! **The frame envelope is conformant. The blocks inside `blocks` are not.** +//! +//! This repository encodes CIDs *inside* commit blocks as CBOR text strings +//! rather than as DAG-CBOR links with tag 42 — see `at_repo::commit` and +//! `at_crypto::signing::sign_dag_cbor`. That convention predates this module +//! and is load-bearing: the block bytes determine every CID in the system, +//! including the `did:plc:` derivation, so changing it would re-address every +//! repo in the database. It is explicitly out of scope here. +//! +//! The consequence, stated plainly: a foreign atproto consumer can connect, +//! parse every frame, read `seq` / `repo` / `rev` / `ops`, and follow the +//! stream. It will then fail when it tries to *validate* the payload — the +//! CAR in `blocks` parses fine and the block CIDs hash correctly over their +//! own bytes, but decoding a commit block as DAG-CBOR yields `"prev"` and +//! `"data"` as strings where the spec demands links, and MST traversal +//! against a stock implementation will not work. The AppView in this +//! workspace reads the frames with the same conventions this crate writes +//! them, which is why it works there. +//! +//! `tooBig` is therefore always `false`: it exists so a producer can say "the +//! diff was too large, go fetch the repo yourself", and we never make that +//! call — every commit here is one record change, and its CAR is small. +//! +//! # Lagging consumers +//! +//! The broadcast channel is bounded ([`FIREHOSE_CHANNEL_CAPACITY`]). A +//! consumer that reads slower than the PDS writes will eventually be lapped, +//! and `tokio::sync::broadcast` reports that as `RecvError::Lagged(n)`. +//! +//! The write path must never wait on a reader, so the channel cannot be made +//! blocking. When a reader lags we send it an `#info` / `OutdatedCursor` +//! frame and **fall back to the database replay** from the last sequence it +//! actually received, then resume live. We do not disconnect it: the events +//! are durable in `firehose_events`, so the fallback is lossless, whereas +//! dropping the socket would force the client to reconnect and perform +//! exactly the same replay after two extra round trips. The only thing a +//! disconnect would buy is protection against a client that lags forever, and +//! that is handled separately by capping consecutive recoveries +//! ([`MAX_LAG_RECOVERIES`]) before closing with an error frame. +//! +//! # Retention +//! +//! Nothing prunes `firehose_events`. It grows by one row per repo write, each +//! carrying the CAR of that commit's new blocks. That is unbounded, and this +//! deployment has no retention job — an operator who wants one has to add it. +//! Pruning is safe by design: the cursor handshake compares the requested +//! cursor against the oldest surviving row and answers a too-old cursor with +//! `#info` / `OutdatedCursor` followed by a replay from the oldest row that +//! still exists, rather than pretending the gap is not there. + +use crate::car::CarWriter; +use crate::dag_cbor::{encode_into, Value}; +use anyhow::{anyhow, Result}; +use at_crypto::cid::cid_from_multihash_bytes; +use chrono::{DateTime, SecondsFormat, Utc}; +use cid::Cid; +use serde_json::json; +use std::sync::Arc; +use tokio::sync::broadcast; + +/// How many events the live broadcast channel buffers per subscriber before +/// the slowest one starts reporting `Lagged`. +/// +/// 1024 is chosen so a consumer can stall for the length of a garbage +/// collection or a slow network write without falling back to the database, +/// while the memory ceiling stays bounded: the channel holds `Arc`s, so the +/// cost is one CAR blob per queued event, shared across all subscribers. +pub const FIREHOSE_CHANNEL_CAPACITY: usize = 1024; + +/// How many times in a row a single connection may be rescued from a lag +/// before we give up and close it. A client that cannot keep up even with a +/// database replay in between is not going to start; at that point the honest +/// answer is an error frame rather than an endless catch-up loop that burns +/// queries on its behalf. +pub const MAX_LAG_RECOVERIES: u32 = 5; + +/// Rows returned per replay query. Bounded so a client reconnecting with +/// `cursor=0` after a long uptime streams the backlog in chunks instead of +/// materialising the whole table (and every CAR in it) in memory at once. +pub const REPLAY_PAGE_SIZE: i64 = 200; + +/// The advisory-lock key that serialises `firehose_events` INSERTs. Any +/// constant works as long as every writer uses the same one; this is +/// `"fhose"` read as ASCII, which makes it recognisable in `pg_locks`. +/// +/// What it costs: the lock is global, not per-repo, so the tail of every +/// repo write — INSERT plus COMMIT — is serialised across all accounts. +/// That is deliberate (a per-repo lock would order each repo's events but +/// not the shared `seq` a consumer paginates on), and it bounds write +/// throughput to how fast Postgres can commit one small INSERT at a time. +/// If that ever becomes the ceiling, the fix is a different sequence +/// design — a per-repo cursor, or handing out `seq` from a single writer +/// task — not a weaker lock: a gap in `seq` is silent data loss for every +/// consumer replaying from a cursor. +pub const FIREHOSE_ADVISORY_LOCK_KEY: i64 = 0x66_68_6f_73_65; + +// -- ops -------------------------------------------------------------------- + +/// What a single repo operation did to one record. +/// +/// `Update` exists separately from `Create` because a consumer that keeps a +/// materialised view needs to know whether to insert or replace; the MST +/// itself does not distinguish them, so the write path resolves it by looking +/// the key up before writing. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RepoOpAction { + Create, + Update, + Delete, +} + +impl RepoOpAction { + pub fn as_str(self) -> &'static str { + match self { + RepoOpAction::Create => "create", + RepoOpAction::Update => "update", + RepoOpAction::Delete => "delete", + } + } + + pub fn parse(s: &str) -> Result { + match s { + "create" => Ok(RepoOpAction::Create), + "update" => Ok(RepoOpAction::Update), + "delete" => Ok(RepoOpAction::Delete), + other => Err(anyhow!("unknown repo op action `{other}`")), + } + } +} + +/// One entry of a commit frame's `ops` array. +/// +/// `path` is `"/"` — the MST key, not an `at://` URI. That +/// is what the wire format specifies, and it is also what the MST is actually +/// keyed by, so there is no reassembly step on either side. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RepoOp { + pub action: RepoOpAction, + pub path: String, + /// The record value's CID for `create` / `update`; `None` for `delete`, + /// where there is no resulting value to point at. + pub cid: Option, +} + +impl RepoOp { + pub fn create(collection: &str, rkey: &str, cid: Cid) -> Self { + Self { + action: RepoOpAction::Create, + path: format!("{collection}/{rkey}"), + cid: Some(cid), + } + } + + pub fn update(collection: &str, rkey: &str, cid: Cid) -> Self { + Self { + action: RepoOpAction::Update, + path: format!("{collection}/{rkey}"), + cid: Some(cid), + } + } + + pub fn delete(collection: &str, rkey: &str) -> Self { + Self { + action: RepoOpAction::Delete, + path: format!("{collection}/{rkey}"), + cid: None, + } + } + + /// Pick `create` or `update` from whether the key already existed. + pub fn put(collection: &str, rkey: &str, cid: Cid, existed: bool) -> Self { + if existed { + Self::update(collection, rkey, cid) + } else { + Self::create(collection, rkey, cid) + } + } + + /// JSON shape stored in `firehose_events.ops`. The CID is a string here + /// (JSONB has no link type); it becomes a tag-42 link again on the wire. + pub fn to_json(&self) -> serde_json::Value { + json!({ + "action": self.action.as_str(), + "path": self.path, + "cid": self.cid.map(|c| c.to_string()), + }) + } + + pub fn from_json(v: &serde_json::Value) -> Result { + let action = RepoOpAction::parse( + v.get("action") + .and_then(|a| a.as_str()) + .ok_or_else(|| anyhow!("op missing `action`"))?, + )?; + let path = v + .get("path") + .and_then(|p| p.as_str()) + .ok_or_else(|| anyhow!("op missing `path`"))? + .to_string(); + let cid = match v.get("cid") { + None | Some(serde_json::Value::Null) => None, + Some(serde_json::Value::String(s)) => { + Some(s.parse::().map_err(|e| anyhow!("op cid: {e}"))?) + } + Some(other) => return Err(anyhow!("op `cid` must be a string or null, got {other}")), + }; + Ok(Self { action, path, cid }) + } + + fn to_dag_cbor(&self) -> Value { + Value::map([ + ("action", Value::text(self.action.as_str())), + ("path", Value::text(&self.path)), + ( + "cid", + match self.cid { + Some(c) => Value::Link(c), + None => Value::Null, + }, + ), + ]) + } +} + +// -- events ----------------------------------------------------------------- + +/// One row of `firehose_events`, ready to be framed. +/// +/// Both the live path (built at write time) and the replay path (read back +/// from Postgres) produce this exact struct, which is what makes a replayed +/// frame byte-identical to the live one — including `time`, which comes from +/// the stored `created_at` rather than from the clock at send time. +#[derive(Debug, Clone)] +pub struct FirehoseEvent { + pub seq: i64, + pub did: String, + pub rev: String, + pub since: Option, + pub commit: Cid, + /// CAR v1 file: the commit block as root, plus the blocks this commit + /// newly created. + pub blocks: Vec, + pub ops: Vec, + pub time: DateTime, +} + +impl FirehoseEvent { + /// The `#commit` body as a DAG-CBOR value. + pub fn to_body(&self) -> Value { + Value::map([ + ("seq", Value::Int(self.seq)), + ("rebase", Value::Bool(false)), + ("tooBig", Value::Bool(false)), + ("repo", Value::text(&self.did)), + ("commit", Value::Link(self.commit)), + ("rev", Value::text(&self.rev)), + ( + "since", + match &self.since { + Some(s) => Value::text(s), + None => Value::Null, + }, + ), + ("blocks", Value::Bytes(self.blocks.clone())), + ( + "ops", + Value::Array(self.ops.iter().map(RepoOp::to_dag_cbor).collect()), + ), + ("blobs", Value::Array(Vec::new())), + ( + "time", + Value::text(self.time.to_rfc3339_opts(SecondsFormat::Micros, true)), + ), + ]) + } + + /// The full binary WebSocket payload: `#commit` header then body. + pub fn to_frame(&self) -> Vec { + frame(&header_value("#commit"), &self.to_body()) + } +} + +/// Build the CAR that goes into a commit event's `blocks` field. +/// +/// Root is the commit block; the remaining entries are the blocks this commit +/// newly wrote (MST nodes and record values). Blocks that already existed in +/// the repo are deliberately left out — that is the whole point of a diff +/// stream, and a consumer that needs an ancestor block asks +/// `com.atproto.sync.getBlocks` for it. +pub fn build_blocks_car( + commit_cid: Cid, + commit_block: &[u8], + new_blocks: &[(Cid, Vec)], +) -> Vec { + let mut w = CarWriter::new(); + w.append(commit_cid, commit_block); + for (cid, data) in new_blocks { + w.append(*cid, data); + } + w.finish(&[commit_cid]) +} + +// -- frame encoding --------------------------------------------------------- + +/// `{"op": 1, "t": ""}` — the header of a regular frame. +fn header_value(t: &str) -> Value { + Value::map([("op", Value::Int(1)), ("t", Value::text(t))]) +} + +/// Concatenate a header and a body into one binary WebSocket payload. +fn frame(header: &Value, body: &Value) -> Vec { + let mut out = Vec::new(); + encode_into(&mut out, header); + encode_into(&mut out, body); + out +} + +/// An `#info` frame: `{"op":1,"t":"#info"}` + `{"name":…,"message":…}`. +/// +/// Informational, not fatal — the stream continues after it. We send it when +/// a cursor is older than anything we still have, and when a live subscriber +/// lagged and is being put back on the database replay. +pub fn encode_info_frame(name: &str, message: &str) -> Vec { + frame( + &header_value("#info"), + &Value::map([("name", Value::text(name)), ("message", Value::text(message))]), + ) +} + +/// An error frame: `{"op":-1}` + `{"error":…,"message":…}`. +/// +/// Terminal — the server closes the socket right after sending it. +pub fn encode_error_frame(error: &str, message: &str) -> Vec { + frame( + &Value::map([("op", Value::Int(-1))]), + &Value::map([ + ("error", Value::text(error)), + ("message", Value::text(message)), + ]), + ) +} + +// -- broadcast -------------------------------------------------------------- + +/// The in-process fan-out from the write path to connected subscribers. +/// +/// Cloneable and cheap: it is a `broadcast::Sender` plus nothing. Events are +/// wrapped in an `Arc` so a burst of subscribers does not multiply the CAR +/// blobs. +#[derive(Clone)] +pub struct Firehose { + tx: broadcast::Sender>, +} + +impl Default for Firehose { + fn default() -> Self { + Self::new() + } +} + +impl Firehose { + pub fn new() -> Self { + let (tx, _rx) = broadcast::channel(FIREHOSE_CHANNEL_CAPACITY); + Self { tx } + } + + /// Publish an event to every live subscriber. + /// + /// Never blocks and never fails in a way the caller must handle: with no + /// subscribers the send returns `Err`, which is the normal state of a PDS + /// nobody is watching. The event is already durable in Postgres by the + /// time we get here, so a dropped broadcast costs a consumer nothing + /// beyond having to replay by cursor. + pub fn publish(&self, event: FirehoseEvent) { + let _ = self.tx.send(Arc::new(event)); + } + + pub fn subscribe(&self) -> broadcast::Receiver> { + self.tx.subscribe() + } + + /// Number of live subscribers. Used for logging / the healthz surface. + #[allow(dead_code)] + pub fn subscriber_count(&self) -> usize { + self.tx.receiver_count() + } +} + +// -- persistence ------------------------------------------------------------ + +/// Append one event inside an open transaction and return it with its +/// assigned `seq` and `created_at`. +/// +/// The advisory lock taken first is what makes the sequence usable as a +/// cursor: without it two concurrent writers can be assigned seq 5 and 6 and +/// commit in the other order, so a reader polling in between sees 6, records +/// it as its cursor, and never learns about 5. Holding +/// `pg_advisory_xact_lock` from just before the INSERT until COMMIT forces +/// commit order to match seq order. It is taken *after* the per-repo +/// `SELECT … FOR UPDATE` in [`crate::routes::helpers::apply_repo_write`], and +/// every writer takes the two in that same order, so the pair cannot deadlock. +pub async fn insert_event_in_tx( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + did: &str, + rev: &str, + since: Option<&str>, + commit: Cid, + blocks: Vec, + ops: Vec, +) -> Result { + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(FIREHOSE_ADVISORY_LOCK_KEY) + .execute(&mut **tx) + .await?; + + let ops_json = serde_json::Value::Array(ops.iter().map(RepoOp::to_json).collect()); + + let (seq, created_at): (i64, DateTime) = sqlx::query_as( + r#"INSERT INTO firehose_events (did, rev, since, commit_cid, blocks, ops) + VALUES ($1, $2, $3, $4, $5, $6) + RETURNING seq, created_at"#, + ) + .bind(did) + .bind(rev) + .bind(since) + .bind(commit.to_bytes()) + .bind(&blocks) + .bind(&ops_json) + .fetch_one(&mut **tx) + .await?; + + Ok(FirehoseEvent { + seq, + did: did.to_string(), + rev: rev.to_string(), + since: since.map(|s| s.to_string()), + commit, + blocks, + ops, + time: created_at, + }) +} + +/// The `(min_seq, max_seq)` currently in the table, or `None` when it is +/// empty. Used by the cursor handshake to tell "from the future" apart from +/// "too old to still have". +pub async fn seq_bounds(db: &sqlx::PgPool) -> Result, sqlx::Error> { + let row: (Option, Option) = + sqlx::query_as("SELECT MIN(seq), MAX(seq) FROM firehose_events") + .fetch_one(db) + .await?; + Ok(match row { + (Some(min), Some(max)) => Some((min, max)), + _ => None, + }) +} + +/// Read up to [`REPLAY_PAGE_SIZE`] events with `seq > after`, oldest first. +pub async fn load_events_after( + db: &sqlx::PgPool, + after: i64, + limit: i64, +) -> Result> { + let rows: Vec<( + i64, + String, + String, + Option, + Vec, + Vec, + serde_json::Value, + DateTime, + )> = sqlx::query_as( + r#"SELECT seq, did, rev, since, commit_cid, blocks, ops, created_at + FROM firehose_events + WHERE seq > $1 + ORDER BY seq ASC + LIMIT $2"#, + ) + .bind(after) + .bind(limit) + .fetch_all(db) + .await + .map_err(|e| anyhow!("firehose_events replay read: {e}"))?; + + rows.into_iter() + .map(|(seq, did, rev, since, commit_cid, blocks, ops, created_at)| { + let commit = cid_from_multihash_bytes(&commit_cid) + .map_err(|e| anyhow!("firehose_events.commit_cid at seq {seq}: {e}"))?; + let ops = ops + .as_array() + .ok_or_else(|| anyhow!("firehose_events.ops at seq {seq} is not an array"))? + .iter() + .map(RepoOp::from_json) + .collect::>>()?; + Ok(FirehoseEvent { + seq, + did, + rev, + since, + commit, + blocks, + ops, + time: created_at, + }) + }) + .collect() +} + +// -- cursor handshake ------------------------------------------------------- + +/// What the connection handler should do with the cursor the client sent. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum CursorPlan { + /// No cursor: send live events only, nothing from the log. + LiveOnly, + /// Replay everything after `from`, then go live. + Replay { from: i64 }, + /// The requested cursor predates the oldest row we still have. Warn with + /// an `#info` frame, then replay from `from` (the oldest surviving row + /// minus one) so the client at least gets everything that does exist. + OutdatedCursor { from: i64, message: String }, + /// The cursor names an event that has not happened. This is a client bug + /// (or a cursor from a different server's log), and continuing would + /// silently strand it — so it is a terminal error frame. + FutureCursor { message: String }, +} + +/// Decide what to do with `cursor` given the log's current `(min, max)`. +/// +/// Split out from the socket handler so the boundary conditions are testable +/// without a database or a WebSocket. +/// +/// Semantics of the cursor: it is the seq of the last event the client +/// *already has*, so a replay yields `seq > cursor`. `cursor = 0` therefore +/// means "everything", and `cursor = max` means "nothing yet, just go live" — +/// which is a valid, empty replay rather than a future cursor. +pub fn plan_cursor(cursor: Option, bounds: Option<(i64, i64)>) -> CursorPlan { + let cursor = match cursor { + None => return CursorPlan::LiveOnly, + Some(c) => c, + }; + if cursor < 0 { + return CursorPlan::FutureCursor { + message: format!("cursor {cursor} is negative"), + }; + } + let (min, max) = match bounds { + // An empty log accepts only cursor 0 ("give me everything, there is + // nothing"). Anything else refers to an event we never had. + None => { + return if cursor == 0 { + CursorPlan::Replay { from: 0 } + } else { + CursorPlan::FutureCursor { + message: format!("cursor {cursor} is ahead of an empty log"), + } + }; + } + Some(b) => b, + }; + if cursor > max { + return CursorPlan::FutureCursor { + message: format!("cursor {cursor} is ahead of the latest event {max}"), + }; + } + // `cursor >= min - 1` means the next event the client wants (cursor + 1) + // is still on disk. Below that, rows have been pruned and the client has + // a hole it can never fill. + if cursor < min - 1 { + return CursorPlan::OutdatedCursor { + from: min - 1, + message: format!( + "cursor {cursor} predates the oldest retained event {min}; \ + resuming from {min} — events {}..{} are gone", + cursor + 1, + min - 1 + ), + }; + } + CursorPlan::Replay { from: cursor } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::dag_cbor::{decode, decode_one}; + use at_crypto::cid::cid_for_cbor; + + fn sample_event() -> FirehoseEvent { + let commit = cid_for_cbor(b"commit block").unwrap(); + let value = cid_for_cbor(b"record value").unwrap(); + FirehoseEvent { + seq: 7, + did: "did:plc:alice".into(), + rev: "3lxxxxxxxx2".into(), + since: Some("3lxxxxxxxx1".into()), + commit, + blocks: build_blocks_car( + commit, + b"commit block", + &[(value, b"record value".to_vec())], + ), + ops: vec![RepoOp::create("app.twi.post", "3lrkey", value)], + time: DateTime::parse_from_rfc3339("2026-09-10T12:00:00.123456Z") + .unwrap() + .with_timezone(&Utc), + } + } + + // -- frame encoding ---------------------------------------------------- + + #[test] + fn commit_frame_header_then_body() { + let ev = sample_event(); + let bytes = ev.to_frame(); + let (header, used) = decode_one(&bytes).unwrap(); + assert_eq!(header.get("op").and_then(Value::as_i64), Some(1)); + assert_eq!(header.get("t").and_then(Value::as_str), Some("#commit")); + // Everything after the header is exactly one more value — no padding, + // no length prefix. + let body = decode(&bytes[used..]).unwrap(); + assert_eq!(body.get("seq").and_then(Value::as_i64), Some(7)); + assert_eq!( + body.get("repo").and_then(Value::as_str), + Some("did:plc:alice") + ); + assert_eq!(body.get("rev").and_then(Value::as_str), Some("3lxxxxxxxx2")); + assert_eq!( + body.get("since").and_then(Value::as_str), + Some("3lxxxxxxxx1") + ); + assert_eq!(body.get("rebase").and_then(Value::as_bool), Some(false)); + assert_eq!(body.get("tooBig").and_then(Value::as_bool), Some(false)); + assert!(body.get("blobs").unwrap().as_array().unwrap().is_empty()); + assert_eq!( + body.get("time").and_then(Value::as_str), + Some("2026-09-10T12:00:00.123456Z") + ); + } + + /// The header's exact bytes, locked in. The AppView builds its reader + /// against this, and canonical key ordering means `t` precedes `op` — + /// which is easy to get wrong and produces a frame a strict DAG-CBOR + /// decoder rejects. + #[test] + fn commit_header_has_the_exact_expected_bytes() { + let bytes = sample_event().to_frame(); + assert_eq!( + &bytes[..15], + &[ + 0xA2, // map(2) + 0x61, b't', // text(1) "t" + 0x67, b'#', b'c', b'o', b'm', b'm', b'i', b't', // text(7) "#commit" + 0x62, b'o', b'p', // text(2) "op" + 0x01, // 1 + ] + ); + } + + #[test] + fn info_and_error_headers_have_the_exact_expected_bytes() { + let info = encode_info_frame("OutdatedCursor", "x"); + assert_eq!( + &info[..12], + &[0xA2, 0x61, b't', 0x65, b'#', b'i', b'n', b'f', b'o', 0x62, b'o', b'p'] + ); + assert_eq!(info[12], 0x01); + // Error: map(1) { "op": -1 }. -1 is major type 1 with argument 0. + let err = encode_error_frame("FutureCursor", "x"); + assert_eq!(&err[..5], &[0xA1, 0x62, b'o', b'p', 0x20]); + } + + #[test] + fn commit_cid_is_a_tag_42_link_not_a_string() { + // The whole point of the hand-rolled encoder: `ciborium` would have + // written this as a string or a newtype struct. + let ev = sample_event(); + let bytes = ev.to_frame(); + let (_h, used) = decode_one(&bytes).unwrap(); + let body = decode(&bytes[used..]).unwrap(); + assert_eq!(body.get("commit").and_then(Value::as_link), Some(&ev.commit)); + // Locate the tag bytes directly, to prove it is not the decoder being + // generous: 0xD8 0x2A is tag(42). + assert!( + bytes.windows(2).any(|w| w == [0xD8, 0x2A]), + "frame must contain a tag-42 head" + ); + } + + #[test] + fn ops_encode_action_path_and_link() { + let ev = sample_event(); + let bytes = ev.to_frame(); + let (_h, used) = decode_one(&bytes).unwrap(); + let body = decode(&bytes[used..]).unwrap(); + let ops = body.get("ops").unwrap().as_array().unwrap(); + assert_eq!(ops.len(), 1); + assert_eq!(ops[0].get("action").and_then(Value::as_str), Some("create")); + assert_eq!( + ops[0].get("path").and_then(Value::as_str), + Some("app.twi.post/3lrkey") + ); + assert!(ops[0].get("cid").unwrap().as_link().is_some()); + } + + #[test] + fn delete_op_has_a_null_cid() { + let mut ev = sample_event(); + ev.ops = vec![RepoOp::delete("app.bsky.feed.like", "3lrkey")]; + let bytes = ev.to_frame(); + let (_h, used) = decode_one(&bytes).unwrap(); + let body = decode(&bytes[used..]).unwrap(); + let ops = body.get("ops").unwrap().as_array().unwrap(); + assert_eq!(ops[0].get("action").and_then(Value::as_str), Some("delete")); + assert!(ops[0].get("cid").unwrap().is_null()); + } + + #[test] + fn first_commit_has_a_null_since() { + let mut ev = sample_event(); + ev.since = None; + let bytes = ev.to_frame(); + let (_h, used) = decode_one(&bytes).unwrap(); + let body = decode(&bytes[used..]).unwrap(); + assert!(body.get("since").unwrap().is_null()); + } + + #[test] + fn blocks_is_a_parsable_car_rooted_at_the_commit() { + let ev = sample_event(); + let bytes = ev.to_frame(); + let (_h, used) = decode_one(&bytes).unwrap(); + let body = decode(&bytes[used..]).unwrap(); + let car = body.get("blocks").and_then(Value::as_bytes).unwrap(); + let (header, blocks) = crate::car::parse(car).unwrap(); + assert_eq!(header.version, 1); + assert_eq!(header.roots, vec![ev.commit]); + assert_eq!(blocks.len(), 2, "commit block + one new record block"); + assert_eq!(blocks[0].cid, ev.commit); + assert_eq!(blocks[0].data, b"commit block"); + } + + #[test] + fn info_frame_round_trips() { + let bytes = encode_info_frame("OutdatedCursor", "resuming from 12"); + let (header, used) = decode_one(&bytes).unwrap(); + assert_eq!(header.get("op").and_then(Value::as_i64), Some(1)); + assert_eq!(header.get("t").and_then(Value::as_str), Some("#info")); + let body = decode(&bytes[used..]).unwrap(); + assert_eq!( + body.get("name").and_then(Value::as_str), + Some("OutdatedCursor") + ); + assert_eq!( + body.get("message").and_then(Value::as_str), + Some("resuming from 12") + ); + } + + #[test] + fn error_frame_uses_op_minus_one_and_carries_no_t() { + let bytes = encode_error_frame("FutureCursor", "cursor 99 is ahead"); + let (header, used) = decode_one(&bytes).unwrap(); + assert_eq!(header.get("op").and_then(Value::as_i64), Some(-1)); + assert!( + header.get("t").is_none(), + "an error header carries op only" + ); + let body = decode(&bytes[used..]).unwrap(); + assert_eq!( + body.get("error").and_then(Value::as_str), + Some("FutureCursor") + ); + assert_eq!( + body.get("message").and_then(Value::as_str), + Some("cursor 99 is ahead") + ); + } + + // -- ops derivation ---------------------------------------------------- + + #[test] + fn put_picks_create_or_update_from_prior_existence() { + let cid = cid_for_cbor(b"v").unwrap(); + assert_eq!( + RepoOp::put("c", "r", cid, false).action, + RepoOpAction::Create + ); + assert_eq!( + RepoOp::put("c", "r", cid, true).action, + RepoOpAction::Update + ); + } + + #[test] + fn op_path_is_collection_slash_rkey() { + let cid = cid_for_cbor(b"v").unwrap(); + assert_eq!( + RepoOp::create("app.bsky.feed.repost", "3lk", cid).path, + "app.bsky.feed.repost/3lk" + ); + } + + #[test] + fn op_json_round_trips_through_the_jsonb_shape() { + let cid = cid_for_cbor(b"v").unwrap(); + for op in [ + RepoOp::create("app.twi.post", "a", cid), + RepoOp::update("app.bsky.actor.profile", "self", cid), + RepoOp::delete("app.bsky.graph.follow", "b"), + ] { + let back = RepoOp::from_json(&op.to_json()).unwrap(); + assert_eq!(back, op); + } + } + + #[test] + fn op_json_rejects_an_unknown_action() { + let v = json!({"action": "rebase", "path": "a/b", "cid": null}); + assert!(RepoOp::from_json(&v).is_err()); + } + + // -- cursor edge cases ------------------------------------------------- + + #[test] + fn no_cursor_is_live_only() { + assert_eq!(plan_cursor(None, Some((1, 10))), CursorPlan::LiveOnly); + assert_eq!(plan_cursor(None, None), CursorPlan::LiveOnly); + } + + #[test] + fn cursor_zero_replays_everything() { + assert_eq!( + plan_cursor(Some(0), Some((1, 10))), + CursorPlan::Replay { from: 0 } + ); + } + + #[test] + fn cursor_at_the_head_is_an_empty_replay_not_an_error() { + // The client is fully caught up. Replaying `seq > 10` yields nothing + // and it goes straight live — that must not be a FutureCursor. + assert_eq!( + plan_cursor(Some(10), Some((1, 10))), + CursorPlan::Replay { from: 10 } + ); + } + + #[test] + fn cursor_past_the_head_is_a_future_cursor() { + assert!(matches!( + plan_cursor(Some(11), Some((1, 10))), + CursorPlan::FutureCursor { .. } + )); + } + + #[test] + fn negative_cursor_is_a_future_cursor() { + assert!(matches!( + plan_cursor(Some(-1), Some((1, 10))), + CursorPlan::FutureCursor { .. } + )); + } + + #[test] + fn empty_log_accepts_zero_and_refuses_anything_else() { + assert_eq!(plan_cursor(Some(0), None), CursorPlan::Replay { from: 0 }); + assert!(matches!( + plan_cursor(Some(1), None), + CursorPlan::FutureCursor { .. } + )); + } + + #[test] + fn cursor_exactly_one_below_the_oldest_row_is_still_exact() { + // min = 5 means seq 5 is the oldest surviving event. A client whose + // cursor is 4 wants 5 next — nothing is missing. + assert_eq!( + plan_cursor(Some(4), Some((5, 10))), + CursorPlan::Replay { from: 4 } + ); + } + + #[test] + fn cursor_below_the_pruned_window_is_outdated() { + match plan_cursor(Some(2), Some((5, 10))) { + CursorPlan::OutdatedCursor { from, message } => { + assert_eq!(from, 4, "resume so the next delivered event is 5"); + assert!(message.contains('5'), "message should name the gap: {message}"); + } + other => panic!("expected OutdatedCursor, got {other:?}"), + } + } +} diff --git a/crates/pds-server/src/main.rs b/crates/pds-server/src/main.rs index 58512dc..ae361b2 100644 --- a/crates/pds-server/src/main.rs +++ b/crates/pds-server/src/main.rs @@ -1,5 +1,7 @@ mod appview_push; mod car; +mod dag_cbor; +mod firehose; mod jwt_issuer; mod keys; mod password; @@ -146,6 +148,13 @@ pub fn router(state: AppState) -> Router { "/xrpc/com.atproto.sync.getBlob", get(routes::blob::get_blob), ) + // The firehose. A WebSocket upgrade arrives as a plain GET, so this + // is a normal `get` route whose handler happens to return an + // upgrade response. + .route( + "/xrpc/com.atproto.sync.subscribeRepos", + get(routes::subscribe_repos::subscribe_repos), + ) .route( "/blob/:cid", get(routes::blob::get_blob_by_cid), diff --git a/crates/pds-server/src/routes/feed.rs b/crates/pds-server/src/routes/feed.rs index f5b961b..fd21dfd 100644 --- a/crates/pds-server/src/routes/feed.rs +++ b/crates/pds-server/src/routes/feed.rs @@ -17,7 +17,10 @@ //! removed from the MST, a new commit is signed, the AppView is //! told to drop the row, and we return the new commit CID + rev. -use crate::routes::helpers::{apply_repo_write, err, lookup_handle, to_sqlx_error, RepoWriteOutcome}; +use crate::firehose::RepoOp; +use crate::routes::helpers::{ + apply_repo_write, err, lookup_handle, to_sqlx_error, RepoWriteOutcome, RepoWriteResult, +}; use at_repo::blockstore::Blockstore; use crate::routes::types::ErrorBody; use crate::state::AppState; @@ -189,11 +192,16 @@ fn build_like_record(req: &CreateLikeReq) -> Result( state: &AppState, did: &str, f: F, -) -> Result)> +) -> Result)> where F: for<'b> FnOnce( &'b mut at_repo::repo::Repo, @@ -201,7 +209,7 @@ where Box> + Send + 'b>, >, { - apply_repo_write(state, did, f).await.map(|o| o.commit) + apply_repo_write(state, did, f).await } // -- handlers --------------------------------------------------------------- @@ -275,7 +283,7 @@ pub async fn create_like( let push_rkey = rkey.clone(); let push_handle_str: Option = lookup_handle(&state, &did).await; - let commit = apply_and_commit(&state, &did, move |repo| { + let write = apply_and_commit(&state, &did, move |repo| { let value_cid = value_cid; let rkey = rkey; let record_buf = record_buf; @@ -297,16 +305,21 @@ pub async fn create_like( commit, head_cid_bytes, head_commit_bytes, + // Always a create: the rkey is a freshly minted TID, so it + // cannot collide with an existing entry. + ops: vec![RepoOp::create(LIKE_COLLECTION, &rkey, value_cid)], }) }) }) .await?; + let commit = write.commit; info!( collection = LIKE_COLLECTION, rkey = %push_rkey, cid = %value_cid, commit = %commit.cid, + seq = write.event.seq, "like created" ); @@ -366,10 +379,21 @@ pub async fn delete_record( // `Repo::delete_record` is idempotent at the MST level (returns // an unchanged tree if the key isn't present), so we always // sign a new commit — the spec says 200 on a no-op delete. - let commit = apply_and_commit(&state, &did, move |repo| { + let write = apply_and_commit(&state, &did, move |repo| { let collection = collection; let rkey = rkey; Box::pin(async move { + // Report the op only when there was something to remove. + // `delete_record` is idempotent — deleting a missing rkey signs + // an unchanged tree — and announcing a delete for a record that + // never existed would make a consumer drop a row it may legitimately + // hold under a different rkey, or (worse) log a phantom deletion + // on every retry of a duplicate unlike. + let existed = repo + .get_record(&collection, &rkey) + .await + .map_err(to_sqlx_error)? + .is_some(); repo.delete_record(&collection, &rkey) .await .map_err(to_sqlx_error)?; @@ -380,15 +404,22 @@ pub async fn delete_record( commit, head_cid_bytes, head_commit_bytes, + ops: if existed { + vec![RepoOp::delete(&collection, &rkey)] + } else { + Vec::new() + }, }) }) }) .await?; + let commit = write.commit; info!( collection = %push_collection, rkey = %push_rkey, commit = %commit.cid, + seq = write.event.seq, "record deleted" ); diff --git a/crates/pds-server/src/routes/helpers.rs b/crates/pds-server/src/routes/helpers.rs index 6641776..102e1ef 100644 --- a/crates/pds-server/src/routes/helpers.rs +++ b/crates/pds-server/src/routes/helpers.rs @@ -12,6 +12,7 @@ //! writers for the same DID can't trample each other's MST updates //! (Phase 5b review C1). +use crate::firehose::{self, FirehoseEvent, RepoOp}; use crate::routes::types::ErrorBody; use crate::state::AppState; use at_crypto::cid::cid_from_multihash_bytes; @@ -24,6 +25,7 @@ use cid::Cid; use k256::ecdsa::SigningKey; use k256::SecretKey; use sqlx::Postgres; +use std::collections::HashSet; use std::sync::Arc; /// Load every block belonging to `did` from the `repo_blocks` table into a @@ -184,14 +186,31 @@ pub fn to_sqlx_error(e: anyhow::Error) -> sqlx::Error { // transaction commits or rolls back, so concurrent writers serialise // behind the holder rather than racing on the head_commit column. -/// Result of a successful repo write: the new signed commit, the CID -/// pointing at the freshly-written head block, and the new revision -/// string. Callers use the commit for AppView ingest pushes. +/// What the closure handed to [`apply_repo_write`] returns: the new signed +/// commit, the CID pointing at the freshly-written head block, and the record +/// operations the closure performed. +/// +/// `ops` is not derivable from the commit — the MST stores the resulting +/// tree, not the edit that produced it, and it cannot tell a create from an +/// update at all. Only the closure knows what it did, so it says so, and the +/// firehose event is built from that. Every write path must fill this in +/// truthfully: an empty `ops` produces a commit frame that tells the AppView +/// "something changed, guess what". #[derive(Debug, Clone)] pub struct RepoWriteOutcome { pub commit: at_repo::commit::Commit, pub head_cid_bytes: Vec, pub head_commit_bytes: Vec, + pub ops: Vec, +} + +/// What [`apply_repo_write`] returns to the route handler: the commit (used +/// for the response body and the AppView push) plus the firehose event that +/// was appended in the same transaction and has already been broadcast. +#[derive(Debug, Clone)] +pub struct RepoWriteResult { + pub commit: at_repo::commit::Commit, + pub event: FirehoseEvent, } /// Apply a write to the user's repo under a row-level lock on the @@ -218,11 +237,27 @@ pub struct RepoWriteOutcome { /// write the row), but the transaction stays open until after the /// `UPDATE`. If the closure or `UPDATE` fails, the transaction rolls /// back and no head pointer or block row changes are visible. +/// +/// ## The firehose event rides in the same transaction +/// +/// Between step 7 and the COMMIT we append one row to `firehose_events` +/// (see [`crate::firehose`]). It has to be the *same* transaction, not a +/// follow-up write: if the event were appended afterwards, a crash in the +/// window between the two would leave a repo whose head has moved but whose +/// event log never mentions it — and since a consumer's cursor only ever +/// moves forward, that commit would be invisible to every subscriber +/// permanently. Sharing the transaction makes "the head moved" and "an event +/// exists for it" one atomic fact. Conversely, a rollback discards both, so +/// no subscriber ever sees an event for a commit that did not happen. +/// +/// The broadcast to live subscribers happens *after* `COMMIT`, for the same +/// reason in reverse: a subscriber must never receive an event that a +/// rollback then erases. pub async fn apply_repo_write( state: &AppState, did: &str, f: F, -) -> Result)> +) -> Result)> where F: for<'b> FnOnce( &'b mut Repo, @@ -240,8 +275,12 @@ where // 2. Take the row-level write lock. Postgres parks competing // transactions here until we COMMIT/ROLLBACK. - let head_row: Option<(Vec, Vec, Option>)> = sqlx::query_as( - "SELECT head_cid, head_commit, prev_commit + // `rev` comes along because it is the *previous* commit's revision, + // which the firehose frame publishes as `since` — a consumer uses it to + // notice that it skipped an intermediate commit. It has to be read here, + // under the lock, before the UPDATE below overwrites it. + let head_row: Option<(Vec, Vec, Option>, String)> = sqlx::query_as( + "SELECT head_cid, head_commit, prev_commit, rev FROM repos WHERE did = $1 FOR UPDATE", @@ -257,8 +296,8 @@ where ) })?; - let (head_cid_blob, head_commit_blob) = match head_row { - Some(r) => (r.0, r.1), + let (head_cid_blob, head_commit_blob, prev_rev) = match head_row { + Some(r) => (r.0, r.1, r.3), None => { return Err(( StatusCode::NOT_FOUND, @@ -354,6 +393,26 @@ where // to run additional queries under the row lock, we'd extend // this helper to also hand out a `&mut PgConnection` (which // doesn't have the lifetime headache of `&mut Transaction`). + // Snapshot the CIDs the repo already had *before* the closure runs. + // Diffing against this afterwards is what tells us which blocks are + // new in this commit — the firehose CAR carries only those, because a + // diff stream that re-sent the whole repo on every post would be + // useless. The snapshot is taken here, after the head block re-seed + // above, so the existing head commit does not look new. + let blocks_before: HashSet = blockstore + .list() + .await + .map_err(|e| { + err( + StatusCode::INTERNAL_SERVER_ERROR, + "InternalServerError", + format!("blockstore list: {e:#}"), + ) + })? + .into_iter() + .map(|(cid, _)| cid) + .collect(); + let outcome: RepoWriteOutcome = f(&mut repo).await.map_err(|e| { err( StatusCode::INTERNAL_SERVER_ERROR, @@ -409,6 +468,51 @@ where ) })?; + // 8. Append the firehose event. Same transaction as everything above — + // see the "rides in the same transaction" note on this function. + // + // The CAR carries the commit block as its root plus every block that + // was not in the repo when we started: the new MST nodes and the new + // record value. Blocks that already existed are omitted; a consumer + // that needs an ancestor asks `com.atproto.sync.getBlocks` for it. + let new_blocks: Vec<(Cid, Vec)> = all_blocks + .iter() + .filter(|(cid, _)| !blocks_before.contains(*cid) && **cid != outcome.commit.cid) + .map(|(cid, bytes)| (*cid, bytes.clone())) + .collect(); + let blocks_car = firehose::build_blocks_car( + outcome.commit.cid, + &outcome.head_commit_bytes, + &new_blocks, + ); + + // A repo whose head was the all-zero sentinel had no previous commit, so + // there is no previous revision to report — `since` is null rather than + // the `"0"` placeholder `createAccount` seeds the row with. + let since: Option<&str> = if is_zero_blob(&head_cid_blob) { + None + } else { + Some(prev_rev.as_str()) + }; + + let event = firehose::insert_event_in_tx( + &mut tx, + did, + &outcome.commit.rev, + since, + outcome.commit.cid, + blocks_car, + outcome.ops.clone(), + ) + .await + .map_err(|e| { + err( + StatusCode::INTERNAL_SERVER_ERROR, + "InternalServerError", + format!("firehose_events insert: {e}"), + ) + })?; + tx.commit().await.map_err(|e| { err( StatusCode::INTERNAL_SERVER_ERROR, @@ -417,7 +521,18 @@ where ) })?; - Ok(outcome) + // 9. Only now, with the commit durable, hand the event to live + // subscribers. Publishing is non-blocking and cannot fail in a way + // that matters: a slow subscriber is dealt with on its own side (see + // the lag policy in `crate::firehose`), and with no subscribers at all + // the send is a no-op. The row is on disk either way, so nothing is + // lost if this reaches nobody. + state.firehose.publish(event.clone()); + + Ok(RepoWriteResult { + commit: outcome.commit, + event, + }) } /// Persist every block in `blocks` into `repo_blocks` using the open diff --git a/crates/pds-server/src/routes/mod.rs b/crates/pds-server/src/routes/mod.rs index f76c011..3e28de8 100644 --- a/crates/pds-server/src/routes/mod.rs +++ b/crates/pds-server/src/routes/mod.rs @@ -5,5 +5,6 @@ pub mod helpers; pub mod identity; pub mod profile; pub mod repo; +pub mod subscribe_repos; pub mod sync; pub mod types; diff --git a/crates/pds-server/src/routes/profile.rs b/crates/pds-server/src/routes/profile.rs index 3531e9a..311c0bb 100644 --- a/crates/pds-server/src/routes/profile.rs +++ b/crates/pds-server/src/routes/profile.rs @@ -16,6 +16,7 @@ //! overwrite the corresponding fields. Best-effort push to the //! AppView follows so the `profiles` cache reflects the new avatar / //! display name / bio without waiting for the Jetstream replay. +use crate::firehose::RepoOp; use crate::jwt_issuer; use crate::routes::helpers::{ apply_repo_write, err, load_head_commit, load_signing_key, load_user_blockstore, @@ -98,6 +99,9 @@ pub async fn set_profile( // Fetch the existing record, if any. let existing = read_profile_record(&state, &did).await?; + // Remembered before `existing` is consumed by the merge — the firehose op + // needs to know whether this is the first profile write for the account. + let existing_present = existing.is_some(); // For any blob CIDs in the request, look up the real // `mime_type` / `size` from the `blobs` table — and verify @@ -152,6 +156,12 @@ pub async fn set_profile( }; let next_for_block = next.clone(); + // `existing` was read before the merge above: a profile record that was + // already there makes this an `update` on the firehose, a first-ever + // `setProfile` a `create`. The rkey is the fixed `self`, so this is the + // one write path where updates are the common case rather than the + // exception. + let profile_existed = existing_present; let outcome = apply_repo_write(&state, &did, move |repo| { let value_cid = value_cid; let next_for_block = next_for_block; @@ -175,6 +185,12 @@ pub async fn set_profile( commit, head_cid_bytes, head_commit_bytes, + ops: vec![RepoOp::put( + "app.bsky.actor.profile", + "self", + value_cid, + profile_existed, + )], }) }) }) @@ -183,6 +199,7 @@ pub async fn set_profile( info!( did = %did, cid = %outcome.commit.cid, + seq = outcome.event.seq, "profile record created" ); diff --git a/crates/pds-server/src/routes/repo.rs b/crates/pds-server/src/routes/repo.rs index 8579177..9428078 100644 --- a/crates/pds-server/src/routes/repo.rs +++ b/crates/pds-server/src/routes/repo.rs @@ -1,3 +1,4 @@ +use crate::firehose::RepoOp; use crate::routes::helpers::{ apply_repo_write, err, lookup_handle, to_sqlx_error, RepoWriteOutcome, }; @@ -110,6 +111,17 @@ pub async fn create_record( let record_buf = record_buf; let collection = collection; Box::pin(async move { + // Ask the MST whether the key is already there *before* writing. + // The firehose distinguishes `create` from `update` and the tree + // itself cannot: after `put_record` both look identical. A + // caller-supplied `rkey` (rather than the generated TID) is the + // case that actually hits this — an overwrite of an existing + // record must not be announced as a create. + let existed = repo + .get_record(&collection, &rkey) + .await + .map_err(to_sqlx_error)? + .is_some(); // Repo assumes the value block is already in the // blockstore — that's the caller's responsibility. repo.blockstore @@ -127,23 +139,34 @@ pub async fn create_record( commit, head_cid_bytes, head_commit_bytes, + ops: vec![RepoOp::put(&collection, &rkey, value_cid, existed)], }) }) }) .await?; let uri = format!("at://{did}/{push_coll}/{push_rkey}"); + let seq = outcome.event.seq; let commit = outcome.commit; - info!(uri = %uri, cid = %value_cid, commit = %commit.cid, "record created"); + info!( + uri = %uri, + cid = %value_cid, + commit = %commit.cid, + seq, + "record created" + ); // 10. Best-effort push to the AppView's `/internal/ingest-commit`. // We send the full record value (not just the CID) because the // AppView's indexer reads `embed` and `reply` off it. // // **Spawned** (not awaited) so a transient AppView outage never - // blocks the user's write response. If the push fails, the - // global Jetstream feed will eventually replay the commit to - // the AppView. + // blocks the user's write response. Losing the push is no longer + // terminal: the same commit was appended to `firehose_events` in + // the write transaction above (see `seq` in the log line), so an + // AppView that reconnects to `com.atproto.sync.subscribeRepos` + // with its cursor picks it up. The push is now purely a latency + // optimisation, not the only delivery path. tokio::spawn(async move { if let Err(e) = push_handle .push_create( diff --git a/crates/pds-server/src/routes/subscribe_repos.rs b/crates/pds-server/src/routes/subscribe_repos.rs new file mode 100644 index 0000000..75ef5a7 --- /dev/null +++ b/crates/pds-server/src/routes/subscribe_repos.rs @@ -0,0 +1,301 @@ +//! `GET /xrpc/com.atproto.sync.subscribeRepos` — the firehose WebSocket. +//! +//! The frame format, the deviation from the atproto spec, the lag policy and +//! the retention story all live in the module header of [`crate::firehose`]; +//! this file is only the socket. +//! +//! ## The handshake, and why it is ordered the way it is +//! +//! ```text +//! 1. subscribe to the live broadcast <-- BEFORE any DB read +//! 2. read (min, max) from firehose_events +//! 3. decide what the cursor means <-- firehose::plan_cursor +//! 4. drain the DB replay, remembering the highest seq sent +//! 5. forward live events with seq > that high-water mark +//! ``` +//! +//! Step 1 has to come first. If we read the database and *then* subscribed, +//! an event committed in between would be in neither: too late for the replay +//! query, too early for the receiver. Subscribing first turns that race into +//! a duplicate instead of a gap — the event is both replayed from the table +//! and sitting in the channel — and a duplicate is something we can filter, +//! which is what the high-water mark in step 5 does. +//! +//! The filter is exact rather than approximate because the write path +//! serialises `firehose_events` inserts under an advisory lock (see +//! [`crate::firehose::insert_event_in_tx`]): if the replay query saw seq `N`, +//! then every seq below `N` is already committed and was also seen. So +//! "everything the replay covered" is precisely "seq <= N", and every event +//! that arrives on the channel afterwards has seq > N. No gap, no duplicate, +//! at the handover. +//! +//! Without a cursor there is no replay at all and the high-water mark stays +//! at zero: a `broadcast::Receiver` only ever yields messages sent after it +//! was created, so "live only" needs no filtering. + +use crate::firehose::{ + self, CursorPlan, FirehoseEvent, MAX_LAG_RECOVERIES, REPLAY_PAGE_SIZE, +}; +use crate::state::AppState; +use axum::extract::ws::{CloseFrame, Message, WebSocket, WebSocketUpgrade}; +use axum::extract::{Query, State}; +use axum::response::Response; +use serde::Deserialize; +use tokio::sync::broadcast::error::RecvError; +use tracing::{debug, info, warn}; + +use futures::{SinkExt, StreamExt}; + +#[derive(Debug, Deserialize)] +pub struct SubscribeQuery { + /// The seq of the last event the client already has. Everything with a + /// larger seq is replayed before the live stream starts. Absent means + /// "live only". + pub cursor: Option, +} + +/// The upgrade handler. Everything interesting happens in [`run`]. +pub async fn subscribe_repos( + State(state): State, + Query(q): Query, + ws: WebSocketUpgrade, +) -> Response { + ws.on_upgrade(move |socket| run(socket, state, q.cursor)) +} + +/// Drive one subscriber for the life of its connection. +async fn run(socket: WebSocket, state: AppState, cursor: Option) { + // 1. Subscribe first — see the ordering note in the module header. + let mut rx = state.firehose.subscribe(); + + let (mut sink, mut stream) = socket.split(); + + // A firehose subscriber sends nothing after the upgrade, but we still + // have to read the socket: that is the only way a Close frame (or a + // client that vanishes without sending data) is noticed while we are + // parked waiting for an event that may not come for hours. The reader + // task does nothing but detect the end of the connection and say so. + let (dead_tx, mut dead_rx) = tokio::sync::oneshot::channel::<()>(); + tokio::spawn(async move { + while let Some(msg) = stream.next().await { + match msg { + Ok(Message::Close(_)) | Err(_) => break, + // Ping/Pong are handled by axum itself; anything else a + // client sends on this endpoint is meaningless and ignored + // rather than treated as an error. + Ok(_) => continue, + } + } + let _ = dead_tx.send(()); + }); + + // 2./3. Work out what the cursor asks for. + let bounds = match firehose::seq_bounds(&state.db).await { + Ok(b) => b, + Err(e) => { + warn!(error = %e, "subscribeRepos: firehose_events bounds read failed"); + let _ = send_error(&mut sink, "InternalServerError", "event log unavailable").await; + return; + } + }; + + let mut replay_from = match firehose::plan_cursor(cursor, bounds) { + CursorPlan::LiveOnly => None, + CursorPlan::Replay { from } => Some(from), + CursorPlan::OutdatedCursor { from, message } => { + // Not fatal: the client keeps its connection and gets everything + // we still have. It is told about the hole so it can decide + // whether to backfill some other way. + info!(cursor = ?cursor, %message, "subscribeRepos: outdated cursor"); + if sink + .send(Message::Binary(firehose::encode_info_frame( + "OutdatedCursor", + &message, + ))) + .await + .is_err() + { + return; + } + Some(from) + } + CursorPlan::FutureCursor { message } => { + // Fatal. Continuing would leave the client waiting for events + // that will be numbered below its cursor and therefore filtered + // out forever — silence is the worst possible answer here. + info!(cursor = ?cursor, %message, "subscribeRepos: future cursor"); + let _ = send_error(&mut sink, "FutureCursor", &message).await; + return; + } + }; + + info!( + cursor = ?cursor, + replay_from = ?replay_from, + subscribers = state.firehose.subscriber_count(), + "subscribeRepos: client connected" + ); + + // 4. Drain the replay. `high_water` ends up as the last seq the client + // has been given, which is exactly the boundary the live filter needs. + let mut high_water: i64 = 0; + if let Some(from) = replay_from.take() { + match replay(&mut sink, &state, from, &mut dead_rx).await { + Ok(last) => high_water = last, + // The socket died mid-replay, or the log became unreadable. + // Either way there is nothing left to do for this connection. + Err(()) => return, + } + } + + // 5. Live. + let mut lag_recoveries: u32 = 0; + loop { + tokio::select! { + _ = &mut dead_rx => { + debug!("subscribeRepos: client closed"); + return; + } + recv = rx.recv() => match recv { + Ok(event) => { + if event.seq <= high_water { + // Already delivered by the replay. This is the + // duplicate the subscribe-first ordering trades the + // gap for. + continue; + } + if !send_event(&mut sink, &event).await { + return; + } + high_water = event.seq; + } + Err(RecvError::Lagged(skipped)) => { + lag_recoveries += 1; + warn!( + skipped, + attempt = lag_recoveries, + high_water, + "subscribeRepos: subscriber lagged; falling back to the database replay" + ); + if lag_recoveries > MAX_LAG_RECOVERIES { + // See the lag policy in `crate::firehose`: we rescue a + // slow client, repeatedly, but not forever. + let _ = send_error( + &mut sink, + "ConsumerTooSlow", + "consumer fell behind repeatedly; reconnect with a cursor", + ) + .await; + return; + } + let message = format!( + "consumer lagged by {skipped} events; resuming from seq {high_water} \ + via the durable log" + ); + if sink + .send(Message::Binary(firehose::encode_info_frame( + "OutdatedCursor", + &message, + ))) + .await + .is_err() + { + return; + } + // Nothing is lost: every event is in `firehose_events`, + // so re-reading from the high-water mark is the same + // stream the channel dropped. The advisory-lock ordering + // guarantees the new high-water mark is again an exact + // boundary for the live filter. + match replay(&mut sink, &state, high_water, &mut dead_rx).await { + Ok(last) => high_water = last.max(high_water), + Err(()) => return, + } + } + Err(RecvError::Closed) => { + // Only happens at process shutdown, when the AppState + // (and with it the sender) is dropped. + debug!("subscribeRepos: broadcast channel closed"); + return; + } + } + } + } +} + +/// Stream every event with `seq > from` out of the database, page by page. +/// +/// Returns the highest seq actually sent (or `from` when there was nothing to +/// send), or `Err(())` when the connection or the database gave out — in +/// which case the caller should drop the connection. +/// +/// Paging matters: a client reconnecting with `cursor=0` after a long uptime +/// would otherwise pull the whole table, CARs and all, into memory at once. +/// It also gives the loop a natural place to notice a client that closed the +/// socket halfway through a large backfill. +async fn replay( + sink: &mut futures::stream::SplitSink, + state: &AppState, + from: i64, + dead_rx: &mut tokio::sync::oneshot::Receiver<()>, +) -> Result { + let mut cursor = from; + loop { + if dead_rx.try_recv().is_ok() { + return Err(()); + } + let page = match firehose::load_events_after(&state.db, cursor, REPLAY_PAGE_SIZE).await { + Ok(p) => p, + Err(e) => { + warn!(error = %format!("{e:#}"), "subscribeRepos: replay read failed"); + let _ = send_error(sink, "InternalServerError", "event log read failed").await; + return Err(()); + } + }; + if page.is_empty() { + return Ok(cursor); + } + for event in &page { + if !send_event(sink, event).await { + return Err(()); + } + cursor = event.seq; + } + // A short page means we reached the end of the log. + if (page.len() as i64) < REPLAY_PAGE_SIZE { + return Ok(cursor); + } + } +} + +/// Send one `#commit` frame. Returns `false` when the socket is gone. +async fn send_event( + sink: &mut futures::stream::SplitSink, + event: &FirehoseEvent, +) -> bool { + sink.send(Message::Binary(event.to_frame())).await.is_ok() +} + +/// Send a terminal error frame and close the socket. +/// +/// The close is explicit (rather than just dropping the sink) so a client +/// distinguishes "the server said no" from "the connection broke". +async fn send_error( + sink: &mut futures::stream::SplitSink, + error: &str, + message: &str, +) -> bool { + if sink + .send(Message::Binary(firehose::encode_error_frame(error, message))) + .await + .is_err() + { + return false; + } + sink.send(Message::Close(Some(CloseFrame { + code: axum::extract::ws::close_code::NORMAL, + reason: error.to_string().into(), + }))) + .await + .is_ok() +} diff --git a/crates/pds-server/src/state.rs b/crates/pds-server/src/state.rs index 596b4be..2439eba 100644 --- a/crates/pds-server/src/state.rs +++ b/crates/pds-server/src/state.rs @@ -1,4 +1,5 @@ use crate::appview_push::AppViewPushClient; +use crate::firehose::Firehose; use at_blob::S3BlobStore; use at_identity::plc::PlcClient; use at_lexicon::{Lex, LexRegistry}; @@ -16,6 +17,14 @@ pub struct AppState { pub blockstore: Arc, pub plc: PlcClient, pub appview: AppViewPushClient, + /// Live fan-out for `com.atproto.sync.subscribeRepos`. + /// + /// Lives on the shared state rather than in the route module because the + /// *write* paths publish into it — `routes::helpers::apply_repo_write` + /// hands every committed event over here — while the WebSocket handler + /// only subscribes. Cloning `AppState` clones the sender, which is the + /// intended way to reach it from a handler. + pub firehose: Firehose, } impl AppState { @@ -26,9 +35,9 @@ impl AppState { Lex::from_json(include_str!("../../../lexicons/app/twi/post.json")).unwrap(), ); // AT-Protocol standard collections: only the records the user - // might legitimately create server-side (feed.like + feed.repost). - // The full atproto collection library is out of scope — for - // anything else, callers pass `validate: false` in the + // might legitimately create server-side (feed.like, feed.repost, + // graph.follow). The full atproto collection library is out of + // scope — for anything else, callers pass `validate: false` in the // createRecord body. lex.lexicons.insert( "app.bsky.feed.like".to_string(), @@ -38,6 +47,17 @@ impl AppState { "app.bsky.feed.repost".to_string(), Lex::from_json(include_str!("../../../lexicons/app/bsky/feed/repost.json")).unwrap(), ); + // Follow record. Its absence was a real outage: the desktop + // client creates follows through `createRecord`, which validates + // by default, so every follow came back + // `unknown lexicon: app.bsky.graph.follow` — the button could + // never have worked. `subject` is a bare DID string here, not a + // strongRef like like/repost use, matching what the client sends + // and what the AppView's `follow_subject_did` reads. + lex.lexicons.insert( + "app.bsky.graph.follow".to_string(), + Lex::from_json(include_str!("../../../lexicons/app/bsky/graph/follow.json")).unwrap(), + ); // Profile record — avatar/banner/display name/description. // Validates the createRecord body when the Tauri client calls // its setProfile command. Other fields stay optional so a @@ -63,6 +83,7 @@ impl AppState { blockstore: Arc::new(MemoryBlockstore::new()), plc: PlcClient::new(plc_url), appview, + firehose: Firehose::new(), } } } diff --git a/crates/pds-server/tests/firehose_integration.rs b/crates/pds-server/tests/firehose_integration.rs new file mode 100644 index 0000000..b2780a5 --- /dev/null +++ b/crates/pds-server/tests/firehose_integration.rs @@ -0,0 +1,677 @@ +//! Integration tests for `com.atproto.sync.subscribeRepos`. +//! +//! Same contract as the other integration suites in this crate: they talk to a +//! PDS listening on `127.0.0.1:2583` and **fail open** — if nothing answers +//! `/healthz`, the test prints a note and returns green rather than failing a +//! developer's `cargo test` on a machine with no server running. Start the +//! server (`./target/debug/pds-server` with `.env` sourced) to actually +//! exercise them. +//! +//! ## Why the frames are decoded by hand here +//! +//! `pds-server` is a binary, so a test crate cannot import its `dag_cbor` +//! module — and that is a feature, not a limitation. These tests are the +//! *consumer* side of the wire contract, and a consumer that reuses the +//! producer's encoder proves nothing: it would happily agree with a frame +//! that no other implementation can read. The decoder below is written from +//! the spec (tag 42, identity prefix, length-first map keys) and knows +//! nothing about how the server produced the bytes. + +use serde_json::{json, Value}; +use std::collections::BTreeMap; +use std::time::Duration; + +use futures::{SinkExt, StreamExt}; +use tokio_tungstenite::tungstenite::Message; + +const PDS_URL: &str = "http://127.0.0.1:2583"; +const PDS_WS: &str = "ws://127.0.0.1:2583"; + +// -- harness --------------------------------------------------------------- + +fn http() -> reqwest::Client { + reqwest::Client::builder() + .timeout(Duration::from_secs(5)) + .build() + .unwrap() +} + +async fn wait_for_pds() -> bool { + let c = http(); + for _ in 0..20 { + if let Ok(r) = c.get(format!("{PDS_URL}/healthz")).send().await { + if r.status().is_success() { + return true; + } + } + tokio::time::sleep(Duration::from_millis(250)).await; + } + false +} + +async fn fresh_user(prefix: &str) -> (reqwest::Client, String, String) { + let c = http(); + let handle = format!( + "{}_{}.maarcadetweet.local", + prefix, + uuid::Uuid::new_v4().simple() + ); + let acc: Value = c + .post(format!("{PDS_URL}/xrpc/com.atproto.server.createAccount")) + .json(&json!({"handle": handle, "password": "hunter2hunter2"})) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + let did = acc["did"].as_str().expect("createAccount did").to_string(); + let jwt = acc["access_jwt"].as_str().expect("access_jwt").to_string(); + (c, did, jwt) +} + +async fn create_post(c: &reqwest::Client, did: &str, jwt: &str, text: &str) -> Value { + c.post(format!("{PDS_URL}/xrpc/com.atproto.repo.createRecord")) + .bearer_auth(jwt) + .json(&json!({ + "repo": did, + "collection": "app.twi.post", + "record": { "text": text, "createdAt": "2026-09-10T12:00:00Z" }, + })) + .send() + .await + .unwrap() + .json() + .await + .unwrap() +} + +type Socket = tokio_tungstenite::WebSocketStream< + tokio_tungstenite::MaybeTlsStream, +>; + +async fn subscribe(cursor: Option) -> Socket { + let url = match cursor { + Some(c) => format!("{PDS_WS}/xrpc/com.atproto.sync.subscribeRepos?cursor={c}"), + None => format!("{PDS_WS}/xrpc/com.atproto.sync.subscribeRepos"), + }; + let (socket, _resp) = tokio_tungstenite::connect_async(&url) + .await + .unwrap_or_else(|e| panic!("subscribeRepos connect to {url}: {e}")); + socket +} + +/// Read the next **binary** message, or `None` on timeout. +/// +/// Text messages would be a protocol violation on this endpoint and are +/// asserted against rather than skipped. +async fn next_frame(socket: &mut Socket) -> Option> { + let deadline = Duration::from_secs(10); + loop { + match tokio::time::timeout(deadline, socket.next()).await { + Err(_) => return None, + Ok(None) => return None, + Ok(Some(Ok(Message::Binary(b)))) => return Some(b), + Ok(Some(Ok(Message::Ping(_)))) | Ok(Some(Ok(Message::Pong(_)))) => continue, + Ok(Some(Ok(Message::Close(_)))) => return None, + Ok(Some(Ok(other))) => panic!("subscribeRepos sent a non-binary frame: {other:?}"), + Ok(Some(Err(e))) => panic!("subscribeRepos socket error: {e}"), + } + } +} + +/// Give the server a moment to finish `on_upgrade` and actually subscribe to +/// the broadcast channel before we trigger a write. +/// +/// The TCP handshake completing does not mean the handler has run. Without +/// this the test would occasionally write before the subscription exists and +/// then wait for a frame that was published to nobody. (The *cursor* replay +/// path is the real fix for that race in production; the live-only test is +/// deliberately testing the raceable path, so it waits.) +async fn settle() { + tokio::time::sleep(Duration::from_millis(300)).await; +} + +// -- an independent DAG-CBOR reader ---------------------------------------- + +#[derive(Debug, Clone, PartialEq)] +enum Cbor { + Null, + Bool(bool), + Int(i64), + Bytes(Vec), + Text(String), + Array(Vec), + Map(BTreeMap), + /// tag(42) + bytes(0x00 || cid) — the binary CID is kept raw and + /// re-encoded to a `bafy…` string on demand, so the test never depends on + /// the server's own CID formatting. + Link(Vec), +} + +impl Cbor { + fn get(&self, key: &str) -> &Cbor { + match self { + Cbor::Map(m) => m + .get(key) + .unwrap_or_else(|| panic!("missing key `{key}` in {self:?}")), + other => panic!("not a map: {other:?}"), + } + } + fn opt(&self, key: &str) -> Option<&Cbor> { + match self { + Cbor::Map(m) => m.get(key), + _ => None, + } + } + fn int(&self) -> i64 { + match self { + Cbor::Int(i) => *i, + other => panic!("not an int: {other:?}"), + } + } + fn text(&self) -> &str { + match self { + Cbor::Text(s) => s, + other => panic!("not text: {other:?}"), + } + } + fn bool(&self) -> bool { + match self { + Cbor::Bool(b) => *b, + other => panic!("not a bool: {other:?}"), + } + } + fn bytes(&self) -> &[u8] { + match self { + Cbor::Bytes(b) => b, + other => panic!("not bytes: {other:?}"), + } + } + fn array(&self) -> &[Cbor] { + match self { + Cbor::Array(a) => a, + other => panic!("not an array: {other:?}"), + } + } + /// The link's CID rendered as a base32 `bafy…` string, for comparison + /// against what the XRPC JSON responses return. + fn link_cid(&self) -> String { + match self { + Cbor::Link(raw) => cid::Cid::read_bytes(&raw[..]) + .expect("tag-42 payload must be a valid binary CID") + .to_string(), + other => panic!("not a link: {other:?}"), + } + } +} + +fn read_head(b: &[u8], p: usize) -> (u8, u64, usize) { + let first = b[p]; + let major = first >> 5; + let low = first & 0x1f; + let (arg, extra) = match low { + 0..=23 => (low as u64, 0usize), + 24 => (b[p + 1] as u64, 1), + 25 => (u16::from_be_bytes([b[p + 1], b[p + 2]]) as u64, 2), + 26 => ( + u32::from_be_bytes([b[p + 1], b[p + 2], b[p + 3], b[p + 4]]) as u64, + 4, + ), + 27 => { + let mut n = 0u64; + for i in 0..8 { + n = (n << 8) | b[p + 1 + i] as u64; + } + (n, 8) + } + other => panic!("indefinite or reserved CBOR head 0x{other:02x} — illegal in DAG-CBOR"), + }; + (major, arg, p + 1 + extra) +} + +fn decode_at(b: &[u8], p: usize) -> (Cbor, usize) { + let (major, arg, mut p) = read_head(b, p); + match major { + 0 => (Cbor::Int(arg as i64), p), + 1 => (Cbor::Int(-(arg as i64) - 1), p), + 2 => { + let end = p + arg as usize; + (Cbor::Bytes(b[p..end].to_vec()), end) + } + 3 => { + let end = p + arg as usize; + ( + Cbor::Text(std::str::from_utf8(&b[p..end]).unwrap().to_string()), + end, + ) + } + 4 => { + let mut items = Vec::new(); + for _ in 0..arg { + let (v, next) = decode_at(b, p); + items.push(v); + p = next; + } + (Cbor::Array(items), p) + } + 5 => { + let mut m = BTreeMap::new(); + let mut prev_key: Option = None; + for _ in 0..arg { + let (k, next) = decode_at(b, p); + p = next; + let key = k.text().to_string(); + // Canonical DAG-CBOR order: shorter keys first, then + // bytewise. Asserted here because a consumer written against + // a strict codec (cborg's `dag-cbor` decoder, for one) will + // reject a frame whose keys are out of order. + if let Some(prev) = &prev_key { + let ordered = (prev.len(), prev.as_bytes()) < (key.len(), key.as_bytes()); + assert!(ordered, "map keys out of canonical order: {prev:?} then {key:?}"); + } + prev_key = Some(key.clone()); + let (v, next) = decode_at(b, p); + p = next; + m.insert(key, v); + } + (Cbor::Map(m), p) + } + 6 => { + assert_eq!(arg, 42, "DAG-CBOR permits only tag 42"); + let (inner, next) = decode_at(b, p); + let raw = match inner { + Cbor::Bytes(v) => v, + other => panic!("tag 42 must wrap bytes, got {other:?}"), + }; + assert_eq!( + raw.first(), + Some(&0x00), + "a binary CID link must start with the 0x00 multibase identity prefix" + ); + (Cbor::Link(raw[1..].to_vec()), next) + } + 7 => match arg { + 20 => (Cbor::Bool(false), p), + 21 => (Cbor::Bool(true), p), + 22 => (Cbor::Null, p), + other => panic!("unsupported CBOR simple value {other}"), + }, + other => panic!("unsupported CBOR major type {other}"), + } +} + +/// Split one binary frame into its header and body values, and assert that +/// the two together consume the whole message — a frame with trailing bytes +/// would silently desynchronise a streaming consumer. +fn parse_frame(bytes: &[u8]) -> (Cbor, Cbor) { + let (header, after_header) = decode_at(bytes, 0); + let (body, end) = decode_at(bytes, after_header); + assert_eq!(end, bytes.len(), "frame must be exactly two CBOR values"); + (header, body) +} + +/// Minimal CAR v1 reader: returns the root CIDs and the block CIDs, both as +/// `bafy…` strings. +fn parse_car(bytes: &[u8]) -> (Vec, Vec) { + fn varint(b: &[u8], p: &mut usize) -> u64 { + let (mut v, mut shift) = (0u64, 0u32); + loop { + let byte = b[*p]; + *p += 1; + v |= ((byte & 0x7f) as u64) << shift; + if byte & 0x80 == 0 { + return v; + } + shift += 7; + } + } + let mut p = 0usize; + let header_len = varint(bytes, &mut p) as usize; + let header = &bytes[p..p + header_len]; + p += header_len; + + // The CAR header is a CBOR map; walk it with the same head reader. Note + // that this server's CAR header tags its roots *without* the 0x00 + // identity prefix (a documented deviation in `car.rs`), so the roots are + // read as plain tagged byte strings rather than through `decode_at`. + let mut hp = 0usize; + let (major, n, next) = read_head(header, hp); + assert_eq!(major, 5, "CAR header must be a map"); + hp = next; + let mut roots = Vec::new(); + for _ in 0..n { + let (m, len, next) = read_head(header, hp); + assert_eq!(m, 3); + hp = next; + let key = std::str::from_utf8(&header[hp..hp + len as usize]).unwrap().to_string(); + hp += len as usize; + if key == "version" { + let (m, v, next) = read_head(header, hp); + assert_eq!(m, 0); + assert_eq!(v, 1, "CAR must be v1"); + hp = next; + } else if key == "roots" { + let (m, count, next) = read_head(header, hp); + assert_eq!(m, 4); + hp = next; + for _ in 0..count { + let (m, tag, next) = read_head(header, hp); + assert_eq!((m, tag), (6, 42), "root must be a tag-42 link"); + hp = next; + let (m, len, next) = read_head(header, hp); + assert_eq!(m, 2); + hp = next; + // A DAG-CBOR link wraps `0x00 || `; the 0x00 is + // the multibase identity prefix and is not part of the CID. + // Assert on it rather than skipping it silently — this + // reader stands in for a foreign consumer, and dropping the + // check would let the header regress unnoticed. + let raw = &header[hp..hp + len as usize]; + assert_eq!( + raw.first(), + Some(&0x00), + "CAR root link must carry the multibase identity prefix" + ); + roots.push(cid::Cid::read_bytes(&raw[1..]).unwrap().to_string()); + hp += len as usize; + } + } else { + panic!("unexpected CAR header key {key}"); + } + } + + let mut blocks = Vec::new(); + while p < bytes.len() { + let section_len = varint(bytes, &mut p) as usize; + let section = &bytes[p..p + section_len]; + let cid = cid::Cid::read_bytes(section).unwrap(); + blocks.push(cid.to_string()); + p += section_len; + } + (roots, blocks) +} + +// -- tests ----------------------------------------------------------------- + +/// A live subscriber receives a `#commit` frame for a record created after it +/// connected, and every field of that frame says what it should. +#[tokio::test] +async fn live_subscriber_receives_a_commit_frame() { + if !wait_for_pds().await { + eprintln!("pds not running, skipping"); + return; + } + let (c, did, jwt) = fresh_user("fhlive").await; + + let mut socket = subscribe(None).await; + settle().await; + + let created = create_post(&c, &did, &jwt, "hello firehose").await; + let record_cid = created["cid"].as_str().expect("createRecord cid").to_string(); + let commit_cid = created["commit"]["cid"] + .as_str() + .expect("createRecord commit.cid") + .to_string(); + let commit_rev = created["commit"]["rev"].as_str().unwrap().to_string(); + let uri = created["uri"].as_str().unwrap().to_string(); + let rkey = uri.rsplit('/').next().unwrap().to_string(); + + // The account was created moments ago and has never written before, so + // the first frame we see for it is this post's. Other accounts may be + // writing concurrently, so filter by DID rather than taking frame 1. + let (header, body) = loop { + let bytes = next_frame(&mut socket) + .await + .expect("expected a #commit frame within the timeout"); + let (header, body) = parse_frame(&bytes); + if header.opt("t").map(|t| t.text()) == Some("#commit") + && body.get("repo").text() == did + { + break (header, body); + } + }; + + assert_eq!(header.get("op").int(), 1, "regular frames carry op = 1"); + assert_eq!(header.get("t").text(), "#commit"); + + assert!(body.get("seq").int() > 0, "seq must be a real cursor value"); + assert_eq!(body.get("repo").text(), did); + assert_eq!(body.get("rev").text(), commit_rev); + assert_eq!(body.get("commit").link_cid(), commit_cid); + assert!(!body.get("rebase").bool()); + assert!(!body.get("tooBig").bool()); + assert!(body.get("blobs").array().is_empty()); + // First-ever commit on a brand new repo — there is no predecessor. + assert_eq!(*body.get("since"), Cbor::Null, "first commit has since = null"); + // RFC 3339 with a Z offset. + let time = body.get("time").text(); + assert!( + chrono_like_rfc3339(time), + "time must be RFC 3339, got {time}" + ); + + // ops + let ops = body.get("ops").array(); + assert_eq!(ops.len(), 1, "one record write means one op"); + assert_eq!(ops[0].get("action").text(), "create"); + assert_eq!(ops[0].get("path").text(), format!("app.twi.post/{rkey}")); + assert_eq!(ops[0].get("cid").link_cid(), record_cid); + + // blocks: a CAR rooted at the commit, containing the commit block and the + // new record block. + let (roots, blocks) = parse_car(body.get("blocks").bytes()); + assert_eq!(roots, vec![commit_cid.clone()], "CAR root is the commit"); + assert!( + blocks.contains(&commit_cid), + "CAR must carry the commit block itself; got {blocks:?}" + ); + assert!( + blocks.contains(&record_cid), + "CAR must carry the new record block; got {blocks:?}" + ); + + let _ = socket.send(Message::Close(None)).await; +} + +/// The same event, fetched again from the durable log with a cursor, is +/// byte-identical to the live frame. +/// +/// Byte-identity is the strong form of the claim and the one that matters: a +/// consumer that deduplicates by hashing frames, or that verifies a signature +/// over them, must not see two different representations of one event. +#[tokio::test] +async fn cursor_replay_reproduces_the_live_frame_exactly() { + if !wait_for_pds().await { + eprintln!("pds not running, skipping"); + return; + } + let (c, did, jwt) = fresh_user("fhreplay").await; + + let mut live = subscribe(None).await; + settle().await; + create_post(&c, &did, &jwt, "replay me").await; + + let live_bytes = loop { + let bytes = next_frame(&mut live) + .await + .expect("expected a live #commit frame"); + let (header, body) = parse_frame(&bytes); + if header.opt("t").map(|t| t.text()) == Some("#commit") + && body.get("repo").text() == did + { + break bytes; + } + }; + let (_h, live_body) = parse_frame(&live_bytes); + let seq = live_body.get("seq").int(); + let _ = live.send(Message::Close(None)).await; + + // Reconnect asking for everything after the event *before* ours, so the + // replay's first matching frame is the one we just saw. + let mut replayed = subscribe(Some(seq - 1)).await; + let replay_bytes = loop { + let bytes = next_frame(&mut replayed) + .await + .expect("expected the event to come back from the replay"); + let (header, body) = parse_frame(&bytes); + if header.opt("t").map(|t| t.text()) == Some("#commit") + && body.get("seq").int() == seq + { + break bytes; + } + }; + + assert_eq!( + replay_bytes, live_bytes, + "a replayed frame must be byte-identical to the live one" + ); + let _ = replayed.send(Message::Close(None)).await; +} + +/// Handing over from replay to live loses nothing and duplicates nothing. +/// +/// Connect with a cursor at the current head (so the replay is empty), then +/// write twice: both events must arrive, in order, exactly once each. +#[tokio::test] +async fn replay_to_live_handover_has_no_gap_and_no_duplicate() { + if !wait_for_pds().await { + eprintln!("pds not running, skipping"); + return; + } + let (c, did, jwt) = fresh_user("fhhandover").await; + + // Establish where the log currently ends by writing one event and reading + // its seq off the live stream. + let mut probe = subscribe(None).await; + settle().await; + create_post(&c, &did, &jwt, "probe").await; + let head_seq = loop { + let bytes = next_frame(&mut probe).await.expect("probe frame"); + let (header, body) = parse_frame(&bytes); + if header.opt("t").map(|t| t.text()) == Some("#commit") + && body.get("repo").text() == did + { + break body.get("seq").int(); + } + }; + let _ = probe.send(Message::Close(None)).await; + + // Now reconnect at that exact cursor: nothing to replay, straight to live. + let mut socket = subscribe(Some(head_seq)).await; + settle().await; + + create_post(&c, &did, &jwt, "after handover one").await; + create_post(&c, &did, &jwt, "after handover two").await; + + let mut seen: Vec = Vec::new(); + while seen.len() < 2 { + let bytes = next_frame(&mut socket) + .await + .expect("expected both post-handover frames"); + let (header, body) = parse_frame(&bytes); + if header.opt("t").map(|t| t.text()) != Some("#commit") { + continue; + } + let seq = body.get("seq").int(); + assert!( + seq > head_seq, + "the cursor said we already had seq {head_seq}; got {seq} again" + ); + if body.get("repo").text() == did { + assert!(!seen.contains(&seq), "event {seq} delivered twice"); + seen.push(seq); + } + } + assert_eq!(seen.len(), 2); + assert!(seen[0] < seen[1], "events must arrive in seq order: {seen:?}"); + let _ = socket.send(Message::Close(None)).await; +} + +/// A cursor past the end of the log is a terminal error frame, not silence. +#[tokio::test] +async fn future_cursor_gets_an_error_frame() { + if !wait_for_pds().await { + eprintln!("pds not running, skipping"); + return; + } + let mut socket = subscribe(Some(i64::MAX / 2)).await; + let bytes = next_frame(&mut socket) + .await + .expect("expected an error frame for a future cursor"); + let (header, body) = parse_frame(&bytes); + assert_eq!(header.get("op").int(), -1, "error frames carry op = -1"); + assert!( + header.opt("t").is_none(), + "an error header has no `t`, only `op`" + ); + assert_eq!(body.get("error").text(), "FutureCursor"); + assert!( + !body.get("message").text().is_empty(), + "the error should say what went wrong" + ); +} + +/// A record deleted through `deleteRecord` produces a `delete` op with a null +/// CID — the one op shape that is not a link. +#[tokio::test] +async fn delete_produces_a_delete_op_with_a_null_cid() { + if !wait_for_pds().await { + eprintln!("pds not running, skipping"); + return; + } + let (c, did, jwt) = fresh_user("fhdelete").await; + let created = create_post(&c, &did, &jwt, "to be deleted").await; + let rkey = created["uri"] + .as_str() + .unwrap() + .rsplit('/') + .next() + .unwrap() + .to_string(); + + let mut socket = subscribe(None).await; + settle().await; + + let resp = c + .post(format!("{PDS_URL}/xrpc/com.atproto.repo.deleteRecord")) + .bearer_auth(&jwt) + .json(&json!({ + "repo": did, + "collection": "app.twi.post", + "rkey": rkey, + })) + .send() + .await + .unwrap(); + assert!(resp.status().is_success(), "deleteRecord: {:?}", resp.status()); + + let (_header, body) = loop { + let bytes = next_frame(&mut socket).await.expect("expected a delete frame"); + let (header, body) = parse_frame(&bytes); + if header.opt("t").map(|t| t.text()) == Some("#commit") + && body.get("repo").text() == did + { + break (header, body); + } + }; + + let ops = body.get("ops").array(); + assert_eq!(ops.len(), 1); + assert_eq!(ops[0].get("action").text(), "delete"); + assert_eq!(ops[0].get("path").text(), format!("app.twi.post/{rkey}")); + assert_eq!(*ops[0].get("cid"), Cbor::Null, "a delete has no resulting CID"); + + // The second commit on this repo, so `since` is the previous revision. + assert!( + matches!(body.get("since"), Cbor::Text(_)), + "a follow-up commit must name its predecessor's rev, got {:?}", + body.get("since") + ); + let _ = socket.send(Message::Close(None)).await; +} + +/// Shape check for the frame's `time`: RFC 3339, UTC, with a `Z` suffix. +fn chrono_like_rfc3339(s: &str) -> bool { + chrono::DateTime::parse_from_rfc3339(s).is_ok() && s.ends_with('Z') +} diff --git a/migrations/pds/0003_firehose.sql b/migrations/pds/0003_firehose.sql new file mode 100644 index 0000000..e33a1c9 --- /dev/null +++ b/migrations/pds/0003_firehose.sql @@ -0,0 +1,112 @@ +-- PDS database schema 0003: the firehose event log. +-- +-- Why +-- +-- Until now the PDS produced no `com.atproto.sync.subscribeRepos` stream at +-- all. The only way a local record reached the AppView was the best-effort +-- HTTP push in `appview_push.rs` — a fire-and-forget `tokio::spawn` that is +-- explicitly documented as "the Jetstream replay will catch up". There is no +-- Jetstream replay for records that only exist on this PDS, so a dropped push +-- meant the post was simply never indexed. Nothing retried it, and nothing +-- could: the commit lived in `repos` / `repo_blocks` but there was no ordered +-- log of *what changed* for a consumer to walk. +-- +-- This table is that log. Every repo write appends exactly one row, in the +-- same transaction as the head-pointer update, so the sequence and the repo +-- head can never disagree. A consumer that reconnects with a cursor replays +-- from here; a consumer that is live gets the same rows pushed over a +-- broadcast channel. +-- +-- Column choices +-- +-- seq BIGSERIAL PRIMARY KEY — the cursor. It has to be a single +-- monotonically increasing integer because that is what the +-- `subscribeRepos` wire contract hands the client and takes +-- back as `?cursor=`. BIGSERIAL (not an `(timestamp, id)` +-- keyset like the AppView's notifications table) because the +-- protocol's cursor is opaque-but-numeric and clients compare +-- it with `>`. +-- +-- Sequence values are handed out at INSERT time, which by +-- itself does NOT guarantee that they become *visible* in seq +-- order — two transactions can grab 5 and 6 and commit in the +-- opposite order, leaving a reader that polls in between with a +-- gap it would never fill. The write path therefore takes +-- `pg_advisory_xact_lock` on a fixed key immediately before +-- this INSERT (see `routes::helpers::apply_repo_write`), which +-- serialises the tail of every firehose-writing transaction so +-- commit order == seq order. That is what makes "give me +-- everything with seq > N" an exact, gap-free replay rather +-- than a best guess. +-- +-- did the repo the event belongs to. Not a FK to `users(did)`: +-- the log outlives the account. If a user is deleted we still +-- want consumers that are mid-replay to see the events that +-- already happened rather than have the rows cascade out from +-- under their cursor. +-- +-- rev the new commit's revision (TID string), mirrored from +-- `repos.rev`. Goes out as the frame's `rev`. +-- +-- since the *previous* commit's rev, or NULL for the first commit on +-- a repo. The frame's `since` field; a consumer uses it to +-- detect that it missed an intermediate commit. +-- +-- commit_cid BYTEA holding the raw binary CID of the new commit, stored +-- the same way `repos.head_cid` stores it so the two are +-- directly comparable with `=` and no text/binary conversion +-- is needed to join them. +-- +-- blocks BYTEA holding a complete CAR v1 file: the commit block as the +-- root plus every block this commit newly created (MST nodes +-- and record values). Stored pre-serialised rather than +-- reassembled from `repo_blocks` at read time because the +-- *diff* — which blocks were new for this particular commit — +-- is only knowable at write time. Recomputing it later would +-- mean diffing two MST snapshots on every replayed event. +-- +-- ops JSONB array of `{action, path, cid}`, the same objects that +-- go into the frame's `ops` field. JSONB rather than a child +-- table because it is always read as a whole, is never queried +-- by content, and a child table would need its own ordering +-- column to reproduce the array faithfully. +-- +-- created_at when the event was appended. This is what the frame's `time` +-- field carries, so a replayed frame is byte-identical to the +-- live one that was broadcast at commit time — a consumer that +-- deduplicates by hashing frames does not see two different +-- frames for one event. +-- +-- Retention: there is none +-- ------------------------ +-- Nothing prunes this table. It grows by one row per repo write, and each row +-- carries a CAR of the commit's new blocks (a few hundred bytes for a plain +-- post, more when a record is large). At the volume this deployment sees that +-- is fine for a long time, but it is unbounded, and an operator who wants a +-- bound has to add one. Deleting the oldest rows is safe: a client whose +-- cursor points before the surviving range gets an `#info`/`OutdatedCursor` +-- frame and resumes from the oldest row that still exists. See the module +-- header of `crates/pds-server/src/firehose.rs`. + +CREATE TABLE IF NOT EXISTS firehose_events ( + seq BIGSERIAL PRIMARY KEY, + did TEXT NOT NULL, + rev TEXT NOT NULL, + since TEXT, + commit_cid BYTEA NOT NULL, + blocks BYTEA NOT NULL, + ops JSONB NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- Cursor replay is `WHERE seq > $1 ORDER BY seq LIMIT $2`, which the +-- BIGSERIAL primary key's own index already serves — no second index for +-- that, on purpose: an extra index on `seq` would be pure write amplification +-- on the hottest path in this table. +-- +-- What the PK does *not* serve is "replay one repo", which is how an operator +-- re-drives a single account into the AppView after an ingest bug, and how +-- `getRepo`-style backfills are debugged. `(did, seq)` covers that and keeps +-- the per-repo scan in seq order. +CREATE INDEX IF NOT EXISTS firehose_events_did_seq_idx + ON firehose_events (did, seq);