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);