forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
23.1 KiB
1//! The replication worker. One task drains the queue: debounced per-repo pushes (5 s after the
2//! last ref update), a database snapshot every `db_snapshot_minutes`, and a reconcile pass every
3//! `reconcile_minutes` that pushes every repo, removes deleted repos from the standby, and pings
4//! the healthcheck. Runs start once at boot, so a fresh standby is filled without waiting.
5//!
6//! Every outcome is recorded in `replication_status` (see `forge_core::store::replication`), which
7//! `/admin/replication` shows.
8
9use std::path::{Path, PathBuf};
10use std::time::{Duration, Instant};
11
12use forge_core::Store;
13use forge_core::store::replication::DB_SUBJECT;
14use forge_core::store::snapshot::verify_snapshot;
15use forge_git::GitEnv;
16use forge_git::backup::refs_checksum;
17
18use super::{
19 KEEP_SNAPSHOTS, Layout, SshTransport, StandbyTransport, Target, is_snapshot_name, push_repo,
20 remote_refs_checksum, sh_quote, snapshot_stamp, wiki_subject, wiki_subject_repo,
21};
22use crate::backup::healthcheck;
23use crate::state::AppState;
24
25/// How often the worker checks its queue and timers when nothing wakes it.
26const TICK: Duration = Duration::from_secs(1);
27
28/// Everything a replication pass needs. The web side builds one from `AppState`; tests build one
29/// around a `LocalTransport`.
30pub struct Ctx<T> {
31 pub store: Store,
32 pub git: GitEnv,
33 pub data_dir: PathBuf,
34 pub layout: Layout,
35 pub transport: T,
36}
37
38/// What a reconcile pass did. `failures` holds one line per repo or step that failed.
39#[derive(Debug, Default, PartialEq, Eq)]
40pub struct ReconcileReport {
41 pub replicated: usize,
42 pub failures: Vec<String>,
43}
44
45/// Starts the worker task. It runs until the process exits. An invalid target is logged and the
46/// worker does not start; config validation makes that unlikely.
47pub fn spawn(state: AppState) {
48 tokio::spawn(run(state));
49}
50
51async fn run(state: AppState) {
52 let config = state.config.replication.clone();
53 let target = match Target::parse(&config.target, config.ssh_port) {
54 Ok(target) => target,
55 Err(e) => {
56 tracing::error!(error = %e, "replication: invalid target, worker not started");
57 return;
58 }
59 };
60 let layout = match Layout::new(&target.base) {
61 Ok(layout) => layout,
62 Err(e) => {
63 tracing::error!(error = %e, "replication: invalid remote base, worker not started");
64 return;
65 }
66 };
67 let transport = SshTransport::new(
68 target,
69 config.identity_file.clone(),
70 config.known_hosts_file.clone(),
71 );
72 let ctx = Ctx {
73 store: state.store.clone(),
74 git: (*state.git).clone(),
75 data_dir: state.git.data_dir.clone(),
76 layout,
77 transport,
78 };
79 tracing::info!(target = %config.target, "replication: worker started");
80
81 let db_every = Duration::from_secs(config.db_snapshot_minutes.max(1) * 60);
82 let reconcile_every = Duration::from_secs(config.reconcile_minutes.max(1) * 60);
83 let mut next_db = Instant::now();
84 let mut next_reconcile = Instant::now();
85 loop {
86 tokio::select! {
87 () = state.replication.wait() => {}
88 () = tokio::time::sleep(TICK) => {}
89 }
90 for repo_id in state.replication.take_due(Instant::now()) {
91 // A repo deleted since the update is skipped; the next reconcile removes its copy.
92 match ctx.store.repo_by_id(&repo_id).await {
93 Ok(Some(_)) => {
94 let _ = replicate_repo(&ctx, &repo_id).await;
95 }
96 Ok(None) => {}
97 Err(e) => tracing::error!(error = %e, "replication: could not look up repo"),
98 }
99 }
100 let now = Instant::now();
101 if now >= next_db {
102 next_db = now + db_every;
103 if let Err(e) = ship_db(&ctx).await {
104 tracing::error!(error = %e, "replication: database snapshot not shipped");
105 }
106 }
107 // Read the request even when the timer is due, so one request is one pass.
108 let requested = state.replication.take_reconcile_request();
109 if requested || now >= next_reconcile {
110 next_reconcile = Instant::now() + reconcile_every;
111 let report = reconcile_once(&ctx).await;
112 tracing::info!(
113 replicated = report.replicated,
114 failures = report.failures.len(),
115 "replication: reconcile finished"
116 );
117 ping(&config.healthcheck_url, &report).await;
118 }
119 }
120}
121
122/// Pings the healthcheck after a reconcile: success when every step worked, failure otherwise.
123async fn ping(url: &str, report: &ReconcileReport) {
124 if url.is_empty() {
125 return;
126 }
127 if report.failures.is_empty() {
128 healthcheck::success(url).await;
129 } else {
130 healthcheck::failure(url, &report.failures.join("; ")).await;
131 }
132}
133
134/// Records an outcome for `subject` and passes the error on. `refs` is the checksum to store on
135/// success (repos only).
136async fn record(
137 store: &Store,
138 subject: &str,
139 result: Result<Option<String>, String>,
140) -> Result<(), String> {
141 match result {
142 Ok(refs) => {
143 if let Err(e) = store.record_replication_ok(subject, refs.as_deref()).await {
144 tracing::error!(error = %e, subject, "replication: could not record success");
145 }
146 Ok(())
147 }
148 Err(e) => {
149 tracing::warn!(subject, error = %e, "replication: step failed");
150 if let Err(db) = store.record_replication_error(subject, &e).await {
151 tracing::error!(error = %db, subject, "replication: could not record failure");
152 }
153 Err(e)
154 }
155 }
156}
157
158/// Pushes one repo to the standby and checks that its refs match. A mismatch triggers one
159/// re-push; a second mismatch is an error. Records the outcome for the repo.
160///
161/// The repo's wiki, when it has one, is replicated the same way after the repo, under the subject
162/// `<id>.wiki`. A failure in either is returned.
163pub async fn replicate_repo<T: StandbyTransport>(
164 ctx: &Ctx<T>,
165 repo_id: &str,
166) -> Result<(), String> {
167 let result = replicate_inner(ctx, repo_id).await;
168 let repo_outcome = record(&ctx.store, repo_id, result).await;
169 let wiki = forge_git::repo::wiki_path(&ctx.data_dir, repo_id).map_err(|e| e.to_string())?;
170 let wiki_outcome = if wiki.join("HEAD").is_file() {
171 let result = replicate_wiki_inner(ctx, repo_id).await;
172 record(&ctx.store, &wiki_subject(repo_id), result).await
173 } else {
174 Ok(())
175 };
176 repo_outcome?;
177 wiki_outcome
178}
179
180async fn replicate_inner<T: StandbyTransport>(
181 ctx: &Ctx<T>,
182 repo_id: &str,
183) -> Result<Option<String>, String> {
184 let local = forge_git::repo::repo_path(&ctx.data_dir, repo_id).map_err(|e| e.to_string())?;
185 if !local.is_dir() {
186 return Err(format!(
187 "local repo directory is missing: {}",
188 local.display()
189 ));
190 }
191 let remote = ctx.layout.repo_dir(repo_id)?;
192 replicate_dir(ctx, &local, &remote).await
193}
194
195async fn replicate_wiki_inner<T: StandbyTransport>(
196 ctx: &Ctx<T>,
197 repo_id: &str,
198) -> Result<Option<String>, String> {
199 let local = forge_git::repo::wiki_path(&ctx.data_dir, repo_id).map_err(|e| e.to_string())?;
200 let remote = ctx.layout.wiki_dir(repo_id)?;
201 replicate_dir(ctx, &local, &remote).await
202}
203
204/// Pushes the bare repo at `local` to `remote` and checks that the refs match, retrying once.
205async fn replicate_dir<T: StandbyTransport>(
206 ctx: &Ctx<T>,
207 local: &Path,
208 remote: &str,
209) -> Result<Option<String>, String> {
210 // Idempotent: creates the bare repo on first use, and does nothing to an existing one.
211 ctx.transport
212 .run(&format!("git init --bare --quiet {}", sh_quote(remote)))
213 .await?;
214
215 push_repo(&ctx.transport, &ctx.git, local, remote).await?;
216 let mut local_sum = local_checksum(&ctx.git, local).await?;
217 let mut remote_sum = remote_refs_checksum(&ctx.transport, remote).await?;
218 if remote_sum != local_sum {
219 push_repo(&ctx.transport, &ctx.git, local, remote).await?;
220 local_sum = local_checksum(&ctx.git, local).await?;
221 remote_sum = remote_refs_checksum(&ctx.transport, remote).await?;
222 if remote_sum != local_sum {
223 return Err(format!(
224 "refs checksum mismatch after re-push: local {local_sum}, standby {remote_sum}"
225 ));
226 }
227 }
228 Ok(Some(local_sum))
229}
230
231async fn local_checksum(git: &GitEnv, local: &Path) -> Result<String, String> {
232 refs_checksum(git, local).await.map_err(|e| e.to_string())
233}
234
235/// Snapshots the database, verifies the copy, ships it as `db/forge-<stamp>.db`, points `latest`
236/// at it and prunes the standby to the newest [`KEEP_SNAPSHOTS`]. Records the outcome as `db`.
237pub async fn ship_db<T: StandbyTransport>(ctx: &Ctx<T>) -> Result<(), String> {
238 let result = ship_db_inner(ctx).await;
239 record(&ctx.store, DB_SUBJECT, result.map(|()| None)).await
240}
241
242async fn ship_db_inner<T: StandbyTransport>(ctx: &Ctx<T>) -> Result<(), String> {
243 let run_dir = ctx.data_dir.join("run");
244 std::fs::create_dir_all(&run_dir)
245 .map_err(|e| format!("cannot create {}: {e}", run_dir.display()))?;
246 let snap = run_dir.join("replica-snap.db");
247 remove_snapshot_files(&snap)?;
248 ctx.store
249 .snapshot_to(&snap)
250 .await
251 .map_err(|e| e.to_string())?;
252 let shipped = ship_snapshot(ctx, &snap).await;
253 // The local copy is only a staging file; a failure to delete it is not a replication failure.
254 let _ = remove_snapshot_files(&snap);
255 shipped
256}
257
258async fn ship_snapshot<T: StandbyTransport>(ctx: &Ctx<T>, snap: &Path) -> Result<(), String> {
259 verify_snapshot(snap).await.map_err(|e| e.to_string())?;
260
261 let db_dir = ctx.layout.db_dir();
262 ctx.transport
263 .run(&format!("mkdir -p {}", sh_quote(&db_dir)))
264 .await?;
265
266 let stamp = snapshot_stamp();
267 let name = format!("forge-{stamp}.db");
268 ctx.transport
269 .upload(snap, &ctx.layout.snapshot_file(&stamp))
270 .await?;
271
272 // `ln -sfn` writes the link under a temporary name, and `mv -T` swaps it in atomically.
273 let latest_tmp = format!("{db_dir}/.latest-tmp");
274 ctx.transport
275 .run(&format!(
276 "ln -sfn -- {} {} && mv -T -- {} {}",
277 sh_quote(&name),
278 sh_quote(&latest_tmp),
279 sh_quote(&latest_tmp),
280 sh_quote(&ctx.layout.latest_link())
281 ))
282 .await?;
283
284 prune_snapshots(ctx, &name).await
285}
286
287/// Deletes all but the newest [`KEEP_SNAPSHOTS`] snapshot files. `keep_always` is never deleted,
288/// even when clock skew makes it sort below older files.
289async fn prune_snapshots<T: StandbyTransport>(
290 ctx: &Ctx<T>,
291 keep_always: &str,
292) -> Result<(), String> {
293 let db_dir = ctx.layout.db_dir();
294 let listing = ctx
295 .transport
296 .run(&format!("ls -1 -- {}", sh_quote(&db_dir)))
297 .await?;
298 let text = String::from_utf8_lossy(&listing);
299 let mut names: Vec<&str> = text.lines().filter(|name| is_snapshot_name(name)).collect();
300 names.sort_unstable();
301 let excess = names.len().saturating_sub(KEEP_SNAPSHOTS);
302 let victims: Vec<String> = names
303 .iter()
304 .take(excess)
305 .filter(|name| **name != keep_always)
306 .map(|name| sh_quote(&format!("{db_dir}/{name}")))
307 .collect();
308 if victims.is_empty() {
309 return Ok(());
310 }
311 ctx.transport
312 .run(&format!("rm -f -- {}", victims.join(" ")))
313 .await
314 .map(|_| ())
315}
316
317/// Removes `snap` and its `-wal` and `-shm` sidecars when present.
318fn remove_snapshot_files(snap: &Path) -> Result<(), String> {
319 for suffix in ["", "-wal", "-shm"] {
320 let mut path = snap.as_os_str().to_owned();
321 path.push(suffix);
322 match std::fs::remove_file(&path) {
323 Ok(()) => {}
324 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
325 Err(e) => return Err(format!("cannot remove {}: {e}", Path::new(&path).display())),
326 }
327 }
328 Ok(())
329}
330
331/// Replicates every repo, removes the standby copies of repos that no longer exist, and reports.
332/// A repo that fails is counted and does not stop the others.
333pub async fn reconcile_once<T: StandbyTransport>(ctx: &Ctx<T>) -> ReconcileReport {
334 let mut report = ReconcileReport::default();
335 let ids = match ctx.store.repo_ids().await {
336 Ok(ids) => ids,
337 Err(e) => {
338 report.failures.push(format!("cannot list repos: {e}"));
339 return report;
340 }
341 };
342 for id in &ids {
343 match replicate_repo(ctx, id).await {
344 Ok(()) => report.replicated += 1,
345 Err(e) => report.failures.push(format!("{id}: {e}")),
346 }
347 }
348
349 let statuses = match ctx.store.list_replication_status().await {
350 Ok(statuses) => statuses,
351 Err(e) => {
352 report
353 .failures
354 .push(format!("cannot list replication status: {e}"));
355 return report;
356 }
357 };
358 for status in statuses {
359 // A wiki's status belongs to its repo: it stays while the repo does.
360 let owner = wiki_subject_repo(&status.subject).unwrap_or(&status.subject);
361 if status.subject == DB_SUBJECT || ids.iter().any(|id| id == owner) {
362 continue;
363 }
364 match remove_remote_repo(ctx, &status.subject).await {
365 Ok(()) => {
366 if let Err(e) = ctx.store.delete_replication_status(&status.subject).await {
367 report.failures.push(format!("{}: {e}", status.subject));
368 }
369 }
370 Err(e) => report
371 .failures
372 .push(format!("{}: remove failed: {e}", status.subject)),
373 }
374 }
375 report
376}
377
378/// Deletes exactly `<base>/repos/<id>.git` on the standby. The id is validated first.
379async fn remove_remote_repo<T: StandbyTransport>(
380 ctx: &Ctx<T>,
381 subject: &str,
382) -> Result<(), String> {
383 let remote = match wiki_subject_repo(subject) {
384 Some(repo_id) => ctx.layout.wiki_dir(repo_id)?,
385 None => ctx.layout.repo_dir(subject)?,
386 };
387 ctx.transport
388 .run(&format!("rm -rf -- {}", sh_quote(&remote)))
389 .await
390 .map(|_| ())
391}
392
393#[cfg(test)]
394mod tests {
395 use std::path::Path;
396
397 use forge_core::Store;
398 use forge_core::Visibility;
399 use forge_core::store::replication::DB_SUBJECT;
400 use forge_core::store::snapshot::verify_snapshot;
401 use forge_git::GitEnv;
402 use forge_git::backup::refs_checksum;
403
404 use super::super::{Layout, LocalTransport, remote_refs_checksum};
405 use super::{Ctx, ReconcileReport, reconcile_once, replicate_repo, ship_db};
406
407 /// Plain git against a bare repo, for making commits in the fixture.
408 fn plumbing(git_dir: &Path, args: &[&str]) -> String {
409 let out = std::process::Command::new("git")
410 .arg(format!("--git-dir={}", git_dir.display()))
411 .args(args)
412 .env("GIT_AUTHOR_NAME", "t")
413 .env("GIT_AUTHOR_EMAIL", "t@example.com")
414 .env("GIT_COMMITTER_NAME", "t")
415 .env("GIT_COMMITTER_EMAIL", "t@example.com")
416 .output()
417 .expect("git runs");
418 assert!(out.status.success(), "git {args:?}: {out:?}");
419 String::from_utf8_lossy(&out.stdout).trim().to_string()
420 }
421
422 /// Points `refname` at a new commit with `message` in `git_dir`.
423 fn commit_to(git_dir: &Path, refname: &str, message: &str) {
424 let tree = plumbing(git_dir, &["hash-object", "-t", "tree", "-w", "--stdin"]);
425 let commit = plumbing(git_dir, &["commit-tree", &tree, "-m", message]);
426 plumbing(git_dir, &["update-ref", refname, &commit]);
427 }
428
429 struct Fixture {
430 _dir: tempfile::TempDir,
431 ctx: Ctx<LocalTransport>,
432 standby: std::path::PathBuf,
433 repo_id: String,
434 local_repo: std::path::PathBuf,
435 }
436
437 async fn fixture() -> Fixture {
438 let dir = tempfile::tempdir().expect("tempdir");
439 let data_dir = dir.path().join("data");
440 std::fs::create_dir_all(&data_dir).expect("data dir");
441 let store = Store::open(&dir.path().join("forge.db"))
442 .await
443 .expect("store");
444 let alice = store
445 .create_user("Alice", None, "hash", false)
446 .await
447 .expect("user");
448 let repo = store
449 .create_repo(&alice.id, "demo", "", Visibility::Private, "main")
450 .await
451 .expect("repo");
452 let git = GitEnv {
453 git_bin: "git".into(),
454 data_dir: data_dir.clone(),
455 };
456 let local_repo = forge_git::repo::repo_path(&data_dir, &repo.id).expect("path");
457 forge_git::repo::init_bare(&git, &local_repo, "main")
458 .await
459 .expect("init bare");
460 commit_to(&local_repo, "refs/heads/main", "first");
461
462 let standby = dir.path().join("standby");
463 std::fs::create_dir_all(&standby).expect("standby");
464 let standby_str = standby.to_str().expect("utf8 path");
465 let ctx = Ctx {
466 store,
467 git,
468 data_dir,
469 layout: Layout::new(standby_str).expect("layout"),
470 transport: LocalTransport,
471 };
472 Fixture {
473 _dir: dir,
474 ctx,
475 standby,
476 repo_id: repo.id,
477 local_repo,
478 }
479 }
480
481 #[tokio::test]
482 async fn push_records_matching_checksum_and_follows_new_refs() {
483 let f = fixture().await;
484 replicate_repo(&f.ctx, &f.repo_id)
485 .await
486 .expect("first push");
487
488 let remote = f.ctx.layout.repo_dir(&f.repo_id).expect("remote dir");
489 assert!(
490 Path::new(&remote).join("HEAD").is_file(),
491 "bare repo created"
492 );
493 let local_sum = refs_checksum(&f.ctx.git, &f.local_repo).await.expect("sum");
494 let remote_sum = remote_refs_checksum(&f.ctx.transport, &remote)
495 .await
496 .expect("remote sum");
497 assert_eq!(local_sum, remote_sum);
498 let status = f
499 .ctx
500 .store
501 .replication_status(&f.repo_id)
502 .await
503 .expect("status")
504 .expect("row");
505 assert!(status.last_ok_at.is_some());
506 assert_eq!(status.refs_sha256.as_deref(), Some(local_sum.as_str()));
507 assert_eq!(status.last_error, None);
508
509 commit_to(&f.local_repo, "refs/heads/main", "second");
510 commit_to(&f.local_repo, "refs/tags/v1", "tagged");
511 replicate_repo(&f.ctx, &f.repo_id)
512 .await
513 .expect("second push");
514 let local_sum = refs_checksum(&f.ctx.git, &f.local_repo).await.expect("sum");
515 let remote_sum = remote_refs_checksum(&f.ctx.transport, &remote)
516 .await
517 .expect("remote sum");
518 assert_eq!(
519 local_sum, remote_sum,
520 "tags and the new head reached the standby"
521 );
522 }
523
524 #[tokio::test]
525 async fn a_failed_push_is_recorded_and_keeps_the_last_success() {
526 let f = fixture().await;
527 replicate_repo(&f.ctx, &f.repo_id).await.expect("push");
528 let ok = f
529 .ctx
530 .store
531 .replication_status(&f.repo_id)
532 .await
533 .expect("status")
534 .expect("row");
535
536 // Break the local repo so the next run cannot read it.
537 std::fs::remove_dir_all(&f.local_repo).expect("remove local repo");
538 assert!(replicate_repo(&f.ctx, &f.repo_id).await.is_err());
539 let failed = f
540 .ctx
541 .store
542 .replication_status(&f.repo_id)
543 .await
544 .expect("status")
545 .expect("row");
546 assert!(failed.last_error.is_some());
547 assert_eq!(failed.last_ok_at, ok.last_ok_at);
548 }
549
550 #[tokio::test]
551 async fn deleted_repo_is_removed_from_the_standby_at_reconcile() {
552 let f = fixture().await;
553 let report = reconcile_once(&f.ctx).await;
554 assert_eq!(
555 report,
556 ReconcileReport {
557 replicated: 1,
558 failures: vec![]
559 }
560 );
561
562 let remote = f.ctx.layout.repo_dir(&f.repo_id).expect("remote dir");
563 assert!(Path::new(&remote).is_dir());
564
565 f.ctx
566 .store
567 .delete_repo(&f.repo_id)
568 .await
569 .expect("delete repo");
570 let report = reconcile_once(&f.ctx).await;
571 assert!(report.failures.is_empty(), "{:?}", report.failures);
572 assert!(!Path::new(&remote).exists(), "standby copy is gone");
573 assert!(
574 f.ctx
575 .store
576 .replication_status(&f.repo_id)
577 .await
578 .expect("status")
579 .is_none(),
580 "status row is gone"
581 );
582 assert!(
583 f.standby.join("repos").is_dir(),
584 "only the repo dir is removed"
585 );
586 }
587
588 #[tokio::test]
589 async fn database_snapshot_ships_with_latest_link_and_prunes_to_24() {
590 let f = fixture().await;
591 // Older snapshots already on the standby: 30 names that sort before any real stamp.
592 let db_dir = f.standby.join("db");
593 std::fs::create_dir_all(&db_dir).expect("db dir");
594 for i in 0..30 {
595 std::fs::write(db_dir.join(format!("forge-19990101T{i:06}Z.db")), b"old")
596 .expect("old snapshot");
597 }
598 std::fs::write(db_dir.join("forge-19990101T000000Z.db.tmp"), b"x").expect("stray");
599
600 ship_db(&f.ctx).await.expect("ship");
601
602 let mut names: Vec<String> = std::fs::read_dir(&db_dir)
603 .expect("list")
604 .map(|e| e.expect("entry").file_name().to_string_lossy().into_owned())
605 .filter(|n| n.starts_with("forge-") && n.ends_with(".db"))
606 .collect();
607 names.sort();
608 assert_eq!(names.len(), 24, "pruned to the newest 24: {names:?}");
609 assert!(!db_dir.join("forge-19990101T000000Z.db").exists());
610 assert!(db_dir.join("forge-19990101T000029Z.db").exists());
611 assert!(
612 db_dir.join("forge-19990101T000000Z.db.tmp").exists(),
613 "foreign files stay"
614 );
615
616 let newest = names.last().expect("newest").clone();
617 let link = std::fs::read_link(db_dir.join("latest")).expect("latest is a symlink");
618 assert_eq!(
619 link.to_string_lossy(),
620 newest,
621 "latest points at the new snapshot"
622 );
623 let info = verify_snapshot(&db_dir.join(&newest))
624 .await
625 .expect("shipped copy verifies");
626 assert!(info.repos.iter().any(|r| r.0 == f.repo_id));
627
628 let status = f
629 .ctx
630 .store
631 .replication_status(DB_SUBJECT)
632 .await
633 .expect("status")
634 .expect("row");
635 assert!(status.last_ok_at.is_some());
636 assert_eq!(status.last_error, None);
637 assert!(
638 !f.ctx.data_dir.join("run").join("replica-snap.db").exists(),
639 "the local staging copy is removed"
640 );
641 }
642
643 #[tokio::test]
644 async fn reconcile_reports_failures_without_stopping_other_repos() {
645 let f = fixture().await;
646 std::fs::remove_dir_all(&f.local_repo).expect("remove local repo");
647 let report = reconcile_once(&f.ctx).await;
648 assert_eq!(report.replicated, 0);
649 assert_eq!(report.failures.len(), 1, "{:?}", report.failures);
650 }
651}