~ knotless Search/
↑↓ move↵ open⇧↵ all results

chadtmiller.com/knotless
Switch repository
Loading…

git that lives in your Atmosphere account

knotless.dev
1 collaborator

knotless 0.13.0 Latest 2 packages, 1 image

1
10

knotless-serve: An account is dropped as deleted only when its host says so

Merged chadtmiller.com asked to merge into main from chadtmiller.com/knotless:serve/account-deleted
crates/knotless-serve/src/accounts.rs+132−15
View file
@@ -4,11 +4,22 @@
//! with what was indexed of it kept. One deleted is gone for good: what
//! this server indexed from it, the mirrors of its repositories and the
//! sessions it kept for it go with it.
//!
//! An event saying an account was deleted is not taken at its word: the
//! host its identity names now is asked first ([`gone`]), and only an
//! answer that says so drops anything. One that cannot be had, PLC or the
//! host down or answering oddly, is asked for again, then the account kept
//! ([`settle`]), since nothing dropped of it would come back.
use crate::App;
use crate::telemetry;
use knotless::env::Ctx;
use knotless::telemetry::Level;
use knotless::xrpc::{Xrpc, XrpcError};
use serde_json::Value;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
/// Takes in that `did`'s host says `status` of it.
pub(crate) fn changed(app: &Arc<App>, did: &str, status: &str) {
@@ -19,25 +30,131 @@ pub(crate) fn changed(app: &Arc<App>, did: &str, status: &str) {
return;
}
let (app, did) = (app.clone(), did.to_string());
tokio::spawn(async move {
if deleted(&app.ctx, &did).await {
purge(&app, &did);
}
});
tokio::spawn(async move { settle(&app, &did, Patience::default()).await });
}
/// Whether `did` is gone from the host its identity now names: one the
/// host still answers for as active moved there, and was deleted only
/// where it was before.
async fn deleted(ctx: &Ctx, did: &str) -> bool {
let Ok(id) = knotless::identity::resolve(ctx, did).await else {
return true;
/// What the host `did`'s identity names now says of it.
#[derive(Debug, Clone, PartialEq)]
pub enum Asked {
/// Gone: the host has no such repo (`RepoNotFound`) or says it was
/// deleted (`active: false`, `status: deleted`), or the identity itself
/// is gone (a PLC tombstone, `410`, or nothing there, `404`).
Gone,
/// The host has it: active, as it is where it moved to, or inactive in
/// a way that can be undone (deactivated, suspended, taken down).
Active,
/// Nothing that says either: the identity or the host could not be
/// read, or answered with what says neither. Why, for the log.
Unknown(String),
}
/// Asks after `did`, said to be deleted, where its identity says it lives
/// now ([`Asked`]): one that moved was deleted only where it was before.
pub async fn gone(ctx: &Ctx, did: &str) -> Asked {
let doc = match knotless::identity::document(ctx, did).await {
Ok(doc) => doc,
Err(e) if matches!(http_status(&e), Some(404 | 410)) => return Asked::Gone,
Err(e) => return Asked::Unknown(format!("{e:#}")),
};
let Some(pds) = knotless::identity::service(&doc, "#atproto_pds") else {
return Asked::Unknown(format!("{did} names no #atproto_pds service"));
};
let xrpc = knotless::xrpc::Xrpc::new(ctx.http.clone(), &id.pds);
let xrpc = Xrpc::new(ctx.http.clone(), &pds);
let asked = [("did", did)];
let status = xrpc.query::<serde_json::Value>("com.atproto.sync.getRepoStatus", &asked, None);
let active = status.await.ok().and_then(|s| s["active"].as_bool());
active != Some(true)
let status = xrpc.query::<Value>("com.atproto.sync.getRepoStatus", &asked, None);
match status.await {
Ok(status) => match (status["active"].as_bool(), status["status"].as_str()) {
(Some(true), _) => Asked::Active,
(Some(false), Some("deleted")) => Asked::Gone,
// Deactivated, suspended or taken down, which can be undone:
// the account is still there.
(Some(false), why) => {
let said = "an account inactive at its host, not deleted";
telemetry::account_asked(Level::INFO, said, did, why, None, None);
Asked::Active
}
(None, _) => Asked::Unknown(format!("{pds} answered {status}")),
},
Err(e) => match XrpcError::of(&e) {
Some(refused) if refused.error == "RepoNotFound" => Asked::Gone,
_ => Asked::Unknown(format!("{e:#}")),
},
}
}
/// The HTTP status a request in `e` was answered with, if it was.
fn http_status(e: &anyhow::Error) -> Option<u16> {
let answered = e.chain().find_map(|c| c.downcast_ref::<reqwest::Error>());
answered
.and_then(reqwest::Error::status)
.map(|s| s.as_u16())
}
/// How long to keep asking after an account no answer has settled.
#[derive(Debug, Clone, Copy)]
pub struct Patience {
/// The wait before asking again, doubled each time.
pub first: Duration,
/// The longest wait.
pub most: Duration,
/// How many times to ask in all.
pub tries: u32,
}
impl Default for Patience {
/// Six asks over about half an hour, the last waits capped at ten
/// minutes: long enough for PLC or a host to come back.
fn default() -> Patience {
Patience {
first: Duration::from_secs(30),
most: Duration::from_secs(600),
tries: 6,
}
}
}
/// What came of an account said to be deleted.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum Settled {
/// Its host has it: it stays.
Kept,
/// Gone: everything held from it was dropped ([`purge`]).
Purged,
/// No answer said, however often asked: it stays, as nothing dropped
/// of it would come back.
GaveUp,
}
/// Asks after `did` until an answer says, waiting longer each time, and
/// drops what is held from it only if it is [`Asked::Gone`].
pub async fn settle(app: &App, did: &str, patience: Patience) -> Settled {
let mut wait = patience.first;
for tried in 1..=patience.tries {
match gone(&app.ctx, did).await {
Asked::Active => {
let said = "an account said to be deleted is there at its host";
telemetry::account_asked(Level::INFO, said, did, None, Some(tried), None);
return Settled::Kept;
}
Asked::Gone => {
purge(app, did);
let said = "an account was deleted";
telemetry::account_asked(Level::INFO, said, did, None, Some(tried), None);
return Settled::Purged;
}
Asked::Unknown(why) if tried < patience.tries => {
let said = "an account said to be deleted, asked after again later";
telemetry::account_asked(Level::WARN, said, did, None, Some(tried), Some(why));
tokio::time::sleep(wait).await;
wait = (wait * 2).min(patience.most);
}
Asked::Unknown(why) => {
let said = "an account said to be deleted, never settled: kept";
telemetry::account_asked(Level::ERROR, said, did, None, Some(tried), Some(why));
}
}
}
Settled::GaveUp
}
/// Drops everything this server holds from `did`, deleted: what both
crates/knotless-serve/src/lib.rs+1−1
View file
@@ -14,7 +14,7 @@
//! TTL, or kept warm by the firehose or Jetstream. Every object it serves is named by its
//! hash, so a client can check it.
mod accounts;
pub mod accounts;
mod api;
mod appview;
pub mod card;
crates/knotless-serve/src/telemetry.rs+21−0
View file
@@ -239,6 +239,27 @@ pub fn feed(name: &str, url: &str, connected: bool) {
telemetry::note(level, said, &fields);
}
/// Logs `said` at `level` of `did`, an account said to be deleted, as it
/// was asked after ([`crate::accounts::settle`]): the status its host
/// holds it at, the asks so far, and why none settled it.
pub fn account_asked(
level: Level,
said: &str,
did: &str,
status: Option<&str>,
tried: Option<u32>,
why: Option<String>,
) {
let fields = Fields {
did: Some(did),
status,
count: tried.map(u64::from),
error: why,
..Fields::default()
};
telemetry::note(level, said, &fields);
}
/// Logs, at warn, an event of `did`'s the index did not take: one the
/// index will miss until a sweep fills it.
pub fn indexed<T, E: std::fmt::Display>(did: &str, done: Result<T, E>) {
crates/knotless-serve/tests/accounts.rsnew+179−0
View file
@@ -0,0 +1,179 @@
//! Asking after an account said to be deleted, against a stand-in PLC
//! directory that is also each account's host: only an answer that says
//! so is gone, and one that cannot be had is asked again, then kept.
use axum::extract::{Query, State};
use axum::http::{StatusCode, Uri};
use axum::response::{IntoResponse, Response};
use axum::{Json, Router};
use knotless_serve::App;
use knotless_serve::accounts::{self, Asked, Patience, Settled};
use serde_json::{Value, json};
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
/// Each account by what its identity and its host say of it.
const HERE: &str = "did:plc:hereaaaaaaaaaaaaaaaaaaaa";
const DELETED: &str = "did:plc:deletedaaaaaaaaaaaaaaaa";
const DEACTIVATED: &str = "did:plc:deactivatedaaaaaaaaaaaa";
const SUSPENDED: &str = "did:plc:suspendedaaaaaaaaaaaaaa";
const TAKENDOWN: &str = "did:plc:takendownaaaaaaaaaaaaaa";
const NO_REPO: &str = "did:plc:norepoaaaaaaaaaaaaaaaaa";
const TOMBSTONE: &str = "did:plc:tombstoneaaaaaaaaaaaaaa";
const NEVER: &str = "did:plc:neveraaaaaaaaaaaaaaaaaa";
const PLC_DOWN: &str = "did:plc:plcdownaaaaaaaaaaaaaaaa";
const HOST_DOWN: &str = "did:plc:hostdownaaaaaaaaaaaaaaa";
const UNREACHABLE: &str = "did:plc:unreachableaaaaaaaaaaaa";
const ODD: &str = "did:plc:oddaaaaaaaaaaaaaaaaaaaaa";
const NO_HOST: &str = "did:plc:nohostaaaaaaaaaaaaaaaaa";
/// Its PLC fails the first time it is asked, then its host says it was
/// deleted.
const FLAKY: &str = "did:plc:flakyaaaaaaaaaaaaaaaaaa";
struct Stand {
base: String,
/// A port nothing listens on.
closed: String,
asked: AtomicUsize,
}
fn doc(did: &str, pds: &str) -> Value {
json!({
"id": did,
"service": [{
"id": "#atproto_pds",
"type": "AtprotoPersonalDataServer",
"serviceEndpoint": pds,
}],
})
}
/// A status the host holds `did` at, not active.
fn inactive(did: &str, status: &str) -> Response {
Json(json!({ "did": did, "active": false, "status": status })).into_response()
}
async fn answer(
State(stand): State<Arc<Stand>>,
uri: Uri,
Query(q): Query<HashMap<String, String>>,
) -> Response {
let refused = |status: StatusCode, error: &str| {
let body = json!({ "error": error, "message": error });
(status, Json(body)).into_response()
};
if uri.path() == "/xrpc/com.atproto.sync.getRepoStatus" {
let did = q["did"].as_str();
return match did {
HERE => Json(json!({ "did": did, "active": true })).into_response(),
DELETED | FLAKY => inactive(did, "deleted"),
DEACTIVATED => inactive(did, "deactivated"),
SUSPENDED => inactive(did, "suspended"),
TAKENDOWN => inactive(did, "takendown"),
NO_REPO => refused(StatusCode::BAD_REQUEST, "RepoNotFound"),
HOST_DOWN => refused(StatusCode::BAD_GATEWAY, "InternalServerError"),
_ => Json(json!({ "did": did })).into_response(),
};
}
let did = uri.path().trim_start_matches('/');
match did {
TOMBSTONE => refused(StatusCode::GONE, "DID not available"),
PLC_DOWN => refused(StatusCode::SERVICE_UNAVAILABLE, "down"),
FLAKY if stand.asked.fetch_add(1, Ordering::SeqCst) == 0 => {
refused(StatusCode::INTERNAL_SERVER_ERROR, "busy")
}
UNREACHABLE => Json(doc(did, &stand.closed)).into_response(),
NO_HOST => Json(json!({ "id": did, "service": [] })).into_response(),
NEVER => refused(StatusCode::NOT_FOUND, "DID not registered"),
_ => Json(doc(did, &stand.base)).into_response(),
}
}
/// The stand-in, served on `rt`, and an appview that reads identities
/// from it, its cache under `dir`.
fn stand_in(rt: &tokio::runtime::Runtime, dir: &std::path::Path) -> Arc<App> {
let listener = rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let base = format!("http://{}", listener.local_addr().unwrap());
let closed = {
let spare = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
format!("http://{}", spare.local_addr().unwrap())
};
let stand = Arc::new(Stand {
base: base.clone(),
closed,
asked: AtomicUsize::new(0),
});
let router = Router::new().fallback(answer).with_state(stand);
rt.spawn(async move { axum::serve(listener, router).await });
let (config, cache) = (dir.join("config"), dir.join("cache"));
let env = knotless::env::Env::from_vars(|name| match name {
"KNOTLESS_PLC_URL" => Some(base.clone()),
"KNOTLESS_CONFIG_DIR" => Some(config.to_string_lossy().into()),
"KNOTLESS_CACHE_DIR" => Some(cache.to_string_lossy().into()),
"KNOTLESS_APPVIEW" => Some(String::new()),
_ => None,
});
let ctx = knotless::env::Ctx::new(env);
App::build(ctx, Duration::ZERO, None, None)
}
/// Quick to try again, and soon done.
const BRIEF: Patience = Patience {
first: Duration::from_millis(10),
most: Duration::from_millis(20),
tries: 3,
};
#[test]
fn only_an_answer_that_says_so_is_gone() {
let rt = tokio::runtime::Runtime::new().unwrap();
let dir = tempfile::tempdir().unwrap();
let app = stand_in(&rt, dir.path());
let asked = |did: &str| rt.block_on(accounts::gone(&app.ctx, did));
// Active, or inactive in a way that can be undone: still there.
for did in [HERE, DEACTIVATED, SUSPENDED, TAKENDOWN] {
assert_eq!(asked(did), Asked::Active, "{did}");
}
for did in [DELETED, NO_REPO, TOMBSTONE, NEVER] {
assert_eq!(asked(did), Asked::Gone, "{did}");
}
for did in [PLC_DOWN, HOST_DOWN, UNREACHABLE, ODD, NO_HOST] {
let asked = asked(did);
assert!(matches!(asked, Asked::Unknown(_)), "{did}: {asked:?}");
}
}
#[test]
fn an_account_not_settled_is_asked_again_then_kept() {
let rt = tokio::runtime::Runtime::new().unwrap();
let dir = tempfile::tempdir().unwrap();
let app = stand_in(&rt, dir.path());
// A mirror of each, which goes only with an account gone.
let mirrors = |did: &str| {
let dir = did.replace(':', "_");
app.ctx.env.cache_dir.join("mirrors").join(dir)
};
for did in [HERE, DEACTIVATED, FLAKY, PLC_DOWN] {
std::fs::create_dir_all(mirrors(did).join("repo.git")).unwrap();
}
let settle = |did: &str| rt.block_on(accounts::settle(&app, did, BRIEF));
assert_eq!(settle(HERE), Settled::Kept);
// Deactivated where it lives: kept, with its mirrors.
assert_eq!(settle(DEACTIVATED), Settled::Kept);
// PLC fails once, then the host says it was deleted.
assert_eq!(settle(FLAKY), Settled::Purged);
// Never answered: asked each time, then kept.
assert_eq!(settle(PLC_DOWN), Settled::GaveUp);
let kept = [HERE, DEACTIVATED, FLAKY, PLC_DOWN].map(|did| mirrors(did).exists());
assert_eq!(kept, [true, true, false, true]);
}
#[test]
fn patience_is_about_half_an_hour() {
let p = Patience::default();
assert_eq!((p.first.as_secs(), p.most.as_secs(), p.tries), (30, 600, 6));
}
docs/design/serve.md+12−2
View file
@@ -184,8 +184,18 @@ identity names now whether it still has it
(`com.atproto.sync.getRepoStatus`). An account that moved to another host
is deleted from the old one, and a server following that host's
firehose directly hears it; the new host still answers for it, so it
stays. One whose identity no longer resolves is taken at the event's
word.
stays. Only an answer that says the account is gone drops anything:
| What is asked | Answers | The account |
| --- | --- | --- |
| its DID document, from PLC or its `did:web` host | `404` or `410` (a tombstone) | gone |
| `getRepoStatus` at the host the document names | `RepoNotFound`, or `active: false` with `status: deleted` | gone |
| `getRepoStatus` | `active: true`, or inactive for any other reason (deactivated, suspended, taken down) | kept |
| either | anything else: no answer, a timeout, a `5xx`, no host named, no `active` | asked again |
One asked again waits 30 s, then twice as long each time up to 10
minutes, six asks in all (about half an hour, long enough for PLC or a
host to come back), and is kept if none of them says.
Details: