start_command/
execution_resume_flow.rs

1//! Apply a durable resume plan without holding the database lock over Docker.
2use super::*;
3
4/// Resume a tracked execution by UUID or session name.
5pub fn resume_execution(
6    store: Option<&ExecutionStore>,
7    identifier: &str,
8    command: Option<&str>,
9    output_format: Option<&str>,
10) -> ExecutionResumeResult {
11    resume_execution_with(
12        store,
13        identifier,
14        command,
15        output_format,
16        &SystemCommandRunner,
17        &SystemResumeHooks,
18    )
19}
20
21/// Resume with an injectable command runner and hooks, so tests never touch
22/// docker.
23pub fn resume_execution_with<R: CommandRunner, H: ResumeHooks>(
24    store: Option<&ExecutionStore>,
25    identifier: &str,
26    command: Option<&str>,
27    output_format: Option<&str>,
28    runner: &R,
29    hooks: &H,
30) -> ExecutionResumeResult {
31    resume_execution_with_resources(
32        store,
33        identifier,
34        command,
35        output_format,
36        runner,
37        hooks,
38        &json!({}),
39    )
40}
41
42pub fn resume_execution_with_options(
43    store: Option<&ExecutionStore>,
44    identifier: &str,
45    command: Option<&str>,
46    output_format: Option<&str>,
47    options: &crate::args_parser::WrapperOptions,
48) -> ExecutionResumeResult {
49    resume_execution_with_resources(
50        store,
51        identifier,
52        command,
53        output_format,
54        &SystemCommandRunner,
55        &SystemResumeHooks,
56        &crate::docker_resource_options::specs(options),
57    )
58}
59
60#[allow(clippy::too_many_arguments)]
61pub fn resume_execution_with_resources<R: CommandRunner, H: ResumeHooks>(
62    store: Option<&ExecutionStore>,
63    identifier: &str,
64    command: Option<&str>,
65    output_format: Option<&str>,
66    runner: &R,
67    hooks: &H,
68    overrides: &Value,
69) -> ExecutionResumeResult {
70    let Some(store) = store else {
71        return ExecutionResumeResult {
72            success: false,
73            output: None,
74            error: Some("Execution tracking is disabled.".to_string()),
75        };
76    };
77
78    let Some(mut record) = store.get(identifier) else {
79        return ExecutionResumeResult {
80            success: false,
81            output: None,
82            error: Some(format!(
83                "No execution found with UUID or session name: {}",
84                identifier
85            )),
86        };
87    };
88
89    if crate::launch_owner::has_active_launch(&record) {
90        return ExecutionResumeResult {
91            success: false,
92            output: None,
93            error: Some("An execution launch is already reserved.".into()),
94        };
95    }
96    let previous = record.clone();
97    let probe = probe_session(&record, runner);
98    let resources = crate::resume_resources::prepare(&record, &probe, overrides, runner);
99    let (limits, resolved, old_memory) = match resources {
100        Ok(result) => result,
101        Err(error) => {
102            return ExecutionResumeResult {
103                success: false,
104                output: None,
105                error: Some(error),
106            }
107        }
108    };
109    let mut planning_record = record.clone();
110    if let Some(limits) = &limits {
111        planning_record
112            .options
113            .insert("resourceLimits".into(), json!(limits));
114    }
115    let mut plan =
116        match build_resume_plan_with_limits(&planning_record, command, &probe, limits.clone()) {
117            Ok(plan) => plan,
118            Err(error) => {
119                return ExecutionResumeResult {
120                    success: false,
121                    output: None,
122                    error: Some(error),
123                }
124            }
125        };
126    if let Some(limits) = limits {
127        plan.resource_limits = limits.clone();
128        let changed = ["memory", "memorySwap", "cpus"]
129            .iter()
130            .any(|k| overrides.get(k).and_then(Value::as_str).is_some())
131            || record.options.contains_key("cpuPenaltyConfig");
132        if plan.mode == ResumeMode::DockerStart && changed {
133            let mut args = vec!["update".into()];
134            args.extend(limits.into_iter().filter(|s| {
135                [
136                    "--memory=",
137                    "--memory-swap=",
138                    "--cpus=",
139                    "--cpu-quota=",
140                    "--cpu-period=",
141                ]
142                .iter()
143                .any(|f| s.starts_with(f))
144            }));
145            args.push(plan.session_name.clone());
146            plan.steps.insert(
147                0,
148                ResumeStep {
149                    command: docker_command().to_string_lossy().into(),
150                    args,
151                    description: "Apply resume limits before start".into(),
152                },
153            );
154        }
155    }
156    let mut container_id: Option<String> = None;
157    let mut attempt = crate::execution_attempt::create_attempt(
158        &record,
159        plan.mode.as_str(),
160        active_session_name(&plan),
161    );
162    let mut lifecycle_record = record.clone();
163    lifecycle_record.attempt = Some(attempt.clone());
164    crate::execution_attempt::append_lifecycle(&lifecycle_record, "resume-started", json!({}));
165    apply_resume_to_record(&mut record, &plan, None);
166    record.attempt = Some(attempt.clone());
167    if let Some(resolved) = resolved {
168        record.options.insert("resolvedLimits".into(), resolved);
169    }
170    let mut requested = record
171        .options
172        .get("resourceLimitSpecs")
173        .filter(|value| value.is_object())
174        .cloned()
175        .unwrap_or_else(|| json!({}));
176    for k in ["memory", "memorySwap", "cpus"] {
177        if let Some(value) = overrides.get(k).filter(|v| v.is_string()) {
178            requested[k] = value.clone();
179        }
180    }
181    record
182        .options
183        .insert("resourceLimitSpecs".into(), requested);
184    if record.options.contains_key("cpuPenaltyConfig")
185        && overrides.get("cpus").and_then(Value::as_str).is_some()
186    {
187        record
188            .options
189            .insert("baseResourceLimits".into(), json!(&plan.resource_limits));
190    }
191    crate::launch_owner::mark_launch(&mut record);
192    if let Err(error) = store.reserve_launch(&record, &previous) {
193        return ExecutionResumeResult {
194            success: false,
195            output: None,
196            error: Some(format!("Launch reservation failed: {}", error)),
197        };
198    }
199
200    if let Some(spec) = overrides.get("memory").and_then(Value::as_str) {
201        let new = record
202            .options
203            .get("resolvedLimits")
204            .and_then(|v| v.get("memory"))
205            .cloned()
206            .unwrap_or(Value::Null);
207        if !record.log_path.is_empty() {
208            append_log_file(
209                &PathBuf::from(&record.log_path),
210                &format!(
211                    "Resume memory limit: {} -> {} ({})\n",
212                    old_memory, new, spec
213                ),
214            );
215        }
216    }
217
218    if plan.mode == ResumeMode::Relaunch {
219        let launch_options = plan.launch_options.clone().unwrap_or_default();
220        let launch_result = hooks.relaunch(&plan.backend, &plan.command, &launch_options);
221        if !launch_result.success {
222            let _ = store.save(&previous);
223            crate::execution_attempt::append_lifecycle(
224                &lifecycle_record,
225                "launch-failed",
226                json!({ "error": launch_result.message }),
227            );
228            return ExecutionResumeResult {
229                success: false,
230                output: None,
231                error: Some(format!(
232                    "Failed to relaunch {} session \"{}\": {}",
233                    plan.backend, plan.session_name, launch_result.message
234                )),
235            };
236        }
237        container_id = launch_result.container_id;
238    } else {
239        for step in &plan.steps {
240            let result = runner.run(&step.command, &step.args);
241            if !result.success {
242                let _ = store.save(&previous);
243                let detail = if !result.stderr.trim().is_empty() {
244                    result.stderr.trim().to_string()
245                } else {
246                    result.error.clone().unwrap_or_else(|| {
247                        format!(
248                            "exit code {}",
249                            result
250                                .status
251                                .map(|code| code.to_string())
252                                .unwrap_or_else(|| "unknown".to_string())
253                        )
254                    })
255                };
256                crate::execution_attempt::append_lifecycle(
257                    &lifecycle_record,
258                    "launch-failed",
259                    json!({ "error": detail }),
260                );
261                return ExecutionResumeResult {
262                    success: false,
263                    output: None,
264                    error: Some(format!(
265                        "Failed to resume {} session \"{}\": {}",
266                        plan.backend, plan.session_name, detail
267                    )),
268                };
269            }
270            let stdout = result.stdout.trim().to_string();
271            if !stdout.is_empty() {
272                container_id = Some(stdout);
273            }
274        }
275
276        if let Some(line) = build_resource_limits_status_line(&plan.resource_limits) {
277            if !record.log_path.is_empty() {
278                append_log_file(&PathBuf::from(&record.log_path), &format!("{}\n", line));
279            }
280        }
281    }
282
283    let previous_session_name = plan
284        .new_session_name
285        .as_ref()
286        .map(|_| plan.session_name.clone());
287    if let Some(id) = container_id.as_deref() {
288        record.options.insert("containerId".into(), json!(id));
289    }
290    record.options.insert("launchPending".into(), json!(false));
291    attempt.launch_accepted_at = Some(Utc::now().to_rfc3339());
292    record
293        .options
294        .insert("resumedAt".into(), json!(attempt.started_at));
295    record.attempt = Some(attempt.clone());
296    if let Err(error) = store.save(&record) {
297        if plan.backend == "docker" {
298            let _ = hooks.attach_watcher(active_session_name(&plan), &record);
299        }
300        let stopped = plan.backend == "docker"
301            && runner
302                .run(
303                    &docker_command().to_string_lossy(),
304                    &["stop".into(), active_session_name(&plan).into()],
305                )
306                .success;
307        if stopped {
308            record.status = ExecutionStatus::Executed;
309            record.exit_code = Some(-1);
310            record.end_time = Some(Utc::now().to_rfc3339());
311            record.options.insert("launchPending".into(), json!(false));
312            let _ = store.save(&record);
313        }
314        return ExecutionResumeResult {
315            success: false,
316            output: None,
317            error: Some(
318                json!({"code": "LAUNCH_PERSISTENCE_FAILED", "uuid": record.uuid,
319                "containerName": active_session_name(&plan), "running": !stopped,
320                "error": error})
321                .to_string(),
322            ),
323        };
324    }
325    crate::execution_attempt::append_lifecycle(&record, "launch-accepted", json!({}));
326    if plan.backend == "docker" {
327        let attachment = hooks.attach_watcher(active_session_name(&plan), &record);
328        let fields = match &attachment {
329            Ok(()) => json!({"watcherAttachedAt": Utc::now().to_rfc3339()}),
330            Err(error) => json!({"watcherError": error}),
331        };
332        match store.patch_attempt(&record.uuid, attempt.number, fields) {
333            Ok(Some(current)) => crate::execution_attempt::append_lifecycle(
334                &current,
335                if attachment.is_ok() {
336                    "watcher-attached"
337                } else {
338                    "watcher-attachment-failed"
339                },
340                json!({"error": attachment.as_ref().err()}),
341            ),
342            Ok(None) => {}
343            Err(error) => {
344                return ExecutionResumeResult {
345                    success: false,
346                    output: None,
347                    error: Some(error),
348                }
349            }
350        }
351        if let Err(error) = attachment {
352            return ExecutionResumeResult {
353                success: false,
354                output: None,
355                error: Some(format!(
356                    "Launch accepted, but completion watcher attachment failed: {}",
357                    error
358                )),
359            };
360        }
361    }
362
363    let session_name = record_option(&record, "sessionName")
364        .unwrap_or(&plan.session_name)
365        .to_string();
366
367    ExecutionResumeResult {
368        success: true,
369        output: Some(format_resume_result(
370            &ResumeResultFields {
371                identifier,
372                uuid: &record.uuid,
373                mode: plan.mode,
374                backend: &plan.backend,
375                session_name: &session_name,
376                previous_session_name: previous_session_name.as_deref(),
377                snapshot_image: plan.snapshot_image.as_deref(),
378                resource_limits: &plan.resource_limits,
379                command: &plan.command,
380                message: &plan.message,
381            },
382            output_format,
383        )),
384        error: None,
385    }
386}
387
388#[cfg(test)]
389#[path = "execution_resume_cases.rs"]
390mod tests;
start_command/execution_resume_flow.rs

start_command/
execution_resume_flow.rs
