diff --git a/consumer/src/firehose/mod.rs b/consumer/src/firehose/mod.rs --- a/consumer/src/firehose/mod.rs +++ b/consumer/src/firehose/mod.rs @@ -117,6 +117,21 @@ FirehoseEvent::Label(event) } + "#sync" => { + counter!("firehose_events.total", "event" => "sync").increment(1); + let event: AtpSyncEvent = + serde_ipld_dagcbor::from_reader(&mut reader)?; + + // increment the seq + if self.seq < event.seq { + self.seq = event.seq; + } else { + tracing::error!("Event sequence was not greater than previous seq, exiting. {} <= {}", event.seq, self.seq); + return Ok(FirehoseOutput::Close); + } + + FirehoseEvent::Sync(event) + } _ => { tracing::warn!("unknown event type {ty}"); return Ok(FirehoseOutput::Continue); diff --git a/consumer/src/firehose/types.rs b/consumer/src/firehose/types.rs --- a/consumer/src/firehose/types.rs +++ b/consumer/src/firehose/types.rs @@ -31,6 +31,7 @@ Account(AtpAccountEvent), Commit(AtpCommitEvent), Label(AtpLabelEvent), + Sync(AtpSyncEvent), } #[derive(Debug, Deserialize)] @@ -48,6 +49,8 @@ Suspended, Deleted, Deactivated, + Throttled, + Desynchronized, } impl AtpAccountStatus { @@ -57,6 +60,8 @@ AtpAccountStatus::Suspended => "suspended", AtpAccountStatus::Deleted => "deleted", AtpAccountStatus::Deactivated => "deactivated", + AtpAccountStatus::Throttled => "throttled", + AtpAccountStatus::Desynchronized => "desynchronized", } } } @@ -68,6 +73,9 @@ AtpAccountStatus::Suspended => parakeet_db::types::ActorStatus::Suspended, AtpAccountStatus::Deleted => parakeet_db::types::ActorStatus::Deleted, AtpAccountStatus::Deactivated => parakeet_db::types::ActorStatus::Deactivated, + AtpAccountStatus::Throttled | AtpAccountStatus::Desynchronized => { + parakeet_db::types::ActorStatus::Active + } } } } @@ -90,19 +98,24 @@ pub since: Option, pub commit: Cid, #[serde(rename = "tooBig")] + #[deprecated] pub too_big: bool, #[serde(default)] pub blocks: ByteBuf, #[serde(default)] pub ops: Vec, #[serde(default)] + #[deprecated] pub blobs: Vec, + #[serde(rename = "prevData")] + pub prev_data: Option, } #[derive(Debug, Deserialize)] pub struct CommitOp { pub action: String, pub cid: Option, + pub prev: Option, pub path: String, } @@ -123,4 +136,14 @@ pub struct AtpLabelEvent { pub seq: u64, pub labels: Vec, +} + +#[derive(Debug, Deserialize)] +pub struct AtpSyncEvent { + pub seq: u64, + pub did: String, + pub time: DateTime, + pub rev: String, + #[serde(default)] + pub blocks: ByteBuf, } diff --git a/consumer/src/indexer/mod.rs b/consumer/src/indexer/mod.rs --- a/consumer/src/indexer/mod.rs +++ b/consumer/src/indexer/mod.rs @@ -1,8 +1,8 @@ use crate::config::HistoryMode; use crate::db; use crate::firehose::{ - AtpAccountEvent, AtpCommitEvent, AtpIdentityEvent, CommitOp, FirehoseConsumer, FirehoseEvent, - FirehoseOutput, + AtpAccountEvent, AtpCommitEvent, AtpIdentityEvent, AtpSyncEvent, CommitOp, FirehoseConsumer, + FirehoseEvent, FirehoseOutput, }; use crate::indexer::types::{ AggregateDeltaStore, BackfillItem, BackfillItemInner, CollectionType, RecordTypes, @@ -107,6 +107,9 @@ FirehoseEvent::Commit(commit) => { index_commit(&mut state, &mut conn, &mut rc, commit).await } + FirehoseEvent::Sync(sync) => { + process_sync(&state, &mut conn, &mut rc, sync).await + } FirehoseEvent::Label(_) => unreachable!(), }; @@ -188,6 +191,7 @@ FirehoseEvent::Identity(identity) => self.hasher.hash_one(&identity.did) % threads, FirehoseEvent::Account(account) => self.hasher.hash_one(&account.did) % threads, FirehoseEvent::Commit(commit) => self.hasher.hash_one(&commit.repo) % threads, + FirehoseEvent::Sync(sync) => self.hasher.hash_one(&sync.did) % threads, FirehoseEvent::Label(_) => { // We handle all labels through direct connections to labelers tracing::warn!("got #labels from the relay"); @@ -199,6 +203,26 @@ tracing::error!("Error sending event: {e}"); } } +} + +#[instrument(skip_all, fields(seq = sync.seq, repo = sync.did))] +async fn process_sync( + state: &RelayIndexerState, + conn: &mut Object, + rc: &mut MultiplexedConnection, + sync: AtpSyncEvent, +) -> eyre::Result<()> { + let Some((sync_state, Some(current_rev))) = db::actor_get_repo_status(conn, &sync.did).await? else { + return Ok(()); + }; + + // don't care if we're not synced. also no point if !do_backfill bc we might not have a worker + if sync_state == ActorSyncState::Synced && state.do_backfill && sync.rev > current_rev { + tracing::debug!("triggering backfill due to #sync"); + rc.rpush::<_, _, i32>("backfill_queue", sync.did).await?; + } + + Ok(()) } #[instrument(skip_all, fields(seq = identity.seq, repo = identity.did))]