Skip to main content

max / makenotwork

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