From f317c6bd119fd526de67860b4a8e2a07b33cdcae Mon Sep 17 00:00:00 2001 From: "@permadeath.com" Date: Wed, 30 Sep 2026 11:30:32 -0400 Subject: [PATCH] feat(index): read the repositories when a server sends #sync A didbot server restored from a backup resets each live repository with a #sync frame. The index now treats one as a gap and reads the repositories, rather than skipping it as a message it has no name for. Co-Authored-By: Claude Opus 5.5 Change-Id: Id22f6fd0a7bd6a0f852b7b3d8c26b5b6b2f9dd0b --- crates/vibescrobble-index/src/firehose.rs | 12 +++++-- crates/vibescrobble-index/src/frames.rs | 42 +++++++++++++++++++++++ docs/running-locally.md | 4 ++- 3 files changed, 55 insertions(+), 3 deletions(-) diff --git a/crates/vibescrobble-index/src/firehose.rs b/crates/vibescrobble-index/src/firehose.rs index 1223bd6b..27915e42 100644 --- a/crates/vibescrobble-index/src/firehose.rs +++ b/crates/vibescrobble-index/src/firehose.rs @@ -137,8 +137,9 @@ pub enum Update { /// Frames were missed and cannot be recovered from the stream. /// /// 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. + /// or names a frame the server never sent, or the server reset a + /// repository with `#sync`, so the only way to find out what is in the + /// gap is to read the repositories. Gap { /// Which server said so. server: ServerRef, @@ -327,6 +328,13 @@ pub async fn subscribe_from(server: ServerRef, cursor: Option, out: mpsc::S account: Box::new(account), }] } + // A restored server resets each live repository with one of + // these. The records it held before may be gone, and the + // stream will not say which, so the repositories are read. + Message::Sync { did, .. } => vec![Update::Gap { + server: server.clone(), + message: format!("{did} was reset to a new commit"), + }], Message::Info(info) if info.name == OUTDATED_CURSOR => { warn!( server = server.base_url(), diff --git a/crates/vibescrobble-index/src/frames.rs b/crates/vibescrobble-index/src/frames.rs index 55af6f45..106d8954 100644 --- a/crates/vibescrobble-index/src/frames.rs +++ b/crates/vibescrobble-index/src/frames.rs @@ -42,6 +42,14 @@ pub enum Message { Identity(Identity), /// `#account`: an account appeared, went away or changed status. Account(Account), + /// `#sync`: the repository was reset to a new commit, and what the stream + /// said about it before this no longer adds up to what it holds. + Sync { + /// Where the frame sits in the stream. + seq: u64, + /// The repository reset. + did: String, + }, /// `#info`: something the server has to say about the connection. Info(Info), /// An error frame. The server closes the connection after it. @@ -70,6 +78,7 @@ impl Message { Self::Commit(commit) => Some(commit.seq), Self::Identity(identity) => Some(identity.seq), Self::Account(account) => Some(account.seq), + Self::Sync { seq, .. } => Some(*seq), Self::Other { seq, .. } => *seq, Self::Info(_) | Self::Error { .. } => None, } @@ -81,6 +90,7 @@ impl Message { Self::Commit(commit) => Some(&commit.did), Self::Identity(identity) => Some(&identity.did), Self::Account(account) => Some(&account.did), + Self::Sync { did, .. } => Some(did), Self::Info(_) | Self::Error { .. } | Self::Other { .. } => None, } } @@ -229,6 +239,10 @@ pub fn decode(bytes: &[u8]) -> Result { active: boolean(&mut body, "active").unwrap_or(false), status: string(&mut body, "status"), }), + "#sync" => Message::Sync { + seq: seq(&mut body)?, + did: required(&mut body, "did")?, + }, "#info" => Message::Info(Info { name: required(&mut body, "name")?, message: string(&mut body, "message").unwrap_or_default(), @@ -414,6 +428,34 @@ mod tests { assert!(matches!(err, DecodeError::Shape(_)), "{err}"); } + #[test] + fn a_sync_names_its_repository_and_is_numbered() { + let header = Ipld::Map(BTreeMap::from([ + ("op".to_owned(), Ipld::Integer(1)), + ("t".to_owned(), Ipld::String("#sync".to_owned())), + ])); + let body = Ipld::Map(BTreeMap::from([ + ("seq".to_owned(), Ipld::Integer(41)), + ( + "did".to_owned(), + Ipld::String("did:web:kestrel.agents.localhost".to_owned()), + ), + ("rev".to_owned(), Ipld::String("3mwq2ab7cde2x".to_owned())), + ("blocks".to_owned(), Ipld::Bytes(Vec::new())), + ( + "time".to_owned(), + Ipld::String("2026-09-27T14:08:49Z".to_owned()), + ), + ])); + let mut frame = atproto_dasl::to_vec(&header).expect("encodes"); + frame.extend(atproto_dasl::to_vec(&body).expect("encodes")); + + let message = decode(&frame).expect("decodes"); + assert_eq!(message.seq(), Some(41)); + assert_eq!(message.did(), Some("did:web:kestrel.agents.localhost")); + assert!(matches!(message, Message::Sync { .. }), "{message:?}"); + } + #[test] fn a_link_and_bytes_read_as_atproto_spells_them_in_json() { let cid = CarBlock::from_data(vec![0xa0]).cid; diff --git a/docs/running-locally.md b/docs/running-locally.md index b777eef3..26978cc7 100644 --- a/docs/running-locally.md +++ b/docs/running-locally.md @@ -254,7 +254,9 @@ 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. +`FutureCursor` error, and the index treats that as the same gap. So is a +`#sync`, which a server restored from a backup sends for each repository it +reset. 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 -- knotmirror 0.0.1