diff --git a/src/ingest/relay/context.rs b/src/ingest/relay/context.rs index 3dc77de..aa78bdd 100644 --- a/src/ingest/relay/context.rs +++ b/src/ingest/relay/context.rs @@ -60,8 +60,8 @@ impl WorkerContext<'_> { is_suppressed } - pub(crate) fn stage_repo_state(&mut self, repo_state: &RepoState<'_>) { - self.staged = Some(repo_state.clone().into_static()); + pub(crate) fn stage_repo_state(&mut self, repo_state: RepoState<'static>) { + self.staged = Some(repo_state); } /// puts the row this message staged on `txn`. the relay holds no lock while it resolves a @@ -199,7 +199,7 @@ impl WorkerContext<'_> { let doc = doc?; repo_state.update_from_doc(doc); repo_state.touch(); - self.stage_repo_state(repo_state); + self.stage_repo_state(repo_state.clone().into_static()); Ok(()) })(); self.stats.record_refresh_doc(refresh_started.elapsed()); @@ -538,7 +538,7 @@ impl WorkerContext<'_> { let mut repo_state = RepoState::synced(); repo_state.update_from_doc(doc); - self.stage_repo_state(&repo_state); + self.stage_repo_state(repo_state.clone().into_static()); self.sink.new_repo(did.clone()); diff --git a/src/ingest/relay/handlers.rs b/src/ingest/relay/handlers.rs index ca25461..0a98c65 100644 --- a/src/ingest/relay/handlers.rs +++ b/src/ingest/relay/handlers.rs @@ -13,7 +13,7 @@ use super::{RelayWorker, WorkerContext}; impl RelayWorker { pub(crate) fn handle_commit( ctx: &mut WorkerContext, - repo_state: &mut RepoState, + mut repo_state: RepoState<'static>, firehose: &Url, commit: Commit<'static>, ) -> Result<()> { @@ -23,7 +23,7 @@ impl RelayWorker { repo_state.advance_message_time(commit.time.0.timestamp_millis()); - let Some(validated) = ctx.validate_commit(repo_state, &commit)? else { + let Some(validated) = ctx.validate_commit(&mut repo_state, &commit)? else { return Ok(()); }; let ValidatedCommit { @@ -56,7 +56,7 @@ impl RelayWorker { pub(crate) fn handle_sync( ctx: &mut WorkerContext, - repo_state: &mut RepoState, + mut repo_state: RepoState<'static>, firehose: &Url, sync: Sync<'static>, ) -> Result<()> { @@ -66,7 +66,7 @@ impl RelayWorker { repo_state.advance_message_time(sync.time.0.timestamp_millis()); - let Some(validated) = ctx.validate_sync(repo_state, &sync)? else { + let Some(validated) = ctx.validate_sync(&mut repo_state, &sync)? else { return Ok(()); }; @@ -81,7 +81,7 @@ impl RelayWorker { pub(crate) fn handle_identity( ctx: &mut WorkerContext, - repo_state: &mut RepoState, + mut repo_state: RepoState<'static>, firehose: &Url, mut identity: Identity, is_pds: bool, @@ -132,7 +132,7 @@ impl RelayWorker { pub(crate) fn handle_account( ctx: &mut WorkerContext, - repo_state: &mut RepoState, + mut repo_state: RepoState<'static>, firehose: &Url, account: Account<'static>, _is_pds: bool, @@ -146,7 +146,7 @@ impl RelayWorker { repo_state.advance_account_time(event_ms); let was_active = repo_state.active; - let snapshot = ctx.sink.account_snapshot(repo_state); + let snapshot = ctx.sink.account_snapshot(&repo_state); repo_state.active = account.active; if !account.active { @@ -177,7 +177,7 @@ impl RelayWorker { &mut ctx.batch, firehose, account, - repo_state, + &repo_state, snapshot, was_active, )?; diff --git a/src/ingest/relay/worker.rs b/src/ingest/relay/worker.rs index 8665b57..c6782e9 100644 --- a/src/ingest/relay/worker.rs +++ b/src/ingest/relay/worker.rs @@ -279,7 +279,7 @@ impl RelayWorker { SubscribeReposMessage::Commit(commit) => { trace!("processing commit"); let started = StatsInstant::now(); - let result = Self::handle_commit(ctx, &mut repo_state, &msg.firehose, *commit); + let result = Self::handle_commit(ctx, repo_state, &msg.firehose, *commit); ctx.stats.record_handle_message( crate::ingest::firehose_stats::RelayMessageKind::Commit, started.elapsed(), @@ -289,7 +289,7 @@ impl RelayWorker { SubscribeReposMessage::Sync(sync) => { debug!("processing sync"); let started = StatsInstant::now(); - let result = Self::handle_sync(ctx, &mut repo_state, &msg.firehose, *sync); + let result = Self::handle_sync(ctx, repo_state, &msg.firehose, *sync); ctx.stats.record_handle_message( crate::ingest::firehose_stats::RelayMessageKind::Sync, started.elapsed(), @@ -301,7 +301,7 @@ impl RelayWorker { let started = StatsInstant::now(); let result = Self::handle_identity( ctx, - &mut repo_state, + repo_state, &msg.firehose, *identity, msg.is_pds, @@ -318,7 +318,7 @@ impl RelayWorker { debug!("processing account"); let started = StatsInstant::now(); let result = - Self::handle_account(ctx, &mut repo_state, &msg.firehose, *account, msg.is_pds); + Self::handle_account(ctx, repo_state, &msg.firehose, *account, msg.is_pds); ctx.stats.record_handle_message( crate::ingest::firehose_stats::RelayMessageKind::Account, started.elapsed(),