tomdebone/maarcadetweet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(appview): PDS-Firehose konsumieren

Browse Source 
Gegenstück zum subscribeRepos-Endpoint: WebSocket-Consumer mit
persistiertem seq-Cursor, Reconnect-Backoff und Behandlung von
#info/OutdatedCursor.

Eigene Cursor-Tabelle statt einer Zeile in jetstream_cursor: dort steht ein
time_us in der Größenordnung 1.7e15, die seq ist ein kleiner Zähler ab 1.
Geteilt hätte GREATEST den PDS-Cursor sofort in eine Zukunft geschoben, die
die PDS nie erreicht.

Kein neuer Indexer-Pfad — jede Op wird in die Single-Op-Form übersetzt, die
apply_commit schon vom Jetstream kennt. Push und Firehose liefern denselben
Commit doppelt; das ist unkritisch, weil die Schreibpfade Upserts sind und
der Dedupe-Index der Notifications den Rest abfängt. Mit einem Test
festgehalten statt vorausgesetzt.

Der CAR-Reader ist neu (es gab nur einen Writer, und der liegt in einem
Binary-Crate ohne lib-Target). Der CBOR-Reader arbeitet mit explizitem
Offset, weil ein Frame zwei hintereinander geschriebene Werte sind, und
akzeptiert CID-Links in beiden Schreibweisen — die Blöcke tragen Strings.

/healthz meldet beide Ströme getrennt; sie fallen unabhängig voneinander
aus.

Verifiziert mit totem Push-Ziel: der Post kam trotzdem an.

Co-Authored-By: Eifelduds <noreply@eifelduds.de>
Claude-Session: https://claude.ai/code/session_013HC9HLrUU1LNwkzp8nkDLX
This commit is contained in:
tomdebone
2026-09-10 07:08:23 +02:00
co-authored by Eifelduds
parent cd68049640
commit df776531df
11 changed files with 2849 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/appview/src/cbor.rs
+442
View File
@@ -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(<cid>)`. `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::<serde_json::Value, _>`, 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<u8>),
Text(String),
Array(Vec<Cbor>),
/// 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<Cbor>),
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<i64> {
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<bool> {
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<Cid> {
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::<Cid>().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<Cbor> {
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<u64> {
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<usize> {
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<u8> {
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(<cid>) 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());
}
}