Something went wrong. Try again.
lightweight `com.atproto.sync.listReposByCollection`
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259//! `#sync` firehose event processing.//!//! A `#sync` event means the PDS has declared a repo discontinuity (e.g. a//! repo migration, a rebase, or a bulk import).//!//! ## Validation steps//!//! 1. Resolve the DID; drop if resolution fails.//! 2. Check account status; drop if not active.//! 3. Verify the commit signature; on failure, evict the cached identity and//! retry once with a fresh resolution (only if the key actually changed).//! 4. Verify that the event's `repo` DID and `rev` match the deserialized//! commit object found in the CAR.//! 6. Drop if `rev` is not strictly newer than `repo_prev.rev`.//! 7. Drop if `rev` timestamp is implausibly far in the future.//!//! On success the repo is marked Desynchronized and enqueued for a full//! resync (the CAR blocks are passed along for the resync worker to use).
use jacquard_api::com_atproto::sync::subscribe_repos::Sync;use jacquard_common::types::{string::Did, tid::Tid};use tracing::{debug, info};
use super::validate::{self, CarDrop};use crate::identity::Resolver;use crate::storage::{ self, DbRef, pds_host::{self, Sync11Mode}, repo::{AccountStatus, RepoInfo, RepoState}, resync_queue::ResyncItem,};use jacquard_common::url::Host;
// ---------------------------------------------------------------------------// Public entry point// ---------------------------------------------------------------------------
/// Outcome of [`check_and_enqueue`].enum SyncCheckOutcome { /// Event handled (either processed or dropped for non-status reasons). Done, /// Account is locally inactive. Caller may check upstream before dropping. InactiveAccount,}
pub(super) async fn process_sync_event( sync: Box<Sync<'static>>, resolver: &Resolver, db: &DbRef, client: &crate::http::ThrottledClient,) -> crate::error::Result<()> { let did = sync.did.clone();
// ── Step 1: Resolve DID ────────────────────────────────────────────────── let Some(resolved) = validate::resolve(&did, resolver, "sync").await else { return Ok(()); };
// ── Steps 3–4: CAR parse + signature verification + field consistency ──── let (rev, mst_root) = match validate_car(&sync, &resolved.pubkey).await { Ok(v) => v, Err(CarDrop::InvalidSignature) => { let Some(fresh) = validate::fresh_key_after_sig_failure(&did, resolver, &resolved, "sync").await else { return Ok(()); }; match validate_car(&sync, &fresh.pubkey).await { Ok(v) => v, Err(e) => { debug!(did = %did, reason = %e, "sync dropped: still invalid with fresh key"); return Ok(()); } } } Err(e) => { debug!(did = %did, reason = %e, "sync dropped"); return Ok(()); } };
let pds_host: Option<Host> = resolved.pds.host().map(|h| h.to_owned()); let blocks = sync.blocks.to_vec(); let now = std::time::SystemTime::now();
// ── Steps 2, 6, 7: Blocking storage checks + mark desync + enqueue ─────── let outcome = { let db = db.clone(); let did = did.clone(); let rev = rev.clone(); let mst_root = mst_root.clone(); let blocks = blocks.clone(); let pds_host = pds_host.clone(); tokio::task::spawn_blocking(move || { check_and_enqueue(&db, did, rev, mst_root, blocks, pds_host, now) }) .await?? };
if let SyncCheckOutcome::InactiveAccount = outcome { // Our local record says inactive, but we may have missed a // reactivation #account event. Check upstream before dropping. if !validate::upstream_says_active(&pds_host, &did, client).await { return Ok(()); } // Upstream says active — persist the reactivation and re-run // check_and_enqueue with the updated status. metrics::counter!("lightrail_account_reactivated_total", "trigger" => "sync") .increment(1); info!(did = %did, "account reactivated via upstream check (triggered by #sync)"); let db_owned = db.clone(); tokio::task::spawn_blocking(move || { let Some((mut repo_info, _)) = storage::repo::get(&db_owned, &did)? else { return Ok(()); }; repo_info.status = AccountStatus::Active; let mut batch = db_owned.database.batch(); storage::repo::put_info_into(&mut batch, &db_owned, &did, &repo_info); batch .commit() .map_err(Into::<crate::storage::StorageError>::into)?; check_and_enqueue(&db_owned, did, rev, mst_root, blocks, pds_host, now).map(|_| ()) }) .await??; }
Ok(())}
// ---------------------------------------------------------------------------// CAR + signature validation (async, in-memory)// ---------------------------------------------------------------------------
/// Parse the CAR, deserialize the commit, verify the signature, and check/// that the event fields are internally consistent with the commit object.////// Unlike `#commit` events there is no explicit commit CID in the event/// envelope; we use the CAR header's root CID to locate the commit block.////// Returns `(rev, mst_root_bytes)` where `mst_root_bytes` is the raw CID bytes/// of the commit's MST root (`commit.data`), in the same format stored in/// [`RepoPrev::prev_data`].async fn validate_car( sync: &Sync<'static>, pubkey: &jacquard_common::types::crypto::PublicKey<'_>,) -> Result<(Tid, Vec<u8>), CarDrop> { let parsed = jacquard_repo::car::parse_car_bytes(sync.blocks.as_ref()) .await .map_err(|_| CarDrop::CarParse)?;
// For #sync events the CAR header root is the commit CID. let block = parsed .blocks .get(&parsed.root) .ok_or(CarDrop::CommitBlockMissing)?;
// Deserialize, verify signature, and check DID + rev consistency. let repo_commit = validate::deserialize_and_verify( block.as_ref(), pubkey, sync.did.clone(), sync.rev.clone().into(), )?;
Ok((repo_commit.rev, repo_commit.data.to_bytes()))}
// ---------------------------------------------------------------------------// Storage checks (blocking)// ---------------------------------------------------------------------------
/// Perform the storage-backed validation steps (2, 6, 7).////// Returns [`SyncCheckOutcome::InactiveAccount`] when the local record says/// the account is inactive, so the async caller can probe upstream and/// optionally retry. All other outcomes (processed or dropped) map to/// [`SyncCheckOutcome::Done`].fn check_and_enqueue( db: &DbRef, did: Did<'static>, rev: Tid, mst_root: Vec<u8>, blocks: Vec<u8>, pds_host: Option<Host>, now: std::time::SystemTime,) -> crate::error::Result<SyncCheckOutcome> { // A #sync event is unambiguously sync1.1 — upgrade the PDS mode eagerly, // before any drop checks, since the presence of a valid signed event is // enough to confirm the PDS supports sync1.1. if let Some(ref host) = pds_host { let mut pds_info = pds_host::get(db, host)?; if pds_info.sync11_mode == Sync11Mode::Lenient { pds_info.sync11_mode = Sync11Mode::Strict; pds_host::put(db, host, &pds_info)?; metrics::counter!("lightrail_pds_mode_upgraded_total", "trigger" => "sync") .increment(1); info!(pds = %host, "PDS upgraded to strict sync1.1 mode via #sync event"); } }
let (info, prev) = match storage::repo::get(db, &did)? { Some(r) => r, None => return Ok(SyncCheckOutcome::Done), // repo not indexed by us; drop silently };
// If the sync event's commit exactly matches our stored prev_data, the repo // state hasn't changed — this is a no-op re-announcement (e.g. a PDS // migration where the relay didn't miss any commits) and there is nothing // to resync. if let Some(prev) = &prev && rev.compare_to(&prev.rev) == 0 && mst_root == prev.prev_data { metrics::counter!("lightrail_resync_avoided_total", "reason" => "no_op").increment(1); debug!(did = %did, "sync dropped: commit state unchanged (no-op, likely post-migration)"); return Ok(SyncCheckOutcome::Done); }
// Separate inactive check so the caller can probe upstream before dropping. if !info.status.is_active() { return Ok(SyncCheckOutcome::InactiveAccount); }
// Steps 6, 7: shared drop checks (active status already verified above). if validate::should_drop(&info, prev.as_ref(), &rev, "sync", &did, now) { return Ok(SyncCheckOutcome::Done); }
// All checks passed — mark desync and enqueue resync. let mut batch = db.database.batch(); storage::repo::put_info_into( &mut batch, db, &did, &RepoInfo { state: RepoState::Desynchronized, status: info.status, error: None, }, ); storage::resync_queue::enqueue_into( &mut batch, db, now, &ResyncItem { did, retry_count: 0, retry_reason: "#sync".to_string(), commit_cbor: blocks, }, ); batch .commit() .map_err(Into::<crate::storage::StorageError>::into)?; Ok(SyncCheckOutcome::Done)}