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<u8>,
},
/// 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<AtpCommit> 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<AtpCommit> {
// 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<Self>,
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<Self>, 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<Self>,
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::<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,
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::<miette::Result<Vec<_>>>()?;
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<i64>) -> 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::<Vec<_>>();
assert_eq!(
encoded,
[
"81a2563181a54170706c7991c4080000000000000007",
"81a25631a54572617365",
"81a2563181a45769706591c408000000000000002a",
]
);
for job in jobs {
let bytes = rmp_serde::to_vec(&job).unwrap();
assert_eq!(rmp_serde::from_slice::<BulkJob>(&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::<Vec<_>>();
assert_eq!(
encoded.iter().map(crate::util::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,
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::<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);
}
}