forked from
tranquil.farm/tranquil-pds
Our Personal Data Server from scratch!
refactor(sync): extract MST parsing and serialization utilities
authored by
did:plc:mb5to35neicxt4gemstoro…
and committed by
tangled.org
f07bcb82
e38343ce
+307
-279
+38
-60
crates/tranquil-pds/src/sync/import.rs
+38
-60
crates/tranquil-pds/src/sync/import.rs
···
192
192
prev_key: &[u8],
193
193
records: &mut Vec<ImportedRecord>,
194
194
) -> Result<(), ImportError> {
195
+
use super::mst::{entries, left_child, parse_mst_entry, reconstruct_key};
196
+
195
197
let block = blocks
196
198
.get(cid)
197
199
.ok_or_else(|| ImportError::BlockNotFound(cid.to_string()))?;
198
-
let value: Ipld = serde_ipld_dagcbor::from_slice(block)
200
+
let node: Ipld = serde_ipld_dagcbor::from_slice(block)
199
201
.map_err(|e| ImportError::InvalidCbor(e.to_string()))?;
200
202
201
-
if let Ipld::Map(ref obj) = value {
202
-
if let Some(Ipld::Link(left_cid)) = obj.get("l") {
203
-
walk_mst_node(blocks, left_cid, prev_key, records)?;
204
-
}
203
+
if let Some(left_cid) = left_child(&node) {
204
+
walk_mst_node(blocks, &left_cid, prev_key, records)?;
205
+
}
205
206
206
-
let mut current_key = prev_key.to_vec();
207
+
let mut current_key = prev_key.to_vec();
207
208
208
-
if let Some(Ipld::List(entries)) = obj.get("e") {
209
-
for entry in entries {
210
-
if let Ipld::Map(entry_obj) = entry {
211
-
let prefix_len = entry_obj
212
-
.get("p")
213
-
.and_then(|p| match p {
214
-
Ipld::Integer(n) => usize::try_from(*n).ok(),
215
-
_ => None,
216
-
})
217
-
.unwrap_or(0);
209
+
if let Some(entry_list) = entries(&node) {
210
+
entry_list
211
+
.iter()
212
+
.filter_map(parse_mst_entry)
213
+
.try_for_each(|entry| {
214
+
if let Some(ref suffix) = entry.key_suffix {
215
+
reconstruct_key(&mut current_key, entry.prefix_len, suffix);
216
+
}
218
217
219
-
let key_suffix = entry_obj.get("k").and_then(|k| {
220
-
if let Ipld::Bytes(b) = k {
221
-
Some(b.clone())
222
-
} else {
223
-
None
224
-
}
225
-
});
226
-
227
-
if let Some(suffix) = key_suffix {
228
-
current_key.truncate(prefix_len);
229
-
current_key.extend_from_slice(&suffix);
230
-
}
231
-
232
-
if let Some(Ipld::Link(tree_cid)) = entry_obj.get("t") {
233
-
walk_mst_node(blocks, tree_cid, ¤t_key, records)?;
234
-
}
235
-
236
-
let record_cid = entry_obj.get("v").and_then(|v| {
237
-
if let Ipld::Link(cid) = v {
238
-
Some(*cid)
239
-
} else {
240
-
None
241
-
}
242
-
});
218
+
if let Some(tree_cid) = entry.subtree {
219
+
walk_mst_node(blocks, &tree_cid, ¤t_key, records)?;
220
+
}
243
221
244
-
if let Some(record_cid) = record_cid
245
-
&& let Ok(full_key) = String::from_utf8(current_key.clone())
246
-
&& let Some(record_block) = blocks.get(&record_cid)
247
-
&& let Ok(record_value) =
248
-
serde_ipld_dagcbor::from_slice::<Ipld>(record_block)
249
-
{
250
-
let blob_refs = find_blob_refs_ipld(&record_value, 0);
251
-
let parts: Vec<&str> = full_key.split('/').collect();
252
-
if parts.len() >= 2 {
253
-
let collection = parts[..parts.len() - 1].join("/");
254
-
let rkey = parts[parts.len() - 1].to_string();
255
-
records.push(ImportedRecord {
256
-
collection,
257
-
rkey,
258
-
cid: record_cid,
259
-
blob_refs,
260
-
});
261
-
}
222
+
if let Some(record_cid) = entry.value
223
+
&& let Ok(full_key) = String::from_utf8(current_key.clone())
224
+
&& let Some(record_block) = blocks.get(&record_cid)
225
+
&& let Ok(record_value) =
226
+
serde_ipld_dagcbor::from_slice::<Ipld>(record_block)
227
+
{
228
+
let blob_refs = find_blob_refs_ipld(&record_value, 0);
229
+
let parts: Vec<&str> = full_key.split('/').collect();
230
+
if parts.len() >= 2 {
231
+
let collection = parts[..parts.len() - 1].join("/");
232
+
let rkey = parts[parts.len() - 1].to_string();
233
+
records.push(ImportedRecord {
234
+
collection,
235
+
rkey,
236
+
cid: record_cid,
237
+
blob_refs,
238
+
});
262
239
}
263
240
}
264
-
}
265
-
}
241
+
242
+
Ok::<_, ImportError>(())
243
+
})?;
266
244
}
267
245
Ok(())
268
246
}
+1
crates/tranquil-pds/src/sync/mod.rs
+1
crates/tranquil-pds/src/sync/mod.rs
+67
crates/tranquil-pds/src/sync/mst.rs
+67
crates/tranquil-pds/src/sync/mst.rs
···
1
+
use cid::Cid;
2
+
use ipld_core::ipld::Ipld;
3
+
4
+
pub struct MstEntry {
5
+
pub prefix_len: usize,
6
+
pub key_suffix: Option<Vec<u8>>,
7
+
pub subtree: Option<Cid>,
8
+
pub value: Option<Cid>,
9
+
}
10
+
11
+
pub fn parse_mst_entry(entry: &Ipld) -> Option<MstEntry> {
12
+
let obj = match entry {
13
+
Ipld::Map(m) => m,
14
+
_ => return None,
15
+
};
16
+
let prefix_len = obj
17
+
.get("p")
18
+
.and_then(|p| match p {
19
+
Ipld::Integer(n) => usize::try_from(*n).ok(),
20
+
_ => None,
21
+
})
22
+
.unwrap_or(0);
23
+
let key_suffix = obj.get("k").and_then(|k| match k {
24
+
Ipld::Bytes(b) => Some(b.clone()),
25
+
Ipld::String(s) => Some(s.as_bytes().to_vec()),
26
+
_ => None,
27
+
});
28
+
let subtree = obj.get("t").and_then(|t| match t {
29
+
Ipld::Link(cid) => Some(*cid),
30
+
_ => None,
31
+
});
32
+
let value = obj.get("v").and_then(|v| match v {
33
+
Ipld::Link(cid) => Some(*cid),
34
+
_ => None,
35
+
});
36
+
Some(MstEntry {
37
+
prefix_len,
38
+
key_suffix,
39
+
subtree,
40
+
value,
41
+
})
42
+
}
43
+
44
+
pub fn left_child(node: &Ipld) -> Option<Cid> {
45
+
match node {
46
+
Ipld::Map(obj) => match obj.get("l") {
47
+
Some(Ipld::Link(cid)) => Some(*cid),
48
+
_ => None,
49
+
},
50
+
_ => None,
51
+
}
52
+
}
53
+
54
+
pub fn entries(node: &Ipld) -> Option<&Vec<Ipld>> {
55
+
match node {
56
+
Ipld::Map(obj) => match obj.get("e") {
57
+
Some(Ipld::List(entries)) => Some(entries),
58
+
_ => None,
59
+
},
60
+
_ => None,
61
+
}
62
+
}
63
+
64
+
pub fn reconstruct_key(prev_key: &mut Vec<u8>, prefix_len: usize, suffix: &[u8]) {
65
+
prev_key.truncate(prefix_len);
66
+
prev_key.extend_from_slice(suffix);
67
+
}
+155
-165
crates/tranquil-pds/src/sync/util.rs
+155
-165
crates/tranquil-pds/src/sync/util.rs
···
205
205
dt.format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string()
206
206
}
207
207
208
-
fn format_identity_event(event: &SequencedEvent) -> Result<Vec<u8>, SyncFrameError> {
209
-
let frame = IdentityFrame {
210
-
did: event.did.clone(),
211
-
handle: event.handle.as_ref().map(|h| h.to_string()),
212
-
seq: event.seq.as_i64(),
213
-
time: format_atproto_time(event.created_at),
214
-
};
215
-
let header = FrameHeader {
216
-
op: 1,
217
-
t: FrameType::Identity,
218
-
};
219
-
let mut bytes = Vec::with_capacity(256);
220
-
serde_ipld_dagcbor::to_writer(&mut bytes, &header)?;
221
-
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
208
+
fn serialize_cbor_pair<H: serde::Serialize, P: serde::Serialize>(
209
+
header: &H,
210
+
payload: &P,
211
+
capacity: usize,
212
+
) -> Result<Vec<u8>, SyncFrameError> {
213
+
let mut bytes = Vec::with_capacity(capacity);
214
+
serde_ipld_dagcbor::to_writer(&mut bytes, header)?;
215
+
serde_ipld_dagcbor::to_writer(&mut bytes, payload)?;
222
216
Ok(bytes)
223
217
}
224
218
219
+
fn serialize_event_frame<P: serde::Serialize>(
220
+
frame_type: FrameType,
221
+
payload: &P,
222
+
capacity: usize,
223
+
) -> Result<Vec<u8>, SyncFrameError> {
224
+
serialize_cbor_pair(&FrameHeader { op: 1, t: frame_type }, payload, capacity)
225
+
}
226
+
227
+
fn format_identity_event(event: &SequencedEvent) -> Result<Vec<u8>, SyncFrameError> {
228
+
serialize_event_frame(
229
+
FrameType::Identity,
230
+
&IdentityFrame {
231
+
did: event.did.clone(),
232
+
handle: event.handle.as_ref().map(|h| h.to_string()),
233
+
seq: event.seq.as_i64(),
234
+
time: format_atproto_time(event.created_at),
235
+
},
236
+
256,
237
+
)
238
+
}
239
+
225
240
fn format_account_event(event: &SequencedEvent) -> Result<Vec<u8>, SyncFrameError> {
226
241
let frame = AccountFrame {
227
242
did: event.did.clone(),
···
230
245
seq: event.seq.as_i64(),
231
246
time: format_atproto_time(event.created_at),
232
247
};
233
-
let header = FrameHeader {
234
-
op: 1,
235
-
t: FrameType::Account,
236
-
};
237
-
let mut bytes = Vec::with_capacity(256);
238
-
serde_ipld_dagcbor::to_writer(&mut bytes, &header)?;
239
-
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
248
+
let bytes = serialize_event_frame(FrameType::Account, &frame, 256)?;
240
249
let hex_str: String = bytes.iter().map(|b| format!("{:02x}", b)).collect();
241
250
tracing::info!(
242
251
did = %frame.did,
···
269
278
extract_rev_from_commit_bytes(&commit_bytes).ok_or(SyncFrameError::RevExtraction)?
270
279
};
271
280
let car_bytes = write_car_blocks(commit_cid, Some(commit_bytes), BTreeMap::new()).await?;
272
-
let frame = SyncFrame {
273
-
did: event.did.clone(),
274
-
rev,
275
-
blocks: car_bytes,
276
-
seq: event.seq.as_i64(),
277
-
time: format_atproto_time(event.created_at),
278
-
};
279
-
let header = FrameHeader {
280
-
op: 1,
281
-
t: FrameType::Sync,
282
-
};
283
-
let mut bytes = Vec::with_capacity(512);
284
-
serde_ipld_dagcbor::to_writer(&mut bytes, &header)?;
285
-
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
286
-
Ok(bytes)
281
+
serialize_event_frame(
282
+
FrameType::Sync,
283
+
&SyncFrame {
284
+
did: event.did.clone(),
285
+
rev,
286
+
blocks: car_bytes,
287
+
seq: event.seq.as_i64(),
288
+
time: format_atproto_time(event.created_at),
289
+
},
290
+
512,
291
+
)
287
292
}
288
293
289
-
pub async fn format_event_for_sending(
290
-
state: &AppState,
291
-
event: SequencedEvent,
292
-
) -> Result<Vec<u8>, SyncFrameError> {
293
-
match event.event_type {
294
-
RepoEventType::Identity => return format_identity_event(&event),
295
-
RepoEventType::Account => return format_account_event(&event),
296
-
RepoEventType::Sync => return format_sync_event(state, &event).await,
297
-
RepoEventType::Commit => {}
298
-
}
294
+
struct CommitEventContext {
295
+
frame: CommitFrame,
296
+
commit_cid: Cid,
297
+
prev_cid: Option<Cid>,
298
+
block_cids: Vec<Cid>,
299
+
}
300
+
301
+
fn prepare_commit_event(event: SequencedEvent) -> Result<CommitEventContext, SyncFrameError> {
299
302
let block_cids_str = event.blocks_cids.clone().unwrap_or_default();
300
303
let prev_cid_link = event.prev_cid.clone();
301
304
let prev_data_cid_link = event.prev_data_cid.clone();
···
314
317
let prev_cid = prev_cid_link
315
318
.as_ref()
316
319
.and_then(|c| Cid::from_str(c.as_str()).ok());
317
-
let mut all_cids: Vec<Cid> = block_cids_str
320
+
let mut block_cids: Vec<Cid> = block_cids_str
318
321
.iter()
319
322
.filter_map(|s| Cid::from_str(s).ok())
320
323
.filter(|c| Some(*c) != prev_cid)
321
324
.collect();
322
-
if !all_cids.contains(&commit_cid) {
323
-
all_cids.push(commit_cid);
325
+
if !block_cids.contains(&commit_cid) {
326
+
block_cids.push(commit_cid);
324
327
}
325
-
if let Some(ref pc) = prev_cid
328
+
Ok(CommitEventContext {
329
+
frame,
330
+
commit_cid,
331
+
prev_cid,
332
+
block_cids,
333
+
})
334
+
}
335
+
336
+
fn partition_blocks(
337
+
block_cids: impl IntoIterator<Item = (Cid, Bytes)>,
338
+
commit_cid: Cid,
339
+
) -> (Option<Bytes>, BTreeMap<Cid, Bytes>) {
340
+
let (commit_data, other_blocks): (Vec<_>, Vec<_>) = block_cids
341
+
.into_iter()
342
+
.partition(|(cid, _)| *cid == commit_cid);
343
+
let commit_bytes = commit_data.into_iter().next().map(|(_, data)| data);
344
+
let other = other_blocks.into_iter().collect();
345
+
(commit_bytes, other)
346
+
}
347
+
348
+
async fn finalize_commit_frame(
349
+
mut frame: CommitFrame,
350
+
commit_cid: Cid,
351
+
commit_bytes: Option<Bytes>,
352
+
other_blocks: BTreeMap<Cid, Bytes>,
353
+
) -> Result<Vec<u8>, SyncFrameError> {
354
+
if let Some(ref cb) = commit_bytes
355
+
&& let Some(rev) = extract_rev_from_commit_bytes(cb)
356
+
{
357
+
frame.rev = rev;
358
+
}
359
+
frame.blocks = write_car_blocks(commit_cid, commit_bytes, other_blocks).await?;
360
+
let capacity = frame.blocks.len() + 512;
361
+
serialize_event_frame(FrameType::Commit, &frame, capacity)
362
+
}
363
+
364
+
pub async fn format_event_for_sending(
365
+
state: &AppState,
366
+
event: SequencedEvent,
367
+
) -> Result<Vec<u8>, SyncFrameError> {
368
+
match event.event_type {
369
+
RepoEventType::Identity => return format_identity_event(&event),
370
+
RepoEventType::Account => return format_account_event(&event),
371
+
RepoEventType::Sync => return format_sync_event(state, &event).await,
372
+
RepoEventType::Commit => {}
373
+
}
374
+
let ctx = prepare_commit_event(event)?;
375
+
let mut frame = ctx.frame;
376
+
if let Some(ref pc) = ctx.prev_cid
326
377
&& let Ok(Some(prev_bytes)) = state.block_store.get(pc).await
327
378
&& let Some(rev) = extract_rev_from_commit_bytes(&prev_bytes)
328
379
{
329
380
frame.since = Some(rev);
330
381
}
331
-
let car_bytes = if !all_cids.is_empty() {
332
-
let fetched = state.block_store.get_many(&all_cids).await?;
333
-
let (commit_data, other_blocks): (Vec<_>, Vec<_>) = all_cids
334
-
.iter()
335
-
.zip(fetched.iter())
336
-
.filter_map(|(cid, data_opt)| data_opt.as_ref().map(|data| (*cid, data.clone())))
337
-
.partition(|(cid, _)| *cid == commit_cid);
338
-
let commit_bytes = commit_data.into_iter().next().map(|(_, data)| data);
339
-
if let Some(ref cb) = commit_bytes
340
-
&& let Some(rev) = extract_rev_from_commit_bytes(cb)
341
-
{
342
-
frame.rev = rev;
343
-
}
344
-
let blocks: std::collections::BTreeMap<Cid, Bytes> = other_blocks.into_iter().collect();
345
-
write_car_blocks(commit_cid, commit_bytes, blocks).await?
346
-
} else {
347
-
Vec::new()
348
-
};
349
-
frame.blocks = car_bytes;
350
-
let header = FrameHeader {
351
-
op: 1,
352
-
t: FrameType::Commit,
353
-
};
354
-
let mut bytes = Vec::with_capacity(frame.blocks.len() + 512);
355
-
serde_ipld_dagcbor::to_writer(&mut bytes, &header)?;
356
-
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
357
-
Ok(bytes)
382
+
if ctx.block_cids.is_empty() {
383
+
frame.blocks = Vec::new();
384
+
let capacity = frame.blocks.len() + 512;
385
+
return serialize_event_frame(FrameType::Commit, &frame, capacity);
386
+
}
387
+
let fetched = state.block_store.get_many(&ctx.block_cids).await?;
388
+
let resolved = ctx
389
+
.block_cids
390
+
.iter()
391
+
.zip(fetched.iter())
392
+
.filter_map(|(cid, data_opt)| data_opt.as_ref().map(|data| (*cid, data.clone())));
393
+
let (commit_bytes, other_blocks) = partition_blocks(resolved, ctx.commit_cid);
394
+
finalize_commit_frame(frame, ctx.commit_cid, commit_bytes, other_blocks).await
358
395
}
359
396
360
397
pub async fn prefetch_blocks_for_events(
···
413
450
Some(commit_bytes.clone()),
414
451
BTreeMap::new(),
415
452
))?;
416
-
let frame = SyncFrame {
417
-
did: event.did.clone(),
418
-
rev,
419
-
blocks: car_bytes,
420
-
seq: event.seq.as_i64(),
421
-
time: format_atproto_time(event.created_at),
422
-
};
423
-
let header = FrameHeader {
424
-
op: 1,
425
-
t: FrameType::Sync,
426
-
};
427
-
let mut bytes = Vec::new();
428
-
serde_ipld_dagcbor::to_writer(&mut bytes, &header)?;
429
-
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
430
-
Ok(bytes)
453
+
serialize_event_frame(
454
+
FrameType::Sync,
455
+
&SyncFrame {
456
+
did: event.did.clone(),
457
+
rev,
458
+
blocks: car_bytes,
459
+
seq: event.seq.as_i64(),
460
+
time: format_atproto_time(event.created_at),
461
+
},
462
+
512,
463
+
)
431
464
}
432
465
433
466
pub async fn format_event_with_prefetched_blocks(
···
440
473
RepoEventType::Sync => return format_sync_event_with_prefetched(&event, prefetched),
441
474
RepoEventType::Commit => {}
442
475
}
443
-
let block_cids_str = event.blocks_cids.clone().unwrap_or_default();
444
-
let prev_cid_link = event.prev_cid.clone();
445
-
let prev_data_cid_link = event.prev_data_cid.clone();
446
-
let mut frame: CommitFrame =
447
-
event
448
-
.try_into()
449
-
.map_err(|e: crate::sync::frame::CommitFrameError| {
450
-
SyncFrameError::InvalidEvent(e.to_string())
451
-
})?;
452
-
if let Some(ref pdc) = prev_data_cid_link
453
-
&& let Ok(cid) = Cid::from_str(pdc.as_str())
454
-
{
455
-
frame.prev_data = Some(cid);
456
-
}
457
-
let commit_cid = frame.commit;
458
-
let prev_cid = prev_cid_link
459
-
.as_ref()
460
-
.and_then(|c| Cid::from_str(c.as_str()).ok());
461
-
let mut all_cids: Vec<Cid> = block_cids_str
462
-
.iter()
463
-
.filter_map(|s| Cid::from_str(s).ok())
464
-
.filter(|c| Some(*c) != prev_cid)
465
-
.collect();
466
-
if !all_cids.contains(&commit_cid) {
467
-
all_cids.push(commit_cid);
468
-
}
469
-
if let Some(commit_bytes) = prefetched.get(&commit_cid)
470
-
&& let Some(rev) = extract_rev_from_commit_bytes(commit_bytes)
471
-
{
472
-
frame.rev = rev;
473
-
}
474
-
if let Some(ref pc) = prev_cid
476
+
let ctx = prepare_commit_event(event)?;
477
+
let mut frame = ctx.frame;
478
+
if let Some(ref pc) = ctx.prev_cid
475
479
&& let Some(prev_bytes) = prefetched.get(pc)
476
480
&& let Some(rev) = extract_rev_from_commit_bytes(prev_bytes)
477
481
{
478
482
frame.since = Some(rev);
479
483
}
480
-
let car_bytes = if !all_cids.is_empty() {
481
-
let (commit_data, other_blocks): (Vec<_>, Vec<_>) = all_cids
482
-
.into_iter()
483
-
.filter_map(|cid| prefetched.get(&cid).map(|data| (cid, data.clone())))
484
-
.partition(|(cid, _)| *cid == commit_cid);
485
-
let commit_bytes_for_car = commit_data.into_iter().next().map(|(_, data)| data);
486
-
let blocks: BTreeMap<Cid, Bytes> = other_blocks.into_iter().collect();
487
-
write_car_blocks(commit_cid, commit_bytes_for_car, blocks).await?
488
-
} else {
489
-
Vec::new()
490
-
};
491
-
frame.blocks = car_bytes;
492
-
let header = FrameHeader {
493
-
op: 1,
494
-
t: FrameType::Commit,
495
-
};
496
-
let mut bytes = Vec::with_capacity(frame.blocks.len() + 512);
497
-
serde_ipld_dagcbor::to_writer(&mut bytes, &header)?;
498
-
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
499
-
Ok(bytes)
484
+
if ctx.block_cids.is_empty() {
485
+
frame.blocks = Vec::new();
486
+
let capacity = frame.blocks.len() + 512;
487
+
return serialize_event_frame(FrameType::Commit, &frame, capacity);
488
+
}
489
+
let resolved = ctx
490
+
.block_cids
491
+
.into_iter()
492
+
.filter_map(|cid| prefetched.get(&cid).map(|data| (cid, data.clone())));
493
+
let (commit_bytes, other_blocks) = partition_blocks(resolved, ctx.commit_cid);
494
+
finalize_commit_frame(frame, ctx.commit_cid, commit_bytes, other_blocks).await
500
495
}
501
496
502
497
pub fn format_info_frame(
503
498
name: InfoFrameName,
504
499
message: Option<&str>,
505
500
) -> Result<Vec<u8>, SyncFrameError> {
506
-
let header = FrameHeader {
507
-
op: 1,
508
-
t: FrameType::Info,
509
-
};
510
-
let frame = InfoFrame {
511
-
name,
512
-
message: message.map(String::from),
513
-
};
514
-
let mut bytes = Vec::with_capacity(128);
515
-
serde_ipld_dagcbor::to_writer(&mut bytes, &header)?;
516
-
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
517
-
Ok(bytes)
501
+
serialize_event_frame(
502
+
FrameType::Info,
503
+
&InfoFrame {
504
+
name,
505
+
message: message.map(String::from),
506
+
},
507
+
128,
508
+
)
518
509
}
519
510
520
511
pub fn format_error_frame(
521
512
error: ErrorFrameName,
522
513
message: Option<&str>,
523
514
) -> Result<Vec<u8>, SyncFrameError> {
524
-
let header = ErrorFrameHeader { op: -1 };
525
-
let frame = ErrorFrameBody {
526
-
error,
527
-
message: message.map(String::from),
528
-
};
529
-
let mut bytes = Vec::with_capacity(128);
530
-
serde_ipld_dagcbor::to_writer(&mut bytes, &header)?;
531
-
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
532
-
Ok(bytes)
515
+
serialize_cbor_pair(
516
+
&ErrorFrameHeader { op: -1 },
517
+
&ErrorFrameBody {
518
+
error,
519
+
message: message.map(String::from),
520
+
},
521
+
128,
522
+
)
533
523
}
+46
-54
crates/tranquil-pds/src/sync/verify.rs
+46
-54
crates/tranquil-pds/src/sync/verify.rs
···
205
205
data_cid: &Cid,
206
206
blocks: &HashMap<Cid, Bytes>,
207
207
) -> Result<(), VerifyError> {
208
+
use super::mst::{entries, left_child, parse_mst_entry};
208
209
use ipld_core::ipld::Ipld;
209
210
210
211
let mut stack = vec![*data_cid];
···
227
228
.ok_or_else(|| VerifyError::BlockNotFound(cid.to_string()))?;
228
229
let node: Ipld = serde_ipld_dagcbor::from_slice(block)
229
230
.map_err(|e| VerifyError::InvalidCbor(e.to_string()))?;
230
-
if let Ipld::Map(ref obj) = node {
231
-
if let Some(Ipld::Link(left_cid)) = obj.get("l") {
232
-
if !blocks.contains_key(left_cid) {
233
-
return Err(VerifyError::BlockNotFound(format!(
234
-
"MST left pointer {} not in CAR",
235
-
left_cid
236
-
)));
237
-
}
238
-
stack.push(*left_cid);
231
+
if let Some(left_cid) = left_child(&node) {
232
+
if !blocks.contains_key(&left_cid) {
233
+
return Err(VerifyError::BlockNotFound(format!(
234
+
"MST left pointer {} not in CAR",
235
+
left_cid
236
+
)));
239
237
}
240
-
if let Some(Ipld::List(entries)) = obj.get("e") {
241
-
let mut last_full_key: Vec<u8> = Vec::new();
242
-
for entry in entries {
243
-
if let Ipld::Map(entry_obj) = entry {
244
-
let prefix_len = entry_obj
245
-
.get("p")
246
-
.and_then(|p| match p {
247
-
Ipld::Integer(i) => usize::try_from(*i).ok(),
248
-
_ => None,
249
-
})
250
-
.unwrap_or(0);
251
-
let key_suffix = entry_obj.get("k").and_then(|k| match k {
252
-
Ipld::Bytes(b) => Some(b.clone()),
253
-
Ipld::String(s) => Some(s.as_bytes().to_vec()),
254
-
_ => None,
255
-
});
256
-
if let Some(suffix) = key_suffix {
257
-
let mut full_key = Vec::new();
258
-
if prefix_len > 0 && prefix_len <= last_full_key.len() {
259
-
full_key.extend_from_slice(&last_full_key[..prefix_len]);
260
-
}
261
-
full_key.extend_from_slice(&suffix);
262
-
if !last_full_key.is_empty() && full_key <= last_full_key {
263
-
return Err(VerifyError::MstValidationFailed(
264
-
"MST keys not in sorted order".to_string(),
265
-
));
266
-
}
267
-
last_full_key = full_key;
238
+
stack.push(left_cid);
239
+
}
240
+
if let Some(entry_list) = entries(&node) {
241
+
let mut last_full_key: Vec<u8> = Vec::new();
242
+
entry_list
243
+
.iter()
244
+
.filter_map(parse_mst_entry)
245
+
.try_for_each(|entry| {
246
+
if let Some(ref suffix) = entry.key_suffix {
247
+
let mut full_key = Vec::new();
248
+
if entry.prefix_len > 0
249
+
&& entry.prefix_len <= last_full_key.len()
250
+
{
251
+
full_key
252
+
.extend_from_slice(&last_full_key[..entry.prefix_len]);
268
253
}
269
-
if let Some(Ipld::Link(tree_cid)) = entry_obj.get("t") {
270
-
if !blocks.contains_key(tree_cid) {
271
-
return Err(VerifyError::BlockNotFound(format!(
272
-
"MST subtree {} not in CAR",
273
-
tree_cid
274
-
)));
275
-
}
276
-
stack.push(*tree_cid);
254
+
full_key.extend_from_slice(suffix);
255
+
if !last_full_key.is_empty() && full_key <= last_full_key {
256
+
return Err(VerifyError::MstValidationFailed(
257
+
"MST keys not in sorted order".to_string(),
258
+
));
277
259
}
278
-
if let Some(Ipld::Link(value_cid)) = entry_obj.get("v")
279
-
&& !blocks.contains_key(value_cid)
280
-
{
281
-
warn!(
282
-
"Record block {} referenced in MST not in CAR (may be expected for partial export)",
283
-
value_cid
284
-
);
260
+
last_full_key = full_key;
261
+
}
262
+
if let Some(tree_cid) = entry.subtree {
263
+
if !blocks.contains_key(&tree_cid) {
264
+
return Err(VerifyError::BlockNotFound(format!(
265
+
"MST subtree {} not in CAR",
266
+
tree_cid
267
+
)));
285
268
}
269
+
stack.push(tree_cid);
286
270
}
287
-
}
288
-
}
271
+
if let Some(value_cid) = entry.value
272
+
&& !blocks.contains_key(&value_cid)
273
+
{
274
+
warn!(
275
+
"Record block {} referenced in MST not in CAR (may be expected for partial export)",
276
+
value_cid
277
+
);
278
+
}
279
+
Ok(())
280
+
})?;
289
281
}
290
282
}
291
283
debug!(