From 51c2d8325c9125206e8a2cbbc8b4b4c2fa85d165 Mon Sep 17 00:00:00 2001 From: "@permadeath.com" Date: Wed, 16 Sep 2026 17:39:49 -0400 Subject: [PATCH] feat(index)!: follow a server over com.atproto.sync.subscribeRepos didbot serves only the protocol's own stream now, so `--pds` reads it: a WebSocket of DAG-CBOR frames whose commits carry their records as a CAR. `frames` decodes them with the DASL crate already in the tree, the cursor is a bare sequence number, the vectors are frames recorded from a real server, and `vibescrobble-replay` records and plays back the same bytes over a WebSocket, so an index follows a replay exactly as it follows a server. Co-Authored-By: Claude Fable 5.1 Change-Id: I4407ee14653b02eeb5d04855ad43997b84022621 --- Cargo.lock | 23 +- Cargo.toml | 19 +- crates/vibescrobble-index/Cargo.toml | 6 +- .../src/bin/vibescrobble-index.rs | 8 +- .../src/bin/vibescrobble-replay.rs | 49 +- crates/vibescrobble-index/src/firehose.rs | 711 +++++++----------- crates/vibescrobble-index/src/frames.rs | 427 +++++++++++ crates/vibescrobble-index/src/jetstream.rs | 13 +- crates/vibescrobble-index/src/lib.rs | 10 +- crates/vibescrobble-index/src/replay.rs | 392 +++++----- crates/vibescrobble-index/src/sse.rs | 70 +- .../tests/account_frames.rs | 6 +- .../vibescrobble-index/tests/firehose_wire.rs | 208 ++--- crates/vibescrobble-index/tests/reconnect.rs | 197 +++-- docs/running-locally.md | 76 +- plan/ingest.md | 9 +- vectors/firehose/README.md | 47 -- vectors/firehose/account-active.json | 11 - vectors/firehose/account-deactivated.json | 12 - vectors/firehose/account-deleted.json | 12 - vectors/firehose/commit-sparse.json | 18 - vectors/firehose/commit.json | 22 - vectors/firehose/identity-no-handle.json | 10 - vectors/firehose/identity.json | 11 - vectors/firehose/info-live.json | 11 - vectors/firehose/info-outdated-cursor.json | 11 - vectors/firehose/info-unknown-name.json | 11 - vectors/subscribe-repos/README.md | 44 ++ vectors/subscribe-repos/account-active.json | 6 + .../subscribe-repos/account-deactivated.json | 6 + vectors/subscribe-repos/account-deleted.json | 6 + vectors/subscribe-repos/commit.json | 6 + .../subscribe-repos/error-future-cursor.json | 4 + vectors/subscribe-repos/identity.json | 6 + .../subscribe-repos/info-outdated-cursor.json | 4 + 35 files changed, 1338 insertions(+), 1144 deletions(-) create mode 100644 crates/vibescrobble-index/src/frames.rs delete mode 100644 vectors/firehose/README.md delete mode 100644 vectors/firehose/account-active.json delete mode 100644 vectors/firehose/account-deactivated.json delete mode 100644 vectors/firehose/account-deleted.json delete mode 100644 vectors/firehose/commit-sparse.json delete mode 100644 vectors/firehose/commit.json delete mode 100644 vectors/firehose/identity-no-handle.json delete mode 100644 vectors/firehose/identity.json delete mode 100644 vectors/firehose/info-live.json delete mode 100644 vectors/firehose/info-outdated-cursor.json delete mode 100644 vectors/firehose/info-unknown-name.json create mode 100644 vectors/subscribe-repos/README.md create mode 100644 vectors/subscribe-repos/account-active.json create mode 100644 vectors/subscribe-repos/account-deactivated.json create mode 100644 vectors/subscribe-repos/account-deleted.json create mode 100644 vectors/subscribe-repos/commit.json create mode 100644 vectors/subscribe-repos/error-future-cursor.json create mode 100644 vectors/subscribe-repos/identity.json create mode 100644 vectors/subscribe-repos/info-outdated-cursor.json diff --git a/Cargo.lock b/Cargo.lock index 9856a7d7..dfa1d4a0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -174,6 +174,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" dependencies = [ "axum-core", + "base64 0.22.1", "bytes", "form_urlencoded", "futures-util", @@ -192,8 +193,10 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", + "sha1", "sync_wrapper", "tokio", + "tokio-tungstenite", "tower", "tower-layer", "tower-service", @@ -2387,13 +2390,13 @@ dependencies = [ [[package]] name = "sha1" -version = "0.11.0" +version = "0.10.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "aacc4cc499359472b4abe1bf11d0b12e688af9a805fa5e3016f9a386dc2d0214" +checksum = "a978451301f4db1d02937a4ab3ccce137717b81826e79b7d49ffe3244a13c3b8" dependencies = [ "cfg-if", - "cpufeatures 0.3.0", - "digest 0.11.3", + "cpufeatures 0.2.17", + "digest 0.10.7", ] [[package]] @@ -2771,9 +2774,9 @@ dependencies = [ [[package]] name = "tokio-tungstenite" -version = "0.30.0" +version = "0.29.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "17a073bfed563fa236697a068031408a93cd9522e08abf9933ead3e73411bd71" +checksum = "8f72a05e828585856dacd553fba484c242c46e391fb0e58917c942ee9202915c" dependencies = [ "futures-util", "log", @@ -2917,16 +2920,16 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "tungstenite" -version = "0.30.0" +version = "0.29.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e48ac77174b19c110a50ab2128b24215ac9cb40e0e12e093fb602d175c569d22" +checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8" dependencies = [ "bytes", "data-encoding", "http", "httparse", "log", - "rand 0.10.2", + "rand 0.9.5", "rustls", "rustls-pki-types", "sha1", @@ -3060,8 +3063,10 @@ dependencies = [ name = "vibescrobble-index" version = "0.1.0" dependencies = [ + "atproto-dasl", "atproto-identity", "axum", + "base64 0.22.1", "futures-core", "futures-util", "model2vec-rs", diff --git a/Cargo.toml b/Cargo.toml index b256f1c3..fc77e9c0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,8 +26,12 @@ vibescrobble-query = { version = "0.1.0", path = "crates/vibescrobble-query" } # `lru` for a cache this crate does not use. What is wanted here is the syntax # and the document model. atproto-identity = { version = "0.14", default-features = false } +# The same release `atproto-identity` builds on, for DAG-CBOR and CAR: what a +# `com.atproto.sync.subscribeRepos` frame is made of. +atproto-dasl = { version = "0.14", default-features = false } -axum = "0.8" +# `ws` is what a replay serves `subscribeRepos` on. +axum = { version = "0.8", features = ["ws"] } base64 = "0.22" # DPoP proofs are ES256, which is P-256: the curve this workspace's server # side verifies and the one every atproto authorization server accepts. @@ -45,14 +49,13 @@ reqwest = { version = "0.12", default-features = false, features = ["json", "rus serde = { version = "1", features = ["derive"] } serde_json = "1" thiserror = "2" -# Vouch records carry `createdAt` and `expiresAt`, and an expired vouch has to -# be dropped rather than followed. +# RFC 3339 timestamps, formatted and compared. time = { version = "0.3", features = ["formatting", "parsing", "macros"] } -# The relay's own output is DAG-CBOR over a WebSocket; Jetstream turns that -# into JSON and filters it by collection, which is why this is a WebSocket -# client and not a CAR reader. rustls rather than native TLS, to match reqwest -# and keep one TLS stack in the binary. -tokio-tungstenite = { version = "0.30", default-features = false, features = ["connect", "rustls-tls-webpki-roots"] } +# Both streams this reads -- a server's `com.atproto.sync.subscribeRepos` and a +# Jetstream -- are WebSockets. The version is the one axum's `ws` feature +# builds on, so the graph carries one copy. rustls rather than native TLS, to +# match reqwest and keep one TLS stack in the binary. +tokio-tungstenite = { version = "0.29", default-features = false, features = ["connect", "rustls-tls-webpki-roots"] } # Only to hand tokio-tungstenite a config with DIDBOT_EXTRA_CA_CERTS added to # the webpki roots it would otherwise use. ring, the provider reqwest uses. rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12"] } diff --git a/crates/vibescrobble-index/Cargo.toml b/crates/vibescrobble-index/Cargo.toml index cd3f3854..73226d7f 100644 --- a/crates/vibescrobble-index/Cargo.toml +++ b/crates/vibescrobble-index/Cargo.toml @@ -18,8 +18,6 @@ reqwest.workspace = true serde.workspace = true serde_json.workspace = true thiserror.workspace = true -# Vouch records carry `createdAt` and `expiresAt`, and an expired vouch has to -# be dropped rather than followed. time.workspace = true tokio.workspace = true tokio-tungstenite.workspace = true @@ -32,6 +30,10 @@ tower-http.workspace = true # Containment is checked here with the same comparison the server mints under, # rather than a second implementation of the label walk. atproto-identity.workspace = true +# `subscribeRepos` frames are DAG-CBOR, and a commit's records ride in a CAR. +atproto-dasl.workspace = true +# A recorded frame is bytes, and a recording is a text file. +base64.workspace = true # Static embeddings: a lookup table and a mean, so a status line can be # embedded on the way past without a GPU. See `work.rs`. model2vec-rs = { version = "0.2.1", default-features = false, features = ["hf-hub", "fancy-regex"] } diff --git a/crates/vibescrobble-index/src/bin/vibescrobble-index.rs b/crates/vibescrobble-index/src/bin/vibescrobble-index.rs index 99687362..7dc8835f 100644 --- a/crates/vibescrobble-index/src/bin/vibescrobble-index.rs +++ b/crates/vibescrobble-index/src/bin/vibescrobble-index.rs @@ -885,14 +885,10 @@ async fn firehose_loop( update = incoming.recv() => { let Some(update) = update else { break; }; match update { - Update::Gap { - server, - oldest, - message, - } => { + Update::Gap { server, message } => { warn!( server = server.base_url(), - oldest, message, "the firehose left a gap; reading every repository" + message, "the stream left a gap; reading every repository" ); shared.pulse.following(&following, false).await; catch_up_now.notify_one(); diff --git a/crates/vibescrobble-index/src/bin/vibescrobble-replay.rs b/crates/vibescrobble-index/src/bin/vibescrobble-replay.rs index dda15dd6..a801df27 100644 --- a/crates/vibescrobble-index/src/bin/vibescrobble-replay.rs +++ b/crates/vibescrobble-index/src/bin/vibescrobble-replay.rs @@ -1,4 +1,4 @@ -//! Record a firehose, and play it back with nothing else running. +//! Record a server's stream, and play it back with nothing else running. //! //! ```text //! vibescrobble-replay record --pds http://localhost:3000 --out session.jsonl @@ -15,19 +15,21 @@ use std::io::Write; use std::path::PathBuf; use tokio::sync::mpsc; +use vibescrobble_index::firehose::{read_frames, subscribe_url, FRAME_BUFFER}; +use vibescrobble_index::frames::NSID; use vibescrobble_index::replay::{router, Pace, Recorded, Recording}; -use vibescrobble_index::sse::{read_frames, Frame}; use vibescrobble_index::ServerRef; /// What `--help` prints. const USAGE: &str = "\ -vibescrobble-replay — record a firehose, and play it back +vibescrobble-replay — record a server's stream, and play it back - vibescrobble-replay record --out [--pds ] [--frames ] + vibescrobble-replay record --out [--pds ] [--cursor ] [--frames ] vibescrobble-replay serve --file [--listen ] [--rate ] Options --pds Server to record from (default http://localhost:3000) + --cursor Resume after this sequence number rather than from the head --out Where to write the recording --frames Stop after this many frames (default: until interrupted) --file Recording to play back @@ -35,8 +37,9 @@ Options --rate Frames per second, or 0 for as fast as it will go (default 5) -A recording is one JSON object per line, so trimming it to the frames that -reproduce something is a text editor and nothing else. +A recording is one JSON object per line — a frame, its sequence number and +the repository it is about — so trimming it to the frames that reproduce +something is a text editor and nothing else. "; /// Where to record from when nothing says otherwise. @@ -74,7 +77,7 @@ async fn main() -> Result<(), Box> { } } -/// Subscribes to a firehose and writes what it sees. +/// Subscribes to a server's stream and writes what it sees. /// /// Appends as it goes rather than buffering to the end, so that a recording /// interrupted with Ctrl-C is a recording rather than nothing. That is the @@ -83,6 +86,7 @@ async fn main() -> Result<(), Box> { async fn record(args: &[String]) -> Result<(), Box> { let mut pds = DEFAULT_PDS.to_owned(); let mut out: Option = None; + let mut cursor: Option = None; let mut limit: Option = None; let mut rest = args.iter(); @@ -90,6 +94,13 @@ async fn record(args: &[String]) -> Result<(), Box> { match arg.as_str() { "--pds" => pds = rest.next().ok_or("--pds wants a url")?.clone(), "--out" => out = Some(PathBuf::from(rest.next().ok_or("--out wants a path")?)), + "--cursor" => { + cursor = Some( + rest.next() + .ok_or("--cursor wants a sequence number")? + .parse()?, + ); + } "--frames" => { limit = Some(rest.next().ok_or("--frames wants a count")?.parse()?); } @@ -102,20 +113,28 @@ async fn record(args: &[String]) -> Result<(), Box> { let file = std::fs::File::create(&out)?; let mut file = std::io::BufWriter::new(file); - let (sender, mut frames) = mpsc::channel::(vibescrobble_index::sse::FRAME_BUFFER); - let url = format!("{}/firehose", server.base_url()); + let (sender, mut incoming) = mpsc::channel::>(FRAME_BUFFER); + let url = subscribe_url(&server, cursor); tracing::info!(%url, path = %out.display(), "recording"); - let http = vibescrobble_index::http::stream_client(); let reading = tokio::spawn(async move { - if let Err(err) = read_frames(&http, &url, &sender).await { + if let Err(err) = read_frames(&url, &sender).await { tracing::warn!(error = %err, "the stream ended"); } }); let mut written = 0usize; - while let Some(frame) = frames.recv().await { - let recorded = Recorded::from(frame); + while let Some(bytes) = incoming.recv().await { + let recorded = match Recorded::from_frame(&bytes) { + Ok(recorded) => recorded, + Err(err) => { + // Kept out rather than kept opaque: a playback of it would be + // refused by the reader for the same reason, at a line number + // nobody could do anything about. + tracing::warn!(error = %err, "a frame this recorder could not read; dropped"); + continue; + } + }; writeln!(file, "{}", serde_json::to_string(&recorded)?)?; // Flushed per frame: the ordinary end of a recording is Ctrl-C, and a // buffer that had not reached the disk would take the interesting part @@ -131,7 +150,7 @@ async fn record(args: &[String]) -> Result<(), Box> { Ok(()) } -/// Serves a recording as a firehose. +/// Serves a recording as a server's stream. async fn serve(args: &[String]) -> Result<(), Box> { let mut file: Option = None; let mut listen: Option = None; @@ -167,7 +186,7 @@ async fn serve(args: &[String]) -> Result<(), Box> { "replaying {} — {frames} frame(s), {accounts} account(s)", file.display() ); - println!(" firehose http://{addr}/firehose"); + println!(" stream ws://{addr}/xrpc/{NSID}"); println!( " follow it vibescrobble-index --pds http://localhost:{}", addr.port() diff --git a/crates/vibescrobble-index/src/firehose.rs b/crates/vibescrobble-index/src/firehose.rs index bf7061f4..1223bd6b 100644 --- a/crates/vibescrobble-index/src/firehose.rs +++ b/crates/vibescrobble-index/src/firehose.rs @@ -1,53 +1,53 @@ -//! Subscribing to a server's record stream, and resuming after a gap. +//! Following one server's `com.atproto.sync.subscribeRepos`, and resuming +//! after a gap. //! //! The survey answers "which accounts exist"; this answers "what have they -//! just written". Before it there was no answer to the second question at all, -//! so the index re-read every repository on a timer and worked out what was -//! new by comparing AT-URIs. That still happens — see the argument in -//! [`crate::survey`] and in the index binary — but it is now the backstop +//! just written". The index re-reads every repository on a timer as well and +//! works out what is new by comparing AT-URIs — see the argument in +//! [`crate::survey`] and in the index binary — but that is the backstop //! rather than the mechanism. //! //! # The cursor is the whole point //! -//! A subscriber remembers the last frame it applied and hands the server that -//! cursor when it reconnects. The server replays what came after it. What that -//! buys is the two failure modes a stream without one has to choose between: -//! resume from nothing and miss whatever happened during the gap, or re-read -//! everything and pay for it every time. Neither is necessary when the server -//! can say "here is what you missed". +//! A subscriber remembers the sequence number of the last frame it applied +//! and hands it to the server when it reconnects. The server replays what +//! came after it. What that buys is the two failure modes a stream without +//! one has to choose between: resume from nothing and miss whatever happened +//! during the gap, or re-read everything and pay for it every time. Neither +//! is necessary when the server can say "here is what you missed". //! -//! It cannot always say that. The buffer is finite and lives in the server's -//! memory, so a subscriber that was away too long, or that is talking to a -//! server which has restarted, gets [`Update::Gap`] instead — the server -//! saying, in as many words, that there are records between the cursor and the -//! oldest frame it still holds. That is not an error and it is not recoverable -//! from the stream; the only thing that can fill it is reading the -//! repositories, which is why the poll does not go away. +//! It cannot always say that. The replay buffer is finite and lives in the +//! server's memory, so a subscriber that was away too long, or one whose +//! server has restarted, gets [`Update::Gap`] instead — the server saying, +//! in as many words, that there are frames between the cursor and the oldest +//! one it still holds. That is not an error and it is not recoverable from +//! the stream; the only thing that can fill it is reading the repositories, +//! which is why the poll does not go away. //! //! # What a frame says about the repository //! -//! A frame names the commit it landed in — the record's CID, the commit's CID -//! and its revision. That is what makes the stream reconcilable rather -//! than merely believable: an index that has applied every frame up to some -//! sequence number holds exactly the records the commit named by that frame -//! covers, and can check that by asking the server for +//! A commit frame names the commit it is — the commit's CID and its +//! revision. That is what makes the stream reconcilable rather than merely +//! believable: an index that has applied every frame up to some sequence +//! number holds exactly the records the commit named by that frame covers, +//! and can check that by asking the server for //! `com.atproto.sync.getLatestCommit` or by reading an export. `/health` //! reports the head this index last heard for each repository, which is the //! left-hand side of that comparison. //! -//! The blocks that would *prove* any of it are not in the frame and are a -//! request away at `com.atproto.sync.getRecord`. Nothing here verifies one: -//! this crate has no DAG-CBOR decoder, and the frame does not -//! carry the proof for it. +//! The records ride along in the frame's CAR and are read out of it by +//! [`crate::frames`]. The tree nodes beside them, which are what would +//! *prove* a record is in the commit, are read past: this crate reads +//! records, and a proof is a request away at `com.atproto.sync.getRecord`. //! //! # What is deliberately not trusted //! -//! A frame names a repository, and a server can name any repository it likes. -//! Nothing here checks that — parsing a frame does not know what a consumer -//! is allowed to believe about it — so the check is made once a [`Commit`], -//! [`Account`] or [`Identity`] reaches the index binary, against the server's -//! own zone (see [`crate::survey::FeedSource::seen`]). A frame naming -//! something outside it is dropped there rather than here. +//! A frame names a repository, and a server can name any repository it +//! likes. Nothing here checks that — reading a frame does not know what a +//! consumer is allowed to believe about it — so the check is made once a +//! [`Commit`], [`Account`] or [`Identity`] reaches the index binary, against +//! the server's own zone (see [`crate::survey::FeedSource::seen`]). A frame +//! naming something outside it is dropped there rather than here. //! //! # Three things a server says, not one //! @@ -55,30 +55,37 @@ //! appearing, ending, or changing status — `active: false` with no further //! detail, or a `status` naming why. An [`Identity`] is a handle its server //! now claims for it. All three carry a `seq`, and all three advance the -//! cursor the same way a commit does: a frame this index does nothing with is -//! still a frame it has seen, and asking for it again after a reconnect would -//! replay it forever. +//! cursor the same way: a frame this index does nothing with is still a +//! frame it has seen, and asking for it again after a reconnect would replay +//! it forever. + +use std::time::Duration; -use serde::{Deserialize, Serialize}; use serde_json::Value; use tokio::sync::mpsc; use tracing::{debug, info, warn}; +use crate::frames::{decode, Message, FUTURE_CURSOR, NSID, OUTDATED_CURSOR}; +use crate::http::{TransportError, CONNECT_TIMEOUT, IDLE_TIMEOUT}; use crate::server::ServerRef; -use crate::sse::{read_frames, Frame, FRAME_BUFFER, RETRY}; -/// One record write, as the server puts it on the wire. +pub use crate::frames::{Account, Identity, Info}; + +/// How long to wait before reconnecting a stream that ended. +pub const RETRY: Duration = Duration::from_secs(3); + +/// How many frames are buffered between the socket and whatever reads them. +pub const FRAME_BUFFER: usize = 256; + +/// One record written, as this index reads it off a commit. /// -/// A mirror of the server's own frame type, declared here rather than shared: -/// an index reading somebody else's server must not compile against that -/// server's types, or it can only ever read servers built from one repository. -#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)] -#[serde(rename_all = "camelCase")] +/// A `#commit` frame carries every write in the commit; this is one of them, +/// with the commit's own facts beside it. A commit with two writes is two of +/// these with the same `seq`, `commit` and `rev`. +#[derive(Debug, Clone, PartialEq)] pub struct Commit { - /// Where the frame sits in the stream. + /// Where the frame sits in the stream. The cursor to resume after. pub seq: u64, - /// Which run of the server minted [`Self::seq`]. - pub instance: String, /// The repository written to. pub did: String, /// The collection the record went into. @@ -88,143 +95,21 @@ pub struct Commit { /// The AT-URI naming the record. pub uri: String, /// The CID of the record itself. - /// - /// Defaulted, like every field a server might not send: an index reads - /// servers it did not build, and one that predates the commit being named - /// must not stop it dead. An empty string is the honest reading of "this - /// server did not say". - #[serde(default)] pub cid: String, /// The CID of the signed commit the write landed in. /// /// This is what a repository is reconciled on. A revision is derived on - /// the server that sent it and does not move for every write, - /// so two frames can share one and still cover different records. - #[serde(default)] + /// the server that sent it and does not move for every write, so two + /// frames can share one and still cover different records. pub commit: String, /// The revision that commit is at. - #[serde(default)] pub rev: String, - /// When the server stored it, RFC 3339. - #[serde(default)] + /// When the server emitted the frame, RFC 3339. pub time: String, - /// The record itself. + /// The record itself, as atproto's JSON spells it. pub record: Value, } -impl Commit { - /// The cursor to hand back to resume after this frame. - pub fn cursor(&self) -> String { - format!("{}:{}", self.instance, self.seq) - } -} - -/// An account's server claims it now answers to a handle, as the wire has it. -/// -/// The shape atproto's own `#identity` carries: `did`, `seq`, `time` and an -/// optional `handle`. Declared here rather than shared, for the same reason -/// [`Commit`] is: an index reading somebody else's server must not compile -/// against that server's types. -#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)] -#[serde(rename_all = "camelCase")] -pub struct Identity { - /// Where the frame sits in the stream. - pub seq: u64, - /// Which run of the server minted [`Self::seq`]. - #[serde(default)] - pub instance: String, - /// The account. - pub did: String, - /// When the server saw it, RFC 3339. - #[serde(default)] - pub time: String, - /// The handle the account's server now claims for it, when it named one. - /// - /// Still only a claim: it is the account's own server saying so. - /// Checking it is [`crate::handle::verify`], which is a request and not - /// on this path. - #[serde(default)] - pub handle: Option, -} - -impl Identity { - /// The cursor to hand back to resume after this frame. - pub fn cursor(&self) -> String { - format!("{}:{}", self.instance, self.seq) - } -} - -/// An account appeared, ended, or changed status, as the wire has it. -/// -/// The shape atproto's own `#account` carries: `did`, `seq`, `time`, -/// `active`, and an optional `status` naming why when it is not. -#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)] -#[serde(rename_all = "camelCase")] -pub struct Account { - /// Where the frame sits in the stream. - pub seq: u64, - /// Which run of the server minted [`Self::seq`]. - #[serde(default)] - pub instance: String, - /// The account. - pub did: String, - /// When the server saw it, RFC 3339. - #[serde(default)] - pub time: String, - /// Whether the account is still being served. - pub active: bool, - /// Why not, when it is not: `deactivated`, `suspended`, `deleted` and so - /// on. An open set, carried rather than interpreted. - #[serde(default)] - pub status: Option, -} - -impl Account { - /// The cursor to hand back to resume after this frame. - pub fn cursor(&self) -> String { - format!("{}:{}", self.instance, self.seq) - } - - /// Whether this says the account has ended for good. - /// - /// Deliberately narrower than `!active`: a deactivated or suspended - /// account is not being served right now and may still come back, which - /// is not the same as gone. [`crate::identity::Registry::ended`] - /// tombstones an identity on the strength of this, so it should only - /// answer `true` when the account is really not coming back. - #[must_use] - pub fn is_deleted(&self) -> bool { - !self.active && self.status.as_deref() == Some("deleted") - } -} - -/// The stream's opening frame, as the server describes itself. -/// -/// Unknown `name` values are carried rather than refused: a server that grows -/// a fourth thing to say about a connection should not stop an index that has -/// not heard of it, and every name except `OutdatedCursor` means "carry on". -#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)] -#[serde(rename_all = "camelCase")] -pub struct StreamInfo { - /// `Live`, `Replay`, `OutdatedCursor`, or something newer. - pub name: String, - /// The run of the server that minted the sequence numbers. - #[serde(default)] - pub instance: String, - /// The oldest sequence number still replayable. - #[serde(default)] - pub oldest: u64, - /// The newest sequence number assigned. - #[serde(default)] - pub newest: u64, - /// What the server has to say about it, for a log line. - #[serde(default)] - pub message: String, -} - -/// The name a server uses when a cursor is out of reach. -const OUTDATED: &str = "OutdatedCursor"; - /// Something the stream told this index. #[derive(Debug, Clone, PartialEq)] pub enum Update { @@ -249,173 +134,256 @@ pub enum Update { /// The change. account: Box, }, - /// Records were missed and cannot be recovered from the stream. + /// Frames were missed and cannot be recovered from the stream. /// - /// The consumer's cursor is older than anything the server still holds, so - /// the only way to find out what is in the gap is to read the - /// repositories. + /// The consumer's cursor is older than anything the server still holds, + /// or names a frame the server never sent, so the only way to find out + /// what is in the gap is to read the repositories. Gap { /// Which server said so. server: ServerRef, - /// The oldest sequence number it can still replay. - oldest: u64, /// What it said, for a log line. message: String, }, } -/// Where a server's record stream is read from, resuming after `cursor`. -fn firehose_url(server: &ServerRef, cursor: Option<&str>) -> String { +/// Where a server's stream is read from, resuming after `cursor`. +/// +/// The WebSocket scheme that matches the server's HTTP one: an index pointed +/// at `https://` must not open a cleartext socket to the same host. +pub fn subscribe_url(server: &ServerRef, cursor: Option) -> String { + let base = server.base_url(); + let base = base + .strip_prefix("https://") + .map(|rest| format!("wss://{rest}")) + .or_else(|| { + base.strip_prefix("http://") + .map(|rest| format!("ws://{rest}")) + }) + .unwrap_or_else(|| base.to_owned()); match cursor { - // A cursor is `:` and the colon is reserved in a query, - // so it is encoded rather than interpolated. The server percent-decodes - // once on arrival and gets back exactly what it minted. - Some(cursor) => format!( - "{}/firehose?cursor={}", - server.base_url(), - encode_query_value(cursor) - ), - None => format!("{}/firehose", server.base_url()), + Some(cursor) => format!("{base}/xrpc/{NSID}?cursor={cursor}"), + None => format!("{base}/xrpc/{NSID}"), } } -/// Percent-encodes everything outside the unreserved set. -fn encode_query_value(value: &str) -> String { - let mut out = String::with_capacity(value.len()); - for byte in value.bytes() { - match byte { - b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => { - out.push(byte as char); +/// Reads one connection's frames into `out`, as the bytes that arrived, +/// until the server closes it. +/// +/// Raw on purpose: a recorder keeps what came off the wire and a subscriber +/// decodes it, and the two must not disagree about what the wire carried. +pub async fn read_frames(url: &str, out: &mpsc::Sender>) -> Result<(), TransportError> { + use futures_core::Stream; + use futures_util::SinkExt; + use std::pin::Pin; + use std::task::Context; + use tokio_tungstenite::tungstenite::Message as Ws; + + let (mut socket, _) = tokio::time::timeout( + CONNECT_TIMEOUT, + tokio_tungstenite::connect_async_tls_with_config(url, None, false, crate::tls::connector()), + ) + .await + .map_err(|_| TransportError::new(url, format!("no connection within {CONNECT_TIMEOUT:?}")))? + .map_err(|err| TransportError::new(url, err.to_string()))?; + // A quiet half of IDLE_TIMEOUT sends a ping. A second quiet half, with + // not even the pong back, drops the connection. + let mut pinged = false; + loop { + let next = tokio::time::timeout( + IDLE_TIMEOUT / 2, + std::future::poll_fn(|cx: &mut Context<'_>| Pin::new(&mut socket).poll_next(cx)), + ) + .await; + let message = match next { + Ok(message) => message, + Err(_) if !pinged => { + debug!(url, "subscribeRepos is quiet; pinging"); + socket + .send(Ws::Ping(Vec::new().into())) + .await + .map_err(|err| TransportError::new(url, err.to_string()))?; + pinged = true; + continue; + } + Err(_) => { + return Err(TransportError::new( + url, + format!("nothing arrived for {IDLE_TIMEOUT:?}"), + )) } - other => out.push_str(&format!("%{other:02X}")), + }; + pinged = false; + let Some(message) = message else { + return Ok(()); + }; + match message.map_err(|err| TransportError::new(url, err.to_string()))? { + Ws::Binary(bytes) => { + if out.send(bytes.to_vec()).await.is_err() { + return Ok(()); + } + } + Ws::Close(_) => return Ok(()), + // A ping is answered by the socket itself; anything else on a + // binary stream is nothing this reads. + other => debug!(url, kind = ?other, "a non-binary message; ignored"), } } - out } -/// Follows one server's firehose from its head, forever. +/// Follows one server's stream from its head, forever. /// /// Reconnects on its own, and does so *with* the cursor it has been keeping, /// which is what makes a restart of either side cost a replay rather than a -/// re-survey. Returns only when the receiver is gone. +/// re-read. Returns only when the receiver is gone. pub async fn subscribe(server: ServerRef, out: mpsc::Sender) { subscribe_from(server, None, out).await; } -/// Follows one server's firehose, starting after `cursor`. +/// Follows one server's stream, starting after `cursor`. /// /// The starting cursor is a parameter rather than always `None` because an /// index that outlives one run of itself would hand back what it had stored, /// and because a resume is not a thing that can be tested end to end if the /// only way to reach one is to wait for a connection to break. -pub async fn subscribe_from(server: ServerRef, cursor: Option, out: mpsc::Sender) { - let http = crate::http::stream_client(); +pub async fn subscribe_from(server: ServerRef, cursor: Option, out: mpsc::Sender) { let mut cursor = cursor; loop { - let url = firehose_url(&server, cursor.as_deref()); - debug!(url, "connecting to the firehose"); + let url = subscribe_url(&server, cursor); + debug!(url, "connecting to subscribeRepos"); - let (frames, mut incoming) = mpsc::channel::(FRAME_BUFFER); + let (sink, mut incoming) = mpsc::channel::>(FRAME_BUFFER); let reader = tokio::spawn({ - let http = http.clone(); let url = url.clone(); async move { - match read_frames(&http, &url, &frames).await { - Ok(()) => debug!(url, "firehose ended"), - Err(err) => warn!(url, %err, "firehose failed"), + match read_frames(&url, &sink).await { + Ok(()) => debug!(url, "subscribeRepos ended"), + Err(err) => warn!(url, %err, "subscribeRepos failed"), } } }); let mut applied = 0usize; - while let Some(frame) = incoming.recv().await { - match frame.event.as_deref() { - Some("info") => { - let Some(info) = frame.parse::() else { - continue; - }; - if info.name == OUTDATED { - warn!( - server = server.base_url(), - oldest = info.oldest, - message = info.message, - "the firehose cannot replay from our cursor" - ); - // The cursor is worthless now: keeping it would ask - // for the same impossible replay on every reconnect. - cursor = None; - let gap = Update::Gap { - server: server.clone(), - oldest: info.oldest, - message: info.message, - }; - if out.send(gap).await.is_err() { - reader.abort(); - return; - } - } else { - info!( - server = server.base_url(), - name = info.name, - oldest = info.oldest, - newest = info.newest, - "firehose connected" - ); - } + while let Some(bytes) = incoming.recv().await { + let message = match decode(&bytes) { + Ok(message) => message, + Err(err) => { + warn!(server = server.base_url(), %err, "a frame this index could not read; skipped"); + continue; } - Some("commit") => { - let Some(commit) = frame.parse::() else { - continue; - }; - // Advanced before the send and regardless of what the - // consumer does with it: a frame this index chose to - // ignore has still been seen, and asking for it again - // after a reconnect would replay it forever. - cursor = Some(commit.cursor()); + }; + // Advanced on every numbered frame, whatever it was and before + // anything is done with it: a frame this index chose to ignore has + // still been seen, and asking for it again after a reconnect would + // replay it forever. + if let Some(seq) = message.seq() { + cursor = Some(seq); + } + let updates: Vec = match message { + Message::Commit(frame) => { applied += 1; - let update = Update::Commit { - server: server.clone(), - commit: Box::new(commit), - }; - if out.send(update).await.is_err() { - reader.abort(); - return; + let mut updates = Vec::new(); + for write in frame.writes { + // A delete carries no record, and neither does a + // commit the server marked `tooBig`: nothing to read + // here, and the sweep reads the repository anyway. + let Some(record) = write.record else { + debug!( + uri = write.uri, + action = write.action, + "no record in the frame; ignored" + ); + continue; + }; + updates.push(Update::Commit { + server: server.clone(), + commit: Box::new(Commit { + seq: frame.seq, + did: frame.did.clone(), + collection: write.collection, + rkey: write.rkey, + uri: write.uri, + cid: write.cid.unwrap_or_default(), + commit: frame.commit.clone(), + rev: frame.rev.clone(), + time: frame.time.clone(), + record, + }), + }); } + updates } - Some("identity") => { - let Some(identity) = frame.parse::() else { - continue; - }; - cursor = Some(identity.cursor()); + Message::Identity(identity) => { applied += 1; - let update = Update::Identity { + vec![Update::Identity { server: server.clone(), identity: Box::new(identity), - }; - if out.send(update).await.is_err() { - reader.abort(); - return; - } + }] } - Some("account") => { - let Some(account) = frame.parse::() else { - continue; - }; - cursor = Some(account.cursor()); + Message::Account(account) => { applied += 1; - let update = Update::Account { + vec![Update::Account { server: server.clone(), account: Box::new(account), - }; - if out.send(update).await.is_err() { - reader.abort(); - return; - } + }] + } + Message::Info(info) if info.name == OUTDATED_CURSOR => { + warn!( + server = server.base_url(), + message = info.message, + "the stream cannot replay from our cursor" + ); + // The cursor is worthless now: keeping it would ask for the + // same impossible replay on every reconnect. + cursor = None; + vec![Update::Gap { + server: server.clone(), + message: info.message, + }] + } + Message::Info(info) => { + info!( + server = server.base_url(), + name = info.name, + message = info.message, + "subscribeRepos said" + ); + continue; + } + Message::Error { error, message } if error == FUTURE_CURSOR => { + // The cursor names a frame this server never sent: it is + // from a different server, or a run of this one that did + // not keep its numbering. Everything between is a hole. + warn!( + server = server.base_url(), + message, "our cursor is ahead of the stream" + ); + cursor = None; + vec![Update::Gap { + server: server.clone(), + message, + }] + } + Message::Error { error, message } => { + warn!( + server = server.base_url(), + error, message, "subscribeRepos refused" + ); + continue; + } + // Carried past rather than refused, and the cursor has moved + // over it, because a server that grows a kind of message must + // not strand an older consumer. + Message::Other { t, .. } => { + debug!(t, "a message type this index has no name for; ignored"); + continue; + } + }; + for update in updates { + if out.send(update).await.is_err() { + reader.abort(); + return; } - // A frame this index has no name for. Carried past rather than - // refused, and the cursor is untouched, because a server that - // grows a frame type must not strand an older consumer. - other => debug!(event = other, "firehose frame ignored"), } } @@ -427,7 +395,7 @@ pub async fn subscribe_from(server: ServerRef, cursor: Option, out: mpsc server = server.base_url(), applied, resuming = cursor.is_some(), - "firehose disconnected; retrying" + "subscribeRepos disconnected; retrying" ); tokio::time::sleep(RETRY).await; } @@ -444,173 +412,22 @@ mod tests { #[test] fn a_cursorless_subscription_asks_for_the_head() { assert_eq!( - firehose_url(&server(), None), - "http://localhost:3000/firehose" + subscribe_url(&server(), None), + "ws://localhost:3000/xrpc/com.atproto.sync.subscribeRepos" ); } #[test] - fn a_cursor_is_encoded_rather_than_interpolated() { - // The colon is reserved in a query string; interpolating it works by - // luck and stops working the moment the instance identifier does not. + fn a_cursor_is_the_bare_sequence_number() { assert_eq!( - firehose_url(&server(), Some("18e4a2f:97")), - "http://localhost:3000/firehose?cursor=18e4a2f%3A97" + subscribe_url(&server(), Some(97)), + "ws://localhost:3000/xrpc/com.atproto.sync.subscribeRepos?cursor=97" ); } #[test] - fn a_commit_frame_reads_back_into_a_cursor() { - let frame = Frame { - id: None, - event: Some("commit".to_owned()), - data: serde_json::json!({ - "seq": 12, - "instance": "18e4a2f", - "did": "did:web:a.agents.localhost%3A3000", - "collection": "com.vibescrobble.scrobble", - "rkey": "3lq", - "uri": "at://did:web:a/com.vibescrobble.scrobble/3lq", - "cid": "bafyreiarecord", - "commit": "bafyreiacommit", - "rev": "3lqmnopqrstuv", - "time": "2026-08-25T10:00:00Z", - "record": {"text": "reading the cart", "emoji": "🔍"}, - }) - .to_string(), - }; - let commit = frame.parse::().expect("a commit frame"); - assert_eq!(commit.cursor(), "18e4a2f:12"); - assert_eq!(commit.record["emoji"], "🔍"); - assert_eq!(commit.commit, "bafyreiacommit"); - } - - /// A server that names no commit is still readable. An index reads - /// servers it did not build, and refusing a frame for a missing field - /// would make one older server stop this one dead. - #[test] - fn a_frame_from_a_server_that_names_no_commit_still_parses() { - let frame = Frame { - id: None, - event: Some("commit".to_owned()), - data: serde_json::json!({ - "seq": 1, - "instance": "18e4a2f", - "did": "did:web:a", - "collection": "com.vibescrobble.scrobble", - "rkey": "3lq", - "uri": "at://did:web:a/com.vibescrobble.scrobble/3lq", - "record": {}, - }) - .to_string(), - }; - let commit = frame.parse::().expect("a commit frame"); - assert!(commit.commit.is_empty()); - } - - #[test] - fn an_info_frame_this_index_has_not_heard_of_is_not_a_gap() { - let info: StreamInfo = - serde_json::from_str(r#"{"name":"Rebalancing","instance":"x"}"#).expect("readable"); - assert_ne!(info.name, OUTDATED); - assert_eq!(info.oldest, 0); - } - - #[test] - fn an_identity_frame_carries_the_claimed_handle() { - let frame = Frame { - id: None, - event: Some("identity".to_owned()), - data: serde_json::json!({ - "seq": 7, - "instance": "18e4a2f", - "did": "did:web:a.agents.localhost%3A3000", - "time": "2026-09-08T10:00:00Z", - "handle": "a.agents.localhost", - }) - .to_string(), - }; - let identity = frame.parse::().expect("an identity frame"); - assert_eq!(identity.cursor(), "18e4a2f:7"); - assert_eq!(identity.handle.as_deref(), Some("a.agents.localhost")); - } - - #[test] - fn an_identity_frame_naming_no_handle_still_parses() { - let frame = Frame { - id: None, - event: Some("identity".to_owned()), - data: serde_json::json!({ - "seq": 8, - "instance": "18e4a2f", - "did": "did:web:a.agents.localhost%3A3000", - }) - .to_string(), - }; - let identity = frame.parse::().expect("an identity frame"); - assert_eq!(identity.handle, None); - } - - #[test] - fn an_active_account_frame_is_not_deleted() { - let frame = Frame { - id: None, - event: Some("account".to_owned()), - data: serde_json::json!({ - "seq": 9, - "instance": "18e4a2f", - "did": "did:web:a.agents.localhost%3A3000", - "time": "2026-09-08T10:00:00Z", - "active": true, - }) - .to_string(), - }; - let account = frame.parse::().expect("an account frame"); - assert_eq!(account.cursor(), "18e4a2f:9"); - assert!(account.active); - assert!(!account.is_deleted()); - } - - #[test] - fn a_deleted_account_frame_reads_as_deleted() { - let frame = Frame { - id: None, - event: Some("account".to_owned()), - data: serde_json::json!({ - "seq": 10, - "instance": "18e4a2f", - "did": "did:web:a.agents.localhost%3A3000", - "time": "2026-09-08T10:00:00Z", - "active": false, - "status": "deleted", - }) - .to_string(), - }; - let account = frame.parse::().expect("an account frame"); - assert!(account.is_deleted()); - } - - /// A deactivated or suspended account is inactive and reversible, which - /// is not the same as gone: [`Account::is_deleted`] must not conflate the - /// two, or an account this index should still hold would be tombstoned on - /// the strength of a status its own server could reverse. - #[test] - fn a_deactivated_account_is_inactive_but_not_deleted() { - let frame = Frame { - id: None, - event: Some("account".to_owned()), - data: serde_json::json!({ - "seq": 11, - "instance": "18e4a2f", - "did": "did:web:a.agents.localhost%3A3000", - "time": "2026-09-08T10:00:00Z", - "active": false, - "status": "deactivated", - }) - .to_string(), - }; - let account = frame.parse::().expect("an account frame"); - assert!(!account.active); - assert!(!account.is_deleted()); + fn a_tls_server_is_read_over_a_tls_socket() { + let server = ServerRef::parse("https://pds.example").expect("valid"); + assert!(subscribe_url(&server, None).starts_with("wss://pds.example/")); } } diff --git a/crates/vibescrobble-index/src/frames.rs b/crates/vibescrobble-index/src/frames.rs new file mode 100644 index 00000000..55af6f45 --- /dev/null +++ b/crates/vibescrobble-index/src/frames.rs @@ -0,0 +1,427 @@ +//! `com.atproto.sync.subscribeRepos` frames, decoded. +//! +//! One WebSocket binary message is two DAG-CBOR values back to back: a +//! header saying what kind of message it is, then the body the lexicon gives +//! that kind. Nothing separates the two; the body starts where the header's +//! decode stopped. +//! +//! A `#commit` body carries its records as a CAR — the commit block, the +//! tree nodes on one path and the record itself — and the record this index +//! wants is the block the op's `cid` names. Nothing here walks the tree or +//! checks the signature: the index reads records, and the blocks that would +//! prove one are a request away at `com.atproto.sync.getRecord`. +//! +//! Declared here rather than taken from a server's crate on purpose: an +//! index reading somebody else's server must not compile against that +//! server's types, or it can only ever read servers built from one +//! repository. `vectors/subscribe-repos/` holds frames recorded from a real +//! server, which is what keeps this reader honest instead. + +use std::collections::BTreeMap; +use std::io::Cursor; + +use atproto_dasl::{CarBlock, CarHeader, Ipld}; +use base64::Engine; +use serde_json::Value; + +/// The method a server answers on. +pub const NSID: &str = "com.atproto.sync.subscribeRepos"; + +/// The `#info` name that says the cursor is out of reach. +pub const OUTDATED_CURSOR: &str = "OutdatedCursor"; + +/// The error a server answers a cursor ahead of its stream with. +pub const FUTURE_CURSOR: &str = "FutureCursor"; + +/// One message off the stream. +#[derive(Debug, Clone, PartialEq)] +pub enum Message { + /// `#commit`: one signed commit and the writes in it. + Commit(CommitFrame), + /// `#identity`: an account's document or handle changed. + Identity(Identity), + /// `#account`: an account appeared, went away or changed status. + Account(Account), + /// `#info`: something the server has to say about the connection. + Info(Info), + /// An error frame. The server closes the connection after it. + Error { + /// The lexicon's error name, such as [`FUTURE_CURSOR`]. + error: String, + /// What the server said about it. + message: String, + }, + /// A message type this index has no name for. + /// + /// Carried rather than refused, sequence number and all: a server that + /// grows a kind of message must not strand an older consumer. + Other { + /// The header's `t`. + t: String, + /// The body's `seq`, when it has one. + seq: Option, + }, +} + +impl Message { + /// The sequence number the message carries, if it is numbered. + pub fn seq(&self) -> Option { + match self { + Self::Commit(commit) => Some(commit.seq), + Self::Identity(identity) => Some(identity.seq), + Self::Account(account) => Some(account.seq), + Self::Other { seq, .. } => *seq, + Self::Info(_) | Self::Error { .. } => None, + } + } + + /// The repository the message is about, if it names one. + pub fn did(&self) -> Option<&str> { + match self { + Self::Commit(commit) => Some(&commit.did), + Self::Identity(identity) => Some(&identity.did), + Self::Account(account) => Some(&account.did), + Self::Info(_) | Self::Error { .. } | Self::Other { .. } => None, + } + } +} + +/// One signed commit and the writes in it. +#[derive(Debug, Clone, PartialEq)] +pub struct CommitFrame { + /// Where the frame sits in the stream. The cursor to resume after. + pub seq: u64, + /// The repository written to. + pub did: String, + /// The CID of the signed commit. What a repository is reconciled on. + pub commit: String, + /// The revision that commit is at. + pub rev: String, + /// The revision before it, when the server named one. + pub since: Option, + /// When the server emitted the frame, RFC 3339. + pub time: String, + /// The writes, in the order the commit lists them. + pub writes: Vec, +} + +/// One record a commit created, changed or removed. +#[derive(Debug, Clone, PartialEq)] +pub struct Write { + /// `create`, `update` or `delete`. + pub action: String, + /// The collection the record is in. + pub collection: String, + /// The record key. + pub rkey: String, + /// The AT-URI naming the record. + pub uri: String, + /// The CID of the record, for anything but a delete. + pub cid: Option, + /// The record, when the commit carried it. + /// + /// A delete carries none. Neither does a commit the server marked + /// `tooBig`, whose blocks are left out of the frame and have to be read + /// from the repository. + pub record: Option, +} + +/// An account's document or handle changed. +#[derive(Debug, Clone, PartialEq)] +pub struct Identity { + /// Where the frame sits in the stream. + pub seq: u64, + /// The account. + pub did: String, + /// When the server emitted the frame, RFC 3339. + pub time: String, + /// The handle it now has, when the server said. + pub handle: Option, +} + +/// An account appeared, went away or changed status. +#[derive(Debug, Clone, PartialEq)] +pub struct Account { + /// Where the frame sits in the stream. + pub seq: u64, + /// The account. + pub did: String, + /// When the server emitted the frame, RFC 3339. + pub time: String, + /// Whether its repository can be read. + pub active: bool, + /// Why not, when it cannot: `deleted`, `deactivated`, `suspended` or + /// `takendown`. An open set, carried rather than interpreted. + pub status: Option, +} + +impl Account { + /// Whether this says the account has ended for good. + /// + /// Deliberately narrower than `!active`: a deactivated or suspended + /// account is not being served right now and may still come back, which + /// is not the same as gone. [`crate::identity::Registry::ended`] + /// tombstones an identity on the strength of this, so it should only + /// answer `true` when the account is really not coming back. + #[must_use] + pub fn is_deleted(&self) -> bool { + !self.active && self.status.as_deref() == Some("deleted") + } +} + +/// Something the server has to say about the connection. +#[derive(Debug, Clone, PartialEq)] +pub struct Info { + /// [`OUTDATED_CURSOR`], or something newer. + pub name: String, + /// What it said, for a log line. + pub message: String, +} + +/// Why a frame could not be read. +#[derive(Debug, thiserror::Error)] +pub enum DecodeError { + /// The header or the body is not DAG-CBOR. + #[error("not DAG-CBOR: {0}")] + Cbor(#[from] atproto_dasl::DecodeError), + /// The commit's blocks are not a CAR. + #[error("the commit's blocks are not a CAR: {0}")] + Car(#[from] atproto_dasl::CarError), + /// The frame decodes but is not shaped as the lexicon says. + #[error("{0}")] + Shape(String), +} + +/// The fields of one DAG-CBOR map. +type Fields = BTreeMap; + +/// Reads one frame: the header, then the body the header's type calls for. +pub fn decode(bytes: &[u8]) -> Result { + // Lenient on purpose. DAG-CBOR is canonical by definition, but an index + // reads servers it did not build, and an integer encoded a byte too wide + // is not a reason to stop reading one. + let mut cursor = Cursor::new(bytes); + let header: Ipld = atproto_dasl::from_reader_non_strict(&mut cursor)?; + let body: Ipld = atproto_dasl::from_slice_non_strict(&bytes[cursor.position() as usize..])?; + let mut header = fields(header, "the header")?; + let mut body = fields(body, "the body")?; + + let op = integer(&mut header, "op").ok_or_else(|| shape("the header names no op"))?; + if op < 0 { + return Ok(Message::Error { + error: string(&mut body, "error").unwrap_or_default(), + message: string(&mut body, "message").unwrap_or_default(), + }); + } + let t = string(&mut header, "t").ok_or_else(|| shape("the header names no type"))?; + Ok(match t.as_str() { + "#commit" => Message::Commit(commit(body)?), + "#identity" => Message::Identity(Identity { + seq: seq(&mut body)?, + did: required(&mut body, "did")?, + time: string(&mut body, "time").unwrap_or_default(), + handle: string(&mut body, "handle"), + }), + "#account" => Message::Account(Account { + seq: seq(&mut body)?, + did: required(&mut body, "did")?, + time: string(&mut body, "time").unwrap_or_default(), + active: boolean(&mut body, "active").unwrap_or(false), + status: string(&mut body, "status"), + }), + "#info" => Message::Info(Info { + name: required(&mut body, "name")?, + message: string(&mut body, "message").unwrap_or_default(), + }), + _ => Message::Other { + t, + seq: integer(&mut body, "seq").and_then(|seq| u64::try_from(seq).ok()), + }, + }) +} + +/// The `#commit` body, with each op's record pulled out of the CAR. +fn commit(mut body: Fields) -> Result { + let seq = seq(&mut body)?; + let did = required(&mut body, "repo")?; + let too_big = boolean(&mut body, "tooBig").unwrap_or(false); + let blocks = if too_big { + BTreeMap::new() + } else { + car_blocks(&bytes(&mut body, "blocks").unwrap_or_default())? + }; + + let mut writes = Vec::new(); + for op in list(&mut body, "ops") { + let mut op = fields(op, "an op")?; + let path = required(&mut op, "path")?; + let Some((collection, rkey)) = path.split_once('/') else { + return Err(shape(format!("op path {path:?} is not collection/rkey"))); + }; + let cid = link(&mut op, "cid"); + let record = cid + .as_deref() + .and_then(|cid| blocks.get(cid)) + .map(|data| atproto_dasl::from_slice_non_strict::(data)) + .transpose()? + .map(json); + writes.push(Write { + action: string(&mut op, "action").unwrap_or_default(), + collection: collection.to_owned(), + rkey: rkey.to_owned(), + uri: format!("at://{did}/{path}"), + cid, + record, + }); + } + + Ok(CommitFrame { + seq, + did, + commit: link(&mut body, "commit").unwrap_or_default(), + rev: string(&mut body, "rev").unwrap_or_default(), + since: string(&mut body, "since"), + time: string(&mut body, "time").unwrap_or_default(), + writes, + }) +} + +/// Every block in a CAR, by CID. +/// +/// The header's roots are read past rather than used: the block an op names +/// is found by its CID, and which block is the commit is already said by the +/// body's `commit` field. +fn car_blocks(car: &[u8]) -> Result>, DecodeError> { + let mut blocks = BTreeMap::new(); + if car.is_empty() { + return Ok(blocks); + } + let (_header, mut at) = CarHeader::from_bytes(car)?; + while at < car.len() { + let (block, used) = CarBlock::from_bytes(&car[at..])?; + at += used; + blocks.insert(block.cid.to_string(), block.data); + } + Ok(blocks) +} + +/// A DAG-CBOR value as atproto's JSON spells it. +/// +/// A link becomes `{"$link": cid}` and bytes become `{"$bytes": base64}`, +/// which is the representation `com.atproto.repo.getRecord` answers with — +/// so a record off the stream and the same record read back over XRPC read +/// the same. +fn json(value: Ipld) -> Value { + match value { + Ipld::Null => Value::Null, + Ipld::Bool(bool) => Value::Bool(bool), + Ipld::Integer(int) => i64::try_from(int) + .map(Value::from) + .or_else(|_| u64::try_from(int).map(Value::from)) + .unwrap_or(Value::Null), + Ipld::Float(float) => serde_json::Number::from_f64(float) + .map(Value::Number) + .unwrap_or(Value::Null), + Ipld::String(string) => Value::String(string), + Ipld::Bytes(bytes) => serde_json::json!({ + "$bytes": base64::engine::general_purpose::STANDARD.encode(bytes) + }), + Ipld::List(items) => Value::Array(items.into_iter().map(json).collect()), + Ipld::Map(fields) => Value::Object( + fields + .into_iter() + .map(|(key, value)| (key, json(value))) + .collect(), + ), + Ipld::Link(cid) => serde_json::json!({ "$link": cid.to_string() }), + } +} + +fn shape(reason: impl Into) -> DecodeError { + DecodeError::Shape(reason.into()) +} + +fn fields(value: Ipld, what: &str) -> Result { + match value { + Ipld::Map(fields) => Ok(fields), + _ => Err(shape(format!("{what} is not a map"))), + } +} + +fn string(fields: &mut Fields, key: &str) -> Option { + match fields.remove(key) { + Some(Ipld::String(string)) => Some(string), + _ => None, + } +} + +fn required(fields: &mut Fields, key: &str) -> Result { + string(fields, key).ok_or_else(|| shape(format!("no {key}"))) +} + +fn integer(fields: &mut Fields, key: &str) -> Option { + match fields.remove(key) { + Some(Ipld::Integer(int)) => Some(int), + _ => None, + } +} + +fn seq(fields: &mut Fields) -> Result { + integer(fields, "seq") + .and_then(|seq| u64::try_from(seq).ok()) + .ok_or_else(|| shape("no seq")) +} + +fn boolean(fields: &mut Fields, key: &str) -> Option { + match fields.remove(key) { + Some(Ipld::Bool(bool)) => Some(bool), + _ => None, + } +} + +fn link(fields: &mut Fields, key: &str) -> Option { + match fields.remove(key) { + Some(Ipld::Link(cid)) => Some(cid.to_string()), + _ => None, + } +} + +fn bytes(fields: &mut Fields, key: &str) -> Option> { + match fields.remove(key) { + Some(Ipld::Bytes(bytes)) => Some(bytes), + _ => None, + } +} + +fn list(fields: &mut Fields, key: &str) -> Vec { + match fields.remove(key) { + Some(Ipld::List(items)) => items, + _ => Vec::new(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_frame_that_is_not_a_frame_is_refused_by_name() { + // A reserved initial byte is not CBOR at all. + let err = decode(&[0xff, 0x00]).expect_err("not cbor"); + assert!(matches!(err, DecodeError::Cbor(_)), "{err}"); + // Text happens to read as a CBOR string, which is not a header. + let err = decode(b"event: commit\n").expect_err("not a header"); + assert!(matches!(err, DecodeError::Shape(_)), "{err}"); + } + + #[test] + fn a_link_and_bytes_read_as_atproto_spells_them_in_json() { + let cid = CarBlock::from_data(vec![0xa0]).cid; + let value = json(Ipld::Map(BTreeMap::from([ + ("ref".to_owned(), Ipld::Link(cid.into())), + ("raw".to_owned(), Ipld::Bytes(vec![1, 2, 3])), + ]))); + assert_eq!(value["ref"]["$link"], cid.to_string()); + assert_eq!(value["raw"]["$bytes"], "AQID"); + } +} diff --git a/crates/vibescrobble-index/src/jetstream.rs b/crates/vibescrobble-index/src/jetstream.rs index 80ba8aa2..7e41bec1 100644 --- a/crates/vibescrobble-index/src/jetstream.rs +++ b/crates/vibescrobble-index/src/jetstream.rs @@ -6,12 +6,13 @@ //! //! # Why not the relay's own stream //! -//! Because a `#commit` frame carries its records as a CAR slice, and getting a -//! scrobble out of one means decoding DAG-CBOR, walking blocks by CID and -//! checking them against the commit — work this project would be doing to -//! arrive at a JSON object that Jetstream already hands over. The index is a -//! reader of records, not a verifier of repositories: the relay verifies, and -//! doing it twice would be a second implementation to keep correct. +//! Because a relay's `subscribeRepos` carries every collection on the +//! network, and Jetstream filters it to the ones a consumer names before +//! anything crosses the wire. [`crate::firehose`] reads one server's own +//! `subscribeRepos` directly — DAG-CBOR, CAR and all — and one server's +//! traffic is small enough to read whole; the whole network's is not, and +//! a relay verifies what it forwards, so what Jetstream hands over is +//! already checked once. //! //! What is given up is the relay's own sequence numbers, and nothing else: //! every Jetstream event carries a `seq` of its own, which is the ordering diff --git a/crates/vibescrobble-index/src/lib.rs b/crates/vibescrobble-index/src/lib.rs index ac605dc6..858b0a6a 100644 --- a/crates/vibescrobble-index/src/lib.rs +++ b/crates/vibescrobble-index/src/lib.rs @@ -6,10 +6,11 @@ //! part of that — no HTTP surface of its own, no opinion about what is drawn //! with what it collects. //! -//! # An account arrives on the firehose, not only by writing something +//! # An account arrives on the stream, not only by writing something //! -//! `/firehose` can carry three things: a commit, whose repository — once -//! checked — is an account this index knows about; an `#account` frame, +//! A server's `com.atproto.sync.subscribeRepos` carries three things: a +//! `#commit`, whose repository — once checked — is an account this index +//! knows about; an `#account` frame, //! saying a repository exists, ended, or changed status, before or instead of //! anything being written to it; and an `#identity` frame, a handle its //! server now claims for it. [`intake::Intake`] is where all three become the @@ -100,6 +101,7 @@ pub mod did; pub mod directory; pub mod feed; pub mod firehose; +pub mod frames; pub mod handle; pub mod http; pub mod identity; @@ -122,7 +124,7 @@ pub use backfill::{backfill, Filled}; pub use did::{canonical_did, AgentDid, DidError, DocumentExt, HandleError}; pub use directory::{Authority, Discovered, ServerDirectory, StaticDirectory, Target}; pub use feed::{read_record, FeedPage, Scrobble}; -pub use firehose::{subscribe, subscribe_from, Commit, StreamInfo, Update}; +pub use firehose::{subscribe, subscribe_from, Commit, Update}; pub use firehose::{Account as FirehoseAccount, Identity as FirehoseIdentity}; pub use handle::{verify as verify_handle, HandleFailure}; pub use http::{HttpGet, HttpResponse, ReqwestGet, TransportError}; diff --git a/crates/vibescrobble-index/src/replay.rs b/crates/vibescrobble-index/src/replay.rs index 5858201b..eb01b4d3 100644 --- a/crates/vibescrobble-index/src/replay.rs +++ b/crates/vibescrobble-index/src/replay.rs @@ -1,4 +1,4 @@ -//! A recorded firehose, and a server that plays it back. +//! A recorded stream, and a server that plays it back. //! //! # What this is for //! @@ -6,7 +6,8 @@ //! server and a swarm in front of it: three terminals and a population, before //! anything downstream has a byte to look at. The canvas already ships a mock //! for exactly that reason. This is the same idea one layer down — a recording -//! of a real stream, played back with nothing else running. +//! of a real `com.atproto.sync.subscribeRepos`, played back with nothing else +//! running. //! //! What it buys over keeping a `--data` directory is that a recording is a //! *sequence*, not a state. The same frames arrive in the same order every @@ -16,67 +17,81 @@ //! //! # The format is one frame per line //! -//! JSON Lines, each carrying the server-sent event's name and its data. Not a -//! clever container: the point is that a person can open a recording, delete -//! the frames that are not the bug, and hand the file to somebody else. +//! JSON Lines. A frame is DAG-CBOR, so each line carries it as base64 — the +//! bytes exactly as they came off the wire, which is what makes a playback +//! the same stream and not a re-enactment of one. Beside the bytes sit the +//! frame's sequence number and the repository it is about, so that a person +//! trimming a recording can tell the lines apart. Those two are checked +//! against the frame when a recording is read: a line whose annotation +//! disagrees with its bytes is refused by line number rather than believed. //! //! # What it does not pretend to be //! -//! A personal data server. It answers the firehose and the account listing an -//! index asks for on every pass, and nothing else — no provisioning, no -//! repositories, no blobs. An index following it fills in what it can and -//! retries what it cannot, which is what it does against a real server that is -//! part way up. +//! A personal data server. It answers the stream and `/health`, and nothing +//! else — no repositories, no records to list, no blobs. An index following +//! it fills in what it can and retries what it cannot, which is what it does +//! against a real server that is part way up. use std::sync::Arc; +use std::time::Duration; +use axum::extract::ws::{Message as Ws, WebSocket, WebSocketUpgrade}; use axum::extract::{Query, State}; -use axum::response::sse::Sse; use axum::response::IntoResponse; use axum::routing::get; use axum::Router; +use base64::Engine; use serde::{Deserialize, Serialize}; -use crate::sse::Frame; +use crate::frames::{decode, DecodeError, NSID}; /// One frame as a recording holds it. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct Recorded { - /// The `event:` name the frame carried, if any. + /// The frame's sequence number, when it is numbered. #[serde(default, skip_serializing_if = "Option::is_none")] - pub event: Option, - /// Everything from the frame's `data:` lines. - pub data: String, + pub seq: Option, + /// The repository the frame is about, when it names one. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub did: Option, + /// The frame, base64. + pub frame: String, } -impl From for Recorded { - fn from(frame: Frame) -> Self { - Self { - event: frame.event, - data: frame.data, - } +impl Recorded { + /// Keeps one frame as it came off the wire, annotated from its contents. + pub fn from_frame(bytes: &[u8]) -> Result { + let message = decode(bytes)?; + Ok(Self { + seq: message.seq(), + did: message.did().map(str::to_owned), + frame: base64::engine::general_purpose::STANDARD.encode(bytes), + }) } -} -impl Recorded { - /// The sequence number this frame carries, if it is a commit. - /// - /// Read out of the data rather than stored beside it, so that a recording - /// somebody has edited by hand cannot disagree with itself. - pub fn seq(&self) -> Option { - serde_json::from_str::(&self.data) - .ok()? - .get("seq")? - .as_u64() + /// The frame's bytes. + pub fn bytes(&self) -> Result, base64::DecodeError> { + base64::engine::general_purpose::STANDARD.decode(&self.frame) } - /// The DID this frame is about, if it names one. - pub fn did(&self) -> Option { - serde_json::from_str::(&self.data) - .ok()? - .get("did")? - .as_str() - .map(str::to_owned) + /// Fills in the annotations from the frame, or refuses ones that disagree + /// with it. + fn check(self) -> Result { + let bytes = self.bytes().map_err(|err| format!("not base64: {err}"))?; + let from_frame = Self::from_frame(&bytes).map_err(|err| err.to_string())?; + if self.seq.is_some_and(|seq| Some(seq) != from_frame.seq) { + return Err(format!( + "says seq {:?} but the frame carries {:?}", + self.seq, from_frame.seq + )); + } + if self.did.is_some() && self.did != from_frame.did { + return Err(format!( + "says did {:?} but the frame carries {:?}", + self.did, from_frame.did + )); + } + Ok(from_frame) } } @@ -99,11 +114,17 @@ impl Recording { if line.trim().is_empty() { continue; } - let frame = serde_json::from_str(line).map_err(|source| RecordingError::Line { - line: number + 1, - source, + let line_number = number + 1; + let recorded: Recorded = + serde_json::from_str(line).map_err(|err| RecordingError::Line { + line: line_number, + reason: err.to_string(), + })?; + let recorded = recorded.check().map_err(|reason| RecordingError::Line { + line: line_number, + reason, })?; - frames.push(frame); + frames.push(recorded); } Ok(Self { frames }) } @@ -120,16 +141,15 @@ impl Recording { /// Every account the recording mentions, in the order first seen. /// - /// An index asks a server what accounts it has on every survey pass. A - /// recording knows only the accounts that wrote something, which is fewer - /// than a real server would list and is the honest answer: nothing here - /// saw the others. + /// A recording knows only the accounts that were announced or wrote + /// something, which is fewer than a real server would list and is the + /// honest answer: nothing here saw the others. pub fn accounts(&self) -> Vec { let mut seen = Vec::new(); for frame in &self.frames { - if let Some(did) = frame.did() { - if !seen.contains(&did) { - seen.push(did); + if let Some(did) = &frame.did { + if !seen.contains(did) { + seen.push(did.clone()); } } } @@ -140,7 +160,7 @@ impl Recording { /// /// An index refuses to survey a server whose `/health` does not name a /// zone — that is how it tells a vibescrobble server from anything else - /// answering on a port — so a replay has to name one or it is a firehose + /// answering on a port — so a replay has to name one or it is a stream /// nobody will read past. /// /// It is derived rather than recorded: a `did:web` is the zone with a @@ -192,19 +212,17 @@ impl Recording { /// The frames after `cursor`, or all of them when there is none. /// - /// A cursor is `:` as the firehose defines it, and only the - /// sequence number is used: a recording is one instance by construction. - pub fn after(&self, cursor: Option<&str>) -> &[Recorded] { - let Some(seq) = cursor - .and_then(|cursor| cursor.rsplit(':').next()) - .and_then(|seq| seq.parse::().ok()) - else { + /// A cursor is the sequence number of the last frame applied, as + /// `subscribeRepos` defines it. A frame with no number — an `#info` — is + /// skipped along with whatever numbered frame it preceded. + pub fn after(&self, cursor: Option) -> &[Recorded] { + let Some(cursor) = cursor else { return &self.frames; }; let start = self .frames .iter() - .position(|frame| frame.seq().is_some_and(|found| found > seq)) + .position(|frame| frame.seq.is_some_and(|seq| seq > cursor)) .unwrap_or(self.frames.len()); &self.frames[start..] } @@ -214,12 +232,12 @@ impl Recording { #[derive(Debug, thiserror::Error)] pub enum RecordingError { /// One line is not a frame. - #[error("line {line} is not a recorded frame: {source}")] + #[error("line {line} is not a recorded frame: {reason}")] Line { /// Which line, counting from one. line: usize, - /// What serde said. - source: serde_json::Error, + /// What was wrong with it. + reason: String, }, } @@ -232,9 +250,9 @@ pub struct Pace { impl Pace { /// The gap between two frames, if there is one. - pub fn gap(self) -> Option { + pub fn gap(self) -> Option { let rate = self.per_second?; - (rate > 0.0).then(|| std::time::Duration::from_secs_f64(1.0 / rate)) + (rate > 0.0).then(|| Duration::from_secs_f64(1.0 / rate)) } } @@ -245,30 +263,56 @@ struct Playback { pace: Pace, } -/// A router that serves `recording` as a firehose. +/// A router that serves `recording` as a server's stream. pub fn router(recording: Recording, pace: Pace) -> Router { let state = Playback { recording: Arc::new(recording), pace, }; Router::new() - .route("/firehose", get(firehose)) + .route(&format!("/xrpc/{NSID}"), get(subscribe_repos)) .route("/health", get(health)) - .route("/xrpc/bot.did.listAgents", get(list_agents)) .with_state(state) } -/// The recording, as a server-sent event stream. -async fn firehose( +/// What `subscribeRepos` accepts. +#[derive(Debug, Deserialize)] +struct Resume { + /// The last sequence number the consumer applied. + cursor: Option, +} + +/// The recording, as one WebSocket of frames. +async fn subscribe_repos( State(playback): State, - Query(params): Query>, + Query(resume): Query, + upgrade: WebSocketUpgrade, ) -> impl IntoResponse { - let cursor = params.get("cursor").cloned(); - let frames: Vec = playback.recording.after(cursor.as_deref()).to_vec(); + let frames: Vec = playback.recording.after(resume.cursor).to_vec(); let gap = playback.pace.gap(); + upgrade.on_upgrade(move |socket| play(socket, frames, gap)) +} - let stream = async_stream::stream(frames, gap); - Sse::new(stream) +/// Hands `frames` over, waiting `gap` between them, then closes. +/// +/// The end of a recording is the end of the stream, and a subscriber that +/// reconnects gets it again from its cursor — which is what a real server +/// does when it has nothing more. +async fn play(mut socket: WebSocket, frames: Vec, gap: Option) { + for frame in frames { + let Ok(bytes) = frame.bytes() else { + // Checked when the recording was read; unreachable, and a frame + // that cannot be sent is not a reason to stop the rest. + continue; + }; + if socket.send(Ws::Binary(bytes.into())).await.is_err() { + return; + } + if let Some(gap) = gap { + tokio::time::sleep(gap).await; + } + } + socket.send(Ws::Close(None)).await.ok(); } /// What this is and how much of it there is. @@ -284,97 +328,45 @@ async fn health(State(playback): State) -> impl IntoResponse { })) } -/// The accounts the recording mentions, shaped as the real listing is. -async fn list_agents(State(playback): State) -> impl IntoResponse { - let agents: Vec = playback - .recording - .accounts() - .into_iter() - .map(|did| serde_json::json!({ "did": did })) - .collect(); - axum::Json(serde_json::json!({ "agents": agents })) -} - -/// The stream half, kept apart because it is the only part that is a `Stream`. -mod async_stream { - use std::convert::Infallible; - use std::future::Future; - use std::pin::Pin; - use std::task::{Context, Poll}; - use std::time::Duration; - - use axum::response::sse::Event; - use futures_core::Stream; - - use super::Recorded; - - /// A stream that hands over `frames`, waiting `gap` between them. - /// - /// Hand-rolled for the reason [`crate::sse::read_frames`] is: the crate - /// that would provide this also provides a great deal that nothing here - /// uses, and the state machine is one index and one sleep. - pub(super) fn stream( - frames: Vec, - gap: Option, - ) -> impl Stream> + Send + 'static { - Playing { - frames, - at: 0, - gap, - waiting: None, - } - } - - struct Playing { - frames: Vec, - at: usize, - gap: Option, - waiting: Option>>, - } - - impl Stream for Playing { - type Item = Result; - - fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - let this = self.get_mut(); - if let Some(sleep) = this.waiting.as_mut() { - match sleep.as_mut().poll(cx) { - Poll::Pending => return Poll::Pending, - Poll::Ready(()) => this.waiting = None, - } - } - let Some(frame) = this.frames.get(this.at) else { - // The end of a recording is the end of the stream, and a - // subscriber that reconnects gets it again from its cursor — - // which is what a real server does when it has nothing more. - return Poll::Ready(None); - }; - this.at += 1; - let mut event = Event::default().data(frame.data.clone()); - if let Some(name) = &frame.event { - event = event.event(name.clone()); - } - if let Some(gap) = this.gap { - this.waiting = Some(Box::pin(tokio::time::sleep(gap))); - } - Poll::Ready(Some(Ok(event))) - } - } -} - #[cfg(test)] mod tests { use super::*; + use atproto_dasl::Ipld; + use std::collections::BTreeMap; + + /// An `#account` frame, which is the smallest numbered frame there is. + fn account_frame(seq: u64, did: &str) -> Vec { + let header = Ipld::Map(BTreeMap::from([ + ("op".to_owned(), Ipld::Integer(1)), + ("t".to_owned(), Ipld::String("#account".to_owned())), + ])); + let body = Ipld::Map(BTreeMap::from([ + ("active".to_owned(), Ipld::Bool(true)), + ("did".to_owned(), Ipld::String(did.to_owned())), + ("seq".to_owned(), Ipld::Integer(seq.into())), + ( + "time".to_owned(), + Ipld::String("2026-09-16T12:00:00Z".to_owned()), + ), + ])); + let mut bytes = atproto_dasl::to_vec(&header).expect("a header encodes"); + bytes.extend(atproto_dasl::to_vec(&body).expect("a body encodes")); + bytes + } + + fn line(seq: u64, did: &str) -> String { + serde_json::to_string(&Recorded::from_frame(&account_frame(seq, did)).expect("readable")) + .expect("serializes") + } fn recording() -> Recording { - Recording::parse( - r#"{"event":"info","data":"{\"instance\":\"abc\",\"oldest\":1}"} -{"event":"commit","data":"{\"seq\":1,\"did\":\"did:web:a\"}"} -{"event":"commit","data":"{\"seq\":2,\"did\":\"did:web:b\"}"} -{"event":"commit","data":"{\"seq\":3,\"did\":\"did:web:a\"}"} -"#, - ) - .expect("the fixture parses") + let text = format!( + "{}\n{}\n{}\n", + line(1, "did:web:a.agents.localhost"), + line(2, "did:web:b.agents.localhost"), + line(3, "did:web:a.agents.localhost"), + ); + Recording::parse(&text).expect("the fixture parses") } #[test] @@ -386,48 +378,67 @@ mod tests { #[test] fn a_blank_line_is_not_a_frame() { - let parsed = Recording::parse("\n\n{\"data\":\"{}\"}\n\n").expect("parses"); + let text = format!("\n\n{}\n\n", line(1, "did:web:a.agents.localhost")); + let parsed = Recording::parse(&text).expect("parses"); assert_eq!(parsed.frames.len(), 1); } #[test] fn a_line_that_will_not_parse_says_which_line() { - let err = Recording::parse("{\"data\":\"{}\"}\nnot json\n").expect_err("refused"); + let text = format!("{}\nnot json\n", line(1, "did:web:a.agents.localhost")); + let err = Recording::parse(&text).expect_err("refused"); assert!(matches!(err, RecordingError::Line { line: 2, .. }), "{err}"); } + /// The annotations are for people; the bytes are the frame. A line whose + /// two halves disagree is a line somebody edited by hand, and believing + /// either half would make the recording lie about itself. + #[test] + fn an_annotation_that_disagrees_with_its_frame_is_refused() { + let mut edited: Recorded = + serde_json::from_str(&line(1, "did:web:a.agents.localhost")).expect("parses"); + edited.seq = Some(9); + let text = serde_json::to_string(&edited).expect("serializes"); + let err = Recording::parse(&text).expect_err("refused"); + assert!(err.to_string().contains("seq"), "{err}"); + } + + #[test] + fn a_frame_without_annotations_is_annotated_on_read() { + let bare = serde_json::json!({ + "frame": base64::engine::general_purpose::STANDARD + .encode(account_frame(4, "did:web:c.agents.localhost")), + }); + let parsed = Recording::parse(&bare.to_string()).expect("parses"); + assert_eq!(parsed.frames[0].seq, Some(4)); + assert_eq!( + parsed.frames[0].did.as_deref(), + Some("did:web:c.agents.localhost") + ); + } + #[test] fn a_cursor_hands_back_only_what_came_after_it() { let recording = recording(); - let after = recording.after(Some("abc:1")); + let after = recording.after(Some(1)); assert_eq!(after.len(), 2); - assert_eq!(after[0].seq(), Some(2)); + assert_eq!(after[0].seq, Some(2)); } #[test] fn no_cursor_is_the_whole_recording() { - assert_eq!(recording().after(None).len(), 4); + assert_eq!(recording().after(None).len(), 3); } #[test] fn a_cursor_past_the_end_hands_back_nothing() { - assert!(recording().after(Some("abc:99")).is_empty()); - } - - #[test] - fn a_cursor_that_is_not_one_is_ignored_rather_than_refused() { - // A recording is a development tool and a mistyped cursor should show - // you the stream, not an error page. - assert_eq!(recording().after(Some("nonsense")).len(), 4); + assert!(recording().after(Some(99)).is_empty()); } #[test] fn the_zone_comes_off_the_accounts_rather_than_being_declared() { - let recording = Recording::parse( - r#"{"event":"commit","data":"{\"seq\":1,\"did\":\"did:web:marmot.agents.localhost%3A3400\"}"} -"#, - ) - .expect("parses"); + let text = line(1, "did:web:marmot.agents.localhost%3A3400"); + let recording = Recording::parse(&text).expect("parses"); // Without the port: that is how a server declares its own zone, and // an index compares the two. assert_eq!(recording.zone().as_deref(), Some("agents.localhost")); @@ -438,21 +449,15 @@ mod tests { // Not a convenience: an index checks that a server's zone sits under // the host it was pointed at, so a playback anywhere else is refused // however correct its frames are. - let recording = Recording::parse( - r#"{"event":"commit","data":"{\"seq\":1,\"did\":\"did:web:marmot.agents.localhost%3A3400\"}"} -"#, - ) - .expect("parses"); + let text = line(1, "did:web:marmot.agents.localhost%3A3400"); + let recording = Recording::parse(&text).expect("parses"); assert_eq!(recording.listen_hint().as_deref(), Some("127.0.0.1:3400")); } #[test] fn a_recording_with_no_port_suggests_no_address() { - let recording = Recording::parse( - r#"{"event":"commit","data":"{\"seq\":1,\"did\":\"did:web:kestrel.agents.example\"}"} -"#, - ) - .expect("parses"); + let text = line(1, "did:web:kestrel.agents.example"); + let recording = Recording::parse(&text).expect("parses"); assert_eq!(recording.zone().as_deref(), Some("agents.example")); assert_eq!(recording.listen_hint(), None); } @@ -465,8 +470,11 @@ mod tests { } #[test] - fn the_accounts_are_the_ones_that_said_something() { - assert_eq!(recording().accounts(), vec!["did:web:a", "did:web:b"]); + fn the_accounts_are_the_ones_that_were_mentioned() { + assert_eq!( + recording().accounts(), + vec!["did:web:a.agents.localhost", "did:web:b.agents.localhost"] + ); } #[test] @@ -482,7 +490,7 @@ mod tests { per_second: Some(4.0) } .gap(), - Some(std::time::Duration::from_millis(250)) + Some(Duration::from_millis(250)) ); } } diff --git a/crates/vibescrobble-index/src/sse.rs b/crates/vibescrobble-index/src/sse.rs index db367462..8b1639a2 100644 --- a/crates/vibescrobble-index/src/sse.rs +++ b/crates/vibescrobble-index/src/sse.rs @@ -7,8 +7,8 @@ //! Frames are handed over raw, event name and all, because a stream that //! carries more than one kind of thing cannot be told apart by its payload //! alone without every payload growing a discriminant it only needs because -//! the name was thrown away. [`crate::firehose`] is the reader that wants -//! that. +//! the name was thrown away. The query service, mirroring an index's +//! `/events`, is the reader that wants that. use std::time::Duration; @@ -26,9 +26,9 @@ pub const FRAME_BUFFER: usize = 256; /// One frame off a server-sent-event stream, before anything is made of it. /// /// The event name is kept because a stream that carries more than one kind of -/// thing — the firehose carries `info` and `commit` — cannot be told apart by -/// its payload alone without every payload growing a discriminant it only -/// needs because the name was thrown away. +/// thing — an index's `/events` carries view events and an `outdated` — cannot +/// be told apart by its payload alone without every payload growing a +/// discriminant it only needs because the name was thrown away. #[derive(Debug, Clone, PartialEq, Eq)] pub struct Frame { /// The frame's `id`, when it carries one. @@ -51,62 +51,14 @@ impl Frame { } } -/// Follows any server-sent-event endpoint, forwarding frames it can read. -/// -/// Reconnects on its own. A service restarting is an ordinary event in -/// development and should cost its readers a gap, not a process. The gap is -/// real, though: whatever happened while the connection was down is missed, -/// which is why everything that reads one of these also has a way to catch -/// up from a snapshot. -pub async fn stream( - url: String, - out: mpsc::Sender, -) { - let http = stream_client(); - loop { - match stream_once(&http, &url, &out).await { - Ok(()) => debug!(url, "event stream ended"), - Err(err) => warn!(url, %err, "event stream failed"), - } - if out.is_closed() { - return; - } - tokio::time::sleep(RETRY).await; - } -} - -/// One connection's worth of frames, translated into `T`. -/// -/// The reader runs in its own task because a frame has to be handed on while -/// the next one is still arriving; doing both in one task means a `select!` -/// over a pinned stream for no gain over a channel and a spawn. -async fn stream_once( - http: &reqwest::Client, - url: &str, - out: &mpsc::Sender, -) -> Result<(), TransportError> { - let (frames, mut incoming) = mpsc::channel::(FRAME_BUFFER); - let out = out.clone(); - let forwarding = tokio::spawn(async move { - while let Some(frame) = incoming.recv().await { - if let Some(value) = frame.parse::() { - if out.send(value).await.is_err() { - return; - } - } - } - }); - let result = read_frames(http, url, &frames).await; - drop(frames); - forwarding.await.ok(); - result -} - /// Follows an endpoint, forwarding raw frames. /// -/// [`stream`] parses each frame into one fixed type and throws the envelope -/// away. A consumer that needs the frame's `id` — its position in somebody -/// else's order — needs the envelope, which is what this keeps. +/// Raw, envelope and all: a consumer that needs the frame's `id` — its +/// position in somebody else's order — needs the envelope. Reconnects on its +/// own. A service restarting is an ordinary event in development and should +/// cost its readers a gap, not a process; the gap is real, though, which is +/// why everything that reads one of these also has a way to catch up from a +/// snapshot. pub async fn frames(url: String, out: mpsc::Sender) { let http = stream_client(); loop { diff --git a/crates/vibescrobble-index/tests/account_frames.rs b/crates/vibescrobble-index/tests/account_frames.rs index 7b5b4d82..e86547a3 100644 --- a/crates/vibescrobble-index/tests/account_frames.rs +++ b/crates/vibescrobble-index/tests/account_frames.rs @@ -12,7 +12,8 @@ //! //! The transport is a table of fixed answers, mutable mid-test, rather than a //! running personal data server: what a real server puts on the wire is -//! `vectors/firehose`'s job, and this is the consumer's own state machine. +//! `vectors/subscribe-repos`'s job, and this is the consumer's own state +//! machine. use std::collections::HashMap; use std::future::Future; @@ -111,7 +112,6 @@ fn server() -> ServerRef { fn account(did: &str, seq: u64, active: bool, status: Option<&str>) -> Account { Account { seq, - instance: "test-instance".to_owned(), did: did.to_owned(), time: "2026-09-08T10:00:00Z".to_owned(), active, @@ -122,7 +122,6 @@ fn account(did: &str, seq: u64, active: bool, status: Option<&str>) -> Account { fn identity(did: &str, seq: u64, handle: Option<&str>) -> Identity { Identity { seq, - instance: "test-instance".to_owned(), did: did.to_owned(), time: "2026-09-08T10:00:01Z".to_owned(), handle: handle.map(str::to_owned), @@ -132,7 +131,6 @@ fn identity(did: &str, seq: u64, handle: Option<&str>) -> Identity { fn commit(did: &str, seq: u64) -> Commit { Commit { seq, - instance: "test-instance".to_owned(), did: did.to_owned(), collection: vibescrobble_index::nsid::SCROBBLE.to_owned(), rkey: format!("3lqm{seq:08}"), diff --git a/crates/vibescrobble-index/tests/firehose_wire.rs b/crates/vibescrobble-index/tests/firehose_wire.rs index 4056bec3..89571079 100644 --- a/crates/vibescrobble-index/tests/firehose_wire.rs +++ b/crates/vibescrobble-index/tests/firehose_wire.rs @@ -1,124 +1,98 @@ -//! What this index reads off a firehose, against the recorded vectors. +//! What this index reads off `com.atproto.sync.subscribeRepos`, against the +//! recorded vectors. //! -//! The consumer's half of `vectors/firehose/`. It asserts that the recorded -//! bodies deserialize into the values this crate acts on, so a server built -//! somewhere else stays readable and a change to the types here fails against -//! the record rather than in production. -//! -//! Several of the vectors exist only on this side. `commit-sparse` and -//! `info-unknown-name` record what an index must tolerate from a server that -//! is older or newer than itself; `identity`, `identity-no-handle` and the -//! three `account-*` vectors record the shape atproto's own `#identity` and -//! `#account` carry, ahead of any server built from this repository sending -//! them on `/firehose` — see `vectors/firehose/README.md`. +//! The consumer's half of `vectors/subscribe-repos/`. Each vector is one +//! frame as a didbot server put it on the wire, and what is asserted is that +//! the bytes decode into the values this crate acts on — so a server built +//! somewhere else stays readable, and a change to the reader here fails +//! against the record rather than in production. use std::path::PathBuf; +use base64::Engine; use serde_json::Value; -use vibescrobble_index::firehose::{Account, Commit, Identity, StreamInfo}; +use vibescrobble_index::frames::{decode, Message, FUTURE_CURSOR, OUTDATED_CURSOR}; +use vibescrobble_index::replay::Recorded; + +const KESTREL: &str = "did:web:kestrel.agents.localhost%3A3931"; -/// Reads one vector, returning its event name and body. -fn vector(name: &str) -> (String, Value) { +/// Reads one vector, returning its frame's bytes. +fn vector(name: &str) -> Vec { let path = PathBuf::from(env!("CARGO_MANIFEST_DIR")) - .join("../../vectors/firehose") + .join("../../vectors/subscribe-repos") .join(format!("{name}.json")); let text = std::fs::read_to_string(&path) .unwrap_or_else(|err| panic!("reading {}: {err}", path.display())); let doc: Value = serde_json::from_str(&text) .unwrap_or_else(|err| panic!("parsing {}: {err}", path.display())); - let event = doc["event"].as_str().expect("an event name").to_owned(); - (event, doc["data"].clone()) -} - -#[test] -fn the_opening_frame_of_a_live_connection_reads() { - let (event, data) = vector("info-live"); - let info: StreamInfo = serde_json::from_value(data).expect("a readable info frame"); - - assert_eq!(event, "info"); - assert_eq!(info.name, "Live"); - assert_eq!(info.instance, "01JEXAMPLEINSTANCE"); - assert_eq!((info.oldest, info.newest), (0, 0)); + let frame = doc["frame"].as_str().expect("a base64 frame"); + base64::engine::general_purpose::STANDARD + .decode(frame) + .unwrap_or_else(|err| panic!("{}: not base64: {err}", path.display())) } -#[test] -fn an_unreachable_cursor_reads_as_the_name_that_means_a_gap() { - let (_, data) = vector("info-outdated-cursor"); - let info: StreamInfo = serde_json::from_value(data).expect("a readable info frame"); - - // The one name that is not "carry on". The index compares this string, so - // the vector is what stops it drifting from what a server sends. - assert_eq!(info.name, "OutdatedCursor"); - assert_eq!(info.oldest, 41); -} - -#[test] -fn an_info_name_this_index_has_not_heard_of_is_carried_rather_than_refused() { - let (_, data) = vector("info-unknown-name"); - let info: StreamInfo = serde_json::from_value(data).expect("a readable info frame"); - - assert_eq!(info.name, "Rebalancing"); - assert_ne!(info.name, "OutdatedCursor", "this must not read as a gap"); -} - -#[test] -fn a_write_reads_with_every_field_a_server_sent() { - let (event, data) = vector("commit"); - let commit: Commit = serde_json::from_value(data).expect("a readable commit frame"); - - assert_eq!(event, "commit"); - assert_eq!(commit.seq, 42); - assert_eq!(commit.did, "did:web:kestrel.agents.localhost"); - assert_eq!(commit.collection, "com.vibescrobble.scrobble"); - assert_eq!(commit.rkey, "3lqm0000abcd2"); - assert_eq!(commit.cursor(), "01JEXAMPLEINSTANCE:42"); - assert_eq!(commit.record["text"], "reading the firehose reconnect path"); +fn read(name: &str) -> Message { + let bytes = vector(name); + decode(&bytes).unwrap_or_else(|err| panic!("{name}: {err}")) } #[test] -fn a_write_from_a_server_that_predates_four_fields_still_reads() { - let (_, data) = vector("commit-sparse"); - let commit: Commit = serde_json::from_value(data).expect("a readable commit frame"); - - // The point of the vector: an index reads servers it did not build, and - // one that does not name a commit must not stop it dead. Empty is the - // honest reading of "this server did not say". - assert_eq!(commit.seq, 43); - assert_eq!(commit.cid, ""); - assert_eq!(commit.commit, ""); - assert_eq!(commit.rev, ""); - assert_eq!(commit.time, ""); - assert_eq!(commit.cursor(), "01JEXAMPLEINSTANCE:43"); - assert_eq!(commit.record["text"], "a frame from an older server"); +fn a_write_reads_with_its_record_out_of_the_car() { + let Message::Commit(commit) = read("commit") else { + panic!("not a commit"); + }; + + assert_eq!(commit.seq, 10); + assert_eq!(commit.did, KESTREL); + assert_eq!( + commit.commit, + "bafyreidtoezye2gvsy4qdvwkrox4uk5dls5p7n32zqylhbhpzquvqqtpzu" + ); + assert_eq!(commit.rev, "3mvo3im3uhtx2"); + assert_eq!(commit.since.as_deref(), Some("3mvo3ikxahhx2")); + assert_eq!(commit.time, "2026-09-16T21:29:35Z"); + + let [write] = commit.writes.as_slice() else { + panic!("one write, not {}", commit.writes.len()); + }; + assert_eq!(write.action, "create"); + assert_eq!(write.collection, "com.vibescrobble.scrobble"); + assert_eq!(write.rkey, "3mvo3im3u4mqb"); + assert_eq!( + write.uri, + format!("at://{KESTREL}/com.vibescrobble.scrobble/3mvo3im3u4mqb") + ); + assert_eq!( + write.cid.as_deref(), + Some("bafyreiexrnsbgab5thj5attcgeqk3vmipzczgqzaj2xa6lqhagm5n44equ") + ); + // The record, as `com.atproto.repo.getRecord` would spell it. + let record = write.record.as_ref().expect("a create carries its record"); + assert_eq!(record["$type"], "com.vibescrobble.scrobble"); + assert_eq!(record["text"], "reading the subscribeRepos reconnect path"); + assert_eq!(record["emoji"], "🧵"); + assert_eq!(record["createdAt"], "2026-09-16T12:00:00Z"); } #[test] fn a_claimed_handle_reads() { - let (event, data) = vector("identity"); - let identity: Identity = serde_json::from_value(data).expect("a readable identity frame"); + let Message::Identity(identity) = read("identity") else { + panic!("not an identity frame"); + }; - assert_eq!(event, "identity"); - assert_eq!(identity.did, "did:web:kestrel.agents.localhost"); - assert_eq!(identity.cursor(), "01JEXAMPLEINSTANCE:44"); + assert_eq!(identity.seq, 2); + assert_eq!(identity.did, KESTREL); assert_eq!(identity.handle.as_deref(), Some("kestrel.agents.localhost")); } -#[test] -fn an_identity_frame_naming_no_handle_still_reads() { - let (_, data) = vector("identity-no-handle"); - let identity: Identity = serde_json::from_value(data).expect("a readable identity frame"); - - assert_eq!(identity.handle, None); -} - #[test] fn an_active_account_carries_no_status() { - let (event, data) = vector("account-active"); - let account: Account = serde_json::from_value(data).expect("a readable account frame"); + let Message::Account(account) = read("account-active") else { + panic!("not an account frame"); + }; - assert_eq!(event, "account"); - assert_eq!(account.did, "did:web:kestrel.agents.localhost"); - assert_eq!(account.cursor(), "01JEXAMPLEINSTANCE:46"); + assert_eq!(account.seq, 3); + assert_eq!(account.did, KESTREL); assert!(account.active); assert_eq!(account.status, None); assert!(!account.is_deleted()); @@ -126,8 +100,9 @@ fn an_active_account_carries_no_status() { #[test] fn a_deleted_account_reads_as_deleted() { - let (_, data) = vector("account-deleted"); - let account: Account = serde_json::from_value(data).expect("a readable account frame"); + let Message::Account(account) = read("account-deleted") else { + panic!("not an account frame"); + }; assert!(!account.active); assert_eq!(account.status.as_deref(), Some("deleted")); @@ -138,10 +113,49 @@ fn a_deleted_account_reads_as_deleted() { /// not mean gone, only a `status` of `deleted` does. #[test] fn a_deactivated_account_is_inactive_but_not_deleted() { - let (_, data) = vector("account-deactivated"); - let account: Account = serde_json::from_value(data).expect("a readable account frame"); + let Message::Account(account) = read("account-deactivated") else { + panic!("not an account frame"); + }; assert!(!account.active); assert_eq!(account.status.as_deref(), Some("deactivated")); assert!(!account.is_deleted()); } + +#[test] +fn an_unreachable_cursor_reads_as_the_name_that_means_a_gap() { + let message = read("info-outdated-cursor"); + let Message::Info(info) = &message else { + panic!("not an info frame"); + }; + + // The index compares this string, so the vector is what stops it drifting + // from what a server sends. + assert_eq!(info.name, OUTDATED_CURSOR); + assert!(!info.message.is_empty()); + // Unnumbered: an info frame must not move the cursor. + assert_eq!(message.seq(), None); +} + +#[test] +fn a_cursor_ahead_of_the_stream_reads_as_the_error_that_names_it() { + let Message::Error { error, message } = read("error-future-cursor") else { + panic!("not an error frame"); + }; + + assert_eq!(error, FUTURE_CURSOR); + assert!(message.contains("999999"), "{message}"); +} + +/// A recording annotates each frame from its own bytes, so what a person +/// sees beside the base64 is what the frame says. +#[test] +fn a_recorded_frame_is_annotated_from_its_contents() { + let recorded = Recorded::from_frame(&vector("commit")).expect("readable"); + assert_eq!(recorded.seq, Some(10)); + assert_eq!(recorded.did.as_deref(), Some(KESTREL)); + + let recorded = Recorded::from_frame(&vector("info-outdated-cursor")).expect("readable"); + assert_eq!(recorded.seq, None); + assert_eq!(recorded.did, None); +} diff --git a/crates/vibescrobble-index/tests/reconnect.rs b/crates/vibescrobble-index/tests/reconnect.rs index 9608108a..cf4ead16 100644 --- a/crates/vibescrobble-index/tests/reconnect.rs +++ b/crates/vibescrobble-index/tests/reconnect.rs @@ -8,22 +8,26 @@ //! The failure it is about is ordinary: a developer restarts the personal //! data server while an index is following it. //! -//! The server here is a stub that speaks the firehose and nothing else. A real -//! one would drag the whole personal data server in for a property that is not -//! about it — what the real server puts on the wire is asserted on its own -//! side, and that the two agree is `vectors/firehose/`'s job. +//! The server here is a stub that speaks `com.atproto.sync.subscribeRepos` +//! and nothing else. A real one would drag the whole personal data server in +//! for a property that is not about it — what the real server puts on the +//! wire is asserted on its own side, and that the two agree is +//! `vectors/subscribe-repos/`'s job. //! //! This is the slow test in this crate and unavoidably so: the subscriber //! waits out its reconnect interval, which is a number chosen for a developer //! watching a terminal rather than for a test. +use std::collections::BTreeMap; use std::net::SocketAddr; use std::sync::{Arc, Mutex}; use std::time::Duration; -use serde_json::{json, Value}; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use atproto_dasl::{CarBlock, CarHeader, Ipld}; +use futures_util::SinkExt; use tokio::sync::mpsc; +use tokio_tungstenite::tungstenite::handshake::server::{Request, Response}; +use tokio_tungstenite::tungstenite::Message as Ws; use vibescrobble_index::firehose::{subscribe_from, Update}; use vibescrobble_index::ServerRef; @@ -31,13 +35,10 @@ use vibescrobble_index::ServerRef; /// thing it protects against is a test that would otherwise never finish. const PATIENCE: Duration = Duration::from_secs(20); -/// The run of the "server" that minted the sequence numbers. -const INSTANCE: &str = "stub-instance"; - const DID: &str = "did:web:stormcock.agents.localhost"; const COLLECTION: &str = "com.vibescrobble.scrobble"; -/// A firehose and nothing else, on a port nobody chose. +/// A stream and nothing else, on a port nobody chose. /// /// Written against the socket rather than on a web framework because the test /// needs to take the connection away from a client that is *already @@ -49,7 +50,7 @@ const COLLECTION: &str = "com.vibescrobble.scrobble"; struct Stub { addr: SocketAddr, /// Every commit frame written so far, in sequence order. - frames: Arc>>, + frames: Arc>>>, /// The accept loop and every connection it has opened. live: Arc>>>, } @@ -107,16 +108,7 @@ impl Stub { fn write(&self, text: &str) -> u64 { let mut frames = self.frames.lock().expect("stub lock"); let seq = frames.len() as u64 + 1; - let rkey = format!("3lqm{seq:08}"); - frames.push(json!({ - "seq": seq, - "instance": INSTANCE, - "did": DID, - "collection": COLLECTION, - "rkey": rkey, - "uri": format!("at://{DID}/{COLLECTION}/{rkey}"), - "record": { "$type": COLLECTION, "text": text }, - })); + frames.push(commit_frame(seq, text)); seq } @@ -125,43 +117,102 @@ impl Stub { } } -/// Serves one connection: the opening frame, whatever the cursor missed, and -/// everything published from then on. -async fn serve(mut socket: tokio::net::TcpStream, frames: Arc>>) { - let Some(request) = read_request(&mut socket).await else { - return; - }; - let after = cursor_of(&request); +fn map(pairs: [(&str, Ipld); N]) -> Ipld { + Ipld::Map( + pairs + .into_iter() + .map(|(key, value)| (key.to_owned(), value)) + .collect::>(), + ) +} + +fn text(value: &str) -> Ipld { + Ipld::String(value.to_owned()) +} - // No content length and no chunking: the body ends when the connection - // does, which is exactly what a stream that is cut should look like. - let opening = json!({ - "name": if after.is_some() { "Replay" } else { "Live" }, - "instance": INSTANCE, - "oldest": 1, - "newest": frames.lock().expect("stub lock").len(), - "message": "a stub firehose", - }); - let head = format!( - "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\n\r\n\ - event: info\r\ndata: {opening}\n\n" +/// One `#commit` frame carrying one scrobble, shaped as the lexicon says. +/// +/// The CAR holds a commit block and the record block, which is the least a +/// real server sends; the tree nodes between them are what the consumer reads +/// past, so their absence is not something it can tell. +fn commit_frame(seq: u64, message: &str) -> Vec { + let rkey = format!("3lqm{seq:08}"); + let rev = format!("3lqm{seq:08}rev"); + let record = CarBlock::from_data( + atproto_dasl::to_vec(&map([ + ("$type", text(COLLECTION)), + ("text", text(message)), + ("createdAt", text("2026-09-16T12:00:00Z")), + ])) + .expect("a record encodes"), + ); + let commit = CarBlock::from_data( + atproto_dasl::to_vec(&map([ + ("did", text(DID)), + ("rev", text(&rev)), + ("version", Ipld::Integer(3)), + ])) + .expect("a commit encodes"), ); - if socket.write_all(head.as_bytes()).await.is_err() { + let mut car = CarHeader::with_root(commit.cid.into()) + .to_bytes() + .expect("a header encodes"); + car.extend(commit.to_bytes().expect("a block encodes")); + car.extend(record.to_bytes().expect("a block encodes")); + + let body = map([ + ("blobs", Ipld::List(Vec::new())), + ("blocks", Ipld::Bytes(car)), + ("commit", Ipld::Link(commit.cid.into())), + ( + "ops", + Ipld::List(vec![map([ + ("action", text("create")), + ("cid", Ipld::Link(record.cid.into())), + ("path", text(&format!("{COLLECTION}/{rkey}"))), + ])]), + ), + ("rebase", Ipld::Bool(false)), + ("repo", text(DID)), + ("rev", text(&rev)), + ("seq", Ipld::Integer(seq.into())), + ("since", Ipld::Null), + ("time", text("2026-09-16T12:00:00Z")), + ("tooBig", Ipld::Bool(false)), + ]); + let header = map([("op", Ipld::Integer(1)), ("t", text("#commit"))]); + let mut bytes = atproto_dasl::to_vec(&header).expect("a header encodes"); + bytes.extend(atproto_dasl::to_vec(&body).expect("a body encodes")); + bytes +} + +/// Serves one connection: whatever the cursor missed, and everything +/// published from then on. +/// +/// The handshake callback's error type is tungstenite's, and its size is not +/// this test's to choose. +#[allow(clippy::result_large_err)] +async fn serve(socket: tokio::net::TcpStream, frames: Arc>>>) { + let mut cursor: Option = None; + let read_cursor = |request: &Request, response: Response| { + cursor = cursor_of(request.uri().query().unwrap_or_default()); + Ok(response) + }; + let Ok(mut socket) = tokio_tungstenite::accept_hdr_async(socket, read_cursor).await else { return; - } + }; - let mut sent = after.unwrap_or(0); + let mut sent = cursor.unwrap_or(0) as usize; loop { - let pending: Vec = frames + let pending: Vec> = frames .lock() .expect("stub lock") .iter() - .skip(sent as usize) + .skip(sent) .cloned() .collect(); for frame in pending { - let block = format!("event: commit\ndata: {frame}\n\n"); - if socket.write_all(block.as_bytes()).await.is_err() { + if socket.send(Ws::Binary(frame.into())).await.is_err() { return; } sent += 1; @@ -170,33 +221,13 @@ async fn serve(mut socket: tokio::net::TcpStream, frames: Arc>> } } -/// Reads request bytes up to the blank line that ends the headers. -async fn read_request(socket: &mut tokio::net::TcpStream) -> Option { - let mut buffered = Vec::new(); - let mut chunk = [0u8; 1024]; - loop { - let read = socket.read(&mut chunk).await.ok()?; - if read == 0 { - return None; - } - buffered.extend_from_slice(&chunk[..read]); - if buffered.windows(4).any(|window| window == b"\r\n\r\n") { - return Some(String::from_utf8_lossy(&buffered).into_owned()); - } - } -} - /// The sequence number a request's cursor names, if it carries one. -/// -/// A cursor is `:` and the colon is reserved in a query, so a -/// client encodes it. Decoding just that one escape is enough here. -fn cursor_of(request: &str) -> Option { - let line = request.lines().next()?; - let query = line.split_whitespace().nth(1)?.split_once('?')?.1; - let value = query +fn cursor_of(query: &str) -> Option { + query .split('&') - .find_map(|pair| pair.strip_prefix("cursor="))?; - value.replace("%3A", ":").split_once(':')?.1.parse().ok() + .find_map(|pair| pair.strip_prefix("cursor="))? + .parse() + .ok() } /// The next update, or a failure naming what was being waited for. @@ -216,6 +247,16 @@ fn commit(update: Update) -> Box { } } +/// The scrobble a stub commit carries, by its text. +fn text_of(commit: &vibescrobble_index::Commit) -> String { + assert_eq!(commit.collection, COLLECTION); + assert_eq!(commit.did, DID); + commit.record["text"] + .as_str() + .unwrap_or_default() + .to_owned() +} + #[tokio::test] async fn a_subscriber_survives_the_server_going_away_and_misses_nothing() { let stub = Stub::start().await; @@ -225,7 +266,8 @@ async fn a_subscriber_survives_the_server_going_away_and_misses_nothing() { // Connected, and reading. stub.write("before the outage"); let first = commit(next(&mut incoming, "the first frame").await); - assert_eq!(first.record["text"], "before the outage"); + assert_eq!(text_of(&first), "before the outage"); + assert_eq!(first.did, DID); // The server goes away. The writes do not: the stream fills with frames // nobody is connected to read. @@ -243,10 +285,7 @@ async fn a_subscriber_survives_the_server_going_away_and_misses_nothing() { next(&mut incoming, "a frame from after the outage").await, )); } - let texts: Vec<&str> = missed - .iter() - .map(|frame| frame.record["text"].as_str().unwrap_or_default()) - .collect(); + let texts: Vec = missed.iter().map(|frame| text_of(frame)).collect(); assert_eq!( texts, [ @@ -273,7 +312,9 @@ async fn a_subscriber_survives_the_server_going_away_and_misses_nothing() { // that works in a test and not on a machine. stub.write("and still following"); assert_eq!( - commit(next(&mut incoming, "a frame after the recovery").await).record["text"], + text_of(&commit( + next(&mut incoming, "a frame after the recovery").await + )), "and still following" ); diff --git a/docs/running-locally.md b/docs/running-locally.md index c3e8c298..b777eef3 100644 --- a/docs/running-locally.md +++ b/docs/running-locally.md @@ -94,18 +94,19 @@ A recording removes the upstream entirely. `serve` listens on the port the recording came from, because an index checks that a server's zone sits under the host it was reached at and a playback -anywhere else is refused however correct its frames are. It answers the -firehose and the account listing, which is what an index reads on every pass, -and nothing else. +anywhere else is refused however correct its frames are. It answers +`com.atproto.sync.subscribeRepos` and `/health`, and nothing else. -A recording is one JSON object per line, so trimming it to the frames that -reproduce something is a text editor. What it buys over a server holding state -is that a recording is a *sequence* rather than a state: the same frames in the -same order every time, as fast or as slowly as `--rate` says, so a read-side -bug is reproduced rather than re-provoked. +A recording is one JSON object per line — the frame as it came off the wire, +base64, with its sequence number and repository written beside it — so +trimming it to the frames that reproduce something is a text editor. What it +buys over a server holding state is that a recording is a *sequence* rather +than a state: the same frames in the same order every time, as fast or as +slowly as `--rate` says, so a read-side bug is reproduced rather than +re-provoked. -`vectors/firehose/` is the same idea one level down — the frame shapes the -reader is checked against, without a recording or a process. +`vectors/subscribe-repos/` is the same idea one level down — single recorded +frames the reader is checked against, without a recording or a process. ## Asking the query service things @@ -229,41 +230,44 @@ or below the host it was reached at, so a commit naming somebody else's name is dropped and logged. On `https://agents.localhost` that host is `agents.localhost`, which covers every account the development server mints. -## The event stream it reads +## The stream it reads -For a `--pds` server, the index reads `GET /firehose`: every record the server -stores, one Server-Sent Event each, with the record itself as JSON. -`vibescrobble-replay serve` answers it. didbot serves -`com.atproto.sync.subscribeRepos` instead, which is what a relay crawls. +`com.atproto.sync.subscribeRepos` on a server is the protocol's own firehose: +a WebSocket of framed DAG-CBOR carrying every commit the server makes, with +account and identity events folded in. `vibescrobble-replay record` is the +easiest way to look at one: ```sh -curl -N localhost:3400/firehose # a replay listening on 3400 +./target/debug/vibescrobble-replay record --pds https://agents.localhost --out session.jsonl ``` -The first frame is always an `info` frame saying which run of the server this -is and which sequence numbers it can still replay. Every `commit` frame after -it carries a `seq`, and a consumer that remembers the last one it applied -resumes with it: +Every `#commit`, `#account` and `#identity` frame carries a `seq`, and a +consumer that remembers the last one it applied resumes after it: ```sh -curl -N 'localhost:3400/firehose?cursor=18e4a2f1b:97' +./target/debug/vibescrobble-replay record --pds https://agents.localhost --cursor 97 --out later.jsonl ``` -A cursor is `:`, not a bare number, because sequence numbers -restart with the process and an integer from a previous run would name a frame -that exists and is the wrong one. Hand back a cursor the server no longer -holds — too far behind its buffer, or from a previous run — and it answers with -an `info` frame naming `OutdatedCursor` and the oldest sequence number it still -has, then replays everything it does have. The records between the two are gone -from the stream forever and can only be read back from the repositories, which -is why the index still sweeps them. - -Every frame names the commit it landed in: `cid` is the record, `commit` is -the signed commit that now covers it, and `rev` is that commit's revision. So -this index can check itself against the repository rather than taking the -stream's word for it — apply frames until the stream goes quiet, then ask -`com.atproto.sync.getLatestCommit` for the same DID and compare `cid` there -with `commit` here. Equal means caught up, exactly. +A didbot server's sequence numbers survive a restart, but its replay buffer +does not. Hand back a cursor it no longer holds and it answers with an `#info` +frame naming `OutdatedCursor`, then whatever it does still hold; the frames +between are gone from the stream forever and can only be read back from the +repositories, which is why the index still sweeps them. A cursor ahead of the +stream — from a different server, say — is refused outright with a +`FutureCursor` error, and the index treats that as the same gap. + +A `#commit` carries its records as a CAR: the commit block, the tree nodes on +each written record's path, and the records themselves. The index reads a +record out of it by the `cid` its op names and reads past the rest — the +proof that a record is in the commit is a request away at +`com.atproto.sync.getRecord`, and this is a reader of records rather than a +verifier of repositories. What it keeps beside the record is what makes the +stream reconcilable: `commit` is the signed commit that now covers it, and +`rev` is that commit's revision. So this index can check itself against the +repository rather than taking the stream's word for it — apply frames until +the stream goes quiet, then ask `com.atproto.sync.getLatestCommit` for the +same DID and compare `cid` there with `commit` here. Equal means caught up, +exactly. Compare on the commit and not on `rev`. A revision there is derived from the repository's record keys, so a write to a fixed key such as `self` moves diff --git a/plan/ingest.md b/plan/ingest.md index 73ac0fc6..d374bd88 100644 --- a/plan/ingest.md +++ b/plan/ingest.md @@ -50,11 +50,10 @@ and it is protocol now. than reimplemented for this path. A commit whose account does not resolve yet still mints it nameless as it always has, but now also queues a retry rather than staying nameless forever if no later commit - happens to try again. No server built from this repository sends - either frame on `/firehose` yet — `vectors/firehose/` pins the shape - atproto's own frames carry, ahead of a producer — so a `--pds` - deployment still discovers most accounts from their first commit, and - the survey remains the backstop for whatever a frame misses. + happens to try again. The frames are the protocol's own, read off + `com.atproto.sync.subscribeRepos` — `vectors/subscribe-repos/` holds + them as a didbot server sent them — and the survey remains the backstop + for whatever a frame misses. - [x] **A cold start can see what was already there.** `--backfill` enumerates a server over `com.atproto.sync.listRepos` and reads every repository it holds before following it, which is the one thing no cursor reaches: a diff --git a/vectors/firehose/README.md b/vectors/firehose/README.md deleted file mode 100644 index a931e1db..00000000 --- a/vectors/firehose/README.md +++ /dev/null @@ -1,47 +0,0 @@ -# Firehose wire vectors - -What a server writes on `/firehose`, recorded so that a producer and a consumer -built from different repositories can each check themselves against it without -either compiling against the other. - -The frame shape is not owned by either side. A server declares it as -`vibescrobble_serve::firehose::{CommitFrame, Info}`, an index declares it as -`vibescrobble_index::firehose::{Commit, StreamInfo}`, and the two declarations -are deliberately not the same type: an index that compiled against a server's -types could only read servers built from this repository. These files are what -holds the two in step instead. - -Each file is one frame: - -```json -{ "note": "why this case is here", "event": "commit", "data": { … } } -``` - -`event` is the SSE event name. `data` is the JSON body of that event. Those -two are the contract a producer and a consumer check themselves against; -`note` is prose for whoever is reading the file, and can be reworded without -changing what the vector proves. - -A producer asserts that serializing its own type yields `data`. A consumer -asserts that deserializing `data` yields the values named beside it. Neither -side reads the other's assertions. - -Two of these exist to pin a tolerance rather than a shape: - -- `info-unknown-name.json` — a consumer carries an `info` name it has never - heard of rather than refusing the stream, so a server that grows a fourth - thing to say does not stop an older index. -- `commit-sparse.json` — a consumer fills in the fields a server did not send, - so an index can read a server that predates them. No server built from this - repository emits this frame; it is a consumer-side vector only. - -`identity.json`, `identity-no-handle.json`, `account-active.json`, -`account-deleted.json` and `account-deactivated.json` are consumer-side only -in a different sense: no server built from this repository emits `identity` -or `account` on `/firehose` yet, so there is no producer assertion to make -these mirror. What they pin instead is the shape atproto's own `#identity` and -`#account` carry, which is what a server that grows the frames is expected to -send — `did`, `seq`, `time` and a `handle` for the first; `did`, `seq`, `time`, -`active` and an optional `status` for the second. `account-deactivated.json` -exists to pin a distinction that matters more than the shape: `active: false` -alone is not the same as gone, and only `status: "deleted"` is. diff --git a/vectors/firehose/account-active.json b/vectors/firehose/account-active.json deleted file mode 100644 index c56cb731..00000000 --- a/vectors/firehose/account-active.json +++ /dev/null @@ -1,11 +0,0 @@ -{ - "note": "An `account` frame announces a repository's serving status, mirroring atproto's own #account event on /firehose. A producer sends one whenever that status changes; an active account carries no `status` field at all, rather than one saying it is fine.", - "event": "account", - "data": { - "seq": 46, - "instance": "01JEXAMPLEINSTANCE", - "did": "did:web:kestrel.agents.localhost", - "time": "2026-09-08T10:00:00Z", - "active": true - } -} diff --git a/vectors/firehose/account-deactivated.json b/vectors/firehose/account-deactivated.json deleted file mode 100644 index 0cc4fdbf..00000000 --- a/vectors/firehose/account-deactivated.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "note": "`active: false` with any status other than `deleted` is reversible — deactivated, suspended and so on — and a reader should stop serving the repository for now without treating it as ended, since the same `did` can turn active again.", - "event": "account", - "data": { - "seq": 48, - "instance": "01JEXAMPLEINSTANCE", - "did": "did:web:kestrel.agents.localhost", - "time": "2026-09-08T10:00:03Z", - "active": false, - "status": "deactivated" - } -} diff --git a/vectors/firehose/account-deleted.json b/vectors/firehose/account-deleted.json deleted file mode 100644 index a04279fc..00000000 --- a/vectors/firehose/account-deleted.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "note": "`active: false` paired with `status: \"deleted\"` is the one combination a reader should treat as final: nothing more will be served for the repository, and its identity can be retired rather than watched for a return.", - "event": "account", - "data": { - "seq": 47, - "instance": "01JEXAMPLEINSTANCE", - "did": "did:web:kestrel.agents.localhost", - "time": "2026-09-08T10:00:02Z", - "active": false, - "status": "deleted" - } -} diff --git a/vectors/firehose/commit-sparse.json b/vectors/firehose/commit-sparse.json deleted file mode 100644 index 4aa3959d..00000000 --- a/vectors/firehose/commit-sparse.json +++ /dev/null @@ -1,18 +0,0 @@ -{ - "note": "Consumer-side only. A server that predates cid, commit, rev and time sends none of them; an index reads servers it did not build, so it fills them in rather than stopping.", - "event": "commit", - "data": { - "seq": 43, - "instance": "01JEXAMPLEINSTANCE", - "did": "did:web:kestrel.agents.localhost", - "collection": "com.vibescrobble.scrobble", - "rkey": "3lqm0000abcf6", - "uri": "at://did:web:kestrel.agents.localhost/com.vibescrobble.scrobble/3lqm0000abcf6", - "record": { - "$type": "com.vibescrobble.scrobble", - "text": "a frame from an older server", - "emoji": "🕰️", - "createdAt": "2026-08-24T12:00:01Z" - } - } -} diff --git a/vectors/firehose/commit.json b/vectors/firehose/commit.json deleted file mode 100644 index ea779c39..00000000 --- a/vectors/firehose/commit.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "note": "One write, with every field a server built from this repository sends.", - "event": "commit", - "data": { - "seq": 42, - "instance": "01JEXAMPLEINSTANCE", - "did": "did:web:kestrel.agents.localhost", - "collection": "com.vibescrobble.scrobble", - "rkey": "3lqm0000abcd2", - "uri": "at://did:web:kestrel.agents.localhost/com.vibescrobble.scrobble/3lqm0000abcd2", - "cid": "bafyreigh2akiscaildcqabsyg3dfr6chu3fgpregiymsck7e7aqa4s52zy", - "commit": "bafyreidfuxdmqvxvtqcyvzkxpvbdcvwjxlrqvbn5xk3vg6mfumzsrxvz4a", - "rev": "3lqm0000abce4", - "time": "2026-08-24T12:00:00Z", - "record": { - "$type": "com.vibescrobble.scrobble", - "text": "reading the firehose reconnect path", - "emoji": "🧵", - "createdAt": "2026-08-24T12:00:00Z" - } - } -} diff --git a/vectors/firehose/identity-no-handle.json b/vectors/firehose/identity-no-handle.json deleted file mode 100644 index a208f695..00000000 --- a/vectors/firehose/identity-no-handle.json +++ /dev/null @@ -1,10 +0,0 @@ -{ - "note": "A deployment that issues no handles sends no `handle` field at all, rather than a null one; a reader must take its absence as 'nothing claimed', not as a parse failure.", - "event": "identity", - "data": { - "seq": 45, - "instance": "01JEXAMPLEINSTANCE", - "did": "did:web:kestrel.agents.localhost", - "time": "2026-09-08T10:00:01Z" - } -} diff --git a/vectors/firehose/identity.json b/vectors/firehose/identity.json deleted file mode 100644 index f637ce8e..00000000 --- a/vectors/firehose/identity.json +++ /dev/null @@ -1,11 +0,0 @@ -{ - "note": "An `identity` frame announces the handle a repository's server currently claims, mirroring atproto's own #identity event on /firehose. A producer sends one whenever that claim changes; a reader takes `did`, `seq`, `time` and `handle` as the claim now in effect.", - "event": "identity", - "data": { - "seq": 44, - "instance": "01JEXAMPLEINSTANCE", - "did": "did:web:kestrel.agents.localhost", - "time": "2026-09-08T10:00:00Z", - "handle": "kestrel.agents.localhost" - } -} diff --git a/vectors/firehose/info-live.json b/vectors/firehose/info-live.json deleted file mode 100644 index a7e836b0..00000000 --- a/vectors/firehose/info-live.json +++ /dev/null @@ -1,11 +0,0 @@ -{ - "note": "The opening frame on a connection that gave no cursor. Sent unconditionally, so a consumer learns the instance and the buffer's range without provoking an error.", - "event": "info", - "data": { - "name": "Live", - "instance": "01JEXAMPLEINSTANCE", - "oldest": 0, - "newest": 0, - "message": "streaming from now; nothing was replayed" - } -} diff --git a/vectors/firehose/info-outdated-cursor.json b/vectors/firehose/info-outdated-cursor.json deleted file mode 100644 index feb7969e..00000000 --- a/vectors/firehose/info-outdated-cursor.json +++ /dev/null @@ -1,11 +0,0 @@ -{ - "note": "The cursor named a frame this server no longer holds, or one from a previous run. Everything still held follows, and the consumer has a hole only a re-read of the repositories can fill.", - "event": "info", - "data": { - "name": "OutdatedCursor", - "instance": "01JEXAMPLEINSTANCE", - "oldest": 41, - "newest": 96, - "message": "cursor 01JOLDINSTANCE:7 is older than the buffer; 41 is the oldest still held" - } -} diff --git a/vectors/firehose/info-unknown-name.json b/vectors/firehose/info-unknown-name.json deleted file mode 100644 index 4a8db568..00000000 --- a/vectors/firehose/info-unknown-name.json +++ /dev/null @@ -1,11 +0,0 @@ -{ - "note": "Consumer-side only. Every info name except OutdatedCursor means carry on, including one this index has never heard of.", - "event": "info", - "data": { - "name": "Rebalancing", - "instance": "01JEXAMPLEINSTANCE", - "oldest": 41, - "newest": 96, - "message": "moving the buffer; the stream continues" - } -} diff --git a/vectors/subscribe-repos/README.md b/vectors/subscribe-repos/README.md new file mode 100644 index 00000000..6e491d18 --- /dev/null +++ b/vectors/subscribe-repos/README.md @@ -0,0 +1,44 @@ +# subscribeRepos wire vectors + +Frames a didbot server put on `com.atproto.sync.subscribeRepos`, recorded so +that this index can check its reader against a real producer without +compiling against it. + +The frame shape is the protocol's, not either side's. A server encodes what +the lexicon says; `vibescrobble_index::frames` decodes it into the values the +index acts on; and the two are deliberately not the same type — an index that +compiled against a server's types could only read servers built from one +repository. These files are what holds the two in step instead. + +Each file is one frame: + +```json +{ "note": "why this case is here", "frame": "", "seq": 10, "did": "…" } +``` + +`frame` is the WebSocket binary message exactly as it arrived: two DAG-CBOR +values back to back, a header naming the message type and then its body. +That is the contract the consumer checks itself against. `seq` and `did` are +read out of the frame and written beside it so a person can tell the files +apart; `note` is prose for whoever is reading, and can be reworded without +changing what the vector proves. + +`crates/vibescrobble-index/tests/firehose_wire.rs` asserts that each frame +decodes to the values named there. To re-record them, run a didbot server and +`vibescrobble-replay record --pds `: a recording is the same base64 per +line, and a vector is one of its lines with a note. + +What each one pins: + +- `commit.json` — a `#commit` with one `create`. The record is read out of the + CAR by the op's `cid`; the commit block and tree nodes beside it are read + past. +- `identity.json` — an `#identity` naming the handle the server issued. +- `account-active.json`, `account-deactivated.json`, `account-deleted.json` — + the three `#account` shapes, and the distinction that matters more than the + shape: `active: false` alone is not the same as gone, and only + `status: "deleted"` is. +- `info-outdated-cursor.json` — the unnumbered `#info` a server answers a + cursor it no longer holds with. The index compares its `name`. +- `error-future-cursor.json` — an error frame, header `op: -1`, for a cursor + ahead of the stream. diff --git a/vectors/subscribe-repos/account-active.json b/vectors/subscribe-repos/account-active.json new file mode 100644 index 00000000..25d67733 --- /dev/null +++ b/vectors/subscribe-repos/account-active.json @@ -0,0 +1,6 @@ +{ + "note": "An `#account` frame for a repository that is being served: `active: true` and no `status` at all, rather than one saying it is fine.", + "frame": "omF0aCNhY2NvdW50Ym9wAaRjZGlkeCdkaWQ6d2ViOmtlc3RyZWwuYWdlbnRzLmxvY2FsaG9zdCUzQTM5MzFjc2VxA2R0aW1ldDIwMjYtMDktMTZUMjE6Mjk6MzRaZmFjdGl2ZfU=", + "seq": 3, + "did": "did:web:kestrel.agents.localhost%3A3931" +} diff --git a/vectors/subscribe-repos/account-deactivated.json b/vectors/subscribe-repos/account-deactivated.json new file mode 100644 index 00000000..81fd0100 --- /dev/null +++ b/vectors/subscribe-repos/account-deactivated.json @@ -0,0 +1,6 @@ +{ + "note": "`active: false` with `status: \"deactivated\"`. Not being served right now and reversible, which is not the same as gone: only `status: \"deleted\"` is.", + "frame": "omF0aCNhY2NvdW50Ym9wAaVjZGlkeCdkaWQ6d2ViOmtlc3RyZWwuYWdlbnRzLmxvY2FsaG9zdCUzQTM5MzFjc2VxDWR0aW1ldDIwMjYtMDktMTZUMjE6MzA6MDVaZmFjdGl2ZfRmc3RhdHVza2RlYWN0aXZhdGVk", + "seq": 13, + "did": "did:web:kestrel.agents.localhost%3A3931" +} diff --git a/vectors/subscribe-repos/account-deleted.json b/vectors/subscribe-repos/account-deleted.json new file mode 100644 index 00000000..04fe21df --- /dev/null +++ b/vectors/subscribe-repos/account-deleted.json @@ -0,0 +1,6 @@ +{ + "note": "`active: false` with `status: \"deleted\"`: the one combination a reader treats as final. Nothing more will be served for the repository, and its identity can be retired rather than watched for a return.", + "frame": "omF0aCNhY2NvdW50Ym9wAaVjZGlkeCdkaWQ6d2ViOmtlc3RyZWwuYWdlbnRzLmxvY2FsaG9zdCUzQTM5MzFjc2VxEGR0aW1ldDIwMjYtMDktMTZUMjE6MzA6MDdaZmFjdGl2ZfRmc3RhdHVzZ2RlbGV0ZWQ=", + "seq": 16, + "did": "did:web:kestrel.agents.localhost%3A3931" +} diff --git a/vectors/subscribe-repos/commit.json b/vectors/subscribe-repos/commit.json new file mode 100644 index 00000000..f518f714 --- /dev/null +++ b/vectors/subscribe-repos/commit.json @@ -0,0 +1,6 @@ +{ + "note": "One `#commit` carrying one write: a `create` of a com.vibescrobble.scrobble record. The CAR in `blocks` holds the commit block, the tree nodes on the record's path and the record itself; the consumer reads the record by the op's `cid` and nothing else.", + "frame": "omF0ZyNjb21taXRib3ABrGNvcHOBo2NjaWTYKlglAAFxEiCXi2QTAD2Z09BOYjEgrdWIfkWTQyBOrg8uBwGZ1vOEhWRwYXRoeCdjb20udmliZXNjcm9iYmxlLnNjcm9iYmxlLzNtdm8zaW0zdTRtcWJmYWN0aW9uZmNyZWF0ZWNyZXZtM212bzNpbTN1aHR4MmNzZXEKZHJlcG94J2RpZDp3ZWI6a2VzdHJlbC5hZ2VudHMubG9jYWxob3N0JTNBMzkzMWR0aW1ldDIwMjYtMDktMTZUMjE6Mjk6MzVaZWJsb2JzgGVzaW5jZW0zbXZvM2lreGFoaHgyZmJsb2Nrc1kD7DqiZXJvb3RzgdgqWCUAAXESIHNxM4Jo1ZY5AdbKi6/KK6Ncuv+3eswws4TvzClYQm/NZ3ZlcnNpb24BjwIBcRIgc3EzgmjVljkB1sqLr8oro1y6/7d6zDCzhO/MKVhCb82mY2RpZHgnZGlkOndlYjprZXN0cmVsLmFnZW50cy5sb2NhbGhvc3QlM0EzOTMxY3Jldm0zbXZvM2ltM3VodHgyY3NpZ1hArfwqYPAquH3GD5wVIhjp0NbbD6t4CqeS3vP1BWvUJNR1T7H9iYilAGhVa3e7wKb7MZqxXXRAhmcQV5D4YZLiVmRkYXRh2CpYJQABcRIgp+VTKk0l4SMmM0orGSIRmrGH8lVLJeEcEFZo1iIoaC5kcHJldtgqWCUAAXESIGYy8NqRkc5xhY1gDQ3xQpMgojqS2ZQ7suozTYqin8wNZ3ZlcnNpb24DUwFxEiBjLH4z0iVZ3KsXqRHD900L/YicJN7l7OO3TAsgdaejRKJhZYBhbNgqWCUAAXESILtWlowVf70PksyQZEWfiLF0JIa6r7EiOoY85q/G5DrBoAEBcRIgl4tkEwA9mdPQTmIxIK3ViH5Fk0MgTq4PLgcBmdbzhIWkZHRleHR4KXJlYWRpbmcgdGhlIHN1YnNjcmliZVJlcG9zIHJlY29ubmVjdCBwYXRoZSR0eXBleBljb20udmliZXNjcm9iYmxlLnNjcm9iYmxlZWVtb2ppZPCfp7VpY3JlYXRlZEF0dDIwMjYtMDktMTZUMTI6MDA6MDBaygEBcRIgp+VTKk0l4SMmM0orGSIRmrGH8lVLJeEcEFZo1iIoaC6iYWWBpGFrWBlib3QuZGlkLnJlZ2lzdHJhdGlvbi9zZWxmYXAAYXTYKlglAAFxEiC6ycFxavdgL9E8qaqWvYkXDA2fWDcSAvwaiiIqvit6AWF22CpYJQABcRIgUri021ssl2ML5rSsRSOdjZb6aZ12tu6MU1M+HnefMlFhbNgqWCUAAXESINi8yiRWKZ+DhedgrnUIzvYLZWB3j5KMzrjwQ3BMUJ3jUwFxEiC6ycFxavdgL9E8qaqWvYkXDA2fWDcSAvwaiiIqvit6AaJhZYBhbNgqWCUAAXESIGMsfjPSJVncqxepEcP3TQv9iJwk3uXs47dMCyB1p6NEiAEBcRIgu1aWjBV/vQ+SzJBkRZ+IsXQkhrqvsSI6hjzmr8bkOsGiYWWBpGFrWCdjb20udmliZXNjcm9iYmxlLnNjcm9iYmxlLzNtdm8zaW0zdTRtcWJhcABhdPZhdtgqWCUAAXESIJeLZBMAPZnT0E5iMSCt1Yh+RZNDIE6uDy4HAZnW84SFYWz2ZmNvbW1pdNgqWCUAAXESIHNxM4Jo1ZY5AdbKi6/KK6Ncuv+3eswws4TvzClYQm/NZnJlYmFzZfRmdG9vQmln9GhwcmV2RGF0YdgqWCUAAXESINv95XJmkBdHsN8d2Y8OFE8SCWo3ygeb2E+umeUDBnig", + "seq": 10, + "did": "did:web:kestrel.agents.localhost%3A3931" +} diff --git a/vectors/subscribe-repos/error-future-cursor.json b/vectors/subscribe-repos/error-future-cursor.json new file mode 100644 index 00000000..ba8764d2 --- /dev/null +++ b/vectors/subscribe-repos/error-future-cursor.json @@ -0,0 +1,4 @@ +{ + "note": "An error frame — header `op: -1` — for a cursor ahead of the stream. The server closes the connection after it.", + "frame": "oWJvcCCiZWVycm9ybEZ1dHVyZUN1cnNvcmdtZXNzYWdleERjdXJzb3IgOTk5OTk5IGlzIGFoZWFkIG9mIHRoaXMgc3RyZWFtOyB0aGUgbmV3ZXN0IGZyYW1lIHNlbnQgaXMgMTAyNA==" +} diff --git a/vectors/subscribe-repos/identity.json b/vectors/subscribe-repos/identity.json new file mode 100644 index 00000000..63b29e27 --- /dev/null +++ b/vectors/subscribe-repos/identity.json @@ -0,0 +1,6 @@ +{ + "note": "An `#identity` frame: the server issued the account a handle. Sent at provisioning, before the account's first commit.", + "frame": "omF0aSNpZGVudGl0eWJvcAGkY2RpZHgnZGlkOndlYjprZXN0cmVsLmFnZW50cy5sb2NhbGhvc3QlM0EzOTMxY3NlcQJkdGltZXQyMDI2LTA5LTE2VDIxOjI5OjM0WmZoYW5kbGV4GGtlc3RyZWwuYWdlbnRzLmxvY2FsaG9zdA==", + "seq": 2, + "did": "did:web:kestrel.agents.localhost%3A3931" +} diff --git a/vectors/subscribe-repos/info-outdated-cursor.json b/vectors/subscribe-repos/info-outdated-cursor.json new file mode 100644 index 00000000..ecc845a4 --- /dev/null +++ b/vectors/subscribe-repos/info-outdated-cursor.json @@ -0,0 +1,4 @@ +{ + "note": "The `#info` a server answers a cursor it no longer holds with. It carries no sequence number; everything before the replay buffer is gone from the stream and must be read from the repositories.", + "frame": "omF0ZSNpbmZvYm9wAaJkbmFtZW5PdXRkYXRlZEN1cnNvcmdtZXNzYWdleHh0aGlzIGN1cnNvciBpcyBvdXRzaWRlIHRoZSByZXBsYXkgYnVmZmVyLCB3aGljaCBob2xkcyAwIHRvIDEwMjQuIEV2ZXJ5dGhpbmcgYmVmb3JlIGl0IG11c3QgYmUgcmVhZCBmcm9tIHRoZSByZXBvc2l0b3JpZXM=" +} -- knotmirror 0.0.1