forge
ExploreLog in

Tahinli/forge

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