forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
19.9 KiB
1//! Scheduled backups and fsck, the run slot they share with manual and CLI triggers, and the job
2//! bodies. Every run is recorded in `backup_runs`. Only one run is active at a time: a second
3//! trigger gets "already running" while the first is in progress.
4
5use std::path::PathBuf;
6use std::sync::{Arc, Mutex, PoisonError};
7use std::time::{Duration, SystemTime};
8
9use forge_core::store::backup_runs::{BackupRun, RunKind, RunTrigger};
10use forge_core::{Error, Result};
11use time::OffsetDateTime;
12use time::format_description::well_known::Rfc3339;
13
14use super::{
15 BackupOptions, BackupSource, BackupSummary, FsckSummary, create_backup, healthcheck, retention,
16 run_fsck, verify_backup,
17};
18use crate::state::AppState;
19
20/// After a failed run, the next try waits at least this long.
21const 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.
24const RECHECK: Duration = Duration::from_secs(30);
25
26/// The one run slot of this process.
27#[derive(Default)]
28pub struct Jobs {
29 running: Mutex<Option<RunKind>>,
30}
31
32/// Holds the run slot. Dropping it frees the slot.
33pub struct JobGuard {
34 jobs: Arc<Jobs>,
35 kind: RunKind,
36}
37
38impl Jobs {
39 /// Takes the run slot for `kind`. Fails with "already running" while another run holds it.
40 pub fn claim(self: &Arc<Self>, kind: RunKind) -> Result<JobGuard> {
41 let mut slot = self.running.lock().unwrap_or_else(PoisonError::into_inner);
42 if let Some(current) = *slot {
43 return Err(Error::Conflict(format!(
44 "already running: a {} run is in progress",
45 current.as_str()
46 )));
47 }
48 *slot = Some(kind);
49 Ok(JobGuard {
50 jobs: Arc::clone(self),
51 kind,
52 })
53 }
54
55 /// The kind of run in progress, if any.
56 pub fn running(&self) -> Option<RunKind> {
57 *self.running.lock().unwrap_or_else(PoisonError::into_inner)
58 }
59}
60
61impl Drop for JobGuard {
62 fn drop(&mut self) {
63 let mut slot = self
64 .jobs
65 .running
66 .lock()
67 .unwrap_or_else(PoisonError::into_inner);
68 if *slot == Some(self.kind) {
69 *slot = None;
70 }
71 }
72}
73
74fn started_unix(run: &BackupRun) -> Option<i64> {
75 OffsetDateTime::parse(&run.started_at, &Rfc3339)
76 .ok()
77 .map(|t| t.unix_timestamp())
78}
79
80/// Seconds until a job of this kind is due. Zero means run now.
81///
82/// The base is the start of the last successful run plus `every_secs`. With no success on record
83/// the job is due at once. When the latest attempt failed, the next try also waits
84/// `retry_after_secs` from that attempt, so a broken setup does not retry every tick.
85pub fn secs_until_due(
86 now: OffsetDateTime,
87 last_ok: Option<&BackupRun>,
88 last_attempt: Option<&BackupRun>,
89 every_secs: i64,
90 retry_after_secs: i64,
91) -> i64 {
92 let mut due = match last_ok.and_then(started_unix) {
93 Some(t) => t.saturating_add(every_secs),
94 None => now.unix_timestamp(),
95 };
96 if let Some(attempt) = last_attempt.filter(|run| run.ok != Some(true))
97 && let Some(t) = started_unix(attempt)
98 {
99 due = due.max(t.saturating_add(retry_after_secs));
100 }
101 (due - now.unix_timestamp()).max(0)
102}
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.
108pub fn spawn(state: AppState) -> tokio::task::JoinHandle<()> {
109 tokio::spawn(async move {
110 loop {
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;
125 }
126 })
127}
128
129/// How often a job of this kind is due, in seconds.
130fn interval_secs(state: &AppState, kind: RunKind) -> i64 {
131 let cfg = &state.config.backup;
132 match kind {
133 RunKind::Backup => {
134 i64::try_from(cfg.interval_hours.saturating_mul(3600)).unwrap_or(i64::MAX)
135 }
136 RunKind::Fsck => {
137 i64::try_from(cfg.fsck_interval_days.saturating_mul(86_400)).unwrap_or(i64::MAX)
138 }
139 }
140}
141
142/// Seconds until a job of this kind is due, from its run history. Zero means run now.
143async 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.
155async 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.
163async 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;
174 }
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");
186 }
187 drop(guard);
188 }
189}
190
191async fn due_inputs(
192 state: &AppState,
193 kind: RunKind,
194) -> Result<(Option<BackupRun>, Option<BackupRun>)> {
195 let ok = state.store.last_ok(kind).await?;
196 let attempt = state.store.last_run(kind).await?;
197 Ok((ok, attempt))
198}
199
200/// The configured backup directory as an absolute path.
201pub fn configured_dir(state: &AppState) -> Result<PathBuf> {
202 std::path::absolute(&state.config.backup.dir).map_err(Error::Io)
203}
204
205async fn scheduled_backup(state: &AppState) -> Result<BackupSummary> {
206 let opts = BackupOptions {
207 out: configured_dir(state)?,
208 include_key: false,
209 scope: super::Scope::All,
210 };
211 backup_job(state, RunTrigger::Schedule, opts, true).await
212}
213
214/// Runs one backup set and records it. The caller holds the run slot.
215///
216/// The set is verified before the run counts as ok. `prune` applies retention to `opts.out` after
217/// a good run. Healthcheck pings are sent for scheduled runs only.
218pub async fn backup_job(
219 state: &AppState,
220 trigger: RunTrigger,
221 opts: BackupOptions,
222 prune: bool,
223) -> Result<BackupSummary> {
224 let run_id = state.store.start_run(RunKind::Backup, trigger).await?;
225 let healthcheck_url = Some(state.config.backup.healthcheck_url.as_str())
226 .filter(|url| trigger == RunTrigger::Schedule && !url.is_empty());
227 if let Some(url) = healthcheck_url {
228 healthcheck::start(url).await;
229 }
230
231 let outcome = backup_set(state, &opts).await;
232 match &outcome {
233 Ok(summary) => {
234 let path = summary.path.to_string_lossy().into_owned();
235 record_finish(state, &run_id, true, Some(&path), Some(summary.bytes), None).await;
236 }
237 Err(e) => {
238 record_finish(state, &run_id, false, None, None, Some(&e.to_string())).await;
239 }
240 }
241 if outcome.is_ok() && prune {
242 match retention::prune(&opts.out, state.config.backup.keep, SystemTime::now()) {
243 Ok(report) => tracing::info!(
244 sets = report.sets_removed.len(),
245 partials = report.partials_removed.len(),
246 "backup retention applied"
247 ),
248 Err(e) => tracing::warn!(error = %e, "backup retention failed"),
249 }
250 }
251 if let Some(url) = healthcheck_url {
252 match &outcome {
253 Ok(_) => healthcheck::success(url).await,
254 Err(e) => healthcheck::failure(url, &e.to_string()).await,
255 }
256 }
257 outcome
258}
259
260async fn backup_set(state: &AppState, opts: &BackupOptions) -> Result<BackupSummary> {
261 let source = BackupSource {
262 store: state.store.clone(),
263 git: (*state.git).clone(),
264 base_url: state.config.base_url.clone(),
265 };
266 let summary = create_backup(&source, opts).await?;
267 let report = verify_backup(&summary.path, &source.git).await?;
268 if !report.is_ok() {
269 return Err(Error::Invalid(format!(
270 "the new backup set failed verification: {}",
271 report.problems.join("; ")
272 )));
273 }
274 Ok(summary)
275}
276
277/// Runs fsck over every repo and the database, and records it. The caller holds the run slot.
278pub async fn fsck_job(state: &AppState, trigger: RunTrigger, full: bool) -> Result<FsckSummary> {
279 let run_id = state.store.start_run(RunKind::Fsck, trigger).await?;
280 let outcome = run_fsck(&state.store, &state.git, full).await;
281 match &outcome {
282 Ok(summary) if summary.is_ok() => {
283 record_finish(state, &run_id, true, None, None, None).await;
284 }
285 Ok(summary) => {
286 let text = summary.problems.join("\n");
287 record_finish(state, &run_id, false, None, None, Some(&text)).await;
288 }
289 Err(e) => {
290 record_finish(state, &run_id, false, None, None, Some(&e.to_string())).await;
291 }
292 }
293 outcome
294}
295
296/// Writes the outcome of a run. A failure to record is logged, because the run itself has already
297/// happened.
298async fn record_finish(
299 state: &AppState,
300 run_id: &str,
301 ok: bool,
302 path: Option<&str>,
303 bytes: Option<u64>,
304 error: Option<&str>,
305) {
306 if let Err(e) = state.store.finish_run(run_id, ok, path, bytes, error).await {
307 tracing::error!(error = %e, run_id, "could not record the outcome of a backup run");
308 }
309}
310
311/// Starts a manual run in the background and returns at once. The caller claimed the run slot
312/// (so it could refuse with "already running" and record the audit entry first). Fails with a
313/// config error, freeing the slot, when no backup directory is set.
314pub fn start_manual(state: &AppState, guard: JobGuard) -> Result<()> {
315 if !state.config.backup.enabled() {
316 return Err(Error::Config(
317 "backups are not configured: set [backup] dir in the config file".into(),
318 ));
319 }
320 let kind = guard.kind;
321 let state = state.clone();
322 tokio::spawn(async move {
323 let _guard = guard;
324 let outcome: Result<()> = match kind {
325 RunKind::Backup => match configured_dir(&state) {
326 Ok(out) => backup_job(
327 &state,
328 RunTrigger::Manual,
329 BackupOptions {
330 out,
331 include_key: false,
332 scope: super::Scope::All,
333 },
334 true,
335 )
336 .await
337 .map(|_| ()),
338 Err(e) => Err(e),
339 },
340 RunKind::Fsck => fsck_job(&state, RunTrigger::Manual, false)
341 .await
342 .map(|_| ()),
343 };
344 if let Err(e) = outcome {
345 tracing::error!(error = %e, kind = kind.as_str(), "manual run failed");
346 }
347 });
348 Ok(())
349}
350
351/// Runs a backup now and waits for it, holding the run slot in `guard`. For the CLI through the
352/// internal socket.
353pub async fn run_backup_now(
354 state: &AppState,
355 guard: JobGuard,
356 trigger: RunTrigger,
357 opts: BackupOptions,
358 prune: bool,
359) -> Result<BackupSummary> {
360 let _guard = guard;
361 backup_job(state, trigger, opts, prune).await
362}
363
364/// Runs fsck now and waits for it, holding the run slot in `guard`. For the CLI through the
365/// internal socket.
366pub async fn run_fsck_now(
367 state: &AppState,
368 guard: JobGuard,
369 trigger: RunTrigger,
370 full: bool,
371) -> Result<FsckSummary> {
372 let _guard = guard;
373 fsck_job(state, trigger, full).await
374}
375
376#[cfg(test)]
377mod tests {
378 use super::*;
379 use forge_core::model::Visibility;
380 use forge_git::repo::{init_bare, repo_path};
381 use tokio::io::{AsyncReadExt, AsyncWriteExt};
382 use tokio::net::TcpListener;
383
384 fn at(stamp: &str) -> OffsetDateTime {
385 OffsetDateTime::parse(stamp, &Rfc3339).expect("stamp")
386 }
387
388 fn run(kind: RunKind, started: &str, ok: Option<bool>) -> BackupRun {
389 BackupRun {
390 id: format!("id-{started}"),
391 kind,
392 trigger: RunTrigger::Schedule,
393 started_at: started.to_string(),
394 finished_at: None,
395 ok,
396 path: None,
397 bytes: None,
398 error: None,
399 }
400 }
401
402 const DAY: i64 = 86_400;
403 const HOUR: i64 = 3600;
404
405 #[test]
406 fn never_run_is_due_now() {
407 let now = at("2026-10-15T12:00:00Z");
408 assert_eq!(secs_until_due(now, None, None, DAY, HOUR), 0);
409 }
410
411 #[test]
412 fn next_run_follows_the_last_success() {
413 let now = at("2026-10-15T12:00:00Z");
414 let last = run(RunKind::Backup, "2026-10-15T06:00:00Z", Some(true));
415 // Last success six hours ago with a 24 hour interval: due in eighteen hours.
416 assert_eq!(
417 secs_until_due(now, Some(&last), Some(&last), DAY, HOUR),
418 18 * HOUR
419 );
420 let old = run(RunKind::Backup, "2026-10-13T00:00:00Z", Some(true));
421 assert_eq!(secs_until_due(now, Some(&old), Some(&old), DAY, HOUR), 0);
422 }
423
424 #[test]
425 fn a_failed_attempt_waits_before_retrying() {
426 let now = at("2026-10-15T12:00:00Z");
427 let old_ok = run(RunKind::Backup, "2026-10-13T00:00:00Z", Some(true));
428 // The latest attempt failed 10 minutes ago: the retry waits the rest of the hour.
429 let failed = run(RunKind::Backup, "2026-10-15T11:50:00Z", Some(false));
430 assert_eq!(
431 secs_until_due(now, Some(&old_ok), Some(&failed), DAY, HOUR),
432 50 * 60
433 );
434 // A run that never finished counts as a failed attempt.
435 let unfinished = run(RunKind::Backup, "2026-10-15T11:50:00Z", None);
436 assert_eq!(
437 secs_until_due(now, None, Some(&unfinished), DAY, HOUR),
438 50 * 60
439 );
440 // Failed long ago: due now.
441 let stale = run(RunKind::Backup, "2026-10-15T09:00:00Z", Some(false));
442 assert_eq!(secs_until_due(now, None, Some(&stale), DAY, HOUR), 0);
443 }
444
445 #[test]
446 fn the_slot_is_exclusive_and_freed_on_drop() {
447 let jobs = Arc::new(Jobs::default());
448 let guard = jobs.claim(RunKind::Backup).expect("first claim");
449 assert_eq!(jobs.running(), Some(RunKind::Backup));
450 match jobs.claim(RunKind::Fsck) {
451 Err(Error::Conflict(message)) => assert!(message.contains("already running")),
452 other => panic!("expected a conflict, got {:?}", other.map(|_| ())),
453 }
454 drop(guard);
455 assert_eq!(jobs.running(), None);
456 assert!(jobs.claim(RunKind::Fsck).is_ok());
457 }
458
459 /// A tiny server that answers `count` requests and records the request lines.
460 async fn ping_server(count: usize) -> (String, Arc<Mutex<Vec<String>>>) {
461 let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
462 let addr = listener.local_addr().expect("addr");
463 let seen = Arc::new(Mutex::new(Vec::new()));
464 let log = Arc::clone(&seen);
465 tokio::spawn(async move {
466 for _ in 0..count {
467 let Ok((mut sock, _)) = listener.accept().await else {
468 return;
469 };
470 let mut buf = vec![0u8; 8192];
471 let n = sock.read(&mut buf).await.unwrap_or(0);
472 let text = String::from_utf8_lossy(&buf[..n]).into_owned();
473 log.lock()
474 .unwrap_or_else(PoisonError::into_inner)
475 .push(text.lines().next().unwrap_or_default().to_owned());
476 let _ = sock
477 .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
478 .await;
479 }
480 });
481 (format!("http://{addr}/hc/token-123"), seen)
482 }
483
484 #[tokio::test]
485 async fn manual_start_without_a_dir_is_a_config_error_and_frees_the_slot() {
486 let dir = tempfile::tempdir().expect("tempdir");
487 let state = crate::test_support::test_state(dir.path()).await;
488 let guard = state.jobs.claim(RunKind::Backup).expect("claim");
489 assert!(matches!(start_manual(&state, guard), Err(Error::Config(_))));
490 assert_eq!(state.jobs.running(), None);
491 }
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
500 #[tokio::test]
501 async fn scheduled_backup_is_recorded_verified_and_pinged() {
502 let dir = tempfile::tempdir().expect("tempdir");
503 let data_dir = dir.path().join("data");
504 std::fs::create_dir_all(&data_dir).expect("data dir");
505 let out = dir.path().join("out");
506 let (url, pings) = ping_server(2).await;
507
508 let mut state = crate::test_support::test_state(&data_dir).await;
509 let mut config = crate::test_support::test_config(&data_dir);
510 config.backup.dir = out.clone();
511 config.backup.keep = 1;
512 config.backup.healthcheck_url = url;
513 state.config = Arc::new(config);
514
515 let owner = state
516 .store
517 .create_user("alice", None, "hash", false)
518 .await
519 .expect("user");
520 let repo = state
521 .store
522 .create_repo(&owner.id, "demo", "", Visibility::Private, "main")
523 .await
524 .expect("repo");
525 let path = repo_path(&data_dir, &repo.id).expect("path");
526 init_bare(&state.git, &path, "main").await.expect("init");
527
528 let first = scheduled_backup(&state).await.expect("first backup");
529 assert!(first.path.starts_with(&out));
530 // A second run: retention keeps only the newest complete set. Set names have one-second
531 // stamps, so the runs must not share a second.
532 tokio::time::sleep(Duration::from_millis(1100)).await;
533 let second = scheduled_backup(&state).await.expect("second backup");
534 assert_ne!(first.path, second.path);
535 let sets: Vec<_> = std::fs::read_dir(&out)
536 .expect("read out")
537 .filter_map(|e| e.ok().map(|e| e.file_name()))
538 .collect();
539 assert_eq!(sets.len(), 1, "{sets:?}");
540
541 let runs = state.store.recent_runs(10).await.expect("runs");
542 assert_eq!(runs.len(), 2);
543 assert!(runs.iter().all(|r| r.ok == Some(true)));
544 assert!(runs.iter().all(|r| r.trigger == RunTrigger::Schedule));
545 assert!(runs[0].path.is_some());
546
547 // The first scheduled run pings start and success; the second run's pings are not needed.
548 for _ in 0..100 {
549 if pings.lock().unwrap_or_else(PoisonError::into_inner).len() >= 2 {
550 break;
551 }
552 tokio::time::sleep(Duration::from_millis(10)).await;
553 }
554 let seen = pings.lock().unwrap_or_else(PoisonError::into_inner).clone();
555 assert_eq!(seen[0], "GET /hc/token-123/start HTTP/1.1");
556 assert_eq!(seen[1], "GET /hc/token-123 HTTP/1.1");
557 }
558}