strop_containers/
engine.rs

1//! The local engine conversation: probe, discovery and identity
2//! resolution. Every `docker` invocation is a supervised capture — argv
3//! arrays, a caller's [`CancelToken`], a deadline and bounded retention —
4//! and every non-zero exit is classified into a typed refusal.
5
6use crate::identity::{validate_name, ContainerIdentity, ContainerRef};
7use crate::ContainerError;
8use std::process::Command;
9use std::time::Duration;
10use strop_core::process::{
11    capture_with, stream_with, CaptureError, CapturePolicy, StdinPolicy, StreamError, StreamPolicy,
12};
13use strop_core::worker::CancelToken;
14
15/// Wall-clock budget for the `docker info` probe.
16const INFO_DEADLINE: Duration = Duration::from_secs(10);
17/// Budget for discovery/inspect metadata commands.
18const META_DEADLINE: Duration = Duration::from_secs(15);
19/// Budget for one filesystem read (`docker cp` tar stream).
20pub(crate) const READ_DEADLINE: Duration = Duration::from_secs(30);
21/// Retained stderr for diagnostics (tails only ever reach errors).
22const STDERR_LIMIT: u64 = 64 * 1024;
23/// Retained stdout for metadata commands (`info`/`ps`/`inspect`).
24const META_LIMIT: u64 = 4 * 1024 * 1024;
25/// Retained metadata for one directory listing: the direct children's
26/// names and kinds. The listing's tar stream itself is consumed
27/// incrementally and never retained, so a subtree's bulk no longer
28/// counts — only a listing with an absurd direct-child set is refused,
29/// never presented partially.
30pub(crate) const LIST_LIMIT: u64 = 16 * 1024 * 1024;
31
32/// The local Docker engine, proven reachable by a bounded probe.
33///
34/// v1 is local-only by construction: there is no endpoint field to
35/// parse, and there is deliberately no way to construct one without the
36/// probe — a remote-engine field is earned with DC5, not reserved.
37#[derive(Debug, Clone)]
38pub struct EngineRef {
39    server_version: String,
40}
41
42impl EngineRef {
43    /// `ServerVersion` as the probe reported it.
44    pub fn server_version(&self) -> &str {
45        &self.server_version
46    }
47}
48
49/// One supervised `docker` run's retained bytes — public for the
50/// `exec_capture` boundary.
51pub struct Captured {
52    pub code: Option<i32>,
53    pub stdout: Vec<u8>,
54    pub stderr: Vec<u8>,
55    pub stdout_dropped: u64,
56}
57
58/// Run `docker <args>` under supervision: own process group, the caller's
59/// token, `deadline`, bounded pipes, no stdin. Spawn failure means the
60/// CLI itself is missing — that is the engine being unavailable.
61pub(crate) fn capture(
62    args: &[&str],
63    stdout_limit: u64,
64    deadline: Duration,
65    token: &CancelToken,
66) -> Result<Captured, ContainerError> {
67    let mut command = Command::new("docker");
68    command.args(args);
69    let policy = CapturePolicy {
70        stdout_limit,
71        stderr_limit: STDERR_LIMIT,
72        stderr_tail: 0,
73        deadline,
74        stdin: StdinPolicy::Null,
75    };
76    let output = capture_with(&mut command, token, &policy).map_err(|error| match error {
77        CaptureError::Spawn(detail) => ContainerError::EngineUnavailable { detail },
78        CaptureError::Cancelled => ContainerError::Cancelled,
79        CaptureError::TimedOut(deadline) => ContainerError::Io {
80            detail: format!(
81                "docker {} timed out after {}s",
82                args.first().copied().unwrap_or("<none>"),
83                deadline.as_secs()
84            ),
85        },
86        CaptureError::Failure(failure) => ContainerError::Io {
87            detail: failure.message,
88        },
89    })?;
90    Ok(Captured {
91        code: output.status.code(),
92        stdout: output.stdout,
93        stderr: output.stderr,
94        stdout_dropped: output.stdout_dropped,
95    })
96}
97
98/// One supervised `docker` run whose stdout streamed through a consumer.
99pub(crate) struct Streamed {
100    pub code: Option<i32>,
101    pub stderr: Vec<u8>,
102}
103
104/// Run `docker <args>` under [`capture`]'s supervision while stdout
105/// streams through `consume` chunk by chunk: nothing is retained by the
106/// supervisor, so a transfer larger than any retention limit stays
107/// bounded by what the consumer keeps. Stderr retention, the deadline
108/// and cancellation behave exactly as in [`capture`]; a consumer error
109/// kills the child and surfaces as its own typed [`ContainerError`].
110pub(crate) fn stream(
111    args: &[&str],
112    deadline: Duration,
113    token: &CancelToken,
114    consume: impl FnMut(&[u8]) -> Result<(), ContainerError>,
115) -> Result<Streamed, ContainerError> {
116    let mut command = Command::new("docker");
117    command.args(args);
118    let policy = StreamPolicy {
119        stderr_limit: STDERR_LIMIT,
120        stderr_tail: 0,
121        deadline,
122        hold_stdin: false,
123    };
124    let output =
125        stream_with(&mut command, token, &policy, consume).map_err(|error| match error {
126            StreamError::Spawn(detail) => ContainerError::EngineUnavailable { detail },
127            StreamError::Cancelled => ContainerError::Cancelled,
128            StreamError::TimedOut(deadline) => ContainerError::Io {
129                detail: format!(
130                    "docker {} timed out after {}s",
131                    args.first().copied().unwrap_or("<none>"),
132                    deadline.as_secs()
133                ),
134            },
135            StreamError::Failure(failure) => ContainerError::Io {
136                detail: failure.message,
137            },
138            StreamError::Consumer(error) => error,
139        })?;
140    Ok(Streamed {
141        code: output.status.code(),
142        stderr: output.stderr,
143    })
144}
145
146/// The last bytes of stderr, lossy-decoded and single-lined — bounded
147/// context for a typed failure, never an unbounded dump.
148pub(crate) fn stderr_tail(stderr: &[u8]) -> String {
149    let text = String::from_utf8_lossy(stderr);
150    let tail: String = text.chars().rev().take(300).collect::<String>();
151    tail.chars().rev().collect::<String>().replace('\n', " ")
152}
153
154/// Probe the local engine: `docker info` must succeed and report a
155/// server version. CLI missing, daemon down or a timed-out probe are all
156/// [`ContainerError::EngineUnavailable`].
157pub fn engine(token: &CancelToken) -> Result<EngineRef, ContainerError> {
158    let output = capture(
159        &["info", "--format", "{{json .ServerVersion}}"],
160        META_LIMIT,
161        INFO_DEADLINE,
162        token,
163    )?;
164    if output.code != Some(0) {
165        return Err(ContainerError::EngineUnavailable {
166            detail: stderr_tail(&output.stderr),
167        });
168    }
169    let server_version: String =
170        serde_json::from_slice(&output.stdout).map_err(|error| ContainerError::Protocol {
171            detail: format!("docker info reported no server version: {error}"),
172        })?;
173    if server_version.is_empty() {
174        return Err(ContainerError::Protocol {
175            detail: "docker info reported an empty server version".into(),
176        });
177    }
178    Ok(EngineRef { server_version })
179}
180
181/// The engine's running containers as full identities: `ps` for the id
182/// set, one batched `inspect` for the records. Containers that stop in
183/// between are dropped — this lists running containers only.
184pub fn list_running(
185    engine: &EngineRef,
186    token: &CancelToken,
187) -> Result<Vec<ContainerIdentity>, ContainerError> {
188    let _ = engine;
189    let output = capture(
190        &["ps", "--quiet", "--no-trunc"],
191        META_LIMIT,
192        META_DEADLINE,
193        token,
194    )?;
195    if output.code != Some(0) {
196        return Err(ContainerError::EngineUnavailable {
197            detail: stderr_tail(&output.stderr),
198        });
199    }
200    let text = String::from_utf8(output.stdout).map_err(|_| ContainerError::Protocol {
201        detail: "docker ps answered in non-UTF-8".into(),
202    })?;
203    let ids: Vec<&str> = text.lines().filter(|line| !line.is_empty()).collect();
204    for id in &ids {
205        if id.len() != 64 || !id.bytes().all(|b| b.is_ascii_hexdigit()) {
206            return Err(ContainerError::Protocol {
207                detail: format!("docker ps reported a non-canonical id {id:?}"),
208            });
209        }
210    }
211    if ids.is_empty() {
212        return Ok(Vec::new());
213    }
214    let mut args = vec!["inspect"];
215    args.extend(ids);
216    let output = capture(&args, META_LIMIT, META_DEADLINE, token)?;
217    inspect_records("docker ps ids", &output)?
218        .into_iter()
219        .filter(|record| record.state.running)
220        .map(identity)
221        .collect()
222}
223
224/// Resolve a name or id prefix to the container's identity, canonical
225/// 64-hex id included. Names are validated before the engine sees them;
226/// an unknown name is [`ContainerError::NoSuchContainer`], never a
227/// best-effort guess.
228pub fn inspect(
229    engine: &EngineRef,
230    name_or_id: &str,
231    token: &CancelToken,
232) -> Result<ContainerIdentity, ContainerError> {
233    let _ = engine;
234    validate_name(name_or_id)?;
235    let output = capture(&["inspect", name_or_id], META_LIMIT, META_DEADLINE, token)?;
236    identity(inspect_record(name_or_id, &output)?)
237}
238
239/// Re-resolve a previously inspected identity by name. The stale-identity
240/// refusal lives here: a name that now maps to a different id, or the same
241/// id restarted since, must not silently inherit the old identity's
242/// reads, caches or completions.
243pub fn revalidate(
244    engine: &EngineRef,
245    held: &ContainerIdentity,
246    token: &CancelToken,
247) -> Result<ContainerRef, ContainerError> {
248    let current = inspect(engine, &held.name, token)?;
249    if current.id != held.id || current.started_at != held.started_at {
250        return Err(ContainerError::StaleIdentity {
251            name: held.name.clone(),
252            expected: format!("{}@{}", held.id, held.started_at),
253            found: format!("{}@{}", current.id, current.started_at),
254        });
255    }
256    ContainerRef::of(&current)
257}
258
259/// The cheap pre-read re-check: the container behind `reference` must
260/// still exist, still be the same incarnation (`StartedAt`), and still be
261/// running. One bounded `inspect` per read — the price of never serving
262/// bytes from a restarted container as if they were the old one's.
263pub(crate) fn refresh(
264    engine: &EngineRef,
265    reference: &ContainerRef,
266    token: &CancelToken,
267) -> Result<(), ContainerError> {
268    let _ = engine;
269    let id = reference.id().as_str();
270    let output = capture(&["inspect", id], META_LIMIT, META_DEADLINE, token)?;
271    let record = inspect_record(id, &output)?;
272    if record.state.started_at != reference.started_at() {
273        return Err(ContainerError::StaleIdentity {
274            name: id.to_string(),
275            expected: reference.incarnation(),
276            found: format!("{id}@{}", record.state.started_at),
277        });
278    }
279    if !record.state.running {
280        return Err(ContainerError::NotRunning { id: id.to_string() });
281    }
282    Ok(())
283}
284
285/// The fields of `docker inspect`'s JSON record this backend consumes.
286#[derive(serde::Deserialize)]
287struct InspectRecord {
288    #[serde(rename = "Id")]
289    id: String,
290    #[serde(rename = "Name", default)]
291    name: String,
292    #[serde(rename = "Config", default)]
293    config: InspectConfig,
294    #[serde(rename = "State", default)]
295    state: InspectState,
296}
297
298#[derive(Default, serde::Deserialize)]
299struct InspectConfig {
300    #[serde(rename = "Image", default)]
301    image: String,
302    #[serde(rename = "User", default)]
303    user: String,
304    /// The container's working directory (Config.WorkingDir); empty means
305    /// the image default ("/").
306    #[serde(rename = "WorkingDir", default)]
307    workdir: String,
308}
309
310#[derive(Default, serde::Deserialize)]
311struct InspectState {
312    #[serde(rename = "Running", default)]
313    running: bool,
314    #[serde(rename = "StartedAt", default)]
315    started_at: String,
316}
317
318/// Decode the single-record inspect answer, classifying failure output.
319fn inspect_record(name_or_id: &str, output: &Captured) -> Result<InspectRecord, ContainerError> {
320    Ok(inspect_records(name_or_id, output)?.remove(0))
321}
322
323/// Decode an inspect answer's JSON array; a non-zero exit naming a
324/// missing object is [`ContainerError::NoSuchContainer`].
325fn inspect_records(what: &str, output: &Captured) -> Result<Vec<InspectRecord>, ContainerError> {
326    if output.code != Some(0) {
327        let tail = stderr_tail(&output.stderr);
328        // Older daemons print "No such object", newer ones "no such object".
329        if tail.to_ascii_lowercase().contains("no such object") {
330            return Err(ContainerError::NoSuchContainer {
331                name: what.to_string(),
332            });
333        }
334        return Err(ContainerError::Io {
335            detail: format!("docker inspect failed: {tail}"),
336        });
337    }
338    if output.stdout_dropped > 0 {
339        return Err(ContainerError::OutputTooLarge {
340            what: "docker inspect output".into(),
341        });
342    }
343    let records: Vec<InspectRecord> =
344        serde_json::from_slice(&output.stdout).map_err(|error| ContainerError::Protocol {
345            detail: format!("docker inspect answered malformed JSON: {error}"),
346        })?;
347    if records.is_empty() {
348        return Err(ContainerError::Protocol {
349            detail: format!("docker inspect of {what} returned no record"),
350        });
351    }
352    Ok(records)
353}
354
355/// The identity view of one inspect record; the id must already be the
356/// canonical 64-hex form (a short-id answer would mean the engine broke
357/// its own contract).
358fn identity(record: InspectRecord) -> Result<ContainerIdentity, ContainerError> {
359    strop_workspace::ContainerId::canonical(record.id.clone()).map_err(|_| {
360        ContainerError::Protocol {
361            detail: format!("inspect id {:?} is not the canonical 64-hex id", record.id),
362        }
363    })?;
364    Ok(ContainerIdentity {
365        id: record.id,
366        name: record.name.trim_start_matches('/').to_string(),
367        image: record.config.image,
368        started_at: record.state.started_at,
369        user: record.config.user,
370        workdir: record.config.workdir,
371    })
372}
373
374#[cfg(test)]
375mod tests {
376    use super::*;
377
378    fn captured(code: Option<i32>, stdout: &[u8], stderr: &[u8]) -> Captured {
379        Captured {
380            code,
381            stdout: stdout.to_vec(),
382            stderr: stderr.to_vec(),
383            stdout_dropped: 0,
384        }
385    }
386
387    const INSPECT_JSON: &str = r#"[{
388        "Id": "5b04229f99d2c4b8ae3c4e38b7a887cf1c5c1ea51f2a1b2c3d4e5f60718293a4",
389        "Name": "/fixture",
390        "Config": {"Image": "busybox:latest", "User": "root"},
391        "State": {"Running": true, "StartedAt": "2026-09-10T08:00:00.123Z"}
392    }]"#;
393
394    #[test]
395    fn inspect_json_decodes_to_identity() {
396        let output = captured(Some(0), INSPECT_JSON.as_bytes(), b"");
397        let identity = identity(inspect_record("fixture", &output).unwrap()).unwrap();
398        assert_eq!(identity.id.len(), 64);
399        assert_eq!(identity.name, "fixture", "leading slash stripped");
400        assert_eq!(identity.image, "busybox:latest");
401        assert_eq!(identity.started_at, "2026-09-10T08:00:00.123Z");
402        assert_eq!(identity.user, "root");
403    }
404
405    #[test]
406    fn missing_object_and_daemon_errors_classify() {
407        let missing = captured(
408            Some(1),
409            b"[]",
410            b"Error response from daemon: No such object: ghost\n",
411        );
412        assert!(matches!(
413            inspect_record("ghost", &missing),
414            Err(ContainerError::NoSuchContainer { .. })
415        ));
416        // Docker 29's lowercase daemon spelling.
417        let lowercase = captured(Some(1), b"[]", b"error: no such object: ghost\n");
418        assert!(matches!(
419            inspect_record("ghost", &lowercase),
420            Err(ContainerError::NoSuchContainer { .. })
421        ));
422        let down = captured(
423            Some(1),
424            b"",
425            b"Cannot connect to the Docker daemon at unix:///var/run/docker.sock",
426        );
427        assert!(matches!(
428            inspect_record("ghost", &down),
429            Err(ContainerError::Io { .. })
430        ));
431        let malformed = captured(Some(0), b"[{]", b"");
432        assert!(matches!(
433            inspect_record("ghost", &malformed),
434            Err(ContainerError::Protocol { .. })
435        ));
436    }
437
438    #[test]
439    fn non_canonical_ids_are_protocol_violations() {
440        let json = INSPECT_JSON.replace(
441            "5b04229f99d2c4b8ae3c4e38b7a887cf1c5c1ea51f2a1b2c3d4e5f60718293a4",
442            "5b04229f99d2",
443        );
444        let output = captured(Some(0), json.as_bytes(), b"");
445        assert!(matches!(
446            inspect_record("x", &output).and_then(identity),
447            Err(ContainerError::Protocol { .. })
448        ));
449    }
450
451    #[test]
452    fn stderr_tail_is_bounded_and_single_line() {
453        let noisy = format!("{}\nfinal line", "x".repeat(1000));
454        let tail = stderr_tail(noisy.as_bytes());
455        assert!(tail.len() <= 300);
456        assert!(tail.ends_with("final line"));
457        assert!(!tail.contains('\n'));
458    }
459}
strop_containers/engine.rs

strop_containers/
engine.rs
