forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
25.1 KiB
1//! Standby replication: pushes every repo and a database snapshot to a passive SSH host (see
2//! `docs/backup.md`, "Standby replication"). The standby is a dedicated unix account that holds
3//! bare repos and snapshots. It never runs forge.
4//!
5//! Everything that reaches the standby goes through [`StandbyTransport`]. [`SshTransport`] runs
6//! the system `ssh`; [`LocalTransport`] (tests only) runs the same shell commands with `sh -c` on
7//! this machine, where the remote base is a temporary directory. Remote paths are validated and
8//! quoted before they are placed in any command.
9
10pub mod worker;
11
12use std::collections::HashMap;
13use std::future::Future;
14use std::path::{Path, PathBuf};
15use std::sync::atomic::{AtomicBool, Ordering};
16use std::sync::{Mutex, PoisonError};
17use std::time::{Duration, Instant};
18
19use forge_git::GitEnv;
20use forge_git::command::GitCommand;
21use tokio::process::Command;
22use tokio::sync::Notify;
23
24use crate::state::AppState;
25
26/// How long a repo waits after its last ref update before it is pushed.
27pub const PUSH_DEBOUNCE: Duration = Duration::from_secs(5);
28/// Newest database snapshots kept on the standby.
29pub const KEEP_SNAPSHOTS: usize = 24;
30/// Name of the snapshot directory on the standby, under the base.
31const DB_DIR: &str = "db";
32/// Name of the repo directory on the standby, under the base.
33const REPOS_DIR: &str = "repos";
34/// Suffix of the replication status subject that names a repo's wiki.
35const WIKI_SUBJECT_SUFFIX: &str = ".wiki";
36/// Prefix and suffix of a snapshot file name: `forge-<UTC stamp>.db`.
37const SNAPSHOT_PREFIX: &str = "forge-";
38const SNAPSHOT_SUFFIX: &str = ".db";
39/// Stamp layout in snapshot names: `YYYYMMDDTHHMMSSZ`.
40const STAMP_LEN: usize = 16;
41const REFS_FORMAT: &str = "--format=%(objectname) %(refname)";
42
43/// Checks a remote base or repo path: an absolute path of `[A-Za-z0-9._/-]`, with no `..`, no
44/// empty components and no trailing slash.
45pub fn validate_remote_path(path: &str) -> Result<(), String> {
46 let well_formed = path.starts_with('/')
47 && path.len() > 1
48 && !path.ends_with('/')
49 && !path.contains("//")
50 && !path.contains("..")
51 && path
52 .bytes()
53 .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'/' | b'-'));
54 if well_formed {
55 Ok(())
56 } else {
57 Err(format!(
58 "remote path must be absolute, use only [A-Za-z0-9._/-], and have no '..': {path:?}"
59 ))
60 }
61}
62
63/// The replication status subject of a repo's wiki: `<id>.wiki`.
64pub fn wiki_subject(repo_id: &str) -> String {
65 format!("{repo_id}{WIKI_SUBJECT_SUFFIX}")
66}
67
68/// The repo id a wiki subject names, or `None` for any other subject.
69pub fn wiki_subject_repo(subject: &str) -> Option<&str> {
70 subject.strip_suffix(WIKI_SUBJECT_SUFFIX)
71}
72
73/// Repo ids are ULIDs: 26 ASCII alphanumerics. Checked before an id is used in a remote path.
74pub fn validate_repo_id(id: &str) -> Result<(), String> {
75 if id.len() == 26 && id.bytes().all(|b| b.is_ascii_alphanumeric()) {
76 Ok(())
77 } else {
78 Err(format!("not a repo id: {id:?}"))
79 }
80}
81
82/// Quotes one argument for a POSIX shell: wraps it in single quotes and writes each embedded
83/// single quote as `'\''`. The result is one word whatever the input holds.
84pub fn sh_quote(arg: &str) -> String {
85 format!("'{}'", arg.replace('\'', "'\\''"))
86}
87
88/// Where the standby is and how to reach it: `[user@]host:/absolute/base` plus the SSH port.
89#[derive(Debug, Clone, PartialEq, Eq)]
90pub struct Target {
91 pub user: Option<String>,
92 pub host: String,
93 pub port: u16,
94 /// Validated absolute base directory on the standby.
95 pub base: String,
96}
97
98impl Target {
99 /// Parses the `target` setting. Host and user may use `[A-Za-z0-9.-]` and `[A-Za-z0-9._-]`;
100 /// neither may start with `-`, so neither can be read by ssh as an option.
101 pub fn parse(spec: &str, port: u16) -> Result<Target, String> {
102 let (dest, base) = spec
103 .split_once(':')
104 .ok_or_else(|| "replication target must be [user@]host:/absolute/path".to_string())?;
105 let (user, host) = match dest.split_once('@') {
106 Some((user, host)) => (Some(user), host),
107 None => (None, dest),
108 };
109 if let Some(user) = user {
110 let ok = !user.is_empty()
111 && !user.starts_with('-')
112 && user
113 .bytes()
114 .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'-'));
115 if !ok {
116 return Err(format!("invalid user in replication target: {user:?}"));
117 }
118 }
119 let host_ok = !host.is_empty()
120 && !host.starts_with('-')
121 && host
122 .bytes()
123 .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'-'));
124 if !host_ok {
125 return Err(format!("invalid host in replication target: {host:?}"));
126 }
127 validate_remote_path(base)?;
128 Ok(Target {
129 user: user.map(str::to_owned),
130 host: host.to_owned(),
131 port,
132 base: base.to_owned(),
133 })
134 }
135
136 /// `user@host`, or `host` without a user.
137 pub fn destination(&self) -> String {
138 match &self.user {
139 Some(user) => format!("{user}@{}", self.host),
140 None => self.host.clone(),
141 }
142 }
143
144 /// The `ssh://` URL git pushes a repo to.
145 pub fn repo_url(&self, remote_repo_dir: &str) -> String {
146 format!(
147 "ssh://{}:{}{remote_repo_dir}",
148 self.destination(),
149 self.port
150 )
151 }
152}
153
154/// Remote layout under the base. `repos/<id>.git` holds bare repos, `db/` holds snapshots and the
155/// `latest` link. `keys/` belongs to the operator and is never written here.
156#[derive(Debug, Clone, PartialEq, Eq)]
157pub struct Layout {
158 pub base: String,
159}
160
161impl Layout {
162 pub fn new(base: &str) -> Result<Layout, String> {
163 validate_remote_path(base)?;
164 Ok(Layout {
165 base: base.to_owned(),
166 })
167 }
168
169 pub fn repos_dir(&self) -> String {
170 format!("{}/{REPOS_DIR}", self.base)
171 }
172
173 pub fn db_dir(&self) -> String {
174 format!("{}/{DB_DIR}", self.base)
175 }
176
177 /// `<base>/repos/<id>.git`. Fails for an id that is not a ULID.
178 pub fn repo_dir(&self, repo_id: &str) -> Result<String, String> {
179 validate_repo_id(repo_id)?;
180 Ok(format!("{}/{repo_id}.git", self.repos_dir()))
181 }
182
183 /// `<base>/repos/<id>.wiki.git`: the standby copy of the repo's wiki.
184 pub fn wiki_dir(&self, repo_id: &str) -> Result<String, String> {
185 validate_repo_id(repo_id)?;
186 Ok(format!("{}/{repo_id}.wiki.git", self.repos_dir()))
187 }
188
189 /// `<base>/db/forge-<stamp>.db`.
190 pub fn snapshot_file(&self, stamp: &str) -> String {
191 format!(
192 "{}/{SNAPSHOT_PREFIX}{stamp}{SNAPSHOT_SUFFIX}",
193 self.db_dir()
194 )
195 }
196
197 pub fn latest_link(&self) -> String {
198 format!("{}/latest", self.db_dir())
199 }
200}
201
202/// A snapshot name `forge-<16-char stamp>.db` with a stamp of digits, `T` and `Z`.
203pub fn is_snapshot_name(name: &str) -> bool {
204 let Some(stamp) = name
205 .strip_prefix(SNAPSHOT_PREFIX)
206 .and_then(|rest| rest.strip_suffix(SNAPSHOT_SUFFIX))
207 else {
208 return false;
209 };
210 stamp.len() == STAMP_LEN
211 && stamp.bytes().enumerate().all(|(i, b)| match i {
212 8 => b == b'T',
213 15 => b == b'Z',
214 _ => b.is_ascii_digit(),
215 })
216}
217
218/// The UTC stamp used in snapshot names, `YYYYMMDDTHHMMSSZ`.
219pub fn snapshot_stamp() -> String {
220 const FORMAT: &[time::format_description::FormatItem<'static>] =
221 time::macros::format_description!("[year][month][day]T[hour][minute][second]Z");
222 time::OffsetDateTime::now_utc()
223 .format(FORMAT)
224 .unwrap_or_else(|_| "00000000T000000Z".to_owned())
225}
226
227/// Moves data to the standby and runs commands there. Implemented by [`SshTransport`] and, in
228/// tests, [`LocalTransport`]. The worker only calls these three methods.
229pub trait StandbyTransport: Send + Sync {
230 /// Runs one shell command (already quoted) on the standby and returns its stdout.
231 fn run(&self, remote_cmd: &str) -> impl Future<Output = Result<Vec<u8>, String>> + Send;
232
233 /// The URL git pushes the repo at `remote_repo_dir` to.
234 fn repo_url(&self, remote_repo_dir: &str) -> String;
235
236 /// Whether git must be told to allow a plain local path as a transport.
237 fn local_git(&self) -> bool;
238
239 /// The `GIT_SSH_COMMAND` git must use to reach the standby, if any.
240 fn push_ssh_command(&self) -> Option<String>;
241
242 /// Streams the local file `local` to `remote_path` on the standby. The data goes to a temporary
243 /// file in the same directory first, and is renamed into place only once complete.
244 fn upload(
245 &self,
246 local: &Path,
247 remote_path: &str,
248 ) -> impl Future<Output = Result<(), String>> + Send;
249}
250
251/// Builds the `cat > tmp && mv tmp final` command used by every upload.
252fn upload_command(remote_path: &str) -> String {
253 let (dir, name) = remote_path.rsplit_once('/').unwrap_or(("", remote_path));
254 let tmp = format!("{dir}/.tmp-{name}");
255 format!(
256 "cat > {} && mv -f {} {}",
257 sh_quote(&tmp),
258 sh_quote(&tmp),
259 sh_quote(remote_path)
260 )
261}
262
263/// Runs `command` with its stdin fed from `file`, and fails on a non-zero exit.
264async fn pipe_file(mut command: Command, file: &Path) -> Result<(), String> {
265 use std::process::Stdio;
266 use tokio::io::AsyncWriteExt;
267 command
268 .stdin(Stdio::piped())
269 .stdout(Stdio::null())
270 .stderr(Stdio::piped())
271 .kill_on_drop(true);
272 let mut child = command
273 .spawn()
274 .map_err(|e| format!("cannot start ssh: {e}"))?;
275 let mut stdin = child
276 .stdin
277 .take()
278 .ok_or_else(|| "ssh stdin is not available".to_string())?;
279 let mut source = tokio::fs::File::open(file)
280 .await
281 .map_err(|e| format!("cannot read {}: {e}", file.display()))?;
282 tokio::io::copy(&mut source, &mut stdin)
283 .await
284 .map_err(|e| format!("upload failed: {e}"))?;
285 stdin
286 .shutdown()
287 .await
288 .map_err(|e| format!("upload failed: {e}"))?;
289 drop(stdin);
290 let out = child
291 .wait_with_output()
292 .await
293 .map_err(|e| format!("upload failed: {e}"))?;
294 if out.status.success() {
295 Ok(())
296 } else {
297 Err(format!(
298 "upload exited with {}: {}",
299 out.status,
300 String::from_utf8_lossy(&out.stderr).trim()
301 ))
302 }
303}
304
305/// Runs `command` and returns stdout. A non-zero exit is an error carrying stderr.
306async fn capture(mut command: Command) -> Result<Vec<u8>, String> {
307 use std::process::Stdio;
308 command.stdin(Stdio::null()).kill_on_drop(true);
309 let out = command
310 .output()
311 .await
312 .map_err(|e| format!("cannot start ssh: {e}"))?;
313 if out.status.success() {
314 Ok(out.stdout)
315 } else {
316 Err(format!(
317 "remote command exited with {}: {}",
318 out.status,
319 String::from_utf8_lossy(&out.stderr).trim()
320 ))
321 }
322}
323
324/// Pushes every ref of the local bare repo `repo_dir` to `url` with `--mirror --force`.
325async fn push_mirror(
326 git: &GitEnv,
327 repo_dir: &Path,
328 url: &str,
329 local_git: bool,
330 ssh_command: Option<String>,
331) -> Result<(), String> {
332 let mut cmd = GitCommand::new(git, "push")
333 .git_dir(repo_dir)
334 .arg("--mirror")
335 .arg("--force")
336 .arg("--quiet");
337 if local_git {
338 cmd = cmd.allow_local_transport();
339 }
340 if let Some(ssh) = ssh_command {
341 cmd = cmd.env("GIT_SSH_COMMAND", ssh);
342 }
343 cmd.arg("--end-of-options")
344 .user_arg(url)
345 .map_err(|e| e.to_string())?
346 .status()
347 .await
348 .map_err(|e| e.to_string())
349}
350
351/// Replication over the system `ssh`, with the options that pin the standby's host key and use
352/// only the configured identity. Argument construction is unit-tested; no test connects.
353#[derive(Debug, Clone)]
354pub struct SshTransport {
355 target: Target,
356 identity_file: PathBuf,
357 known_hosts_file: PathBuf,
358}
359
360impl SshTransport {
361 pub fn new(target: Target, identity_file: PathBuf, known_hosts_file: PathBuf) -> SshTransport {
362 SshTransport {
363 target,
364 identity_file,
365 known_hosts_file,
366 }
367 }
368
369 /// The `-i`/`-o`/`-p` options, in order. Every one is a fixed word or a configured path.
370 pub fn ssh_options(&self) -> Vec<String> {
371 vec![
372 "-i".to_owned(),
373 self.identity_file.to_string_lossy().into_owned(),
374 "-o".to_owned(),
375 "IdentitiesOnly=yes".to_owned(),
376 "-o".to_owned(),
377 format!(
378 "UserKnownHostsFile={}",
379 self.known_hosts_file.to_string_lossy()
380 ),
381 "-o".to_owned(),
382 "StrictHostKeyChecking=yes".to_owned(),
383 "-o".to_owned(),
384 "BatchMode=yes".to_owned(),
385 "-o".to_owned(),
386 "ConnectTimeout=20".to_owned(),
387 "-p".to_owned(),
388 self.target.port.to_string(),
389 ]
390 }
391
392 /// The full argv after the program name: options, destination, then the remote command.
393 pub fn ssh_argv(&self, remote_cmd: &str) -> Vec<String> {
394 let mut argv = self.ssh_options();
395 argv.push(self.target.destination());
396 argv.push(remote_cmd.to_owned());
397 argv
398 }
399
400 /// The value for `GIT_SSH_COMMAND`: the same ssh invocation, each word quoted for the shell.
401 pub fn git_ssh_command(&self) -> String {
402 std::iter::once("ssh".to_owned())
403 .chain(self.ssh_options())
404 .map(|word| sh_quote(&word))
405 .collect::<Vec<_>>()
406 .join(" ")
407 }
408
409 fn command(&self, remote_cmd: &str) -> Command {
410 let mut cmd = Command::new("ssh");
411 cmd.args(self.ssh_argv(remote_cmd));
412 cmd
413 }
414}
415
416impl StandbyTransport for SshTransport {
417 fn run(&self, remote_cmd: &str) -> impl Future<Output = Result<Vec<u8>, String>> + Send {
418 let cmd = self.command(remote_cmd);
419 capture(cmd)
420 }
421
422 fn repo_url(&self, remote_repo_dir: &str) -> String {
423 self.target.repo_url(remote_repo_dir)
424 }
425
426 fn local_git(&self) -> bool {
427 false
428 }
429
430 fn push_ssh_command(&self) -> Option<String> {
431 Some(self.git_ssh_command())
432 }
433
434 fn upload(
435 &self,
436 local: &Path,
437 remote_path: &str,
438 ) -> impl Future<Output = Result<(), String>> + Send {
439 let cmd = self.command(&upload_command(remote_path));
440 let local = local.to_path_buf();
441 async move { pipe_file(cmd, &local).await }
442 }
443}
444
445/// Runs the same shell commands as [`SshTransport`], but with `sh -c` on this machine. The remote
446/// base is then a local directory. Test only.
447#[cfg(test)]
448#[derive(Debug, Clone, Copy, Default)]
449pub struct LocalTransport;
450
451#[cfg(test)]
452impl StandbyTransport for LocalTransport {
453 fn run(&self, remote_cmd: &str) -> impl Future<Output = Result<Vec<u8>, String>> + Send {
454 let mut cmd = Command::new("sh");
455 cmd.arg("-c").arg(remote_cmd);
456 capture(cmd)
457 }
458
459 fn repo_url(&self, remote_repo_dir: &str) -> String {
460 remote_repo_dir.to_owned()
461 }
462
463 fn local_git(&self) -> bool {
464 true
465 }
466
467 fn push_ssh_command(&self) -> Option<String> {
468 None
469 }
470
471 fn upload(
472 &self,
473 local: &Path,
474 remote_path: &str,
475 ) -> impl Future<Output = Result<(), String>> + Send {
476 let mut cmd = Command::new("sh");
477 cmd.arg("-c").arg(upload_command(remote_path));
478 let local = local.to_path_buf();
479 async move { pipe_file(cmd, &local).await }
480 }
481}
482
483/// Pushes one local repo with `--mirror --force` through a transport. The remote directory must
484/// already exist (see `worker::replicate_repo`).
485pub async fn push_repo<T: StandbyTransport>(
486 transport: &T,
487 git: &GitEnv,
488 repo_dir: &Path,
489 remote_repo_dir: &str,
490) -> Result<(), String> {
491 push_mirror(
492 git,
493 repo_dir,
494 &transport.repo_url(remote_repo_dir),
495 transport.local_git(),
496 transport.push_ssh_command(),
497 )
498 .await
499}
500
501/// The refs checksum of the bare repo at `remote_repo_dir`, read through the transport.
502pub async fn remote_refs_checksum<T: StandbyTransport>(
503 transport: &T,
504 remote_repo_dir: &str,
505) -> Result<String, String> {
506 let cmd = format!(
507 "git -C {} for-each-ref {}",
508 sh_quote(remote_repo_dir),
509 sh_quote(REFS_FORMAT)
510 );
511 let out = transport.run(&cmd).await?;
512 Ok(forge_git::backup::refs_checksum_of_listing(&out))
513}
514
515/// The queue the web side writes and the replication worker drains. Holds repos waiting for a
516/// debounced push, and the "reconcile now" request.
517#[derive(Default)]
518pub struct ReplicationQueue {
519 notify: Notify,
520 pending: Mutex<HashMap<String, Instant>>,
521 reconcile: AtomicBool,
522}
523
524impl ReplicationQueue {
525 /// Queues `repo_id` for a push once [`PUSH_DEBOUNCE`] has passed without another update.
526 pub fn enqueue(&self, repo_id: &str) {
527 let due = Instant::now() + PUSH_DEBOUNCE;
528 self.pending
529 .lock()
530 .unwrap_or_else(PoisonError::into_inner)
531 .insert(repo_id.to_owned(), due);
532 self.notify.notify_one();
533 }
534
535 /// Removes and returns the repos whose debounce has passed at `now`, sorted by id.
536 pub fn take_due(&self, now: Instant) -> Vec<String> {
537 let mut pending = self.pending.lock().unwrap_or_else(PoisonError::into_inner);
538 let mut due: Vec<String> = pending
539 .iter()
540 .filter(|(_, at)| **at <= now)
541 .map(|(id, _)| id.clone())
542 .collect();
543 due.sort();
544 for id in &due {
545 pending.remove(id);
546 }
547 due
548 }
549
550 /// Asks the worker for a full reconcile pass as soon as it can.
551 pub fn request_reconcile(&self) {
552 self.reconcile.store(true, Ordering::SeqCst);
553 self.notify.notify_one();
554 }
555
556 /// Returns true once per request.
557 pub fn take_reconcile_request(&self) -> bool {
558 self.reconcile.swap(false, Ordering::SeqCst)
559 }
560
561 /// Resolves when something is queued or a reconcile is requested.
562 pub async fn wait(&self) {
563 self.notify.notified().await;
564 }
565}
566
567/// Queues a repo after any ref update. Does nothing when replication is disabled.
568pub fn enqueue_repo(state: &AppState, repo_id: &str) {
569 if state.config.replication.enabled() {
570 state.replication.enqueue(repo_id);
571 }
572}
573
574#[cfg(test)]
575mod tests {
576 use super::*;
577
578 #[test]
579 fn remote_paths_must_be_plain_absolute_paths() {
580 for ok in ["/srv/forge-standby", "/srv/a.b_c-d/e", "/x"] {
581 assert!(validate_remote_path(ok).is_ok(), "{ok} should pass");
582 }
583 for bad in [
584 "",
585 "/",
586 "srv/forge",
587 "./srv",
588 "/srv/../etc",
589 "/srv/..",
590 "/srv/a/../b",
591 "/srv//x",
592 "/srv/x/",
593 "/srv/with space",
594 "/srv/quote'",
595 "/srv/$HOME",
596 "/srv/a;rm",
597 "/srv/a`b`",
598 "/srv/é",
599 "/srv/x\n",
600 ] {
601 assert!(validate_remote_path(bad).is_err(), "{bad:?} should fail");
602 }
603 }
604
605 #[test]
606 fn repo_ids_must_be_ulid_shaped() {
607 assert!(validate_repo_id("01ARZ3NDEKTSV4RRFFQ69G5FAV").is_ok());
608 assert!(validate_repo_id("01ARZ3NDEKTSV4RRFFQ69G5FA").is_err());
609 assert!(validate_repo_id("../../etc/passwd0000000000").is_err());
610 assert!(validate_repo_id("01ARZ3NDEKTSV4RRFFQ69G5F'V").is_err());
611 }
612
613 #[test]
614 fn quoting_survives_spaces_quotes_and_metacharacters() {
615 assert_eq!(sh_quote("/srv/x"), "'/srv/x'");
616 assert_eq!(sh_quote(""), "''");
617 assert_eq!(sh_quote("a b"), "'a b'");
618 assert_eq!(sh_quote("it's"), "'it'\\''s'");
619 assert_eq!(sh_quote("$(rm -rf /)`x`;|&"), "'$(rm -rf /)`x`;|&'");
620 assert_eq!(sh_quote("'"), "''\\'''");
621 // Round trip through a real shell: each quoted word comes back unchanged.
622 let words = ["plain", "it's", "a b", "$HOME", "x'y'z", "back\\slash"];
623 let line = words
624 .iter()
625 .map(|w| sh_quote(w))
626 .collect::<Vec<_>>()
627 .join(" ");
628 let script = format!("printf '%s\\0' {line}");
629 let out = std::process::Command::new("sh")
630 .arg("-c")
631 .arg(&script)
632 .output()
633 .expect("sh runs");
634 let got: Vec<String> = out
635 .stdout
636 .split(|b| *b == 0)
637 .filter(|s| !s.is_empty())
638 .map(|s| String::from_utf8_lossy(s).into_owned())
639 .collect();
640 assert_eq!(got, words);
641 }
642
643 #[test]
644 fn target_parses_user_host_and_base() {
645 let t = Target::parse("sb@standby.example:/srv/forge-standby", 2222).expect("parse");
646 assert_eq!(t.user.as_deref(), Some("sb"));
647 assert_eq!(t.host, "standby.example");
648 assert_eq!(t.port, 2222);
649 assert_eq!(t.base, "/srv/forge-standby");
650 assert_eq!(
651 t.repo_url("/srv/forge-standby/repos/X.git"),
652 "ssh://sb@standby.example:2222/srv/forge-standby/repos/X.git"
653 );
654 let bare = Target::parse("standby:/srv/x", 22).expect("parse");
655 assert_eq!(bare.destination(), "standby");
656 }
657
658 #[test]
659 fn target_rejects_bad_specs() {
660 for bad in [
661 "no-colon",
662 "sb@host",
663 "sb@host:relative",
664 "sb@host:/srv/../x",
665 "-oProxyCommand=x@host:/srv",
666 "@host:/srv",
667 "sb@:/srv",
668 "sb@-x:/srv",
669 "sb x@host:/srv",
670 "sb@ho st:/srv",
671 "sb@host:/srv/a b",
672 ] {
673 assert!(Target::parse(bad, 22).is_err(), "{bad:?} should be refused");
674 }
675 }
676
677 #[test]
678 fn layout_paths_follow_the_contract() {
679 let layout = Layout::new("/srv/sb").expect("layout");
680 assert_eq!(
681 layout.repo_dir("01ARZ3NDEKTSV4RRFFQ69G5FAV").expect("id"),
682 "/srv/sb/repos/01ARZ3NDEKTSV4RRFFQ69G5FAV.git"
683 );
684 assert!(layout.repo_dir("../x").is_err());
685 assert_eq!(
686 layout.snapshot_file("20260101T000000Z"),
687 "/srv/sb/db/forge-20260101T000000Z.db"
688 );
689 assert_eq!(layout.latest_link(), "/srv/sb/db/latest");
690 assert!(Layout::new("relative").is_err());
691 }
692
693 #[test]
694 fn snapshot_names_are_recognised_strictly() {
695 assert!(is_snapshot_name("forge-20260101T120000Z.db"));
696 assert!(!is_snapshot_name("forge-20260101T120000Z.db.tmp"));
697 assert!(!is_snapshot_name("latest"));
698 assert!(!is_snapshot_name(".tmp-forge-20260101T120000Z.db"));
699 assert!(!is_snapshot_name("forge-2026-01-01T12:00:00Z.db"));
700 assert!(!is_snapshot_name("forge-20260101X120000Z.db"));
701 }
702
703 #[test]
704 fn snapshot_stamp_has_the_expected_shape() {
705 let stamp = snapshot_stamp();
706 assert!(is_snapshot_name(&format!("forge-{stamp}.db")), "{stamp}");
707 }
708
709 #[test]
710 fn ssh_argv_pins_the_host_key_and_identity() {
711 let target = Target::parse("sb@standby.example:/srv/sb", 2200).expect("target");
712 let transport = SshTransport::new(
713 target,
714 PathBuf::from("/etc/forge/standby_key"),
715 PathBuf::from("/etc/forge/standby_known_hosts"),
716 );
717 let argv = transport.ssh_argv("true");
718 let expected = [
719 "-i",
720 "/etc/forge/standby_key",
721 "-o",
722 "IdentitiesOnly=yes",
723 "-o",
724 "UserKnownHostsFile=/etc/forge/standby_known_hosts",
725 "-o",
726 "StrictHostKeyChecking=yes",
727 "-o",
728 "BatchMode=yes",
729 "-o",
730 "ConnectTimeout=20",
731 "-p",
732 "2200",
733 "sb@standby.example",
734 "true",
735 ];
736 assert_eq!(argv, expected);
737 assert!(!argv.iter().any(|a| a.contains("StrictHostKeyChecking=no")));
738 assert_eq!(
739 transport.git_ssh_command(),
740 "'ssh' '-i' '/etc/forge/standby_key' '-o' 'IdentitiesOnly=yes' \
741 '-o' 'UserKnownHostsFile=/etc/forge/standby_known_hosts' \
742 '-o' 'StrictHostKeyChecking=yes' '-o' 'BatchMode=yes' \
743 '-o' 'ConnectTimeout=20' '-p' '2200'"
744 );
745 }
746
747 #[test]
748 fn upload_command_writes_a_temp_then_renames() {
749 assert_eq!(
750 upload_command("/srv/sb/db/forge-20260101T000000Z.db"),
751 "cat > '/srv/sb/db/.tmp-forge-20260101T000000Z.db' && mv -f \
752 '/srv/sb/db/.tmp-forge-20260101T000000Z.db' '/srv/sb/db/forge-20260101T000000Z.db'"
753 );
754 }
755
756 #[test]
757 fn queue_debounces_and_drains_due_repos() {
758 let queue = ReplicationQueue::default();
759 queue.enqueue("b");
760 queue.enqueue("a");
761 assert!(queue.take_due(Instant::now()).is_empty(), "not due yet");
762 let later = Instant::now() + PUSH_DEBOUNCE + Duration::from_millis(10);
763 assert_eq!(queue.take_due(later), ["a", "b"]);
764 assert!(queue.take_due(later).is_empty(), "drained");
765 assert!(!queue.take_reconcile_request());
766 queue.request_reconcile();
767 assert!(queue.take_reconcile_request());
768 assert!(!queue.take_reconcile_request(), "one request, one pass");
769 }
770}