lightweight
com.atproto.sync.listReposByCollection
rework resync queue
+80
-51
+50
-31
src/storage/keys.rs
+50
-31
src/storage/keys.rs
···
148
148
key
149
149
}
150
150
151
-
/// `"repoResyncQueue"\0<ts_be:u64>\0<did>` — timestamp-ordered resync queue.
151
+
/// `"repoResyncQueue"\0<ts_be:u64>\0<did>`: timestamp-ordered resync queue.
152
152
///
153
153
/// Big-endian timestamp gives natural chronological ordering.
154
154
pub fn resync_queue(ts_be: u64, did: Did<'_>) -> Vec<u8> {
···
162
162
key
163
163
}
164
164
165
-
/// `"repoResyncQueue"\0` — prefix for scanning the entire resync queue.
165
+
/// `"repoResyncQueue"\0`: prefix for scanning the entire resync queue.
166
166
pub fn resync_queue_prefix_all() -> Vec<u8> {
167
167
let mut key = Vec::with_capacity(PREFIX_RESYNC_QUEUE.len() + 1);
168
168
key.extend_from_slice(PREFIX_RESYNC_QUEUE);
···
170
170
key
171
171
}
172
172
173
-
/// `"repoResyncQueue"\0<ts_be:u64>\0` — prefix for scanning resync queue up to a timestamp.
174
-
pub fn resync_queue_prefix(ts_be: u64) -> Vec<u8> {
175
-
let mut key = Vec::with_capacity(PREFIX_RESYNC_QUEUE.len() + 1 + 8 + 1);
176
-
key.extend_from_slice(PREFIX_RESYNC_QUEUE);
177
-
key.push(NUL_SEP);
173
+
/// `<ts_be:u64>\0`: timestamp middle serialization for resync queue
174
+
///
175
+
/// can be directly concattenated after resync_queue_prefix_all()
176
+
pub fn resync_queue_ts_midfix(ts_be: u64) -> Vec<u8> {
177
+
let mut key = Vec::with_capacity(8 + 1);
178
178
key.extend_from_slice(&ts_be.to_be_bytes());
179
179
key.push(NUL_SEP);
180
180
key
···
184
184
///
185
185
/// Key layout: `"repoResyncQueue"\0<ts_be:u64>\0<did>`
186
186
/// Returns `None` if the key is structurally invalid.
187
-
pub fn resync_queue_parse(key: &[u8]) -> Option<(u64, Did<'static>)> {
188
-
// TODO: error on unparseable!
187
+
pub fn resync_queue_parse(key: &[u8]) -> StorageResult<(u64, Did<'static>)> {
188
+
let key_str = String::from_utf8_lossy(key);
189
189
let rest = key
190
-
.strip_prefix(PREFIX_RESYNC_QUEUE)?
191
-
.strip_prefix(&[NUL_SEP])?;
190
+
.strip_prefix(PREFIX_RESYNC_QUEUE)
191
+
.ok_or(StorageError::Corrupt {
192
+
key: key_str.to_string(),
193
+
reason: "wrong prefix for resync queue",
194
+
})?
195
+
.strip_prefix(&[NUL_SEP])
196
+
.ok_or(StorageError::Corrupt {
197
+
key: key_str.to_string(),
198
+
reason: "wrong prefix for resync queue",
199
+
})?;
192
200
if rest.len() < 9 {
193
201
// Need at least 8 bytes for timestamp + 1 separator byte.
194
-
return None;
202
+
return Err(StorageError::Corrupt {
203
+
key: key_str.to_string(),
204
+
reason: "not enough suffix bytes for resync queue",
205
+
});
195
206
}
196
-
let ts = u64::from_be_bytes(rest[..8].try_into().ok()?);
197
-
let rest = rest[8..].strip_prefix(&[NUL_SEP])?;
198
-
let did_str = std::str::from_utf8(rest).ok()?;
199
-
Did::new_owned(did_str).ok().map(|did| (ts, did))
207
+
let ts_bytes = rest[..8].try_into().map_err(|_| StorageError::Corrupt {
208
+
key: key_str.to_string(),
209
+
reason: "not enough suffix bytes for resync queue",
210
+
})?;
211
+
let ts = u64::from_be_bytes(ts_bytes);
212
+
let rest = rest[8..]
213
+
.strip_prefix(&[NUL_SEP])
214
+
.ok_or(StorageError::Corrupt {
215
+
key: key_str.to_string(),
216
+
reason: "missing NUL separator in suffix bytes for resync queue",
217
+
})?;
218
+
let did_str = std::str::from_utf8(rest).map_err(|_| StorageError::Corrupt {
219
+
key: key_str.to_string(),
220
+
reason: "invalid bytes for did string for resync queue",
221
+
})?;
222
+
let did = Did::new_owned(did_str).map_err(|_| StorageError::Corrupt {
223
+
key: key_str.to_string(),
224
+
reason: "invalid DID for resync queue",
225
+
})?;
226
+
Ok((ts, did))
200
227
}
201
228
202
229
#[cfg(test)]
···
317
344
}
318
345
319
346
#[test]
320
-
fn resync_queue_prefix_matches_full_key() {
321
-
let prefix = resync_queue_prefix(42);
322
-
let key = resync_queue(42, did("did:web:example.com"));
323
-
assert!(key.starts_with(&prefix));
324
-
}
325
-
326
-
#[test]
327
-
fn resync_queue_prefix_does_not_match_different_timestamp() {
328
-
let prefix = resync_queue_prefix(42);
329
-
let key = resync_queue(43, did("did:web:example.com"));
330
-
assert!(!key.starts_with(&prefix));
331
-
}
332
-
333
-
#[test]
334
347
fn resync_queue_sorts_by_timestamp() {
335
348
let earlier = resync_queue(100, did("did:web:example.com"));
336
349
let later = resync_queue(200, did("did:web:example.com"));
···
356
369
357
370
#[test]
358
371
fn resync_queue_parse_returns_none_for_truncated_key() {
359
-
assert!(resync_queue_parse(b"repoResyncQueue\0").is_none());
372
+
assert_eq!(
373
+
resync_queue_parse(b"repoResyncQueue\0"),
374
+
Err(StorageError::Corrupt {
375
+
key: "repoResyncQueue\0".to_string(),
376
+
reason: "not enough suffix bytes for resync queue",
377
+
})
378
+
);
360
379
}
361
380
}
+30
-20
src/storage/resync.rs
+30
-20
src/storage/resync.rs
···
3
3
//! Keys: `"repoResyncQueue"\0<ts_be:u64>\0<did>`
4
4
//! Values: `[u16 BE retry_count][u16 BE reason_len][reason_bytes][commit_cbor_bytes]`
5
5
6
+
use fjall::util::prefixed_range;
6
7
use jacquard_common::types::string::Did;
7
8
8
9
use crate::storage::{
···
79
80
/// note: deleted accounts aren't removed from the resync queue so we need to
80
81
/// check that (or does the caller deal with it?)
81
82
///
82
-
/// TODO: we actually want to pass in an optional cursor so we can efficiently
83
-
/// skip over tombstones. we don't have to persist the cursor to disk, but the
84
-
/// caller can hold it in memory over the app's lifetime so we only pay the tomb
85
-
/// scan cost once on startup.
86
-
pub fn dequeue_ready(db: &DbRef, now: u64) -> StorageResult<Option<ResyncItem>> {
83
+
/// `since`: we actually want to pass in a cursor so we can efficiently skip
84
+
/// over tombstones. we don't have to persist the cursor to disk, but the caller
85
+
/// can hold it in memory over the app's lifetime so we only pay the tomb scan
86
+
/// cost once on startup.
87
+
pub fn dequeue_ready(
88
+
db: &DbRef,
89
+
now: u64,
90
+
since: Option<Vec<u8>>,
91
+
) -> StorageResult<Option<(ResyncItem, Vec<u8>)>> {
87
92
let prefix = keys::resync_queue_prefix_all();
88
-
for guard in db.ks.prefix(&prefix) {
89
-
let (key_slice, val_slice) = guard.into_inner()?;
90
-
let key_bytes: &[u8] = key_slice.as_ref();
91
-
let Some((ts, did)) = keys::resync_queue_parse(key_bytes) else {
92
-
continue;
93
-
};
94
-
if ts > now {
95
-
break; // queue is ordered; no earlier entries remain
96
-
}
97
-
let key_str = String::from_utf8_lossy(key_bytes).into_owned();
98
-
let item = decode(val_slice.as_ref(), &key_str, did)?;
99
-
db.ks.remove(key_bytes)?;
100
-
return Ok(Some(item));
101
-
}
102
-
Ok(None)
93
+
94
+
let lower_suffix = since.unwrap_or(vec![]);
95
+
let upper_suffix = keys::resync_queue_ts_midfix(now);
96
+
97
+
let Some(guard) = db
98
+
.ks
99
+
.range(prefixed_range(prefix, lower_suffix..upper_suffix))
100
+
.next()
101
+
else {
102
+
return Ok(None);
103
+
};
104
+
105
+
let (key_slice, val_slice) = guard.into_inner()?;
106
+
let key_bytes = key_slice.to_vec();
107
+
let (ts, did) = keys::resync_queue_parse(&key_bytes)?;
108
+
assert!(ts < now);
109
+
let key_str = String::from_utf8_lossy(&key_bytes).into_owned();
110
+
let item = decode(val_slice.as_ref(), &key_str, did)?;
111
+
db.ks.remove(&key_bytes)?;
112
+
Ok(Some((item, key_bytes)))
103
113
}