lightweight
com.atproto.sync.listReposByCollection
remove some .ok() error suppressions
+45
-23
+1
-1
src/mst/mortality.rs
+1
-1
src/mst/mortality.rs
···
1114
1114
let fixture_dir = suite_dir.join("tests/diff/exhaustive");
1115
1115
let mut entries: Vec<_> = std::fs::read_dir(&fixture_dir)
1116
1116
.unwrap()
1117
-
.filter_map(|e| e.ok())
1117
+
.filter_map(|e| e.inspect_err(|e| eprintln!("error reading dir: {e}")).ok())
1118
1118
.filter(|e| e.path().extension().map_or(false, |x| x == "json"))
1119
1119
.collect();
1120
1120
entries.sort_by_key(|e| e.path());
+21
-14
src/storage/collection_index.rs
+21
-14
src/storage/collection_index.rs
···
91
91
}
92
92
93
93
/// Parse the collection NSID from a full cbr key given the prefix length.
94
-
fn cbr_parse_collection(key: &[u8], prefix_len: usize) -> Option<Nsid<'static>> {
95
-
// TODO: we should error on unparseable!
96
-
let nsid_str = std::str::from_utf8(key.get(prefix_len..)?).ok()?;
97
-
Nsid::new_owned(nsid_str).ok()
94
+
fn cbr_parse_collection(key: &[u8], prefix_len: usize) -> StorageResult<Nsid<'static>> {
95
+
let key_str = String::from_utf8_lossy(key);
96
+
let suffix = key.get(prefix_len..).ok_or(StorageError::Corrupt {
97
+
key: key_str.to_string(),
98
+
reason: "invalid prefix when parsing collection in cbr",
99
+
})?;
100
+
let nsid_str = std::str::from_utf8(suffix).map_err(|_| StorageError::Corrupt {
101
+
key: key_str.to_string(),
102
+
reason: "invalid string suffix for NSID in cbr suffix",
103
+
})?;
104
+
Nsid::new_owned(nsid_str).map_err(|_| StorageError::Corrupt {
105
+
key: key_str.to_string(),
106
+
reason: "invalid NSID in cbr suffix",
107
+
})
98
108
}
99
109
100
110
// ---------------------------------------------------------------------------
···
143
153
/// Iterate over collections in the cbr index for `did`, starting after `cursor`.
144
154
///
145
155
/// Returns at most `limit` NSIDs.
146
-
///
147
-
/// TODO: we can fjall range to the collection's next-after-max (might even be
148
-
/// exposed now?) or maybe use prefix + seek for the start?
149
156
pub fn scan_cbr(
150
157
db: &DbRef,
151
158
did: &Did<'_>,
···
176
183
{
177
184
continue;
178
185
}
179
-
if let Some(col) = cbr_parse_collection(&k, prefix_len) {
180
-
cols.push(col);
181
-
}
186
+
let col = cbr_parse_collection(&k, prefix_len)?;
187
+
cols.push(col);
182
188
}
183
189
Ok(cols)
184
190
}
···
325
331
/// rbc and cbr removes into `batch`. Use this when the index cleanup must be
326
332
/// atomic with other writes (e.g. a repo state update on tombstone).
327
333
/// Returns the number of collections removed.
334
+
///
335
+
/// TODO: also pass in a db snapshot??
328
336
pub fn remove_all_into(
329
337
batch: &mut fjall::OwnedWriteBatch,
330
338
db: &DbRef,
···
338
346
.collect::<fjall::Result<_>>()?;
339
347
let prefix_len = prefix.len();
340
348
for cbr_key in &collections {
341
-
if let Some(col) = cbr_parse_collection(cbr_key, prefix_len) {
342
-
batch.remove(&db.index_ks, rbc(col, did));
343
-
}
349
+
let col = cbr_parse_collection(cbr_key, prefix_len)?;
350
+
batch.remove(&db.index_ks, rbc(col, did));
344
351
batch.remove(&db.index_ks, cbr_key.as_slice());
345
352
}
346
353
Ok(collections.len())
···
543
550
let col = nsid("app.bsky.feed.post");
544
551
let key = cbr(&d, col.clone());
545
552
let prefix_len = cbr_prefix(&d).len();
546
-
assert_eq!(cbr_parse_collection(&key, prefix_len), Some(col));
553
+
assert_eq!(cbr_parse_collection(&key, prefix_len), Ok(col));
547
554
}
548
555
549
556
#[test]
+16
-4
src/storage/meta.rs
+16
-4
src/storage/meta.rs
···
12
12
use std::sync::{Arc, Mutex};
13
13
14
14
use cardinality_estimator_safe::{Element, Sketch};
15
-
use tracing::error;
15
+
use tracing::{error, trace};
16
16
17
17
use super::{DbRef, PREFIX_META, error::StorageResult};
18
18
···
137
137
}
138
138
139
139
fn read_sketch(ks: &fjall::Keyspace, suffix: &[u8]) -> Sk {
140
-
let Some(bytes) = ks.get(full_key(suffix)).ok().flatten() else {
140
+
let bytes = match ks.get(full_key(suffix)) {
141
+
Ok(b) => b,
142
+
Err(e) => {
143
+
error!(key = ?suffix, error = %e, "failed to read sketch from db, using fresh sketch");
144
+
return Sk::default();
145
+
}
146
+
};
147
+
let Some(bytes) = bytes else {
148
+
// normal just didn't exist yet (should only happen once per db)
149
+
trace!(key = ?suffix, "sketch not found in db, initializing new");
141
150
return Sk::default();
142
151
};
143
152
match postcard::from_bytes(bytes.as_ref()) {
144
-
Ok(sk) => sk,
153
+
Ok(sk) => {
154
+
trace!(key = ?suffix, "successfully got sketch from db");
155
+
sk
156
+
}
145
157
Err(e) => {
146
-
error!(key = ?suffix, error = %e, "failed to deserialize cardinality sketch; using fresh sketch");
158
+
error!(key = ?suffix, error = %e, "failed to deserialize sketch, using fresh sketch");
147
159
Sk::default()
148
160
}
149
161
}
+7
-4
src/sync/firehose/commit_event.rs
+7
-4
src/sync/firehose/commit_event.rs
···
170
170
let rev = commit.rev.clone();
171
171
let since = commit.since.clone();
172
172
// Extract the raw CID bytes from the event's prevData field (sync1.1).
173
-
let incoming_prev_data: Option<Vec<u8>> = commit
173
+
let incoming_prev_data = commit
174
174
.prev_data
175
-
.as_ref()
176
-
.and_then(|cl| cl.to_ipld().ok())
177
-
.map(|cid| cid.to_bytes());
175
+
.map(|cid| {
176
+
cid.to_ipld()
177
+
.map_err(|e| crate::error::Error::Other(format!("bad CID format: {e}")))
178
+
})
179
+
.transpose()?
180
+
.map(|ipld_cid| ipld_cid.to_bytes());
178
181
179
182
tokio::task::spawn_blocking(move || {
180
183
process_blocking(