lightweight
com.atproto.sync.listReposByCollection
return ok not err on inductive proof drop
+41
-98
+1
hacking.md
+1
hacking.md
···
117
117
- [x] no-events-received timeout reconnect
118
118
- [x] account status convergeance: if we receive commits from apparently-inactive accounts, should we check upstream status to make sure we're not stale?
119
119
- [x] resync short-circuit: tiny repos may actually return their entire CAR for getRecord
120
+
- [x] use jacquard's built-in inductive proof methods
120
121
- [~] repo-stream: drop record block contents with processor fn
121
122
- [x] in getRecord before describeRepo
122
123
- [ ] in commit handling
+40
-98
src/sync/firehose/commit_event.rs
+40
-98
src/sync/firehose/commit_event.rs
···
16
16
//! 9. If we have a `repo_prev` and `prevData` is present: drop and queue
17
17
//! resync if the bytes don't match `prev.prev_data`.
18
18
19
-
use std::sync::Arc;
20
-
21
-
use cid::Cid as IpldCid;
22
-
use jacquard_api::com_atproto::sync::subscribe_repos::{Commit, RepoOp};
23
-
use jacquard_common::types::cid::CidLink;
19
+
use jacquard_api::com_atproto::sync::subscribe_repos::Commit;
24
20
use jacquard_common::types::{string::Did, string::Nsid, tid::Tid};
25
-
use jacquard_repo::mst::VerifiedWriteOp;
26
-
use jacquard_repo::{MemoryBlockStore, Mst};
27
-
use tracing::{debug, error, info, warn};
21
+
use jacquard_repo::commit::firehose::{FirehoseCommit, RepoOp as FirehoseRepoOp};
22
+
use tracing::{debug, error, info, trace, warn};
28
23
29
24
use super::validate::{self, CarDrop};
30
25
use crate::identity::Resolver;
···
116
111
};
117
112
118
113
// ── Steps 3–4: CAR parse + signature verification + field consistency ────
119
-
let (new_mst_root_bytes, mst_root_cid, parsed) = match validate_car(&commit, &resolved.pubkey)
120
-
.await
121
-
{
114
+
let (new_mst_root_bytes, parsed) = match validate_car(&commit, &resolved.pubkey).await {
122
115
Ok(v) => v,
123
116
Err(CarDrop::InvalidSignature) => {
124
117
let Some(fresh) =
···
153
146
// Clone the parsed CAR before consuming its blocks into the MST block store.
154
147
// not super-cheap, but the Bytes values are refcounted at least so whatever
155
148
let parsed_clone = parsed.clone();
156
-
157
-
let storage = Arc::new(MemoryBlockStore::new_from_blocks(parsed.blocks));
158
-
let new_mst = Mst::load(Arc::clone(&storage), mst_root_cid, None);
159
149
160
150
// ── Step 5: Inductive proof ───────────────────────────────────────────────
161
-
let strict = pds_mode == Sync11Mode::Strict;
162
-
if !verify_inductive_proof(new_mst, &commit.ops, commit.prev_data.as_ref(), strict).await? {
151
+
if pds_mode == Sync11Mode::Strict
152
+
&& let Err(e) = to_firehose_commit(&commit)
153
+
.validate_v1_1(&resolved.pubkey)
154
+
.await
155
+
{
163
156
metrics::counter!("lightrail_commit_dropped_total", "reason" => "proof_failed")
164
157
.increment(1);
165
-
debug!(did = %did, "commit dropped: inductive proof failed");
158
+
debug!(did = %did, error = %e, "commit dropped: inductive proof failed");
166
159
return Ok(());
167
160
}
168
161
···
219
212
async fn validate_car(
220
213
commit: &Commit<'static>,
221
214
pubkey: &jacquard_common::types::crypto::PublicKey<'_>,
222
-
) -> Result<(Vec<u8>, IpldCid, jacquard_repo::car::ParsedCar), CarDrop> {
215
+
) -> Result<(Vec<u8>, jacquard_repo::car::ParsedCar), CarDrop> {
223
216
// Parse the CAR slice embedded in the firehose event.
224
217
let parsed = jacquard_repo::car::parse_car_bytes(commit.blocks.as_ref())
225
218
.await
···
254
247
// for block store construction) and the parsed CAR (the blocks themselves).
255
248
metrics::histogram!("lightrail_commit_car_bytes").record(commit.blocks.len() as f64);
256
249
let mst_root_cid = repo_commit.data;
257
-
Ok((mst_root_cid.to_bytes(), mst_root_cid, parsed))
250
+
Ok((mst_root_cid.to_bytes(), parsed))
258
251
}
259
252
260
253
// ---------------------------------------------------------------------------
261
254
// Step 5: Inductive proof (async, MST mutations)
262
255
// ---------------------------------------------------------------------------
263
256
264
-
/// Verify the sync1.1 inductive proof for a commit.
265
-
///
266
-
/// Starting from the new MST (`mst`, loaded from this commit's CAR blocks),
267
-
/// invert every op in the event's `ops` list in reverse order. Each inversion
268
-
/// validates the op's `prev` CID in-place via [`Mst::invert_op`]. After all
269
-
/// inversions the resulting MST root must equal `prev_data` (the previous MST
270
-
/// root recorded in the commit object).
271
-
///
272
-
/// Returns `Ok(true)` if the proof passes or is skipped (pre-sync1.1),
273
-
/// `Ok(false)` if it fails (caller should drop the commit), or an `Err` for
274
-
/// a storage-level failure.
275
-
async fn verify_inductive_proof(
276
-
mut mst: Mst<MemoryBlockStore>,
277
-
ops: &[RepoOp<'_>],
278
-
prev_data: Option<&CidLink<'_>>,
279
-
strict: bool,
280
-
) -> crate::error::Result<bool> {
281
-
// Without prevData we cannot check the final root.
282
-
let Some(prev_data_link) = prev_data else {
283
-
// Strict PDSes must always supply prevData.
284
-
return Ok(!strict);
285
-
};
286
-
287
-
let expected_prev_root = match prev_data_link.to_ipld() {
288
-
Ok(cid) => cid,
289
-
Err(_) => {
290
-
// Malformed prevData CID — strict PDSes should never send this.
291
-
return Ok(!strict);
292
-
}
293
-
};
294
-
295
-
// Invert each op in reverse order. The ops are in the CAR proof, so every
296
-
// key's path from root is available in the block store.
297
-
for op in ops.iter().rev() {
298
-
let Some(verified_op) = convert_op(op) else {
299
-
// An op is missing required CIDs — strict PDSes must always include them.
300
-
return Ok(!strict);
301
-
};
302
-
303
-
let ok = mst
304
-
.invert_op(verified_op)
305
-
.await
306
-
.map_err(|e| crate::error::Error::Other(e.to_string()))?;
307
-
if !ok {
308
-
return Ok(false);
309
-
}
310
-
}
311
-
312
-
// After all inversions the tree root must match prevData.
313
-
let actual_prev_root = mst
314
-
.get_pointer()
315
-
.await
316
-
.map_err(|e| crate::error::Error::Other(e.to_string()))?;
317
-
318
-
Ok(actual_prev_root == expected_prev_root)
319
-
}
320
-
321
-
/// Convert a firehose [`RepoOp`] into a [`VerifiedWriteOp`] for
322
-
/// [`Mst::invert_op`].
323
-
///
324
-
/// Returns `None` if a required CID field is absent (pre-sync1.1 commits may
325
-
/// omit `prev` on Update/Delete ops) or if the action is unrecognised.
326
-
fn convert_op(op: &RepoOp<'_>) -> Option<VerifiedWriteOp> {
327
-
let key = op.path.as_ref().into(); // &str → SmolStr
328
-
match op.action.as_ref() {
329
-
"create" => {
330
-
let cid = op.cid.as_ref()?.to_ipld().ok()?;
331
-
Some(VerifiedWriteOp::Create { key, cid })
332
-
}
333
-
"update" => {
334
-
let cid = op.cid.as_ref()?.to_ipld().ok()?;
335
-
let prev = op.prev.as_ref()?.to_ipld().ok()?;
336
-
Some(VerifiedWriteOp::Update { key, cid, prev })
337
-
}
338
-
"delete" => {
339
-
let prev = op.prev.as_ref()?.to_ipld().ok()?;
340
-
Some(VerifiedWriteOp::Delete { key, prev })
341
-
}
342
-
_ => None,
257
+
fn to_firehose_commit<'a>(commit: &Commit<'a>) -> FirehoseCommit<'a> {
258
+
FirehoseCommit {
259
+
repo: commit.repo.clone(),
260
+
rev: commit.rev.clone(),
261
+
seq: commit.seq,
262
+
since: commit.since.clone().unwrap_or_else(|| {
263
+
trace!("putting in a phony TID for FirehoseCommit (None on Commit)");
264
+
Tid::now_0()
265
+
}),
266
+
time: commit.time.clone(),
267
+
commit: commit.commit.clone(),
268
+
blocks: commit.blocks.clone(),
269
+
// HACK: reverse op order for inductive proof -- when upgrading jacquard this is fixed
270
+
ops: commit
271
+
.ops
272
+
.iter()
273
+
.rev()
274
+
.map(|op| FirehoseRepoOp {
275
+
action: op.action.clone(),
276
+
cid: op.cid.clone(),
277
+
path: op.path.clone(),
278
+
prev: op.prev.clone(),
279
+
})
280
+
.collect(),
281
+
prev_data: commit.prev_data.clone(),
282
+
blobs: commit.blobs.clone(),
283
+
too_big: commit.too_big,
284
+
rebase: commit.rebase,
343
285
}
344
286
}
345
287