lightweight
com.atproto.sync.listReposByCollection
wire resync rev/data to repo_prev for sync
+47
-33
+1
-1
src/examples/list_repo_collections.rs
+1
-1
src/examples/list_repo_collections.rs
···
41
41
if snapshot.collections.is_empty() {
42
42
println!("(no collections)");
43
43
} else {
44
-
println!("rev: {}, commit: {}", snapshot.rev, snapshot.commit);
44
+
println!("rev: {}, data: {:?}", snapshot.rev, snapshot.data);
45
45
for col in snapshot.collections {
46
46
println!("{col}");
47
47
}
+5
-4
src/sync/resync/describe_repo.rs
+5
-4
src/sync/resync/describe_repo.rs
···
38
38
IntoStatic,
39
39
error::ClientErrorKind,
40
40
http_client::HttpClient,
41
-
types::string::{Cid, Did, Nsid, RecordKey, Rkey, Tid},
41
+
types::string::{Did, Nsid, RecordKey, Rkey, Tid},
42
42
xrpc::{XrpcError, XrpcExt},
43
43
};
44
44
use repo_stream::{DriverBuilder, LoadError};
···
102
102
.map_err(|e| GetCollectionsError::InvalidData(format!("bad rev in commit: {e}")))?;
103
103
104
104
// shuffle repo-stream raw cid => jacquard cid wrapper
105
-
let mst_root = Cid::ipld(mem_car.commit.data.clone().into());
105
+
let data = mem_car.commit.data.clone().into();
106
106
107
107
let proven_collections =
108
108
crate::mst::collections::extract_from_slice(&mut mem_car).map_err(|e| {
···
138
138
Ok(RepoSnapshot {
139
139
collections,
140
140
rev,
141
-
commit: Some(mst_root),
141
+
data,
142
142
})
143
143
}
144
144
···
174
174
175
175
use bytes::Bytes;
176
176
use jacquard_common::types::nsid::Nsid;
177
+
use jacquard_common::types::string::Cid;
177
178
use jacquard_common::types::string::Did;
178
179
use jacquard_common::types::tid::Tid;
179
180
use jacquard_repo::commit::Commit;
···
343
344
let base = server.uri().parse().unwrap();
344
345
let snapshot = fetch_collections(&client, &base, did()).await.unwrap();
345
346
346
-
assert_eq!(snapshot.commit.unwrap().as_str(), expected_mst_root);
347
+
assert_eq!(Cid::ipld(snapshot.data).as_str(), expected_mst_root);
347
348
assert_eq!(snapshot.rev.as_str(), expected_rev);
348
349
}
349
350
+16
-13
src/sync/resync/get_repo.rs
+16
-13
src/sync/resync/get_repo.rs
···
5
5
//! sending back an unlimited stream of garbage CAR blocks (eventually).
6
6
7
7
use jacquard_api::com_atproto::sync::get_repo::GetRepo;
8
-
use jacquard_common::http_client::{HttpClient, HttpClientExt};
9
-
use jacquard_common::stream::ByteStream;
10
-
use jacquard_common::types::string::Did;
11
-
use jacquard_common::types::tid::Tid;
12
-
use jacquard_common::xrpc::XrpcExt;
8
+
use jacquard_common::{
9
+
http_client::{HttpClient, HttpClientExt},
10
+
stream::ByteStream,
11
+
types::string::{Did, Tid},
12
+
xrpc::XrpcExt,
13
+
};
13
14
use repo_stream::{DriverBuilder, LoadError};
14
15
15
16
use super::{GetCollectionsError, RepoGoneReason, RepoSnapshot};
···
69
70
let rev = Tid::new(&mem_car.commit.rev)
70
71
.map_err(|e| GetCollectionsError::InvalidData(format!("bad rev in commit: {e}")))?;
71
72
73
+
let data = mem_car.commit.data.clone().into();
74
+
72
75
let collections = crate::mst::collections::extract_from_full(mem_car)
73
76
.await
74
77
.map_err(|e| GetCollectionsError::InvalidData(format!("bad CAR walk: {e}")))?;
···
76
79
Ok(RepoSnapshot {
77
80
collections,
78
81
rev,
79
-
commit: None, // load_car does not expose the CAR root (commit) CID
82
+
data,
80
83
})
81
84
}
82
85
···
120
123
121
124
use bytes::Bytes;
122
125
use jacquard_common::types::nsid::Nsid;
126
+
use jacquard_common::types::string::Cid;
123
127
use jacquard_common::types::string::Did;
124
128
use jacquard_common::types::tid::Tid;
125
129
···
139
143
140
144
/// Build a minimal valid getRepo CAR from a list of MST keys.
141
145
///
142
-
/// Returns `(car_bytes, commit_cid_string, rev_string)`.
146
+
/// Returns `(car_bytes, mst_root_cid_string, rev_string)`.
143
147
async fn make_car(keys: &[&str]) -> (Vec<u8>, String, String) {
144
148
let storage = Arc::new(MemoryBlockStore::new());
145
149
let mut mst = Mst::new(storage.clone());
···
150
154
}
151
155
152
156
let (mst_root, mut blocks) = mst.collect_blocks().await.unwrap();
157
+
let mst_root_str = mst_root.to_string();
153
158
154
159
let commit = Commit {
155
160
did: Did::new_owned(DID).unwrap(),
···
160
165
sig: Bytes::from(vec![0u8; 64]),
161
166
};
162
167
let commit_cid = commit.to_cid().unwrap();
163
-
let commit_cid_str = commit_cid.to_string();
164
168
let rev_str = commit.rev.to_string();
165
169
let commit_cbor = Bytes::from(commit.to_cbor().unwrap());
166
170
blocks.insert(commit_cid, commit_cbor);
167
171
168
172
let car = write_car_bytes(commit_cid, blocks).await.unwrap();
169
-
(car, commit_cid_str, rev_str)
173
+
(car, mst_root_str, rev_str)
170
174
}
171
175
172
176
async fn serve_car(car: Vec<u8>) -> MockServer {
···
250
254
}
251
255
252
256
#[tokio::test]
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;
257
+
async fn returns_rev_and_mst_root() {
258
+
let (car, expected_mst_root, expected_rev) = make_car(&["app.bsky.feed.post/abc123"]).await;
256
259
let server = serve_car(car).await;
257
260
258
261
let client = reqwest::Client::new();
···
260
263
let snapshot = fetch_collections(&client, &base, did()).await.unwrap();
261
264
262
265
assert_eq!(snapshot.rev.to_string(), expected_rev);
263
-
assert!(snapshot.commit.is_none());
266
+
assert_eq!(Cid::ipld(snapshot.data).as_str(), expected_mst_root);
264
267
}
265
268
266
269
#[tokio::test]
+25
-15
src/sync/resync/mod.rs
+25
-15
src/sync/resync/mod.rs
···
21
21
22
22
use std::time::Duration;
23
23
24
-
use jacquard_common::IntoStatic;
25
-
use jacquard_common::http_client::{HttpClient, HttpClientExt};
26
-
use jacquard_common::types::{cid::Cid, nsid::Nsid, string::Did, tid::Tid};
24
+
use cid::Cid as RawCid;
25
+
use jacquard_common::{
26
+
IntoStatic,
27
+
http_client::{HttpClient, HttpClientExt},
28
+
types::string::{Did, Nsid, Tid},
29
+
};
27
30
use tracing::info;
28
31
29
32
use crate::storage::{
30
33
DbRef,
31
-
repo::{AccountStatus, RepoInfo, RepoState},
34
+
repo::{AccountStatus, RepoInfo, RepoPrev, RepoState},
32
35
};
33
36
34
37
type Result<T> = std::result::Result<T, ResyncError>;
···
59
62
pub collections: Vec<Nsid<'static>>,
60
63
/// Revision TID of the latest commit.
61
64
pub rev: Tid,
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>>,
65
+
/// MST root CID
66
+
pub data: RawCid,
68
67
}
69
68
70
69
/// Why fetching a repository's collection list failed.
···
126
125
let resolved = resolver.resolve(&did).await?;
127
126
let base = &*resolved.pds;
128
127
129
-
let snapshot = match fetch_collections(
128
+
let repo_snapshot = match fetch_collections(
130
129
client,
131
130
base,
132
131
did.clone(),
···
161
160
};
162
161
163
162
let db = db.clone();
164
-
let collections = snapshot.collections;
163
+
let collections = repo_snapshot.collections;
164
+
let rev = repo_snapshot.rev;
165
+
let prev_data = repo_snapshot.data.to_bytes();
165
166
let (n_inserted, n_removed) = tokio::task::spawn_blocking(move || {
167
+
crate::storage::repo::put_prev(&db, &did, &RepoPrev { rev, prev_data })?;
166
168
crate::storage::collection_index::sync_collections(&db, &did, &collections)
167
169
})
168
170
.await
···
254
256
.await;
255
257
256
258
match describe_result {
257
-
Ok(Ok(snapshot)) if !snapshot.collections.is_empty() => return Ok(snapshot),
259
+
Ok(Ok(snapshot)) if !snapshot.collections.is_empty() => {
260
+
metrics::counter!("lightrail_resync_fetch_total", "source" => "describe_repo")
261
+
.increment(1);
262
+
return Ok(snapshot);
263
+
}
258
264
// Empty list: PDS may have a bug or not paginate large collection sets.
259
265
// Fall through to the full CAR walk.
260
266
Ok(Ok(_)) => {}
···
269
275
Ok(Err(_)) | Err(_) => {}
270
276
}
271
277
272
-
tokio::time::timeout(
278
+
let result = tokio::time::timeout(
273
279
get_repo_timeout,
274
280
get_repo::fetch_collections(client, base, did),
275
281
)
···
278
284
Err(GetCollectionsError::Request(
279
285
"getRepo timed out".to_string(),
280
286
))
281
-
})
287
+
});
288
+
if result.is_ok() {
289
+
metrics::counter!("lightrail_resync_fetch_total", "source" => "get_repo").increment(1);
290
+
}
291
+
result
282
292
}