forge
ExploreLog in

Tahinli/forge

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