diff --git a/src/backfill/mod.rs b/src/backfill/mod.rs index 6053618..b85dd52 100644 --- a/src/backfill/mod.rs +++ b/src/backfill/mod.rs @@ -41,20 +41,19 @@ pub(crate) fn backfilled_state( return Ok(None); }; let current = crate::db::deser_repo_state(&bytes)?.into_static(); - // a pds only hands over an active account's repo, so one marked inactive before the - // fetch came back - if !after.active { - after.active = true; - after.status = RepoStatus::Desynchronized; - } // the doc and the pds both answered when the fetch started, so an identity or account - // change written during it is newer and stays - if current.identity() != before.identity() { + // written during it is newer and stays + if current.identity_written_since(before) { after.set_identity(current.identity()); } - if (current.active, ¤t.status) != (before.active, &before.status) { + if current.account_written_since(before) { after.active = current.active; after.status = current.status.clone(); + } else if !after.active { + // a pds only hands over an active account's repo, so one marked inactive before the + // fetch came back + after.active = true; + after.status = RepoStatus::Desynchronized; } Ok(Some(after.rebased(before, current))) } diff --git a/src/backfill/worker/task.rs b/src/backfill/worker/task.rs index e87174d..429c17c 100644 --- a/src/backfill/worker/task.rs +++ b/src/backfill/worker/task.rs @@ -1537,6 +1537,43 @@ mod tests { Ok(()) } + #[tokio::test] + async fn a_newer_event_repeating_the_old_value_beats_what_the_fetch_saw() -> miette::Result<()> + { + let fixture = Fixture::new().await?; + let mut seeded = rooted("3jzfcijpj2z2a")?; + seeded.active = false; + seeded.status = RepoStatus::Deactivated; + seeded.last_account_time = Some(1_000); + seeded.last_identity_time = Some(1_000); + fixture.seed(&seeded, 7, &[])?; + let body = post("a"); + let car = build_car("3jzfcijpj2z2b", &[("a", body.as_slice())]).await?; + fixture.pds.serve_car(car.bytes).await; + + let result = fixture + .run_changing(keys::pending_key(7), async { + // what the relay writes for a newer deactivation and identity event + // that say what the row already says + let mut row = fixture.repo()?; + row.last_account_time = Some(2_000); + row.last_identity_time = Some(2_000); + fixture.put_repo(&row) + }) + .await?; + + assert!( + matches!(result, Ok(TaskDisposition::Finished)), + "{result:?}" + ); + let after = fixture.repo()?; + assert!(!after.active); + assert_eq!(after.status, RepoStatus::Deactivated); + assert_eq!(after.pds.as_deref(), seeded.pds.as_deref()); + assert_eq!(after.last_account_time, Some(2_000)); + Ok(()) + } + #[tokio::test] async fn an_identity_written_during_the_fetch_outlives_the_backfills_doc() -> miette::Result<()> { diff --git a/src/types.rs b/src/types.rs index d7bce11..5e39cee 100644 --- a/src/types.rs +++ b/src/types.rs @@ -262,14 +262,28 @@ impl<'i> RepoState<'i> { self.signing_key = signing_key; } + /// whether going from `before` to `self` saw an account event or changed the account. + /// the time counts on its own because a newer event can repeat the old status + #[cfg(feature = "indexer")] + pub(crate) fn account_written_since(&self, before: &Self) -> bool { + self.last_account_time != before.last_account_time + || (self.active, &self.status) != (before.active, &before.status) + } + + /// like [`Self::account_written_since`], for the handle, pds and signing key + #[cfg(feature = "indexer")] + pub(crate) fn identity_written_since(&self, before: &Self) -> bool { + self.last_identity_time != before.last_identity_time || self.identity() != before.identity() + } + /// puts what one writer changed, going from `before` to `self`, onto `current`, the /// row as it is now, so whatever another writer changed in between survives. if both /// changed the same part `self` wins, but a root never goes back to an older rev and /// the event times only move forward #[cfg(feature = "indexer")] pub(crate) fn rebased(self, before: &Self, current: Self) -> Self { - let account_changed = (self.active, &self.status) != (before.active, &before.status); - let identity_changed = self.identity() != before.identity(); + let account_changed = self.account_written_since(before); + let identity_changed = self.identity_written_since(before); let root_changed = self.root != before.root; let (active, status) = if account_changed { @@ -518,6 +532,36 @@ mod tests { assert_eq!(merged.root, Some(commit_at("3jzfcijpj2z2b"))); } + #[cfg(feature = "indexer")] + #[test] + fn a_rebase_keeps_a_newer_event_that_repeats_the_old_value() -> miette::Result<()> { + let mut before = RepoState::synced(); + before.active = false; + before.status = RepoStatus::Deactivated; + before.handle = Some(Handle::new_static("alice.test").into_diagnostic()?); + before.last_account_time = Some(10); + before.last_identity_time = Some(10); + + let mut mine = before.clone(); + mine.last_account_time = Some(20); + mine.last_identity_time = Some(20); + + let mut theirs = before.clone(); + theirs.active = true; + theirs.status = RepoStatus::Desynchronized; + theirs.handle = Some(Handle::new_static("bob.test").into_diagnostic()?); + + let merged = mine.rebased(&before, theirs); + assert!(!merged.active); + assert_eq!(merged.status, RepoStatus::Deactivated); + assert_eq!( + merged.handle.as_ref().map(Handle::as_str), + Some("alice.test") + ); + assert_eq!(merged.last_account_time, Some(20)); + Ok(()) + } + #[test] fn commit_and_repo_status_conversion_contracts() { let did: Did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();