use std::fmt::Display; use jacquard_common::types::string::{Did, Handle}; use jacquard_common::{CowStr, IntoStatic}; pub(crate) use crate::mst::AtpCommit; #[cfg(feature = "indexer")] use serde::{Deserialize, Serialize}; use smol_str::ToSmolStr; use crate::db::types::{DbTid, DidKey}; use crate::resolver::MiniDoc; #[cfg(any(feature = "indexer_stream", feature = "relay", feature = "jetstream"))] pub(crate) mod event; #[cfg(test)] pub(crate) mod v14; pub(crate) mod v15; pub(crate) mod v16; pub(crate) mod v2; pub(crate) mod v4; pub(crate) mod v7; #[cfg(any(feature = "indexer_stream", feature = "relay", feature = "jetstream"))] pub(crate) use event::*; pub(crate) use v15::RepoState; pub(crate) use v16::*; /// which record-scoped body copies an operator redaction removes. #[cfg(feature = "indexer")] #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum DeleteBodyTarget { Head, History, All, } #[cfg(feature = "indexer")] impl DeleteBodyTarget { pub(crate) fn includes_head(self) -> bool { matches!(self, Self::Head | Self::All) } pub(crate) fn includes_history(self) -> bool { matches!(self, Self::History | Self::All) } } /// a bulk job the owner shard runs in steps, kept on disk while it runs so a restart resumes /// it. each kind says what the job was started for, so a restart can tell a job that's still /// due from one that something else made pointless meanwhile #[cfg(feature = "indexer")] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub(crate) enum BulkJob { V1(BulkJobV1), } #[cfg(feature = "indexer")] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub(crate) enum BulkJobV1 { /// a backfill writing its records in steps, for the queue entry under this key Apply { #[serde(with = "serde_bytes")] pending_key: Vec, }, /// an operator erase. an exclude it asked for went into the filter in the txn that /// wrote this, so a resumed erase doesn't need to know Erase, /// a deleted account's records moving to history, as of `death_rev` Wipe { death_rev: DbTid }, } /// idempotent result of deleting indexed body bytes while preserving logical /// record identity and repository topology. #[cfg(feature = "indexer")] #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct DeleteBodiesReport { pub heads_redacted: u64, pub history_redacted: u64, pub event_bodies_redacted: u64, pub body_bytes_redacted: u64, pub buffered_commits_deleted: u64, pub buffered_commit_bytes_deleted: u64, } impl From for Commit { fn from(value: AtpCommit) -> Self { Self { data: value.data, prev: value.prev, rev: DbTid::from(&value.rev), sig: value.sig, version: value.version, } } } impl Commit { pub(crate) fn into_atp_commit(self, did: Did) -> Option { // version < 0 is a sentinel used in v2 migration for repos with no commit data if self.version < 0 { return None; } Some(AtpCommit { did, rev: self.rev.to_tid(), data: self.data, prev: self.prev, sig: self.sig, version: self.version, }) } } impl Display for RepoStatus { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { RepoStatus::Synced => write!(f, "synced"), RepoStatus::Error(e) => write!(f, "error({e})"), RepoStatus::Deactivated => write!(f, "deactivated"), RepoStatus::Takendown => write!(f, "takendown"), RepoStatus::Suspended => write!(f, "suspended"), RepoStatus::Deleted => write!(f, "deleted"), RepoStatus::Desynchronized => write!(f, "desynchronized"), RepoStatus::Throttled => write!(f, "throttled"), } } } #[cfg(feature = "indexer")] mod indexer { use super::*; impl RepoMetadata { pub fn backfilling(index_id: u64) -> Self { Self { index_id, tracked: true, retry_count: 0, relay_inactive: None, backfilled_rev: None, refetch: None, refetch_deferred: false, } } /// whether the last car a backfill wrote already has the commit at `rev` pub fn car_covers(&self, rev: &jacquard_common::types::tid::Tid) -> bool { self.backfilled_rev .is_some_and(|car| DbTid::from(rev) <= car) } } impl Allowance { /// `Err` is the unix time the window ends, not how long to wait pub fn check( allowance: &Option, now: i64, limit: u32, window: i64, ) -> Result<(), i64> { match allowance.filter(|allowance| now < allowance.since.saturating_add(window)) { Some(current) if current.used >= limit => Err(current.since.saturating_add(window)), _ => Ok(()), } } /// a charge past the window's end starts the next window pub fn charge(allowance: &mut Option, now: i64, amount: u32, window: i64) { let current = allowance .filter(|allowance| now < allowance.since.saturating_add(window)) .unwrap_or(Self { since: now, used: 0, }); *allowance = Some(Self { used: current.used.saturating_add(amount), ..current }); } pub fn take( allowance: &mut Option, now: i64, limit: u32, window: i64, ) -> Result<(), i64> { Self::check(allowance, now, limit, window)?; Self::charge(allowance, now, 1, window); Ok(()) } } impl ResyncState { pub fn next_backoff(retry_count: u32) -> i64 { // the caller counts the failure before asking, so the first retry waits 2m, then 4m, // 8m... up to 1h let base = 60; let cap = 3600; let mult = 2u64.pow(retry_count.min(10)) as i64; let delay = (base * mult).min(cap); // add +/- 10% jitter let jitter = (rand::random::() * 0.2 - 0.1) * delay as f64; let delay = (delay as f64 + jitter) as i64; chrono::Utc::now().timestamp() + delay } } #[cfg(feature = "indexer_stream")] #[derive(Clone, Debug)] pub(crate) enum BroadcastEvent { Persisted(#[allow(dead_code)] u64), /// a durable record event with optional inline block bytes for live tailing. /// /// used to avoid re-reading the durable event and record body when tailing. LiveRecord(std::sync::Arc), Ephemeral(Box>), } #[derive(Debug, PartialEq, Eq, Clone, Copy)] pub(crate) enum GaugeState { Synced, Pending, Resync(Option), } impl GaugeState { pub fn is_resync(&self) -> bool { matches!(self, GaugeState::Resync(_)) } } } #[cfg(feature = "indexer")] pub(crate) use indexer::*; /// the identity a DID doc owns. it is the only part of a repo state that /// resolving a doc rewrites, so consumers that cache it have to be told when it /// moves instead of discovering it later. #[derive(Debug, Clone, PartialEq)] pub(crate) struct DocIdentity<'i> { handle: Option, pds: Option>, signing_key: Option>, } impl<'i> RepoState<'i> { pub fn backfilling() -> Self { Self { active: true, status: RepoStatus::Desynchronized, root: None, last_updated_at: chrono::Utc::now().timestamp(), handle: None, pds: None, signing_key: None, last_message_time: None, last_identity_time: None, last_account_time: None, identity_resolved_at: None, } } #[cfg(any(test, feature = "relay"))] pub fn synced() -> Self { Self { status: RepoStatus::Synced, ..Self::backfilling() } } // advances the high-water mark to event_ms if it's newer than what we've seen pub fn advance_message_time(&mut self, event_ms: i64) { self.last_message_time = Some(event_ms.max(self.last_message_time.unwrap_or(0))); } pub fn should_process_identity_time(&self, event_ms: i64) -> bool { self.last_identity_time.is_none_or(|t| event_ms > t) } pub fn advance_identity_time(&mut self, event_ms: i64) { self.last_identity_time = Some(event_ms.max(self.last_identity_time.unwrap_or(0))); self.advance_message_time(event_ms); } pub fn should_process_account_time(&self, event_ms: i64) -> bool { self.last_account_time.is_none_or(|t| event_ms > t) } pub fn advance_account_time(&mut self, event_ms: i64) { self.last_account_time = Some(event_ms.max(self.last_account_time.unwrap_or(0))); self.advance_message_time(event_ms); } // updates last_updated_at to now pub fn touch(&mut self) { self.last_updated_at = chrono::Utc::now().timestamp(); } /// the doc-owned identity, the whole of what a resolve can rewrite. pub(crate) fn identity(&self) -> DocIdentity<'i> { DocIdentity { handle: self.handle.clone(), pds: self.pds.clone(), signing_key: self.signing_key.clone(), } } /// whether `other` has the same doc-owned identity, the same as comparing both /// [`Self::identity`]s but without copying them #[cfg(any(test, feature = "indexer"))] pub(crate) fn same_identity(&self, other: &Self) -> bool { (&self.handle, &self.pds, &self.signing_key) == (&other.handle, &other.pds, &other.signing_key) } /// the identity moved, so anyone holding a cached copy must re-resolve it. #[cfg(any(test, feature = "indexer_stream"))] pub(crate) fn identity_moved_from(&self, previous: &Self) -> bool { !self.same_identity(previous) } /// the account lifecycle moved. a first sighting of a pds counts as a move: /// nothing downstream keyed off an identity it never saw. #[cfg(any(test, feature = "indexer_stream"))] pub(crate) fn account_moved_from(&self, previous: &Self) -> bool { self.active != previous.active || self.status != previous.status || previous.pds.is_none() } pub fn update_from_doc(&mut self, doc: MiniDoc) -> bool { let previous = self.identity(); self.pds = Some(CowStr::Owned(doc.pds.to_smolstr())); self.handle = doc.handle; self.signing_key = doc.key.map(From::from); self.identity() != previous } /// [`Self::update_from_doc`] with a doc fetched fresh at `at`, unix secs pub fn update_from_resolve(&mut self, doc: MiniDoc, at: i64) -> bool { self.identity_resolved_at = Some(at); self.update_from_doc(doc) } #[cfg(feature = "indexer")] pub(crate) fn set_identity(&mut self, identity: DocIdentity<'i>) { let DocIdentity { handle, pds, signing_key, } = identity; self.handle = handle; self.pds = pds; 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. the /// identity has no such check, because the relay moves its clock even for an event it got /// no fresh doc for, so there only a changed value counts #[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) } /// whether going from `before` to `self` resolved the doc or otherwise wrote the identity, /// which a refresh that finds the same doc does too, by moving only the resolve time #[cfg(feature = "indexer")] fn identity_written_since(&self, before: &Self) -> bool { !self.same_identity(before) || self.identity_resolved_at != before.identity_resolved_at } /// 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, the /// event times only move forward, and the identity (handle, pds, key and when they were /// resolved, as one) goes to whichever side resolved it last #[cfg(feature = "indexer")] pub(crate) fn rebased(self, before: &Self, current: Self) -> Self { let account_changed = self.account_written_since(before); let identity_changed = self.identity_written_since(before) && (!current.identity_written_since(before) || self.identity_resolved_at >= current.identity_resolved_at); let root_changed = self.root != before.root; let (active, status) = if account_changed { (self.active, self.status) } else { (current.active, current.status) }; let (handle, pds, signing_key, identity_resolved_at) = if identity_changed { ( self.handle, self.pds, self.signing_key, self.identity_resolved_at, ) } else { ( current.handle, current.pds, current.signing_key, current.identity_resolved_at, ) }; let root = match (self.root, current.root) { (_, theirs) if !root_changed => theirs, (Some(mine), Some(theirs)) if theirs.rev > mine.rev => Some(theirs), (mine, _) => mine, }; Self { active, status, root, last_message_time: self.last_message_time.max(current.last_message_time), last_identity_time: self.last_identity_time.max(current.last_identity_time), last_account_time: self.last_account_time.max(current.last_account_time), last_updated_at: self.last_updated_at.max(current.last_updated_at), signing_key, pds, handle, identity_resolved_at, } } /// when the identity was last resolved. a row from before v15 has no time, and reads as /// resolved at a point in the ttl before `epoch` picked by its did, so the rows that were /// there at the upgrade come due for a refresh spread over one ttl instead of all at once pub(crate) fn identity_resolved_or_spread(&self, did: &Did, epoch: i64, ttl: i64) -> i64 { self.identity_resolved_at.unwrap_or_else(|| { let spread = (crate::util::hash(&did.as_str()) % ttl.max(1) as u64) as i64; epoch - spread }) } } impl<'i> IntoStatic for RepoState<'i> { type Output = RepoState<'static>; fn into_static(self) -> Self::Output { RepoState { active: self.active, status: self.status, root: self.root, last_updated_at: self.last_updated_at, handle: self.handle.map(IntoStatic::into_static), pds: self.pds.map(IntoStatic::into_static), signing_key: self.signing_key.map(IntoStatic::into_static), last_message_time: self.last_message_time, last_identity_time: self.last_identity_time, last_account_time: self.last_account_time, identity_resolved_at: self.identity_resolved_at, } } } #[cfg(feature = "indexer")] #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] pub(crate) enum ResyncErrorKind { Ratelimited, Transport, Generic, } #[cfg(feature = "indexer")] #[derive(Debug, Clone, Serialize, Deserialize)] pub(crate) enum ResyncState { Error { kind: ResyncErrorKind, retry_count: u32, next_retry: i64, // unix timestamp }, Gone { status: RepoStatus, // deactivated, takendown, suspended }, } #[cfg(test)] mod tests { use super::*; use miette::IntoDiagnostic; #[test] fn identity_moves_only_for_doc_owned_fields() -> miette::Result<()> { let mut previous = RepoState::synced(); previous.pds = Some(CowStr::Borrowed("https://pds.example")); previous.handle = Some(Handle::new_static("alice.test").into_diagnostic()?); let mut state = previous.clone(); state.advance_message_time(1_000); state.touch(); state.root = previous.root.clone(); assert!(!state.identity_moved_from(&previous)); assert!(!state.account_moved_from(&previous)); let mut handle_moved = previous.clone(); handle_moved.handle = Some(Handle::new_static("carol.test").into_diagnostic()?); assert!(handle_moved.identity_moved_from(&previous)); assert!(!handle_moved.account_moved_from(&previous)); let mut pds_moved = previous.clone(); pds_moved.pds = Some(CowStr::Borrowed("https://other.example")); assert!(pds_moved.identity_moved_from(&previous)); let mut key_moved = previous.clone(); key_moved.signing_key = Some(DidKey::from_did_key( "did:key:zQ3shokFTS3brHcDQrn82RUDfCZESWL1ZdCEJwekUDPQiYBme", )?); assert!(key_moved.identity_moved_from(&previous)); let mut account_moved = previous.clone(); account_moved.status = RepoStatus::Throttled; assert!(account_moved.account_moved_from(&previous)); assert!(!account_moved.identity_moved_from(&previous)); Ok(()) } #[test] fn first_sighting_of_a_pds_moves_the_account() { let mut previous = RepoState::backfilling(); previous.pds = None; let mut state = previous.clone(); state.pds = Some(CowStr::Borrowed("https://pds.example")); assert!(state.account_moved_from(&previous)); assert!(state.identity_moved_from(&previous)); } #[test] fn identity_dedupe_does_not_depend_on_commit_clock() { let mut state = RepoState::backfilling(); state.advance_message_time(2_000); assert!(state.should_process_identity_time(1_500)); state.advance_identity_time(1_500); assert_eq!(state.last_message_time, Some(2_000)); assert_eq!(state.last_identity_time, Some(1_500)); assert!(!state.should_process_identity_time(1_500)); assert!(!state.should_process_identity_time(1_499)); assert!(state.should_process_identity_time(1_501)); } #[test] fn account_dedupe_does_not_depend_on_commit_clock() { let mut state = RepoState::backfilling(); state.advance_message_time(3_000); assert!(state.should_process_account_time(2_500)); state.advance_account_time(2_500); assert_eq!(state.last_message_time, Some(3_000)); assert_eq!(state.last_account_time, Some(2_500)); assert!(!state.should_process_account_time(2_500)); assert!(!state.should_process_account_time(2_499)); assert!(state.should_process_account_time(2_501)); } #[test] fn synced_state_is_active_without_backfilling_status() { let state = RepoState::synced(); assert!(state.active); assert_eq!(state.status, RepoStatus::Synced); assert!(state.root.is_none()); } #[test] fn into_static_preserves_per_event_clocks() -> miette::Result<()> { let mut state = RepoState::backfilling(); state.last_message_time = Some(10); state.last_identity_time = Some(20); state.last_account_time = Some(30); state.handle = Some(Handle::new_static("alice.test").into_diagnostic()?); state.pds = Some(CowStr::Borrowed("https://pds.example")); state.signing_key = Some(DidKey::from_did_key( "did:key:zQ3shokFTS3brHcDQrn82RUDfCZESWL1ZdCEJwekUDPQiYBme", )?); let static_state = state.into_static(); assert_eq!(static_state.last_message_time, Some(10)); assert_eq!(static_state.last_identity_time, Some(20)); assert_eq!(static_state.last_account_time, Some(30)); Ok(()) } #[cfg(feature = "indexer")] fn commit_at(rev: &str) -> Commit { Commit { version: 3, rev: crate::db::types::DbTid::from( &jacquard_common::types::tid::Tid::new(rev).unwrap(), ), data: crate::car::compute_cid(rev.as_bytes()), prev: None, sig: bytes::Bytes::from_static(b"signature"), } } #[cfg(feature = "indexer")] #[test] fn a_rebase_keeps_what_only_the_other_writer_changed() -> miette::Result<()> { let mut before = RepoState::synced(); before.root = Some(commit_at("3jzfcijpj2z2a")); before.handle = Some(Handle::new_static("alice.test").into_diagnostic()?); before.last_message_time = Some(10); let mut mine = before.clone(); mine.root = Some(commit_at("3jzfcijpj2z2b")); mine.last_message_time = Some(30); let mut theirs = before.clone(); theirs.active = false; theirs.status = RepoStatus::Deactivated; theirs.handle = Some(Handle::new_static("bob.test").into_diagnostic()?); theirs.last_message_time = Some(20); theirs.last_account_time = Some(20); let merged = mine.rebased(&before, theirs); assert_eq!(merged.root, Some(commit_at("3jzfcijpj2z2b"))); assert!(!merged.active); assert_eq!(merged.status, RepoStatus::Deactivated); assert_eq!(merged.handle.as_ref().map(Handle::as_str), Some("bob.test")); assert_eq!(merged.last_message_time, Some(30)); assert_eq!(merged.last_account_time, Some(20)); Ok(()) } #[cfg(feature = "indexer")] #[test] fn a_rebase_lets_this_writer_win_a_conflict_but_never_moves_the_root_back() { let before = RepoState::synced(); let mut mine = before.clone(); mine.status = RepoStatus::Takendown; mine.active = false; mine.root = Some(commit_at("3jzfcijpj2z2a")); let mut theirs = before.clone(); theirs.status = RepoStatus::Throttled; theirs.root = Some(commit_at("3jzfcijpj2z2b")); let merged = mine.rebased(&before, theirs); assert_eq!(merged.status, RepoStatus::Takendown); assert!(!merged.active); assert_eq!(merged.root, Some(commit_at("3jzfcijpj2z2b"))); } #[cfg(feature = "indexer")] #[test] fn a_rebase_keeps_a_newer_event_that_repeats_the_old_value() { let mut before = RepoState::synced(); before.active = false; before.status = RepoStatus::Deactivated; before.last_account_time = Some(10); let mut mine = before.clone(); mine.last_account_time = Some(20); let mut theirs = before.clone(); theirs.active = true; theirs.status = RepoStatus::Desynchronized; let merged = mine.rebased(&before, theirs); assert!(!merged.active); assert_eq!(merged.status, RepoStatus::Deactivated); assert_eq!(merged.last_account_time, Some(20)); } #[cfg(feature = "indexer")] #[test] fn an_identity_clock_alone_doesnt_beat_a_newer_identity() { let mut before = RepoState::synced(); before.pds = Some(CowStr::Borrowed("https://a.example/")); before.last_identity_time = Some(10); // the relay handling an identity event it got no fresh doc for let mut mine = before.clone(); mine.last_identity_time = Some(20); // a backfill persisting the doc it resolved let mut theirs = before.clone(); theirs.pds = Some(CowStr::Borrowed("https://b.example/")); let merged = mine.rebased(&before, theirs); assert_eq!(merged.pds.as_deref(), Some("https://b.example/")); assert_eq!(merged.last_identity_time, Some(20)); } #[test] fn rows_without_a_resolve_time_spread_over_one_ttl() -> miette::Result<()> { let ttl = 86_400; let epoch = 1_700_000_000; let ages = (0..1000) .map(|i| { let did = Did::new_owned(format!("did:plc:{i:0>24}")).into_diagnostic()?; let state = RepoState::backfilling(); let at = state.identity_resolved_or_spread(&did, epoch, ttl); assert_eq!(at, state.identity_resolved_or_spread(&did, epoch, ttl)); Ok(epoch - at) }) .collect::>>()?; assert!(ages.iter().all(|age| (0..ttl).contains(age))); // a tenth of the ttl each, give or take for slot in 0..10 { let in_slot = ages.iter().filter(|age| **age * 10 / ttl == slot).count(); assert!((50..150).contains(&in_slot), "slot {slot} has {in_slot}"); } let mut stamped = RepoState::backfilling(); stamped.identity_resolved_at = Some(42); let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").into_diagnostic()?; assert_eq!(stamped.identity_resolved_or_spread(&did, epoch, ttl), 42); Ok(()) } #[cfg(feature = "indexer")] fn resolved(pds: &'static str, at: Option) -> RepoState<'static> { RepoState { pds: Some(CowStr::Borrowed(pds)), identity_resolved_at: at, ..RepoState::synced() } } #[cfg(feature = "indexer")] #[test] fn a_refresh_that_finds_the_same_doc_keeps_its_stamp() { let before = resolved("https://a.example/", Some(10)); let mine = resolved("https://a.example/", Some(20)); let mut theirs = before.clone(); theirs.last_message_time = Some(5); let merged = mine.rebased(&before, theirs); assert_eq!(merged.identity_resolved_at, Some(20)); assert_eq!(merged.last_message_time, Some(5)); } #[cfg(feature = "indexer")] #[test] fn the_newest_resolve_wins_the_whole_identity() { let before = resolved("https://a.example/", Some(10)); // the relay resolved b, then a backfill that resolved after it wrote c let relay = resolved("https://b.example/", Some(20)); let backfill = resolved("https://c.example/", Some(30)); let merged = relay.clone().rebased(&before, backfill.clone()); assert_eq!(merged.pds.as_deref(), Some("https://c.example/")); assert_eq!(merged.identity_resolved_at, Some(30)); // and the other way around let merged = backfill.rebased(&before, relay); assert_eq!(merged.pds.as_deref(), Some("https://c.example/")); assert_eq!(merged.identity_resolved_at, Some(30)); // a writer that never stamps still beats a row nobody else touched let unstamped = resolved("https://d.example/", Some(10)); let mut moved = unstamped.clone(); moved.pds = Some(CowStr::Borrowed("https://e.example/")); let merged = moved.rebased(&unstamped, unstamped.clone()); assert_eq!(merged.pds.as_deref(), Some("https://e.example/")); } #[cfg(feature = "indexer")] #[test] fn marking_an_identity_stale_sticks_unless_someone_resolved_since() { let before = resolved("https://a.example/", Some(10)); let stale = resolved("https://a.example/", Some(0)); let merged = stale.clone().rebased(&before, before.clone()); assert_eq!(merged.identity_resolved_at, Some(0)); let fresh = resolved("https://a.example/", Some(30)); let merged = stale.rebased(&before, fresh); assert_eq!(merged.identity_resolved_at, Some(30)); } #[test] fn commit_and_repo_status_conversion_contracts() { let did: Did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let rev = jacquard_common::types::tid::Tid::new("3jzfcijpj2z2a").unwrap(); let data = crate::car::compute_cid(b"root"); let prev = crate::car::compute_cid(b"prev"); let atp = AtpCommit { did: did.clone(), version: 3, data, rev: rev.clone(), prev: Some(prev), sig: bytes::Bytes::from_static(b"\x01\x02\x03"), }; let stored = Commit::from(atp); assert_eq!(stored.version, 3); assert_eq!(stored.data, data); assert_eq!(stored.prev, Some(prev)); assert_eq!(stored.rev.to_tid(), rev); assert_eq!(stored.sig.as_ref(), b"\x01\x02\x03"); let restored = stored.into_atp_commit(did.clone()).unwrap(); assert_eq!(restored.did, did); assert_eq!(restored.version, 3); assert_eq!(restored.data, data); assert_eq!(restored.prev, Some(prev)); assert_eq!(restored.rev, rev); assert_eq!(restored.sig.as_ref(), b"\x01\x02\x03"); let mut sentinel = Commit::from(restored); sentinel.version = -1; assert!(sentinel.into_atp_commit(did).is_none()); for (status, display) in [ (RepoStatus::Synced, "synced"), (RepoStatus::Error("stinkpot".into()), "error(stinkpot)"), (RepoStatus::Deactivated, "deactivated"), (RepoStatus::Takendown, "takendown"), (RepoStatus::Suspended, "suspended"), (RepoStatus::Deleted, "deleted"), (RepoStatus::Desynchronized, "desynchronized"), (RepoStatus::Throttled, "throttled"), ] { assert_eq!(status.to_string(), display); } } #[cfg(feature = "indexer")] #[test] fn bulk_job_golden_fixtures() { let jobs = [ BulkJob::V1(BulkJobV1::Apply { pending_key: vec![0, 0, 0, 0, 0, 0, 0, 7], }), BulkJob::V1(BulkJobV1::Erase), BulkJob::V1(BulkJobV1::Wipe { death_rev: DbTid::new_from_bytes(42_u64.to_be_bytes()), }), ]; let encoded = jobs .iter() .map(|job| crate::util::hex::encode(rmp_serde::to_vec(job).unwrap())) .collect::>(); assert_eq!( encoded, [ "81a2563181a54170706c7991c4080000000000000007", "81a25631a54572617365", "81a2563181a45769706591c408000000000000002a", ] ); for job in jobs { let bytes = rmp_serde::to_vec(&job).unwrap(); assert_eq!(rmp_serde::from_slice::(&bytes).unwrap(), job); } } #[cfg(feature = "indexer")] #[test] fn current_persisted_state_golden_fixtures() { let states = [ ResyncState::Error { kind: ResyncErrorKind::Ratelimited, retry_count: 7, next_retry: -42, }, ResyncState::Error { kind: ResyncErrorKind::Transport, retry_count: 8, next_retry: 0, }, ResyncState::Error { kind: ResyncErrorKind::Generic, retry_count: 9, next_retry: 42, }, ResyncState::Gone { status: RepoStatus::Deleted, }, ]; let encoded = states .iter() .map(|state| rmp_serde::to_vec(state).unwrap()) .collect::>(); assert_eq!( encoded.iter().map(crate::util::hex::encode).collect::>(), [ "81a54572726f7293ab526174656c696d6974656407d0d6", "81a54572726f7293a95472616e73706f72740800", "81a54572726f7293a747656e65726963092a", "81a4476f6e6591a744656c65746564", ] ); assert!(matches!( rmp_serde::from_slice::(&encoded[0]).unwrap(), ResyncState::Error { kind: ResyncErrorKind::Ratelimited, retry_count: 7, next_retry: -42, } )); assert!(matches!( rmp_serde::from_slice::(&encoded[1]).unwrap(), ResyncState::Error { kind: ResyncErrorKind::Transport, retry_count: 8, next_retry: 0, } )); assert!(matches!( rmp_serde::from_slice::(&encoded[2]).unwrap(), ResyncState::Error { kind: ResyncErrorKind::Generic, retry_count: 9, next_retry: 42, } )); assert!(matches!( rmp_serde::from_slice::(&encoded[3]).unwrap(), ResyncState::Gone { status: RepoStatus::Deleted, } )); let metadata = RepoMetadata { tracked: true, index_id: 7, retry_count: 2, relay_inactive: None, backfilled_rev: None, refetch: None, refetch_deferred: false, }; let encoded = rmp_serde::to_vec(&metadata).unwrap(); assert_eq!(crate::util::hex::encode(&encoded), "97c30702c0c0c0c2"); let decoded: RepoMetadata = rmp_serde::from_slice(&encoded).unwrap(); assert!(decoded.tracked); assert_eq!(decoded.index_id, 7); assert_eq!(decoded.retry_count, 2); } #[cfg(feature = "indexer")] #[test] fn operator_redaction_json_contracts() { for (target, expected) in [ (DeleteBodyTarget::Head, "\"head\""), (DeleteBodyTarget::History, "\"history\""), (DeleteBodyTarget::All, "\"all\""), ] { let encoded = serde_json::to_string(&target).unwrap(); assert_eq!(encoded, expected); assert_eq!( serde_json::from_str::(&encoded).unwrap(), target ); } let report = DeleteBodiesReport { heads_redacted: 1, history_redacted: 7, event_bodies_redacted: 3, body_bytes_redacted: 11, buffered_commits_deleted: 2, buffered_commit_bytes_deleted: 13, }; let encoded = serde_json::to_value(&report).unwrap(); assert_eq!(encoded["heads_redacted"], 1); assert_eq!(encoded["history_redacted"], 7); assert_eq!(encoded["event_bodies_redacted"], 3); assert_eq!(encoded["body_bytes_redacted"], 11); assert_eq!(encoded["buffered_commits_deleted"], 2); assert_eq!(encoded["buffered_commit_bytes_deleted"], 13); } }