0 0 ⚡ perf: replace polling loops with event-driven wakes Tahinli/forge
⚡ perf: replace polling loops with event-driven wakes
- workers sleep until their next due time, with a 15-minute backstop
- ci claim wakes all runners
11 changed files +278 -114 @@ -450,6 +450,21 @@ impl Store { 450 450 finish(tx, outcome).await 451 451 } 452 452 453 + /// The earliest `next_attempt_at` among pending and failed deliveries, or `None` when none is 454 + /// pending. A claimed delivery carries its lease expiry here, so the lease counts as due time. 455 + pub async fn next_delivery_due(&self) -> Result<Option<String>> { 456 + let conn = self.read_conn().await; 457 + fetch_opt( 458 + &conn, 459 + "SELECT MIN(next_attempt_at) FROM webhook_deliveries \ 460 + WHERE state IN ('pending', 'failed')", 461 + (), 462 + |row| col_opt_text(row, 0), 463 + ) 464 + .await 465 + .map(Option::flatten) 466 + } 467 + 453 468 /// Records the outcome of one attempt. See [`AttemptOutcome`] for the resulting state. 454 469 /// On success `delivered_at` is stamped and `next_attempt_at` is cleared. The response body 455 470 /// is truncated to 64 KiB. `NotFound` if the delivery does not exist.
@@ -1,8 +1,10 @@ 1 1 //! The advisory worker. When `[advisories] enabled = true`, it keeps an OSV dump per configured 2 -//! ecosystem in `<data_dir>/advisories/<ecosystem>/all.zip`. A dump older than `refresh_hours` is 3 -//! downloaded again (capped at [`MAX_DUMP_BYTES`], with timeouts, no redirects). A dump that is 4 -//! newer than the last import is imported, and then every repo is rescanned. It checks once an 5 -//! hour, and once at boot, so an existing dump is imported without a download. 2 +//! ecosystem in `<data_dir>/advisories/<ecosystem>/all.zip`. Each dump is downloaded again 3 +//! `refresh_hours` after its last refresh (the file's modification time), sleeping until then 4 +//! (capped at [`MAX_DUMP_BYTES`], with timeouts, no redirects). A failed download is retried with 5 +//! a backoff that doubles from [`RETRY_FIRST`] up to [`RETRY_MAX`]. A dump that is newer than the 6 +//! last import is imported, and then every repo is rescanned. At boot, a fresh dump is imported 7 +//! without a download. 6 8 7 9 use std::collections::HashMap; 8 10 use std::fs::{self, File}; @@ -16,8 +18,10 @@ use forge_core::{Error, Result}; 16 18 use super::scan::scan_all; 17 19 use crate::state::AppState; 18 20 19 -/// How often the worker checks the dumps. 20 -const CHECK_EVERY: Duration = Duration::from_secs(60 * 60); 21 +/// First wait after a failed download. It doubles with each further failure. 22 +const RETRY_FIRST: Duration = Duration::from_secs(5 * 60); 23 +/// Longest wait after a failed download. 24 +const RETRY_MAX: Duration = Duration::from_secs(60 * 60); 21 25 /// Largest dump accepted, in bytes. 22 26 const MAX_DUMP_BYTES: u64 = 500 * 1024 * 1024; 23 27 const CONNECT_TIMEOUT: Duration = Duration::from_secs(30); @@ -57,40 +61,83 @@ async fn run(state: AppState) { 57 61 return; 58 62 } 59 63 }; 64 + let refresh_after = Duration::from_secs(state.config.advisories.refresh_hours.max(1) * 3600); 60 65 let mut imported: HashMap<String, SystemTime> = HashMap::new(); 66 + // When each ecosystem's dump is next downloaded, and how many downloads in a row failed. 67 + let mut due: HashMap<String, SystemTime> = HashMap::new(); 68 + let mut failures: HashMap<String, u32> = HashMap::new(); 61 69 loop { 70 + let mut wake_at: Option<SystemTime> = None; 62 71 for ecosystem in &state.config.advisories.ecosystems { 63 72 if !KNOWN_ECOSYSTEMS.contains(&ecosystem.as_str()) { 64 73 tracing::warn!(ecosystem = %ecosystem, "advisories: unknown ecosystem, skipped"); 65 74 continue; 66 75 } 67 - if let Err(e) = refresh_one(&state, &client, ecosystem, &mut imported).await { 68 - tracing::error!(error = %e, ecosystem = %ecosystem, "advisories: refresh failed"); 76 + let path = dump_path(&state.config.data_dir, ecosystem); 77 + let mut refresh_at = *due 78 + .entry(ecosystem.clone()) 79 + .or_insert_with(|| next_refresh_at(&path, refresh_after)); 80 + let now = SystemTime::now(); 81 + if refresh_at <= now { 82 + refresh_at = match download(&client, ecosystem, &path).await { 83 + Ok(()) => { 84 + failures.remove(ecosystem); 85 + next_refresh_at(&path, refresh_after) 86 + } 87 + Err(e) => { 88 + tracing::error!(error = %e, ecosystem = %ecosystem, "advisories: refresh failed"); 89 + let count = failures.entry(ecosystem.clone()).or_insert(0); 90 + *count += 1; 91 + now + retry_delay(*count) 92 + } 93 + }; 94 + due.insert(ecosystem.clone(), refresh_at); 69 95 } 96 + if let Err(e) = import_if_changed(&state, ecosystem, &path, &mut imported).await { 97 + tracing::error!(error = %e, ecosystem = %ecosystem, "advisories: import failed"); 98 + } 99 + wake_at = Some(wake_at.map_or(refresh_at, |at| at.min(refresh_at))); 70 100 } 71 - tokio::time::sleep(CHECK_EVERY).await; 101 + let Some(wake_at) = wake_at else { 102 + tracing::info!("advisories: no known ecosystems configured, worker stopped"); 103 + return; 104 + }; 105 + let wait = wake_at 106 + .duration_since(SystemTime::now()) 107 + .unwrap_or(Duration::ZERO); 108 + tokio::time::sleep(wait).await; 72 109 } 73 110 } 74 111 75 -/// Downloads `ecosystem`'s dump if it is stale, then imports it if it changed since the last import. 76 -async fn refresh_one( 112 +/// When the dump at `path` is due for its next download: `refresh_after` past its last refresh. 113 +/// A missing dump is due now. 114 +fn next_refresh_at(path: &Path, refresh_after: Duration) -> SystemTime { 115 + match modified_time(path) { 116 + None => SystemTime::now(), 117 + Some(last) => last + refresh_after, 118 + } 119 +} 120 + 121 +/// The wait after the `failures`-th failed download in a row: 5 minutes, doubling, capped at an hour. 122 +fn retry_delay(failures: u32) -> Duration { 123 + let shift = failures.saturating_sub(1).min(16); 124 + (RETRY_FIRST * 2u32.pow(shift)).min(RETRY_MAX) 125 +} 126 + 127 +/// Imports the dump of `ecosystem` if it changed since the last import, then rescans every repo. 128 +async fn import_if_changed( 77 129 state: &AppState, 78 - client: &reqwest::Client, 79 130 ecosystem: &str, 131 + path: &Path, 80 132 imported: &mut HashMap<String, SystemTime>, 81 133 ) -> Result<()> { 82 - let path = dump_path(&state.config.data_dir, ecosystem); 83 - let max_age = Duration::from_secs(state.config.advisories.refresh_hours.max(1) * 3600); 84 - if is_older_than(&path, max_age) { 85 - download(client, ecosystem, &path).await?; 86 - } 87 - let Some(modified) = modified_time(&path) else { 134 + let Some(modified) = modified_time(path) else { 88 135 return Ok(()); 89 136 }; 90 137 if imported.get(ecosystem) == Some(&modified) { 91 138 return Ok(()); 92 139 } 93 - import(state, ecosystem, &path).await?; 140 + import(state, ecosystem, path).await?; 94 141 imported.insert(ecosystem.to_owned(), modified); 95 142 scan_all(state).await; 96 143 Ok(()) @@ -100,16 +147,6 @@ fn modified_time(path: &Path) -> Option<SystemTime> { 100 147 fs::metadata(path).and_then(|m| m.modified()).ok() 101 148 } 102 149 103 -/// True when the file is missing or was written more than `max_age` ago. 104 -fn is_older_than(path: &Path, max_age: Duration) -> bool { 105 - match modified_time(path) { 106 - None => true, 107 - Some(at) => SystemTime::now() 108 - .duration_since(at) 109 - .is_ok_and(|age| age > max_age), 110 - } 111 -} 112 - 113 150 /// Streams the dump to `<path>.part` and renames it into place once complete. A response that is 114 151 /// too large, or that fails midway, leaves the previous dump in place. 115 152 async fn download(client: &reqwest::Client, ecosystem: &str, path: &Path) -> Result<()> {
@@ -17,10 +17,11 @@ use super::{ 17 17 }; 18 18 use crate::state::AppState; 19 19 20 -/// How often the scheduler looks at the clock. 21 -const TICK: Duration = Duration::from_secs(30); 22 20 /// After a failed run, the next try waits at least this long. 23 21 const RETRY_AFTER_SECS: i64 = 60 * 60; 22 +/// Pause after each pass. A due job that did not clear its due time (it could not take the run 23 +/// slot while a manual run held it, say) waits this long instead of spinning the loop. 24 +const RECHECK: Duration = Duration::from_secs(30); 24 25 25 26 /// The one run slot of this process. 26 27 #[derive(Default)] @@ -101,61 +102,87 @@ pub fn secs_until_due( 101 102 } 102 103 103 104 /// Starts the scheduler loop. The caller checks that `[backup] dir` is set. 105 +/// 106 +/// Each pass sleeps until the earliest of the backup and fsck is due, runs what is due, and then 107 +/// works out the next instant from the run history again. 104 108 pub fn spawn(state: AppState) -> tokio::task::JoinHandle<()> { 105 109 tokio::spawn(async move { 106 110 loop { 107 - tick(&state).await; 108 - tokio::time::sleep(TICK).await; 111 + match next_due_in(&state).await { 112 + Ok(wait) => { 113 + let until = tokio::time::Instant::now() 114 + + Duration::from_secs(u64::try_from(wait).unwrap_or(0)); 115 + tokio::time::sleep_until(until).await; 116 + } 117 + Err(e) => { 118 + tracing::error!(error = %e, "backup scheduler could not read its history"); 119 + tokio::time::sleep(RECHECK).await; 120 + continue; 121 + } 122 + } 123 + run_due(&state).await; 124 + tokio::time::sleep(RECHECK).await; 109 125 } 110 126 }) 111 127 } 112 128 113 -/// One pass of the loop: runs the backup and the fsck when either is due. 114 -async fn tick(state: &AppState) { 129 +/// How often a job of this kind is due, in seconds. 130 +fn interval_secs(state: &AppState, kind: RunKind) -> i64 { 115 131 let cfg = &state.config.backup; 116 - let backup_every = i64::try_from(cfg.interval_hours.saturating_mul(3600)).unwrap_or(i64::MAX); 117 - let fsck_every = 118 - i64::try_from(cfg.fsck_interval_days.saturating_mul(86_400)).unwrap_or(i64::MAX); 119 - let backup_wait = match due_inputs(state, RunKind::Backup).await { 120 - Ok((ok, attempt)) => secs_until_due( 121 - OffsetDateTime::now_utc(), 122 - ok.as_ref(), 123 - attempt.as_ref(), 124 - backup_every, 125 - RETRY_AFTER_SECS, 126 - ), 127 - Err(e) => { 128 - tracing::error!(error = %e, "backup scheduler could not read its history"); 129 - return; 132 + match kind { 133 + RunKind::Backup => { 134 + i64::try_from(cfg.interval_hours.saturating_mul(3600)).unwrap_or(i64::MAX) 130 135 } 131 - }; 132 - if backup_wait == 0 { 133 - // A manual run holding the slot just means this run waits for the next tick. 134 - if let Ok(guard) = state.jobs.claim(RunKind::Backup) { 135 - if let Err(e) = scheduled_backup(state).await { 136 - tracing::error!(error = %e, "scheduled backup failed"); 137 - } 138 - drop(guard); 136 + RunKind::Fsck => { 137 + i64::try_from(cfg.fsck_interval_days.saturating_mul(86_400)).unwrap_or(i64::MAX) 139 138 } 140 139 } 141 - let fsck_wait = match due_inputs(state, RunKind::Fsck).await { 142 - Ok((ok, attempt)) => secs_until_due( 143 - OffsetDateTime::now_utc(), 144 - ok.as_ref(), 145 - attempt.as_ref(), 146 - fsck_every, 147 - RETRY_AFTER_SECS, 148 - ), 149 - Err(e) => { 150 - tracing::error!(error = %e, "fsck scheduler could not read its history"); 151 - return; 140 +} 141 + 142 +/// Seconds until a job of this kind is due, from its run history. Zero means run now. 143 +async fn secs_until_run(state: &AppState, kind: RunKind) -> Result<i64> { 144 + let (ok, attempt) = due_inputs(state, kind).await?; 145 + Ok(secs_until_due( 146 + OffsetDateTime::now_utc(), 147 + ok.as_ref(), 148 + attempt.as_ref(), 149 + interval_secs(state, kind), 150 + RETRY_AFTER_SECS, 151 + )) 152 +} 153 + 154 +/// Seconds until the earlier of the backup and the fsck is due. Zero means one is due now. 155 +async fn next_due_in(state: &AppState) -> Result<i64> { 156 + let backup = secs_until_run(state, RunKind::Backup).await?; 157 + let fsck = secs_until_run(state, RunKind::Fsck).await?; 158 + Ok(backup.min(fsck)) 159 +} 160 + 161 +/// Runs each job that is due now. A job that cannot take the run slot, because a manual run holds 162 +/// it, is left for the next pass. 163 +async fn run_due(state: &AppState) { 164 + for kind in [RunKind::Backup, RunKind::Fsck] { 165 + let due = match secs_until_run(state, kind).await { 166 + Ok(wait) => wait == 0, 167 + Err(e) => { 168 + tracing::error!(error = %e, kind = kind.as_str(), "scheduler could not read its history"); 169 + false 170 + } 171 + }; 172 + if !due { 173 + continue; 152 174 } 153 - }; 154 - if fsck_wait == 0 155 - && let Ok(guard) = state.jobs.claim(RunKind::Fsck) 156 - { 157 - if let Err(e) = fsck_job(state, RunTrigger::Schedule, false).await { 158 - tracing::error!(error = %e, "scheduled fsck failed"); 175 + let Ok(guard) = state.jobs.claim(kind) else { 176 + continue; 177 + }; 178 + let outcome = match kind { 179 + RunKind::Backup => scheduled_backup(state).await.map(|_| ()), 180 + RunKind::Fsck => fsck_job(state, RunTrigger::Schedule, false) 181 + .await 182 + .map(|_| ()), 183 + }; 184 + if let Err(e) = outcome { 185 + tracing::error!(error = %e, kind = kind.as_str(), "scheduled run failed"); 159 186 } 160 187 drop(guard); 161 188 } @@ -463,6 +490,13 @@ mod tests { 463 490 assert_eq!(state.jobs.running(), None); 464 491 } 465 492 493 + #[tokio::test] 494 + async fn a_fresh_install_wakes_at_once() { 495 + let dir = tempfile::tempdir().expect("tempdir"); 496 + let state = crate::test_support::test_state(dir.path()).await; 497 + assert_eq!(next_due_in(&state).await.expect("next due"), 0); 498 + } 499 + 466 500 #[tokio::test] 467 501 async fn scheduled_backup_is_recorded_verified_and_pinged() { 468 502 let dir = tempfile::tempdir().expect("tempdir");
@@ -58,6 +58,8 @@ pub async fn reap_before(state: &AppState, older_than: &str) -> usize { 58 58 let mut seen = HashSet::new(); 59 59 let mut wake = false; 60 60 for job in &jobs { 61 + state.ci_live.close_job(&job.id); 62 + state.ci_live.bump_run(&job.run_id); 61 63 if !seen.insert(job.run_id.clone()) { 62 64 continue; 63 65 } @@ -73,7 +75,7 @@ pub async fn reap_before(state: &AppState, older_than: &str) -> usize { 73 75 } 74 76 } 75 77 if wake { 76 - state.ci_waker.notify_one(); 78 + state.ci_waker.notify_waiters(); 77 79 } 78 80 if !jobs.is_empty() { 79 81 tracing::warn!(count = jobs.len(), "ci: reaped lost jobs");
@@ -243,7 +243,7 @@ async fn create( 243 243 .await; 244 244 } 245 245 if sync_run_statuses(state, &run).await? { 246 - state.ci_waker.notify_one(); 246 + state.ci_waker.notify_waiters(); 247 247 } 248 248 Ok(run) 249 249 }
@@ -13,8 +13,8 @@ use time::format_description::well_known::Rfc3339; 13 13 14 14 use crate::state::AppState; 15 15 16 -/// The longest the worker sleeps when nothing wakes it. 17 -const POLL: Duration = Duration::from_secs(60); 16 +/// The longest the worker sleeps. It guards against a missed wake, not polling. 17 +const BACKSTOP: Duration = Duration::from_secs(15 * 60); 18 18 /// The shortest sleep, so a due time that has already passed does not spin the loop. 19 19 const MIN_SLEEP: Duration = Duration::from_secs(1); 20 20 /// Messages leased per pass. @@ -33,7 +33,7 @@ async fn run(state: AppState) { 33 33 Ok(nap) => nap, 34 34 Err(e) => { 35 35 tracing::error!(error = %e, "mail: could not run a pass"); 36 - POLL 36 + BACKSTOP 37 37 } 38 38 }; 39 39 tokio::select! { @@ -47,7 +47,8 @@ async fn run(state: AppState) { 47 47 async fn tick(state: &AppState) -> forge_core::Result<Duration> { 48 48 let settings = state.store.mail_settings().await?; 49 49 if !settings.enabled { 50 - return Ok(POLL); 50 + // Turning mail on notifies the waker; the backstop only guards a missed wake. 51 + return Ok(BACKSTOP); 51 52 } 52 53 let now = now_stamp(); 53 54 let lease_until = stamp_after(LEASE); @@ -93,13 +94,13 @@ async fn deliver(state: &AppState, entry: &OutboxEntry) { 93 94 } 94 95 95 96 /// How long the worker sleeps: until `next_due` (an RFC 3339 stamp), clamped to 96 -/// `MIN_SLEEP..=POLL`. `None` or an unparsable stamp gives `POLL`. 97 +/// `MIN_SLEEP..=BACKSTOP`. `None` (nothing queued) or an unparsable stamp gives `BACKSTOP`. 97 98 pub fn sleep_for(next_due: Option<&str>, now: OffsetDateTime) -> Duration { 98 99 let Some(due) = next_due.and_then(|s| OffsetDateTime::parse(s, &Rfc3339).ok()) else { 99 - return POLL; 100 + return BACKSTOP; 100 101 }; 101 102 let until = Duration::try_from(due - now).unwrap_or(Duration::ZERO); 102 - until.clamp(MIN_SLEEP, POLL) 103 + until.clamp(MIN_SLEEP, BACKSTOP) 103 104 } 104 105 105 106 #[cfg(test)] @@ -140,14 +141,14 @@ mod tests { 140 141 141 142 let now = datetime!(2026-10-15 12:00:00 UTC); 142 143 let stamp = |t: OffsetDateTime| t.format(&Rfc3339).expect("stamp"); 143 - assert_eq!(sleep_for(None, now), POLL); 144 + assert_eq!(sleep_for(None, now), BACKSTOP); 144 145 let in_10s = stamp(now + time::Duration::seconds(10)); 145 146 assert_eq!(sleep_for(Some(&in_10s), now), Duration::from_secs(10)); 146 147 let in_1h = stamp(now + time::Duration::hours(1)); 147 - assert_eq!(sleep_for(Some(&in_1h), now), POLL); 148 + assert_eq!(sleep_for(Some(&in_1h), now), BACKSTOP); 148 149 let past = stamp(now - time::Duration::seconds(5)); 149 150 assert_eq!(sleep_for(Some(&past), now), MIN_SLEEP); 150 - assert_eq!(sleep_for(Some("soon"), now), POLL); 151 + assert_eq!(sleep_for(Some("soon"), now), BACKSTOP); 151 152 } 152 153 153 154 #[tokio::test] @@ -197,7 +198,7 @@ mod tests { 197 198 enqueue(&state, "k", "a@example.com", "Hi", "body") 198 199 .await 199 200 .expect("queued"); 200 - assert_eq!(tick(&state).await.expect("pass"), POLL); 201 + assert_eq!(tick(&state).await.expect("pass"), BACKSTOP); 201 202 assert!(fake.sent().is_empty()); 202 203 let row = state.store.recent_mail(10).await.expect("rows"); 203 204 assert_eq!(row[0].status(), OutboxStatus::Queued);
@@ -1,6 +1,6 @@ 1 -//! The mirror sync worker. It polls for due mirrors every minute, and wakes at once when 2 -//! `AppState::mirror_waker` is notified (a "sync now", a new mirror, or a push). At most two runs 3 -//! are in flight, and never two on the same repo. 1 +//! The mirror sync worker. It sleeps until the next due mirror, and wakes at once when 2 +//! `AppState::mirror_waker` is notified (a "sync now", a new mirror, or a push). With nothing due 3 +//! it waits on the waker alone. At most two runs are in flight, and never two on the same repo. 4 4 5 5 use std::collections::HashSet; 6 6 use std::time::Duration; @@ -14,8 +14,8 @@ use super::{policy, run_one}; 14 14 use crate::state::AppState; 15 15 use crate::webhooks::client::WebhookPolicy; 16 16 17 -/// The longest the worker sleeps when nothing wakes it. 18 -const POLL: Duration = Duration::from_secs(60); 17 +/// The longest the worker sleeps. It guards against a missed wake, not polling. 18 +const BACKSTOP: Duration = Duration::from_secs(15 * 60); 19 19 /// The shortest sleep, so a due time that has already passed does not spin the loop. 20 20 const MIN_SLEEP: Duration = Duration::from_secs(1); 21 21 /// Runs in flight at once. @@ -44,14 +44,13 @@ async fn run(state: AppState) { 44 44 .await; 45 45 // Wake when the earliest enabled mirror is due, so a push mirror queued with a short 46 46 // debounce runs on time instead of waiting for the next poll. 47 - let next = match state.store.next_mirror_due().await { 48 - Ok(next) => next, 47 + let nap = match state.store.next_mirror_due().await { 48 + Ok(next) => sleep_for(next.as_deref(), OffsetDateTime::now_utc()), 49 49 Err(e) => { 50 50 tracing::error!(error = %e, "mirrors: could not read the next due time"); 51 - None 51 + BACKSTOP 52 52 } 53 53 }; 54 - let nap = sleep_for(next.as_deref(), OffsetDateTime::now_utc()); 55 54 tokio::select! { 56 55 _ = tokio::time::sleep(nap) => {} 57 56 _ = state.mirror_waker.notified() => {} @@ -64,14 +63,14 @@ async fn run(state: AppState) { 64 63 } 65 64 66 65 /// How long the worker sleeps before its next pass: until `next_due` (an RFC 3339 stamp, the 67 -/// earliest enabled mirror), clamped to `MIN_SLEEP..=POLL`. `None` or an unparsable stamp gives 68 -/// `POLL`. 66 +/// earliest enabled mirror), clamped to `MIN_SLEEP..=BACKSTOP`. `None` (nothing enabled) or an 67 +/// unparsable stamp gives `BACKSTOP`. 69 68 pub fn sleep_for(next_due: Option<&str>, now: OffsetDateTime) -> Duration { 70 69 let Some(due) = next_due.and_then(|s| OffsetDateTime::parse(s, &Rfc3339).ok()) else { 71 - return POLL; 70 + return BACKSTOP; 72 71 }; 73 72 let until = Duration::try_from(due - now).unwrap_or(Duration::ZERO); 74 - until.clamp(MIN_SLEEP, POLL) 73 + until.clamp(MIN_SLEEP, BACKSTOP) 75 74 } 76 75 77 76 /// Starts runs for due mirrors until the concurrency limit, skipping mirrors and repos already @@ -130,22 +129,22 @@ mod tests { 130 129 131 130 let now = datetime!(2026-10-15 12:00:00 UTC); 132 131 let stamp = |t: OffsetDateTime| t.format(&Rfc3339).expect("stamp"); 133 - // Nothing enabled: the full poll interval. 134 - assert_eq!(super::sleep_for(None, now), super::POLL); 135 - // A push queued with a 10 s debounce: sleep until it is due, not the full poll. 132 + // Nothing enabled: the backstop. 133 + assert_eq!(super::sleep_for(None, now), super::BACKSTOP); 134 + // A push queued with a 10 s debounce: sleep until it is due. 136 135 let in_10s = stamp(now + time::Duration::seconds(10)); 137 136 assert_eq!( 138 137 super::sleep_for(Some(&in_10s), now), 139 138 Duration::from_secs(10) 140 139 ); 141 - // Due later than the poll interval: wake at the poll interval. 140 + // Due later than the backstop: wake at the backstop. 142 141 let in_1h = stamp(now + time::Duration::hours(1)); 143 - assert_eq!(super::sleep_for(Some(&in_1h), now), super::POLL); 142 + assert_eq!(super::sleep_for(Some(&in_1h), now), super::BACKSTOP); 144 143 // Already due (or in flight and still due): do not spin, wait the minimum. 145 144 let past = stamp(now - time::Duration::seconds(5)); 146 145 assert_eq!(super::sleep_for(Some(&past), now), super::MIN_SLEEP); 147 - // A stamp that does not parse falls back to the poll interval. 148 - assert_eq!(super::sleep_for(Some("soon"), now), super::POLL); 146 + // A stamp that does not parse falls back to the backstop. 147 + assert_eq!(super::sleep_for(Some("soon"), now), super::BACKSTOP); 149 148 } 150 149 151 150 /// Plain git against a bare repo, for setting up the fake upstream.
@@ -633,8 +633,9 @@ async fn store_mirror( 633 633 .store 634 634 .schedule_mirror(&mirror.id, &now_stamp()) 635 635 .await?; 636 - state.mirror_waker.notify_one(); 637 636 } 637 + // A new mirror is due now, push or pull. 638 + state.mirror_waker.notify_one(); 638 639 Ok(Ok(mirror)) 639 640 } 640 641 @@ -931,6 +932,8 @@ async fn mirror_update(cx: &Cx, Form(form): Form<EditForm>) -> Result<impl View> 931 932 form.enabled.is_some(), 932 933 ) 933 934 .await?; 935 + // The interval and the enabled flag decide when this mirror runs next. 936 + state.mirror_waker.notify_one(); 934 937 audit::record( 935 938 &state.store, 936 939 Some(cx),
@@ -547,6 +547,16 @@ impl ReplicationQueue { 547 547 due 548 548 } 549 549 550 + /// The earliest debounce due time among the queued repos, or `None` when nothing is queued. 551 + pub fn next_due(&self) -> Option<Instant> { 552 + self.pending 553 + .lock() 554 + .unwrap_or_else(PoisonError::into_inner) 555 + .values() 556 + .min() 557 + .copied() 558 + } 559 + 550 560 /// Asks the worker for a full reconcile pass as soon as it can. 551 561 pub fn request_reconcile(&self) { 552 562 self.reconcile.store(true, Ordering::SeqCst); @@ -762,6 +772,7 @@ mod tests { 762 772 let later = Instant::now() + PUSH_DEBOUNCE + Duration::from_millis(10); 763 773 assert_eq!(queue.take_due(later), ["a", "b"]); 764 774 assert!(queue.take_due(later).is_empty(), "drained"); 775 + assert_eq!(queue.next_due(), None, "nothing left to wait for"); 765 776 assert!(!queue.take_reconcile_request()); 766 777 queue.request_reconcile(); 767 778 assert!(queue.take_reconcile_request());
@@ -22,8 +22,8 @@ use super::{ 22 22 use crate::backup::healthcheck; 23 23 use crate::state::AppState; 24 24 25 -/// How often the worker checks its queue and timers when nothing wakes it. 26 -const TICK: Duration = Duration::from_secs(1); 25 +/// Longest the worker sleeps with nothing due. Guards against a missed wake, not polling. 26 +const BACKSTOP: Duration = Duration::from_secs(15 * 60); 27 27 28 28 /// Everything a replication pass needs. The web side builds one from `AppState`; tests build one 29 29 /// around a `LocalTransport`. @@ -83,9 +83,21 @@ async fn run(state: AppState) { 83 83 let mut next_db = Instant::now(); 84 84 let mut next_reconcile = Instant::now(); 85 85 loop { 86 + // Sleep until the earliest debounce or calendar timer, or until something is queued. 87 + let backstop = Instant::now() + BACKSTOP; 88 + let next = [ 89 + state.replication.next_due(), 90 + Some(next_db), 91 + Some(next_reconcile), 92 + Some(backstop), 93 + ] 94 + .into_iter() 95 + .flatten() 96 + .min() 97 + .unwrap_or(backstop); 86 98 tokio::select! { 99 + () = tokio::time::sleep_until(tokio::time::Instant::from_std(next)) => {} 87 100 () = state.replication.wait() => {} 88 - () = tokio::time::sleep(TICK) => {} 89 101 } 90 102 for repo_id in state.replication.take_due(Instant::now()) { 91 103 // A repo deleted since the update is skipped; the next reconcile removes its copy.
@@ -15,6 +15,8 @@ use forge_core::time::{now_stamp, stamp_after, stamp_ago}; 15 15 use forge_core::webhooks::{AttemptOutcome, Delivery}; 16 16 use futures_util::StreamExt; 17 17 use serde_json::{Map, Value}; 18 +use time::OffsetDateTime; 19 +use time::format_description::well_known::Rfc3339; 18 20 use tokio::sync::Notify; 19 21 20 22 use super::client::{HookClient, SendOutcome, WebhookPolicy, sign, validate_url}; @@ -27,8 +29,10 @@ pub type WebhookWaker = Arc<Notify>; 27 29 const BATCH: u32 = 16; 28 30 /// Deliveries sent at once. 29 31 const CONCURRENCY: usize = 4; 30 -/// How long the worker sleeps when nothing wakes it. 31 -const IDLE_POLL: Duration = Duration::from_secs(30); 32 +/// The shortest sleep, so a due time that has already passed does not spin the loop. 33 +const MIN_SLEEP: Duration = Duration::from_secs(1); 34 +/// The longest sleep. It only guards against a missed wake, it is not polling. 35 +const BACKSTOP: Duration = Duration::from_secs(15 * 60); 32 36 /// Deliveries older than this are purged. 33 37 const RETENTION: Duration = Duration::from_secs(14 * 24 * 60 * 60); 34 38 /// How often the purge runs. @@ -86,14 +90,41 @@ async fn run(state: AppState, waker: WebhookWaker) { 86 90 }; 87 91 // A full batch may mean more is due, so go straight back for it. 88 92 if claimed < usize::try_from(BATCH).unwrap_or(usize::MAX) { 93 + // Wake when the earliest pending delivery is due (a claimed one wakes at its lease 94 + // expiry), or at the next purge. With nothing pending, sleep up to `BACKSTOP`. 95 + let next = match state.store.next_delivery_due().await { 96 + Ok(next) => next, 97 + Err(e) => { 98 + tracing::error!(error = %e, "webhooks: could not read the next due time"); 99 + None 100 + } 101 + }; 102 + let nap = next 103 + .as_deref() 104 + .map_or(BACKSTOP, |due| sleep_for(due, OffsetDateTime::now_utc())); 105 + let purge_nap = last_purge.map_or(Duration::ZERO, |at| { 106 + PURGE_EVERY.saturating_sub(at.elapsed()) 107 + }); 89 108 tokio::select! { 90 109 () = waker.notified() => {} 91 - () = tokio::time::sleep(IDLE_POLL) => {} 110 + () = tokio::time::sleep(nap) => {} 111 + () = tokio::time::sleep(purge_nap) => {} 92 112 } 93 113 } 94 114 } 95 115 } 96 116 117 +/// How long the worker sleeps until `next_due` (an RFC 3339 stamp, the earliest pending 118 +/// delivery), clamped to `MIN_SLEEP..=BACKSTOP`. An unparsable stamp gives `MIN_SLEEP`, so it is 119 +/// retried promptly rather than stalling. 120 +fn sleep_for(next_due: &str, now: OffsetDateTime) -> Duration { 121 + let Ok(due) = OffsetDateTime::parse(next_due, &Rfc3339) else { 122 + return MIN_SLEEP; 123 + }; 124 + let until = Duration::try_from(due - now).unwrap_or(Duration::ZERO); 125 + until.clamp(MIN_SLEEP, BACKSTOP) 126 +} 127 + 97 128 /// Claims up to [`BATCH`] due deliveries at `now`, sends them, and records each result. Returns 98 129 /// how many were claimed. 99 130 pub(crate) async fn process_due( @@ -499,6 +530,25 @@ mod tests { 499 530 assert!(row.next_attempt_at.is_some()); 500 531 } 501 532 533 + #[test] 534 + fn sleep_stops_at_the_next_due_time_and_never_spins() { 535 + use time::macros::datetime; 536 + 537 + let now = datetime!(2026-10-15 12:00:00 UTC); 538 + let stamp = |t: OffsetDateTime| t.format(&Rfc3339).expect("stamp"); 539 + // Due in 10 minutes: sleep until then. 540 + let in_10m = stamp(now + time::Duration::minutes(10)); 541 + assert_eq!(sleep_for(&in_10m, now), Duration::from_secs(600)); 542 + // Due later than the backstop: wake at the backstop. 543 + let in_1h = stamp(now + time::Duration::hours(1)); 544 + assert_eq!(sleep_for(&in_1h, now), BACKSTOP); 545 + // Already due, or a lease that has just passed: wait the minimum. 546 + let past = stamp(now - time::Duration::seconds(5)); 547 + assert_eq!(sleep_for(&past, now), MIN_SLEEP); 548 + // A stamp that does not parse also waits the minimum. 549 + assert_eq!(sleep_for("soon", now), MIN_SLEEP); 550 + } 551 + 502 552 #[test] 503 553 fn backoff_schedule_matches_the_contract() { 504 554 assert_eq!(next_attempt_at(0), None);