From 6fe83ed02ee4e8473ab83e2123174de6d19dd0fd Mon Sep 17 00:00:00 2001 From: "@permadeath.com" Date: Wed, 9 Sep 2026 10:30:32 -0400 Subject: [PATCH] test(swarm): run the swarm against a server in this process MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Over a real socket, the stream's numbering under concurrent writes, the blob index after records pointing at blobs are deleted, and a frozen repository beside its siblings — each read off the server rather than the swarm, with the swarm's ledger checked against the server's. Co-Authored-By: Claude Fable 5.1 Change-Id: If65c7e43afd927346517d3dcafc4927b977a5e62 --- Cargo.lock | 5 + crates/didbot-swarm/Cargo.toml | 11 +- crates/didbot-swarm/tests/write_load.rs | 413 ++++++++++++++++++++++++ 3 files changed, 426 insertions(+), 3 deletions(-) create mode 100644 crates/didbot-swarm/tests/write_load.rs diff --git a/Cargo.lock b/Cargo.lock index 51c419c5..d8ff175f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1297,8 +1297,13 @@ dependencies = [ name = "didbot-swarm" version = "0.1.0" dependencies = [ + "axum", + "didbot-data", + "didbot-dns", "didbot-http", + "didbot-identity", "didbot-pds", + "didbot-serve", "rand 0.9.5", "reqwest", "serde", diff --git a/crates/didbot-swarm/Cargo.toml b/crates/didbot-swarm/Cargo.toml index 3b9d3bcd..3bce0fd8 100644 --- a/crates/didbot-swarm/Cargo.toml +++ b/crates/didbot-swarm/Cargo.toml @@ -21,10 +21,15 @@ tracing.workspace = true tracing-subscriber.workspace = true [dev-dependencies] -# The personal data server validates every record against its lexicon, so the -# swarm's generated records are checked against the same code that will -# refuse them rather than against a hand-written approximation of it. +# The swarm is run against a server in the test's own process, over a real +# socket, so what it exercises is the write pipeline the deployment runs and +# what it asserts is read straight off that server's stores and stream. +axum.workspace = true +didbot-data.workspace = true +didbot-dns.workspace = true +didbot-identity.workspace = true didbot-pds.workspace = true +didbot-serve.workspace = true [lints] workspace = true diff --git a/crates/didbot-swarm/tests/write_load.rs b/crates/didbot-swarm/tests/write_load.rs new file mode 100644 index 00000000..b0dcdc60 --- /dev/null +++ b/crates/didbot-swarm/tests/write_load.rs @@ -0,0 +1,413 @@ +//! The swarm against a server in this process, over a real socket. +//! +//! What the swarm is for is exercising the write pipeline before real agents +//! do, so what is asserted here is read off the server rather than off the +//! swarm: the `subscribeRepos` stream's numbering under concurrent writes, +//! the blob index after records pointing at blobs have been deleted, and a +//! frozen repository's contents next to its siblings'. The swarm's own +//! ledger is checked against the server's wherever the two could disagree, +//! because a load generator whose picture of a repository drifts from the +//! server's is one that has stopped listening to refusals. + +use std::collections::BTreeSet; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use didbot_data::{dag_cbor, Value}; +use didbot_dns::LoopbackDns; +use didbot_identity::Zone; +use didbot_pds::blobs::BlobLimits; +use didbot_pds::records::ListParams; +use didbot_pds::{ + Actor, Lock, MemoryAccountStore, MemoryBlobStore, Provisioner, Registry, Sequence, +}; +use didbot_serve::{ + app_with_repos, AuthState, BroadcastSink, Exclusions, Firehose, HealthState, Repos, ReposStream, +}; +use didbot_swarm::vocabulary::{OTHER, THING}; +use didbot_swarm::{Action, Cadence, Pds, Swarm, Verb}; +use tokio::sync::Mutex; + +const ZONE: &str = "agents.localhost"; + +/// How long a step may take before it counts as a hang. +const PATIENCE: Duration = Duration::from_secs(20); + +/// How many frames the replay buffer holds, which bounds how far behind the +/// reader in [`collect_seqs`] may fall. +const REPLAY: usize = 8192; + +/// How long the stream may go quiet before the reader decides it has seen +/// everything the load produced. +const QUIET: Duration = Duration::from_secs(2); + +/// A server, its registry, and the stream producer behind it. +struct Server { + registry: Arc>, + repos: Repos, + base_url: String, +} + +impl Server { + /// Starts a server on a port nobody chose, with no grace on blob + /// collection: a blob nothing points at is a candidate the instant its + /// last reference goes, so the count is the only thing deciding. + async fn start() -> Self { + let zone = Zone::new(ZONE).expect("zone host is valid"); + let repos = Repos::new(Arc::new(Sequence::in_memory()), REPLAY); + let limits = BlobLimits { + collection_grace: Duration::ZERO, + ..BlobLimits::default() + }; + let registry = Arc::new( + Provisioner::new( + "did:web:owner.example", + zone, + format!("http://{ZONE}"), + LoopbackDns::new(), + MemoryAccountStore::new(), + ) + .with_repo_sink(Arc::new(repos.clone())) + .with_blob_store(Arc::new(MemoryBlobStore::with_limits(limits))), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("loopback is bindable"); + let addr = listener.local_addr().expect("the port reads back"); + let dynamic: Arc = registry.clone(); + let router = app_with_repos( + dynamic, + BroadcastSink::default(), + Firehose::default(), + repos.clone(), + AuthState::default(), + HealthState::new(), + ); + tokio::spawn(async move { + axum::serve(listener, router).await.ok(); + }); + Self { + registry, + repos, + base_url: format!("http://{addr}"), + } + } + + /// A swarm against this server, writing on every beat it can. + fn swarm(&self, seed: u64) -> Swarm { + Swarm::new(Pds::new(&self.base_url), seed).with_cadence(Cadence { + writes_per_minute: f64::MAX, + beats_per_second: 1.0, + }) + } + + /// Every record key `did` holds, across both sample collections. + fn keys(&self, did: &str) -> BTreeSet { + [THING, OTHER] + .into_iter() + .flat_map(|collection| { + self.registry + .list_records(did, collection, &ListParams::new(10_000)) + .expect("the repository lists") + .into_iter() + .map(|(rkey, _)| rkey) + }) + .collect() + } +} + +/// Runs `beats` beats through `swarm` on `workers` concurrent drivers, each +/// choosing by `choose`. +/// +/// The lock is held to plan and to settle and never across the request, so +/// the server sees `workers` requests at once while the decisions stay one +/// seeded sequence. +async fn drive( + swarm: &Arc>, + beats: usize, + workers: usize, + choose: fn(&mut Swarm) -> Action, +) { + let pds = swarm.lock().await.pds().clone(); + let remaining = Arc::new(AtomicUsize::new(beats)); + let mut handles = Vec::with_capacity(workers); + for _ in 0..workers { + let swarm = swarm.clone(); + let pds = pds.clone(); + let remaining = remaining.clone(); + handles.push(tokio::spawn(async move { + loop { + if remaining + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |left| { + left.checked_sub(1) + }) + .is_err() + { + return; + } + let plan = { + let mut held = swarm.lock().await; + let action = choose(&mut held); + held.plan(action) + }; + let outcome = pds.perform(&plan).await; + if let Err(err) = swarm.lock().await.settle(plan, outcome) { + panic!("a beat failed against a healthy server: {err}"); + } + } + })); + } + for handle in handles { + tokio::time::timeout(PATIENCE, handle) + .await + .expect("a driver finishes") + .expect("a driver does not panic"); + } +} + +fn population(swarm: &mut Swarm) -> Action { + swarm.next_action(6) +} + +fn writes(swarm: &mut Swarm) -> Action { + swarm.next_write() +} + +/// Reads frames off `stream` until `wanted` have arrived or the stream goes +/// [`QUIET`], answering their sequence numbers in arrival order. +async fn collect_seqs(mut stream: ReposStream, wanted: u64) -> Vec { + let mut seqs = Vec::new(); + while (seqs.len() as u64) < wanted { + match tokio::time::timeout(QUIET, stream.next()).await { + Ok(Some(bytes)) => seqs.push(seq_of(&bytes)), + Ok(None) | Err(_) => break, + } + } + seqs +} + +/// The `seq` in a frame's body. +/// +/// A frame is a header and a body, two DAG-CBOR values back to back, and +/// only the encoder knows where the split is; trying every offset is cheap +/// and proves the two really are concatenated. +fn seq_of(bytes: &[u8]) -> u64 { + for split in 1..bytes.len() { + let (Ok(_), Ok(Value::Map(body))) = ( + dag_cbor::decode(&bytes[..split]), + dag_cbor::decode(&bytes[split..]), + ) else { + continue; + }; + let Some(Value::Integer(seq)) = body.get("seq") else { + panic!("a frame body carries no integer seq: {body:?}"); + }; + return u64::try_from(*seq).expect("a sequence number is not negative"); + } + panic!("a frame did not decode as two DAG-CBOR values"); +} + +/// Under concurrent writes, provisioning and deletion, every number from one +/// to the newest the producer admits to reaches a subscriber exactly once: +/// no gap, no repeat. +/// +/// The reader subscribes before the load starts and runs alongside it, so +/// what it sees is the live stream rather than a replay tidied up after the +/// fact. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn the_stream_stays_contiguous_under_concurrent_writes() { + let server = Server::start().await; + let stream = server + .repos + .subscribe(None, Exclusions::default()) + .expect("a live subscription"); + + let swarm = Arc::new(Mutex::new(server.swarm(3))); + for _ in 0..6 { + swarm + .lock() + .await + .spawn() + .await + .expect("an agent provisions"); + } + drive(&swarm, 240, 8, population).await; + + let newest = server.repos.newest(); + let snapshot = swarm.lock().await.tally.snapshot(); + assert_eq!(snapshot.failed(), 0, "{snapshot}"); + for verb in [ + Verb::CreateRecord, + Verb::PutRecord, + Verb::DeleteRecord, + Verb::UploadBlob, + Verb::DeleteAgent, + ] { + assert!(snapshot.get(verb).ok > 0, "{verb:?} never ran:\n{snapshot}"); + } + + let mut seqs = tokio::time::timeout(PATIENCE, collect_seqs(stream, newest)) + .await + .expect("the stream delivers what the producer numbered"); + seqs.sort_unstable(); + let expected: Vec = (1..=newest).collect(); + assert_eq!( + seqs, expected, + "the stream skipped or repeated a number under load" + ); +} + +/// A blob a deleted record pointed at is what the collector takes, and +/// exactly that: every blob the swarm orphaned is collectable, nothing a live +/// record names is, and a pass takes them all. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_blob_orphaned_by_a_deleted_record_is_collectable() { + let server = Server::start().await; + let swarm = Arc::new(Mutex::new(server.swarm(5))); + for _ in 0..4 { + swarm + .lock() + .await + .spawn() + .await + .expect("an agent provisions"); + } + drive(&swarm, 300, 4, writes).await; + + let snapshot = swarm.lock().await.tally.snapshot(); + assert_eq!(snapshot.failed(), 0, "{snapshot}"); + assert!( + snapshot.orphaned > 0, + "the swarm deleted no record that pointed at a blob:\n{snapshot}" + ); + let live_attachments = swarm + .lock() + .await + .agents() + .iter() + .flat_map(|agent| agent.records.iter()) + .filter(|held| held.attachment.is_some()) + .count() as u64; + + let stats = server.registry.stats().blobs; + assert_eq!( + stats.collectable.count, snapshot.orphaned, + "what the collector would take is not what the swarm orphaned: {stats:?}" + ); + // The referenced ones are every attachment still held, plus one avatar + // per account. + assert_eq!( + stats.referenced.count, + live_attachments + swarm.lock().await.len() as u64, + "{stats:?}" + ); + + let collected = server.registry.collect_blobs(); + assert_eq!(collected.len() as u64, snapshot.orphaned); + let after = server.registry.stats().blobs; + assert_eq!(after.collectable.count, 0, "{after:?}"); + assert_eq!(after.referenced, stats.referenced, "{after:?}"); +} + +/// A frozen account's writes are refused and the swarm stops sending them, +/// while its siblings go on writing; and the swarm's picture of every +/// repository, frozen or not, is what the server holds. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_frozen_agent_is_refused_while_its_siblings_proceed() { + let server = Server::start().await; + let swarm = Arc::new(Mutex::new(server.swarm(11))); + for _ in 0..4 { + swarm + .lock() + .await + .spawn() + .await + .expect("an agent provisions"); + } + drive(&swarm, 40, 4, writes).await; + + let dids: Vec = swarm + .lock() + .await + .agents() + .iter() + .map(|agent| agent.did.clone()) + .collect(); + let frozen = dids[0].clone(); + let before: Vec> = dids.iter().map(|did| server.keys(did)).collect(); + server + .registry + .lock(&frozen, Lock::Frozen, Actor::Operator("write-load")) + .expect("the account freezes"); + let landed_before = swarm + .lock() + .await + .tally + .snapshot() + .get(Verb::CreateRecord) + .ok; + + // Refusals are failures the driver would otherwise panic on; here they + // are the point, so the beats are run by hand. + let pds = swarm.lock().await.pds().clone(); + let mut refusals = 0; + for _ in 0..80 { + let plan = { + let mut held = swarm.lock().await; + let action = held.next_write(); + held.plan(action) + }; + let outcome = pds.perform(&plan).await; + if let Err(err) = swarm.lock().await.settle(plan, outcome) { + assert!(err.is_freeze(), "a write failed for another reason: {err}"); + refusals += 1; + } + } + + let snapshot = swarm.lock().await.tally.snapshot(); + assert!( + refusals > 0, + "the frozen account refused nothing:\n{snapshot}" + ); + assert_eq!(snapshot.refused, refusals); + assert!( + snapshot.get(Verb::CreateRecord).ok > landed_before, + "the siblings wrote nothing after the freeze:\n{snapshot}" + ); + + let held = swarm.lock().await; + let agent = held + .agents() + .iter() + .find(|agent| agent.did == frozen) + .expect("the frozen agent is still in the population"); + assert!( + agent.frozen, + "the swarm did not take the freeze at its word" + ); + assert_eq!( + server.keys(&frozen), + before[0], + "the frozen repository moved" + ); + for (index, did) in dids.iter().enumerate() { + let ledger: BTreeSet = held + .agents() + .iter() + .find(|agent| &agent.did == did) + .expect("every agent is still in the population") + .records + .iter() + .map(|held| held.rkey.clone()) + .collect(); + assert_eq!( + ledger, + server.keys(did), + "the swarm's picture of {did} is not the server's" + ); + if index > 0 { + assert_ne!(server.keys(did), before[index], "{did} wrote nothing"); + } + } +} -- 2.51.2