Skip to main content

max / makenotwork

29.9 KB · 785 lines History Blame Raw
1 //! Fetch the prod backup that `migration_dry_run` runs against.
2 //!
3 //! Sources supported:
4 //! - `file:///abs/path/to/dump.sql.gz` — local copy (dev).
5 //! - `rsync://host/module/path` — rsync daemon protocol.
6 //! - `ssh://user@host[:port]/path/file.sql.gz` — rsync-over-ssh. Used to pull
7 //! prod backups from `backup-puller@alpha-west-1`.
8 //!
9 //! The fetch is command-driven: the operator triggers it via /backup/fetch, it
10 //! is not implicit in promote. That keeps the slowest, most failure-prone step
11 //! visible in the TUI rather than buried inside a deploy.
12
13 use crate::config::AppConfig;
14 use crate::topology::Topology;
15 use anyhow::{Context, Result, bail};
16 use chrono::Utc;
17 use ops_exec::{CapabilitySet, Executor, SshExec, SyncOpts};
18 use sqlx::SqlitePool;
19 use std::path::Path;
20 use std::sync::Arc;
21 use tokio::process::Command;
22
23 #[derive(Debug, Clone)]
24 pub struct FetchedBackup {
25 /// Which configured dump this is (`BackupConfig::name`), so the caller can
26 /// tell the server's from multithreaded's in one response.
27 pub name: String,
28 pub source: String,
29 pub local_path: String,
30 pub byte_size: Option<i64>,
31 }
32
33 /// Parsed `backup.source` URL. Owned strings so the parsed form outlives the
34 /// (possibly transient) URL we read from config.
35 #[derive(Debug, Clone, PartialEq, Eq)]
36 pub(crate) enum BackupSource {
37 /// Local file copy. Path follows the `file://` prefix.
38 File { path: String },
39 /// rsync daemon protocol. Full URL stays intact (rsync handles it).
40 RsyncDaemon { url: String },
41 /// rsync-over-ssh. Port is optional.
42 Ssh {
43 user_host: String,
44 port: Option<u16>,
45 path: String,
46 },
47 }
48
49 /// Parse a `backup.source` URL into a `BackupSource`. Rejects unsupported
50 /// schemes and malformed `ssh://` URLs (no path part).
51 pub(crate) fn parse_source(s: &str) -> Result<BackupSource> {
52 if let Some(rest) = s.strip_prefix("file://") {
53 if rest.is_empty() {
54 bail!("file:// URL is missing a path: {s}");
55 }
56 return Ok(BackupSource::File { path: rest.into() });
57 }
58 if s.starts_with("rsync://") {
59 return Ok(BackupSource::RsyncDaemon { url: s.into() });
60 }
61 if let Some(rest) = s.strip_prefix("ssh://") {
62 let (user_host_port, path_rest) = rest
63 .split_once('/')
64 .with_context(|| format!("ssh:// URL missing path: {s}"))?;
65 if user_host_port.is_empty() {
66 bail!("ssh:// URL missing user@host: {s}");
67 }
68 let path = format!("/{path_rest}");
69 let (user_host, port) = match user_host_port.rsplit_once(':') {
70 Some((uh, p)) => {
71 // Heuristic: trailing `:digits` after the final `:` is the port.
72 // Anything else (IPv6 literal, etc.) gets left alone.
73 match p.parse::<u16>() {
74 Ok(n) => (uh.to_string(), Some(n)),
75 Err(_) => (user_host_port.to_string(), None),
76 }
77 }
78 None => (user_host_port.to_string(), None),
79 };
80 if user_host.is_empty() {
81 bail!("ssh:// URL has empty host (port {port:?})");
82 }
83 return Ok(BackupSource::Ssh {
84 user_host,
85 port,
86 path,
87 });
88 }
89 bail!("unsupported backup source scheme: {s}");
90 }
91
92 /// Absolute floor: backups smaller than this are an empty or header-only file
93 /// (an rsync that wrote zero bytes, an empty source). This is the fallback when
94 /// there is no prior backup to compare against (the first-ever fetch).
95 const MIN_BACKUP_BYTES: u64 = 64;
96
97 /// Plausibility floor as a fraction of the last verified backup's size. A real
98 /// dump never abruptly halves; a source-side truncation that still closed a
99 /// valid gzip would pass `gzip -t` and the absolute floor, but not this. Every
100 /// row in `backups` is a previously-verified dump, so the last one is a sound
101 /// reference. Integer-halved at the call site.
102 ///
103 /// A dump *can* legitimately halve, though — a retention prune landing, a bloated
104 /// table finally being swept. When that happens the floor is self-sealing: no new
105 /// row is written unless a fetch clears it, so the reference can never advance and
106 /// every later fetch fails on the same stale number. That is not hypothetical;
107 /// Sando sat wedged from 2026-06-12 to 2026-07-27 after MNW started deleting
108 /// expired `tower_sessions` rows and the dump dropped 43 MB -> 6 MB overnight.
109 /// `force` (operator-supplied, via `POST /backup/fetch {"force":true}`) is the way
110 /// out: it drops to the absolute floor for one fetch, so the accepted dump becomes
111 /// the new reference. It never skips `gzip -t` — a truncated file is still refused.
112 const MIN_BACKUP_FRACTION_DENOM: i64 = 2;
113
114 /// Verify a freshly-downloaded backup before it is allowed to become the live
115 /// dump: reject anything below `min_bytes` (the plausibility floor derived from
116 /// history), and for a gzip require a complete, valid stream (`gzip -t` fails on
117 /// truncation/corruption). `is_gz` is taken from the *destination* name, not the
118 /// temp path (which carries a `.partial` suffix).
119 async fn verify_backup(tmp_path: &str, is_gz: bool, min_bytes: u64) -> Result<()> {
120 let meta = tokio::fs::metadata(tmp_path)
121 .await
122 .with_context(|| format!("stat fetched backup {tmp_path}"))?;
123 anyhow::ensure!(
124 meta.len() >= min_bytes,
125 "fetched backup {tmp_path} is implausibly small ({} bytes, floor {min_bytes}); \
126 treating as a failed/truncated transfer",
127 meta.len(),
128 );
129 if is_gz {
130 let out = Command::new("gzip")
131 .arg("-t")
132 .arg(tmp_path)
133 .output()
134 .await
135 .with_context(|| format!("spawning gzip -t {tmp_path}"))?;
136 anyhow::ensure!(
137 out.status.success(),
138 "fetched backup {tmp_path} failed gzip integrity check (truncated/corrupt): {}",
139 String::from_utf8_lossy(&out.stderr),
140 );
141 }
142 Ok(())
143 }
144
145 /// Pull every configured prod dump, or just `only` when named.
146 ///
147 /// Each dump is fetched independently: one source being down must not leave the
148 /// others un-refreshed, because a stale dump is a *blocked* gate and the whole
149 /// point of having more than one is that each database gets its own. So every
150 /// entry is attempted, and the errors are aggregated at the end — a caller
151 /// (`/backup/fetch`, and the daily timer through it) still sees a failure, it
152 /// just sees it after the work that could succeed did.
153 ///
154 /// `force` re-baselines the plausibility floor: see `MIN_BACKUP_FRACTION_DENOM`.
155 /// Pass `false` for anything automated — it is an operator escape hatch, not a
156 /// retry strategy.
157 pub async fn fetch(
158 pool: &SqlitePool,
159 cfg: &Arc<AppConfig>,
160 topo: &Arc<Topology>,
161 force: bool,
162 only: Option<&str>,
163 ) -> Result<Vec<FetchedBackup>> {
164 let selected: Vec<&crate::topology::BackupConfig> = match only {
165 Some(name) => vec![topo.backup_named(name).with_context(|| {
166 format!(
167 "no backup named {name:?} in the topology (have: {})",
168 topo.backup
169 .iter()
170 .map(|b| b.name.as_str())
171 .collect::<Vec<_>>()
172 .join(", ")
173 )
174 })?],
175 None => topo.backup.iter().collect(),
176 };
177
178 let mut fetched = Vec::new();
179 let mut failures = Vec::new();
180 for backup in selected {
181 match fetch_one(pool, cfg, backup, force).await {
182 Ok(fb) => fetched.push(fb),
183 Err(e) => {
184 tracing::error!(backup = %backup.name, error = %e, "backup fetch failed");
185 failures.push(format!("{}: {e:#}", backup.name));
186 }
187 }
188 }
189 anyhow::ensure!(
190 failures.is_empty(),
191 "{} of {} backup fetch(es) failed: {}",
192 failures.len(),
193 failures.len() + fetched.len(),
194 failures.join("; "),
195 );
196 Ok(fetched)
197 }
198
199 /// Pull one configured dump into its `local_path`.
200 async fn fetch_one(
201 pool: &SqlitePool,
202 cfg: &Arc<AppConfig>,
203 backup: &crate::topology::BackupConfig,
204 force: bool,
205 ) -> Result<FetchedBackup> {
206 let name = backup.name.clone();
207 let source = backup.source.clone();
208 let local_path = backup.local_path.clone();
209
210 if let Some(parent) = Path::new(&local_path).parent() {
211 tokio::fs::create_dir_all(parent).await?;
212 }
213
214 // Download to a sibling temp path, verify integrity, then atomically rename
215 // into place. The live `local_path` is never the write target, so a partial
216 // or corrupt transfer can never become the backup `migration_dry_run`
217 // restores (CF4). `--inplace`/`--partial` are deliberately NOT used — those
218 // keep a truncated file on failure, the opposite of what we want here.
219 let tmp_path = format!("{local_path}.partial");
220 let is_gz = std::path::Path::new(&local_path)
221 .extension()
222 .is_some_and(|ext| ext.eq_ignore_ascii_case("gz"));
223
224 // Plausibility floor: half the last verified backup's size, never below the
225 // absolute floor. Scoped to this dump's name — the server's dump is two
226 // orders of magnitude larger than multithreaded's, so a shared floor would
227 // reject every mt fetch as implausibly small and, on the other side, let a
228 // truncated server dump through. The first-ever fetch of a name (no prior
229 // row) falls back to the absolute floor.
230 let last_size: Option<i64> = sqlx::query_scalar(
231 "SELECT byte_size FROM backups
232 WHERE app = ? AND name = ? ORDER BY fetched_at DESC LIMIT 1",
233 )
234 .bind(&cfg.id)
235 .bind(&name)
236 .fetch_optional(pool)
237 .await?;
238 let min_bytes = if force {
239 tracing::warn!(
240 last_verified_bytes = last_size,
241 "force: re-baselining the backup plausibility floor to the absolute minimum; \
242 this fetch's size becomes the new reference"
243 );
244 MIN_BACKUP_BYTES
245 } else {
246 last_size.map_or(MIN_BACKUP_BYTES, |s| {
247 ((s / MIN_BACKUP_FRACTION_DENOM) as u64).max(MIN_BACKUP_BYTES)
248 })
249 };
250
251 let parsed = parse_source(&source)?;
252 let downloaded: Result<()> = async {
253 match parsed {
254 BackupSource::File { path } => {
255 tokio::fs::copy(&path, &tmp_path)
256 .await
257 .with_context(|| format!("copy {path} -> {tmp_path}"))?;
258 }
259 BackupSource::RsyncDaemon { url } => {
260 let out = Command::new("rsync")
261 .args(["-az", &url, &tmp_path])
262 .output()
263 .await
264 .context("spawning rsync")?;
265 anyhow::ensure!(
266 out.status.success(),
267 "rsync (daemon) failed: {}",
268 String::from_utf8_lossy(&out.stderr),
269 );
270 }
271 BackupSource::Ssh {
272 user_host,
273 path,
274 port,
275 } => {
276 // Through the executor rather than a hand-rolled `rsync -e ssh`:
277 // one transport, one set of SSH flags. The sync plane is now
278 // gated (ops_exec::gate_pull): grant `observe:artifact` and
279 // confine to the dump's own directory. `path` is operator config
280 // (the BackupSource in sando.toml), not attacker input, so the
281 // parent-dir root is a formality here — but it keeps this pull
282 // fail-closed like every other, rather than an open read of the
283 // remote host.
284 let caps = CapabilitySet::from_tokens(Vec::<&str>::new(), ["artifact"]);
285 let exec = SshExec::new(user_host, caps).with_port(port);
286 let exec = match Path::new(&path).parent() {
287 Some(dir) if !dir.as_os_str().is_empty() => exec.with_pull_root(dir),
288 // A bare filename with no directory: confine to the path
289 // itself (starts_with is reflexive), still fail-closed.
290 _ => exec.with_pull_root(&path),
291 };
292 // NOT `--partial`: a truncated leftover here is dangerous, not
293 // useful — a resumed fetch could splice two different dumps into
294 // one plausible-looking file (CF4). NOT `-z` either: the dump is
295 // already compressed.
296 let opts = SyncOpts {
297 compress: false,
298 partial: false,
299 ..SyncOpts::default()
300 };
301 exec.pull_file(Path::new(&path), Path::new(&tmp_path), &opts)
302 .await
303 .context("rsync (ssh) failed")?;
304 }
305 }
306 verify_backup(&tmp_path, is_gz, min_bytes).await
307 }
308 .await;
309
310 // On any download/verify failure, remove the temp file so a corrupt
311 // `.partial` never lingers, and leave the existing live backup untouched.
312 if let Err(e) = downloaded {
313 let _ = tokio::fs::remove_file(&tmp_path).await;
314 return Err(e);
315 }
316
317 tokio::fs::rename(&tmp_path, &local_path)
318 .await
319 .with_context(|| format!("atomic rename {tmp_path} -> {local_path}"))?;
320
321 let meta = tokio::fs::metadata(&local_path).await?;
322 let size = meta.len() as i64;
323
324 // Record the new backup and prune stale rows in one transaction, so a crash
325 // between the two can't leave the insert without the prune (or, worse, lose
326 // the insert while keeping a half-applied delete). The on-disk file is
327 // overwritten each fetch (single `local_path`), so rows older than 30 days
328 // reference a path that no longer exists — keep the table from growing.
329 let mut tx = pool.begin().await?;
330 sqlx::query(
331 "INSERT INTO backups (app, name, fetched_at, source, local_path, byte_size) \
332 VALUES (?, ?, ?, ?, ?, ?)",
333 )
334 .bind(&cfg.id)
335 .bind(&name)
336 .bind(Utc::now().to_rfc3339())
337 .bind(&source)
338 .bind(&local_path)
339 .bind(size)
340 .execute(&mut *tx)
341 .await?;
342 // Scoped: one product's fetch is not a licence to prune another's history,
343 // and the 30-day window is about this product's own overwritten dumps.
344 sqlx::query("DELETE FROM backups WHERE app = ? AND fetched_at < datetime('now', '-30 days')")
345 .bind(&cfg.id)
346 .execute(&mut *tx)
347 .await?;
348 tx.commit().await?;
349
350 Ok(FetchedBackup {
351 name,
352 source,
353 local_path,
354 byte_size: Some(size),
355 })
356 }
357
358 #[cfg(test)]
359 mod tests {
360 use super::*;
361 use crate::topology::{BackupConfig, RepoConfig};
362
363 // ---- CF4: atomic write + integrity ----
364
365 async fn mem_pool() -> SqlitePool {
366 let pool = sqlx::sqlite::SqlitePoolOptions::new()
367 .max_connections(1)
368 .connect("sqlite::memory:")
369 .await
370 .unwrap();
371 crate::db::migrate(&pool).await.unwrap();
372 pool
373 }
374
375 fn topo_with_backup(source: String, local_path: String) -> Topology {
376 Topology {
377 repo: Some(RepoConfig {
378 bare_path: "/tmp/x.git".into(),
379 branch: "main".into(),
380 upstream: None,
381 }),
382 backup: vec![BackupConfig {
383 name: "server".into(),
384 source,
385 local_path,
386 }],
387 tiers: vec![],
388 aux_repos: Vec::new(),
389 }
390 }
391
392 /// ~4 KB of poorly-compressible bytes so a gzip of it stays well above the
393 /// size floor and a half-truncation lands mid-stream (failing `gzip -t`).
394 fn incompressible(n: usize) -> Vec<u8> {
395 (0..n)
396 .map(|i| (i.wrapping_mul(2_654_435_761) >> 13) as u8)
397 .collect()
398 }
399
400 async fn write_valid_gz(path: &Path) {
401 let plain = path.with_extension("plain");
402 tokio::fs::write(&plain, incompressible(4096))
403 .await
404 .unwrap();
405 let out = Command::new("sh")
406 .arg("-c")
407 .arg(format!("gzip -c {} > {}", plain.display(), path.display()))
408 .output()
409 .await
410 .unwrap();
411 assert!(out.status.success(), "gzip shim failed");
412 }
413
414 #[tokio::test]
415 async fn fetch_file_source_writes_atomically_and_records_row() {
416 let tmp = tempfile::tempdir().unwrap();
417 let src = tmp.path().join("src.sql.gz");
418 write_valid_gz(&src).await;
419 let dest = tmp.path().join("backups/latest.sql.gz");
420 let topo = Arc::new(topo_with_backup(
421 format!("file://{}", src.display()),
422 dest.to_string_lossy().into_owned(),
423 ));
424 let pool = mem_pool().await;
425 let cfg = Arc::new(AppConfig::for_tests());
426
427 let fb = fetch(&pool, &cfg, &topo, false, None)
428 .await
429 .unwrap()
430 .remove(0);
431 assert!(dest.exists(), "live backup written");
432 assert!(
433 !dest.with_file_name("latest.sql.gz.partial").exists(),
434 "temp file consumed by the atomic rename",
435 );
436 assert!(fb.byte_size.unwrap() > 0);
437 let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM backups")
438 .fetch_one(&pool)
439 .await
440 .unwrap();
441 assert_eq!(count.0, 1, "a row is recorded for a successful fetch");
442 }
443
444 #[tokio::test]
445 async fn fetch_rejects_a_dump_far_below_the_last_backup_size() {
446 // Plausibility floor: a complete, valid gzip that is far smaller than the
447 // last verified backup is a likely source-side truncation and is rejected
448 // even though `gzip -t` passes — the gap the 64-byte absolute floor missed.
449 let tmp = tempfile::tempdir().unwrap();
450 let src = tmp.path().join("src.sql.gz");
451 write_valid_gz(&src).await; // ~4 KB valid gzip
452 let dest = tmp.path().join("backups/latest.sql.gz");
453 let topo = Arc::new(topo_with_backup(
454 format!("file://{}", src.display()),
455 dest.to_string_lossy().into_owned(),
456 ));
457 let pool = mem_pool().await;
458 let cfg = Arc::new(AppConfig::for_tests());
459
460 // Seed a prior verified backup far larger than the incoming one; the floor
461 // becomes 500_000, well above the ~4 KB dump.
462 sqlx::query("INSERT INTO backups (fetched_at, source, local_path, byte_size) VALUES (?, 'x', ?, 1000000)")
463 .bind(Utc::now().to_rfc3339())
464 .bind(dest.to_string_lossy().into_owned())
465 .execute(&pool).await.unwrap();
466
467 let err = fetch(&pool, &cfg, &topo, false, None)
468 .await
469 .unwrap_err()
470 .to_string();
471 assert!(err.contains("implausibly small"), "{err}");
472 assert!(
473 !dest.exists(),
474 "a rejected dump never becomes the live backup"
475 );
476 let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM backups")
477 .fetch_one(&pool)
478 .await
479 .unwrap();
480 assert_eq!(count.0, 1, "the rejected fetch records no new row");
481 }
482
483 #[tokio::test]
484 async fn the_plausibility_floor_is_scoped_to_one_dump() {
485 // The server's dump is two orders of magnitude larger than
486 // multithreaded's. A shared floor would reject every mt fetch as
487 // implausibly small (and, the other way round, let a badly truncated
488 // server dump through on mt's reference). Seed a large `server` row and
489 // fetch a small `multithreaded` one: it must be accepted.
490 let tmp = tempfile::tempdir().unwrap();
491 let src = tmp.path().join("mt.sql.gz");
492 write_valid_gz(&src).await; // ~4 KB valid gzip
493 let dest = tmp.path().join("backups/mt-latest.sql.gz");
494 let mut topo = topo_with_backup(
495 format!("file://{}", src.display()),
496 dest.to_string_lossy().into_owned(),
497 );
498 topo.backup[0].name = "multithreaded".into();
499 let topo = Arc::new(topo);
500 let pool = mem_pool().await;
501 let cfg = Arc::new(AppConfig::for_tests());
502
503 sqlx::query(
504 "INSERT INTO backups (name, fetched_at, source, local_path, byte_size) \
505 VALUES ('server', ?, 'x', '/tmp/server.sql.gz', 1000000)",
506 )
507 .bind(Utc::now().to_rfc3339())
508 .execute(&pool)
509 .await
510 .unwrap();
511
512 let fetched = fetch(&pool, &cfg, &topo, false, None)
513 .await
514 .expect("the server's size must not set multithreaded's floor");
515 assert_eq!(fetched.len(), 1);
516 assert_eq!(fetched[0].name, "multithreaded");
517 let recorded: (String,) =
518 sqlx::query_as("SELECT name FROM backups ORDER BY id DESC LIMIT 1")
519 .fetch_one(&pool)
520 .await
521 .unwrap();
522 assert_eq!(recorded.0, "multithreaded", "the row is recorded by name");
523 }
524
525 #[tokio::test]
526 async fn fetching_an_unknown_name_is_an_error_not_a_silent_no_op() {
527 // A typo'd `{"name":"mt"}` must not report success having fetched
528 // nothing — the operator would read that as a refreshed dump.
529 let tmp = tempfile::tempdir().unwrap();
530 let topo = Arc::new(topo_with_backup(
531 "file:///nope".into(),
532 tmp.path().join("x.sql.gz").to_string_lossy().into_owned(),
533 ));
534 let pool = mem_pool().await;
535 let cfg = Arc::new(AppConfig::for_tests());
536
537 let err = fetch(&pool, &cfg, &topo, false, Some("mt"))
538 .await
539 .unwrap_err()
540 .to_string();
541 assert!(err.contains("no backup named"), "{err}");
542 }
543
544 #[tokio::test]
545 async fn one_failing_source_does_not_skip_the_others() {
546 // Each dump gates a different database, and a stale dump is a blocked
547 // gate — so a broken source must not cost the working one its refresh.
548 let tmp = tempfile::tempdir().unwrap();
549 let good_src = tmp.path().join("good.sql.gz");
550 write_valid_gz(&good_src).await;
551 let good_dest = tmp.path().join("backups/good.sql.gz");
552 let mut topo = topo_with_backup(
553 "file:///nonexistent/sando-test-missing.sql.gz".into(),
554 tmp.path()
555 .join("backups/bad.sql.gz")
556 .to_string_lossy()
557 .into_owned(),
558 );
559 topo.backup.push(crate::topology::BackupConfig {
560 name: "multithreaded".into(),
561 source: format!("file://{}", good_src.display()),
562 local_path: good_dest.to_string_lossy().into_owned(),
563 });
564 let topo = Arc::new(topo);
565 let pool = mem_pool().await;
566 let cfg = Arc::new(AppConfig::for_tests());
567
568 let err = fetch(&pool, &cfg, &topo, false, None)
569 .await
570 .unwrap_err()
571 .to_string();
572 assert!(err.contains("server:"), "the failure names its dump: {err}");
573 assert!(
574 good_dest.exists(),
575 "the reachable dump is still fetched after the unreachable one fails"
576 );
577 }
578
579 #[tokio::test]
580 async fn force_rebaselines_the_floor_after_a_legitimate_shrink() {
581 // The wedge this exists for: the floor is derived from a row that only a
582 // passing fetch can replace, so a dump that legitimately halves locks the
583 // fetch out permanently. `force` accepts one undersized dump and makes it
584 // the new reference, unwedging the next ordinary fetch.
585 let tmp = tempfile::tempdir().unwrap();
586 let src = tmp.path().join("src.sql.gz");
587 write_valid_gz(&src).await;
588 let dest = tmp.path().join("backups/latest.sql.gz");
589 let topo = Arc::new(topo_with_backup(
590 format!("file://{}", src.display()),
591 dest.to_string_lossy().into_owned(),
592 ));
593 let pool = mem_pool().await;
594 let cfg = Arc::new(AppConfig::for_tests());
595 sqlx::query("INSERT INTO backups (fetched_at, source, local_path, byte_size) VALUES (?, 'x', ?, 1000000)")
596 .bind(Utc::now().to_rfc3339())
597 .bind(dest.to_string_lossy().into_owned())
598 .execute(&pool).await.unwrap();
599
600 // Same dump the un-forced fetch rejects above.
601 let fb = fetch(&pool, &cfg, &topo, true, None)
602 .await
603 .unwrap()
604 .remove(0);
605 assert!(dest.exists(), "the forced dump becomes the live backup");
606 let recorded = fb.byte_size.unwrap();
607 assert!(
608 recorded < 1_000_000,
609 "the accepted dump really is the small one"
610 );
611
612 // The new row is now the reference, so the next fetch passes unforced.
613 fetch(&pool, &cfg, &topo, false, None)
614 .await
615 .expect("floor re-baselined to the forced fetch's size");
616 }
617
618 #[tokio::test]
619 async fn force_still_rejects_a_corrupt_gzip() {
620 // `force` relaxes the size floor only. A truncated dump is refused either
621 // way — otherwise the escape hatch would be a way to install a broken
622 // backup as the thing migration_dry_run restores.
623 let tmp = tempfile::tempdir().unwrap();
624 let src = tmp.path().join("src.sql.gz");
625 write_valid_gz(&src).await;
626 let whole = tokio::fs::read(&src).await.unwrap();
627 tokio::fs::write(&src, &whole[..whole.len() / 2])
628 .await
629 .unwrap();
630 let dest = tmp.path().join("backups/latest.sql.gz");
631 let topo = Arc::new(topo_with_backup(
632 format!("file://{}", src.display()),
633 dest.to_string_lossy().into_owned(),
634 ));
635 let pool = mem_pool().await;
636 let cfg = Arc::new(AppConfig::for_tests());
637
638 let err = fetch(&pool, &cfg, &topo, true, None)
639 .await
640 .unwrap_err()
641 .to_string();
642 assert!(err.contains("gzip integrity check"), "{err}");
643 assert!(
644 !dest.exists(),
645 "a corrupt dump never becomes the live backup"
646 );
647 }
648
649 #[tokio::test]
650 async fn fetch_rejects_truncated_gz_and_leaves_no_live_file() {
651 let tmp = tempfile::tempdir().unwrap();
652 let src = tmp.path().join("src.sql.gz");
653 write_valid_gz(&src).await;
654 // Truncate to half: a valid gzip prefix that fails `gzip -t` mid-stream.
655 let full = tokio::fs::read(&src).await.unwrap();
656 assert!(
657 full.len() / 2 > MIN_BACKUP_BYTES as usize,
658 "half must clear the size floor to exercise gzip -t"
659 );
660 tokio::fs::write(&src, &full[..full.len() / 2])
661 .await
662 .unwrap();
663
664 let dest = tmp.path().join("backups/latest.sql.gz");
665 let topo = Arc::new(topo_with_backup(
666 format!("file://{}", src.display()),
667 dest.to_string_lossy().into_owned(),
668 ));
669 let pool = mem_pool().await;
670 let cfg = Arc::new(AppConfig::for_tests());
671
672 let res = fetch(&pool, &cfg, &topo, false, None).await;
673 assert!(res.is_err(), "a truncated gzip must fail the fetch");
674 assert!(
675 !dest.exists(),
676 "no live backup file results from a failed fetch"
677 );
678 assert!(
679 !dest.with_file_name("latest.sql.gz.partial").exists(),
680 "the corrupt temp file is cleaned up",
681 );
682 let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM backups")
683 .fetch_one(&pool)
684 .await
685 .unwrap();
686 assert_eq!(count.0, 0, "no row recorded for a failed fetch");
687 }
688
689 #[test]
690 fn parses_file_url() {
691 let s = parse_source("file:///opt/backups/latest.sql.gz").unwrap();
692 assert_eq!(
693 s,
694 BackupSource::File {
695 path: "/opt/backups/latest.sql.gz".into()
696 }
697 );
698 }
699
700 #[test]
701 fn file_url_without_path_errors() {
702 assert!(parse_source("file://").is_err());
703 }
704
705 #[test]
706 fn parses_rsync_daemon_url() {
707 let s = parse_source("rsync://astra/mnw/latest.sql.gz").unwrap();
708 assert_eq!(
709 s,
710 BackupSource::RsyncDaemon {
711 url: "rsync://astra/mnw/latest.sql.gz".into()
712 }
713 );
714 }
715
716 #[test]
717 fn parses_ssh_url_with_port() {
718 let s = parse_source("ssh://backup-puller@alpha-west-1:2200/latest.sql.gz").unwrap();
719 assert_eq!(
720 s,
721 BackupSource::Ssh {
722 user_host: "backup-puller@alpha-west-1".into(),
723 port: Some(2200),
724 path: "/latest.sql.gz".into(),
725 }
726 );
727 }
728
729 #[test]
730 fn parses_ssh_url_without_port() {
731 let s = parse_source("ssh://max@astra/opt/backups/mnw/latest.sql.gz").unwrap();
732 assert_eq!(
733 s,
734 BackupSource::Ssh {
735 user_host: "max@astra".into(),
736 port: None,
737 path: "/opt/backups/mnw/latest.sql.gz".into(),
738 }
739 );
740 }
741
742 #[test]
743 fn ssh_url_without_path_errors() {
744 // `split_once('/')` — `ssh://user@host` has no `/` after the scheme.
745 assert!(parse_source("ssh://backup-puller@alpha-west-1").is_err());
746 }
747
748 #[test]
749 fn ssh_url_without_user_host_errors() {
750 // Empty user@host: `ssh:///foo`. Caught by the empty-prefix check.
751 assert!(parse_source("ssh:///latest.sql.gz").is_err());
752 }
753
754 #[test]
755 fn ssh_url_with_non_numeric_after_colon_treats_as_part_of_host() {
756 // `host:notaport` should NOT parse `notaport` as a port. Leave the
757 // colon part of user_host; libssh/rsync will reject if truly wrong.
758 let s = parse_source("ssh://user@host:notaport/path").unwrap();
759 assert_eq!(
760 s,
761 BackupSource::Ssh {
762 user_host: "user@host:notaport".into(),
763 port: None,
764 path: "/path".into(),
765 }
766 );
767 }
768
769 #[test]
770 fn rejects_unknown_scheme() {
771 assert!(parse_source("ftp://example.com/file").is_err());
772 assert!(parse_source("just-a-path.sql.gz").is_err());
773 assert!(parse_source("").is_err());
774 }
775
776 #[test]
777 fn ssh_url_preserves_multi_segment_path() {
778 let s = parse_source("ssh://a@b:22/opt/foo/bar/baz.sql.gz").unwrap();
779 match s {
780 BackupSource::Ssh { path, .. } => assert_eq!(path, "/opt/foo/bar/baz.sql.gz"),
781 _ => panic!("wrong variant"),
782 }
783 }
784 }
785