use jacquard_common::types::{did::Did, string::Handle};
use smol_str::SmolStr;
use crate::control::repos::MiniDocError;
use crate::db::types::DidKey;
use super::*;
#[derive(Serialize, Deserialize)]
pub struct ResolveMiniDocOutput {
pub did: Did,
pub handle: Handle,
#[serde(serialize_with = "crate::util::did_key_serialize_str")]
#[serde(deserialize_with = "crate::util::did_key_deserialize_str")]
pub signing_key: DidKey<'static>,
pub pds: SmolStr,
}
#[derive(Serialize, Deserialize)]
pub struct ResolveMiniDocRequestData {
pub identifier: AtIdentifier,
}
pub(super) async fn resolve_mini_doc(
hydrant: &Hydrant,
identifier: &AtIdentifier,
) -> XrpcResult<ResolveMiniDocOutput> {
let did = hydrant
.state
.resolver
.resolve_did(identifier)
.await
.map_err(|e| bad_request(format!("can't resolve identifier: {e}")))?;
mini_doc_of(hydrant, &did, identifier).await
}
/// [`resolve_mini_doc`] for the did `identifier` already resolved to
pub(super) async fn mini_doc_of(
hydrant: &Hydrant,
did: &Did,
identifier: &AtIdentifier,
) -> XrpcResult<ResolveMiniDocOutput> {
let requested_handle = match identifier {
AtIdentifier::Handle(handle) => Some(handle),
AtIdentifier::Did(_) => None,
};
let repo = hydrant.repos.get(did);
let doc = match repo.mini_doc_for(requested_handle).await {
// slingshot answers for any did, so what we don't index comes from the resolver
Err(MiniDocError::RepoNotFound | MiniDocError::NotSynced) => {
repo.mini_doc_from_resolver().await
}
indexed => indexed,
}
.map_err(|e| match e {
MiniDocError::NotSynced => bad_request("repo not synced"),
MiniDocError::RepoNotFound => bad_request("repo not found"),
MiniDocError::CouldNotResolveIdentity => {
upstream_error("identity could not be resolved")
}
MiniDocError::Other(e) => internal_error(e),
})?;
Ok(ResolveMiniDocOutput {
did: doc.did,
handle: doc.handle,
pds: doc.pds.to_smolstr(),
signing_key: doc.signing_key,
})
}
pub async fn handle(
State(hydrant): State<Hydrant>,
ExtractXrpc(req): ExtractXrpc<ResolveMiniDocRequestData>,
) -> XrpcResult<Json<ResolveMiniDocOutput>> {
Ok(Json(resolve_mini_doc(&hydrant, &req.identifier).await?))
}
#[cfg(all(test, feature = "indexer_stream"))]
mod tests {
use std::borrow::Cow;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::time::Duration;
use axum::extract::State;
use axum::http::{StatusCode, Uri, header};
use axum::response::{IntoResponse, Response};
use axum::{Json, Router};
use jacquard_common::IntoStatic;
use jacquard_common::types::ident::AtIdentifier;
use jacquard_common::types::string::{Did, Handle};
use miette::IntoDiagnostic;
use tokio::sync::{RwLock, Semaphore};
use tokio::time::timeout;
use url::Url;
use super::*;
use crate::config::Config;
use crate::db::{self, Txn, keys};
use crate::ingest::indexer::FirehoseWorker;
use crate::resolver::Resolver;
use crate::types::{
BroadcastEvent, IdentityEvt, RepoMetadata, RepoState, RepoStatus, ResyncState,
};
const DID: &str = "did:plc:ewvi7nxzyoun6zhxrhs64oiz";
const OLD_HANDLE: &str = "old.test";
const NEW_HANDLE: &str = "new.test";
/// a handle that resolves to somebody else, so it never verifies
const LIAR_HANDLE: &str = "liar.test";
const OTHER_DID: &str = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa";
const SIGNING_DID_KEY: &str = "did:key:zQ3shokFTS3brHcDQrn82RUDfCZESWL1ZdCEJwekUDPQiYBme";
const SIGNING_MULTIBASE: &str = "zQ3shokFTS3brHcDQrn82RUDfCZESWL1ZdCEJwekUDPQiYBme";
#[derive(Clone)]
struct StubResponse {
status: StatusCode,
body: String,
}
struct StubIdentityInner {
did: Did,
response: RwLock<StubResponse>,
doc_requests: AtomicUsize,
handle_requests: AtomicUsize,
block_next_doc: AtomicBool,
doc_started: Semaphore,
doc_release: Semaphore,
}
#[derive(Clone)]
struct StubIdentity(Arc<StubIdentityInner>);
impl StubIdentity {
async fn set_doc(&self, body: String) {
*self.0.response.write().await = StubResponse {
status: StatusCode::OK,
body,
};
}
async fn fail_docs(&self) {
*self.0.response.write().await = StubResponse {
status: StatusCode::INTERNAL_SERVER_ERROR,
body: "upstream failed".to_owned(),
};
}
fn doc_requests(&self) -> usize {
self.0.doc_requests.load(Ordering::SeqCst)
}
fn handle_requests(&self) -> usize {
self.0.handle_requests.load(Ordering::SeqCst)
}
fn block_next_doc(&self) {
self.0.block_next_doc.store(true, Ordering::SeqCst);
}
async fn wait_until_doc_started(&self) {
timeout(Duration::from_secs(5), self.0.doc_started.acquire())
.await
.expect("fresh DID document request should start")
.expect("request-start semaphore should stay open")
.forget();
}
fn release_doc(&self) {
self.0.doc_release.add_permits(1);
}
}
async fn stub_identity_handler(State(stub): State<StubIdentity>, uri: Uri) -> Response {
if uri.path() == "/xrpc/com.atproto.identity.resolveHandle" {
stub.0.handle_requests.fetch_add(1, Ordering::SeqCst);
let did = match uri.query().is_some_and(|q| q.contains(LIAR_HANDLE)) {
true => OTHER_DID,
false => stub.0.did.as_str(),
};
return (
StatusCode::OK,
[(header::CONTENT_TYPE, "application/json")],
serde_json::json!({ "did": did }).to_string(),
)
.into_response();
}
stub.0.doc_requests.fetch_add(1, Ordering::SeqCst);
let response = stub.0.response.read().await.clone();
if stub.0.block_next_doc.swap(false, Ordering::SeqCst) {
stub.0.doc_started.add_permits(1);
stub.0
.doc_release
.acquire()
.await
.expect("document-release semaphore should stay open")
.forget();
}
(
response.status,
[(header::CONTENT_TYPE, "application/json")],
response.body,
)
.into_response()
}
async fn spawn_identity_server(body: String) -> (Url, StubIdentity) {
let stub = StubIdentity(Arc::new(StubIdentityInner {
did: Did::new_static(DID).unwrap(),
response: RwLock::new(StubResponse {
status: StatusCode::OK,
body,
}),
doc_requests: AtomicUsize::new(0),
handle_requests: AtomicUsize::new(0),
block_next_doc: AtomicBool::new(false),
doc_started: Semaphore::new(0),
doc_release: Semaphore::new(0),
}));
let app = Router::new()
.fallback(stub_identity_handler)
.with_state(stub.clone());
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
(Url::parse(&format!("http://{addr}/")).unwrap(), stub)
}
fn did_doc(handle: &str, pds: &str) -> String {
serde_json::json!({
"@context": ["https://www.w3.org/ns/did/v1"],
"id": DID,
"alsoKnownAs": [format!("at://{handle}")],
"service": [{
"id": "#atproto_pds",
"type": "AtprotoPersonalDataServer",
"serviceEndpoint": pds,
}],
"verificationMethod": [{
"id": format!("{DID}#atproto"),
"type": "Multikey",
"controller": DID,
"publicKeyMultibase": SIGNING_MULTIBASE,
}],
})
.to_string()
}
fn keyless_did_doc(handle: &str, pds: &str) -> String {
serde_json::json!({
"@context": ["https://www.w3.org/ns/did/v1"],
"id": DID,
"alsoKnownAs": [format!("at://{handle}")],
"service": [{
"id": "#atproto_pds",
"type": "AtprotoPersonalDataServer",
"serviceEndpoint": pds,
}],
})
.to_string()
}
fn pdsless_did_doc(handle: &str) -> String {
serde_json::json!({
"@context": ["https://www.w3.org/ns/did/v1"],
"id": DID,
"alsoKnownAs": [format!("at://{handle}")],
"verificationMethod": [{
"id": format!("{DID}#atproto"),
"type": "Multikey",
"controller": DID,
"publicKeyMultibase": SIGNING_MULTIBASE,
}],
})
.to_string()
}
fn old_signing_key() -> DidKey<'static> {
DidKey(Cow::Borrowed(&[
0xe7, 0x01, 0x02, 0x79, 0xbe, 0x66, 0x7e, 0xf9, 0xdc, 0xbb, 0xac, 0x55, 0xa0, 0x62,
0x95, 0xce, 0x87, 0x0b, 0x07, 0x02, 0x9b, 0xfc, 0xdb, 0x2d, 0xce, 0x28, 0xd9, 0x59,
0xf2, 0x81, 0x5b, 0x16, 0xf8, 0x17, 0x98,
]))
}
fn resolved_signing_key() -> DidKey<'static> {
DidKey::from_did_key(SIGNING_DID_KEY).unwrap()
}
fn complete_state(handle: &str, pds: &str) -> RepoState<'static> {
let mut state = RepoState::synced();
state.handle = Some(Handle::new(handle).unwrap().into_static());
state.pds = Some(pds.to_owned().into());
state.signing_key = Some(old_signing_key());
state.last_message_time = Some(101);
state.last_identity_time = Some(102);
state.last_account_time = Some(103);
state.last_updated_at = 104;
state
}
struct Fixture {
hydrant: Hydrant,
stub: StubIdentity,
did: Did,
_tmp: tempfile::TempDir,
}
impl Fixture {
async fn new(doc: String) -> miette::Result<Self> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let (local_base, stub) = spawn_identity_server(doc).await;
let public_base = Url::parse("http://public.example/").into_diagnostic()?;
let address = std::net::SocketAddr::new(
std::net::Ipv4Addr::LOCALHOST.into(),
local_base.port().expect("test server URL has a port"),
);
let public = crate::net::PublicHttpClient::test(
reqwest::Client::builder()
.no_proxy()
.resolve("public.example", address)
.build()
.into_diagnostic()?,
);
let config = Config {
database_path: tmp.path().to_path_buf(),
plc_urls: vec![public_base.clone()],
..Default::default()
};
let resolver = Resolver::stubbed_for_test(public_base, 10, public);
let hydrant = Hydrant::new_with_resolver(config, resolver).await?;
Ok(Self {
hydrant,
stub,
did: Did::new_static(DID).unwrap(),
_tmp: tmp,
})
}
fn seed(&self, state: &RepoState<'_>) -> miette::Result<()> {
let mut batch = self.hydrant.state.db.inner.batch();
batch.insert(
&self.hydrant.state.db.repos,
keys::repo_key(&self.did),
db::ser_repo_state(state)?,
);
batch.insert(
&self.hydrant.state.db.repo_metadata,
keys::repo_metadata_key(&self.did),
db::ser_repo_meta(&RepoMetadata::backfilling(7))?,
);
batch.commit().into_diagnostic()
}
fn seed_gone(&self, status: RepoStatus) -> miette::Result<()> {
self.hydrant
.state
.db
.indexer
.resync
.insert(
keys::repo_key(&self.did),
rmp_serde::to_vec(&ResyncState::Gone { status }).into_diagnostic()?,
)
.into_diagnostic()
}
fn seed_refetches(&self, used: u32) {
let mut metadata = RepoMetadata::backfilling(7);
metadata.refetch = Some(crate::types::Allowance {
since: chrono::Utc::now().timestamp(),
used,
});
crate::ingest::indexer::refetch::tests::put_metadata(
&self.hydrant.state,
&self.did,
&metadata,
);
}
fn host_refetches(&self) -> miette::Result<Option<u32>> {
let bytes = self
.hydrant
.state
.db
.cursors
.get(keys::refetch_allowance_key("old-pds.example"))
.into_diagnostic()?;
bytes
.map(|bytes| {
rmp_serde::from_slice::<crate::types::Allowance>(&bytes)
.map(|allowance| allowance.used)
.into_diagnostic()
})
.transpose()
}
fn seed_pending(&self) -> miette::Result<()> {
let db = &self.hydrant.state.db;
let mut batch = db.inner.batch();
db.indexer
.stage_pending_insert(&mut batch, &self.did, &keys::pending_key(7));
batch.commit().into_diagnostic()
}
fn durable_state(&self) -> miette::Result<RepoState<'static>> {
let bytes = self
.hydrant
.state
.db
.repos
.get(keys::repo_key(&self.did))
.into_diagnostic()?
.expect("durable repository state");
Ok(db::deser_repo_state(&bytes)?.into_static())
}
fn repo_bytes(&self) -> miette::Result<Vec<u8>> {
Ok(self
.hydrant
.state
.db
.repos
.get(keys::repo_key(&self.did))
.into_diagnostic()?
.expect("durable repository state")
.to_vec())
}
fn pending_rows(&self) -> miette::Result<Vec<(Vec<u8>, Vec<u8>)>> {
self.hydrant
.state
.db
.indexer
.pending_iter()
.map(|entry| {
let (key, value) = entry.into_inner().into_diagnostic()?;
Ok((key.to_vec(), value.to_vec()))
})
.collect()
}
fn resync_rows(&self) -> miette::Result<Vec<(Vec<u8>, Vec<u8>)>> {
self.hydrant
.state
.db
.indexer
.resync
.iter()
.map(|entry| {
let (key, value) = entry.into_inner().into_diagnostic()?;
Ok((key.to_vec(), value.to_vec()))
})
.collect()
}
}
fn mini_doc_result<T>(result: Result<T, MiniDocError>) -> miette::Result<T> {
result.map_err(|error| miette::miette!("{error}"))
}
fn identity_events(
receiver: &mut tokio::sync::broadcast::Receiver<BroadcastEvent>,
) -> Vec<IdentityEvt> {
let mut events = Vec::new();
loop {
match receiver.try_recv() {
Ok(BroadcastEvent::Ephemeral(event)) => {
events.push(event.identity.expect("identity event"));
}
Ok(other) => panic!("unexpected broadcast event: {other:?}"),
Err(tokio::sync::broadcast::error::TryRecvError::Empty) => break,
Err(error) => panic!("could not receive broadcast event: {error}"),
}
}
events
}
fn assert_resolved_identity(state: &RepoState<'_>, handle: &str, pds: &str) {
assert_eq!(state.handle.as_ref().map(Handle::as_str), Some(handle));
assert_eq!(state.pds.as_deref(), Some(pds));
assert_eq!(state.signing_key.as_ref(), Some(&resolved_signing_key()));
}
async fn finish_backfill(fixture: &Fixture) -> miette::Result<()> {
let [(pending_key, _)] = fixture
.pending_rows()?
.try_into()
.expect("one queued backfill");
let (tx, shard) = FirehoseWorker::spawn_one_shard(fixture.hydrant.state.clone());
let did = fixture.did.clone();
let drained = tx
.on_owner(&fixture.did, move |state, shard| {
FirehoseWorker::finish_for_test(state, shard, &did, &pending_key)
})
.await??;
drop(tx);
timeout(
Duration::from_secs(5),
tokio::task::spawn_blocking(move || shard.join()),
)
.await
.expect("indexer shard should stop")
.into_diagnostic()?
.map_err(|_| miette::miette!("indexer shard panicked"))?;
assert!(drained, "the queued repo wasn't drained");
Ok(())
}
#[tokio::test]
async fn stale_requested_handle_repairs_durable_identity_and_mini_doc() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
let mut initial = complete_state(OLD_HANDLE, "https://old-pds.example/");
initial.root = Some(crate::types::Commit {
version: 3,
rev: crate::db::types::DbTid::from(
&jacquard_common::types::string::Tid::new("3jzfcijpj2z2a").into_diagnostic()?,
),
data: crate::car::compute_cid(b"root"),
prev: None,
sig: bytes::Bytes::from_static(b"signature"),
});
fixture.seed(&initial)?;
let requested = Handle::new_static(NEW_HANDLE).unwrap();
let mut events = fixture.hydrant.state.db.stream.event_tx.subscribe();
let asked_at = chrono::Utc::now().timestamp();
let mini = mini_doc_result(
fixture
.hydrant
.repos
.get(&fixture.did)
.mini_doc_for(Some(&requested))
.await,
)?;
assert_eq!(mini.handle.as_str(), NEW_HANDLE);
assert_eq!(mini.pds.as_str(), "https://new-pds.example/");
assert_eq!(mini.signing_key, resolved_signing_key());
let durable = fixture.durable_state()?;
assert_resolved_identity(&durable, NEW_HANDLE, "https://new-pds.example/");
assert_eq!(durable.status, RepoStatus::Synced);
assert_eq!(durable.last_message_time, Some(101));
assert_eq!(durable.last_identity_time, Some(102));
assert_eq!(durable.last_account_time, Some(103));
assert_eq!(durable.last_updated_at, 104);
assert_eq!(durable.root.as_ref().map(|root| root.version), Some(3));
assert!(
durable
.identity_resolved_at
.is_some_and(|at| at >= asked_at),
"{:?}",
durable.identity_resolved_at
);
assert_eq!(fixture.stub.doc_requests(), 1);
assert_eq!(fixture.stub.handle_requests(), 1);
let events = identity_events(&mut events);
assert_eq!(events.len(), 1);
assert_eq!(events[0].did, fixture.did);
assert_eq!(
events[0].handle.as_ref().map(Handle::as_str),
Some(NEW_HANDLE)
);
Ok(())
}
#[tokio::test]
async fn missing_pds_and_signing_key_each_trigger_reconciliation() -> miette::Result<()> {
for missing_pds in [true, false] {
let fixture = Fixture::new(did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
let mut state = complete_state(NEW_HANDLE, "https://old-pds.example/");
if missing_pds {
state.pds = None;
} else {
state.signing_key = None;
}
fixture.seed(&state)?;
let mini = mini_doc_result(fixture.hydrant.repos.get(&fixture.did).mini_doc().await)?;
assert_eq!(mini.handle.as_str(), NEW_HANDLE);
assert_eq!(mini.pds.as_str(), "https://new-pds.example/");
assert_resolved_identity(
&fixture.durable_state()?,
NEW_HANDLE,
"https://new-pds.example/",
);
assert_eq!(fixture.stub.doc_requests(), 1);
}
Ok(())
}
#[tokio::test]
async fn repeated_deactivated_reads_do_not_refresh_did_doc() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(OLD_HANDLE, "https://old-pds.example")).await?;
let mut state = complete_state(OLD_HANDLE, "https://old-pds.example/");
state.status = RepoStatus::Deactivated;
state.signing_key = Some(resolved_signing_key());
fixture.seed(&state)?;
for _ in 0..5 {
let mini = mini_doc_result(fixture.hydrant.repos.get(&fixture.did).mini_doc().await)?;
assert_eq!(mini.handle.as_str(), OLD_HANDLE);
}
assert_eq!(fixture.stub.doc_requests(), 0);
Ok(())
}
#[tokio::test]
async fn repeated_keyless_reads_refresh_did_doc_once_per_cool_off() -> miette::Result<()> {
let fixture = Fixture::new(keyless_did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
let mut state = complete_state(NEW_HANDLE, "https://new-pds.example/");
state.signing_key = None;
fixture.seed(&state)?;
for _ in 0..5 {
let result = fixture.hydrant.repos.get(&fixture.did).mini_doc().await;
assert!(matches!(result, Err(MiniDocError::CouldNotResolveIdentity)));
}
assert_eq!(fixture.stub.doc_requests(), 1);
Ok(())
}
#[tokio::test]
async fn repeated_pdsless_reads_refresh_did_doc_once_per_cool_off() -> miette::Result<()> {
let fixture = Fixture::new(pdsless_did_doc(NEW_HANDLE)).await?;
let mut state = complete_state(NEW_HANDLE, "https://old-pds.example/");
state.pds = None;
state.signing_key = Some(resolved_signing_key());
fixture.seed(&state)?;
for _ in 0..5 {
let result = fixture.hydrant.repos.get(&fixture.did).mini_doc().await;
assert!(matches!(result, Err(MiniDocError::CouldNotResolveIdentity)));
}
assert_eq!(fixture.stub.doc_requests(), 1);
Ok(())
}
#[tokio::test]
async fn pdsless_gone_recovery_queues_from_cooldown_without_refetch() -> miette::Result<()> {
let fixture = Fixture::new(pdsless_did_doc(NEW_HANDLE)).await?;
let mut state = complete_state(NEW_HANDLE, "https://stored-pds.example/");
state.status = RepoStatus::Deactivated;
state.signing_key = Some(resolved_signing_key());
fixture.seed(&state)?;
fixture.seed_gone(RepoStatus::Deactivated)?;
let first = fixture.hydrant.repos.get(&fixture.did).mini_doc().await;
assert!(matches!(first, Err(MiniDocError::CouldNotResolveIdentity)));
assert!(fixture.pending_rows()?.is_empty());
assert_eq!(fixture.resync_rows()?.len(), 1);
assert_eq!(fixture.stub.doc_requests(), 1);
let mini = mini_doc_result(fixture.hydrant.repos.get(&fixture.did).mini_doc().await)?;
assert_eq!(mini.pds.as_str(), "https://stored-pds.example/");
assert_eq!(fixture.pending_rows()?.len(), 1);
assert!(fixture.resync_rows()?.is_empty());
assert_eq!(fixture.stub.doc_requests(), 1);
let repeat = fixture.hydrant.repos.get(&fixture.did).mini_doc().await;
assert!(matches!(repeat, Err(MiniDocError::NotSynced)));
assert_eq!(fixture.pending_rows()?.len(), 1);
assert_eq!(fixture.stub.doc_requests(), 1);
Ok(())
}
#[tokio::test]
async fn fresh_fetch_failure_leaves_state_queues_and_events_unchanged() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
fixture.seed(&complete_state(OLD_HANDLE, "https://old-pds.example/"))?;
fixture.seed_gone(RepoStatus::Deactivated)?;
fixture.stub.fail_docs().await;
let before_repo = fixture.repo_bytes()?;
let before_pending = fixture.pending_rows()?;
let before_resync = fixture.resync_rows()?;
let before_event_id = fixture.hydrant.state.db.stream.ids.next();
let mut events = fixture.hydrant.state.db.stream.event_tx.subscribe();
let result = resolve_mini_doc(&fixture.hydrant, &AtIdentifier::Did(fixture.did.clone()))
.await;
let Err(error) = result else {
panic!("failed fresh fetch should return the upstream error");
};
assert_eq!(error.status, StatusCode::BAD_GATEWAY);
assert_eq!(
serde_json::to_value(error.error).into_diagnostic()?["error"],
"UpstreamError"
);
assert_eq!(fixture.repo_bytes()?, before_repo);
assert_eq!(fixture.pending_rows()?, before_pending);
assert_eq!(fixture.resync_rows()?, before_resync);
assert_eq!(fixture.hydrant.state.db.stream.ids.next(), before_event_id);
assert!(identity_events(&mut events).is_empty());
assert_eq!(fixture.stub.doc_requests(), 1);
Ok(())
}
#[tokio::test]
async fn deleted_repo_never_reconciles_or_recovers_gone() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
let mut deleted = complete_state(OLD_HANDLE, "https://old-pds.example/");
deleted.status = RepoStatus::Deleted;
fixture.seed(&deleted)?;
fixture.seed_gone(RepoStatus::Deactivated)?;
let before = fixture.repo_bytes()?;
let before_resync = fixture.resync_rows()?;
let requested = Handle::new_static(NEW_HANDLE).unwrap();
let mut events = fixture.hydrant.state.db.stream.event_tx.subscribe();
let mini = mini_doc_result(
fixture
.hydrant
.repos
.get(&fixture.did)
.mini_doc_for(Some(&requested))
.await,
)?;
assert_eq!(mini.handle.as_str(), OLD_HANDLE);
assert_eq!(fixture.repo_bytes()?, before);
assert!(fixture.pending_rows()?.is_empty());
assert_eq!(fixture.resync_rows()?, before_resync);
assert!(identity_events(&mut events).is_empty());
assert_eq!(fixture.stub.doc_requests(), 0);
Ok(())
}
#[tokio::test]
async fn gone_recovery_queues_once_and_backfill_finished_syncs() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
let mut gone = complete_state(OLD_HANDLE, "https://old-pds.example/");
gone.status = RepoStatus::Deactivated;
fixture.seed(&gone)?;
fixture.seed_gone(RepoStatus::Deactivated)?;
let mut events = fixture.hydrant.state.db.stream.event_tx.subscribe();
let mini = timeout(
Duration::from_secs(5),
fixture.hydrant.repos.get(&fixture.did).mini_doc(),
)
.await
.expect("gone recovery must not self-deadlock");
let mini = mini_doc_result(mini)?;
assert_eq!(mini.handle.as_str(), NEW_HANDLE);
assert_eq!(mini.pds.as_str(), "https://new-pds.example/");
let pending = fixture.pending_rows()?;
assert_eq!(pending.len(), 1);
assert_eq!(pending[0].1, keys::repo_key(&fixture.did));
assert!(fixture.resync_rows()?.is_empty());
assert_eq!(fixture.durable_state()?.status, RepoStatus::Deactivated);
let first_events = identity_events(&mut events);
assert_eq!(first_events.len(), 1);
assert_eq!(fixture.stub.doc_requests(), 1);
let repeat = fixture.hydrant.repos.get(&fixture.did).mini_doc().await;
assert!(matches!(repeat, Err(MiniDocError::NotSynced)));
assert_eq!(fixture.pending_rows()?, pending);
assert!(fixture.resync_rows()?.is_empty());
assert!(identity_events(&mut events).is_empty());
assert_eq!(fixture.stub.doc_requests(), 1);
finish_backfill(&fixture).await?;
assert_eq!(fixture.durable_state()?.status, RepoStatus::Synced);
assert!(fixture.pending_rows()?.is_empty());
Ok(())
}
#[tokio::test]
async fn gone_recovery_with_no_refetch_left_queues_and_charges_nothing() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
let mut gone = complete_state(OLD_HANDLE, "https://old-pds.example/");
gone.status = RepoStatus::Deactivated;
gone.active = false;
fixture.seed(&gone)?;
fixture.seed_gone(RepoStatus::Deactivated)?;
fixture.seed_refetches(6);
let resync = fixture.resync_rows()?;
for _ in 0..3 {
let _ = fixture.hydrant.repos.get(&fixture.did).mini_doc().await;
}
assert!(fixture.pending_rows()?.is_empty());
assert_eq!(fixture.resync_rows()?, resync);
assert_eq!(
fixture.host_refetches()?,
None,
"a denial costs the host nothing"
);
Ok(())
}
#[tokio::test]
async fn a_refetch_gone_recovery_is_denied_keeps_the_repo_parked() -> miette::Result<()> {
let fixture = Fixture::new(pdsless_did_doc(NEW_HANDLE)).await?;
// `active` stays true, an inactive repo gets denied without being parked
let mut state = complete_state(NEW_HANDLE, "https://old-pds.example/");
state.status = RepoStatus::Deactivated;
state.signing_key = Some(resolved_signing_key());
fixture.seed(&state)?;
fixture.seed_gone(RepoStatus::Deactivated)?;
fixture.seed_refetches(6);
// the first read starts the cooldown, the second recovers from what's stored
for _ in 0..2 {
let _ = fixture.hydrant.repos.get(&fixture.did).mini_doc().await;
}
assert!(fixture.pending_rows()?.is_empty());
assert_eq!(fixture.durable_state()?.status, RepoStatus::Throttled);
let [(_, entry)] = fixture.resync_rows()?.try_into().expect("one resync entry");
assert!(matches!(
rmp_serde::from_slice::<ResyncState>(&entry).into_diagnostic()?,
ResyncState::Error {
kind: crate::types::ResyncErrorKind::Ratelimited,
..
}
));
Ok(())
}
#[tokio::test]
async fn gone_recovery_charges_its_host_once() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
let mut gone = complete_state(OLD_HANDLE, "https://old-pds.example/");
gone.status = RepoStatus::Deactivated;
fixture.seed(&gone)?;
fixture.seed_gone(RepoStatus::Deactivated)?;
let _ = fixture.hydrant.repos.get(&fixture.did).mini_doc().await;
assert_eq!(fixture.pending_rows()?.len(), 1);
assert_eq!(fixture.host_refetches()?, Some(1));
let _ = fixture.hydrant.repos.get(&fixture.did).mini_doc().await;
assert_eq!(
fixture.host_refetches()?,
Some(1),
"asking again is no new fetch"
);
Ok(())
}
/// the stamp is what tells the relay a stored key is fresh, so a fetch that changed nothing
/// still moves it
#[tokio::test]
async fn a_fetch_that_finds_the_same_identity_still_stamps_it() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
let mut gone = complete_state(NEW_HANDLE, "https://new-pds.example/");
gone.signing_key = Some(resolved_signing_key());
gone.status = RepoStatus::Deactivated;
fixture.seed(&gone)?;
fixture.seed_gone(RepoStatus::Deactivated)?;
let mut events = fixture.hydrant.state.db.stream.event_tx.subscribe();
let asked_at = chrono::Utc::now().timestamp();
mini_doc_result(fixture.hydrant.repos.get(&fixture.did).mini_doc().await)?;
assert_eq!(fixture.stub.doc_requests(), 1);
let durable = fixture.durable_state()?;
assert_eq!(durable.identity(), gone.identity());
assert!(
durable
.identity_resolved_at
.is_some_and(|at| at >= asked_at),
"{:?}",
durable.identity_resolved_at
);
assert!(identity_events(&mut events).is_empty());
Ok(())
}
#[tokio::test]
async fn a_mini_doc_writes_on_the_repos_indexer_shard() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
let mut gone = complete_state(OLD_HANDLE, "https://old-pds.example/");
gone.status = RepoStatus::Deactivated;
fixture.seed(&gone)?;
fixture.seed_gone(RepoStatus::Deactivated)?;
let before = fixture.repo_bytes()?;
// the repo's shard never runs, so a write that goes through it never happens
let _shards = crate::ingest::indexer::FirehoseWorker::serve_shards(
fixture.hydrant.state.clone(),
1,
Some(0),
);
let mini = fixture.hydrant.repos.get(&fixture.did).mini_doc().await;
assert!(matches!(mini, Err(MiniDocError::Other(_))));
assert_eq!(fixture.repo_bytes()?, before);
assert!(fixture.pending_rows()?.is_empty());
assert_eq!(fixture.resync_rows()?.len(), 1);
Ok(())
}
#[tokio::test]
async fn backfill_finished_consumes_pending_gone_row_before_next_read() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(OLD_HANDLE, "https://old-pds.example")).await?;
let mut state = complete_state(OLD_HANDLE, "https://old-pds.example/");
state.status = RepoStatus::Deactivated;
state.signing_key = Some(resolved_signing_key());
fixture.seed(&state)?;
fixture.seed_pending()?;
fixture.seed_gone(RepoStatus::Deactivated)?;
assert_eq!(fixture.pending_rows()?.len(), 1);
assert_eq!(fixture.resync_rows()?.len(), 1);
finish_backfill(&fixture).await?;
assert_eq!(fixture.durable_state()?.status, RepoStatus::Synced);
assert!(fixture.pending_rows()?.is_empty());
assert!(fixture.resync_rows()?.is_empty());
let mini = mini_doc_result(fixture.hydrant.repos.get(&fixture.did).mini_doc().await)?;
assert_eq!(mini.handle.as_str(), OLD_HANDLE);
assert!(fixture.pending_rows()?.is_empty());
assert!(fixture.resync_rows()?.is_empty());
assert_eq!(fixture.stub.doc_requests(), 0);
Ok(())
}
#[tokio::test]
async fn raced_fresh_fetch_is_discarded() -> miette::Result<()> {
let fixture = Fixture::new(did_doc("cached.test", "https://cached-pds.example")).await?;
fixture.seed(&complete_state(OLD_HANDLE, "https://old-pds.example/"))?;
let cached = fixture.hydrant.resolver().resolve_doc(&fixture.did).await?;
assert_eq!(cached.pds.as_str(), "https://cached-pds.example/");
fixture
.stub
.set_doc(did_doc(NEW_HANDLE, "https://fetched-pds.example"))
.await;
fixture.stub.block_next_doc();
let requested = Handle::new_static(NEW_HANDLE).unwrap();
let repo = fixture.hydrant.repos.get(&fixture.did);
let mut events = fixture.hydrant.state.db.stream.event_tx.subscribe();
let request = tokio::spawn(async move { repo.mini_doc_for(Some(&requested)).await });
fixture.stub.wait_until_doc_started().await;
let winner = complete_state("winner.test", "https://winner-pds.example/");
let winner_for_db = winner.clone();
let did_for_db = fixture.did.clone();
fixture
.hydrant
.state
.db
.run(move |db| {
let mut txn = Txn::new(db);
txn.batch.insert(
&db.repos,
keys::repo_key(&did_for_db),
db::ser_repo_state(&winner_for_db)?,
);
txn.commit()
})
.await?;
crate::ops::announce_identity(
&fixture.hydrant.state.db,
&fixture.did,
winner.handle.clone(),
)
.await;
fixture.stub.release_doc();
let mini = mini_doc_result(request.await.into_diagnostic()?)?;
assert_eq!(mini.handle.as_str(), "winner.test");
let durable = fixture.durable_state()?;
assert_eq!(
durable.handle.as_ref().map(Handle::as_str),
Some("winner.test")
);
assert_eq!(durable.pds.as_deref(), Some("https://winner-pds.example/"));
assert_eq!(durable.signing_key, winner.signing_key);
let events = identity_events(&mut events);
assert_eq!(events.len(), 1);
assert_eq!(
events[0].handle.as_ref().map(Handle::as_str),
Some("winner.test")
);
assert_eq!(fixture.stub.doc_requests(), 2);
let still_cached = fixture.hydrant.resolver().resolve_doc(&fixture.did).await?;
assert_eq!(still_cached.pds.as_str(), "https://cached-pds.example/");
assert_eq!(fixture.stub.doc_requests(), 2);
Ok(())
}
#[tokio::test]
async fn complete_did_only_read_uses_cached_identity_without_refresh() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(OLD_HANDLE, "https://old-pds.example")).await?;
fixture.seed(&complete_state(OLD_HANDLE, "https://old-pds.example/"))?;
let before = fixture.repo_bytes()?;
fixture.hydrant.resolver().resolve_doc(&fixture.did).await?;
assert_eq!(fixture.stub.doc_requests(), 1);
fixture.stub.fail_docs().await;
let output = resolve_mini_doc(&fixture.hydrant, &AtIdentifier::Did(fixture.did.clone()))
.await
.map_err(|error| miette::miette!("DID-only read failed: {}", error.status))?;
assert_eq!(output.handle.as_str(), OLD_HANDLE);
assert_eq!(output.pds.as_str(), "https://old-pds.example/");
assert_eq!(fixture.repo_bytes()?, before);
assert_eq!(fixture.stub.doc_requests(), 1);
Ok(())
}
#[tokio::test]
async fn describe_repo_reconciles_stale_handle() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
fixture.seed(&complete_state(OLD_HANDLE, "https://old-pds.example/"))?;
let requested = Handle::new_static(NEW_HANDLE).unwrap();
let result = super::super::describe_repo::handle(
State(fixture.hydrant.clone()),
ExtractXrpc(super::super::describe_repo::DescribeRepoRequestData {
identifier: AtIdentifier::Handle(requested),
}),
)
.await;
let Ok(Json(output)) = result else {
panic!("describeRepo should reconcile a stale requested handle");
};
assert_eq!(output.did, fixture.did);
assert_eq!(output.handle.as_str(), NEW_HANDLE);
assert_eq!(output.pds.as_str(), "https://new-pds.example/");
assert_eq!(output.signing_key, resolved_signing_key());
assert_resolved_identity(
&fixture.durable_state()?,
NEW_HANDLE,
"https://new-pds.example/",
);
assert_eq!(fixture.stub.doc_requests(), 1);
assert_eq!(fixture.stub.handle_requests(), 2);
Ok(())
}
async fn http_mini_doc(
fixture: &Fixture,
identifier: AtIdentifier,
) -> XrpcResult<ResolveMiniDocOutput> {
resolve_mini_doc(&fixture.hydrant, &identifier)
.await
}
fn assert_upstream_error(result: XrpcResult<ResolveMiniDocOutput>) -> miette::Result<()> {
let Err(error) = result else {
panic!("an unresolvable identity should be an upstream error");
};
assert_eq!(error.status, StatusCode::BAD_GATEWAY);
assert_eq!(
serde_json::to_value(error.error).into_diagnostic()?["error"],
"UpstreamError"
);
Ok(())
}
fn has_repo_row(fixture: &Fixture) -> miette::Result<bool> {
let row = fixture
.hydrant
.state
.db
.repos
.get(keys::repo_key(&fixture.did))
.into_diagnostic()?;
Ok(row.is_some())
}
#[tokio::test]
async fn untracked_did_answers_from_the_resolver() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
let output = http_mini_doc(&fixture, AtIdentifier::Did(fixture.did.clone()))
.await
.map_err(|error| miette::miette!("untracked did refused: {}", error.status))?;
assert_eq!(output.did, fixture.did);
assert_eq!(output.handle.as_str(), NEW_HANDLE);
assert_eq!(output.pds.as_str(), "https://new-pds.example/");
assert_eq!(output.signing_key, resolved_signing_key());
assert!(!has_repo_row(&fixture)?);
assert_eq!(fixture.stub.doc_requests(), 1);
Ok(())
}
#[tokio::test]
async fn untracked_handle_identifier_answers_with_the_verified_handle() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
let requested = Handle::new_static(NEW_HANDLE).unwrap();
let output = http_mini_doc(&fixture, AtIdentifier::Handle(requested))
.await
.map_err(|error| miette::miette!("untracked handle refused: {}", error.status))?;
assert_eq!(output.did, fixture.did);
assert_eq!(output.handle.as_str(), NEW_HANDLE);
assert!(!has_repo_row(&fixture)?);
Ok(())
}
#[tokio::test]
async fn untracked_did_with_an_unverifiable_handle_gets_handle_invalid() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(LIAR_HANDLE, "https://new-pds.example")).await?;
let output = http_mini_doc(&fixture, AtIdentifier::Did(fixture.did.clone()))
.await
.map_err(|error| miette::miette!("untracked did refused: {}", error.status))?;
assert_eq!(output.handle.as_str(), "handle.invalid");
assert_eq!(output.pds.as_str(), "https://new-pds.example/");
assert_eq!(output.signing_key, resolved_signing_key());
Ok(())
}
#[tokio::test]
async fn not_synced_repo_answers_from_the_resolver() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
fixture.seed(&complete_state(OLD_HANDLE, "https://old-pds.example/"))?;
fixture.seed_pending()?;
let before = fixture.repo_bytes()?;
let before_pending = fixture.pending_rows()?;
let output = http_mini_doc(&fixture, AtIdentifier::Did(fixture.did.clone()))
.await
.map_err(|error| miette::miette!("pending repo refused: {}", error.status))?;
assert_eq!(output.handle.as_str(), NEW_HANDLE);
assert_eq!(output.pds.as_str(), "https://new-pds.example/");
assert_eq!(fixture.repo_bytes()?, before);
assert_eq!(fixture.pending_rows()?, before_pending);
Ok(())
}
#[tokio::test]
async fn unresolvable_untracked_did_is_an_upstream_error() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
fixture.stub.fail_docs().await;
let result = http_mini_doc(&fixture, AtIdentifier::Did(fixture.did.clone())).await;
assert_upstream_error(result)?;
assert!(!has_repo_row(&fixture)?);
Ok(())
}
#[tokio::test]
async fn untracked_did_without_a_signing_key_is_an_upstream_error() -> miette::Result<()> {
let fixture = Fixture::new(keyless_did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
let result = http_mini_doc(&fixture, AtIdentifier::Did(fixture.did.clone())).await;
assert_upstream_error(result)
}
#[tokio::test]
async fn untracked_did_uses_a_cached_resolver_doc_without_fetching() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
fixture.hydrant.resolver().resolve_doc(&fixture.did).await?;
assert_eq!(fixture.stub.doc_requests(), 1);
fixture.stub.fail_docs().await;
let output = http_mini_doc(&fixture, AtIdentifier::Did(fixture.did.clone()))
.await
.map_err(|error| miette::miette!("cached did refused: {}", error.status))?;
assert_eq!(output.pds.as_str(), "https://new-pds.example/");
assert_eq!(fixture.stub.doc_requests(), 1);
Ok(())
}
#[tokio::test]
async fn synced_repo_still_answers_from_its_own_state() -> miette::Result<()> {
let fixture = Fixture::new(did_doc(NEW_HANDLE, "https://new-pds.example")).await?;
let mut state = complete_state(OLD_HANDLE, "https://old-pds.example/");
state.signing_key = Some(resolved_signing_key());
fixture.seed(&state)?;
let output = http_mini_doc(&fixture, AtIdentifier::Did(fixture.did.clone()))
.await
.map_err(|error| miette::miette!("synced repo refused: {}", error.status))?;
assert_eq!(output.handle.as_str(), OLD_HANDLE);
assert_eq!(output.pds.as_str(), "https://old-pds.example/");
assert_eq!(fixture.stub.doc_requests(), 0);
Ok(())
}
}