From 124a90dc07fea78ed045e5d9c3d15020e7e29232 Mon Sep 17 00:00:00 2001 From: tomdebone Date: Thu, 10 Sep 2026 07:08:23 +0200 Subject: [PATCH] feat(appview): PDS-Firehose konsumieren MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Gegenstück zum subscribeRepos-Endpoint: WebSocket-Consumer mit persistiertem seq-Cursor, Reconnect-Backoff und Behandlung von #info/OutdatedCursor. Eigene Cursor-Tabelle statt einer Zeile in jetstream_cursor: dort steht ein time_us in der Größenordnung 1.7e15, die seq ist ein kleiner Zähler ab 1. Geteilt hätte GREATEST den PDS-Cursor sofort in eine Zukunft geschoben, die die PDS nie erreicht. Kein neuer Indexer-Pfad — jede Op wird in die Single-Op-Form übersetzt, die apply_commit schon vom Jetstream kennt. Push und Firehose liefern denselben Commit doppelt; das ist unkritisch, weil die Schreibpfade Upserts sind und der Dedupe-Index der Notifications den Rest abfängt. Mit einem Test festgehalten statt vorausgesetzt. Der CAR-Reader ist neu (es gab nur einen Writer, und der liegt in einem Binary-Crate ohne lib-Target). Der CBOR-Reader arbeitet mit explizitem Offset, weil ein Frame zwei hintereinander geschriebene Werte sind, und akzeptiert CID-Links in beiden Schreibweisen — die Blöcke tragen Strings. /healthz meldet beide Ströme getrennt; sie fallen unabhängig voneinander aus. Verifiziert mit totem Push-Ziel: der Post kam trotzdem an. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013HC9HLrUU1LNwkzp8nkDLX --- crates/appview/Cargo.toml | 9 + crates/appview/src/car.rs | 394 +++++ crates/appview/src/cbor.rs | 442 ++++++ crates/appview/src/firehose.rs | 37 +- crates/appview/src/lib.rs | 3 + crates/appview/src/main.rs | 27 + crates/appview/src/pds_firehose.rs | 1411 +++++++++++++++++ crates/appview/src/routes.rs | 14 + .../appview/tests/pds_firehose_integration.rs | 444 ++++++ crates/at-shared/src/config.rs | 25 + .../appview/0010_pds_firehose_cursor.sql | 44 + 11 files changed, 2849 insertions(+), 1 deletion(-) create mode 100644 crates/appview/src/car.rs create mode 100644 crates/appview/src/cbor.rs create mode 100644 crates/appview/src/pds_firehose.rs create mode 100644 crates/appview/tests/pds_firehose_integration.rs create mode 100644 migrations/appview/0010_pds_firehose_cursor.sql diff --git a/crates/appview/Cargo.toml b/crates/appview/Cargo.toml index 42b1ae5..072a797 100644 --- a/crates/appview/Cargo.toml +++ b/crates/appview/Cargo.toml @@ -40,6 +40,15 @@ uuid = { workspace = true } rustls = { version = "0.23", default-features = false, features = ["ring", "std", "logging", "tls12"] } base64 = { workspace = true } futures = { workspace = true } +# The local PDS firehose (`src/pds_firehose.rs`): a WebSocket carrying +# DAG-CBOR frames whose `blocks` field is a CAR of record blocks. +# `tokio-tungstenite` for the socket, `cid` for the block addresses, +# `ciborium` for the record blocks themselves (they are written with +# `ciborium::into_writer` on the PDS side, so it is their exact inverse). +# The frame envelope is decoded by `src/cbor.rs`, which needs no crate. +tokio-tungstenite = { workspace = true } +cid = { workspace = true } +ciborium = { workspace = true } [dev-dependencies] tokio = { workspace = true } diff --git a/crates/appview/src/car.rs b/crates/appview/src/car.rs new file mode 100644 index 0000000..9c1cea0 --- /dev/null +++ b/crates/appview/src/car.rs @@ -0,0 +1,394 @@ +//! CAR v1 *reader*. +//! +//! The repo has a writer (`pds-server/src/car.rs`) but no reader the +//! AppView could use: `pds-server` is a binary crate with no library +//! target, so its `parse` helper is unreachable from here. The PDS +//! firehose hands us a CAR in every `#commit` frame's `blocks` field, so +//! the AppView needs its own. +//! +//! Format (): +//! +//! ```text +//! [ varint: header_len | DAG-CBOR header ] { version: 1, roots: [CID] } +//! [ varint: section_len | CID | block bytes ] block 1 +//! [ varint: section_len | CID | block bytes ] block 2 +//! ... +//! ``` +//! +//! The header is decoded with [`crate::cbor`], which accepts both the +//! spec's `tag(42) + bytes(0x00 || cid)` link and the bare +//! `tag(42) + bytes(cid)` this codebase's writer emits. +//! +//! Only structure is validated. Block CIDs are *not* re-hashed here: +//! the firehose connection is to our own PDS over the cluster-internal +//! URL, and a mismatch would mean a bug rather than an attack. See +//! [`verify_block_cids`] for the opt-in check the tests use. + +use anyhow::{anyhow, bail, Result}; +use cid::Cid; +use std::collections::HashMap; + +use crate::cbor; + +/// The parsed CAR header. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CarHeader { + pub version: u64, + pub roots: Vec, +} + +/// One `(CID, bytes)` pair out of a CAR file. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CarBlock { + pub cid: Cid, + pub data: Vec, +} + +/// A CAR file's header plus its blocks, in file order. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Car { + pub header: CarHeader, + pub blocks: Vec, +} + +impl Car { + /// Index the blocks by CID for lookup by the commit's `ops`. + /// + /// Duplicate CIDs keep the first occurrence, matching the writer's + /// own de-duplication. + pub fn block_map(&self) -> HashMap { + let mut map = HashMap::with_capacity(self.blocks.len()); + for b in &self.blocks { + map.entry(b.cid).or_insert(b.data.as_slice()); + } + map + } + + /// The first root, if the header declares one. + /// + /// `allow(dead_code)`: the ingest path doesn't need the commit + /// block itself (the ops carry the record CIDs), but a CAR reader + /// that cannot name its root is a reader with a hole in it, and the + /// tests read it. + #[allow(dead_code)] + pub fn root(&self) -> Option { + self.roots().first().copied() + } + + #[allow(dead_code)] + pub fn roots(&self) -> &[Cid] { + &self.header.roots + } +} + +/// Parse a CAR v1 byte stream. +pub fn parse(bytes: &[u8]) -> Result { + let mut p = 0usize; + + let (header_len, n) = read_varint(bytes, p)?; + p += n; + let header_end = checked_end(bytes, p, header_len, "CAR header")?; + let header = decode_header(&bytes[p..header_end])?; + p = header_end; + + let mut blocks = Vec::new(); + while p < bytes.len() { + let (section_len, n) = read_varint(bytes, p)?; + let section_start = p + n; + let section_end = checked_end(bytes, section_start, section_len, "CAR section")?; + let section = &bytes[section_start..section_end]; + let cid = Cid::read_bytes(section) + .map_err(|e| anyhow!("invalid CID in CAR section at offset {section_start}: {e}"))?; + let cid_len = cid.encoded_len(); + if cid_len > section.len() { + bail!("CAR section at {section_start} is shorter than its CID"); + } + blocks.push(CarBlock { + cid, + data: section[cid_len..].to_vec(), + }); + p = section_end; + } + + Ok(Car { header, blocks }) +} + +/// Re-hash every block and compare against its declared CID. +/// +/// Not called on the ingest path (see the module docs); the CAR reader +/// tests use it to prove the reader hands back the bytes the writer put +/// in, unshifted by an off-by-one in the section framing. +#[allow(dead_code)] +pub fn verify_block_cids(car: &Car) -> Result<()> { + for b in &car.blocks { + let recomputed = at_crypto::cid::cid_for_cbor(&b.data)?; + if recomputed != b.cid { + bail!("CAR block CID mismatch: declared {}, computed {recomputed}", b.cid); + } + } + Ok(()) +} + +fn decode_header(bytes: &[u8]) -> Result { + let value = cbor::decode(bytes)?; + let version = value + .get("version") + .and_then(|v| v.as_i64()) + .ok_or_else(|| anyhow!("CAR header missing `version`"))?; + if version != 1 { + bail!("unsupported CAR version {version} (only v1 is defined for atproto)"); + } + // `roots` is required by the spec but may legitimately be empty. + let roots = match value.get("roots") { + Some(v) => v + .as_array() + .ok_or_else(|| anyhow!("CAR header `roots` is not an array"))? + .iter() + .map(|item| { + item.as_cid() + .ok_or_else(|| anyhow!("CAR header root is not a CID link")) + }) + .collect::>>()?, + None => Vec::new(), + }; + Ok(CarHeader { + version: version as u64, + roots, + }) +} + +/// LEB128 unsigned varint, the length prefix CAR uses. +fn read_varint(bytes: &[u8], offset: usize) -> Result<(u64, usize)> { + let mut value: u64 = 0; + let mut shift = 0u32; + let mut i = offset; + loop { + let b = *bytes + .get(i) + .ok_or_else(|| anyhow!("varint extends past end of CAR input at {offset}"))?; + i += 1; + value |= u64::from(b & 0x7f) << shift; + if b & 0x80 == 0 { + return Ok((value, i - offset)); + } + shift += 7; + if shift >= 64 { + bail!("varint longer than 64 bits at offset {offset}"); + } + } +} + +fn checked_end(bytes: &[u8], pos: usize, len: u64, what: &str) -> Result { + let len = usize::try_from(len).map_err(|_| anyhow!("{what} length overflows usize"))?; + let end = pos + .checked_add(len) + .ok_or_else(|| anyhow!("{what} length overflows"))?; + if end > bytes.len() { + bail!("{what} length {len} exceeds input (offset {pos}, total {})", bytes.len()); + } + Ok(end) +} + +#[cfg(test)] +pub(crate) mod test_writer { + //! A byte-for-byte copy of `pds-server/src/car.rs`'s encoder, so the + //! reader's tests exercise *the writer's* output rather than a + //! convenient fiction. + //! + //! Copied rather than imported because `pds-server` has no library + //! target. If the writer ever changes shape, the integration test + //! `pds_firehose_integration::car_reader_parses_a_real_repo_export` + //! is the tripwire: it parses a CAR produced by the running PDS. + + use cid::Cid; + + fn cbor_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 < 0x100 { + out.push(m | 24); + out.push(n as u8); + } else if n < 0x10000 { + out.push(m | 25); + out.push((n >> 8) as u8); + out.push(n as u8); + } else if n < 0x100_0000 { + out.push(m | 26); + out.push((n >> 16) as u8); + out.push((n >> 8) as u8); + out.push(n as u8); + } else { + out.push(m | 27); + out.push((n >> 24) as u8); + out.push((n >> 16) as u8); + out.push((n >> 8) as u8); + out.push(n as u8); + } + } + + fn cbor_text(out: &mut Vec, s: &str) { + cbor_head(out, 3, s.len() as u64); + out.extend_from_slice(s.as_bytes()); + } + + fn cbor_bytes(out: &mut Vec, b: &[u8]) { + cbor_head(out, 2, b.len() as u64); + out.extend_from_slice(b); + } + + pub fn encode_header(roots: &[Cid]) -> Vec { + let mut out = Vec::new(); + cbor_head(&mut out, 5, 2); + cbor_text(&mut out, "version"); + cbor_head(&mut out, 0, 1); + cbor_text(&mut out, "roots"); + cbor_head(&mut out, 4, roots.len() as u64); + for cid in roots { + cbor_head(&mut out, 6, 42); + cbor_bytes(&mut out, &cid.to_bytes()); + } + out + } + + fn write_varint(out: &mut Vec, mut n: u64) { + loop { + let mut byte = (n & 0x7f) as u8; + n >>= 7; + if n != 0 { + byte |= 0x80; + } + out.push(byte); + if n == 0 { + return; + } + } + } + + #[derive(Default)] + pub struct CarWriter { + blocks: Vec<(Cid, Vec)>, + } + + impl CarWriter { + pub fn new() -> Self { + Self::default() + } + + pub fn append(&mut self, cid: Cid, data: &[u8]) { + if self.blocks.iter().any(|(c, _)| *c == cid) { + return; + } + self.blocks.push((cid, data.to_vec())); + } + + pub fn finish(&self, roots: &[Cid]) -> Vec { + let header = encode_header(roots); + let mut out = Vec::new(); + write_varint(&mut out, header.len() as u64); + out.extend_from_slice(&header); + for (cid, data) in &self.blocks { + let cid_bytes = cid.to_bytes(); + write_varint(&mut out, (cid_bytes.len() + data.len()) as u64); + out.extend_from_slice(&cid_bytes); + out.extend_from_slice(data); + } + out + } + } +} + +#[cfg(test)] +mod tests { + use super::test_writer::CarWriter; + use super::*; + use at_crypto::cid::cid_for_cbor; + + #[test] + fn round_trips_a_single_block() { + let data = b"hello world".to_vec(); + let cid = cid_for_cbor(&data).unwrap(); + let mut w = CarWriter::new(); + w.append(cid, &data); + let car = parse(&w.finish(&[cid])).unwrap(); + assert_eq!(car.header.version, 1); + assert_eq!(car.roots(), &[cid]); + assert_eq!(car.blocks.len(), 1); + assert_eq!(car.blocks[0].data, data); + verify_block_cids(&car).unwrap(); + } + + #[test] + fn round_trips_many_blocks_in_order() { + let payloads: Vec> = (0..6) + .map(|i| format!("block-{i}").into_bytes()) + .collect(); + let cids: Vec = payloads.iter().map(|p| cid_for_cbor(p).unwrap()).collect(); + let mut w = CarWriter::new(); + for (cid, data) in cids.iter().zip(&payloads) { + w.append(*cid, data); + } + let car = parse(&w.finish(&[cids[3]])).unwrap(); + assert_eq!(car.root(), Some(cids[3])); + assert_eq!(car.blocks.len(), 6); + for (i, b) in car.blocks.iter().enumerate() { + assert_eq!(b.cid, cids[i]); + assert_eq!(b.data, payloads[i]); + } + verify_block_cids(&car).unwrap(); + } + + #[test] + fn block_map_finds_blocks_by_cid() { + let a = b"record a".to_vec(); + let b = b"record b".to_vec(); + let (ca, cb) = (cid_for_cbor(&a).unwrap(), cid_for_cbor(&b).unwrap()); + let mut w = CarWriter::new(); + w.append(ca, &a); + w.append(cb, &b); + let car = parse(&w.finish(&[ca])).unwrap(); + let map = car.block_map(); + assert_eq!(map.get(&ca).copied(), Some(a.as_slice())); + assert_eq!(map.get(&cb).copied(), Some(b.as_slice())); + assert!(!map.contains_key(&cid_for_cbor(b"absent").unwrap())); + } + + #[test] + fn handles_a_multi_byte_varint_section_length() { + // A >127 byte block forces a two-byte varint, which is where an + // off-by-one in the length prefix would show up. + let data = vec![0x42u8; 500]; + let cid = cid_for_cbor(&data).unwrap(); + let mut w = CarWriter::new(); + w.append(cid, &data); + let car = parse(&w.finish(&[cid])).unwrap(); + assert_eq!(car.blocks[0].data.len(), 500); + verify_block_cids(&car).unwrap(); + } + + #[test] + fn accepts_an_empty_root_list() { + let data = b"orphan".to_vec(); + let cid = cid_for_cbor(&data).unwrap(); + let mut w = CarWriter::new(); + w.append(cid, &data); + let car = parse(&w.finish(&[])).unwrap(); + assert!(car.roots().is_empty()); + assert_eq!(car.blocks.len(), 1); + } + + #[test] + fn rejects_truncated_input() { + let data = b"hello".to_vec(); + let cid = cid_for_cbor(&data).unwrap(); + let mut w = CarWriter::new(); + w.append(cid, &data); + let bytes = w.finish(&[cid]); + // Cut into the last block's payload. + assert!(parse(&bytes[..bytes.len() - 3]).is_err()); + // Cut inside the header. + assert!(parse(&bytes[..3]).is_err()); + // Nothing at all. + assert!(parse(&[]).is_err()); + } +} diff --git a/crates/appview/src/cbor.rs b/crates/appview/src/cbor.rs new file mode 100644 index 0000000..973ac1c --- /dev/null +++ b/crates/appview/src/cbor.rs @@ -0,0 +1,442 @@ +//! A minimal, allocation-honest CBOR reader — just enough to decode the +//! frames of `com.atproto.sync.subscribeRepos` and the header of a CAR +//! file. +//! +//! ## Why not `ciborium`? +//! +//! Two reasons, and both come from what the wire format actually is. +//! +//! 1. **Two values per message.** A subscribeRepos frame is *two* +//! DAG-CBOR values written back to back (header, then body) inside +//! one WebSocket binary message. A `serde`-shaped reader gives us +//! "decode one value from this slice" and no cursor we can resume +//! from, so we would have to guess where the header ended. +//! 2. **Tag 42.** DAG-CBOR encodes a CID link as `tag(42) + +//! bytes()`. `ciborium`'s `serde` mapping turns tags into a +//! private newtype dance that does not survive a round trip through +//! `serde_json::Value`, which is the shape the rest of the AppView +//! speaks. +//! +//! So this module decodes CBOR into its own small [`Cbor`] tree with an +//! explicit byte offset, which makes "read the header, then read the +//! body from where the header stopped" a two-line function. +//! +//! ## What it deliberately does not do +//! +//! No indefinite-length items (DAG-CBOR forbids them; we reject them +//! rather than guess), no half floats beyond a plain `f64` widening, no +//! canonicalisation checks. It is a *reader* for input we already +//! decided to trust at the transport layer, with hard length checks so +//! a malformed frame returns `Err` instead of panicking. +//! +//! Record blocks inside the CAR payload are NOT decoded with this +//! module: they are written by `ciborium::into_writer(&serde_json::Value)` +//! on the PDS side (see `pds-server/src/routes/repo.rs`), which means +//! CIDs inside a record are plain strings, not tag-42 links. Their +//! inverse is `ciborium::from_reader::`, and +//! that is what [`crate::pds_firehose`] uses for them. + +use anyhow::{anyhow, bail, Result}; +use cid::Cid; + +/// A decoded CBOR value. +/// +/// `Nint` carries the already-negated value (CBOR stores `-1 - n`), so +/// callers never have to remember the bias. +#[derive(Debug, Clone, PartialEq)] +pub enum Cbor { + Uint(u64), + Nint(i64), + Bytes(Vec), + Text(String), + Array(Vec), + /// Kept as an ordered key/value list rather than a map: DAG-CBOR + /// keys are text and already canonically ordered, and a `Vec` keeps + /// the decoder free of hashing while the maps we read have a + /// handful of entries at most. + Map(Vec<(Cbor, Cbor)>), + Tag(u64, Box), + Bool(bool), + Null, + Undefined, + Float(f64), +} + +impl Cbor { + /// Look up a text key in a map. `None` for non-maps and misses. + pub fn get(&self, key: &str) -> Option<&Cbor> { + match self { + Cbor::Map(entries) => entries.iter().find_map(|(k, v)| match k { + Cbor::Text(s) if s == key => Some(v), + _ => None, + }), + _ => None, + } + } + + /// The value as a signed integer, accepting both CBOR integer + /// majors. `None` when the value is not an integer, or when an + /// unsigned value exceeds `i64::MAX` (which cannot happen for a + /// `seq`, but silently wrapping would be worse than a miss). + pub fn as_i64(&self) -> Option { + match self { + Cbor::Uint(n) => i64::try_from(*n).ok(), + Cbor::Nint(n) => Some(*n), + _ => None, + } + } + + pub fn as_str(&self) -> Option<&str> { + match self { + Cbor::Text(s) => Some(s.as_str()), + _ => None, + } + } + + pub fn as_bytes(&self) -> Option<&[u8]> { + match self { + Cbor::Bytes(b) => Some(b.as_slice()), + _ => None, + } + } + + pub fn as_bool(&self) -> Option { + match self { + Cbor::Bool(b) => Some(*b), + _ => None, + } + } + + pub fn as_array(&self) -> Option<&[Cbor]> { + match self { + Cbor::Array(items) => Some(items.as_slice()), + _ => None, + } + } + + pub fn is_null(&self) -> bool { + matches!(self, Cbor::Null | Cbor::Undefined) + } + + /// Decode a DAG-CBOR CID link: `tag(42) + bytes(...)`. + /// + /// The spec prefixes the CID bytes with a single `0x00` (the + /// multibase "identity" marker), because a CID inside a byte string + /// has no textual multibase prefix to carry. **This repository's own + /// writer omits that byte** — `pds-server/src/car.rs` writes + /// `cbor_bytes(&cid.to_bytes())` — so we accept both spellings: a + /// leading `0x00` is skipped, anything else is parsed as-is. Being + /// lenient here costs nothing (a real CIDv1 never starts with + /// `0x00`, and CIDv0 starts with `0x12`) and it means the AppView + /// keeps working whether the PDS follows the spec or the house + /// convention. + pub fn as_cid(&self) -> Option { + let inner = match self { + Cbor::Tag(42, inner) => inner.as_ref(), + // A bare byte string where a link is expected: CARs written by + // older builds of this PDS tagged the root CID without the + // `0x00` identity prefix, and some encoders drop the tag + // entirely. Both still have to parse. + Cbor::Bytes(_) => self, + // A record block written from `serde_json::Value` spells a + // CID as a plain string — accept that too, so callers do + // not need a second code path for the block contents. + Cbor::Text(s) => return s.parse::().ok(), + _ => return None, + }; + let raw = inner.as_bytes()?; + let raw = match raw.first() { + Some(0x00) => &raw[1..], + _ => raw, + }; + Cid::read_bytes(raw).ok() + } +} + +/// Decode exactly one CBOR value starting at `pos`. +/// +/// Returns the value and the offset just past it, so a caller can read +/// the next value from the same buffer — which is precisely what a +/// two-value subscribeRepos frame needs. +pub fn decode_at(bytes: &[u8], pos: usize) -> Result<(Cbor, usize)> { + let (major, arg, mut p) = read_head(bytes, pos)?; + match major { + 0 => Ok((Cbor::Uint(arg), p)), + 1 => { + // CBOR negative integers store `-1 - n`. Values below + // `i64::MIN` cannot occur in anything we consume, and + // wrapping them would produce a positive number, so bail. + let n = i64::try_from(arg) + .map_err(|_| anyhow!("negative integer out of i64 range"))?; + Ok((Cbor::Nint(-1 - n), p)) + } + 2 => { + let end = checked_end(bytes, p, arg, "byte string")?; + let v = bytes[p..end].to_vec(); + Ok((Cbor::Bytes(v), end)) + } + 3 => { + let end = checked_end(bytes, p, arg, "text string")?; + let s = std::str::from_utf8(&bytes[p..end]) + .map_err(|e| anyhow!("invalid UTF-8 in CBOR text: {e}"))? + .to_string(); + Ok((Cbor::Text(s), end)) + } + 4 => { + let mut items = Vec::with_capacity(sane_capacity(arg)); + for _ in 0..arg { + let (v, next) = decode_at(bytes, p)?; + items.push(v); + p = next; + } + Ok((Cbor::Array(items), p)) + } + 5 => { + let mut entries = Vec::with_capacity(sane_capacity(arg)); + for _ in 0..arg { + let (k, next) = decode_at(bytes, p)?; + let (v, next) = decode_at(bytes, next)?; + entries.push((k, v)); + p = next; + } + Ok((Cbor::Map(entries), p)) + } + 6 => { + let (inner, next) = decode_at(bytes, p)?; + Ok((Cbor::Tag(arg, Box::new(inner)), next)) + } + 7 => match arg { + 20 => Ok((Cbor::Bool(false), p)), + 21 => Ok((Cbor::Bool(true), p)), + 22 => Ok((Cbor::Null, p)), + 23 => Ok((Cbor::Undefined, p)), + // Floats arrive as the raw bit pattern in `arg`; the width + // is implied by the additional-information byte, which + // `read_head` has already consumed. We only ever see f64 in + // practice (DAG-CBOR requires it), so the narrower widths + // are decoded for completeness rather than need. + _ => Ok((Cbor::Float(f64::from_bits(arg)), p)), + }, + other => bail!("unsupported CBOR major type {other}"), + } +} + +/// Decode a single CBOR value that must span the whole buffer. +pub fn decode(bytes: &[u8]) -> Result { + let (v, end) = decode_at(bytes, 0)?; + if end != bytes.len() { + bail!("trailing bytes after CBOR value ({} left)", bytes.len() - end); + } + Ok(v) +} + +/// Read a CBOR head: major type plus its argument. +/// +/// Indefinite-length encodings (`additional information == 31`) are +/// rejected: DAG-CBOR forbids them, and accepting them would mean +/// implementing break-stop scanning for input that should never carry +/// it. +fn read_head(bytes: &[u8], pos: usize) -> Result<(u8, u64, usize)> { + let first = *bytes + .get(pos) + .ok_or_else(|| anyhow!("CBOR read past end of input at {pos}"))?; + let major = first >> 5; + let low = first & 0x1f; + let (arg, extra) = match low { + 0..=23 => (low as u64, 0usize), + 24 => (read_uint(bytes, pos + 1, 1)?, 1), + 25 => (read_uint(bytes, pos + 1, 2)?, 2), + 26 => (read_uint(bytes, pos + 1, 4)?, 4), + 27 => (read_uint(bytes, pos + 1, 8)?, 8), + 31 => bail!("indefinite-length CBOR item is not valid DAG-CBOR"), + other => bail!("reserved CBOR additional information {other}"), + }; + // For major 7 the "argument" of a float is the raw bit pattern, and + // f32/f16 need widening before `f64::from_bits` makes sense. + let arg = match (major, low) { + (7, 25) => f64::from(half_to_f32(arg as u16)).to_bits(), + (7, 26) => f64::from(f32::from_bits(arg as u32)).to_bits(), + _ => arg, + }; + Ok((major, arg, pos + 1 + extra)) +} + +fn read_uint(bytes: &[u8], pos: usize, len: usize) -> Result { + if pos + len > bytes.len() { + bail!("truncated CBOR integer of {len} byte(s) at {pos}"); + } + let mut n: u64 = 0; + for b in &bytes[pos..pos + len] { + n = (n << 8) | u64::from(*b); + } + Ok(n) +} + +/// IEEE-754 half → single. Only reached for `f16` inputs, which nothing +/// in this protocol emits; kept so a stray value decodes instead of +/// erroring out mid-frame. +fn half_to_f32(bits: u16) -> f32 { + let sign = ((bits >> 15) & 1) as u32; + let exp = ((bits >> 10) & 0x1f) as u32; + let frac = (bits & 0x3ff) as u32; + let out = match exp { + 0 if frac == 0 => sign << 31, + 0 => { + // Subnormal: renormalise. + let mut e = -1i32; + let mut f = frac; + while f & 0x400 == 0 { + f <<= 1; + e -= 1; + } + let exp32 = (127 - 15 + e) as u32; + (sign << 31) | (exp32 << 23) | ((f & 0x3ff) << 13) + } + 0x1f => (sign << 31) | (0xff << 23) | (frac << 13), + _ => (sign << 31) | ((exp + 127 - 15) << 23) | (frac << 13), + }; + f32::from_bits(out) +} + +/// Bounds-check a string/bytes payload before slicing it. +fn checked_end(bytes: &[u8], pos: usize, len: u64, what: &str) -> Result { + let len = usize::try_from(len).map_err(|_| anyhow!("{what} length overflows usize"))?; + let end = pos + .checked_add(len) + .ok_or_else(|| anyhow!("{what} length overflows"))?; + if end > bytes.len() { + bail!("{what} of {len} byte(s) exceeds input at offset {pos}"); + } + Ok(end) +} + +/// Cap the pre-allocation a declared array/map length can trigger. A +/// corrupt frame claiming `map(2^40)` must not make us reserve 40 GiB +/// before the first missing byte errors out; the collection still grows +/// naturally for genuinely large inputs. +fn sane_capacity(declared: u64) -> usize { + usize::try_from(declared.min(1024)).unwrap_or(0) +} + +#[cfg(test)] +mod tests { + use super::*; + use at_crypto::cid::cid_for_cbor; + + /// Build `{"op": 1, "t": "#commit"}` by hand — the exact bytes the + /// PDS writes for a regular frame header. + fn commit_header_bytes() -> Vec { + let mut v = vec![0xA2]; // map(2) + v.push(0x62); // text(2) + v.extend_from_slice(b"op"); + v.push(0x01); // uint 1 + v.push(0x61); // text(1) + v.extend_from_slice(b"t"); + v.push(0x67); // text(7) + v.extend_from_slice(b"#commit"); + v + } + + #[test] + fn decodes_a_frame_header() { + let v = decode(&commit_header_bytes()).unwrap(); + assert_eq!(v.get("op").unwrap().as_i64(), Some(1)); + assert_eq!(v.get("t").unwrap().as_str(), Some("#commit")); + assert!(v.get("missing").is_none()); + } + + #[test] + fn decodes_negative_op_of_an_error_header() { + // {"op": -1} → map(1), text(2)"op", nint(0) = -1 + let bytes = vec![0xA1, 0x62, b'o', b'p', 0x20]; + let v = decode(&bytes).unwrap(); + assert_eq!(v.get("op").unwrap().as_i64(), Some(-1)); + } + + #[test] + fn decodes_multi_byte_integers() { + // uint16 300, uint32 70000, uint64 2^33, nint -300 + assert_eq!(decode(&[0x19, 0x01, 0x2C]).unwrap().as_i64(), Some(300)); + assert_eq!( + decode(&[0x1A, 0x00, 0x01, 0x11, 0x70]).unwrap().as_i64(), + Some(70000) + ); + assert_eq!( + decode(&[0x1B, 0, 0, 0, 2, 0, 0, 0, 0]).unwrap().as_i64(), + Some(8_589_934_592) + ); + assert_eq!(decode(&[0x39, 0x01, 0x2B]).unwrap().as_i64(), Some(-300)); + } + + #[test] + fn decode_at_reads_two_values_back_to_back() { + // This is the whole reason the module exists: a frame is header + // + body concatenated with no separator. + let mut buf = commit_header_bytes(); + let body_start = buf.len(); + buf.extend_from_slice(&[0xA1, 0x63, b's', b'e', b'q', 0x18, 0x2A]); // {"seq": 42} + let (header, next) = decode_at(&buf, 0).unwrap(); + assert_eq!(next, body_start); + assert_eq!(header.get("t").unwrap().as_str(), Some("#commit")); + let (body, end) = decode_at(&buf, next).unwrap(); + assert_eq!(end, buf.len()); + assert_eq!(body.get("seq").unwrap().as_i64(), Some(42)); + } + + #[test] + fn decodes_tag_42_cid_link_both_spellings() { + let cid = cid_for_cbor(b"a block").unwrap(); + let raw = cid.to_bytes(); + + // House spelling: tag(42) + bytes() with no 0x00 prefix. + let mut bare = vec![0xD8, 42]; // tag(42) via 1-byte extension + bare.push(0x58); // bytes, 1-byte length + bare.push(raw.len() as u8); + bare.extend_from_slice(&raw); + assert_eq!(decode(&bare).unwrap().as_cid(), Some(cid)); + + // Spec spelling: the same, with the identity multibase prefix. + let mut prefixed = vec![0xD8, 42, 0x58, (raw.len() + 1) as u8, 0x00]; + prefixed.extend_from_slice(&raw); + assert_eq!(decode(&prefixed).unwrap().as_cid(), Some(cid)); + + // And the string spelling records use. + let text = { + let s = cid.to_string(); + let mut v = vec![0x78, s.len() as u8]; + v.extend_from_slice(s.as_bytes()); + v + }; + assert_eq!(decode(&text).unwrap().as_cid(), Some(cid)); + } + + #[test] + fn decodes_simple_values_and_containers() { + // [true, false, null] → array(3) + let v = decode(&[0x83, 0xF5, 0xF4, 0xF6]).unwrap(); + let items = v.as_array().unwrap(); + assert_eq!(items[0].as_bool(), Some(true)); + assert_eq!(items[1].as_bool(), Some(false)); + assert!(items[2].is_null()); + } + + #[test] + fn rejects_truncated_and_indefinite_input() { + // text(7) claiming 7 bytes but carrying 2. + assert!(decode(&[0x67, b'a', b'b']).is_err()); + // Indefinite-length array. + assert!(decode(&[0x9F, 0x01, 0xFF]).is_err()); + // Trailing garbage after a complete value. + assert!(decode(&[0x01, 0x02]).is_err()); + // Empty input. + assert!(decode(&[]).is_err()); + } + + #[test] + fn oversized_declared_length_errors_instead_of_allocating() { + // map(2^32) with nothing behind it. Must return Err quickly + // rather than trying to reserve the declared capacity. + let bytes = vec![0xBA, 0xFF, 0xFF, 0xFF, 0xFF]; + assert!(decode(&bytes).is_err()); + } +} diff --git a/crates/appview/src/firehose.rs b/crates/appview/src/firehose.rs index 5c9703c..cf57034 100644 --- a/crates/appview/src/firehose.rs +++ b/crates/appview/src/firehose.rs @@ -16,7 +16,7 @@ use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering}; use std::sync::Arc; use std::time::Duration; use tokio::sync::mpsc; -use tracing::{debug, info, trace, warn}; +use tracing::{debug, info, warn}; use crate::indexer; @@ -31,6 +31,18 @@ pub struct Stats { /// writes this; `/healthz` reads it. Wrapped in `Arc` so the consumer /// can hold its own clone without borrowing from us. pub jetstream_connected: Arc, + /// Whether the **local PDS** firehose WebSocket is currently up + /// ([`crate::pds_firehose`]). Separate from `jetstream_connected` + /// because the two streams fail independently and for different + /// reasons: a dead Jetstream means no view of the wider network, a + /// dead PDS firehose means the AppView has lost the guaranteed + /// delivery path for its *own* users' records and is running on the + /// best-effort push alone. `/healthz` has to be able to say which. + pub pds_connected: AtomicBool, + /// Number of `#commit` frames applied from the PDS firehose. + pub pds_frames_processed: AtomicU64, + /// Highest `seq` applied from the PDS firehose in this process. + pub pds_last_seq: AtomicI64, } impl Default for Stats { @@ -40,6 +52,9 @@ impl Default for Stats { last_event_time_us: AtomicI64::new(0), last_cursor_persisted_us: AtomicI64::new(0), jetstream_connected: Arc::new(AtomicBool::new(false)), + pds_connected: AtomicBool::new(false), + pds_frames_processed: AtomicU64::new(0), + pds_last_seq: AtomicI64::new(0), } } } @@ -50,6 +65,9 @@ impl std::fmt::Debug for Stats { .field("events_processed", &self.events_processed()) .field("last_event_time_us", &self.last_event_time_us.load(Ordering::Relaxed)) .field("jetstream_connected", &self.jetstream_connected()) + .field("pds_connected", &self.pds_connected()) + .field("pds_frames_processed", &self.pds_frames_processed()) + .field("pds_last_seq", &self.pds_last_seq()) .finish() } } @@ -85,6 +103,23 @@ impl Stats { pub fn jetstream_connected(&self) -> bool { self.jetstream_connected.load(Ordering::Relaxed) } + + /// Is the local PDS firehose connected right now? + pub fn pds_connected(&self) -> bool { + self.pds_connected.load(Ordering::Relaxed) + } + + pub fn pds_frames_processed(&self) -> u64 { + self.pds_frames_processed.load(Ordering::Relaxed) + } + + /// Highest PDS-firehose `seq` this process has applied. 0 before the + /// first frame — note this is the *in-process* high-water mark, not + /// the persisted cursor, which lives in `pds_firehose_cursor` and + /// survives restarts. + pub fn pds_last_seq(&self) -> i64 { + self.pds_last_seq.load(Ordering::Relaxed) + } } /// The thing the Jetstream consumer calls once per event. diff --git a/crates/appview/src/lib.rs b/crates/appview/src/lib.rs index 9440bc6..877af2c 100644 --- a/crates/appview/src/lib.rs +++ b/crates/appview/src/lib.rs @@ -4,9 +4,12 @@ //! against a stub resolver without booting the binary. pub mod auth; +pub mod car; +pub mod cbor; pub mod firehose; pub mod handle_sync; pub mod indexer; pub mod ingest; +pub mod pds_firehose; pub mod routes; pub mod state; diff --git a/crates/appview/src/main.rs b/crates/appview/src/main.rs index 1087f24..fe8bc17 100644 --- a/crates/appview/src/main.rs +++ b/crates/appview/src/main.rs @@ -8,10 +8,13 @@ use tracing::info; use tracing_subscriber::EnvFilter; mod auth; +mod car; +mod cbor; mod firehose; mod handle_sync; mod indexer; mod ingest; +mod pds_firehose; mod routes; mod state; @@ -82,6 +85,30 @@ async fn main() -> Result<()> { }); } + // Local PDS firehose. The push path (`/internal/ingest-commit`) is + // the fast way a local commit reaches the index; this is the + // guaranteed one — it carries a durable cursor, so anything the + // push dropped while the AppView was down is replayed on connect. + // See the module docs in `pds_firehose.rs` for why both exist. + if cfg.pds_firehose_enabled { + let start_seq = pds_firehose::cursor_get(&db).await.unwrap_or(0); + let base = cfg.pds_base_url(); + info!( + url = %pds_firehose::subscribe_url(&base, (start_seq > 0).then_some(start_seq)), + cursor = start_seq, + "starting the local PDS firehose consumer" + ); + let consumer = pds_firehose::PdsFirehose::new(db.clone(), base, stats.clone()) + .with_max_backoff_secs(30); + tokio::spawn(consumer.run_forever()); + } else { + tracing::warn!( + "PDS_FIREHOSE_ENABLED=false — local commits reach the index only through \ + the best-effort `/internal/ingest-commit` push; a push lost to a restart \ + or a network error will NOT be recovered" + ); + } + let state = AppState::new(cfg.clone(), db.clone(), stats.clone()); // Announce every relaxed security switch before we serve anything. diff --git a/crates/appview/src/pds_firehose.rs b/crates/appview/src/pds_firehose.rs new file mode 100644 index 0000000..7f56b46 --- /dev/null +++ b/crates/appview/src/pds_firehose.rs @@ -0,0 +1,1411 @@ +//! Consumer for the **local PDS firehose** +//! (`com.atproto.sync.subscribeRepos`). +//! +//! ## Why this exists next to the push path +//! +//! The AppView learns about a local user's own records twice, on +//! purpose: +//! +//! - **`POST /internal/ingest-commit`** ([`crate::ingest`]) — the PDS +//! pushes each commit the moment it writes it. It is *fast*: the +//! post is in the index before the client's `createRecord` response +//! has finished rendering. It is also *best effort*: a single HTTP +//! request with no retry, no queue and no acknowledgement the +//! AppView can be held to. If the AppView is restarting, or the +//! request times out, that record is gone from the index for good — +//! the public Jetstream never sees this PDS, so nothing replays it. +//! +//! - **this module** — a WebSocket stream with a *sequence number* the +//! AppView persists. It is slower to arrive and heavier to decode +//! (DAG-CBOR frames carrying a CAR of the commit's blocks), but it +//! is *guaranteed*: after a crash, a network blip or an hour of +//! downtime, the AppView reconnects with `?cursor=` and +//! the PDS replays everything it missed. +//! +//! Fast path plus guaranteed path, and the overlap between them is +//! handled by making every write idempotent rather than by trying to +//! deduplicate at the transport layer — which is also what makes a +//! Jetstream reconnect safe. See [`crate::indexer`]: posts UPSERT on +//! `uri`, likes/reposts `ON CONFLICT (did, post_uri) DO NOTHING` (so +//! the counter bump happens exactly once), follows UPSERT on +//! `(follower_did, subject_did)`, and notifications carry a unique +//! dedupe index. A commit arriving over both paths, or the same frame +//! replayed after a restart, converges to the same rows. +//! +//! ## Frame format +//! +//! One WebSocket **binary** message is two DAG-CBOR values written back +//! to back — a header, then a body: +//! +//! ```text +//! header {"op": 1, "t": "#commit"} regular message +//! header {"op": -1} error, body {"error", "message"} +//! +//! body #commit +//! seq (i64), rebase (bool), tooBig (bool), repo (DID), +//! commit (CID link, tag 42), rev (string), since (string|null), +//! blocks (byte string: CAR v1, root = commit block), +//! ops [{action: create|update|delete, path: "/", +//! cid: CID link|null}], +//! blobs [], time (RFC3339) +//! +//! body #info {"name": "OutdatedCursor", "message": ...} +//! ``` +//! +//! The *frame envelope* is DAG-CBOR with tag-42 CID links. The *block +//! contents* inside `blocks` are not: this codebase writes record and +//! commit blocks with `ciborium::into_writer(&serde_json::Value)` (see +//! `pds-server/src/routes/repo.rs` and `at-repo/src/commit.rs`), which +//! spells CIDs as plain strings. So the envelope is decoded by +//! [`crate::cbor`] and the blocks by `ciborium` into `serde_json::Value` +//! — each with the reader that matches how it was written. +//! +//! ## How ops reach the database +//! +//! Every op is converted into the same [`at_firehose::JetstreamEvent`] +//! shape the public firehose delivers and handed to +//! [`crate::indexer::apply_commit`]. That is deliberate: there is +//! exactly one place in the AppView that decides what a +//! `app.bsky.feed.like` create means, and adding a second transport +//! must not fork it. + +use anyhow::{anyhow, bail, Context, Result}; +use at_firehose::JetstreamEvent; +use cid::Cid; +use futures::StreamExt; +use serde_json::{json, Value}; +use sqlx::PgPool; +use std::collections::HashMap; +use std::sync::atomic::Ordering; +use std::sync::Arc; +use std::time::Duration; +use tokio_tungstenite::tungstenite::Message; +use tracing::{debug, error, info, warn}; + +use crate::car; +use crate::cbor::Cbor; +use crate::firehose::Stats; +use crate::indexer; + +/// XRPC path of the subscription. Kept as a constant so the URL builder +/// and its tests can't drift apart. +pub const SUBSCRIBE_PATH: &str = "/xrpc/com.atproto.sync.subscribeRepos"; + +/// Cursor row id — same single-row-table pattern as `jetstream_cursor`. +const CURSOR_ROW_ID: i32 = 1; + +/// How often the same sequence number may fail to apply before we give +/// up on it and move past. Without this a single poisonous frame (a +/// record the indexer chokes on) would pin the cursor and replay +/// forever on every reconnect. +const MAX_APPLY_ATTEMPTS: u32 = 3; + +// -- URL ------------------------------------------------------------------- + +/// Build the subscription URL from the PDS base URL. +/// +/// `AppConfig::pds_base_url()` yields an *HTTP* URL (`PDS_INTERNAL_URL`, +/// else `PDS_PUBLIC_URL`), because that is what the handle resolver and +/// the signing-key fetch need. A WebSocket needs the matching ws +/// scheme, and getting that mapping wrong is the classic way to end up +/// with a consumer that silently never connects: +/// +/// | in | out | +/// |----------|--------| +/// | `http` | `ws` | +/// | `https` | `wss` | +/// | `ws`/`wss` | unchanged (an operator may configure it directly) | +/// | no scheme | `ws` (dev default, matching `http`) | +/// +/// A trailing slash on the base URL is dropped so we never emit a +/// double slash before `/xrpc`. +pub fn subscribe_url(pds_base_url: &str, cursor: Option) -> String { + let trimmed = pds_base_url.trim().trim_end_matches('/'); + let (scheme, authority) = match trimmed.split_once("://") { + Some(("http", rest)) => ("ws", rest), + Some(("https", rest)) => ("wss", rest), + Some(("ws", rest)) => ("ws", rest), + Some(("wss", rest)) => ("wss", rest), + // Anything else (a scheme we don't know) is passed through + // untouched rather than mangled — connecting will fail loudly, + // which beats connecting to the wrong place quietly. + Some((other, rest)) => (other, rest), + None => ("ws", trimmed), + }; + let base = format!("{scheme}://{authority}{SUBSCRIBE_PATH}"); + match cursor { + // `cursor=0` is meaningful ("from the very beginning") but we + // only ever hold 0 to mean "no cursor yet", and asking for + // everything since the dawn of the repo on a fresh AppView is + // exactly what a full backfill should do — so it is sent. + Some(seq) if seq >= 0 => format!("{base}?cursor={seq}"), + _ => base, + } +} + +// -- cursor ---------------------------------------------------------------- + +/// Read the persisted PDS-firehose sequence number. `0` when no frame +/// has been processed yet, which the caller turns into "subscribe +/// without a cursor" (i.e. start from the PDS's current head) — see +/// [`PdsFirehose::run_forever`]. +pub async fn cursor_get(db: &PgPool) -> Result { + let row: Option<(i64,)> = + sqlx::query_as("SELECT cursor FROM pds_firehose_cursor WHERE id = $1") + .bind(CURSOR_ROW_ID) + .fetch_optional(db) + .await?; + Ok(row.map(|(c,)| c).unwrap_or(0)) +} + +/// Advance the persisted cursor, never backwards (`GREATEST`), exactly +/// like `indexer::cursor_advance` does for Jetstream. Frames can be +/// applied slightly out of order across a reconnect; the highest seq we +/// have durably processed is the one worth resuming from. +pub async fn cursor_advance(db: &PgPool, seq: i64) -> Result<()> { + sqlx::query( + r#"UPDATE pds_firehose_cursor + SET cursor = GREATEST(cursor, $1), updated_at = now() + WHERE id = $2"#, + ) + .bind(seq) + .bind(CURSOR_ROW_ID) + .execute(db) + .await?; + Ok(()) +} + +/// Force the cursor to a value, downwards included. +/// +/// Only used when the *server* tells us our cursor is unusable — +/// `#info OutdatedCursor` (we asked for something older than the PDS +/// still keeps) or an error frame such as `FutureCursor` (we asked for +/// something the PDS has never emitted, which is what a re-created PDS +/// database looks like). In both cases keeping the stale number would +/// mean reconnecting into the same rejection forever. +pub async fn cursor_reset(db: &PgPool, seq: i64) -> Result<()> { + sqlx::query( + r#"UPDATE pds_firehose_cursor + SET cursor = $1, updated_at = now() + WHERE id = $2"#, + ) + .bind(seq) + .bind(CURSOR_ROW_ID) + .execute(db) + .await?; + Ok(()) +} + +// -- frames ---------------------------------------------------------------- + +/// One decoded op out of a `#commit` frame. +#[derive(Debug, Clone, PartialEq)] +pub struct FrameOp { + /// `create` | `update` | `delete`. + pub action: String, + /// `"/"`. + pub path: String, + /// The record's CID. `None` for deletes. + pub cid: Option, +} + +impl FrameOp { + /// The collection NSID — everything before the first `/`. + pub fn collection(&self) -> Option<&str> { + let (collection, rkey) = self.path.split_once('/')?; + (!collection.is_empty() && !rkey.is_empty()).then_some(collection) + } + + /// The record key — everything after the first `/`. + pub fn rkey(&self) -> Option<&str> { + let (collection, rkey) = self.path.split_once('/')?; + (!collection.is_empty() && !rkey.is_empty()).then_some(rkey) + } +} + +/// A decoded `#commit` body. +#[derive(Debug, Clone, PartialEq)] +pub struct CommitFrame { + pub seq: i64, + pub rebase: bool, + pub too_big: bool, + pub repo: String, + pub commit: Option, + pub rev: String, + pub since: Option, + /// Raw CAR bytes. Parsed lazily by [`Self::records`] so a frame + /// with only deletes never pays for it. + pub blocks: Vec, + pub ops: Vec, + pub time: String, +} + +impl CommitFrame { + /// `time` as microseconds since the epoch, for the health/lag + /// counters. Falls back to "now" when the PDS sends something that + /// isn't RFC 3339 — a broken clock in a log line is better than a + /// dropped commit. + pub fn time_us(&self) -> i64 { + chrono::DateTime::parse_from_rfc3339(&self.time) + .map(|t| t.timestamp_micros()) + .unwrap_or_else(|_| chrono::Utc::now().timestamp_micros()) + } +} + +/// A decoded frame. +#[derive(Debug, Clone, PartialEq)] +pub enum Frame { + Commit(Box), + /// `#info` — the server telling us something about the stream + /// itself, most importantly `OutdatedCursor`. + Info { + name: String, + message: Option, + }, + /// `op: -1` — a terminal error frame; the server closes after it. + Error { + error: String, + message: Option, + }, + /// A message type we don't consume (`#identity`, `#account`, + /// `#handle`, …). Recorded so the log says what was skipped. + Other { t: String }, +} + +/// Decode one WebSocket binary message into a [`Frame`]. +pub fn decode_frame(bytes: &[u8]) -> Result { + let (header, body_start) = + crate::cbor::decode_at(bytes, 0).context("decoding frame header")?; + let (body, end) = + crate::cbor::decode_at(bytes, body_start).context("decoding frame body")?; + if end != bytes.len() { + // Not fatal for us — we have both values — but it means the + // sender wrote something we don't understand, so say so. + debug!( + trailing = bytes.len() - end, + "frame has trailing bytes after the body" + ); + } + + let op = header + .get("op") + .and_then(|v| v.as_i64()) + .ok_or_else(|| anyhow!("frame header has no `op`"))?; + + if op == -1 { + return Ok(Frame::Error { + error: body + .get("error") + .and_then(|v| v.as_str()) + .unwrap_or("UnknownError") + .to_string(), + message: body + .get("message") + .and_then(|v| v.as_str()) + .map(str::to_string), + }); + } + if op != 1 { + bail!("unsupported frame op {op}"); + } + + let t = header + .get("t") + .and_then(|v| v.as_str()) + .ok_or_else(|| anyhow!("regular frame header has no `t`"))?; + + match t { + "#commit" => Ok(Frame::Commit(Box::new(decode_commit_body(&body)?))), + "#info" => Ok(Frame::Info { + name: body + .get("name") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(), + message: body + .get("message") + .and_then(|v| v.as_str()) + .map(str::to_string), + }), + other => Ok(Frame::Other { + t: other.to_string(), + }), + } +} + +fn decode_commit_body(body: &Cbor) -> Result { + let seq = body + .get("seq") + .and_then(|v| v.as_i64()) + .ok_or_else(|| anyhow!("#commit body has no `seq`"))?; + let repo = body + .get("repo") + .and_then(|v| v.as_str()) + .ok_or_else(|| anyhow!("#commit body has no `repo`"))? + .to_string(); + + let ops = match body.get("ops") { + Some(Cbor::Array(items)) => items + .iter() + .map(decode_op) + .collect::>>() + .context("decoding #commit ops")?, + // An op-less commit is legal (a rebase, or a commit that only + // moved MST nodes). Treat a missing array the same way. + _ => Vec::new(), + }; + + Ok(CommitFrame { + seq, + rebase: body.get("rebase").and_then(|v| v.as_bool()).unwrap_or(false), + too_big: body.get("tooBig").and_then(|v| v.as_bool()).unwrap_or(false), + repo, + commit: body.get("commit").and_then(|v| v.as_cid()), + rev: body + .get("rev") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(), + since: body + .get("since") + .filter(|v| !v.is_null()) + .and_then(|v| v.as_str()) + .map(str::to_string), + blocks: body + .get("blocks") + .and_then(|v| v.as_bytes()) + .unwrap_or_default() + .to_vec(), + ops, + time: body + .get("time") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(), + }) +} + +fn decode_op(op: &Cbor) -> Result { + Ok(FrameOp { + action: op + .get("action") + .and_then(|v| v.as_str()) + .ok_or_else(|| anyhow!("op has no `action`"))? + .to_string(), + path: op + .get("path") + .and_then(|v| v.as_str()) + .ok_or_else(|| anyhow!("op has no `path`"))? + .to_string(), + cid: op.get("cid").filter(|v| !v.is_null()).and_then(|v| v.as_cid()), + }) +} + +// -- frame → indexer ------------------------------------------------------- + +/// Turn a `#commit` frame into the per-op events the indexer consumes. +/// +/// One event per op, each in the Jetstream *single-op* shape +/// (`commit.collection` + `commit.operation` + `commit.record`), because +/// a single frame may touch several collections while +/// [`indexer::apply_commit`] reads one collection per event. +/// +/// Ops we can't act on are skipped with a log line rather than failing +/// the frame — a frame is all-or-nothing for the cursor, and one +/// unreadable op must not block the commits behind it. +pub fn events_from_frame(frame: &CommitFrame) -> Result> { + // A frame carrying only deletes has no CAR at all — don't demand one. + if frame.blocks.is_empty() { + return Ok(build_events(frame, &HashMap::new())); + } + // The CAR owns the block bytes the map borrows, so it has to stay + // alive until `build_events` returns. + let car = car::parse(&frame.blocks).context("parsing #commit blocks CAR")?; + Ok(build_events(frame, &car.block_map())) +} + +fn build_events(frame: &CommitFrame, blocks: &HashMap) -> Vec { + let time_us = frame.time_us(); + let mut events = Vec::with_capacity(frame.ops.len()); + + for op in &frame.ops { + let (Some(collection), Some(rkey)) = (op.collection(), op.rkey()) else { + warn!(seq = frame.seq, path = %op.path, "skipping op with unusable path"); + continue; + }; + + // `update` is `create` as far as every write in `indexer` is + // concerned — they are all upserts keyed by URI / (did, subject). + // Mapping it here keeps `apply_commit` (written for Jetstream, + // which only ever says create/delete) untouched. + let action = match op.action.as_str() { + "create" | "update" => "create", + "delete" => "delete", + other => { + warn!(seq = frame.seq, action = %other, "skipping op with unknown action"); + continue; + } + }; + + let record = if action == "create" { + match op.cid.as_ref().and_then(|c| blocks.get(c)) { + Some(bytes) => match decode_record_block(bytes) { + Ok(v) => Some(v), + Err(e) => { + warn!( + seq = frame.seq, path = %op.path, error = %e, + "could not decode record block; skipping op" + ); + continue; + } + }, + None => { + // `tooBig` frames legitimately omit blocks: the PDS + // is telling us the commit was too large to inline + // and that a `getRepo` backfill is needed. Anything + // else is a PDS bug worth a louder line. + if frame.too_big { + debug!(seq = frame.seq, path = %op.path, + "tooBig frame omits the record block; skipping op"); + } else { + warn!(seq = frame.seq, path = %op.path, + "create op has no matching block in the CAR; skipping op"); + } + continue; + } + } + } else { + None + }; + + let mut commit = json!({ + "collection": collection, + "operation": action, + "rkey": rkey, + "path": op.path, + }); + if let Some(cid) = &op.cid { + commit["cid"] = json!(cid.to_string()); + } + if let Some(record) = record { + commit["record"] = record; + } + + events.push(JetstreamEvent { + did: frame.repo.clone(), + time_us, + kind: "commit".to_string(), + commit: Some(commit), + identity: None, + account: None, + }); + } + + events +} + +/// Decode one record block. +/// +/// Blocks are written by `ciborium::into_writer(&serde_json::Value)`, so +/// `ciborium` reading back into `serde_json::Value` is their exact +/// inverse — including the house convention that a CID inside a record +/// is a plain string rather than a tag-42 link. +fn decode_record_block(bytes: &[u8]) -> Result { + ciborium::from_reader::(bytes) + .map_err(|e| anyhow!("record block is not CBOR we can read: {e}")) +} + +/// Apply every op of a frame through the shared indexer path. +/// +/// Returns the number of events the indexer reported as applied. Errors +/// propagate: the caller keeps the cursor pinned so the frame is +/// replayed on the next connect. +pub async fn apply_frame(db: &PgPool, frame: &CommitFrame) -> Result { + if frame.rebase { + // A rebase rewrites history without changing the records we + // index. Nothing to do beyond noting it. + info!(seq = frame.seq, repo = %frame.repo, "firehose rebase frame"); + } + if frame.too_big { + warn!( + seq = frame.seq, repo = %frame.repo, + "firehose frame marked tooBig — its record blocks are not inlined; \ + affected ops are skipped and stay reachable only via the push path" + ); + } + + let events = events_from_frame(frame)?; + let mut applied = 0usize; + for ev in &events { + if indexer::apply_commit(db, ev).await? { + applied += 1; + } + } + Ok(applied) +} + +// -- consumer -------------------------------------------------------------- + +/// The long-lived WebSocket consumer. +pub struct PdsFirehose { + pub db: PgPool, + /// HTTP(S) base URL of the PDS — `AppConfig::pds_base_url()`. The + /// ws/wss mapping happens in [`subscribe_url`]. + pub pds_base_url: String, + pub stats: Arc, + /// Reconnect backoff ceiling in seconds. + pub max_backoff_secs: u64, +} + +impl PdsFirehose { + pub fn new(db: PgPool, pds_base_url: String, stats: Arc) -> Self { + Self { + db, + pds_base_url, + stats, + max_backoff_secs: 30, + } + } + + pub fn with_max_backoff_secs(mut self, secs: u64) -> Self { + self.max_backoff_secs = secs.max(1); + self + } + + /// Connect, consume, reconnect — forever. + /// + /// Backoff doubles from 1s to `max_backoff_secs`, the same shape as + /// [`at_firehose::JetstreamConsumer::run`]. This matters more here + /// than for Jetstream: until the PDS's `subscribeRepos` endpoint is + /// live, every attempt fails, and a tight loop would fill the log + /// and the PDS's accept queue. + pub async fn run_forever(self) { + let mut backoff_secs: u64 = 1; + loop { + match self.connect_and_consume().await { + Ok(()) => warn!("pds firehose stream ended, reconnecting"), + Err(e) => warn!(error = %format!("{e:#}"), "pds firehose connection failed"), + } + self.stats.pds_connected.store(false, Ordering::Relaxed); + debug!("reconnecting to the pds firehose in {backoff_secs}s"); + tokio::time::sleep(Duration::from_secs(backoff_secs)).await; + backoff_secs = (backoff_secs * 2).min(self.max_backoff_secs); + } + } + + /// One connection's lifetime. Returns `Ok(())` on a clean close. + async fn connect_and_consume(&self) -> Result<()> { + // Re-read the cursor on every connect rather than caching it in + // the struct: it is the durable record of what we processed, + // and re-reading means a reset written during the previous + // connection (OutdatedCursor / FutureCursor) is honoured here. + let stored = cursor_get(&self.db).await.unwrap_or(0); + let url = subscribe_url(&self.pds_base_url, (stored > 0).then_some(stored)); + + let (mut ws, _resp) = tokio_tungstenite::connect_async(&url) + .await + .with_context(|| format!("connecting to {url}"))?; + info!(url = %url, cursor = stored, "connected to the pds firehose"); + self.stats.pds_connected.store(true, Ordering::Relaxed); + + // Tracks a frame that keeps failing to apply, so one poisonous + // commit can't pin the cursor across reconnects forever. + let mut failing: Option<(i64, u32)> = None; + + while let Some(msg) = ws.next().await { + let msg = msg.context("reading from the pds firehose")?; + let bytes = match msg { + Message::Binary(b) => b, + // The protocol is binary-only; ping/pong are answered + // by tungstenite itself, and a Close ends the stream. + Message::Close(frame) => { + info!(?frame, "pds firehose closed by the server"); + return Ok(()); + } + Message::Text(t) => { + debug!(text = %t, "ignoring unexpected text frame"); + continue; + } + _ => continue, + }; + + let frame = match decode_frame(&bytes) { + Ok(f) => f, + Err(e) => { + // A frame we can't parse is skipped, not fatal: the + // cursor stays where it is, so if it mattered we + // will see it again after the next reconnect. + warn!(error = %format!("{e:#}"), len = bytes.len(), + "could not decode a pds firehose frame; skipping"); + continue; + } + }; + + match frame { + Frame::Commit(commit) => { + self.handle_commit(&commit, &mut failing).await?; + } + Frame::Info { name, message } => { + self.handle_info(&name, message.as_deref()).await?; + } + Frame::Error { error, message } => { + error!( + error = %error, + message = %message.unwrap_or_default(), + "pds firehose error frame" + ); + // `FutureCursor` means we asked for a seq the PDS + // has never emitted — the normal cause is a + // re-created PDS database while the AppView kept + // its cursor. Clearing it lets the next connect + // start from the current head instead of looping + // on the same rejection. + if error == "FutureCursor" { + warn!("resetting the pds firehose cursor after FutureCursor"); + cursor_reset(&self.db, 0).await?; + } + // The server closes the stream after an error + // frame; return so the caller backs off. + return Ok(()); + } + Frame::Other { t } => { + debug!(t = %t, "ignoring pds firehose frame type we don't consume"); + } + } + } + Ok(()) + } + + async fn handle_commit( + &self, + frame: &CommitFrame, + failing: &mut Option<(i64, u32)>, + ) -> Result<()> { + match apply_frame(&self.db, frame).await { + Ok(applied) => { + *failing = None; + self.stats.pds_frames_processed.fetch_add(1, Ordering::Relaxed); + self.stats + .pds_last_seq + .fetch_max(frame.seq, Ordering::Relaxed); + self.stats + .last_event_time_us + .fetch_max(frame.time_us(), Ordering::Relaxed); + // The cursor is flushed per frame rather than batched + // like the Jetstream one. This stream carries only our + // own users' commits — a handful per minute at most — + // so one small UPDATE per commit is cheaper than the + // machinery to avoid it, and it means a crash costs at + // most one replayed (idempotent) frame. + cursor_advance(&self.db, frame.seq).await?; + debug!( + seq = frame.seq, repo = %frame.repo, ops = frame.ops.len(), + applied, "applied a pds firehose commit" + ); + Ok(()) + } + Err(e) => { + let attempts = match failing { + Some((seq, n)) if *seq == frame.seq => { + *n += 1; + *n + } + _ => { + *failing = Some((frame.seq, 1)); + 1 + } + }; + if attempts >= MAX_APPLY_ATTEMPTS { + error!( + seq = frame.seq, repo = %frame.repo, attempts, + error = %format!("{e:#}"), + "giving up on a pds firehose frame after repeated failures; \ + advancing the cursor past it to unblock the stream" + ); + *failing = None; + cursor_advance(&self.db, frame.seq).await?; + return Ok(()); + } + warn!( + seq = frame.seq, repo = %frame.repo, attempts, + error = %format!("{e:#}"), + "failed to apply a pds firehose frame; cursor stays put and the \ + frame will be replayed after reconnect" + ); + // Drop the connection so the PDS replays from the last + // durable cursor. Returning Err takes us through the + // caller's backoff. + Err(e) + } + } + } + + /// `#info` frames. The only one defined today is `OutdatedCursor`: + /// we asked for a sequence the PDS no longer keeps, and it is + /// serving us from the oldest it still has. The stream continues — + /// this is informational, never fatal. + /// + /// The stored cursor is now meaningless, so it is cleared: leaving + /// it would make every future connect ask for the same too-old + /// value and collect the same `#info` again. The frames that follow + /// immediately re-populate it (their seqs are all above the pruned + /// window, so `GREATEST` accepts them). + /// + /// The trade-off: if the connection drops in the short window + /// between the `#info` and the first replayed frame, the next + /// connect starts live-only and the already-pruned backlog is not + /// re-requested. That backlog is, by definition, data the PDS was + /// prepared to drop — and the alternative (keeping a cursor the + /// server rejects) buys nothing else. + async fn handle_info(&self, name: &str, message: Option<&str>) -> Result<()> { + match name { + "OutdatedCursor" => { + warn!( + message = %message.unwrap_or_default(), + "pds firehose reports an outdated cursor; continuing from the \ + server's oldest available sequence and clearing the stored cursor" + ); + cursor_reset(&self.db, 0).await?; + } + other => { + info!(name = %other, message = %message.unwrap_or_default(), + "pds firehose #info"); + } + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::car::test_writer::CarWriter; + use at_crypto::cid::cid_for_cbor; + + // -- tiny DAG-CBOR writer, for building frames the way the PDS does -- + // + // Mirrors `pds-server/src/dag_cbor.rs`. It exists so the decoder is + // tested against bytes shaped like the real wire format instead of + // against its own output. + + fn 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 < 0x100 { + out.push(m | 24); + out.push(n as u8); + } else if n < 0x10000 { + out.push(m | 25); + out.extend_from_slice(&(n as u16).to_be_bytes()); + } else if n < 0x1_0000_0000 { + 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()); + } + } + + fn text(out: &mut Vec, s: &str) { + head(out, 3, s.len() as u64); + out.extend_from_slice(s.as_bytes()); + } + + fn bytes(out: &mut Vec, b: &[u8]) { + head(out, 2, b.len() as u64); + out.extend_from_slice(b); + } + + fn int(out: &mut Vec, n: i64) { + if n >= 0 { + head(out, 0, n as u64); + } else { + head(out, 1, (-1 - n) as u64); + } + } + + /// Spec-correct CID link: tag(42) + bytes(0x00 || cid) — what the + /// PDS's `dag_cbor::write_link` emits for frames. + fn link(out: &mut Vec, cid: &Cid) { + head(out, 6, 42); + let mut raw = vec![0x00]; + raw.extend_from_slice(&cid.to_bytes()); + bytes(out, &raw); + } + + fn regular_header(t: &str) -> Vec { + let mut v = Vec::new(); + head(&mut v, 5, 2); + text(&mut v, "op"); + int(&mut v, 1); + text(&mut v, "t"); + text(&mut v, t); + v + } + + struct OpSpec { + action: &'static str, + path: String, + cid: Option, + } + + /// Build a complete `#commit` frame the way the PDS will. + fn commit_frame(seq: i64, repo: &str, ops: &[OpSpec], car_bytes: &[u8]) -> Vec { + let commit_cid = cid_for_cbor(b"the commit block").unwrap(); + let mut v = regular_header("#commit"); + // map(11): seq, rebase, tooBig, repo, commit, rev, since, + // blocks, ops, blobs, time — the full #commit body. + head(&mut v, 5, 11); + text(&mut v, "seq"); + int(&mut v, seq); + text(&mut v, "rebase"); + v.push(0xF4); // false + text(&mut v, "tooBig"); + v.push(0xF4); + text(&mut v, "repo"); + text(&mut v, repo); + text(&mut v, "commit"); + link(&mut v, &commit_cid); + text(&mut v, "rev"); + text(&mut v, "3kabc"); + text(&mut v, "since"); + v.push(0xF6); // null + text(&mut v, "blocks"); + bytes(&mut v, car_bytes); + text(&mut v, "ops"); + head(&mut v, 4, ops.len() as u64); + for op in ops { + head(&mut v, 5, 3); + text(&mut v, "action"); + text(&mut v, op.action); + text(&mut v, "path"); + text(&mut v, &op.path); + text(&mut v, "cid"); + match &op.cid { + Some(c) => link(&mut v, c), + None => v.push(0xF6), + } + } + text(&mut v, "blobs"); + head(&mut v, 4, 0); + text(&mut v, "time"); + text(&mut v, "2026-09-10T12:00:00Z"); + v + } + + /// CBOR-encode a record the way `pds-server` writes record blocks. + fn record_block(record: &Value) -> (Cid, Vec) { + let mut buf = Vec::new(); + ciborium::into_writer(record, &mut buf).unwrap(); + let cid = cid_for_cbor(&buf).unwrap(); + (cid, buf) + } + + fn post_frame(seq: i64, did: &str, rkey: &str, text_body: &str) -> (Vec, Cid) { + let record = json!({ + "$type": "app.twi.post", + "text": text_body, + "createdAt": "2026-09-10T12:00:00Z", + }); + let (cid, block) = record_block(&record); + let mut w = CarWriter::new(); + w.append(cid, &block); + let car_bytes = w.finish(&[cid]); + let ops = vec![OpSpec { + action: "create", + path: format!("app.twi.post/{rkey}"), + cid: Some(cid), + }]; + (commit_frame(seq, did, &ops, &car_bytes), cid) + } + + // -- URL derivation --------------------------------------------------- + + #[test] + fn url_maps_http_to_ws_and_https_to_wss() { + assert_eq!( + subscribe_url("http://127.0.0.1:2583", None), + "ws://127.0.0.1:2583/xrpc/com.atproto.sync.subscribeRepos" + ); + assert_eq!( + subscribe_url("https://pds.example.com", None), + "wss://pds.example.com/xrpc/com.atproto.sync.subscribeRepos" + ); + } + + #[test] + fn url_keeps_an_explicit_ws_scheme_and_drops_trailing_slashes() { + assert_eq!( + subscribe_url("wss://pds.example.com/", None), + "wss://pds.example.com/xrpc/com.atproto.sync.subscribeRepos" + ); + assert_eq!( + subscribe_url("ws://pds:3000", None), + "ws://pds:3000/xrpc/com.atproto.sync.subscribeRepos" + ); + // Scheme-less config is the dev shorthand; ws matches http. + assert_eq!( + subscribe_url("127.0.0.1:2583", None), + "ws://127.0.0.1:2583/xrpc/com.atproto.sync.subscribeRepos" + ); + } + + #[test] + fn url_appends_the_cursor() { + assert_eq!( + subscribe_url("http://127.0.0.1:2583", Some(42)), + "ws://127.0.0.1:2583/xrpc/com.atproto.sync.subscribeRepos?cursor=42" + ); + // No cursor → no query string at all, which the PDS reads as + // "start from the current head". + assert!(!subscribe_url("http://127.0.0.1:2583", None).contains('?')); + } + + // -- frame decoding --------------------------------------------------- + + #[test] + fn decodes_a_commit_frame_with_a_record() { + let (frame_bytes, record_cid) = + post_frame(7, "did:plc:alice", "3krkey", "hello firehose"); + let frame = decode_frame(&frame_bytes).unwrap(); + let Frame::Commit(commit) = frame else { + panic!("expected a #commit frame, got {frame:?}"); + }; + assert_eq!(commit.seq, 7); + assert_eq!(commit.repo, "did:plc:alice"); + assert_eq!(commit.rev, "3kabc"); + assert_eq!(commit.since, None); + assert!(!commit.rebase && !commit.too_big); + assert!(commit.commit.is_some(), "tag-42 commit link must decode"); + assert_eq!(commit.ops.len(), 1); + assert_eq!(commit.ops[0].action, "create"); + assert_eq!(commit.ops[0].cid, Some(record_cid)); + assert_eq!(commit.ops[0].collection(), Some("app.twi.post")); + assert_eq!(commit.ops[0].rkey(), Some("3krkey")); + assert_eq!(commit.time_us(), 1_789_041_600_000_000); + } + + #[test] + fn decodes_a_delete_op_with_a_null_cid() { + let ops = vec![OpSpec { + action: "delete", + path: "app.twi.post/gone".to_string(), + cid: None, + }]; + let bytes = commit_frame(9, "did:plc:bob", &ops, &[]); + let Frame::Commit(commit) = decode_frame(&bytes).unwrap() else { + panic!("expected #commit"); + }; + assert_eq!(commit.ops[0].action, "delete"); + assert_eq!(commit.ops[0].cid, None); + assert!(commit.blocks.is_empty()); + } + + #[test] + fn decodes_an_info_frame() { + let mut v = regular_header("#info"); + head(&mut v, 5, 2); + text(&mut v, "name"); + text(&mut v, "OutdatedCursor"); + text(&mut v, "message"); + text(&mut v, "cursor too old"); + match decode_frame(&v).unwrap() { + Frame::Info { name, message } => { + assert_eq!(name, "OutdatedCursor"); + assert_eq!(message.as_deref(), Some("cursor too old")); + } + other => panic!("expected #info, got {other:?}"), + } + } + + #[test] + fn decodes_an_error_frame() { + // header {"op": -1}, body {"error": ..., "message": ...} + let mut v = Vec::new(); + head(&mut v, 5, 1); + text(&mut v, "op"); + int(&mut v, -1); + head(&mut v, 5, 2); + text(&mut v, "error"); + text(&mut v, "FutureCursor"); + text(&mut v, "message"); + text(&mut v, "cursor in the future"); + match decode_frame(&v).unwrap() { + Frame::Error { error, message } => { + assert_eq!(error, "FutureCursor"); + assert_eq!(message.as_deref(), Some("cursor in the future")); + } + other => panic!("expected an error frame, got {other:?}"), + } + } + + #[test] + fn decodes_an_unknown_message_type_without_failing() { + let mut v = regular_header("#identity"); + head(&mut v, 5, 1); + text(&mut v, "did"); + text(&mut v, "did:plc:x"); + match decode_frame(&v).unwrap() { + Frame::Other { t } => assert_eq!(t, "#identity"), + other => panic!("expected Other, got {other:?}"), + } + } + + #[test] + fn rejects_malformed_frames() { + // Truncated: header only. + assert!(decode_frame(®ular_header("#commit")).is_err()); + // Header without `op`. + let mut v = Vec::new(); + head(&mut v, 5, 1); + text(&mut v, "t"); + text(&mut v, "#commit"); + head(&mut v, 5, 0); + assert!(decode_frame(&v).is_err()); + // Empty message. + assert!(decode_frame(&[]).is_err()); + } + + #[test] + fn accepts_a_bare_cid_link_without_the_identity_prefix() { + // The PDS's CAR header writer omits the 0x00 prefix. If a future + // frame writer does the same, the AppView must still decode it. + let cid = cid_for_cbor(b"x").unwrap(); + let mut v = regular_header("#commit"); + head(&mut v, 5, 3); + text(&mut v, "seq"); + int(&mut v, 1); + text(&mut v, "repo"); + text(&mut v, "did:plc:a"); + text(&mut v, "commit"); + head(&mut v, 6, 42); + bytes(&mut v, &cid.to_bytes()); // no 0x00 + let Frame::Commit(commit) = decode_frame(&v).unwrap() else { + panic!("expected #commit"); + }; + assert_eq!(commit.commit, Some(cid)); + } + + // -- frame → events --------------------------------------------------- + + #[test] + fn builds_a_create_event_with_the_record_from_the_car() { + let (bytes, cid) = post_frame(3, "did:plc:alice", "rk1", "from the CAR"); + let Frame::Commit(commit) = decode_frame(&bytes).unwrap() else { + panic!("expected #commit"); + }; + let events = events_from_frame(&commit).unwrap(); + assert_eq!(events.len(), 1); + let ev = &events[0]; + assert_eq!(ev.did, "did:plc:alice"); + assert_eq!(ev.kind, "commit"); + let c = ev.commit.as_ref().unwrap(); + assert_eq!(c["collection"], "app.twi.post"); + assert_eq!(c["operation"], "create"); + assert_eq!(c["rkey"], "rk1"); + assert_eq!(c["cid"], cid.to_string()); + assert_eq!(c["record"]["text"], "from the CAR"); + // And the indexer's own parser agrees with the shape we built. + assert_eq!( + indexer::extract_collection(c).as_deref(), + Some("app.twi.post") + ); + let ops = indexer::commit_op_from_jetstream_value(c); + assert_eq!(ops.len(), 1); + assert_eq!(ops[0].action, "create"); + assert_eq!(ops[0].rkey.as_deref(), Some("rk1")); + } + + #[test] + fn maps_update_to_create_because_every_write_is_an_upsert() { + let record = json!({"text": "edited", "createdAt": "2026-09-10T12:00:00Z"}); + let (cid, block) = record_block(&record); + let mut w = CarWriter::new(); + w.append(cid, &block); + let ops = vec![OpSpec { + action: "update", + path: "app.twi.post/rk".to_string(), + cid: Some(cid), + }]; + let bytes = commit_frame(4, "did:plc:alice", &ops, &w.finish(&[cid])); + let Frame::Commit(commit) = decode_frame(&bytes).unwrap() else { + panic!("expected #commit"); + }; + let events = events_from_frame(&commit).unwrap(); + assert_eq!(events[0].commit.as_ref().unwrap()["operation"], "create"); + } + + #[test] + fn builds_a_delete_event_without_a_record() { + let ops = vec![OpSpec { + action: "delete", + path: "app.bsky.feed.like/rk".to_string(), + cid: None, + }]; + let bytes = commit_frame(5, "did:plc:alice", &ops, &[]); + let Frame::Commit(commit) = decode_frame(&bytes).unwrap() else { + panic!("expected #commit"); + }; + let events = events_from_frame(&commit).unwrap(); + let c = events[0].commit.as_ref().unwrap(); + assert_eq!(c["operation"], "delete"); + assert!(c.get("record").is_none()); + assert!(c.get("cid").is_none()); + } + + #[test] + fn one_frame_with_several_collections_yields_one_event_each() { + let post = json!({"text": "p", "createdAt": "2026-09-10T12:00:00Z"}); + let like = json!({ + "subject": {"uri": "at://did:plc:b/app.twi.post/x", "cid": "bafycid"}, + "createdAt": "2026-09-10T12:00:00Z", + }); + let (post_cid, post_block) = record_block(&post); + let (like_cid, like_block) = record_block(&like); + let mut w = CarWriter::new(); + w.append(post_cid, &post_block); + w.append(like_cid, &like_block); + let ops = vec![ + OpSpec { + action: "create", + path: "app.twi.post/p1".to_string(), + cid: Some(post_cid), + }, + OpSpec { + action: "create", + path: "app.bsky.feed.like/l1".to_string(), + cid: Some(like_cid), + }, + ]; + let bytes = commit_frame(6, "did:plc:alice", &ops, &w.finish(&[post_cid])); + let Frame::Commit(commit) = decode_frame(&bytes).unwrap() else { + panic!("expected #commit"); + }; + let events = events_from_frame(&commit).unwrap(); + assert_eq!(events.len(), 2); + assert_eq!( + events[0].commit.as_ref().unwrap()["collection"], + "app.twi.post" + ); + assert_eq!( + events[1].commit.as_ref().unwrap()["collection"], + "app.bsky.feed.like" + ); + } + + #[test] + fn skips_ops_whose_block_is_missing_instead_of_failing_the_frame() { + // Two creates, only one block in the CAR: the frame must still + // produce the event it can, so one bad op doesn't stall the + // cursor behind it. + let good = json!({"text": "good", "createdAt": "2026-09-10T12:00:00Z"}); + let (good_cid, good_block) = record_block(&good); + let orphan_cid = cid_for_cbor(b"never written").unwrap(); + let mut w = CarWriter::new(); + w.append(good_cid, &good_block); + let ops = vec![ + OpSpec { + action: "create", + path: "app.twi.post/orphan".to_string(), + cid: Some(orphan_cid), + }, + OpSpec { + action: "create", + path: "app.twi.post/good".to_string(), + cid: Some(good_cid), + }, + ]; + let bytes = commit_frame(8, "did:plc:alice", &ops, &w.finish(&[good_cid])); + let Frame::Commit(commit) = decode_frame(&bytes).unwrap() else { + panic!("expected #commit"); + }; + let events = events_from_frame(&commit).unwrap(); + assert_eq!(events.len(), 1); + assert_eq!(events[0].commit.as_ref().unwrap()["rkey"], "good"); + } + + #[test] + fn skips_ops_with_an_unusable_path() { + let ops = vec![ + OpSpec { + action: "create", + path: "no-slash".to_string(), + cid: None, + }, + OpSpec { + action: "delete", + path: "app.twi.post/".to_string(), + cid: None, + }, + ]; + let bytes = commit_frame(10, "did:plc:alice", &ops, &[]); + let Frame::Commit(commit) = decode_frame(&bytes).unwrap() else { + panic!("expected #commit"); + }; + assert!(events_from_frame(&commit).unwrap().is_empty()); + } + + // -- cursor persistence ----------------------------------------------- + + /// Same fail-open harness the indexer's own DB tests use: without a + /// reachable `DATABASE_URL_APPVIEW` the test skips. + async fn try_test_db() -> Option { + let url = std::env::var("DATABASE_URL_APPVIEW").ok()?; + match tokio::time::timeout(Duration::from_secs(2), PgPool::connect(&url)).await { + Ok(Ok(pool)) => match sqlx::migrate!("../../migrations/appview").run(&pool).await { + Ok(()) => Some(pool), + Err(_) => None, + }, + _ => None, + } + } + + #[tokio::test] + async fn cursor_survives_and_never_moves_backwards() { + let Some(db) = try_test_db().await else { + eprintln!("DATABASE_URL_APPVIEW unset/unreachable — skipping cursor test"); + return; + }; + // The suite shares one database, so restore whatever was there. + let original = cursor_get(&db).await.unwrap(); + + cursor_reset(&db, 0).await.unwrap(); + assert_eq!(cursor_get(&db).await.unwrap(), 0); + + cursor_advance(&db, 100).await.unwrap(); + assert_eq!(cursor_get(&db).await.unwrap(), 100); + + // A lower value must not win — that is what makes a replayed + // frame after a reconnect harmless. + cursor_advance(&db, 50).await.unwrap(); + assert_eq!(cursor_get(&db).await.unwrap(), 100); + + cursor_advance(&db, 101).await.unwrap(); + assert_eq!(cursor_get(&db).await.unwrap(), 101); + + // The reset path (OutdatedCursor / FutureCursor) may go down. + cursor_reset(&db, 0).await.unwrap(); + assert_eq!(cursor_get(&db).await.unwrap(), 0); + + cursor_reset(&db, original).await.unwrap(); + } + + #[tokio::test] + async fn applying_the_same_frame_twice_is_idempotent() { + let Some(db) = try_test_db().await else { + eprintln!("DATABASE_URL_APPVIEW unset/unreachable — skipping idempotency test"); + return; + }; + let did = format!("did:plc:fhtest{}", uuid::Uuid::new_v4().simple()); + let rkey = "rk1"; + let uri = format!("at://{did}/app.twi.post/{rkey}"); + + let (bytes, _) = post_frame(1, &did, rkey, "idempotent"); + let Frame::Commit(commit) = decode_frame(&bytes).unwrap() else { + panic!("expected #commit"); + }; + + // Apply the same frame three times — the push path, then the + // firehose, then a post-restart replay. + for _ in 0..3 { + apply_frame(&db, &commit).await.unwrap(); + } + + let count: i64 = sqlx::query_scalar("SELECT count(*) FROM posts WHERE uri = $1") + .bind(&uri) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(count, 1, "a replayed frame must not duplicate the post"); + + sqlx::query("DELETE FROM posts WHERE uri = $1") + .bind(&uri) + .execute(&db) + .await + .unwrap(); + } + + #[tokio::test] + async fn a_replayed_like_frame_keeps_the_counter_and_notification_at_one() { + let Some(db) = try_test_db().await else { + eprintln!("DATABASE_URL_APPVIEW unset/unreachable — skipping like idempotency test"); + return; + }; + let author = format!("did:plc:fhauth{}", uuid::Uuid::new_v4().simple()); + let liker = format!("did:plc:fhlike{}", uuid::Uuid::new_v4().simple()); + let post_rkey = "p1"; + let post_uri = format!("at://{author}/app.twi.post/{post_rkey}"); + + // Seed the post through the firehose path too. + let (post_bytes, post_cid) = post_frame(1, &author, post_rkey, "like me"); + let Frame::Commit(post_commit) = decode_frame(&post_bytes).unwrap() else { + panic!("expected #commit"); + }; + apply_frame(&db, &post_commit).await.unwrap(); + + // A like frame from a *different* DID, so a notification is + // warranted (self-likes are filtered by `should_notify`). + let like = json!({ + "subject": {"uri": post_uri, "cid": post_cid.to_string()}, + "createdAt": "2026-09-10T12:00:00Z", + }); + let (like_cid, like_block) = record_block(&like); + let mut w = CarWriter::new(); + w.append(like_cid, &like_block); + let ops = vec![OpSpec { + action: "create", + path: "app.bsky.feed.like/l1".to_string(), + cid: Some(like_cid), + }]; + let like_bytes = commit_frame(2, &liker, &ops, &w.finish(&[like_cid])); + let Frame::Commit(like_commit) = decode_frame(&like_bytes).unwrap() else { + panic!("expected #commit"); + }; + + for _ in 0..3 { + apply_frame(&db, &like_commit).await.unwrap(); + } + + let likes: i64 = sqlx::query_scalar("SELECT count(*) FROM likes WHERE post_uri = $1") + .bind(&post_uri) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(likes, 1, "replay must not insert a second like row"); + + let like_count: i64 = sqlx::query_scalar("SELECT like_count FROM posts WHERE uri = $1") + .bind(&post_uri) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(like_count, 1, "the denormalized counter must be bumped once"); + + let notifications: i64 = sqlx::query_scalar( + "SELECT count(*) FROM notifications \ + WHERE recipient_did = $1 AND author_did = $2 AND kind = 'like'", + ) + .bind(&author) + .bind(&liker) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(notifications, 1, "the dedupe index must hold on replay"); + + sqlx::query("DELETE FROM notifications WHERE recipient_did = $1") + .bind(&author) + .execute(&db) + .await + .unwrap(); + sqlx::query("DELETE FROM likes WHERE post_uri = $1") + .bind(&post_uri) + .execute(&db) + .await + .unwrap(); + sqlx::query("DELETE FROM posts WHERE uri = $1") + .bind(&post_uri) + .execute(&db) + .await + .unwrap(); + } +} diff --git a/crates/appview/src/routes.rs b/crates/appview/src/routes.rs index 0b56455..998f0cb 100644 --- a/crates/appview/src/routes.rs +++ b/crates/appview/src/routes.rs @@ -1538,6 +1538,16 @@ async fn actor_list( // -- healthz ---------------------------------------------------------------- +/// Liveness probe. +/// +/// Both ingest streams report separately. `jetstream_connected` is the +/// public network's view; `pds_firehose_connected` is the local PDS's +/// guaranteed delivery path for our own users' records. A deployment +/// can be perfectly healthy for reads with the first one down, but a +/// `pds_firehose_enabled: true, pds_firehose_connected: false` pair +/// means local commits are riding on the best-effort push alone — which +/// is exactly the state an operator wants to see in a probe rather than +/// discover from a missing post. async fn healthz(State(state): State) -> impl IntoResponse { let stats = &state.stats; Json(json!({ @@ -1545,6 +1555,10 @@ async fn healthz(State(state): State) -> impl IntoResponse { "lag_ms": stats.lag_ms(), "events_processed": stats.events_processed(), "jetstream_connected": stats.jetstream_connected(), + "pds_firehose_enabled": state.cfg.pds_firehose_enabled, + "pds_firehose_connected": stats.pds_connected(), + "pds_firehose_frames": stats.pds_frames_processed(), + "pds_firehose_seq": stats.pds_last_seq(), })) } diff --git a/crates/appview/tests/pds_firehose_integration.rs b/crates/appview/tests/pds_firehose_integration.rs new file mode 100644 index 0000000..e7cde16 --- /dev/null +++ b/crates/appview/tests/pds_firehose_integration.rs @@ -0,0 +1,444 @@ +//! End-to-end tests for the local PDS firehose consumer. +//! +//! These are the only tests that put *real* PDS bytes through +//! [`appview::pds_firehose`]: everything else in the module's own +//! `#[cfg(test)]` section builds frames from a hand-written DAG-CBOR +//! encoder, which proves the decoder matches our reading of the +//! contract but not that the PDS writes what we think it writes. +//! +//! Fail-open, like every other suite in this directory. Each test +//! prints a notice and returns successfully when a precondition is +//! missing: +//! +//! - the PDS isn't running on `:2583`; +//! - `DATABASE_URL_APPVIEW` is unset or the database is unreachable; +//! - **`com.atproto.sync.subscribeRepos` does not exist yet.** The +//! endpoint is being built in `crates/pds-server` in parallel with +//! this consumer. Until it lands, the WebSocket upgrade fails and +//! these tests skip with a message saying so — they are not proof of +//! anything while that line appears in the output. +//! +//! What they cover once the endpoint is live: +//! +//! - `frame_from_the_local_pds_indexes_a_post` — subscribe, create a +//! record over `com.atproto.repo.createRecord`, and drive the frame +//! the PDS emits through the real decode → CAR → indexer path, +//! asserting the row lands in `posts`. +//! - `replaying_the_same_frame_changes_nothing` — the same frame +//! applied twice leaves exactly one row, which is what makes the +//! overlap with the `/internal/ingest-commit` push safe. +//! - `car_reader_parses_a_real_repo_export` — the AppView's CAR reader +//! against a CAR the PDS's *writer* produced (`getRepo`). +//! - `healthz_reports_the_firehose_state` — the running AppView's +//! probe carries the new fields. + +use futures::StreamExt; +use serde_json::{json, Value}; +use sqlx::PgPool; +use std::time::Duration; +use tokio_tungstenite::tungstenite::Message; + +use appview::pds_firehose::{self, Frame}; + +const PDS_URL: &str = "http://127.0.0.1:2583"; +const PDS_WS: &str = "ws://127.0.0.1:2583"; + +fn appview_url() -> String { + std::env::var("APPVIEW_TEST_URL").unwrap_or_else(|_| "http://127.0.0.1:2584".to_string()) +} + +fn client() -> reqwest::Client { + reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build() + .unwrap() +} + +async fn service_up(c: &reqwest::Client, base: &str) -> bool { + for _ in 0..12 { + if let Ok(r) = c.get(format!("{base}/healthz")).send().await { + if r.status().is_success() { + return true; + } + } + tokio::time::sleep(Duration::from_millis(250)).await; + } + false +} + +async fn appview_db() -> Option { + let url = std::env::var("DATABASE_URL_APPVIEW").ok()?; + match tokio::time::timeout(Duration::from_secs(2), PgPool::connect(&url)).await { + Ok(Ok(pool)) => Some(pool), + _ => None, + } +} + +struct Account { + did: String, + access_jwt: String, +} + +async fn create_account(c: &reqwest::Client) -> Option { + let handle = format!("fh_{}.maarcadetweet.local", uuid::Uuid::new_v4().simple()); + let r: Value = c + .post(format!("{PDS_URL}/xrpc/com.atproto.server.createAccount")) + .json(&json!({ "handle": handle, "password": "hunter2hunter2" })) + .send() + .await + .ok()? + .json() + .await + .ok()?; + Some(Account { + did: r["did"].as_str()?.to_string(), + access_jwt: r["access_jwt"].as_str()?.to_string(), + }) +} + +async fn create_post(c: &reqwest::Client, acc: &Account, text: &str) -> Option { + let r: Value = c + .post(format!("{PDS_URL}/xrpc/com.atproto.repo.createRecord")) + .bearer_auth(&acc.access_jwt) + .json(&json!({ + "repo": acc.did, + "collection": "app.twi.post", + "record": { "text": text, "createdAt": "2026-09-10T12:00:00Z" }, + })) + .send() + .await + .ok()? + .json() + .await + .ok()?; + r["uri"].as_str().map(str::to_string) +} + +type Ws = tokio_tungstenite::WebSocketStream< + tokio_tungstenite::MaybeTlsStream, +>; + +/// Subscribe to the PDS firehose, or `None` if the endpoint isn't there +/// yet (see the module docs). +async fn subscribe() -> Option { + let url = pds_firehose::subscribe_url(PDS_WS, None); + match tokio::time::timeout(Duration::from_secs(5), tokio_tungstenite::connect_async(&url)).await + { + Ok(Ok((ws, _))) => Some(ws), + Ok(Err(e)) => { + eprintln!( + "cannot subscribe to {url}: {e} — the PDS endpoint com.atproto.sync.\ + subscribeRepos is probably not implemented yet; skipping" + ); + None + } + Err(_) => { + eprintln!("timed out connecting to {url}; skipping"); + None + } + } +} + +/// Read frames until one is a `#commit` for `did`, or the deadline +/// passes. `#info` frames along the way are tolerated (a fresh +/// subscription may legitimately be told its cursor is outdated). +async fn next_commit_for( + ws: &mut Ws, + did: &str, + timeout: Duration, +) -> Option { + let deadline = tokio::time::Instant::now() + timeout; + loop { + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + if remaining.is_zero() { + return None; + } + let msg = match tokio::time::timeout(remaining, ws.next()).await { + Ok(Some(Ok(m))) => m, + Ok(Some(Err(e))) => { + eprintln!("firehose read error: {e}"); + return None; + } + Ok(None) | Err(_) => return None, + }; + let Message::Binary(bytes) = msg else { continue }; + match pds_firehose::decode_frame(&bytes) { + Ok(Frame::Commit(commit)) if commit.repo == did => return Some(*commit), + Ok(Frame::Commit(_)) => continue, + Ok(Frame::Info { name, .. }) => { + eprintln!("firehose #info: {name}"); + continue; + } + Ok(Frame::Error { error, message }) => { + eprintln!("firehose error frame: {error} {message:?}"); + return None; + } + Ok(Frame::Other { .. }) => continue, + Err(e) => { + // A frame we cannot decode is a contract failure worth + // failing the test over — but only once we know the + // endpoint exists, which we do by this point. + panic!("could not decode a real PDS firehose frame: {e:#}"); + } + } + } +} + +/// Everything a live test needs, or `None` with a printed reason. +async fn ready() -> Option<(reqwest::Client, PgPool, Account, Ws)> { + let c = client(); + if !service_up(&c, PDS_URL).await { + eprintln!("pds not running on {PDS_URL}, skipping"); + return None; + } + let Some(db) = appview_db().await else { + eprintln!("DATABASE_URL_APPVIEW unset or unreachable, skipping"); + return None; + }; + let Some(acc) = create_account(&c).await else { + eprintln!("could not create a PDS account, skipping"); + return None; + }; + // Subscribe *before* writing anything, so the commit we are about + // to make is guaranteed to fall inside the subscription window. + let ws = subscribe().await?; + Some((c, db, acc, ws)) +} + +#[tokio::test] +async fn frame_from_the_local_pds_indexes_a_post() { + let Some((c, db, acc, mut ws)) = ready().await else { + return; + }; + let text = format!("firehose e2e {}", uuid::Uuid::new_v4().simple()); + let Some(uri) = create_post(&c, &acc, &text).await else { + eprintln!("createRecord failed, skipping"); + return; + }; + + let Some(commit) = next_commit_for(&mut ws, &acc.did, Duration::from_secs(15)).await else { + eprintln!("no #commit frame for {} arrived in time, skipping", acc.did); + return; + }; + + // The frame itself must carry what the contract promises. + assert!(commit.seq > 0, "seq must be a positive sequence number"); + assert_eq!(commit.repo, acc.did); + assert!(!commit.rev.is_empty(), "commit frames carry a rev"); + assert!( + !commit.blocks.is_empty(), + "a create commit must inline its record block" + ); + let create = commit + .ops + .iter() + .find(|op| op.action == "create" && op.collection() == Some("app.twi.post")) + .expect("the frame must contain the post create op"); + assert!(create.cid.is_some(), "a create op carries the record CID"); + + // The blocks field must be a CAR our reader understands, and the + // op's CID must resolve inside it. + let car = appview::car::parse(&commit.blocks).expect("blocks must be a readable CAR v1"); + assert!( + car.block_map().contains_key(&create.cid.unwrap()), + "the record block must be present in the CAR" + ); + + // And the whole path — frame → CAR → indexer — must land the row. + let events = pds_firehose::events_from_frame(&commit).expect("events"); + assert!( + events + .iter() + .any(|e| e.commit.as_ref().unwrap()["record"]["text"] == json!(text)), + "the decoded record must carry the text we posted" + ); + + // Remove whatever the AppView's own push path already wrote, so the + // assertion below is about *this* code applying *this* frame. + sqlx::query("DELETE FROM posts WHERE uri = $1") + .bind(&uri) + .execute(&db) + .await + .unwrap(); + + pds_firehose::apply_frame(&db, &commit) + .await + .expect("apply_frame"); + + let stored: Option = sqlx::query_scalar("SELECT text FROM posts WHERE uri = $1") + .bind(&uri) + .fetch_optional(&db) + .await + .unwrap(); + assert_eq!( + stored.as_deref(), + Some(text.as_str()), + "the firehose frame must index the post at {uri}" + ); + + sqlx::query("DELETE FROM posts WHERE uri = $1") + .bind(&uri) + .execute(&db) + .await + .unwrap(); +} + +#[tokio::test] +async fn replaying_the_same_frame_changes_nothing() { + let Some((c, db, acc, mut ws)) = ready().await else { + return; + }; + let text = format!("firehose replay {}", uuid::Uuid::new_v4().simple()); + let Some(uri) = create_post(&c, &acc, &text).await else { + eprintln!("createRecord failed, skipping"); + return; + }; + let Some(commit) = next_commit_for(&mut ws, &acc.did, Duration::from_secs(15)).await else { + eprintln!("no #commit frame for {} arrived in time, skipping", acc.did); + return; + }; + + // Three applications: the push already ran, then the firehose, then + // a post-restart replay of the same seq. + for _ in 0..3 { + pds_firehose::apply_frame(&db, &commit) + .await + .expect("apply_frame"); + } + + let rows: i64 = sqlx::query_scalar("SELECT count(*) FROM posts WHERE uri = $1") + .bind(&uri) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(rows, 1, "replay must not duplicate {uri}"); + + sqlx::query("DELETE FROM posts WHERE uri = $1") + .bind(&uri) + .execute(&db) + .await + .unwrap(); +} + +#[tokio::test] +async fn car_reader_parses_a_real_repo_export() { + // This one needs no firehose: `getRepo` has always served a CAR + // produced by the PDS's own writer, which is exactly the encoder + // the firehose's `blocks` field reuses. + let c = client(); + if !service_up(&c, PDS_URL).await { + eprintln!("pds not running on {PDS_URL}, skipping"); + return; + } + let Some(acc) = create_account(&c).await else { + eprintln!("could not create a PDS account, skipping"); + return; + }; + if create_post(&c, &acc, "car reader fixture").await.is_none() { + eprintln!("createRecord failed, skipping"); + return; + } + + let resp = c + .get(format!("{PDS_URL}/xrpc/com.atproto.sync.getRepo")) + .query(&[("did", acc.did.as_str())]) + .send() + .await + .unwrap(); + if !resp.status().is_success() { + eprintln!("getRepo returned {}, skipping", resp.status()); + return; + } + let bytes = resp.bytes().await.unwrap(); + let car = appview::car::parse(&bytes).expect("getRepo must return a readable CAR v1"); + assert_eq!(car.header.version, 1); + assert!( + !car.blocks.is_empty(), + "a repo with one record has blocks (commit + MST + record)" + ); + // Every block must hash to the CID the file declares — the strongest + // available statement that the reader's section framing is right. + appview::car::verify_block_cids(&car).expect("block CIDs must verify"); +} + +#[tokio::test] +async fn healthz_reports_the_firehose_state() { + let c = client(); + if !service_up(&c, &appview_url()).await { + eprintln!("appview not running, skipping"); + return; + } + let body: Value = c + .get(format!("{}/healthz", appview_url())) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + + // A binary built before this feature has none of these keys; say so + // rather than failing, because "restart the AppView" is the fix. + let Some(enabled) = body.get("pds_firehose_enabled").and_then(Value::as_bool) else { + eprintln!( + "the running AppView predates the PDS firehose (no pds_firehose_enabled \ + in /healthz) — rebuild and restart it; skipping" + ); + return; + }; + assert!( + body.get("pds_firehose_connected") + .and_then(Value::as_bool) + .is_some(), + "/healthz must report pds_firehose_connected: {body}" + ); + assert!( + body.get("pds_firehose_seq").and_then(Value::as_i64).is_some(), + "/healthz must report pds_firehose_seq: {body}" + ); + + if !enabled { + eprintln!("PDS_FIREHOSE_ENABLED=false on the running AppView; nothing more to check"); + return; + } + if !body["pds_firehose_connected"].as_bool().unwrap() { + eprintln!( + "the AppView is not connected to the PDS firehose — expected while \ + com.atproto.sync.subscribeRepos is still being implemented; skipping \ + the live-consumption check" + ); + return; + } + + // Connected: a new record must move the sequence number the AppView + // reports, which is the end-to-end proof that the *service* (not + // just this test process) consumes the stream. + if !service_up(&c, PDS_URL).await { + eprintln!("pds not running, skipping the live-consumption check"); + return; + } + let Some(acc) = create_account(&c).await else { + eprintln!("could not create a PDS account, skipping"); + return; + }; + let before = body["pds_firehose_seq"].as_i64().unwrap_or(0); + if create_post(&c, &acc, "healthz seq probe").await.is_none() { + eprintln!("createRecord failed, skipping"); + return; + } + for _ in 0..40 { + tokio::time::sleep(Duration::from_millis(250)).await; + let now: Value = c + .get(format!("{}/healthz", appview_url())) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + if now["pds_firehose_seq"].as_i64().unwrap_or(0) > before { + return; // consumed + } + } + panic!("the AppView reports the firehose connected but its seq never advanced"); +} diff --git a/crates/at-shared/src/config.rs b/crates/at-shared/src/config.rs index 4d0c1f2..4316e0e 100644 --- a/crates/at-shared/src/config.rs +++ b/crates/at-shared/src/config.rs @@ -18,6 +18,21 @@ fn default_auth_required() -> bool { true } +/// Default for `PDS_FIREHOSE_ENABLED`. +/// +/// `true` — the AppView consumes the local PDS's +/// `com.atproto.sync.subscribeRepos` stream. That stream is the only +/// *guaranteed* path for a local user's own records: the fast +/// `POST /internal/ingest-commit` push is best effort, and the public +/// Jetstream never sees this PDS, so a lost push means a permanently +/// missing post. Defaulting to on means an operator who never heard of +/// the variable gets the durable behaviour; switching it off is the +/// explicit choice (e.g. a PDS too old to serve the endpoint, or a +/// second AppView instance that should not double-index). +fn default_pds_firehose_enabled() -> bool { + true +} + #[derive(Debug, Clone, Deserialize)] pub struct AppConfig { pub pds_host: String, @@ -78,6 +93,12 @@ pub struct AppConfig { /// `APPVIEW_CORS_ORIGINS=tauri://localhost,http://127.0.0.1:1430` #[serde(default)] pub appview_cors_origins: Vec, + /// Whether the AppView subscribes to the local PDS firehose + /// (`com.atproto.sync.subscribeRepos` on + /// [`AppConfig::pds_base_url`]). Default `true` — see + /// [`default_pds_firehose_enabled`] for why. + #[serde(default = "default_pds_firehose_enabled")] + pub pds_firehose_enabled: bool, } impl AppConfig { @@ -122,6 +143,10 @@ impl AppConfig { .ok() .map(|s| parse_csv_env(&s)) .unwrap_or_default(), + pds_firehose_enabled: std::env::var("PDS_FIREHOSE_ENABLED") + .ok() + .map(|s| parse_bool_env(&s)) + .unwrap_or_else(default_pds_firehose_enabled), }) } diff --git a/migrations/appview/0010_pds_firehose_cursor.sql b/migrations/appview/0010_pds_firehose_cursor.sql new file mode 100644 index 0000000..487a8f4 --- /dev/null +++ b/migrations/appview/0010_pds_firehose_cursor.sql @@ -0,0 +1,44 @@ +-- AppView database schema 0010: cursor for the local PDS firehose. +-- +-- Why a second cursor table +-- +-- The AppView now consumes two event streams, and they are numbered in +-- completely different spaces: +-- +-- * `jetstream_cursor.cursor` is a Jetstream `time_us` — microseconds +-- since the epoch, produced by a public relay we do not control. +-- * this table's `cursor` is the `seq` of our own PDS's +-- `com.atproto.sync.subscribeRepos` — a small monotonic counter +-- that starts at 1 in a fresh PDS database. +-- +-- Sharing one row between them would mean the larger of the two values +-- (always the Jetstream timestamp) permanently swallowing the other: +-- `cursor_advance` uses GREATEST, so the very first Jetstream event +-- would push the PDS cursor to ~1.7e15 and every subsequent +-- subscribeRepos connect would ask for a sequence the PDS will never +-- reach. Hence a table of its own, deliberately in the same shape as +-- `jetstream_cursor` so both read/advance the same way. +-- +-- Shape +-- id pinned to 1 by a CHECK — a single-row table, the same +-- pattern `jetstream_cursor` uses. It makes "advance the +-- cursor" a plain UPDATE with no upsert dance and makes a +-- second row impossible to create by accident. +-- cursor the last `seq` we durably applied. 0 means "nothing +-- yet": the consumer then subscribes without a `cursor` +-- query parameter, which the PDS reads as "start from the +-- current head" rather than replaying the entire repo +-- history into a fresh index. +-- updated_at observability only — how stale the stream is can be +-- read straight off the row. +-- +-- The row is inserted here so `cursor_advance`'s UPDATE always has a +-- target; `pds_firehose::cursor_get` still tolerates a missing row and +-- returns 0. + +CREATE TABLE pds_firehose_cursor ( + id INTEGER PRIMARY KEY DEFAULT 1 CHECK (id = 1), + cursor BIGINT NOT NULL DEFAULT 0, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +INSERT INTO pds_firehose_cursor (id, cursor) VALUES (1, 0);