Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566use std::fmt::Display;
use jacquard_common::types::string::{Did, Handle};use jacquard_common::{CowStr, IntoStatic};use jacquard_repo::commit::Commit as 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;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 v7::*;
/// 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) }}
/// 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<S: jacquard_common::BosStr> From<AtpCommit<S>> for Commit { fn from(value: AtpCommit<S>) -> 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<S: jacquard_common::BosStr>( self, did: Did<S>, ) -> Option<AtpCommit<S>> { // 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, } } }
impl ResyncState { pub fn next_backoff(retry_count: u32) -> i64 { // exponential backoff: 1m, 2m, 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::<f64>() * 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<super::LiveRecordEvent>), Ephemeral(Box<MarshallableEvt<'static>>), }
#[derive(Debug, PartialEq, Eq, Clone, Copy)] pub(crate) enum GaugeState { Synced, Pending, Resync(Option<ResyncErrorKind>), }
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<Handle>, pds: Option<CowStr<'i>>, signing_key: Option<DidKey<'i>>,}
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, } }
#[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(), } }
/// 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.identity() != previous.identity() }
/// 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 }}
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, } }}
#[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(()) }
#[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 = jacquard_repo::mst::util::compute_cid(b"root").unwrap(); let prev = jacquard_repo::mst::util::compute_cid(b"prev").unwrap(); 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 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::<Vec<_>>(); assert_eq!( encoded.iter().map(hex::encode).collect::<Vec<_>>(), [ "81a54572726f7293ab526174656c696d6974656407d0d6", "81a54572726f7293a95472616e73706f72740800", "81a54572726f7293a747656e65726963092a", "81a4476f6e6591a744656c65746564", ] ); assert!(matches!( rmp_serde::from_slice::<ResyncState>(&encoded[0]).unwrap(), ResyncState::Error { kind: ResyncErrorKind::Ratelimited, retry_count: 7, next_retry: -42, } )); assert!(matches!( rmp_serde::from_slice::<ResyncState>(&encoded[1]).unwrap(), ResyncState::Error { kind: ResyncErrorKind::Transport, retry_count: 8, next_retry: 0, } )); assert!(matches!( rmp_serde::from_slice::<ResyncState>(&encoded[2]).unwrap(), ResyncState::Error { kind: ResyncErrorKind::Generic, retry_count: 9, next_retry: 42, } )); assert!(matches!( rmp_serde::from_slice::<ResyncState>(&encoded[3]).unwrap(), ResyncState::Gone { status: RepoStatus::Deleted, } ));
let metadata = RepoMetadata { tracked: true, index_id: 7, }; let encoded = rmp_serde::to_vec(&metadata).unwrap(); assert_eq!(hex::encode(&encoded), "92c307"); let decoded: RepoMetadata = rmp_serde::from_slice(&encoded).unwrap(); assert!(decoded.tracked); assert_eq!(decoded.index_id, 7); }
#[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::<DeleteBodyTarget>(&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); }}