start_command/
execution_recovery.rs

1//! Launch-time recovery for killed detached docker sessions (issue #176).
2//!
3//! `$ --isolated docker --detached --on-kill-resume 3 --recovery-command B -- A`
4//! runs `A`; when the main process is killed (exit 137, or `OOMKilled` with no
5//! exit status of its own), the detached completion watcher hands the inspected
6//! facts to this module, which restarts the *same* container so `B` continues
7//! on the same filesystem, with the same resource limits (they live in the
8//! container's HostConfig), under the same execution UUID and appending to the
9//! same log file.
10//!
11//! How `B` replaces `A` inside the same container: a container launched with a
12//! recovery command runs a tiny selector as its command. The selector runs `A`
13//! normally, but runs `B` when the marker file exists. Before `docker start`,
14//! the marker (holding the attempt number) is copied into the stopped
15//! container with `docker cp`. Without a recovery command the original command
16//! is simply started again.
17//!
18//! Like `detached_finalize`, the entry point runs as a short-lived child of the
19//! watcher on a host whose CLI invocation is long gone, so it never panics on
20//! bad input and never prints anything.
21
22use std::fs;
23use std::io::Write;
24use std::path::PathBuf;
25
26use chrono::{SecondsFormat, Utc};
27use serde_json::{json, Value};
28
29use crate::cgroup_memory::{
30    cgroup_shell_vars, format_cgroup_memory_log_line, parse_cgroup_memory_sample,
31};
32use crate::detached_finalize::{normalize_bool, normalize_container_error};
33use crate::docker_cleanup::{
34    docker_command, get_docker_container_cleanup_policy,
35    start_detached_docker_completion_watcher_with, DockerWatcherOptions,
36};
37use crate::docker_post_mortem::{format_container_post_mortem, shell_vars, ContainerPostMortem};
38use crate::docker_resource_limits::{
39    build_resource_limits_status_line, read_docker_resource_limits,
40};
41use crate::execution_control::{CommandRunOutput, CommandRunner, SystemCommandRunner};
42use crate::execution_resume::build_launch_options;
43use crate::execution_store::{
44    ExecutionRecord, ExecutionStatus, ExecutionStore, ExecutionStoreOptions,
45};
46use crate::exit_reason::describe_exit_code_str;
47use crate::isolation::isolation_log::shell_quote;
48use crate::recovery_delay::{
49    format_recovery_delay, pick_recovery_delay_ms, record_on_kill_resume_delay, system_random,
50    system_sleep, wait_for_recovery_delay,
51};
52
53/// Hidden argument the detached watcher re-invokes this binary with.
54pub const INTERNAL_RECOVER_FLAG: &str = "--internal-recover-detached-docker";
55
56/// File whose presence makes the selector run the recovery command.
57pub const RECOVERY_MARKER_PATH: &str = "/.start-command-recovery";
58
59/// Environment variable carrying the attempt number into the recovery command.
60pub const RECOVERY_ATTEMPT_ENV: &str = "START_COMMAND_RECOVERY_ATTEMPT";
61
62/// Exit code of a process killed with SIGKILL (the OOM killer's signal).
63pub const KILLED_EXIT_CODE: i32 = 137;
64
65/// POSIX selector run as the container command when a recovery command is set:
66/// `sh -c SELECTOR start-command <shell> <flag> <recovery> <main argv...>`.
67pub fn recovery_selector() -> String {
68    format!(
69        "s=$1; f=$2; r=$3; shift 3; if [ -e {marker} ]; then {env}=$(cat {marker} 2>/dev/null); export {env}; if [ -n \"$f\" ]; then exec \"$s\" \"$f\" -c \"$r\"; fi; exec \"$s\" -c \"$r\"; fi; exec \"$@\"",
70        marker = RECOVERY_MARKER_PATH,
71        env = RECOVERY_ATTEMPT_ENV
72    )
73}
74
75/// Wrap the container command so a later `docker start` can switch to the
76/// recovery command.
77pub fn build_recovery_selector_args(
78    main_args: &[String],
79    shell: &str,
80    shell_flag: Option<&str>,
81    recovery_command: &str,
82) -> Vec<String> {
83    let mut args = vec![
84        "sh".to_string(),
85        "-c".to_string(),
86        recovery_selector(),
87        "start-command".to_string(),
88        if shell.is_empty() { "sh" } else { shell }.to_string(),
89        shell_flag.unwrap_or("").to_string(),
90        recovery_command.to_string(),
91    ];
92    args.extend(main_args.iter().cloned());
93    args
94}
95
96/// Whether the inspected facts describe a killed main process.
97///
98/// Exit 137 (SIGKILL, the OOM killer's signal) always counts. `OOMKilled`
99/// alone does not: Docker sets it when *any* process in the container's
100/// cgroup was OOM-killed (a compiler, a test runner, a child `node`), and it
101/// stays set until the container is started again. A main process that
102/// survived that and then exited 0-127 on its own ran to completion, so the
103/// flag only counts when there is no usable exit status (the watcher's `-1`,
104/// or nothing at all) (issue #178).
105pub fn is_killed_exit(exit_code: &str, oom_killed: &str) -> bool {
106    let code = describe_exit_code_str(exit_code).code;
107    if code == Some(KILLED_EXIT_CODE) {
108        return true;
109    }
110    normalize_bool(oom_killed) == Some(true) && code.is_none_or(|code| code < 0)
111}
112
113/// Shell condition run by the completion watcher: true when the container was
114/// killed and the recovery entry point resumed it. Mirrors `is_killed_exit`:
115/// `OOMKilled` only counts without a non-negative exit code (issue #178).
116pub fn build_recovery_snippet(execution_id: &str) -> String {
117    let executable = std::env::current_exe()
118        .map(|path| path.to_string_lossy().to_string())
119        .unwrap_or_else(|_| "start".to_string());
120    format!(
121        "{{ [ \"${exit}\" = {code} ] || {{ [ \"${oom}\" = true ] && ! [ \"${exit}\" -ge 0 ] 2>/dev/null; }}; }} && {exe} {flag} {id} \"${exit}\" \"${oom}\" \"${started}\" \"${finished}\" \"${error}\" \"${cgroup}\" >/dev/null 2>&1",
122        exit = shell_vars::EXIT,
123        oom = shell_vars::OOM,
124        started = shell_vars::STARTED,
125        finished = shell_vars::FINISHED,
126        error = shell_vars::ERROR,
127        cgroup = cgroup_shell_vars::SAMPLE,
128        code = KILLED_EXIT_CODE,
129        exe = shell_quote(&executable),
130        flag = INTERNAL_RECOVER_FLAG,
131        id = shell_quote(execution_id),
132    )
133}
134
135/// What the `[Recovery k/N]` separator reports.
136pub struct RecoverySeparator<'a> {
137    pub attempt: u64,
138    pub max_attempts: u64,
139    pub exit_code: &'a str,
140    pub oom_killed: &'a str,
141    pub container_name: &'a str,
142    pub command: Option<&'a str>,
143    /// Random `--on-kill-resume-delay` wait before the resume (issue #181).
144    pub delay_ms: u64,
145}
146
147/// The `[Recovery k/N]` separator written into the session log.
148pub fn format_recovery_separator(params: &RecoverySeparator) -> String {
149    let described = describe_exit_code_str(params.exit_code);
150    let code = described
151        .code
152        .map(|code| code.to_string())
153        .unwrap_or_else(|| "unknown".to_string());
154    let signal = described
155        .signal
156        .map(|name| format!(", {}", name))
157        .unwrap_or_default();
158    let what = match params.command {
159        Some(command) => format!("running recovery command: {}", command),
160        None => "running the original command again".to_string(),
161    };
162    let delay = if params.delay_ms > 0 {
163        format!(" after a {} delay", format_recovery_delay(params.delay_ms))
164    } else {
165        String::new()
166    };
167    format!(
168        "\n[Recovery {}/{}] Main process was killed (exit {}{}, oomKilled={}); resuming container {}{}, {}\n",
169        params.attempt,
170        params.max_attempts,
171        code,
172        signal,
173        normalize_bool(params.oom_killed) == Some(true),
174        params.container_name,
175        delay,
176        what
177    )
178}
179
180/// The docker facts the watcher hands over.
181#[derive(Debug, Clone, Default)]
182pub struct RecoveryFacts {
183    pub exit_code: String,
184    pub oom_killed: String,
185    pub started_at: String,
186    pub finished_at: String,
187    pub container_error: String,
188    /// The watcher's last cgroup v2 sample (issue #182).
189    pub cgroup_memory: String,
190}
191
192/// Outcome of a recovery attempt; `recovered` is true only when the container
193/// was started again.
194#[derive(Debug, Clone, PartialEq, Eq)]
195pub struct RecoveryOutcome {
196    pub recovered: bool,
197    pub reason: String,
198    pub attempt: Option<u64>,
199    /// The `--on-kill-resume-delay` wait, when one is configured (issue #181).
200    pub delay_ms: Option<u64>,
201}
202
203impl RecoveryOutcome {
204    fn not(reason: &str) -> Self {
205        RecoveryOutcome {
206            recovered: false,
207            reason: reason.to_string(),
208            attempt: None,
209            delay_ms: None,
210        }
211    }
212}
213
214/// Where the `--on-kill-resume-delay` wait gets its randomness and sleeps;
215/// tests swap both out (issue #181).
216pub struct RecoveryDelayHooks<'a> {
217    pub random: &'a dyn Fn() -> f64,
218    pub sleep: &'a dyn Fn(u64),
219}
220
221impl Default for RecoveryDelayHooks<'_> {
222    fn default() -> Self {
223        RecoveryDelayHooks {
224            random: &system_random,
225            sleep: &system_sleep,
226        }
227    }
228}
229
230fn is_stop_requested(store: &ExecutionStore, execution_id: &str) -> bool {
231    store
232        .get(execution_id)
233        .is_some_and(|record| option_str(&record, "stopRequestedAt").is_some())
234}
235
236/// Starts the completion watcher that follows a resumed container.
237pub type WatcherStarter<'a> = &'a dyn Fn(&str, &ExecutionRecord, &DockerWatcherOptions);
238
239fn append_to_log(log_path: &str, text: &str) {
240    if log_path.is_empty() {
241        return;
242    }
243    // The log is best-effort here: recovery itself must still proceed.
244    if let Ok(mut file) = fs::OpenOptions::new()
245        .create(true)
246        .append(true)
247        .open(log_path)
248    {
249        let _ = file.write_all(text.as_bytes());
250    }
251}
252
253fn failure_detail(result: &CommandRunOutput) -> String {
254    let stderr = result.stderr.trim();
255    if !stderr.is_empty() {
256        return stderr.to_string();
257    }
258    result.error.clone().unwrap_or_else(|| {
259        format!(
260            "exit code {}",
261            result
262                .status
263                .map(|code| code.to_string())
264                .unwrap_or_else(|| "unknown".to_string())
265        )
266    })
267}
268
269fn write_recovery_marker<R: CommandRunner + ?Sized>(
270    container_name: &str,
271    attempt: u64,
272    runner: &R,
273) -> CommandRunOutput {
274    let dir = std::env::temp_dir().join(format!("start-recovery-{}", uuid::Uuid::new_v4()));
275    let file = dir.join("marker");
276    let written = fs::create_dir_all(&dir).and_then(|_| fs::write(&file, format!("{}\n", attempt)));
277    #[cfg(unix)]
278    {
279        use std::os::unix::fs::PermissionsExt;
280        let _ = fs::set_permissions(&file, fs::Permissions::from_mode(0o644));
281    }
282    let result = match written {
283        Ok(()) => runner.run(
284            &docker_command().to_string_lossy(),
285            &[
286                "cp".to_string(),
287                file.to_string_lossy().to_string(),
288                format!("{}:{}", container_name, RECOVERY_MARKER_PATH),
289            ],
290        ),
291        Err(err) => CommandRunOutput {
292            success: false,
293            error: Some(err.to_string()),
294            ..CommandRunOutput::default()
295        },
296    };
297    let _ = fs::remove_dir_all(&dir);
298    result
299}
300
301fn option_u64(record: &ExecutionRecord, key: &str) -> u64 {
302    match record.options.get(key) {
303        Some(Value::Number(n)) => n.as_u64().unwrap_or(0),
304        Some(Value::String(s)) => s.trim().parse().unwrap_or(0),
305        _ => 0,
306    }
307}
308
309fn option_str<'a>(record: &'a ExecutionRecord, key: &str) -> Option<&'a str> {
310    record
311        .options
312        .get(key)
313        .and_then(Value::as_str)
314        .filter(|value| !value.is_empty())
315}
316
317/// Resume a killed execution in its own container.
318pub fn recover_killed_execution<R: CommandRunner + ?Sized>(
319    store: &ExecutionStore,
320    execution_id: &str,
321    facts: &RecoveryFacts,
322    runner: &R,
323    start_watcher: WatcherStarter,
324) -> RecoveryOutcome {
325    recover_killed_execution_with_delay(
326        store,
327        execution_id,
328        facts,
329        runner,
330        start_watcher,
331        &RecoveryDelayHooks::default(),
332    )
333}
334
335/// [`recover_killed_execution`] after the random `--on-kill-resume-delay`
336/// wait (issue #181), with injectable randomness and sleep.
337pub fn recover_killed_execution_with_delay<R: CommandRunner + ?Sized>(
338    store: &ExecutionStore,
339    execution_id: &str,
340    facts: &RecoveryFacts,
341    runner: &R,
342    start_watcher: WatcherStarter,
343    delay: &RecoveryDelayHooks,
344) -> RecoveryOutcome {
345    if execution_id.is_empty() {
346        return RecoveryOutcome::not("missing-arguments");
347    }
348    let Some(mut record) = store.get(execution_id) else {
349        return RecoveryOutcome::not("record-not-found");
350    };
351    let max_attempts = option_u64(&record, "onKillResume");
352    let container_name = option_str(&record, "sessionName").map(str::to_string);
353    let Some(container_name) = container_name
354        .filter(|_| option_str(&record, "isolated") == Some("docker") && max_attempts >= 1)
355    else {
356        return RecoveryOutcome::not("not-configured");
357    };
358    if !is_killed_exit(&facts.exit_code, &facts.oom_killed) {
359        return RecoveryOutcome::not("not-killed");
360    }
361    let log_path = record.log_path.clone();
362    let log = |text: &str| append_to_log(&log_path, text);
363    if option_str(&record, "stopRequestedAt").is_some() {
364        log("\n[Recovery] Not resuming: the session was stopped on request.\n");
365        return RecoveryOutcome::not("stop-requested");
366    }
367    let used = option_u64(&record, "recoveryAttempts");
368    if used >= max_attempts {
369        log(&format!(
370            "\n[Recovery] Not resuming: all {} recovery attempt(s) used.\n",
371            max_attempts
372        ));
373        return RecoveryOutcome::not("attempts-exhausted");
374    }
375
376    let attempt = used + 1;
377    let delay_range = record_on_kill_resume_delay(record.options.get("onKillResumeDelay"));
378    let delay_ms = pick_recovery_delay_ms(delay_range.as_deref(), delay.random);
379    let give_up = |reason: String| {
380        log(&format!(
381            "[Recovery {}/{}] Failed: {}\n",
382            attempt, max_attempts, reason
383        ));
384        RecoveryOutcome {
385            recovered: false,
386            reason: "resume-failed".to_string(),
387            attempt: Some(attempt),
388            delay_ms: None,
389        }
390    };
391
392    let oom_killed = normalize_bool(&facts.oom_killed);
393    log(&format!(
394        "\n{}",
395        format_container_post_mortem(&ContainerPostMortem {
396            container_name: container_name.clone(),
397            exit_code: describe_exit_code_str(&facts.exit_code).code,
398            oom_killed,
399            started_at: Some(facts.started_at.clone()),
400            finished_at: Some(facts.finished_at.clone()),
401            error: normalize_container_error(&facts.container_error),
402        })
403    ));
404    if let Some(line) = format_cgroup_memory_log_line(&facts.cgroup_memory) {
405        log(&format!("{}\n", line));
406    }
407    let recovery_command = option_str(&record, "recoveryCommand").map(str::to_string);
408    log(&format_recovery_separator(&RecoverySeparator {
409        attempt,
410        max_attempts,
411        exit_code: &facts.exit_code,
412        oom_killed: &facts.oom_killed,
413        container_name: &container_name,
414        command: recovery_command.as_deref(),
415        delay_ms,
416    }));
417
418    // Executions killed by one host-wide OOM event must not all come back in
419    // the same second (issue #181). `--stop` during the wait cancels the
420    // resume: the container has already exited, so `docker stop` only leaves
421    // the `stopRequestedAt` marker this loop polls for.
422    if delay_ms > 0
423        && wait_for_recovery_delay(delay_ms, delay.sleep, &|| {
424            is_stop_requested(store, execution_id)
425        })
426    {
427        log(&format!(
428            "[Recovery {}/{}] Not resuming: the session was stopped on request during the delay.\n",
429            attempt, max_attempts
430        ));
431        return RecoveryOutcome {
432            recovered: false,
433            reason: "stop-requested".to_string(),
434            attempt: Some(attempt),
435            delay_ms: Some(delay_ms),
436        };
437    }
438
439    // Docker keeps `docker update` limits in the HostConfig across restarts;
440    // they are read back so the log and `--status` show what the resumed run
441    // is held to, and so a later relaunch can re-apply them.
442    let resource_limits = read_docker_resource_limits(&container_name, runner).unwrap_or_default();
443    if let Some(line) = build_resource_limits_status_line(&resource_limits) {
444        log(&format!("{}\n", line));
445    }
446
447    if recovery_command.is_some() {
448        let copied = write_recovery_marker(&container_name, attempt, runner);
449        if !copied.success {
450            return give_up(format!(
451                "could not mark the container: {}",
452                failure_detail(&copied)
453            ));
454        }
455    }
456
457    let since = Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true);
458    let started = runner.run(
459        &docker_command().to_string_lossy(),
460        &["start".to_string(), container_name.clone()],
461    );
462    if !started.success {
463        return give_up(format!("docker start failed: {}", failure_detail(&started)));
464    }
465
466    let described = describe_exit_code_str(&facts.exit_code);
467    let mut history = record
468        .options
469        .get("recoveryHistory")
470        .and_then(Value::as_array)
471        .cloned()
472        .unwrap_or_default();
473    let delayed = if delay_range.is_some() {
474        format!(", delayMs={}", delay_ms)
475    } else {
476        String::new()
477    };
478    // The killed run's own counters: the resumed run starts a fresh cgroup.
479    let counted = parse_cgroup_memory_sample(&facts.cgroup_memory)
480        .map(|memory| {
481            [
482                ("oomEvents", memory.oom_events),
483                ("oomKills", memory.oom_kills),
484            ]
485            .iter()
486            .filter_map(|(field, value)| value.map(|value| format!(", {}={}", field, value)))
487            .collect::<String>()
488        })
489        .unwrap_or_default();
490    history.push(json!(format!(
491        "{}: exit {}, oomKilled={}{}{}, resumed at {}",
492        attempt,
493        described
494            .code
495            .map(|code| code.to_string())
496            .unwrap_or_else(|| "unknown".to_string()),
497        oom_killed == Some(true),
498        counted,
499        delayed,
500        since
501    )));
502    let options = &mut record.options;
503    options.insert("recoveryAttempts".to_string(), json!(attempt));
504    options.insert("recoveryHistory".to_string(), Value::Array(history));
505    options.insert("lastRecoveryAt".to_string(), json!(since));
506    if delay_range.is_some() {
507        options.insert("lastRecoveryDelayMs".to_string(), json!(delay_ms));
508    }
509    if !resource_limits.is_empty() {
510        options.insert("resourceLimits".to_string(), json!(resource_limits));
511    }
512    record.status = ExecutionStatus::Executing;
513    record.exit_code = None;
514    record.end_time = None;
515    record.exit_reason = None;
516    record.oom_killed = None;
517    record.cgroup_memory = None;
518    // The container is already running again; even if the save fails, the new
519    // watcher still follows it and finalizes the record when it ends.
520    let _ = store.save(&record);
521
522    start_watcher(
523        &container_name,
524        &record,
525        &DockerWatcherOptions {
526            since: Some(since),
527            recover_on_kill: true,
528        },
529    );
530    RecoveryOutcome {
531        recovered: true,
532        reason: "resumed".to_string(),
533        attempt: Some(attempt),
534        delay_ms: delay_range.map(|_| delay_ms),
535    }
536}
537
538/// The real watcher: a detached shell following the resumed container.
539pub fn start_system_watcher(
540    container_name: &str,
541    record: &ExecutionRecord,
542    watcher: &DockerWatcherOptions,
543) {
544    let log_path = (!record.log_path.is_empty()).then(|| PathBuf::from(&record.log_path));
545    start_detached_docker_completion_watcher_with(
546        container_name,
547        get_docker_container_cleanup_policy(&build_launch_options(record)),
548        log_path.as_ref(),
549        Some(record.uuid.as_str()),
550        watcher,
551    );
552}
553
554/// Entry point for `--internal-recover-detached-docker <uuid> <exit> <oom>
555/// <started> <finished> <error> <cgroup>`. Returns 0 only when the container was
556/// resumed; any other outcome lets the watcher continue with its normal
557/// cleanup, footer and finalization.
558pub fn run_internal_recover(args: &[String]) -> i32 {
559    let execution_id = match args.first() {
560        Some(value) if !value.is_empty() => value.clone(),
561        _ => return 1,
562    };
563    if std::env::var("START_DISABLE_TRACKING").as_deref() == Ok("true") {
564        return 1;
565    }
566    let field = |index: usize| args.get(index).cloned().unwrap_or_default();
567    let store = ExecutionStore::with_options(ExecutionStoreOptions {
568        app_folder: std::env::var("START_APP_FOLDER").ok().map(PathBuf::from),
569        ..ExecutionStoreOptions::default()
570    });
571    let outcome = recover_killed_execution(
572        &store,
573        &execution_id,
574        &RecoveryFacts {
575            exit_code: field(1),
576            oom_killed: field(2),
577            started_at: field(3),
578            finished_at: field(4),
579            container_error: field(5),
580            cgroup_memory: field(6),
581        },
582        &SystemCommandRunner,
583        &start_system_watcher,
584    );
585    if outcome.recovered {
586        0
587    } else {
588        1
589    }
590}
start_command/execution_recovery.rs

start_command/
execution_recovery.rs
