Skip to main content

max / makenotwork

ops-exec: split pull into pull_file/pull_dir; move backup.rs onto the executor Executor::pull had two impls with incompatible path semantics: AgentRpc streamed a single file, while SshExec/LocalExec appended a trailing slash for rsync's directory-contents form. Nothing enforced which you had -- the contract lived in a doc comment -- and transport is a config-level choice (HostTransport::Agent|Local|Ssh), so flipping bento's driver to ssh would have silently produced `rsync mbp:/path/App.dmg/`. push documented itself as "directory/file"; the file half was never true. Make the distinction a type: pull_file / pull_dir / push_dir. AgentRpc implements pull_file and refuses pull_dir + push_dir by design (it serves one file per /pull). The rsync transports do both, trailing slash only on the dir form. Sando's backup.rs then moves off the last bespoke ssh in the daemon. Two things had to grow first, both opt-out so every existing caller is byte-identical: - SyncOpts.compress (-z): backups and DMGs are already compressed, so -z burned CPU for nothing. - SyncOpts.partial (--partial): rsync_command forced it, but backup.rs documents at :142 that a truncated leftover is DANGEROUS here -- a resumed fetch could splice two different dumps into one plausible file (CF4). Forcing it would have re-introduced exactly that. The port lives on RemoteHost (with_port), not on one transport, so the exec path (ssh -p) and the sync path (rsync -e "ssh -p") cannot disagree about which port a host is on. One deliberate behavior change: backup fetches now carry the shared SSH_FLAGS, which adds ConnectTimeout=10 to a path that had none. Also fixes bento-tui, whose test build was broken on main (TailStatus lacked Debug) -- its 5 tests had never been running -- plus two pre-existing collapsible-if lints there. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-16 17:04 UTC
Commit: b0a97aa93bd1e829b88728733fba9d8d66b40a9f
Parent: 28a7dd4
10 files changed, +368 insertions, -112 deletions
M bento/driver/src/lib.rs +3 -3
@@ -180,9 +180,9 @@ fn failure_tail(out: &RunOutput) -> String {
180 180 }
181 181
182 182 /// Drive the full macOS release recipe through `exec`, streaming all output to
183 - /// `sink`. Does NOT pull the artifact — the caller does that (transport choice
184 - /// belongs there: `AgentRpc::pull` for a single DMG, or `SshExec::pull` for a
185 - /// dir). Returns whether Gatekeeper accepted the DMG.
183 + /// `sink`. Does NOT pull the artifact — the caller does that, via
184 + /// `Executor::pull_file` (the DMG is one file; every transport that supports a
185 + /// file pull handles it). Returns whether Gatekeeper accepted the DMG.
186 186 pub async fn run_release(
187 187 exec: &dyn Executor,
188 188 plan: &ReleasePlan,
M bento/driver/src/main.rs +4 -2
@@ -115,10 +115,12 @@ async fn main() -> Result<()> {
115 115 .await
116 116 .with_context(|| format!("creating dest dir {}", cfg.local.dest_dir.display()))?;
117 117 let local_dmg = cfg.local.dest_dir.join(&dmg_name);
118 - exec.pull(
118 + // One file, and an already-compressed one: `precompressed` skips rsync's -z
119 + // if this ever runs over SshExec (AgentRpc streams and ignores the opts).
120 + exec.pull_file(
119 121 std::path::Path::new(&outcome.dmg_remote),
120 122 &local_dmg,
121 - &Default::default(),
123 + &ops_exec::SyncOpts::precompressed(),
122 124 )
123 125 .await
124 126 .context("pulling the signed DMG back")?;
M bento/tui/src/main.rs +51 -51
@@ -114,7 +114,7 @@ struct StepTail {
114 114 status: TailStatus,
115 115 }
116 116
117 - #[derive(Clone, PartialEq, Eq)]
117 + #[derive(Clone, Debug, PartialEq, Eq)]
118 118 enum TailStatus {
119 119 InFlight,
120 120 Finished(String),
@@ -167,10 +167,10 @@ impl Shared {
167 167 }
168 168
169 169 fn open_tail(&mut self, run_id: StepRunId, target: String, step: String) {
170 - if self.tails.len() >= TAILS_CAP {
171 - if let Some((&oldest, _)) = self.tails.iter().next() {
172 - self.tails.remove(&oldest);
173 - }
170 + if self.tails.len() >= TAILS_CAP
171 + && let Some((&oldest, _)) = self.tails.iter().next()
172 + {
173 + self.tails.remove(&oldest);
174 174 }
175 175 self.tails.insert(run_id, StepTail::new(target, step));
176 176 self.focus_run = Some(run_id);
@@ -466,55 +466,55 @@ fn ui_loop<B: Backend>(
466 466 loop {
467 467 term.draw(|f| draw(f, daemon, shared))?;
468 468
469 - if event::poll(Duration::from_millis(120))? {
470 - if let XEvent::Key(k) = event::read()? {
471 - match k.code {
472 - KeyCode::Char('q') | KeyCode::Esc => return Ok(()),
473 - KeyCode::Char('c') if k.modifiers.contains(KeyModifiers::CONTROL) => {
474 - return Ok(());
475 - }
476 - KeyCode::Up | KeyCode::Char('k') => {
477 - let mut g = shared.lock().unwrap();
478 - g.selected = g.selected.saturating_sub(1);
479 - }
480 - KeyCode::Down | KeyCode::Char('j') => {
481 - let mut g = shared.lock().unwrap();
482 - let max = g
483 - .state
484 - .as_ref()
485 - .and_then(|s| s.build.as_ref())
486 - .map(|b| b.targets.len().saturating_sub(1))
487 - .unwrap_or(0);
488 - if g.selected < max {
489 - g.selected += 1;
490 - }
491 - }
492 - KeyCode::Char('b') => {
493 - let app = shared.lock().unwrap().app_name(default_app);
494 - let _ = action_tx.try_send(Action::Build { app });
495 - }
496 - KeyCode::Char('R') => {
497 - let (app, target) = {
498 - let g = shared.lock().unwrap();
499 - (g.app_name(default_app), g.selected_target())
500 - };
501 - if let Some(target) = target {
502 - let _ = action_tx.try_send(Action::Retry { app, target });
503 - }
504 - }
505 - KeyCode::Char('r') => {
506 - shared.lock().unwrap().notice = Some("refresh on next tick".into());
507 - }
508 - KeyCode::Char('[') => {
509 - let mut g = shared.lock().unwrap();
510 - g.focus_run = cycle_focus(&g.tails, g.focus_run, -1);
469 + if event::poll(Duration::from_millis(120))?
470 + && let XEvent::Key(k) = event::read()?
471 + {
472 + match k.code {
473 + KeyCode::Char('q') | KeyCode::Esc => return Ok(()),
474 + KeyCode::Char('c') if k.modifiers.contains(KeyModifiers::CONTROL) => {
475 + return Ok(());
476 + }
477 + KeyCode::Up | KeyCode::Char('k') => {
478 + let mut g = shared.lock().unwrap();
479 + g.selected = g.selected.saturating_sub(1);
480 + }
481 + KeyCode::Down | KeyCode::Char('j') => {
482 + let mut g = shared.lock().unwrap();
483 + let max = g
484 + .state
485 + .as_ref()
486 + .and_then(|s| s.build.as_ref())
487 + .map(|b| b.targets.len().saturating_sub(1))
488 + .unwrap_or(0);
489 + if g.selected < max {
490 + g.selected += 1;
511 491 }
512 - KeyCode::Char(']') => {
513 - let mut g = shared.lock().unwrap();
514 - g.focus_run = cycle_focus(&g.tails, g.focus_run, 1);
492 + }
493 + KeyCode::Char('b') => {
494 + let app = shared.lock().unwrap().app_name(default_app);
495 + let _ = action_tx.try_send(Action::Build { app });
496 + }
497 + KeyCode::Char('R') => {
498 + let (app, target) = {
499 + let g = shared.lock().unwrap();
500 + (g.app_name(default_app), g.selected_target())
501 + };
502 + if let Some(target) = target {
503 + let _ = action_tx.try_send(Action::Retry { app, target });
515 504 }
516 - _ => {}
517 505 }
506 + KeyCode::Char('r') => {
507 + shared.lock().unwrap().notice = Some("refresh on next tick".into());
508 + }
509 + KeyCode::Char('[') => {
510 + let mut g = shared.lock().unwrap();
511 + g.focus_run = cycle_focus(&g.tails, g.focus_run, -1);
512 + }
513 + KeyCode::Char(']') => {
514 + let mut g = shared.lock().unwrap();
515 + g.focus_run = cycle_focus(&g.tails, g.focus_run, 1);
516 + }
517 + _ => {}
518 518 }
519 519 }
520 520 }
M sando/daemon/src/backup.rs +15 -18
@@ -14,6 +14,7 @@ use crate::config::Config;
14 14 use crate::topology::Topology;
15 15 use anyhow::{Context, Result, bail};
16 16 use chrono::Utc;
17 + use ops_exec::{CapabilitySet, Executor, SshExec, SyncOpts};
17 18 use sqlx::SqlitePool;
18 19 use std::path::Path;
19 20 use std::sync::Arc;
@@ -176,25 +177,21 @@ pub async fn fetch(
176 177 String::from_utf8_lossy(&out.stderr),
177 178 );
178 179 }
179 - BackupSource::Ssh { user_host, port, path } => {
180 - let ssh_cmd = match port {
181 - Some(p) => format!("ssh -p {p} -o BatchMode=yes -o StrictHostKeyChecking=accept-new"),
182 - None => "ssh -o BatchMode=yes -o StrictHostKeyChecking=accept-new".into(),
183 - };
184 - let remote = format!("{user_host}:{path}");
185 - let out = Command::new("rsync")
186 - .args(["-a"])
187 - .arg("-e").arg(&ssh_cmd)
188 - .arg(&remote)
189 - .arg(&tmp_path)
190 - .output()
180 + BackupSource::Ssh { user_host, path, port } => {
181 + // Through the executor rather than a hand-rolled `rsync -e ssh`:
182 + // one transport, one set of SSH flags. Fetching a dump needs no
183 + // authority on the host beyond the SSH key itself, so the
184 + // capability set is empty — `pull_file` is data movement, not a
185 + // gated action.
186 + let exec = SshExec::new(user_host, CapabilitySet::default()).with_port(port);
187 + // NOT `--partial`: a truncated leftover here is dangerous, not
188 + // useful — a resumed fetch could splice two different dumps into
189 + // one plausible-looking file (CF4). NOT `-z` either: the dump is
190 + // already compressed.
191 + let opts = SyncOpts { compress: false, partial: false, ..SyncOpts::default() };
192 + exec.pull_file(Path::new(&path), Path::new(&tmp_path), &opts)
191 193 .await
192 - .context("spawning rsync")?;
193 - anyhow::ensure!(
194 - out.status.success(),
195 - "rsync (ssh) failed: {}",
196 - String::from_utf8_lossy(&out.stderr),
197 - );
194 + .context("rsync (ssh) failed")?;
198 195 }
199 196 }
200 197 verify_backup(&tmp_path, is_gz, min_bytes).await
M sando/daemon/src/deploy.rs +1 -1
@@ -165,7 +165,7 @@ async fn deploy_remote(
165 165 // --chmod: F+X preserves the execute bit per-file (binaries land 0755,
166 166 // data files 0644) instead of a blanket 0755.
167 167 executor
168 - .push(staged_release_dir, Path::new(&release_dir), &SyncOpts::release_mirror())
168 + .push_dir(staged_release_dir, Path::new(&release_dir), &SyncOpts::release_mirror())
169 169 .await
170 170 .context("rsync failed (current symlink left intact)")?;
171 171
M shared/ops-exec/src/executor.rs +51 -8
@@ -13,24 +13,46 @@ use async_trait::async_trait;
13 13 use std::path::Path;
14 14 use tokio::process::Command;
15 15
16 - /// Options controlling an rsync push/pull. Default = a plain `-a --partial`
16 + /// Options controlling an rsync push/pull. Default = a plain `-az --partial`
17 17 /// mirror that never prunes the destination (safe for artifact collection).
18 18 /// Sando's release-dir deploy opts in to `delete` + `chmod` to keep its exact
19 19 /// behavior.
20 - #[derive(Clone, Debug, Default)]
20 + #[derive(Clone, Debug)]
21 21 pub struct SyncOpts {
22 22 /// `--delete`: prune files on the destination that are gone from the
23 23 /// source (a true mirror). Off by default.
24 24 pub delete: bool,
25 25 /// `--chmod=<spec>`: force destination permissions per-file.
26 26 pub chmod: Option<String>,
27 + /// `-z`: compress in flight. On by default, which is right for source trees
28 + /// and release dirs. Turn it OFF for already-compressed payloads (a `.gz`
29 + /// dump, a `.dmg`) — `-z` then burns CPU on both ends to save ~nothing.
30 + pub compress: bool,
31 + /// `--partial`: keep a partially-transferred file so a retry can resume it.
32 + /// On by default (worth it for a multi-hundred-MB artifact over a flaky
33 + /// link). Turn it OFF when a truncated leftover is *dangerous* rather than
34 + /// merely useless — e.g. fetching a DB dump, where a resumed transfer could
35 + /// splice two different dumps into one plausible-looking file.
36 + pub partial: bool,
37 + }
38 +
39 + impl Default for SyncOpts {
40 + fn default() -> Self {
41 + Self { delete: false, chmod: None, compress: true, partial: true }
42 + }
27 43 }
28 44
29 45 impl SyncOpts {
30 46 /// The Sando release-dir mirror: prune stale assets and force exec bits the
31 47 /// way `deploy.rs` always has.
32 48 pub fn release_mirror() -> Self {
33 - Self { delete: true, chmod: Some("Du=rwx,Dgo=rx,Fu=rw,Fgo=r,F+X".into()) }
49 + Self { delete: true, chmod: Some("Du=rwx,Dgo=rx,Fu=rw,Fgo=r,F+X".into()), ..Self::default() }
50 + }
51 +
52 + /// For payloads that are already compressed (`.gz`/`.dmg`/`.zip`): skip
53 + /// `-z` so the transfer doesn't re-compress compressed bytes.
54 + pub fn precompressed() -> Self {
55 + Self { compress: false, ..Self::default() }
34 56 }
35 57 }
36 58
@@ -57,12 +79,33 @@ pub trait Executor: Send + Sync {
57 79 /// command is dispatched.
58 80 async fn run_streaming(&self, step: &Step, sink: &mut dyn LogSink) -> Result<RunOutput>;
59 81
60 - /// Pull artifacts a prior step produced back to the caller (rsync from the
61 - /// host into `local`).
62 - async fn pull(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()>;
82 + /// Pull ONE FILE a prior step produced back to the caller: `remote` names
83 + /// the file on the host, `local` the destination path.
84 + ///
85 + /// Split from [`Executor::pull_dir`] because the transports genuinely
86 + /// differ here, and conflating them is a silent foot-gun: the rsync
87 + /// transports need a trailing slash for directory-contents semantics and
88 + /// must NOT have one for a file, while [`crate::AgentRpc`] streams a single
89 + /// file over HTTP and cannot do directories at all. One `pull` taking either
90 + /// shape meant a config-level transport swap (`HostTransport::Agent` →
91 + /// `Ssh`) could turn a working file pull into `rsync host:/x/App.dmg/`.
92 + ///
93 + /// Contract is file→file. Pointing this at a *directory* is not checked and
94 + /// not an error on the rsync transports (`-a` implies `-r`, so the dir is
95 + /// copied nested under `local`); `AgentRpc` will fail on it. Use
96 + /// [`Executor::pull_dir`] when you mean a directory.
97 + async fn pull_file(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()>;
63 98
64 - /// Push a directory/file from the caller to the host (rsync into `remote`).
65 - async fn push(&self, local: &Path, remote: &Path, opts: &SyncOpts) -> Result<()>;
99 + /// Pull the CONTENTS OF A DIRECTORY back to the caller (rsync from the host
100 + /// into `local`). See [`Executor::pull_file`] for why this is separate.
101 + async fn pull_dir(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()>;
102 +
103 + /// Push the contents of a local directory to the host (rsync into `remote`).
104 + ///
105 + /// Directory-only: both rsync transports append a trailing slash, and
106 + /// `AgentRpc` refuses by design (bulk data onto a host goes over
107 + /// SshExec/rsync or a `git pull` step, not the agent's exec surface).
108 + async fn push_dir(&self, local: &Path, remote: &Path, opts: &SyncOpts) -> Result<()>;
66 109
67 110 /// Subscribe to the host's observe stream; `None` if no observe capability
68 111 /// (or, in v1, if no resident observer is wired — E3).
M shared/ops-exec/src/remote.rs +70 -4
@@ -50,10 +50,15 @@ pub trait LogSink: Send {
50 50 }
51 51
52 52 /// A build host: the local machine or an SSH target (a tailnet alias such as
53 - /// `mbp`, or `user@host`).
53 + /// `mbp`, or `user@host`), optionally on a non-default SSH port.
54 54 #[derive(Debug, Clone)]
55 55 pub struct RemoteHost {
56 56 ssh_target: String,
57 + /// `None` = ssh's default (22, or whatever `~/.ssh/config` says for this
58 + /// target). The port lives here rather than on one transport so the exec
59 + /// path (`ssh -p`) and the sync path (`rsync -e "ssh -p"`) can never
60 + /// disagree about which port a host is on.
61 + port: Option<u16>,
57 62 }
58 63
59 64 /// Result of a streamed run: the exit status plus a bounded recent tail of the
@@ -89,9 +94,16 @@ pub(crate) fn push_bounded(buf: &mut Vec<u8>, chunk: &[u8], cap: usize) {
89 94
90 95 impl RemoteHost {
91 96 /// `"local"` (or empty) runs commands directly via `sh -c`; anything else
92 - /// is an SSH target.
97 + /// is an SSH target, on ssh's default port until [`RemoteHost::with_port`].
93 98 pub fn new(ssh_target: impl Into<String>) -> Self {
94 - Self { ssh_target: ssh_target.into() }
99 + Self { ssh_target: ssh_target.into(), port: None }
100 + }
101 +
102 + /// Reach this host on a non-default SSH port. `None` keeps ssh's default,
103 + /// so a caller with an `Option<u16>` can pass it straight through.
104 + pub fn with_port(mut self, port: Option<u16>) -> Self {
105 + self.port = port;
106 + self
95 107 }
96 108
97 109 pub fn is_local(&self) -> bool {
@@ -102,6 +114,23 @@ impl RemoteHost {
102 114 &self.ssh_target
103 115 }
104 116
117 + pub fn port(&self) -> Option<u16> {
118 + self.port
119 + }
120 +
121 + /// The `ssh` argv this host is reached with, minus the target: the shared
122 + /// [`SSH_FLAGS`] plus `-p <port>` when one is set. Shared by the exec path
123 + /// ([`RemoteHost::command`]) and the sync path (rsync's `-e`), so the two
124 + /// cannot drift.
125 + pub(crate) fn ssh_args(&self) -> Vec<String> {
126 + let mut args: Vec<String> = SSH_FLAGS.iter().map(|s| s.to_string()).collect();
127 + if let Some(p) = self.port {
128 + args.push("-p".into());
129 + args.push(p.to_string());
130 + }
131 + args
132 + }
133 +
105 134 /// Build the `Command` that runs `script` as a single `/bin/sh` program,
106 135 /// locally or over SSH. The remote side runs `script` as the argument to
107 136 /// the login shell (ssh joins argv with spaces and hands it to the shell),
@@ -113,7 +142,7 @@ impl RemoteHost {
113 142 cmd
114 143 } else {
115 144 let mut cmd = Command::new("ssh");
116 - cmd.args(SSH_FLAGS).arg(&self.ssh_target).arg(script);
145 + cmd.args(self.ssh_args()).arg(&self.ssh_target).arg(script);
117 146 cmd
118 147 }
119 148 }
@@ -181,6 +210,43 @@ mod tests {
181 210 use super::*;
182 211
183 212 #[test]
213 + fn ssh_args_carry_the_port_and_the_shared_flags() {
214 + // The point of the port living on RemoteHost: the exec path and the
215 + // sync path build their ssh invocation from this one place, so they
216 + // cannot disagree about which port a host is on.
217 + let host = RemoteHost::new("backup@db.example").with_port(Some(2222));
218 + let args = host.ssh_args();
219 + let p = args.iter().position(|a| a == "-p").expect("-p present");
220 + assert_eq!(args[p + 1], "2222");
221 + assert!(args.contains(&"BatchMode=yes".to_string()));
222 +
223 + // Default: no -p, so ssh/ssh_config picks the port.
224 + assert!(!RemoteHost::new("mbp").ssh_args().contains(&"-p".to_string()));
225 + }
226 +
227 + #[test]
228 + fn ported_host_puts_the_port_on_the_ssh_command() {
229 + let host = RemoteHost::new("backup@db.example").with_port(Some(2222));
230 + let cmd = host.command("true");
231 + let args: Vec<String> =
232 + cmd.as_std().get_args().map(|a| a.to_string_lossy().to_string()).collect();
233 + let p = args.iter().position(|a| a == "-p").expect("-p on the ssh argv");
234 + assert_eq!(args[p + 1], "2222");
235 + // The script is still the last arg, after the target.
236 + assert_eq!(args.last().unwrap(), "true");
237 + }
238 +
239 + #[test]
240 + fn local_host_ignores_the_port() {
241 + // "local" runs via `sh -c`; a port is meaningless and must not leak in.
242 + let host = RemoteHost::new("local").with_port(Some(2222));
243 + let cmd = host.command("true");
244 + let args: Vec<String> =
245 + cmd.as_std().get_args().map(|a| a.to_string_lossy().to_string()).collect();
246 + assert_eq!(args, vec!["-c".to_string(), "true".to_string()]);
247 + }
248 +
249 + #[test]
184 250 fn push_bounded_keeps_only_the_last_cap_bytes() {
185 251 let mut buf = Vec::new();
186 252 push_bounded(&mut buf, b"hello", 8);
M shared/ops-exec/src/rpc.rs +14 -3
@@ -135,7 +135,7 @@ impl Executor for AgentRpc {
135 135 Ok(RunOutput { status: exit_status(code), stdout: captured, stderr: Vec::new() })
136 136 }
137 137
138 - async fn pull(&self, remote: &Path, local: &Path, _opts: &SyncOpts) -> Result<()> {
138 + async fn pull_file(&self, remote: &Path, local: &Path, _opts: &SyncOpts) -> Result<()> {
139 139 // Caller-side enforcement (half 1 of 2), same as `run_streaming`. `/pull`
140 140 // is observe-plane: it needs the `artifact` grant. Failing fast here
141 141 // matters more than elsewhere — a pull is the *last* step of a release,
@@ -176,12 +176,23 @@ impl Executor for AgentRpc {
176 176 Ok(())
177 177 }
178 178
179 - async fn push(&self, _local: &Path, _remote: &Path, _opts: &SyncOpts) -> Result<()> {
179 + async fn pull_dir(&self, _remote: &Path, _local: &Path, _opts: &SyncOpts) -> Result<()> {
180 + // `/pull` serves exactly one file per request (a streamed HTTP body);
181 + // there is no directory form, and inventing one would mean walking the
182 + // host's tree over the agent's exec surface. Artifact retrieval is
183 + // per-file by design — collect a directory with SshExec/rsync.
184 + anyhow::bail!(
185 + "AgentRpc::pull_dir is unsupported by design; the agent serves one file per \
186 + /pull — use pull_file, or SshExec/rsync for a directory"
187 + )
188 + }
189 +
190 + async fn push_dir(&self, _local: &Path, _remote: &Path, _opts: &SyncOpts) -> Result<()> {
180 191 // The agent transport is for in-session *execution*. Bulk data movement
181 192 // onto the host uses SshExec/rsync or a `git pull` step inside the
182 193 // recipe — keeping the agent's surface small (one open port, exec only).
183 194 anyhow::bail!(
184 - "AgentRpc::push is unsupported by design; move source/data with SshExec/rsync \
195 + "AgentRpc::push_dir is unsupported by design; move source/data with SshExec/rsync \
185 196 or a `git pull` step, not the agent"
186 197 )
187 198 }
M shared/ops-exec/src/transport.rs +130 -17
@@ -13,7 +13,7 @@
13 13
14 14 use crate::capability::{CapabilityDenied, CapabilitySet};
15 15 use crate::executor::{Executor, SyncOpts, run_command_into_sink};
16 - use crate::remote::{LogSink, RemoteHost, RunOutput, SSH_FLAGS, sh_quote};
16 + use crate::remote::{LogSink, RemoteHost, RunOutput, sh_quote};
17 17 use crate::step::Step;
18 18 use anyhow::{Context, Result};
19 19 use async_trait::async_trait;
@@ -63,9 +63,11 @@ fn validate_env_names(step: &Step) -> Result<()> {
63 63 Ok(())
64 64 }
65 65
66 - /// Build the rsync `Command` shared by local and ssh push/pull. `remote_spec`
67 - /// is either a plain path (local) or `target:path` (ssh).
68 - fn rsync_command(src: &str, dst: &str, over_ssh: bool, opts: &SyncOpts) -> Command {
66 + /// Build the rsync `Command` shared by local and ssh push/pull. `src`/`dst` are
67 + /// either plain paths (local) or `target:path` (ssh). `ssh_args`, when `Some`,
68 + /// is the `ssh` argv rsync should transport over — [`RemoteHost::ssh_args`],
69 + /// so the port and flags match the exec path exactly; `None` = a local rsync.
70 + fn rsync_command(src: &str, dst: &str, ssh_args: Option<Vec<String>>, opts: &SyncOpts) -> Command {
69 71 let mut rsync = Command::new("rsync");
70 72 // Kill the transfer if the caller's future is dropped (e.g. a promote whose
71 73 // HTTP handler was cancelled by a client disconnect). Without this the
@@ -74,15 +76,21 @@ fn rsync_command(src: &str, dst: &str, over_ssh: bool, opts: &SyncOpts) -> Comma
74 76 // deploy. Matches the ssh-exec (remote.rs) and local-exec (executor.rs)
75 77 // paths, which already set it.
76 78 rsync.kill_on_drop(true);
77 - rsync.arg("-az").arg("--partial");
79 + rsync.arg("-a");
80 + if opts.partial {
81 + rsync.arg("--partial");
82 + }
83 + if opts.compress {
84 + rsync.arg("-z");
85 + }
78 86 if opts.delete {
79 87 rsync.arg("--delete");
80 88 }
81 89 if let Some(chmod) = &opts.chmod {
82 90 rsync.arg(format!("--chmod={chmod}"));
83 91 }
84 - if over_ssh {
85 - rsync.arg("-e").arg(format!("ssh {}", SSH_FLAGS.join(" ")));
92 + if let Some(args) = ssh_args {
93 + rsync.arg("-e").arg(format!("ssh {}", args.join(" ")));
86 94 }
87 95 rsync.arg(src).arg(dst);
88 96 rsync
@@ -119,15 +127,21 @@ impl Executor for LocalExec {
119 127 run_command_into_sink(cmd, sink).await
120 128 }
121 129
122 - async fn pull(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()> {
130 + async fn pull_file(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()> {
131 + // No trailing slash: rsync copies the file itself.
132 + let src = remote.to_string_lossy().to_string();
133 + run_rsync(rsync_command(&src, &local.to_string_lossy(), None, opts), "pull_file(local)").await
134 + }
135 +
136 + async fn pull_dir(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()> {
123 137 // Local "pull" is just a local rsync; trailing slash = contents.
124 138 let src = format!("{}/", remote.display());
125 - run_rsync(rsync_command(&src, &local.to_string_lossy(), false, opts), "pull(local)").await
139 + run_rsync(rsync_command(&src, &local.to_string_lossy(), None, opts), "pull_dir(local)").await
126 140 }
127 141
128 - async fn push(&self, local: &Path, remote: &Path, opts: &SyncOpts) -> Result<()> {
142 + async fn push_dir(&self, local: &Path, remote: &Path, opts: &SyncOpts) -> Result<()> {
129 143 let src = format!("{}/", local.display());
130 - run_rsync(rsync_command(&src, &remote.to_string_lossy(), false, opts), "push(local)").await
144 + run_rsync(rsync_command(&src, &remote.to_string_lossy(), None, opts), "push_dir(local)").await
131 145 }
132 146
133 147 fn capabilities(&self) -> &CapabilitySet {
@@ -146,6 +160,14 @@ impl SshExec {
146 160 Self { host: RemoteHost::new(ssh_target), caps }
147 161 }
148 162
163 + /// Reach this host on a non-default SSH port. `None` keeps ssh's default,
164 + /// so a caller holding an `Option<u16>` can pass it straight through. The
165 + /// port applies to both the exec and sync paths (see [`RemoteHost`]).
166 + pub fn with_port(mut self, port: Option<u16>) -> Self {
167 + self.host = self.host.with_port(port);
168 + self
169 + }
170 +
149 171 pub fn ssh_target(&self) -> &str {
150 172 self.host.ssh_target()
151 173 }
@@ -160,15 +182,24 @@ impl Executor for SshExec {
160 182 run_command_into_sink(cmd, sink).await
161 183 }
162 184
163 - async fn pull(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()> {
185 + async fn pull_file(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()> {
186 + // No trailing slash: rsync copies the file itself, not "contents of".
187 + let src = format!("{}:{}", self.host.ssh_target(), remote.display());
188 + let ssh = Some(self.host.ssh_args());
189 + run_rsync(rsync_command(&src, &local.to_string_lossy(), ssh, opts), "pull_file(ssh)").await
190 + }
191 +
192 + async fn pull_dir(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()> {
164 193 let src = format!("{}:{}/", self.host.ssh_target(), remote.display());
165 - run_rsync(rsync_command(&src, &local.to_string_lossy(), true, opts), "pull(ssh)").await
194 + let ssh = Some(self.host.ssh_args());
195 + run_rsync(rsync_command(&src, &local.to_string_lossy(), ssh, opts), "pull_dir(ssh)").await
166 196 }
167 197
168 - async fn push(&self, local: &Path, remote: &Path, opts: &SyncOpts) -> Result<()> {
198 + async fn push_dir(&self, local: &Path, remote: &Path, opts: &SyncOpts) -> Result<()> {
169 199 let src = format!("{}/", local.display());
170 200 let dst = format!("{}:{}/", self.host.ssh_target(), remote.display());
171 - run_rsync(rsync_command(&src, &dst, true, opts), "push(ssh)").await
201 + let ssh = Some(self.host.ssh_args());
202 + run_rsync(rsync_command(&src, &dst, ssh, opts), "push_dir(ssh)").await
172 203 }
173 204
174 205 fn capabilities(&self) -> &CapabilitySet {
@@ -267,9 +298,91 @@ mod tests {
267 298 tokio::fs::create_dir_all(&src).await.unwrap();
268 299 tokio::fs::write(src.join("f.txt"), b"hi").await.unwrap();
269 300 let exec = LocalExec::new(CapabilitySet::default());
270 - exec.push(&src, &mid, &SyncOpts::default()).await.unwrap();
301 + exec.push_dir(&src, &mid, &SyncOpts::default()).await.unwrap();
271 302 assert_eq!(tokio::fs::read(mid.join("f.txt")).await.unwrap(), b"hi");
272 - exec.pull(&mid, &dst, &SyncOpts::default()).await.unwrap();
303 + exec.pull_dir(&mid, &dst, &SyncOpts::default()).await.unwrap();
273 304 assert_eq!(tokio::fs::read(dst.join("f.txt")).await.unwrap(), b"hi");
274 305 }
306 +
307 + /// The distinction the split exists for: a single file lands AS a file.
308 + /// Under the old dir-shaped `pull` this path was `{file}/` — rsync would
309 + /// refuse it.
310 + #[tokio::test]
311 + async fn local_pull_file_copies_one_file() {
312 + let dir = tempfile::tempdir().unwrap();
313 + let src = dir.path().join("dump.sql.gz");
314 + let dst = dir.path().join("fetched.sql.gz");
315 + tokio::fs::write(&src, b"DUMPBYTES").await.unwrap();
316 + let exec = LocalExec::new(CapabilitySet::default());
317 + exec.pull_file(&src, &dst, &SyncOpts::default()).await.unwrap();
318 + assert_eq!(tokio::fs::read(&dst).await.unwrap(), b"DUMPBYTES");
319 + }
320 +
321 + /// Pins what the file/dir split actually buys, which is NOT a type-level
322 + /// refusal: `-a` implies `-r`, so `pull_file` pointed at a directory does
323 + /// rsync's no-trailing-slash thing — copies the dir INTO the destination
324 + /// (`out/adir/f.txt`) rather than erroring. `pull_dir` would have put the
325 + /// contents at `out/f.txt`. Documented here because the difference is
326 + /// silent, and the docs on `pull_file` promise only file→file.
327 + #[tokio::test]
328 + async fn local_pull_file_on_a_dir_nests_rather_than_flattening() {
329 + let dir = tempfile::tempdir().unwrap();
330 + let src = dir.path().join("adir");
331 + let out = dir.path().join("out");
332 + tokio::fs::create_dir_all(&src).await.unwrap();
333 + tokio::fs::create_dir_all(&out).await.unwrap();
334 + tokio::fs::write(src.join("f.txt"), b"hi").await.unwrap();
335 + let exec = LocalExec::new(CapabilitySet::default());
336 + exec.pull_file(&src, &out, &SyncOpts::default()).await.unwrap();
337 + assert_eq!(tokio::fs::read(out.join("adir").join("f.txt")).await.unwrap(), b"hi");
338 + assert!(!out.join("f.txt").exists(), "pull_file does not flatten; that's pull_dir");
339 + }
340 +
341 + #[test]
342 + fn sync_opts_flags_are_opt_out() {
343 + // Default = the historical `-az --partial`.
344 + let cmd = rsync_command("s", "d", None, &SyncOpts::default());
345 + let args = render_args(&cmd);
346 + assert!(args.contains(&"-z".to_string()), "compress on by default: {args:?}");
347 + assert!(args.contains(&"--partial".to_string()), "partial on by default: {args:?}");
348 +
349 + // A precompressed payload drops -z but keeps --partial.
350 + let args = render_args(&rsync_command("s", "d", None, &SyncOpts::precompressed()));
351 + assert!(!args.contains(&"-z".to_string()), "precompressed drops -z: {args:?}");
352 + assert!(args.contains(&"--partial".to_string()));
353 +
354 + // Sando's backup fetch drops both: a resumed dump could splice two
355 + // different backups (CF4).
356 + let opts = SyncOpts { compress: false, partial: false, ..SyncOpts::default() };
357 + let args = render_args(&rsync_command("s", "d", None, &opts));
358 + assert!(!args.contains(&"-z".to_string()));
359 + assert!(!args.contains(&"--partial".to_string()));
360 +
361 + // release_mirror keeps prune + chmod, and still compresses.
362 + let args = render_args(&rsync_command("s", "d", None, &SyncOpts::release_mirror()));
363 + assert!(args.contains(&"--delete".to_string()));
364 + assert!(args.iter().any(|a| a.starts_with("--chmod=")));
365 + assert!(args.contains(&"-z".to_string()));
366 + }
367 +
368 + #[test]
369 + fn rsync_over_ssh_carries_the_port_into_dash_e() {
370 + let host = RemoteHost::new("backup@db.example").with_port(Some(2222));
371 + let cmd = rsync_command("src", "dst", Some(host.ssh_args()), &SyncOpts::default());
372 + let args = render_args(&cmd);
373 + let e = args.iter().position(|a| a == "-e").expect("-e present");
374 + let ssh_spec = &args[e + 1];
375 + assert!(ssh_spec.contains("-p 2222"), "port reaches rsync's ssh: {ssh_spec}");
376 + assert!(ssh_spec.contains("BatchMode=yes"), "shared flags kept: {ssh_spec}");
377 +
378 + // No port set ⇒ no -p at all (ssh/ssh_config decides).
379 + let plain = RemoteHost::new("mbp");
380 + let args = render_args(&rsync_command("s", "d", Some(plain.ssh_args()), &SyncOpts::default()));
381 + let e = args.iter().position(|a| a == "-e").unwrap();
382 + assert!(!args[e + 1].contains("-p"), "no port ⇒ no -p: {}", args[e + 1]);
383 + }
384 +
385 + fn render_args(cmd: &Command) -> Vec<String> {
386 + cmd.as_std().get_args().map(|a| a.to_string_lossy().to_string()).collect()
387 + }
275 388 }
M shared/ops-exec/tests/agent_e2e.rs +29 -5
@@ -165,7 +165,7 @@ async fn agent_pull_serves_an_in_root_file() {
165 165 .await;
166 166 let rpc = AgentRpc::new(base, "mbp", puller());
167 167 let local = dir.path().join("pulled.dmg");
168 - rpc.pull(&artifact, &local, &Default::default()).await.unwrap();
168 + rpc.pull_file(&artifact, &local, &Default::default()).await.unwrap();
169 169 assert_eq!(tokio::fs::read(&local).await.unwrap(), b"DMGBYTES");
170 170 }
171 171
@@ -191,7 +191,7 @@ async fn agent_pull_denied_with_only_build_log_observe() {
191 191 // Client-side grant is deliberately wide so the *agent* is what refuses.
192 192 let rpc = AgentRpc::new(base, "mbp", puller());
193 193 let local = dir.path().join("out.bin");
194 - let err = rpc.pull(&artifact, &local, &Default::default()).await.unwrap_err();
194 + let err = rpc.pull_file(&artifact, &local, &Default::default()).await.unwrap_err();
195 195 assert!(err.to_string().contains("artifact"), "denial names the grant: {err}");
196 196 assert!(!local.exists(), "denied pull must not write a file");
197 197 }
@@ -216,7 +216,7 @@ async fn agent_pull_denied_without_observe_grant() {
216 216 let local = dir.path().join("out.bin");
217 217 // The agent returns 403; AgentRpc surfaces the reason and the file never
218 218 // transfers.
219 - assert!(rpc.pull(&artifact, &local, &Default::default()).await.is_err());
219 + assert!(rpc.pull_file(&artifact, &local, &Default::default()).await.is_err());
220 220 assert!(!local.exists(), "denied pull must not write a file");
221 221 }
222 222
@@ -241,7 +241,7 @@ async fn agent_pull_denied_caller_side_without_declaring_artifact() {
241 241 // Agent would happily serve this; the caller's own set is what refuses.
242 242 let rpc = AgentRpc::new(base, "mbp", CapabilitySet::from_tokens(["build"], ["build-log"]));
243 243 let local = dir.path().join("out.bin");
244 - let err = rpc.pull(&artifact, &local, &Default::default()).await.unwrap_err();
244 + let err = rpc.pull_file(&artifact, &local, &Default::default()).await.unwrap_err();
245 245 assert!(err.to_string().contains("observe:artifact"), "caller-side denial: {err}");
246 246 assert!(!local.exists(), "denied pull must not write a file");
247 247 }
@@ -265,7 +265,31 @@ async fn agent_pull_denied_outside_root() {
265 265 .await;
266 266 let rpc = AgentRpc::new(base, "mbp", puller());
267 267 let local = dir.path().join("out.key");
268 - let err = rpc.pull(&secret, &local, &Default::default()).await.unwrap_err();
268 + let err = rpc.pull_file(&secret, &local, &Default::default()).await.unwrap_err();
269 269 assert!(!local.exists(), "out-of-root pull must not write a file");
270 270 let _ = err; // status is non-2xx; the point is the secret never transfers
271 271 }
272 +
273 + /// The agent serves exactly one file per `/pull`; there is no directory form.
274 + /// `pull_dir` must refuse locally rather than invent one — this is the half of
275 + /// the trait split that keeps a transport swap honest.
276 + #[tokio::test]
277 + async fn agent_pull_dir_is_refused_by_design() {
278 + let dir = tempfile::tempdir().unwrap();
279 + let root = dir.path().join("dist");
280 + tokio::fs::create_dir_all(&root).await.unwrap();
281 +
282 + let grant = GrantConfig { actuate: vec![], observe: vec!["artifact".into()] };
283 + let base = spawn_agent_with_pull(
284 + vec![CallerGrant { identity: "fw13".into(), actuate: vec![], observe: vec!["artifact".into()] }],
285 + grant,
286 + Some(root.clone()),
287 + )
288 + .await;
289 + let rpc = AgentRpc::new(base, "mbp", puller());
290 + let err = rpc
291 + .pull_dir(&root, &dir.path().join("out"), &Default::default())
292 + .await
293 + .expect_err("pull_dir must be refused, not attempted");
294 + assert!(err.to_string().contains("unsupported by design"), "{err}");
295 + }