microcosm.blue / lightrail
lightweight com.atproto.sync.listReposByCollection
45
fork

Configure Feed

Select the types of activity you want to include in your feed.

clean up resync collection-getters

phil d3cd1666 1c8ca535

+434 -370
+3 -2
Cargo.lock
··· 2327 2327 "metrics", 2328 2328 "metrics-exporter-prometheus", 2329 2329 "mini-moka", 2330 + "n0-future", 2330 2331 "repo-stream", 2331 2332 "reqwest", 2332 2333 "rustls", ··· 3326 3327 3327 3328 [[package]] 3328 3329 name = "repo-stream" 3329 - version = "0.5.0-alpha.2" 3330 + version = "0.5.0-alpha.3" 3330 3331 source = "registry+https://github.com/rust-lang/crates.io-index" 3331 - checksum = "244266cae2ea40ce0c33f5f9168950ae263297806675c51add72497e93cafbc0" 3332 + checksum = "7df620415aa6a02c68a83a699f48f23472724367ae2301e6498b8f5957866340" 3332 3333 dependencies = [ 3333 3334 "cid", 3334 3335 "fjall",
+4 -3
Cargo.toml
··· 15 15 http = "1" 16 16 jacquard-api = { version = "0.9.5", default-features = false, features = ["com_atproto", "streaming"] } 17 17 jacquard-axum = { version = "0.9.6", default-features = false, features = ["tracing"] } 18 - jacquard-common = { version = "0.9.5", features = ["websocket", "reqwest-client"] } 18 + jacquard-common = { version = "0.9.5", features = ["websocket", "reqwest-client", "streaming"] } 19 + n0-future = "0.1" 19 20 jacquard-derive = "0.9.5" 20 21 jacquard-identity = "0.9.5" 21 22 jacquard-lexicon = "0.9.5" ··· 24 25 metrics-exporter-prometheus = { version = "0.18.1", features = ["http-listener"] } 25 26 mini-moka = "0.10" 26 27 reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] } 27 - repo-stream = { version = "0.5.0-alpha.2", features = ["jacquard"] } 28 + repo-stream = { version = "0.5.0-alpha.3", features = ["jacquard"] } 28 29 rustls = { version = "0.23", default-features = false, features = ["aws-lc-rs"] } 29 30 rustversion = "1" 30 31 serde = { version = "1", features = ["derive"] } 31 32 serde_json = "1" 32 33 thiserror = "2.0.18" 33 34 tokio = { version = "1.49.0", features = ["full"] } 34 - tokio-util = { version = "0.7", features = ["rt"] } 35 + tokio-util = { version = "0.7", features = ["rt", "io-util"] } 35 36 tracing = "0.1.44" 36 37 tracing-subscriber = { version = "0.3", features = ["env-filter"] } 37 38
+2 -1
hacking.md
··· 73 73 - [x] noop if rev/cid are unchanged 74 74 - [-] drop any preceeding #commits from the queue? (tricky with noop) 75 75 - [ ] "deep crawl" mode for relays that listHosts -> listRepos on host instead of relying on relay listRepos 76 + - [ ] defensive loop-cursor handling 76 77 - [~] actually firehose-index!! 77 78 - [x] extract collections-added/removed directly from CAR slice 78 79 - [ ] (spend some time on tests here) ··· 125 126 126 127 this is not nice because the response has neither the MST root signature nor the repo's `rev`, so we have to do extra work. 127 128 128 - the extra work for `describeRepo` currently entails calling `com.atproto.sync.getLatestCommit` first, to get a `rev` and MST root CID. this isn't sync-1.1 level robust: we're racing in between the two requests against potential new collections being added or removed. but doing cutover from a latest-commit *before* describeRepo was called should make the list converge to the correct state. 129 + the extra work for `describeRepo` currently entails ~~calling `com.atproto.sync.getLatestCommit` first~~ er, calling `com.atproto.sync.getRecord` with an arbitrary key first, to get a `rev` and MST root CID. this isn't sync-1.1 level robust: we're racing in between the two requests against potential new collections being added or removed. but doing cutover from a latest-commit *before* describeRepo was called should make the list converge to the correct state. 129 130 130 131 there are a few other ways we might get a fully-authenticated collections list, see [./authenticated-collection-list.md](./authenticated-collection-list.md) for some possible approaches. 131 132
+38 -1
src/http.rs
··· 15 15 16 16 use dashmap::DashMap; 17 17 use governor::{DefaultDirectRateLimiter, Quota, RateLimiter}; 18 - use jacquard_common::http_client::HttpClient; 18 + use jacquard_common::http_client::{HttpClient, HttpClientExt}; 19 + use jacquard_common::stream::{ByteStream, StreamError}; 19 20 20 21 /// Default per-host request rate: 10 req/s. 21 22 const DEFAULT_RATE_PER_SEC: u32 = 10; ··· 111 112 Ok(builder.body(body).expect("failed to build response")) 112 113 } 113 114 } 115 + 116 + impl HttpClientExt for ThrottledClient { 117 + async fn send_http_streaming( 118 + &self, 119 + request: http::Request<Vec<u8>>, 120 + ) -> Result<http::Response<ByteStream>, Self::Error> { 121 + let (parts, body) = request.into_parts(); 122 + if let Some(host) = parts.uri.host() { 123 + let limiter = self.shared.get_or_create_limiter(host); 124 + while limiter.check().is_err() { 125 + tokio::time::sleep(self.shared.token_interval).await; 126 + } 127 + } 128 + self.inner 129 + .send_http_streaming(http::Request::from_parts(parts, body)) 130 + .await 131 + } 132 + 133 + #[cfg(not(target_arch = "wasm32"))] 134 + async fn send_http_bidirectional<S>( 135 + &self, 136 + parts: http::request::Parts, 137 + body: S, 138 + ) -> Result<http::Response<ByteStream>, Self::Error> 139 + where 140 + S: n0_future::Stream<Item = Result<bytes::Bytes, StreamError>> + Send + 'static, 141 + { 142 + if let Some(host) = parts.uri.host() { 143 + let limiter = self.shared.get_or_create_limiter(host); 144 + while limiter.check().is_err() { 145 + tokio::time::sleep(self.shared.token_interval).await; 146 + } 147 + } 148 + self.inner.send_http_bidirectional(parts, body).await 149 + } 150 + }
-120
src/mst/adjacent.rs
··· 1 - //! Extract adjacent MST keys from a CAR slice. 2 - //! 3 - //! Given the raw blocks included in a `getRecord` CAR slice, this module 4 - //! uses `jacquard-repo` primitives to find neighbouring record keys and detect 5 - //! collection boundaries. 6 - //! 7 - //! `getRecord` CAR slices are *partial* snapshots of the MST: they include only 8 - //! the blocks on the path from the root to the queried key. Sibling subtrees 9 - //! are referenced by CID but their blocks are absent. The cursor walks only 10 - //! the blocks that are present, skipping any subtree whose block is missing. 11 - 12 - use std::sync::Arc; 13 - 14 - use jacquard_repo::{ 15 - MemoryBlockStore, Mst, 16 - car::parse_car_bytes, 17 - commit::Commit, 18 - mst::{CursorPosition, MstCursor}, 19 - }; 20 - 21 - use crate::error::{Error, Result}; 22 - 23 - /// The adjacent keys returned by the MST for a given probe key. 24 - #[derive(Debug, Clone)] 25 - pub struct AdjacentKeys { 26 - /// The key immediately before the probe key in the MST, if any. 27 - pub prev: Option<String>, 28 - /// The key immediately after the probe key in the MST, if any. 29 - pub next: Option<String>, 30 - } 31 - 32 - /// Extract adjacent keys for `probe_key` from the given CAR block bytes. 33 - /// 34 - /// Parses the MST from the CAR, then walks the tree in sorted order using 35 - /// `MstCursor`, collecting the largest visible key less than `probe_key` 36 - /// (prev) and the smallest visible key greater than `probe_key` (next). 37 - /// 38 - /// Subtrees whose blocks are absent from the CAR are silently skipped; this 39 - /// is expected for `getRecord` slices, which contain only the proof path. 40 - pub async fn extract_adjacent(car_bytes: &[u8], probe_key: &str) -> Result<AdjacentKeys> { 41 - // Parse the CAR bytes into a root CID + block map. 42 - let parsed = parse_car_bytes(car_bytes) 43 - .await 44 - .map_err(|e| Error::Other(e.to_string()))?; 45 - 46 - // The CAR root is the signed commit; its `data` field is the MST root CID. 47 - let mst_root = { 48 - let commit_bytes = parsed 49 - .blocks 50 - .get(&parsed.root) 51 - .ok_or_else(|| Error::Other("getRecord CAR has no commit block".into()))?; 52 - let commit = Commit::from_cbor(commit_bytes.as_ref()) 53 - .map_err(|e| Error::Other(format!("bad commit in getRecord CAR: {}", e)))?; 54 - *commit.data() 55 - }; 56 - 57 - // Load all CAR blocks into an in-memory store and mount the MST lazily. 58 - let storage = Arc::new(MemoryBlockStore::new_from_blocks(parsed.blocks)); 59 - let mst = Mst::load(storage, mst_root, None); 60 - 61 - // Walk the tree with a cursor in sorted key order. 62 - // The cursor starts pointing at the MST root (a Tree position); we process 63 - // each position as we encounter it. 64 - let mut cursor = MstCursor::new(mst); 65 - let mut prev: Option<String> = None; 66 - let mut next: Option<String> = None; 67 - 68 - loop { 69 - match cursor.current() { 70 - CursorPosition::End => break, 71 - 72 - CursorPosition::Leaf { key, .. } => { 73 - let k = key.as_str(); 74 - if k < probe_key { 75 - // This is a candidate for prev; keep the latest one seen. 76 - prev = Some(k.to_owned()); 77 - // Advance past this leaf (step_over; never fails). 78 - if cursor.advance().await.is_err() { 79 - break; 80 - } 81 - } else if k > probe_key { 82 - // First key greater than probe_key in sorted order. 83 - next = Some(k.to_owned()); 84 - break; 85 - } else { 86 - // k == probe_key: the record actually exists; skip it. 87 - if cursor.advance().await.is_err() { 88 - break; 89 - } 90 - } 91 - } 92 - 93 - CursorPosition::Tree { .. } => { 94 - // Descend into the subtree. If the block is absent from the 95 - // CAR (the cursor returns Err), skip the subtree instead. 96 - match cursor.advance().await { 97 - Ok(()) => {} 98 - Err(_) => { 99 - if cursor.skip_subtree().await.is_err() { 100 - break; 101 - } 102 - } 103 - } 104 - } 105 - } 106 - } 107 - 108 - Ok(AdjacentKeys { prev, next }) 109 - } 110 - 111 - /// Estimate whether this is a small repo by inspecting the MST level of the 112 - /// highest-level node present in `car_bytes`. 113 - /// 114 - /// Uses `jacquard_repo::mst::util::layer_for_key()` on the highest-level key 115 - /// found in the partial MST slice. Small repos are statistically unlikely to 116 - /// contain high-level keys, since every CAR slice must include all maximum-level 117 - /// keys of the repository. 118 - pub fn is_small_repo(_car_bytes: &[u8], _threshold_level: u8) -> crate::error::Result<bool> { 119 - todo!("parse MST nodes from CAR blocks and check maximum key level") 120 - }
+72
src/mst/collections.rs
··· 1 + use jacquard_common::types::string::Nsid; 2 + use repo_stream::MemCar; 3 + use std::collections::HashSet; 4 + 5 + #[derive(Debug, thiserror::Error)] 6 + pub enum MstCollectionsError { 7 + #[error("repo-stream WalkError: {0}")] 8 + WalkError(#[from] repo_stream::WalkError), 9 + #[error("bad repo path: {0}")] 10 + BadPath(String), 11 + } 12 + 13 + type Result<T> = std::result::Result<T, MstCollectionsError>; 14 + 15 + /// get whatever collections can be found from a possibly-partial CAR 16 + /// 17 + /// linear scan of all present keys, no seeks -- expects a small CAR slice 18 + pub fn extract_from_slice(car: &mut MemCar) -> Result<HashSet<Nsid<'static>>> { 19 + let mut collections = HashSet::new(); 20 + while let Some((path, _)) = car.next_keys()? { 21 + let (nsid_str, _) = path.split_once('/').ok_or(MstCollectionsError::BadPath( 22 + "missing '/' separator".to_string(), 23 + ))?; 24 + let collection = nsid_str 25 + .parse() 26 + .map_err(|e| MstCollectionsError::BadPath(format!("nsid parse: {e}")))?; 27 + collections.insert(collection); 28 + } 29 + Ok(collections) 30 + } 31 + 32 + /// get all collections from a full repo export CAR 33 + /// 34 + /// uses repo-stream's `seek` to skip over unnecesary subtrees 35 + /// 36 + /// ships the work out via spawn_blocking because large repos can take a bit, 37 + /// and disk-spilled repos (eventually) will be even more blocking 38 + /// 39 + /// TODO: have repo_stream include some size estimate so we can avoid 40 + /// spawn_blocking for small repos 41 + pub async fn extract_from_full(car: MemCar) -> Result<Vec<Nsid<'static>>> { 42 + tokio::task::spawn_blocking(move || extract_full_blocking_impl(car)) 43 + .await 44 + .unwrap() // don't suppress tokio-level errors we can't handle 45 + } 46 + 47 + /// skim the repo along collection boundaries to extract them all 48 + /// 49 + /// seeks past each collection using a sentinel that sorts strictly after any 50 + /// valid key in the collection. Rkeys in atproto are capped at 512 chars; 513 51 + /// tildes exceeds that maximum, so `collection/<513 tildes>` can never equal an 52 + /// actual record key and is guaranteed to be greater than 53 + /// `collection/<512 tildes>` (the max valid key). 54 + /// 55 + /// returns a lex-sorted list of all collections in the repo 56 + fn extract_full_blocking_impl(mut car: MemCar) -> Result<Vec<Nsid<'static>>> { 57 + let tilde_max = "~".repeat(513); 58 + 59 + let mut collections = Vec::new(); 60 + while let Some((path, _)) = car.next_keys_strict()? { 61 + let (nsid_str, _) = path.split_once('/').ok_or(MstCollectionsError::BadPath( 62 + "missing '/' separator".to_string(), 63 + ))?; 64 + let collection = nsid_str 65 + .parse() 66 + .map_err(|e| MstCollectionsError::BadPath(format!("nsid parse: {e}")))?; 67 + collections.push(collection); 68 + car.seek(&format!("{nsid_str}/{tilde_max}")).unwrap(); 69 + } 70 + collections.sort_unstable(); 71 + Ok(collections) 72 + }
+2 -7
src/mst/mod.rs
··· 1 - //! MST (Merkle Search Tree) utilities. 2 - //! 3 - //! Thin wrappers around `jacquard-repo` primitives for: 4 - //! - extracting adjacent keys from CAR slices 5 - //! - detecting collection boundaries 6 - //! - small-repo heuristic via MST level inspection 1 + //! mst (merkle search tree, the atproto repo structure) utils 7 2 8 - pub mod adjacent; 3 + pub mod collections;
+1 -1
src/sync/firehose/sync_event.rs
··· 142 142 && mst_root == prev.prev_data 143 143 { 144 144 metrics::counter!("lightrail_resync_avoided_total", "reason" => "no_op").increment(1); 145 - debug!(did = %did, "sync dropped: commit state unchanged (no-op migration)"); 145 + debug!(did = %did, "sync dropped: commit state unchanged (no-op, likely post-migration)"); 146 146 return Ok(()); 147 147 } 148 148
+204 -96
src/sync/resync/describe_repo.rs
··· 1 - //! Enumerate a repository's collections via `com.atproto.repo.describeRepo`. 1 + //! list repo collections with `com.atproto.repo.describeRepo` 2 + //! 3 + //! describeRepo includes a list of collections in the repo, so we can avoid 4 + //! downloading full repo contents just to build our tiny index if we use it. 5 + //! 6 + //! there are two downsides: 7 + //! 8 + //! 1. the list is not authenticated in the AT sense, so we are trusting the PDS 9 + //! to generate it correctly, 10 + //! 2. the response does not include the repo's `rev` or `data` (mst root hash) 11 + //! that we would need to assert that the next commit we see from them on the 12 + //! firehose actually follows from the describeRepo response without gaps. 2 13 //! 3 - //! The PDS returns its collection list directly in the response — one request, 4 - //! no CAR parsing. This is tried first because it is cheap and usually works. 5 - //! A second request to `com.atproto.sync.getLatestCommit` fetches the commit 6 - //! CID and rev that the `get_repo` path gets for free from the CAR. 14 + //! for the second, we try to separately obtain a usable `rev`+`data` pair 15 + //! separately from the `describeRepo` request, by calling `sync.getRecord`. 16 + //! the neat thing is, we don't need to know about an existing record to 17 + //! getRecord! PDSs return "proof-of-nonexistence" if your key doesn't point to 18 + //! anything, so we get a CAR with the Commit Object that we need, whether or 19 + //! not we asked for an existing record. 20 + //! 21 + //! as a bonus, we'll receive a small number of real MST keys in the CAR slice, 22 + //! which we can assert against the `describeRepo` collections as a first-line 23 + //! check of the PDS describeRepo response. 24 + //! 25 + //! getting the rev/data from `getRecord` before `describeRepo` is meant to help 26 + //! to our local state converge to the correct state in case there is an update 27 + //! between our two calls: 28 + //! 29 + //! - if a new collection was introduced: we'll "accidentally" add it early. 30 + //! when it re-arrives on the firehose, re-adding it is harmless. 31 + //! - similar with deletes: deleting what wasn't there is a db no-op. 32 + 33 + use jacquard_api::com_atproto::{ 34 + repo::describe_repo::DescribeRepo, 35 + sync::get_record::{GetRecord, GetRecordError}, 36 + }; 37 + use jacquard_common::{ 38 + IntoStatic, 39 + error::ClientErrorKind, 40 + http_client::HttpClient, 41 + types::string::{Cid, Did, Nsid, RecordKey, Rkey, Tid}, 42 + xrpc::{XrpcError, XrpcExt}, 43 + }; 44 + use repo_stream::{DriverBuilder, LoadError}; 45 + use std::collections::HashSet; 7 46 8 - use jacquard_api::com_atproto::repo::describe_repo::DescribeRepo; 9 - use jacquard_api::com_atproto::sync::get_latest_commit::GetLatestCommit; 10 - use jacquard_common::error::ClientErrorKind; 11 - use jacquard_common::http_client::HttpClient; 12 - use jacquard_common::types::cid::Cid; 13 - use jacquard_common::types::nsid::Nsid; 14 - use jacquard_common::types::string::Did; 15 - use jacquard_common::xrpc::XrpcExt; 47 + use super::{GetCollectionsError, RepoGoneReason, RepoSnapshot}; 16 48 17 - use super::{GetCollectionsError, RepoSnapshot}; 49 + /// for `getRecord` CAR slice (commit block + small MST proof path) 50 + const GET_RECORD_MEM_LIMIT_MB: usize = 1; 18 51 19 - /// Ask the PDS for its collection list and latest commit metadata. 52 + /// get repo collections the cheap (and less-authenticated-data-y) way 20 53 /// 21 - /// Makes two requests: `describeRepo` for the collection list, then 22 - /// `getLatestCommit` for the commit CID and revision. 54 + /// - does a phony `getRecord` for a Commit (-> rev/data sync anchor) and a 55 + /// small set of proof keys we can use to check 56 + /// - the `describeRepo`'s collection list from the repo 23 57 pub async fn fetch_collections<C>( 24 58 client: &C, 25 59 base: &jacquard_common::url::Url, ··· 28 62 where 29 63 C: HttpClient + Sync, 30 64 { 31 - // Step 1: commit CID + rev from getLatestCommit. 32 - // 33 - // We intentionally fetch commit metadata *before* the collection list. 34 - // If a commit lands between the two requests, we end up with a rev that 35 - // is slightly behind the collections we observed — our index may contain 36 - // a collection that didn't exist at the stored rev (a false positive), 37 - // but it will never *miss* a collection relative to that rev (a false 38 - // negative). False positives self-correct: the firehose delivers events 39 - // from our stored rev onward, so the racing commit will arrive and 40 - // reconcile the index. With the opposite order we'd store a rev ahead of 41 - // our collection snapshot; the firehose would then skip the racing commit 42 - // entirely, leaving the index persistently stale. 43 - // 44 - // TODO: think about this more. we really want to avoid adding a collection 45 - // twice so that we can safely do fjall weak deletes. if we can't assure 46 - // that then we should do full regular tombstone deletes. 47 - let req = GetLatestCommit { did: did.clone() }; 48 - let host = base.host_str().unwrap_or("").to_string(); 49 - let resp = client 65 + // step 1: commit rev + data CID from getRecord, plus a few collections 66 + let req = GetRecord { 67 + collection: Nsid::new_static("blue.microcosm.lightrail.probe").unwrap(), 68 + did: did.clone(), 69 + rkey: RecordKey(Rkey::new_static("any").unwrap()), 70 + }; 71 + let host = base 72 + .host_str() 73 + .ok_or(GetCollectionsError::Request(format!( 74 + "failed to get host from {base:?}" 75 + )))? 76 + .to_string(); 77 + 78 + let output = client 50 79 .xrpc(base.clone()) 51 80 .send(&req) 52 81 .await 53 - .map_err(|e| classify_send_error(e, &host))?; 54 - let (rev, commit) = match resp.parse() { 55 - Ok(output) => { 56 - let commit = output.cid.to_ipld().map(Cid::ipld).map_err(|e| { 57 - GetCollectionsError::InvalidData(format!("invalid commit CID: {e}")) 58 - })?; 59 - (output.rev, commit) 60 - } 61 - Err(e) => return Err(GetCollectionsError::Request(e.to_string())), 62 - }; 82 + .map_err(|e| classify_send_error(e, &host))? 83 + .parse() 84 + .map_err(|e| match e { 85 + XrpcError::Xrpc(e) => classify_get_record_error(e), 86 + e => GetCollectionsError::Request(e.to_string()), 87 + })?; 88 + 89 + let reader = tokio::io::BufReader::new(std::io::Cursor::new(output.body)); 90 + let mut mem_car = DriverBuilder::new() 91 + .with_mem_limit_mb(GET_RECORD_MEM_LIMIT_MB) 92 + .load_car(reader) 93 + .await 94 + .map_err(|e| match e { 95 + LoadError::MemoryLimitReached(_) => GetCollectionsError::InvalidData(format!( 96 + "getRecord CAR exceeds {GET_RECORD_MEM_LIMIT_MB} MiB limit" 97 + )), 98 + e => GetCollectionsError::InvalidData(format!("getRecord CAR error: {e}")), 99 + })?; 100 + 101 + let rev = Tid::new(&mem_car.commit.rev) 102 + .map_err(|e| GetCollectionsError::InvalidData(format!("bad rev in commit: {e}")))?; 103 + 104 + // shuffle repo-stream raw cid => jacquard cid wrapper 105 + let mst_root = Cid::ipld(mem_car.commit.data.clone().into()); 106 + 107 + let proven_collections = 108 + crate::mst::collections::extract_from_slice(&mut mem_car).map_err(|e| { 109 + GetCollectionsError::InvalidData(format!("failed to extract collections from CAR: {e}")) 110 + })?; 63 111 64 - // Step 2: collection list from describeRepo. 65 - let req = DescribeRepo { 66 - repo: did.clone().into(), 67 - }; 68 - let resp = client 112 + // step 2: collection list from describeRepo. 113 + let collections: HashSet<Nsid<'_>> = client 69 114 .xrpc(base.clone()) 70 - .send(&req) 115 + .send(&DescribeRepo { 116 + repo: did.clone().into(), 117 + }) 71 118 .await 72 - .map_err(|e| classify_send_error(e, &host))?; 73 - let mut collections: Vec<Nsid<'static>> = match resp.parse() { 74 - Ok(output) => output 75 - .collections 76 - .iter() 77 - .map(|c| { 78 - Nsid::new_owned(c.as_str()).map_err(|e| { 79 - GetCollectionsError::InvalidData(format!("invalid collection NSID: {e}")) 80 - }) 81 - }) 82 - .collect::<Result<Vec<_>, _>>()?, 83 - Err(e) => return Err(GetCollectionsError::Request(e.to_string())), 84 - }; 119 + .map_err(|e| classify_send_error(e, &host))? 120 + .parse() 121 + .map_err(|e| GetCollectionsError::Request(e.to_string()))? // Request => get_repo fallback 122 + .collections 123 + .into_iter() 124 + .map(IntoStatic::into_static) 125 + .collect(); 126 + 127 + // post-processing check: collections we know from getRecord present 128 + if !collections.is_superset(&proven_collections) { 129 + return Err(GetCollectionsError::InvalidData( 130 + "describeRepo inconsistent collection set".to_string(), 131 + )); 132 + } 133 + 134 + // final shape and order 135 + let mut collections = Vec::from_iter(collections); 85 136 collections.sort_unstable(); 86 137 87 138 Ok(RepoSnapshot { 88 139 collections, 89 140 rev, 90 - commit, 141 + commit: Some(mst_root), 91 142 }) 92 143 } 93 144 145 + fn classify_get_record_error(e: GetRecordError<'_>) -> GetCollectionsError { 146 + match e { 147 + GetRecordError::RepoNotFound(_) => GetCollectionsError::RepoNotFound, 148 + GetRecordError::RepoTakendown(_) => { 149 + GetCollectionsError::RepoGone(RepoGoneReason::Takendown) 150 + } 151 + GetRecordError::RepoSuspended(_) => { 152 + GetCollectionsError::RepoGone(RepoGoneReason::Suspended) 153 + } 154 + GetRecordError::RepoDeactivated(_) => { 155 + GetCollectionsError::RepoGone(RepoGoneReason::Deactivated) 156 + } 157 + // RecordNotFound or unknown: treat as transient; getRepo fallback will be attempted. 158 + e => GetCollectionsError::Request(e.to_string()), 159 + } 160 + } 161 + 94 162 fn classify_send_error(e: jacquard_common::error::ClientError, host: &str) -> GetCollectionsError { 95 163 if matches!(e.kind(), ClientErrorKind::Http { status } if status.as_u16() == 429) { 96 164 GetCollectionsError::RateLimited(host.to_string()) ··· 102 170 #[cfg(test)] 103 171 mod tests { 104 172 use super::*; 105 - use wiremock::matchers::{method, path, query_param}; 106 - use wiremock::{Mock, MockServer, ResponseTemplate}; 173 + use std::sync::Arc; 107 174 175 + use bytes::Bytes; 108 176 use jacquard_common::types::nsid::Nsid; 109 177 use jacquard_common::types::string::Did; 178 + use jacquard_common::types::tid::Tid; 179 + use jacquard_repo::commit::Commit; 180 + use jacquard_repo::{BlockStore, MemoryBlockStore, Mst, car::write_car_bytes}; 181 + use wiremock::matchers::{method, path, query_param}; 182 + use wiremock::{Mock, MockServer, ResponseTemplate}; 110 183 111 184 fn nsid(s: &str) -> Nsid<'static> { 112 185 Nsid::new_owned(s).unwrap() 113 186 } 114 187 115 188 const DID: &str = "did:web:example.com"; 116 - // A plausible commit CID and rev for mock responses. 117 - const TEST_CID: &str = "bafyreib2rxk3rybk3aobmv5cjuql3bm2twh4jo5uxgf5kpqcsgz7soitae"; 118 - const TEST_REV: &str = "3lczouzaqmo2e"; 119 189 120 190 fn did() -> Did<'static> { 121 191 Did::new_owned(DID).unwrap() 122 192 } 123 193 194 + /// Build a minimal valid CAR for use as a `getRecord` response. 195 + /// 196 + /// Returns `(car_bytes, mst_root_cid_string, rev_string)`. 197 + async fn make_car(keys: &[&str]) -> (Vec<u8>, String, String) { 198 + let storage = Arc::new(MemoryBlockStore::new()); 199 + let mut mst = Mst::new(storage.clone()); 200 + 201 + let dummy_cid = storage.put(b"record").await.unwrap(); 202 + for key in keys { 203 + mst = mst.add(key, dummy_cid).await.unwrap(); 204 + } 205 + 206 + let (mst_root, mut blocks) = mst.collect_blocks().await.unwrap(); 207 + let mst_root_str = mst_root.to_string(); 208 + 209 + let commit = Commit { 210 + did: Did::new_owned(DID).unwrap(), 211 + version: 3, 212 + data: mst_root, 213 + rev: Tid::now_0(), 214 + prev: None, 215 + sig: Bytes::from(vec![0u8; 64]), 216 + }; 217 + let commit_cid = commit.to_cid().unwrap(); 218 + let rev_str = commit.rev.to_string(); 219 + let commit_cbor = Bytes::from(commit.to_cbor().unwrap()); 220 + blocks.insert(commit_cid, commit_cbor); 221 + 222 + let car = write_car_bytes(commit_cid, blocks).await.unwrap(); 223 + (car, mst_root_str, rev_str) 224 + } 225 + 124 226 fn describe_repo_response(collections: &[&str]) -> serde_json::Value { 125 227 serde_json::json!({ 126 228 "did": DID, ··· 131 233 }) 132 234 } 133 235 134 - fn latest_commit_response() -> serde_json::Value { 135 - serde_json::json!({ "cid": TEST_CID, "rev": TEST_REV }) 136 - } 137 - 138 236 /// Mount mocks for both endpoints on a single server. 139 237 async fn mock_server( 140 238 describe_template: ResponseTemplate, 141 - latest_template: ResponseTemplate, 239 + get_record_car: Vec<u8>, 142 240 ) -> MockServer { 143 241 let server = MockServer::start().await; 144 242 Mock::given(method("GET")) ··· 148 246 .mount(&server) 149 247 .await; 150 248 Mock::given(method("GET")) 151 - .and(path("/xrpc/com.atproto.sync.getLatestCommit")) 249 + .and(path("/xrpc/com.atproto.sync.getRecord")) 152 250 .and(query_param("did", DID)) 153 - .respond_with(latest_template) 251 + .and(query_param("collection", "blue.microcosm.lightrail.probe")) 252 + .and(query_param("rkey", "any")) 253 + .respond_with( 254 + ResponseTemplate::new(200) 255 + .insert_header("content-type", "application/vnd.ipld.car") 256 + .set_body_bytes(get_record_car), 257 + ) 154 258 .mount(&server) 155 259 .await; 156 260 server ··· 158 262 159 263 #[tokio::test] 160 264 async fn returns_collections_sorted() { 265 + let (car, _, _) = make_car(&["app.bsky.actor.profile/self"]).await; 161 266 let server = mock_server( 162 267 ResponseTemplate::new(200).set_body_json(describe_repo_response(&[ 163 268 "app.bsky.feed.post", 164 269 "app.bsky.actor.profile", 165 270 ])), 166 - ResponseTemplate::new(200).set_body_json(latest_commit_response()), 271 + car, 167 272 ) 168 273 .await; 169 274 ··· 181 286 #[tokio::test] 182 287 async fn sorts_server_order() { 183 288 // Server returns collections in reverse lexicographic order. 289 + let (car, _, _) = make_car(&["app.bsky.actor.profile/self"]).await; 184 290 let server = mock_server( 185 291 ResponseTemplate::new(200).set_body_json(describe_repo_response(&[ 186 292 "app.bsky.graph.follow", 187 293 "app.bsky.feed.post", 188 294 "app.bsky.actor.profile", 189 295 ])), 190 - ResponseTemplate::new(200).set_body_json(latest_commit_response()), 296 + car, 191 297 ) 192 298 .await; 193 299 ··· 207 313 208 314 #[tokio::test] 209 315 async fn returns_empty_vec_when_no_collections() { 316 + let (car, _, _) = make_car(&[]).await; 210 317 let server = mock_server( 211 318 ResponseTemplate::new(200).set_body_json(describe_repo_response(&[])), 212 - ResponseTemplate::new(200).set_body_json(latest_commit_response()), 319 + car, 213 320 ) 214 321 .await; 215 322 ··· 221 328 } 222 329 223 330 #[tokio::test] 224 - async fn returns_commit_cid_and_rev() { 331 + async fn returns_mst_root_cid_and_rev() { 332 + // The MST root CID (commit.data) and rev round-trip correctly. 333 + let (car, expected_mst_root, expected_rev) = 334 + make_car(&["app.bsky.actor.profile/self"]).await; 225 335 let server = mock_server( 226 336 ResponseTemplate::new(200) 227 - .set_body_json(describe_repo_response(&["app.bsky.feed.post"])), 228 - ResponseTemplate::new(200).set_body_json(latest_commit_response()), 337 + .set_body_json(describe_repo_response(&["app.bsky.actor.profile"])), 338 + car, 229 339 ) 230 340 .await; 231 341 ··· 233 343 let base = server.uri().parse().unwrap(); 234 344 let snapshot = fetch_collections(&client, &base, did()).await.unwrap(); 235 345 236 - assert_eq!(snapshot.commit.as_str(), TEST_CID); 237 - assert_eq!(snapshot.rev.as_str(), TEST_REV); 346 + assert_eq!(snapshot.commit.unwrap().as_str(), expected_mst_root); 347 + assert_eq!(snapshot.rev.as_str(), expected_rev); 238 348 } 239 349 240 350 #[tokio::test] 241 - async fn errors_on_latest_commit_failure() { 242 - // getLatestCommit fails; describeRepo is never reached. 351 + async fn errors_on_get_record_failure() { 352 + // getRecord fails; describeRepo is never reached. 243 353 let server = MockServer::start().await; 244 354 Mock::given(method("GET")) 245 - .and(path("/xrpc/com.atproto.sync.getLatestCommit")) 355 + .and(path("/xrpc/com.atproto.sync.getRecord")) 246 356 .respond_with(ResponseTemplate::new(500)) 247 357 .mount(&server) 248 358 .await; ··· 256 366 257 367 #[tokio::test] 258 368 async fn errors_on_describe_repo_failure() { 259 - // getLatestCommit succeeds but describeRepo returns an XRPC error. 369 + // getRecord succeeds but describeRepo returns an XRPC error. 370 + let (car, _, _) = make_car(&["app.bsky.actor.profile/self"]).await; 260 371 let server = mock_server( 261 372 ResponseTemplate::new(400).set_body_json(serde_json::json!({ 262 373 "error": "InvalidRequest", 263 374 "message": "bad request", 264 375 })), 265 - ResponseTemplate::new(200).set_body_json(latest_commit_response()), 376 + car, 266 377 ) 267 378 .await; 268 379 ··· 276 387 #[tokio::test] 277 388 async fn errors_on_server_error() { 278 389 // A 5xx on describeRepo is a transient failure; surfaces as Err. 279 - let server = mock_server( 280 - ResponseTemplate::new(500), 281 - ResponseTemplate::new(200).set_body_json(latest_commit_response()), 282 - ) 283 - .await; 390 + let (car, _, _) = make_car(&["app.bsky.actor.profile/self"]).await; 391 + let server = mock_server(ResponseTemplate::new(500), car).await; 284 392 285 393 let client = reqwest::Client::new(); 286 394 let base = server.uri().parse().unwrap();
+4 -10
src/sync/resync/dispatcher.rs
··· 417 417 } 418 418 419 419 fn backoff_secs(retry_count: u16) -> u64 { 420 - // 60 s, 120 s, 240 s, 480 s, 960 s, 1920 s, 3600 s (capped) 420 + // 60s, 120s, 240s, 480s, 960s, 1920s, 3600s (capped) 421 421 let exp = (retry_count as u32).saturating_sub(1).min(5); 422 422 (60u64 * (1u64 << exp)).min(3600) 423 423 } ··· 427 427 /// Returns `None` when we've exhausted retries — the `RepoState::NotFound` 428 428 /// written by `index_repo` is the terminal state; future firehose activity 429 429 /// for the DID will re-trigger processing normally. 430 - /// 431 - /// Schedule (retry_count after incrementing): 432 - /// 1 → 6 h (covers transient mid-migration windows) 433 - /// 2 → 24 h (covers stale DID→PDS cache expiry) 434 - /// 3 → 24 h (one more check before giving up) 435 - /// ≥4 → None 436 430 fn not_found_backoff_secs(retry_count: u16) -> Option<u64> { 437 431 match retry_count { 438 - 1 => Some(6 * 3600), // 6 h 439 - 2 => Some(24 * 3600), // 24 h 440 - 3 => Some(24 * 3600), // 24 h 432 + 1 => Some(6 * 3600), // 6h 433 + 2 => Some(24 * 3600), // 24h 434 + 3 => Some(24 * 3600), // 24h 441 435 _ => None, 442 436 } 443 437 }
+95 -124
src/sync/resync/get_repo.rs
··· 1 - //! Enumerate a repository's collections via `com.atproto.sync.getRepo`. 1 + //! list repo collections with `com.atproto.sync.getRepo` 2 2 //! 3 - //! Fetches the full repository CAR and walks every MST leaf to collect all 4 - //! collection NSIDs. Used as a fallback when `describeRepo` is unavailable or 5 - //! returns an empty list. 3 + //! hooks jacquard's streaming response body up with repo-stream's async read 4 + //! input so that repo-stream's memory limit will stop a malicious server from 5 + //! sending back an unlimited stream of garbage CAR blocks (eventually). 6 6 7 - use bytes::Bytes; 8 - use jacquard_api::com_atproto::sync::get_repo::{GetRepo, GetRepoError}; 9 - use jacquard_common::error::ClientErrorKind; 10 - use jacquard_common::http_client::HttpClient; 11 - use jacquard_common::types::cid::Cid; 12 - use jacquard_common::types::nsid::Nsid; 7 + use jacquard_api::com_atproto::sync::get_repo::GetRepo; 8 + use jacquard_common::http_client::{HttpClient, HttpClientExt}; 9 + use jacquard_common::stream::ByteStream; 13 10 use jacquard_common::types::string::Did; 14 - use jacquard_common::xrpc::{XrpcError, XrpcExt}; 15 - use jacquard_repo::{car::parse_car_bytes, commit::Commit}; 16 - use repo_stream::DriverBuilder; 11 + use jacquard_common::types::tid::Tid; 12 + use jacquard_common::xrpc::XrpcExt; 13 + use repo_stream::{DriverBuilder, LoadError}; 17 14 18 15 use super::{GetCollectionsError, RepoGoneReason, RepoSnapshot}; 19 16 20 - async fn fetch_repo_car<C>( 17 + /// limit for in-memory CAR loading, MiB. 18 + const CAR_MEM_LIMIT_MB: usize = 1024; 19 + 20 + /// collections from full repo car with repo-stream 21 + pub async fn fetch_collections<C>( 21 22 client: &C, 22 23 base: &jacquard_common::url::Url, 23 24 did: Did<'_>, 24 - ) -> std::result::Result<Bytes, GetCollectionsError> 25 + ) -> std::result::Result<RepoSnapshot, GetCollectionsError> 25 26 where 26 - C: HttpClient + Sync, 27 + C: HttpClient + HttpClientExt + Sync, 27 28 { 28 29 let req = GetRepo { did, since: None }; 29 - 30 30 let host = base.host_str().unwrap_or("").to_string(); 31 - let resp = client.xrpc(base.clone()).send(&req).await.map_err(|e| { 32 - if matches!(e.kind(), ClientErrorKind::Http { status } if status.as_u16() == 429) { 33 - GetCollectionsError::RateLimited(host) 34 - } else { 35 - GetCollectionsError::Request(e.to_string()) 36 - } 37 - })?; 38 31 39 - match resp.parse() { 40 - Ok(output) => Ok(output.body), 41 - Err(XrpcError::Xrpc(err)) => match err { 42 - GetRepoError::RepoNotFound(_) => Err(GetCollectionsError::RepoNotFound), 43 - GetRepoError::RepoTakendown(_) => { 44 - Err(GetCollectionsError::RepoGone(RepoGoneReason::Takendown)) 45 - } 46 - GetRepoError::RepoSuspended(_) => { 47 - Err(GetCollectionsError::RepoGone(RepoGoneReason::Suspended)) 48 - } 49 - GetRepoError::RepoDeactivated(_) => { 50 - Err(GetCollectionsError::RepoGone(RepoGoneReason::Deactivated)) 51 - } 52 - GetRepoError::Unknown(ref data) => { 53 - // Some PDSes send {"error":"NotFound"} instead of the 54 - // standardised "RepoNotFound" XRPC error code. 55 - let is_not_found = data 56 - .as_object() 57 - .and_then(|obj| obj.get("error")) 58 - .and_then(|v| v.as_str()) 59 - == Some("NotFound"); 60 - if is_not_found { 61 - Err(GetCollectionsError::RepoNotFound) 62 - } else { 63 - Err(GetCollectionsError::UnexpectedXrpc(format!("{data:?}"))) 64 - } 65 - } 66 - }, 67 - Err(e) => Err(GetCollectionsError::Request(e.to_string())), 32 + let response = client 33 + .xrpc(base.clone()) 34 + .download(&req) 35 + .await 36 + .map_err(|e| GetCollectionsError::Request(e.to_string()))?; 37 + 38 + let status = response.status(); 39 + let (_, body) = response.into_parts(); 40 + 41 + if status.as_u16() == 429 { 42 + return Err(GetCollectionsError::RateLimited(host)); 43 + } 44 + if !status.is_success() { 45 + return Err(classify_xrpc_error(body, status).await); 68 46 } 69 - } 70 47 71 - /// Fetch the full repo CAR and walk every MST leaf to collect all collections. 72 - pub async fn fetch_collections<C>( 73 - client: &C, 74 - base: &jacquard_common::url::Url, 75 - did: Did<'_>, 76 - ) -> std::result::Result<RepoSnapshot, GetCollectionsError> 77 - where 78 - C: HttpClient + Sync, 79 - { 80 - let car = fetch_repo_car(client, base, did).await?; 48 + // Bridge ByteStream → AsyncRead. 49 + // n0_future::Stream is futures_lite::Stream which re-exports 50 + // futures_core::stream::Stream, so StreamReader accepts the inner stream. 51 + use futures::StreamExt as _; 52 + use tokio_util::io::StreamReader; 53 + 54 + let reader = tokio::io::BufReader::new(StreamReader::new( 55 + body.into_inner().map(|r| r.map_err(std::io::Error::other)), 56 + )); 81 57 82 - let parsed = parse_car_bytes(&car) 58 + let mem_car = DriverBuilder::new() 59 + .with_mem_limit_mb(CAR_MEM_LIMIT_MB) 60 + .load_car(reader) 83 61 .await 84 - .map_err(|e| GetCollectionsError::InvalidData(format!("CAR parse error: {e}")))?; 62 + .map_err(|e| match e { 63 + LoadError::MemoryLimitReached(_) => GetCollectionsError::InvalidData(format!( 64 + "repo CAR exceeds {CAR_MEM_LIMIT_MB} MiB memory limit" 65 + )), 66 + e => GetCollectionsError::InvalidData(format!("CAR load error: {e}")), 67 + })?; 85 68 86 - // IpldCid is Copy so we can capture the commit CID before moving `parsed`. 87 - let commit = Cid::ipld(parsed.root); 69 + let rev = Tid::new(&mem_car.commit.rev) 70 + .map_err(|e| GetCollectionsError::InvalidData(format!("bad rev in commit: {e}")))?; 88 71 89 - let rev = { 90 - let commit_bytes = parsed.blocks.get(&parsed.root).ok_or_else(|| { 91 - GetCollectionsError::InvalidData("getRepo CAR has no commit block".into()) 92 - })?; 93 - Commit::from_cbor(commit_bytes.as_ref()) 94 - .map_err(|e| { 95 - GetCollectionsError::InvalidData(format!("bad commit in getRepo CAR: {e}")) 96 - })? 97 - .rev 98 - }; 72 + let collections = crate::mst::collections::extract_from_full(mem_car) 73 + .await 74 + .map_err(|e| GetCollectionsError::InvalidData(format!("bad CAR walk: {e}")))?; 75 + 76 + Ok(RepoSnapshot { 77 + collections, 78 + rev, 79 + commit: None, // load_car does not expose the CAR root (commit) CID 80 + }) 81 + } 99 82 100 - // Load into repo-stream for MST walking (lighter than jacquard's Mst/MstCursor). 101 - let mut mem_car = DriverBuilder::new() 102 - .load_jacquard_parsed_car(parsed) 103 - .map_err(|e| GetCollectionsError::InvalidData(format!("CAR load error: {e}")))?; 83 + /// Read and parse a non-2xx response body as an XRPC error object. 84 + async fn classify_xrpc_error(body: ByteStream, status: http::StatusCode) -> GetCollectionsError { 85 + use futures::StreamExt as _; 86 + use tokio::io::AsyncReadExt as _; 87 + use tokio_util::io::StreamReader; 104 88 105 - // MST keys are `<collection>/<rkey>` in lexicographic order. next_keys() 106 - // yields only record leaves (MST nodes are traversed but not emitted). 107 - // When one collection name is a prefix of another (e.g. `sh.tangled.repo` 108 - // and `sh.tangled.repo.issue`), `.` (46) < `/` (47) means the longer name 109 - // sorts first; we sort after collecting to guarantee lexicographic output. 110 - let mut collections: Vec<Nsid<'static>> = Vec::new(); 111 - let mut last_col: Option<String> = None; 89 + let mut reader = StreamReader::new(body.into_inner().map(|r| r.map_err(std::io::Error::other))); 90 + let mut bytes = Vec::new(); 91 + if let Err(e) = reader.read_to_end(&mut bytes).await { 92 + return GetCollectionsError::Request(e.to_string()); 93 + }; 112 94 113 - while let Some((key, _)) = mem_car 114 - .next_keys() 115 - .map_err(|e| GetCollectionsError::InvalidData(format!("MST traversal error: {e}")))? 116 - { 117 - if let Some(col) = key.split_once('/').map(|(c, _)| c) 118 - && last_col.as_deref() != Some(col) 119 - { 120 - let nsid = Nsid::new_owned(col).map_err(|e| { 121 - GetCollectionsError::InvalidData(format!("invalid collection NSID in MST: {e}")) 122 - })?; 123 - last_col = Some(col.to_owned()); 124 - collections.push(nsid); 95 + // TODO: really we should be parsing directly to a type from jacquard here 96 + if let Ok(val) = serde_json::from_slice::<serde_json::Value>(&bytes) { 97 + match val.get("error").and_then(serde_json::Value::as_str) { 98 + Some("RepoNotFound" | "NotFound") => return GetCollectionsError::RepoNotFound, 99 + Some("RepoTakendown") => { 100 + return GetCollectionsError::RepoGone(RepoGoneReason::Takendown); 101 + } 102 + Some("RepoSuspended") => { 103 + return GetCollectionsError::RepoGone(RepoGoneReason::Suspended); 104 + } 105 + Some("RepoDeactivated") => { 106 + return GetCollectionsError::RepoGone(RepoGoneReason::Deactivated); 107 + } 108 + Some(other) => return GetCollectionsError::UnexpectedXrpc(other.to_owned()), 109 + None => {} 125 110 } 126 111 } 127 112 128 - collections.sort_unstable(); 129 - Ok(RepoSnapshot { 130 - collections, 131 - rev, 132 - commit, 133 - }) 113 + GetCollectionsError::Request(format!("HTTP {status}")) 134 114 } 135 115 136 116 #[cfg(test)] ··· 159 139 160 140 /// Build a minimal valid getRepo CAR from a list of MST keys. 161 141 /// 162 - /// Each key should be in `<collection>/<rkey>` format. All records point 163 - /// to a single dummy CID — only the key structure matters for these tests. 164 - /// The commit is unsigned (zero-filled sig); `fetch_collections` does not 165 - /// verify signatures. 166 - /// 167 - /// Returns `(car_bytes, commit_cid_string, rev_string)` so tests can assert 168 - /// that `fetch_collections` round-trips the commit metadata correctly. 142 + /// Returns `(car_bytes, commit_cid_string, rev_string)`. 169 143 async fn make_car(keys: &[&str]) -> (Vec<u8>, String, String) { 170 144 let storage = Arc::new(MemoryBlockStore::new()); 171 145 let mut mst = Mst::new(storage.clone()); 172 146 173 - // All records share one dummy CID — we only care about key structure. 174 147 let dummy_cid = storage.put(b"record").await.unwrap(); 175 148 for key in keys { 176 149 mst = mst.add(key, dummy_cid).await.unwrap(); ··· 239 212 240 213 #[tokio::test] 241 214 async fn sorts_collections_when_one_is_prefix_of_another() { 242 - // `sh.tangled.repo` is a prefix of `sh.tangled.repo.issue` and 243 - // `sh.tangled.repo.pull`. In the MST, `.` (46) < `/` (47) means 244 - // `sh.tangled.repo.issue/<rkey>` sorts before `sh.tangled.repo/<rkey>`, 245 - // so without an explicit sort the output would be in reverse order. 215 + // `sh.tangled.repo` is a prefix of `sh.tangled.repo.issue`. In the MST, 216 + // `.` (46) < `/` (47) so `sh.tangled.repo.issue/<rkey>` sorts before 217 + // `sh.tangled.repo/<rkey>`. Output must still be sorted lexicographically. 246 218 let (car, _, _) = make_car(&[ 247 219 "sh.tangled.repo/self", 248 220 "sh.tangled.repo.issue/abc123", ··· 278 250 } 279 251 280 252 #[tokio::test] 281 - async fn returns_commit_cid_and_rev() { 282 - // Verify that the commit CID and rev round-trip correctly through the CAR. 283 - let (car, expected_commit, expected_rev) = make_car(&["app.bsky.feed.post/abc123"]).await; 253 + async fn returns_rev() { 254 + // Rev round-trips correctly; commit CID is unavailable via load_car. 255 + let (car, _, expected_rev) = make_car(&["app.bsky.feed.post/abc123"]).await; 284 256 let server = serve_car(car).await; 285 257 286 258 let client = reqwest::Client::new(); 287 259 let base = server.uri().parse().unwrap(); 288 260 let snapshot = fetch_collections(&client, &base, did()).await.unwrap(); 289 261 290 - assert_eq!(snapshot.commit.as_str(), expected_commit); 291 262 assert_eq!(snapshot.rev.to_string(), expected_rev); 263 + assert!(snapshot.commit.is_none()); 292 264 } 293 265 294 266 #[tokio::test] ··· 313 285 #[tokio::test] 314 286 async fn errors_on_server_error() { 315 287 // A 5xx is a transient failure, not a signal that the repo is gone. 316 - // It should surface as Err so the caller can decide whether to retry. 317 288 let server = MockServer::start().await; 318 289 Mock::given(method("GET")) 319 290 .and(path("/xrpc/com.atproto.sync.getRepo"))
+9 -5
src/sync/resync/mod.rs
··· 22 22 use std::time::Duration; 23 23 24 24 use jacquard_common::IntoStatic; 25 - use jacquard_common::http_client::HttpClient; 25 + use jacquard_common::http_client::{HttpClient, HttpClientExt}; 26 26 use jacquard_common::types::{cid::Cid, nsid::Nsid, string::Did, tid::Tid}; 27 27 use tracing::info; 28 28 ··· 59 59 pub collections: Vec<Nsid<'static>>, 60 60 /// Revision TID of the latest commit. 61 61 pub rev: Tid, 62 - /// CID of the latest commit (as returned by `getRepo` or `getLatestCommit`). 63 - pub commit: Cid<'static>, 62 + /// CID of the latest commit, when available. 63 + /// 64 + /// Present when fetched via `describeRepo` + `getLatestCommit`. `None` 65 + /// when determined from a streaming `getRepo` CAR (repo-stream's `load_car` 66 + /// does not expose the CAR root CID). 67 + pub commit: Option<Cid<'static>>, 64 68 } 65 69 66 70 /// Why fetching a repository's collection list failed. ··· 113 117 get_repo_timeout: Duration, 114 118 ) -> Result<()> 115 119 where 116 - C: HttpClient + Sync, 120 + C: HttpClient + HttpClientExt + Sync, 117 121 { 118 122 // Own the DID for the duration of the function so we can move it into 119 123 // spawn_blocking closures (which require 'static captures). ··· 241 245 get_repo_timeout: Duration, 242 246 ) -> std::result::Result<RepoSnapshot, GetCollectionsError> 243 247 where 244 - C: HttpClient + Sync, 248 + C: HttpClient + HttpClientExt + Sync, 245 249 { 246 250 let describe_result = tokio::time::timeout( 247 251 describe_timeout,