[ingest] ignore a #sync that doesn't move the rev forward
64 unmodified lines65666768697071727374207 unmodified linesExpand all282283284285286287288289290291292293294204 unmodified linesExpand all49950050150250350450550650750850951051151251351451551651751851952052152252352452552652752810 unmodified lines64 unmodified lines SizeLimitExceeded, /// CAR could not be parsed #[error("malformed CAR: {0}")] MalformedCar(miette::Report), #[error("future rev")] FutureRev, /// wire message fields differ from the signed commit object #[error("field mismatch in {field}")] FieldMismatch { field: &'static str }, /// signature verification failed207 unmodified linesExpand all if msg.blocks.len() > MAX_BLOCKS_BYTES { return Err(SyncValidationError::SizeLimitExceeded); }
// a future rev would stick as the repo's rev and make honest commits after it look stale if Tid::new(msg.rev.as_ref()).is_ok_and(|rev| is_future_rev(&rev, opts.rev_clock_skew_secs)) { return Err(SyncValidationError::FutureRev); }
// 2. CAR parse, checking each block's cid as it is read let parsed = crate::car::parse_commit_car(msg.blocks.clone(), opts.verify_cids) .map_err(SyncValidationError::MalformedCar)?;
204 unmodified linesExpand all assert!(err.to_string().contains("CAR block CID mismatch"), "{err}"); }
#[test] fn a_sync_dated_past_the_clock_is_rejected_before_its_car_is_read() { let opts = ValidationOptions::from(&crate::config::Config::default()); let sync = |rev: &str| Sync { blocks: bytes::Bytes::new(), did: Did::new_static("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa").unwrap(), rev: rev.to_owned().into(), seq: 1, time: Datetime(chrono::Utc::now().fixed_offset()), };
assert!(matches!( validate_sync(&sync("jzzzzzzzzzzzz"), None, &opts), Err(SyncValidationError::FutureRev) )); // the rev is fine here, it's the empty car that fails assert!(matches!( validate_sync(&sync(Tid::now_0().as_str()), None, &opts), Err(SyncValidationError::MalformedCar(_)) )); }
#[test] fn record_dag_cbor_projection_accepts_data_and_rejects_invalid_bytes() { let encoded = serde_ipld_dagcbor::to_vec(&serde_json::json!({ "$type": "app.bsky.feed.post", "text": "stinkpot",10 unmodified lines71 unmodified lines72737475767778798081828384858687140 unmodified linesExpand all71 unmodified lines if !repo_state.active { return Ok(Handled::Done); }
// replaying an old #sync would make the indexer refetch every time if let Some(root) = &repo_state.root && sync.rev.as_str() <= root.rev.to_tid().as_str() { debug!(rev = %sync.rev, "sync at or below the tracked rev, ignoring"); return Ok(Handled::Done); }
repo_state.advance_message_time(sync.time.0.timestamp_millis());
let validated = match ctx.validate_sync(&mut repo_state, firehose, &sync)? { Checked::Done(Some(validated)) => validated,140 unmodified linesExpand all227 unmodified linesExpand all228229230231232232233234235236237238239240241242243244786 unmodified linesExpand all1031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094677 unmodified linesExpand all227 unmodified linesExpand all did: &Did, seq: i64, key: &k256::ecdsa::SigningKey,) -> SubscribeReposMessage<'static> { let rev = Tid::now_0(); sync_at(did, seq, key, Tid::now_0()).await}
pub(super) async fn sync_at( did: &Did, seq: i64, key: &k256::ecdsa::SigningKey, rev: Tid,) -> SubscribeReposMessage<'static> { let data = crate::car::compute_cid(b"data"); let commit = crate::mst::AtpCommit::sign_k256(did.clone(), data, rev.clone(), None, key); let body = commit.to_cbor().unwrap(); let cid = crate::car::compute_cid(&body);786 unmodified linesExpand all assert_eq!(account_time(&relay, &bob), Some(base)); Ok(())}
#[cfg(feature = "indexer")]#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn a_sync_that_does_not_move_the_rev_forward_is_ignored() -> Result<()> { let relay = Pipeline::start(Options::default()).await?; let alice = did("alice"); let key = signing_key(1); relay.put_row(&alice, &known_row(Some(&key), Duration::from_secs(60)))?; let older = Tid::now_0(); tokio::time::sleep(Duration::from_millis(5)).await; let newer = Tid::now_0();
relay .send(sync_at(&alice, 1, &key, newer.clone()).await) .await; relay .until(Duration::from_secs(5), |relay| relay.cursor() == 1) .await; assert_eq!(relay.forwarded.lock().len(), 1);
relay.send(sync_at(&alice, 2, &key, older).await).await; relay.send(sync_at(&alice, 3, &key, newer).await).await; relay .until(Duration::from_secs(5), |relay| relay.cursor() == 3) .await; assert_eq!(relay.forwarded.lock().len(), 1, "a stale sync got through"); Ok(())}
#[cfg(feature = "indexer")]#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn a_sync_dated_past_the_clock_is_ignored() -> Result<()> { let relay = Pipeline::start(Options::default()).await?; let alice = did("alice"); let key = signing_key(1); relay.put_row(&alice, &known_row(Some(&key), Duration::from_secs(60)))?; let honest = Tid::now_0(); let rev_of = |relay: &Pipeline| relay.row(&alice).unwrap().root.unwrap().rev.to_tid();
let future = Tid::new("jzzzzzzzzzzzz").unwrap(); relay.send(sync_at(&alice, 1, &key, future).await).await; relay .send(sync_at(&alice, 2, &key, honest.clone()).await) .await; relay .until(Duration::from_secs(5), |relay| relay.cursor() == 2) .await;
assert_eq!( relay.forwarded.lock().len(), 1, "only the honest sync got through" ); assert_eq!(rev_of(&relay), honest); Ok(())}
/// with the indexer marking frames done, the relay hands a forwarded frame's ticket on with its/// event and keeps only the frames that forwarded nothing#[cfg(feature = "indexer")]#[tokio::test(flavor = "multi_thread", worker_threads = 2)]677 unmodified linesExpand all