diff --git a/docs/configuration.md b/docs/configuration.md index cd39fb4..fd6c8f8 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -59,6 +59,7 @@ hydrant is configured via environment variables, all prefixed with `HYDRANT_` (e | variable | default | description | | :--- | :--- | :--- | | `BACKFILL_CONCURRENCY_LIMIT` | `16` (`64` full network) | maximum number of concurrent backfill tasks | +| `BACKFILL_MEMORY_BYTES` | `4294967296` (4 GiB) | memory that unfiltered backfills may hold at once. each one counts its car's expected peak, about 6x the car, since it keeps every block, the records and the repo's write batch until it commits. a backfill that would go past it waits, except the oldest one running, which always goes ahead, so peak memory stays near this plus one repo. filtered backfills stream and keep only what they want, so they don't count | | `BACKFILL_STRATEGY` | `auto` | backfill strategy: `full` keeps the existing `getRepo` path, `sparse-filter` attempts authenticated sparse collection backfill before falling back to full, `auto` probes filtered repos and falls back to full for tiny MST roots | | `REPO_FETCH_TIMEOUT` | `5min` | timeout for fetching a repository | | `VERIFY_SIGNATURES` | `full` | signature verification level: `full`, `backfill-only`, or `none` | diff --git a/src/backfill/budget.rs b/src/backfill/budget.rs new file mode 100644 index 0000000..7812cc9 --- /dev/null +++ b/src/backfill/budget.rs @@ -0,0 +1,195 @@ +//! how much memory whole-car backfills may hold at once. a whole car keeps every block, the +//! records from the walk and the repo's write batch until it commits, so its peak follows the +//! car's size, and a few big repos fetched together can take far more than the concurrency +//! limit suggests. + +use std::collections::BTreeMap; +use std::sync::{Arc, Mutex}; + +use tokio::sync::Notify; + +/// peak memory above baseline per byte of car, the larger of the bench repos measured with the +/// perf-backfill probe: jay.bsky.team (15 MB) peaked at about 6.3x its car and pfrazee.com +/// (83 MB) at about 4.5x, so small repos carry more fixed cost per byte +const PEAK_PER_CAR_BYTE: f64 = 6.3; + +/// what a whole-car backfill of `car_bytes` is expected to hold at its peak +pub(crate) fn peak_for(car_bytes: usize) -> usize { + (car_bytes as f64 * PEAK_PER_CAR_BYTE) as usize +} + +#[derive(Clone)] +pub(crate) struct BackfillBudget { + inner: Arc, +} + +struct Inner { + limit: usize, + state: Mutex, + freed: Notify, +} + +#[derive(Default)] +struct State { + used: usize, + next_ticket: u64, + /// bytes held by each reservation, by when it was made + holders: BTreeMap, +} + +impl BackfillBudget { + pub(crate) fn new(limit: usize) -> Self { + Self { + inner: Arc::new(Inner { + limit, + state: Mutex::default(), + freed: Notify::new(), + }), + } + } + + /// an empty reservation, placed behind every one made before it + pub(crate) fn reservation(&self) -> Reservation { + let mut state = self.inner.state.lock().unwrap(); + let ticket = state.next_ticket; + state.next_ticket += 1; + state.holders.insert(ticket, 0); + Reservation { + inner: self.inner.clone(), + ticket, + bytes: 0, + } + } + + #[cfg(test)] + pub(crate) fn used(&self) -> usize { + self.inner.state.lock().unwrap().used + } +} + +/// a share of the budget, given back when dropped, so a backfill that fails, times out or is +/// cancelled can't leak it +pub(crate) struct Reservation { + inner: Arc, + ticket: u64, + bytes: usize, +} + +impl Reservation { + /// waits until this reservation holds at least `bytes`. the oldest reservation never + /// waits and may go over the limit: reservations grow as their cars arrive, so two big + /// ones could otherwise each hold part of the budget and wait on each other forever + pub(crate) async fn grow_to(&mut self, bytes: usize) { + if bytes <= self.bytes { + return; + } + let more = bytes - self.bytes; + loop { + let freed = self.inner.freed.notified(); + tokio::pin!(freed); + // registered before checking, so a release between the check and the wait isn't + // missed + freed.as_mut().enable(); + { + let mut state = self.inner.state.lock().unwrap(); + let oldest = state.holders.keys().next() == Some(&self.ticket); + if oldest || state.used + more <= self.inner.limit { + state.used += more; + *state.holders.get_mut(&self.ticket).unwrap() += more; + self.bytes = bytes; + return; + } + } + freed.await; + } + } +} + +impl std::fmt::Debug for Reservation { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Reservation") + .field("bytes", &self.bytes) + .finish() + } +} + +impl Drop for Reservation { + fn drop(&mut self) { + let mut state = self.inner.state.lock().unwrap(); + state.holders.remove(&self.ticket); + state.used -= self.bytes; + drop(state); + self.inner.freed.notify_waiters(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use futures::FutureExt; + use std::time::Duration; + + #[tokio::test] + async fn small_backfills_never_wait_on_the_default_budget() { + let budget = BackfillBudget::new(crate::config::Config::default().backfill_memory_bytes); + let mut held = Vec::new(); + for _ in 0..64 { + let mut reservation = budget.reservation(); + // a 1 MB car, bigger than most repos + reservation + .grow_to(peak_for(1024 * 1024)) + .now_or_never() + .expect("a small backfill waited for the budget"); + held.push(reservation); + } + drop(held); + assert_eq!(budget.used(), 0); + } + + #[tokio::test] + async fn two_growing_backfills_bigger_than_the_budget_both_finish() { + let budget = BackfillBudget::new(100); + let grow = |budget: BackfillBudget| async move { + let mut reservation = budget.reservation(); + for step in 1..=10 { + reservation.grow_to(step * 30).await; + tokio::task::yield_now().await; + } + reservation.bytes + }; + let both = futures::future::join(grow(budget.clone()), grow(budget.clone())); + let done = tokio::time::timeout(Duration::from_secs(5), both) + .await + .expect("two growing reservations waited on each other"); + assert_eq!(done, (300, 300)); + assert_eq!(budget.used(), 0); + } + + #[tokio::test] + async fn a_newer_reservation_waits_until_an_older_one_lets_go() { + let budget = BackfillBudget::new(100); + let mut older = budget.reservation(); + older.grow_to(150).await; + let mut newer = budget.reservation(); + assert!(newer.grow_to(10).now_or_never().is_none()); + + drop(older); + newer + .grow_to(10) + .now_or_never() + .expect("freed budget was handed on"); + assert_eq!(budget.used(), 10); + } + + #[tokio::test] + async fn a_cancelled_wait_takes_nothing() { + let budget = BackfillBudget::new(100); + let mut older = budget.reservation(); + older.grow_to(100).await; + let mut newer = budget.reservation(); + assert!(newer.grow_to(50).now_or_never().is_none()); + drop(newer); + drop(older); + assert_eq!(budget.used(), 0); + } +} diff --git a/src/backfill/mod.rs b/src/backfill/mod.rs index 83db347..5d868cd 100644 --- a/src/backfill/mod.rs +++ b/src/backfill/mod.rs @@ -1,4 +1,5 @@ mod admission; +pub(crate) mod budget; pub mod client; pub mod error; pub(crate) mod filtered_car; diff --git a/src/backfill/worker/process.rs b/src/backfill/worker/process.rs index 26a20b2..68f2cd2 100644 --- a/src/backfill/worker/process.rs +++ b/src/backfill/worker/process.rs @@ -15,6 +15,7 @@ use jacquard_common::types::cid::{Cid as AtCid, IpldCid}; use jacquard_common::types::did::Did; use crate::backfill::admission::BackfillAdmission; +use crate::backfill::budget::{BackfillBudget, Reservation, peak_for}; use crate::backfill::client::{ThrottledHttpClient, collect_body_bounded}; use crate::backfill::error::BackfillError; use crate::backfill::filtered_car::{FilteredCar, FilteredRepo}; @@ -211,10 +212,14 @@ pub(super) async fn process_did( let throttle = app_state.throttler.get_handle(&pds).await; let tier = app_state.resolve_pds_tier(pds.host_str().unwrap_or("")); let car = FilteredCar::new(filter.clone(), app_state.verify_cids).map_or_else( - || RepoCar::Whole(WholeCar::new(app_state.verify_cids)), + || { + let whole = WholeCar::new(app_state.verify_cids); + RepoCar::Whole(whole, app_state.backfill_budget.clone()) + }, RepoCar::Filtered, ); - let car = match fetch_full_repo_car( + // the budget a whole car takes is held until its batch commits, since that's its peak + let (car, _budget) = match fetch_full_repo_car( http, &pds, did, @@ -225,8 +230,8 @@ pub(super) async fn process_did( ) .await? { - FullRepoOutcome::Car(parsed) => ParsedRepo::Whole(parsed), - FullRepoOutcome::Filtered(repo) => ParsedRepo::Filtered(repo), + FullRepoOutcome::Car { car, budget } => (ParsedRepo::Whole(car), budget), + FullRepoOutcome::Filtered(repo) => (ParsedRepo::Filtered(repo), None), FullRepoOutcome::NotFound => { // `RepoNotFound` is not lifecycle evidence: it says the pds has no repo, not that the // account was deleted, and the repo may only be missing on this target. erasure stays @@ -572,21 +577,24 @@ enum ParsedRepo { /// what a getRepo body is read into as it streams enum RepoCar { - Whole(WholeCar), + Whole(WholeCar, BackfillBudget), Filtered(FilteredCar), } impl RepoCar { fn push(&mut self, chunk: &[u8]) -> Result<()> { match self { - RepoCar::Whole(car) => car.push(chunk), + RepoCar::Whole(car, _) => car.push(chunk), RepoCar::Filtered(car) => car.push(chunk), } } fn finish(self) -> Result { Ok(match self { - RepoCar::Whole(car) => FullRepoOutcome::Car(car.finish()?), + RepoCar::Whole(car, _) => FullRepoOutcome::Car { + car: car.finish()?, + budget: None, + }, RepoCar::Filtered(car) => FullRepoOutcome::Filtered(Box::new(car.finish()?)), }) } @@ -607,9 +615,19 @@ async fn stream_car( if content_length.is_some_and(|len| len > max_bytes as u64) { return Ok(None); } - if let (RepoCar::Whole(whole), Some(len)) = (&mut car, content_length) { - whole.reserve(len as usize); - } + // taken after the backfill permit, the pds admission and the pds connection permit and + // never before them, so a task waiting here holds nothing a budget holder could need + let mut budget = match &mut car { + RepoCar::Whole(whole, shared) => { + let mut reservation = shared.reservation(); + if let Some(len) = content_length { + reservation.grow_to(peak_for(len as usize)).await; + whole.reserve(len as usize); + } + Some(reservation) + } + RepoCar::Filtered(_) => None, + }; let (tx, mut rx) = tokio::sync::mpsc::channel::(CHUNKS_IN_FLIGHT); let reader = tokio::task::spawn_blocking(move || { while let Some(chunk) = rx.blocking_recv() { @@ -631,6 +649,9 @@ async fn stream_car( if received > max_bytes { return Ok(false); } + if let Some(budget) = &mut budget { + budget.grow_to(peak_for(received)).await; + } // a closed channel means the reader stopped on an error, which joining it reports if tx.send(chunk).await.is_err() { break; @@ -640,7 +661,10 @@ async fn stream_car( } .await; drop(tx); - let outcome = reader.await.into_diagnostic()?; + let mut outcome = reader.await.into_diagnostic()?; + if let Ok(FullRepoOutcome::Car { budget: held, .. }) = &mut outcome { + *held = budget; + } match read.map_err(|e| BackfillError::Transport(e.to_string().into()))? { false => Ok(None), true => Ok(Some(outcome?)), @@ -772,7 +796,7 @@ async fn blocks_from_whole_car( FullRepoOutcome::Inactive(status) => { Err(miette::miette!("repo went {status:?} while its records were refetched").into()) } - FullRepoOutcome::Car(_) => { + FullRepoOutcome::Car { .. } => { Err(miette::miette!("a filtered getRepo handed back the whole body").into()) } } @@ -786,8 +810,12 @@ const ERROR_BODY_MAX_BYTES: usize = 64 * 1024; /// rate-limit failures are surfaced as [`BackfillError`] instead. #[derive(Debug)] enum FullRepoOutcome { - /// every block of the car, within the configured size ceiling. - Car(ParsedCar), + /// every block of the car, within the configured size ceiling, and the share of the + /// backfill budget it took + Car { + car: ParsedCar, + budget: Option, + }, /// the parts of the car a filtered backfill can use, kept as the body streamed past. Filtered(Box), /// the PDS has no repository for this DID (`RepoNotFound`). this is a fetch result, not @@ -968,6 +996,10 @@ mod tests { Bytes::from(bytes) } + fn whole_car(budget: &BackfillBudget) -> RepoCar { + RepoCar::Whole(WholeCar::new(true), budget.clone()) + } + fn tangled_only() -> FilteredCar { let mut filter = crate::filter::FilterConfig::new(FilterMode::Filter); filter.collections = vec![SmolStr::new("sh.tangled.repo")]; @@ -978,7 +1010,12 @@ mod tests { pds: &url::Url, max_body_bytes: usize, ) -> Result { - fetch_with(pds, max_body_bytes, RepoCar::Whole(WholeCar::new(true))).await + fetch_with( + pds, + max_body_bytes, + whole_car(&BackfillBudget::new(usize::MAX)), + ) + .await } async fn fetch_with( @@ -1104,9 +1141,9 @@ mod tests { let pds = spawn_get_repo(StatusCode::OK, car.to_vec()).await; let whole = crate::car::parse_car(car, true).unwrap(); match fetch(&pds, 1024).await.unwrap() { - FullRepoOutcome::Car(parsed) => { - assert_eq!(parsed.root, whole.root); - assert_eq!(parsed.blocks.map(), whole.blocks.map()); + FullRepoOutcome::Car { car, .. } => { + assert_eq!(car.root, whole.root); + assert_eq!(car.blocks.map(), whole.blocks.map()); } other => panic!("expected Car, got {other:?}"), } @@ -1128,7 +1165,7 @@ mod tests { let pds = serve(app).await; for car in [ - RepoCar::Whole(WholeCar::new(true)), + whole_car(&BackfillBudget::new(usize::MAX)), RepoCar::Filtered(tangled_only()), ] { let fetch = fetch_with(&pds, 1024, car); @@ -1140,6 +1177,45 @@ mod tests { } } + #[tokio::test] + async fn a_whole_car_holds_its_budget_until_dropped() { + let car = commit_only_car(false).await; + let budget = BackfillBudget::new(usize::MAX); + // one sized body, and one only seen as it arrives + let sized = spawn_get_repo(StatusCode::OK, car.to_vec()).await; + let chunked = + spawn_streamed_get_repo(car.chunks(7).map(Bytes::copy_from_slice).collect(), false) + .await; + for pds in [sized, chunked] { + let outcome = fetch_with(&pds, 1024, whole_car(&budget)).await.unwrap(); + assert!(matches!( + outcome, + FullRepoOutcome::Car { + budget: Some(_), + .. + } + )); + assert_eq!(budget.used(), peak_for(car.len()), "{pds}"); + drop(outcome); + assert_eq!(budget.used(), 0, "{pds}"); + } + } + + #[tokio::test] + async fn a_whole_car_that_fails_partway_gives_its_budget_back() { + let budget = BackfillBudget::new(usize::MAX); + let car = commit_only_car(false).await; + let half = car.slice(..car.len() / 2); + let dropped = spawn_streamed_get_repo(vec![half], true).await; + let forged = spawn_streamed_get_repo(vec![commit_only_car(true).await], false).await; + for pds in [dropped, forged] { + fetch_with(&pds, 1024, whole_car(&budget)) + .await + .unwrap_err(); + assert_eq!(budget.used(), 0, "{pds}"); + } + } + #[tokio::test] async fn full_get_repo_rejects_oversized_body() { let pds = spawn_get_repo(StatusCode::OK, vec![0u8; 4096]).await; diff --git a/src/backfill/worker/task.rs b/src/backfill/worker/task.rs index 964a2cf..0aa0fc0 100644 --- a/src/backfill/worker/task.rs +++ b/src/backfill/worker/task.rs @@ -1231,6 +1231,44 @@ mod tests { Ok(()) } + #[tokio::test] + async fn a_full_backfill_holds_its_budget_until_its_batch_commits() -> miette::Result<()> { + let body = post("a"); + let car = build_car("3jzfcijpj2z2a", &[("a", body.as_slice())]).await?; + let fixture = Fixture::new().await?; + fixture.seed(&RepoState::backfilling().into_static(), 9, &[])?; + fixture.pds.serve_car(car.bytes.clone()).await; + + // the repo lock, held from a thread so the persist waits on it after the fetch is done + let (locked_tx, locked) = tokio::sync::oneshot::channel(); + let (release, release_rx) = std::sync::mpsc::channel::<()>(); + let lock = async { + let state = fixture.state.clone(); + let did = fixture.did.clone(); + std::thread::spawn(move || { + let mut txn = db::Txn::new(&state.db); + txn.hold_repo_write_lock(&did); + locked_tx.send(()).unwrap(); + release_rx.recv().unwrap(); + }); + Ok(()) + }; + let budget = fixture.state.backfill_budget.clone(); + let check = async { + locked.await.unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + let held = budget.used(); + release.send(()).unwrap(); + held + }; + let (result, held) = tokio::join!(fixture.run_changing(keys::pending_key(9), lock), check); + + assert_eq!(result??, TaskDisposition::Finished); + assert_eq!(held, crate::backfill::budget::peak_for(car.bytes.len())); + assert_eq!(budget.used(), 0); + Ok(()) + } + /// a repo with a record filed under a collection other than its `$type`, which a filtered /// car drops and the scan then finds missing async fn car_with_a_misfiled_record() -> miette::Result<(BuiltCar, Vec)> { diff --git a/src/config.rs b/src/config.rs index ee0e80b..5a4183a 100644 --- a/src/config.rs +++ b/src/config.rs @@ -123,6 +123,11 @@ pub struct Config { /// lower this on memory-constrained hosts running high backfill concurrency. /// set via `HYDRANT_MAX_CAR_BODY_BYTES`. pub max_car_body_bytes: usize, + /// memory that whole-car backfills may hold at once, counted as each car's expected peak + /// (a few times its size). a backfill whose car would go past it waits for others to + /// finish, except the oldest one running, which always goes ahead so big repos can't + /// block each other. set via `HYDRANT_BACKFILL_MEMORY_BYTES`. + pub backfill_memory_bytes: usize, /// maximum number of concurrent backfill tasks. /// set via `HYDRANT_BACKFILL_CONCURRENCY_LIMIT`. pub backfill_concurrency_limit: usize, @@ -346,6 +351,9 @@ impl Default for Config { // 2 GiB safety ceiling: never rejects a legitimate repository, only a runaway or // malicious response. memory-constrained/high-concurrency hosts should lower it. max_car_body_bytes: 2 * 1024 * 1024 * 1024, + // a few repos the size of the biggest accounts at once, while the usual repo of a + // few hundred kilobytes never notices it, and one oversized repo still runs alone + backfill_memory_bytes: 4 * 1024 * 1024 * 1024, backfill_concurrency_limit: 16, backfill_strategy: BackfillStrategy::Auto, backfill_proxies: vec![], @@ -613,6 +621,11 @@ impl fmt::Display for Config { "max car body", format_args!("{} mb", self.max_car_body_bytes / (1024 * 1024)) )?; + config_line!( + f, + "backfill memory", + format_args!("{} mb", self.backfill_memory_bytes / (1024 * 1024)) + )?; config_line!(f, "ephemeral", self.ephemeral)?; if let Some(ttl) = self.history_ttl { config_line!(f, "history ttl", humantime::format_duration(ttl))?; diff --git a/src/config/env.rs b/src/config/env.rs index 150f412..9a46cf4 100644 --- a/src/config/env.rs +++ b/src/config/env.rs @@ -155,6 +155,11 @@ impl Config { sec ); let max_car_body_bytes = cfg!(lookup, "MAX_CAR_BODY_BYTES", defaults.max_car_body_bytes); + let backfill_memory_bytes = cfg!( + lookup, + "BACKFILL_MEMORY_BYTES", + defaults.backfill_memory_bytes + ); let ephemeral: bool = cfg!(lookup, "EPHEMERAL", defaults.ephemeral); let ephemeral_ttl = cfg!(lookup, "EPHEMERAL_TTL", defaults.ephemeral_ttl, sec); @@ -424,6 +429,7 @@ impl Config { enable_backfill, repo_fetch_timeout, max_car_body_bytes, + backfill_memory_bytes, backfill_concurrency_limit, backfill_strategy, backfill_proxies, diff --git a/src/state.rs b/src/state.rs index 727c840..fd645ae 100644 --- a/src/state.rs +++ b/src/state.rs @@ -69,6 +69,8 @@ pub struct AppState { pub only_index_links: bool, pub verify_cids: bool, pub max_car_body_bytes: usize, + #[cfg(feature = "indexer")] + pub(crate) backfill_budget: crate::backfill::budget::BackfillBudget, pub get_repo_semaphore: Semaphore, pub maintenance_timeout: Duration, pub(crate) stop: Stop, @@ -170,6 +172,10 @@ impl AppState { only_index_links: config.only_index_links, verify_cids: config.verify_cids, max_car_body_bytes: config.max_car_body_bytes, + #[cfg(feature = "indexer")] + backfill_budget: crate::backfill::budget::BackfillBudget::new( + config.backfill_memory_bytes, + ), get_repo_semaphore: Semaphore::new(config.get_repo_concurrency_limit), maintenance_timeout: config.maintenance_timeout, stop: Stop::default(),