From d5469f9f3d71f3ec41b326bee75bbbab12df23aa Mon Sep 17 00:00:00 2001 From: "@permadeath.com" Date: Fri, 11 Sep 2026 09:37:41 -0400 Subject: [PATCH] feat(serve): run the blob collector on a timer beside the stale sweep `Provisioner::collect_blobs` is reached through a new `BlobCollect` trait and called by `BlobCollector` every hour, first tick at startup, the way `StaleSweeper` runs the retention sweep. Co-Authored-By: Claude Fable 5.1 Change-Id: Ie4a26aca6139a058b56b0966bb72b470601d11ef --- crates/didbot-pds/src/lib.rs | 4 +- crates/didbot-pds/src/provision.rs | 26 +++++- crates/didbot-serve/src/bin/didbot-pds.rs | 39 +++++---- crates/didbot-serve/src/blob_collect.rs | 95 ++++++++++++++++++++++ crates/didbot-serve/src/lib.rs | 2 + crates/didbot-serve/tests/blob_collect.rs | 99 +++++++++++++++++++++++ plan/account-data.md | 2 +- 7 files changed, 244 insertions(+), 23 deletions(-) create mode 100644 crates/didbot-serve/src/blob_collect.rs create mode 100644 crates/didbot-serve/tests/blob_collect.rs diff --git a/crates/didbot-pds/src/lib.rs b/crates/didbot-pds/src/lib.rs index 4ecf35d4..1fe962bc 100644 --- a/crates/didbot-pds/src/lib.rs +++ b/crates/didbot-pds/src/lib.rs @@ -180,8 +180,8 @@ pub use names::{ }; pub use provision::RegistrationFacts; pub use provision::{ - AccountRevokeHook, Confinement, Delta, ProvisionError, ProvisionRequest, Provisioned, - Provisioner, Registry, RegistryStats, RepoExport, RepoHead, ReservationDepth, + AccountRevokeHook, BlobCollect, Confinement, Delta, ProvisionError, ProvisionRequest, + Provisioned, Provisioner, Registry, RegistryStats, RepoExport, RepoHead, ReservationDepth, ReservationPolicy, ReservationRequest, ReservedIdentity, StaleSweep, Swap, DEFAULT_PENDING_RESERVATIONS, DEFAULT_RECORD_TARGET, DEFAULT_RESERVATION_TTL, RESERVATION_BACKEND, UNAUTHENTICATED_BACKEND, diff --git a/crates/didbot-pds/src/provision.rs b/crates/didbot-pds/src/provision.rs index 41af6da9..6f8f08bf 100644 --- a/crates/didbot-pds/src/provision.rs +++ b/crates/didbot-pds/src/provision.rs @@ -1166,6 +1166,16 @@ pub trait StaleSweep: Send + Sync { fn sweep_stale_bounded(&self, max_age: time::Duration, limit: usize) -> Vec; } +/// The one thing a blob collector's timer needs from a registry. +/// +/// Its own trait for the reason [`StaleSweep`] is: the caller answers a +/// clock, not a request, and holds exactly the capability it runs. +pub trait BlobCollect: Send + Sync { + /// Unlinks every blob nothing references, past its grace window, and + /// answers what was taken. See [`Provisioner::collect_blobs`]. + fn collect_blobs(&self) -> Vec; +} + /// The object-safe view of a registry, as the HTTP layer holds it. /// /// Every method takes a DID as `&str` and returns owned values, so the trait @@ -2257,9 +2267,9 @@ where /// Collects every blob nothing references, past its grace window, and /// answers what was taken. /// - /// Called by whatever already runs on a schedule in the deployment, the - /// same way [`Self::sweep_stale_bounded`]'s timer lives in - /// `didbot-serve`: this crate holds no runtime of its own. + /// Called on a timer in `didbot-serve`, through [`BlobCollect`], the + /// same way [`Self::sweep_stale_bounded`] is through [`StaleSweep`]: + /// this crate holds no runtime of its own. /// /// Held under the same write lock every write path holds /// across its own check-apply-commit — this is not a write, but it must @@ -5162,6 +5172,16 @@ where } } +impl BlobCollect for Provisioner +where + D: DnsProvider, + S: AccountStore, +{ + fn collect_blobs(&self) -> Vec { + Provisioner::collect_blobs(self) + } +} + impl Registry for Provisioner where D: DnsProvider, diff --git a/crates/didbot-serve/src/bin/didbot-pds.rs b/crates/didbot-serve/src/bin/didbot-pds.rs index 27afa6e5..e305f193 100644 --- a/crates/didbot-serve/src/bin/didbot-pds.rs +++ b/crates/didbot-serve/src/bin/didbot-pds.rs @@ -45,10 +45,10 @@ use didbot_pds::evaluation_log::{EvaluationLog, FileSink, LOG_FILE}; use didbot_pds::policy_tree::{build_tree, TreePolicyGate}; use didbot_pds::Estop; use didbot_pds::{ - AccountStore, AgentTokenStore, BlobLimits, BlobStore, CommitSink, CommitStore, Durable, - LedgerStore, LifecycleSink, MemoryAccountStore, MemoryBlobStore, Naming, ProvisionRequest, - Provisioner, RecordStore, RegistrationFacts, Registry, RepoEventSink, ReservationPolicy, - ServerLifecycle, StaleSweep, Trust, DEFAULT_GRACE_WINDOW, DEFAULT_HOLD, + AccountStore, AgentTokenStore, BlobCollect, BlobLimits, BlobStore, CommitSink, CommitStore, + Durable, LedgerStore, LifecycleSink, MemoryAccountStore, MemoryBlobStore, Naming, + ProvisionRequest, Provisioner, RecordStore, RegistrationFacts, Registry, RepoEventSink, + ReservationPolicy, ServerLifecycle, StaleSweep, Trust, DEFAULT_GRACE_WINDOW, DEFAULT_HOLD, }; use didbot_serve::ownership_poll; use didbot_serve::policy_poll::PolicyPoll; @@ -56,10 +56,10 @@ use didbot_serve::policy_poll::PolicyPoll; use didbot_serve::serve_tls; use didbot_serve::DEFAULT_REPOS_CAPACITY; use didbot_serve::{ - app_with_repos, estop_admin, init_tracing, serve_router, AuthState, BroadcastSink, - ConfinementPoll, Disclosure, Firehose, HealthState, HealthTick, OwnershipPoll, Repos, - StaleSweeper, SweepSchedule, ZoneReadRetry, ZoneUnderRead, DEFAULT_CONFINEMENT_INTERVAL, - DEFAULT_HEALTH_INTERVAL, DEFAULT_ZONE_READ_INTERVAL, + app_with_repos, estop_admin, init_tracing, serve_router, AuthState, BlobCollector, + BroadcastSink, ConfinementPoll, Disclosure, Firehose, HealthState, HealthTick, OwnershipPoll, + Repos, StaleSweeper, SweepSchedule, ZoneReadRetry, ZoneUnderRead, DEFAULT_COLLECT_INTERVAL, + DEFAULT_CONFINEMENT_INTERVAL, DEFAULT_HEALTH_INTERVAL, DEFAULT_ZONE_READ_INTERVAL, }; use socket2::{Domain, Protocol, Type}; use tokio::net::TcpListener; @@ -1414,7 +1414,7 @@ async fn run(mut args: Args) -> Result<(), String> { .pending_reservations .unwrap_or(didbot_pds::DEFAULT_PENDING_RESERVATIONS), }; - let (registry, sweepable): Assembled = match &durable { + let (registry, sweepable, collectable): Assembled = match &durable { Some(durable) => assemble( args.owner.clone(), zone.clone(), @@ -1695,6 +1695,9 @@ async fn run(mut args: Args) -> Result<(), String> { // withdrawn by the deletion that account's retention earns it. See // `didbot_serve::stale_sweep`. let sweeper = StaleSweeper::spawn(sweepable, SweepSchedule::default()); + // Unlinks blobs nothing references once their grace window has run + // out, on the same kind of clock; see `didbot_serve::blob_collect`. + let collector = BlobCollector::spawn(collectable, DEFAULT_COLLECT_INTERVAL); // The tree of principals, re-derived on the ownership poll's clock, so a // vouch that lapses in a parent's repository confines its subtree // without anyone asking; see `didbot_serve::confinement_poll`. @@ -1714,6 +1717,7 @@ async fn run(mut args: Args) -> Result<(), String> { }; tick.stop().await; sweeper.stop().await; + collector.stop().await; confinement.stop().await; zone_read.stop().await; if let Some(poll) = ownership_poll { @@ -1734,10 +1738,11 @@ async fn run(mut args: Args) -> Result<(), String> { result.map_err(|err| format!("server stopped: {err}")) } -/// One `Provisioner` behind the two object-safe facades the server holds it -/// by: the router answers requests through [`Registry`], and -/// `StaleSweeper` answers a clock through [`StaleSweep`]. -type Assembled = (Arc, Arc); +/// One `Provisioner` behind the three object-safe facades the server holds +/// it by: the router answers requests through [`Registry`], `StaleSweeper` +/// answers a clock through [`StaleSweep`], and `BlobCollector` answers one +/// through [`BlobCollect`]. +type Assembled = (Arc, Arc, Arc); /// Builds the provisioner and hands it back behind the object-safe facade. /// @@ -1824,11 +1829,11 @@ where .ensure_server_account() .map_err(|err| format!("could not adopt the server's own account: {err}"))?; info!(did = account.did.as_str(), "the server has an account"); - // One `Provisioner` behind two erased handles: the router holds the - // request-answering half, `StaleSweeper` holds the clock-answering one, - // and they are the same accounts either way. + // One `Provisioner` behind three erased handles: the router holds the + // request-answering one, `StaleSweeper` and `BlobCollector` each hold a + // clock-answering one, and they are the same accounts either way. let provisioner = Arc::new(provisioner); - Ok((provisioner.clone(), provisioner)) + Ok((provisioner.clone(), provisioner.clone(), provisioner)) } /// The gate every write this server admits is judged by, and the log both diff --git a/crates/didbot-serve/src/blob_collect.rs b/crates/didbot-serve/src/blob_collect.rs new file mode 100644 index 00000000..296539e8 --- /dev/null +++ b/crates/didbot-serve/src/blob_collect.rs @@ -0,0 +1,95 @@ +//! The timer that runs `didbot_pds::Provisioner::collect_blobs`. +//! +//! An upload starts unreferenced, and a blob whose reference count has +//! fallen to zero stays on disk until something unlinks it. `collect_blobs` +//! is that something, and this is its clock — the same shape as +//! [`crate::stale_sweep`], for the same reason: a startup-only pass would +//! leave a long-lived server accumulating bytes nothing names. + +use std::panic::AssertUnwindSafe; +use std::sync::Arc; +use std::time::Duration; + +use didbot_pds::BlobCollect; +use tokio::sync::Notify; +use tokio::task::JoinHandle; +use tokio::time::MissedTickBehavior; +use tracing::{info, warn}; + +/// How often the collector runs, absent configuration. +/// +/// An hour: the floor under a deployment's grace window is +/// `didbot_pds::blobs::MIN_BLOB_COLLECTION_GRACE`, an hour, so a blob past +/// its window waits at most another window's worth before a pass takes it. +pub const DEFAULT_COLLECT_INTERVAL: Duration = Duration::from_secs(60 * 60); + +/// A running collector, stopped by [`BlobCollector::stop`] or by dropping it. +pub struct BlobCollector { + handle: JoinHandle<()>, + stop: Arc, +} + +impl BlobCollector { + /// Starts collecting through `registry` every `interval`. + /// + /// The first tick fires immediately, so a restart takes what the + /// previous run left past its grace before it serves anything. Each + /// pass runs on a blocking thread, as [`crate::stale_sweep`]'s does: it + /// takes the write lock every write path takes and unlinks files. + pub fn spawn(registry: Arc, interval: Duration) -> Self { + let stop = Arc::new(Notify::new()); + let notified = stop.clone(); + let handle = tokio::spawn(async move { + let mut ticker = tokio::time::interval(interval); + ticker.set_missed_tick_behavior(MissedTickBehavior::Delay); + loop { + tokio::select! { + biased; + () = notified.notified() => break, + _ = ticker.tick() => {} + } + collect_once(registry.clone()).await; + } + }); + Self { handle, stop } + } + + /// Stops the collector and waits for the task to finish. + pub async fn stop(mut self) { + self.stop.notify_one(); + let _ = (&mut self.handle).await; + } +} + +impl Drop for BlobCollector { + fn drop(&mut self) { + self.handle.abort(); + } +} + +/// One tick: collect, say what it took. +/// +/// Public and separate from the timer for the reason +/// [`crate::stale_sweep::sweep_once`] is: a caller can run exactly one pass +/// without waiting on a clock. +pub async fn collect_once(registry: Arc) { + let collected = tokio::task::spawn_blocking(move || { + std::panic::catch_unwind(AssertUnwindSafe(|| registry.collect_blobs())) + }) + .await; + + match collected { + Ok(Ok(collected)) if collected.is_empty() => {} + Ok(Ok(collected)) => info!( + blobs = collected.len(), + bytes = collected.iter().map(|blob| blob.bytes).sum::(), + "collected blobs nothing referenced, past their grace window" + ), + // Both arms leave the timer running: what this pass did not take + // is still there for the next one. + Ok(Err(_)) => warn!("the blob collector panicked; the next tick tries again"), + Err(error) => { + warn!(%error, "the blob collector did not finish; the next tick tries again") + } + } +} diff --git a/crates/didbot-serve/src/lib.rs b/crates/didbot-serve/src/lib.rs index e686e239..2f6baf75 100644 --- a/crates/didbot-serve/src/lib.rs +++ b/crates/didbot-serve/src/lib.rs @@ -68,6 +68,7 @@ #![forbid(unsafe_code)] mod auth; +pub mod blob_collect; mod blobs; pub mod confinement_poll; mod dashboard; @@ -93,6 +94,7 @@ mod wire; pub mod zone_read; pub use auth::{Credential, Disclosure, ROUTE_CREDENTIALS}; +pub use blob_collect::{collect_once, BlobCollector, DEFAULT_COLLECT_INTERVAL}; pub use confinement_poll::{ConfinementPoll, DEFAULT_CONFINEMENT_INTERVAL}; pub use error::{repo_surface, ApiError, ErrorBody}; pub use events::{BroadcastSink, EventFrame, EventStream, DEFAULT_EVENT_CAPACITY}; diff --git a/crates/didbot-serve/tests/blob_collect.rs b/crates/didbot-serve/tests/blob_collect.rs new file mode 100644 index 00000000..071daa70 --- /dev/null +++ b/crates/didbot-serve/tests/blob_collect.rs @@ -0,0 +1,99 @@ +//! The scheduled blob collector, against a real `Provisioner`. +//! +//! `didbot-pds`'s own blob tests cover the decision — what is past its +//! grace and unreferenced. This covers the deployment actually making it on +//! a clock: a blob nothing ever referenced is gone once the timer has run, +//! and one a record holds is not. + +use std::sync::Arc; +use std::time::Duration; + +use didbot_dns::LoopbackDns; +use didbot_identity::Zone; +use didbot_pds::{ + BlobCollect, BlobLimits, MemoryAccountStore, MemoryBlobStore, ProvisionRequest, Provisioner, + Registry, Swap, +}; +use didbot_serve::BlobCollector; +use serde_json::json; + +const ZONE_HOST: &str = "agents.localhost"; + +/// Under the atproto floor on purpose, reachable only by building the +/// struct directly — the floor guards configuration, and a test that had +/// to wait an hour would not be one. +const TEST_GRACE: Duration = Duration::from_millis(20); + +fn provisioner() -> Arc> { + let zone = Zone::delegated("localhost", ZONE_HOST).expect("test zone should be constructible"); + Arc::new( + Provisioner::new( + "did:web:owner.example", + zone, + "http://localhost:3000".to_owned(), + LoopbackDns::new(), + MemoryAccountStore::new(), + ) + .with_blob_store(Arc::new(MemoryBlobStore::with_limits(BlobLimits { + collection_grace: TEST_GRACE, + ..BlobLimits::default() + }))), + ) +} + +fn upload(pds: &impl Registry, did: &str, bytes: &[u8]) -> String { + let mut upload = pds + .begin_blob(did, "image/png", Some(bytes.len() as u64)) + .expect("a blob upload should open"); + upload.write(bytes).expect("a chunk should write"); + upload.commit().expect("the upload should commit").cid +} + +#[tokio::test] +async fn the_scheduled_collector_unlinks_an_unreferenced_blob_and_spares_a_referenced_one() { + let pds = provisioner(); + let did = pds + .provision(ProvisionRequest::new("junco", None)) + .expect("provisioning should succeed") + .account + .did + .as_str() + .to_owned(); + let stray = upload(pds.as_ref(), &did, b"never referenced by anything"); + let kept = upload(pds.as_ref(), &did, b"an avatar's worth of bytes"); + pds.put_record( + &did, + "com.example.thing", + None, + json!({"image": {"$type": "blob", "ref": {"$link": kept}, + "mimeType": "image/png", "size": 26}}), + &Swap::default(), + ) + .expect("the write should land"); + + let collector = BlobCollector::spawn( + pds.clone() as Arc, + Duration::from_millis(5), + ); + // Poll rather than sleep a fixed span: wait for a tick past the grace + // window to have actually taken the stray. + let taken = tokio::time::timeout(Duration::from_secs(5), async { + loop { + if pds.fetch_blob(&did, &stray).is_err() { + return true; + } + tokio::time::sleep(Duration::from_millis(1)).await; + } + }) + .await; + collector.stop().await; + + assert!( + taken.is_ok(), + "the scheduled collector should have unlinked the unreferenced blob" + ); + assert!( + pds.fetch_blob(&did, &kept).is_ok(), + "a blob a record references must survive the collector" + ); +} diff --git a/plan/account-data.md b/plan/account-data.md index 0d23fde7..2a51692a 100644 --- a/plan/account-data.md +++ b/plan/account-data.md @@ -177,7 +177,6 @@ binary. The stale sweep has a timer; the collector gets the same one. collection; the Cedar context gains `operation == "delete"` with `existing` populated. - [ ] Heap compaction, with the byte-identity tests extended to it. -- [ ] Schedule the blob collector beside the stale sweep. - [ ] One integration test per row of the table above, each asserting the cell after a restart and a checkpoint. - [ ] The architecture page: the four kinds of record, the table, and the @@ -191,3 +190,4 @@ binary. The stale sweep has a timer; the collector gets the same one. tombstone's value; the roundtrip invariant as a property test over the regex engine and an integration test through `deleteRecord` then `createRecord` across a checkpoint and a restart. +- [x] Schedule the blob collector beside the stale sweep. -- 2.51.2