+23 -10

hacking.md

reviewed

··· 35 35 - [x] #commit and #sync 36 36 - [x] make sure blocking db calls are in `spawn_blocking`!! 37 37 - [x] db queries 38 38 - - [~] configuration 39 39 - - [~] copy applicable from tap 40 40 - - [ ] copy applicable from collectiondir 38 38 + - [x] implement `com.atproto.sync.listRepos` 41 39 - [x] sync1.1!!! 42 40 - [x] verify #commit event 43 41 - [x] verify #sync event 44 42 - [x] inductive proof for #commits 43 43 + - [~] configuration 44 44 + - [~] copy applicable from tap 45 45 + - [ ] copy applicable from collectiondir 45 46 - [~] actually firehose-index!! 46 47 - [x] extract collections-added/removed directly from CAR slice 47 48 - [ ] (spend some time on tests here) 48 49 - [x] do the thing (write them to the db) 49 50 - [ ] swap in repo-stream 51 51 + - [~] metrics 52 52 + - [x] basic metrics 53 53 + - [x] serve prom-style 54 54 + - [ ] copy applicable ones from collectiondir 55 55 + - [~] prefix-merge walker (limit by total collections to be merged?) 56 56 + - [x] add an all-collections index 57 57 + - [ ] swap in repo-stream for backfill 58 58 + - [ ] with memory limit 59 59 + - [ ] with a global concurrency limit for big repos 60 60 + - [ ] with disk spilling for huge repo 61 61 + - [ ] with queueing resync for large repos if resources are taken?? 62 62 + - [ ] (self-reminder: get_repo should be rare in lightrail) 50 63 - [ ] lenient sync1.1 51 64 - [ ] *don't* allow non-validating commits that look like sync1.1 52 65 - [ ] rachet by PDS host: be lenient if we have never seen a sync1.1-looking commit, always strict after we see one. ··· 57 70 - [ ] "deep crawl" mode for relays that listHosts -> listRepos on host instead of relying on relay listRepos 58 71 - [ ] special did:web behaviour to keep reusing a stale resolution on failure 59 72 - [ ] filter dids from inactive accounts 60 60 - - [ ] implement `com.atproto.sync.listRepos` 61 73 - [ ] multi-collection parallel walk/merge 62 62 - - [ ] add collection-list index 63 63 - - [ ] prefix-merge walker (limit by total collections to be merged?) 64 64 - - [ ] do we need 65 65 - - [ ] metrics 66 66 - - [ ] copy applicable ones from collectiondir 67 67 - - [ ] serve prom-style 68 74 - [ ] admin view of backfill state etc 69 75 - [ ] vanity stats for optimizations, like how many in-flight repos were saved from resync due to high-water-mark firehose cursor persistence 70 76 - [ ] account status convergeance: if we receive commits from apparently-inactive accounts, should we check upstream status to make sure we're not stale? ··· 106 112 ## state/db models 107 113 108 114 taking [inspiration from tap](https://github.com/bluesky-social/indigo/blob/main/cmd/tap/models/models.go) here! 115 115 + 116 116 + TODO: fix outdated prefixes here 109 117 110 118 ``` 111 119 main index: ··· 152 160 "repoPrev"||<did> => <rev:string>||<prevData:cid> 153 161 154 162 note: kept separate and small because it very frequently updates! 163 163 + 164 164 + 165 165 + all-collections list: 166 166 + 167 167 + "col"||<collection> => () 155 168 156 169 157 170 resync queue:

+196

src/storage/collection_list.rs

reviewed

··· 1 1 + //! Global collection list — every NSID ever seen across all repos. 2 2 + //! 3 3 + //! Key: `"col"<nsid>` — presence indicates the collection exists (or once existed). 4 4 + //! Value: empty. 5 5 + //! 6 6 + //! Entries are written on every collection birth and are never deleted, so the 7 7 + //! set can retain NSIDs after all repos have removed their last record. This is 8 8 + //! acceptable: the index is a superset of live collections, not an exact set. 9 9 + //! 10 10 + //! Writes are blind overwrites — no read-before-write. With only a few thousand 11 11 + //! collections in the network the key space is tiny and compaction handles the 12 12 + //! redundant writes cheaply. 13 13 + 14 14 + use jacquard_common::types::nsid::Nsid; 15 15 + 16 16 + use crate::storage::{DbRef, PREFIX_COLLECTION_LIST, StorageResult, error::StorageError}; 17 17 + 18 18 + // --------------------------------------------------------------------------- 19 19 + // Key encoding 20 20 + // --------------------------------------------------------------------------- 21 21 + 22 22 + fn key(collection: Nsid<'_>) -> Vec<u8> { 23 23 + let s = collection.as_str(); 24 24 + let mut k = Vec::with_capacity(PREFIX_COLLECTION_LIST.len() + s.len()); 25 25 + k.extend_from_slice(&PREFIX_COLLECTION_LIST); 26 26 + k.extend_from_slice(s.as_bytes()); 27 27 + k 28 28 + } 29 29 + 30 30 + fn parse_nsid(raw_key: &[u8]) -> StorageResult<Nsid<'static>> { 31 31 + let prefix_len = PREFIX_COLLECTION_LIST.len(); 32 32 + let key_str = String::from_utf8_lossy(raw_key).to_string(); 33 33 + let nsid_bytes = raw_key.get(prefix_len..).ok_or(StorageError::Corrupt { 34 34 + key: key_str.clone(), 35 35 + reason: "collection list key shorter than prefix", 36 36 + })?; 37 37 + let nsid_str = std::str::from_utf8(nsid_bytes).map_err(|_| StorageError::Corrupt { 38 38 + key: key_str.clone(), 39 39 + reason: "non-UTF-8 NSID in collection list key", 40 40 + })?; 41 41 + Nsid::new_owned(nsid_str).map_err(|_| StorageError::Corrupt { 42 42 + key: key_str, 43 43 + reason: "invalid NSID in collection list key", 44 44 + }) 45 45 + } 46 46 + 47 47 + // --------------------------------------------------------------------------- 48 48 + // Write 49 49 + // --------------------------------------------------------------------------- 50 50 + 51 51 + /// Add `collection` to the global list within an existing batch. 52 52 + /// 53 53 + /// Blindly overwrites any existing entry — no read needed. 54 54 + pub fn insert_into(batch: &mut fjall::OwnedWriteBatch, db: &DbRef, collection: Nsid<'_>) { 55 55 + batch.insert(&db.ks, key(collection), b""); 56 56 + } 57 57 + 58 58 + // --------------------------------------------------------------------------- 59 59 + // Read 60 60 + // --------------------------------------------------------------------------- 61 61 + 62 62 + /// Iterate over all known collections, starting at `cursor` (inclusive). 63 63 + /// 64 64 + /// Returns at most `limit` NSIDs sorted lexicographically. `next` is the first 65 65 + /// NSID of the following page, or `None` if this is the last page. 66 66 + pub fn scan( 67 67 + db: &DbRef, 68 68 + cursor: Option<Nsid<'_>>, 69 69 + limit: usize, 70 70 + ) -> StorageResult<(Vec<Nsid<'static>>, Option<Nsid<'static>>)> { 71 71 + let start_key: Vec<u8> = { 72 72 + let mut k = PREFIX_COLLECTION_LIST.to_vec(); 73 73 + if let Some(ref nsid) = cursor { 74 74 + k.extend_from_slice(nsid.as_str().as_bytes()); 75 75 + } 76 76 + k 77 77 + }; 78 78 + 79 79 + let mut ranger = db.ks.range(start_key..); 80 80 + let mut collections = Vec::with_capacity(limit); 81 81 + 82 82 + for guard in ranger.by_ref() { 83 83 + let (k, _v) = guard.into_inner()?; 84 84 + if !k.starts_with(&PREFIX_COLLECTION_LIST) { 85 85 + break; 86 86 + } 87 87 + collections.push(parse_nsid(&k)?); 88 88 + if collections.len() >= limit { 89 89 + break; 90 90 + } 91 91 + } 92 92 + 93 93 + let next = loop { 94 94 + let Some(guard) = ranger.next() else { 95 95 + break None; 96 96 + }; 97 97 + let key = guard.key()?; 98 98 + if !key.starts_with(&PREFIX_COLLECTION_LIST) { 99 99 + break None; 100 100 + } 101 101 + match parse_nsid(&key) { 102 102 + Ok(nsid) => break Some(nsid), 103 103 + Err(_) => continue, 104 104 + } 105 105 + }; 106 106 + 107 107 + Ok((collections, next)) 108 108 + } 109 109 + 110 110 + // --------------------------------------------------------------------------- 111 111 + // Tests 112 112 + // --------------------------------------------------------------------------- 113 113 + 114 114 + #[cfg(test)] 115 115 + mod tests { 116 116 + use super::*; 117 117 + use crate::storage::open_temporary; 118 118 + 119 119 + fn nsid(s: &str) -> Nsid<'static> { 120 120 + Nsid::new_owned(s.to_owned()).unwrap() 121 121 + } 122 122 + 123 123 + #[test] 124 124 + fn insert_and_scan_single() { 125 125 + let db = open_temporary().unwrap(); 126 126 + let mut batch = db.database.batch(); 127 127 + insert_into(&mut batch, &db, nsid("app.bsky.feed.post")); 128 128 + batch.commit().unwrap(); 129 129 + 130 130 + let (cols, next) = scan(&db, None, 100).unwrap(); 131 131 + assert_eq!(cols, vec![nsid("app.bsky.feed.post")]); 132 132 + assert!(next.is_none()); 133 133 + } 134 134 + 135 135 + #[test] 136 136 + fn scan_is_sorted() { 137 137 + let db = open_temporary().unwrap(); 138 138 + let mut batch = db.database.batch(); 139 139 + insert_into(&mut batch, &db, nsid("app.bsky.graph.follow")); 140 140 + insert_into(&mut batch, &db, nsid("app.bsky.actor.profile")); 141 141 + insert_into(&mut batch, &db, nsid("app.bsky.feed.post")); 142 142 + batch.commit().unwrap(); 143 143 + 144 144 + let (cols, _) = scan(&db, None, 100).unwrap(); 145 145 + assert_eq!( 146 146 + cols, 147 147 + vec![ 148 148 + nsid("app.bsky.actor.profile"), 149 149 + nsid("app.bsky.feed.post"), 150 150 + nsid("app.bsky.graph.follow"), 151 151 + ] 152 152 + ); 153 153 + } 154 154 + 155 155 + #[test] 156 156 + fn blind_overwrite_is_idempotent() { 157 157 + let db = open_temporary().unwrap(); 158 158 + for _ in 0..3 { 159 159 + let mut batch = db.database.batch(); 160 160 + insert_into(&mut batch, &db, nsid("app.bsky.feed.post")); 161 161 + batch.commit().unwrap(); 162 162 + } 163 163 + let (cols, _) = scan(&db, None, 100).unwrap(); 164 164 + assert_eq!(cols, vec![nsid("app.bsky.feed.post")]); 165 165 + } 166 166 + 167 167 + #[test] 168 168 + fn scan_pagination() { 169 169 + let db = open_temporary().unwrap(); 170 170 + let mut batch = db.database.batch(); 171 171 + for col in &[ 172 172 + "app.bsky.actor.profile", 173 173 + "app.bsky.feed.like", 174 174 + "app.bsky.feed.post", 175 175 + "app.bsky.graph.follow", 176 176 + ] { 177 177 + insert_into(&mut batch, &db, nsid(col)); 178 178 + } 179 179 + batch.commit().unwrap(); 180 180 + 181 181 + let (page1, next) = scan(&db, None, 2).unwrap(); 182 182 + assert_eq!( 183 183 + page1, 184 184 + vec![nsid("app.bsky.actor.profile"), nsid("app.bsky.feed.like")] 185 185 + ); 186 186 + let cursor = next.unwrap(); 187 187 + assert_eq!(cursor, nsid("app.bsky.feed.post")); 188 188 + 189 189 + let (page2, next2) = scan(&db, Some(cursor), 2).unwrap(); 190 190 + assert_eq!( 191 191 + page2, 192 192 + vec![nsid("app.bsky.feed.post"), nsid("app.bsky.graph.follow")] 193 193 + ); 194 194 + assert!(next2.is_none()); 195 195 + } 196 196 + }

+3

src/storage/mod.rs

reviewed

··· 1 1 pub mod backfill_progress; 2 2 pub mod collection_index; 3 3 + pub mod collection_list; 3 4 pub mod error; 4 5 pub mod firehose_cursor; 5 6 pub mod repo; ··· 19 20 /// Fixed-length (3 byte) key prefix per data type 20 21 type KeyPrefix = [u8; 3]; 21 22 23 23 + /// Global collection list (collection → ()). See [`collection_list`]. 24 24 + pub(crate) const PREFIX_COLLECTION_LIST: KeyPrefix = *b"col"; 22 25 /// Main collection index (collection → did). See [`collection_index`]. 23 26 pub(crate) const PREFIX_RBC: KeyPrefix = *b"rbc"; 24 27 /// Reversed collection index (did → collection). See [`collection_index`].

+3

src/storage/repo.rs

reviewed

··· 346 346 Vec<(Did<'static>, RepoInfo, Option<RepoPrev>)>, 347 347 Option<Did<'static>>, 348 348 )> { 349 349 + // TODO: use fjall prefix_range 350 350 + // TODO: probably snapshot so we get a consistent account view? 351 351 + 349 352 let prefix_len = PREFIX_REPO.len(); 350 353 351 354 let start_key: Vec<u8> = {

+4 -2

src/sync/firehose/commit_event.rs

reviewed

··· 494 494 } 495 495 496 496 // All checks passed — atomically update the chain tip and the collection 497 497 - // index (born → insert, died → remove). 497 497 + // index (born → insert, died → remove). Also record each born collection 498 498 + // in the global collection list (blind overwrite, never deleted). 498 499 let mut batch = db.database.batch(); 499 500 storage::repo::put_prev_into( 500 501 &mut batch, ··· 506 507 }, 507 508 ); 508 509 for coll in born { 509 509 - storage::collection_index::insert_into(&mut batch, db, did.clone(), coll); 510 510 + storage::collection_index::insert_into(&mut batch, db, did.clone(), coll.clone()); 511 511 + storage::collection_list::insert_into(&mut batch, db, coll); 510 512 } 511 513 for coll in died { 512 514 storage::collection_index::remove_into(&mut batch, db, did.clone(), coll);