Commit

Author:

Hash:

Timestamp:

+565 -217 +/-9 browse

Kevin Schoon [me@kevinschoon.com]

c1479195665cf8a82dcef6d8be25bac50dcd7a8b

Tue, 25 Aug 2026 18:10:49 +0000 (3 weeks ago)

improve ayllu-build gc, add better tests, refactor FakeRuntime
1diff --git a/ayllu-build/src/error.rs b/ayllu-build/src/error.rs
2index 7969b93..8cac31b 100644
3--- a/ayllu-build/src/error.rs
4+++ b/ayllu-build/src/error.rs
5 @@ -75,4 +75,6 @@ pub enum Error {
6 UnresolvedWorkflow(String),
7 #[error("Unresolved step: {0}")]
8 UnresolvedStep(String),
9+ #[error("Process did not shutdown after: {0:?}")]
10+ ShutdownTimeout(Duration),
11 }
12 diff --git a/ayllu-build/src/executor.rs b/ayllu-build/src/executor.rs
13index 94de111..7838c65 100644
14--- a/ayllu-build/src/executor.rs
15+++ b/ayllu-build/src/executor.rs
16 @@ -55,7 +55,7 @@ impl Executor {
17 })?;
18 }
19 }
20- let duration = handle.shutdown(timeout.add(Duration::from_secs(2)));
21+ let duration = handle.shutdown(timeout.add(Duration::from_secs(2)))?;
22 conn.workflow_finish(
23 *current_workflow_id,
24 state,
25 @@ -263,94 +263,13 @@ mod test {
26 config::Image,
27 error::Error,
28 executor::ExecutorBuilder,
29- runtime::{Handle, Runtime, Sha256Sum, Update},
30+ runtime::{
31+ Handle, Runtime, Sha256Sum, Update,
32+ test_utils::{Action, FakeRuntime},
33+ },
34 source::{Resolved, Source},
35 };
36
37- #[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
38- enum Action {
39- Step(i32, i32, i32),
40- Workflow(i32, i32),
41- }
42-
43- pub struct FakeHandle {}
44-
45- impl Handle for FakeHandle {
46- fn shutdown(&self, timeout: std::time::Duration) -> std::time::Duration {
47- Duration::from_secs(1)
48- }
49-
50- fn sample(&self, timestamp: u64) -> Result<crate::runtime::Sample, Error> {
51- todo!()
52- }
53- }
54-
55- #[derive(Default)]
56- pub struct FakeRuntime {
57- actions: RefCell<Vec<Action>>,
58- execution_failures: HashMap<i32, ()>,
59- }
60-
61- impl Runtime for FakeRuntime {
62- type Handle = FakeHandle;
63-
64- fn start(
65- &self,
66- manifest_id: i32,
67- workflow_id: i32,
68- ) -> Result<Self::Handle, crate::error::Error> {
69- self.actions
70- .borrow_mut()
71- .push(Action::Workflow(manifest_id, workflow_id));
72- Ok(self::FakeHandle {})
73- }
74-
75- fn execute<F>(
76- &self,
77- manifest_id: i32,
78- workflow_id: i32,
79- step_id: i32,
80- handle: &Self::Handle,
81- on_update: F,
82- ) -> Result<(i32, Duration), Error>
83- where
84- F: FnMut(&Update) -> Result<(), Error>,
85- {
86- self.actions
87- .borrow_mut()
88- .push(Action::Step(manifest_id, workflow_id, step_id));
89- if self.execution_failures.contains_key(&step_id) {
90- return Ok((1, Duration::from_secs(1)));
91- }
92- Ok((0, Duration::from_secs(1)))
93- }
94-
95- fn find_output(
96- &self,
97- manifest_id: i32,
98- workflow_id: i32,
99- path: &std::path::Path,
100- ) -> Result<(Sha256Sum, std::fs::File), Error> {
101- todo!()
102- }
103-
104- fn cleanup(&self, manifest_id: i32, workflow_id: i32) -> Result<(), Error> {
105- todo!()
106- }
107-
108- fn shell(&self, manifest_id: i32, workflow_id: i32) -> Result<i32, Error> {
109- todo!()
110- }
111-
112- fn initalize(
113- &mut self,
114- resolved: &Resolved,
115- graph: &crate::graph::Graph,
116- ) -> Result<(), Error> {
117- Ok(())
118- }
119- }
120-
121 const TEST_MANIFEST: &str = r#"
122 {
123 "workflows":
124 diff --git a/ayllu-build/src/main.rs b/ayllu-build/src/main.rs
125index 03e15c0..f77c6b0 100644
126--- a/ayllu-build/src/main.rs
127+++ b/ayllu-build/src/main.rs
128 @@ -1,5 +1,3 @@
129- use std::time::Duration;
130-
131 use ayllu_database::{events::Kind, subscriber::Name};
132 use tabled::builder::Builder;
133 use tracing::Level;
134 @@ -135,7 +133,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
135 }
136 let oci_runtime = rt_builder.build().unwrap();
137 let mut builder = Builder::default();
138- builder.push_record(["id", "pid", "state", "path"]);
139+ builder.push_record(["id", "pid", "running", "path"]);
140 oci_runtime.list()?.into_iter().for_each(|state| {
141 builder.push_record([
142 state.id,
143 @@ -143,8 +141,8 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
144 .pid
145 .map(|pid| pid.to_string())
146 .unwrap_or("-".to_string()),
147+ state.running.to_string(),
148 state.path.to_string_lossy().to_string(),
149- state.status.to_string(),
150 ]);
151 });
152 let table = builder.build();
153 @@ -158,11 +156,12 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
154 } => {
155 let oci_runtime = rt_builder.build().unwrap();
156 let gc = GcBuilder::default()
157- .timeout(timeout.map(Duration::from_secs))
158- .config(cfg.clone())
159+ .timeout_secs(timeout.unwrap_or(5))
160+ .work_dir(cfg.build.work_dir.clone())
161+ .db_path(cfg.common.database.path.clone())
162 .build()
163 .unwrap();
164- gc.run(&oci_runtime, force, workspace)?;
165+ println!("{}", gc.run(&oci_runtime, force, workspace)?);
166 Ok(())
167 }
168 Commands::Debug {
169 diff --git a/ayllu-build/src/runtime/mod.rs b/ayllu-build/src/runtime/mod.rs
170index 271304f..c8d7a7d 100644
171--- a/ayllu-build/src/runtime/mod.rs
172+++ b/ayllu-build/src/runtime/mod.rs
173 @@ -1,9 +1,38 @@
174- use std::{fs::File, path::Path, time::Duration};
175+ use std::{
176+ fs::File,
177+ path::{Path, PathBuf},
178+ time::Duration,
179+ };
180
181 use crate::{error::Error, source::Resolved};
182
183 pub mod oci;
184
185+ /// Describes the state of a workflow per the runtime
186+ #[derive(Clone, Default, Debug)]
187+ pub struct WorkflowState {
188+ pub id: String,
189+ pub pid: Option<i32>,
190+ pub path: PathBuf,
191+ pub running: bool,
192+ }
193+
194+ impl WorkflowState {
195+ pub fn ids(&self) -> (i32, i32) {
196+ let mut split = self.id.split("-");
197+ let _ = split.next().unwrap();
198+ let manifest_id = split
199+ .next()
200+ .map(|part| part.parse::<i32>().unwrap())
201+ .unwrap();
202+ let workflow_id = split
203+ .next()
204+ .map(|part| part.parse::<i32>().unwrap())
205+ .unwrap();
206+ (manifest_id, workflow_id)
207+ }
208+ }
209+
210 pub struct Sha256Sum(String);
211
212 impl std::fmt::Display for Sha256Sum {
213 @@ -47,7 +76,7 @@ pub enum Update<'a> {
214
215 pub trait Handle {
216 fn sample(&self, timestamp: u64) -> Result<Sample, Error>;
217- fn shutdown(&self, timeout: Duration) -> Duration;
218+ fn shutdown(&self, timeout: Duration) -> Result<Duration, Error>;
219 }
220
221 pub trait Runtime: Sized {
222 @@ -74,4 +103,112 @@ pub trait Runtime: Sized {
223 ) -> Result<(Sha256Sum, File), Error>;
224 /// Run garbage collection
225 fn cleanup(&self, manifest_id: i32, workflow_id: i32) -> Result<(), Error>;
226+ /// List all running processes
227+ fn list(&self) -> Result<Vec<WorkflowState>, Error>;
228+ /// Kill a particular workflow
229+ fn kill(&self, manifest_id: i32, workflow_id: i32) -> Result<(), Error>;
230+ }
231+
232+ #[cfg(test)]
233+ pub(crate) mod test_utils {
234+ use std::{cell::RefCell, collections::HashMap, time::Duration};
235+
236+ use crate::{
237+ error::Error,
238+ runtime::{Handle, Runtime, Sha256Sum, Update, WorkflowState},
239+ source::Resolved,
240+ };
241+
242+ #[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
243+ pub enum Action {
244+ Step(i32, i32, i32),
245+ Workflow(i32, i32),
246+ }
247+
248+ pub struct FakeHandle {}
249+
250+ impl Handle for FakeHandle {
251+ fn shutdown(&self, _timeout: std::time::Duration) -> Result<std::time::Duration, Error> {
252+ Ok(Duration::from_secs(1))
253+ }
254+
255+ fn sample(&self, _timestamp: u64) -> Result<crate::runtime::Sample, Error> {
256+ todo!()
257+ }
258+ }
259+
260+ #[derive(Default)]
261+ pub struct FakeRuntime {
262+ pub actions: RefCell<Vec<Action>>,
263+ pub execution_failures: HashMap<i32, ()>,
264+ pub running: Vec<WorkflowState>,
265+ }
266+
267+ impl Runtime for FakeRuntime {
268+ type Handle = FakeHandle;
269+
270+ fn start(
271+ &self,
272+ manifest_id: i32,
273+ workflow_id: i32,
274+ ) -> Result<Self::Handle, crate::error::Error> {
275+ self.actions
276+ .borrow_mut()
277+ .push(Action::Workflow(manifest_id, workflow_id));
278+ Ok(self::FakeHandle {})
279+ }
280+
281+ fn execute<F>(
282+ &self,
283+ manifest_id: i32,
284+ workflow_id: i32,
285+ step_id: i32,
286+ _handle: &Self::Handle,
287+ _on_update: F,
288+ ) -> Result<(i32, Duration), Error>
289+ where
290+ F: FnMut(&Update) -> Result<(), Error>,
291+ {
292+ self.actions
293+ .borrow_mut()
294+ .push(Action::Step(manifest_id, workflow_id, step_id));
295+ if self.execution_failures.contains_key(&step_id) {
296+ return Ok((1, Duration::from_secs(1)));
297+ }
298+ Ok((0, Duration::from_secs(1)))
299+ }
300+
301+ fn find_output(
302+ &self,
303+ _manifest_id: i32,
304+ _workflow_id: i32,
305+ _path: &std::path::Path,
306+ ) -> Result<(Sha256Sum, std::fs::File), Error> {
307+ todo!()
308+ }
309+
310+ fn cleanup(&self, _manifest_id: i32, _workflow_id: i32) -> Result<(), Error> {
311+ Ok(())
312+ }
313+
314+ fn shell(&self, _manifest_id: i32, _workflow_id: i32) -> Result<i32, Error> {
315+ todo!()
316+ }
317+
318+ fn initalize(
319+ &mut self,
320+ _resolved: &Resolved,
321+ _graph: &crate::graph::Graph,
322+ ) -> Result<(), Error> {
323+ Ok(())
324+ }
325+
326+ fn list(&self) -> Result<Vec<WorkflowState>, Error> {
327+ Ok(self.running.clone())
328+ }
329+
330+ fn kill(&self, _manifest_id: i32, _workflow_id: i32) -> Result<(), Error> {
331+ Ok(())
332+ }
333+ }
334 }
335 diff --git a/ayllu-build/src/runtime/oci/gc.rs b/ayllu-build/src/runtime/oci/gc.rs
336index b92349d..634c4ef 100644
337--- a/ayllu-build/src/runtime/oci/gc.rs
338+++ b/ayllu-build/src/runtime/oci/gc.rs
339 @@ -1,61 +1,95 @@
340- use std::{ops::Add, time::Duration};
341+ use std::{ops::Add, path::PathBuf, time::Duration};
342
343- use ayllu_database::{Wrapper as Database, build::State};
344+ use ayllu_database::{
345+ Wrapper as Database,
346+ build::{State, manifests::Manifest},
347+ };
348 use derive_builder::Builder;
349
350 use crate::{
351- config::Config,
352 error::Error,
353 runtime::{
354- Handle, Runtime,
355- oci::{
356- OciHandle, OciRuntime, Workspace, runtime::WorkflowState, workspace::scan_workspaces,
357- },
358+ Handle, Runtime, WorkflowState,
359+ oci::{OciHandle, Workspace, workspace::scan_workspaces},
360 },
361 };
362
363+ #[derive(Clone, Copy, Default, PartialEq, Eq)]
364+ pub struct GcOps {
365+ pub n_stopped_processes: usize,
366+ pub n_deleted_workspaces: usize,
367+ pub n_cancelled_manifests: usize,
368+ }
369+
370+ impl std::fmt::Display for GcOps {
371+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
372+ write!(
373+ f,
374+ "Manifests cancelled: {}, Processes stopped: {}, Workspaces deleted: {}",
375+ self.n_cancelled_manifests, self.n_stopped_processes, self.n_deleted_workspaces
376+ )
377+ }
378+ }
379+
380 #[derive(Builder, Clone)]
381 pub struct Gc {
382- pub config: Config,
383- #[builder(default)]
384- pub timeout: Option<Duration>,
385+ pub db_path: PathBuf,
386+ pub work_dir: PathBuf,
387+ pub timeout_secs: u64,
388 }
389
390 impl Gc {
391- pub fn run(&self, oci_runtime: &OciRuntime, force: bool, workspace: bool) -> Result<(), Error> {
392- let mut db = Database::new(&self.config.database.path)?;
393+ pub fn run(
394+ &self,
395+ oci_runtime: &impl Runtime,
396+ force: bool,
397+ workspace: bool,
398+ ) -> Result<GcOps, Error> {
399+ let mut ops = GcOps::default();
400+ let mut db = Database::new(&self.db_path)?;
401 let mut conn = db.call();
402 let running: Vec<WorkflowState> = oci_runtime
403 .list()?
404 .into_iter()
405- .filter(|state| matches!(state.status, oci_spec::runtime::ContainerState::Running))
406+ .filter(|state| state.running)
407 .collect();
408
409- conn.manifest_list(None, None, None, None)?
410+ tracing::info!("Runtime reports {} running workflows", running.len());
411+
412+ tracing::info!("Checking for manifests which report running but are not");
413+ // TODO: Filter on state in db layer
414+ let manifests = conn.manifest_list(None, None, None, None)?;
415+ let db_running: Vec<&Manifest> = manifests
416 .iter()
417- .try_for_each(|manifest| {
418- if running
419- .iter()
420- .find(|state| state.ids().0 == manifest.id)
421- .is_none()
422- && manifest.state.eq(&State::Running)
423- {
424- tracing::info!(
425- "Cancelling {} since it reports that it is running but is not",
426- manifest.id
427- );
428- conn.manifest_cancel(manifest.id)
429- } else {
430- Ok(())
431- }
432- })?;
433+ .filter(|manifest| manifest.state.eq(&State::Running))
434+ .collect();
435+ db_running.iter().try_for_each(|manifest| {
436+ if running
437+ .iter()
438+ .find(|state| state.ids().0 == manifest.id)
439+ .is_none()
440+ {
441+ tracing::info!(
442+ "Cancelling {} since it reports that it is running but is not",
443+ manifest.id
444+ );
445+ conn.manifest_cancel(manifest.id)?;
446+ ops.n_cancelled_manifests += 1;
447+ Ok::<_, Error>(())
448+ } else {
449+ Ok(())
450+ }
451+ })?;
452
453- tracing::info!("OCI Runtime reports {} running containers", running.len());
454 running.iter().try_for_each(|workflow_state| {
455 tracing::info!(
456 "Found container: {}: {}",
457 workflow_state.id,
458- workflow_state.status
459+ if workflow_state.running {
460+ "Running"
461+ } else {
462+ "?"
463+ }
464 );
465 let (manifest_id, workflow_id) = workflow_state.ids();
466 if let Some(state) = conn.manifest_check(manifest_id)?
467 @@ -68,44 +102,57 @@ impl Gc {
468 return Ok(());
469 };
470
471- let workspace = Workspace::new(&self.config.build.work_dir, manifest_id, workflow_id);
472-
473- tracing::info!("Killing: {manifest_id}: {workflow_id}");
474+ let workspace = Workspace::new(&self.work_dir, manifest_id, workflow_id);
475+ tracing::info!("Killing: Manifest={manifest_id},Workflow={workflow_id}");
476 if let Some(pid) = workflow_state.pid
477 && pid != 0
478+ && workspace.exists()
479 {
480+ tracing::info!("Attempting to shutdown container via workspace");
481 let handle = OciHandle::attach(
482 pid,
483 &workspace.child_pid_path(),
484- self.timeout
485- .unwrap_or(Duration::from_secs(self.config.build.interrupt_timeout)),
486+ Duration::from_secs(self.timeout_secs),
487 )?;
488- handle.shutdown(
489- Duration::from_secs(self.config.build.interrupt_timeout)
490- .add(Duration::from_secs(2)),
491+ let duration = handle
492+ .shutdown(Duration::from_secs(self.timeout_secs).add(Duration::from_secs(2)))?;
493+ ops.n_stopped_processes += 1;
494+ tracing::info!("Process was killed after: {duration:?}");
495+ } else {
496+ tracing::info!(
497+ "OCI runtime suggests container is running but it does not have a workspace, killing via OCI runtime"
498 );
499+ oci_runtime.kill(manifest_id, workflow_id)?;
500+ ops.n_stopped_processes += 1;
501 }
502 if let Err(err) = oci_runtime.cleanup(manifest_id, workflow_id) {
503 tracing::warn!("Failed to delete runtime; {err:?}");
504 }
505- workspace.unmount()?;
506- conn.manifest_cancel(manifest_id)?;
507+ match conn.manifest_cancel(manifest_id) {
508+ Ok(_) => {
509+ tracing::info!("Cancelled hung manifest {manifest_id}");
510+ ops.n_cancelled_manifests += 1
511+ }
512+ Err(e) => {
513+ tracing::info!("Failed to cancel container: {e}");
514+ }
515+ }
516 Ok::<_, crate::error::Error>(())
517 })?;
518
519 let running: Vec<WorkflowState> = oci_runtime
520 .list()?
521 .into_iter()
522- .filter(|state| matches!(state.status, oci_spec::runtime::ContainerState::Running))
523+ .filter(|state| state.running)
524 .collect();
525
526 tracing::info!(
527- "OCI runtime now reports {} running containers",
528+ "OCI runtime now reports {} running workflows",
529 running.len()
530 );
531
532 if workspace {
533- for workspace in scan_workspaces(&self.config.build.work_dir)? {
534+ for workspace in scan_workspaces(&self.work_dir)? {
535 let manifest_id = workspace.id().manifest_id;
536 let manifest_still_active = running
537 .iter()
538 @@ -124,9 +171,180 @@ impl Gc {
539 tracing::error!(
540 "Workspace destruction failed, maybe some lingering mount?: {err:?}"
541 )
542- }
543+ };
544+ ops.n_deleted_workspaces += 1;
545 }
546 }
547- Ok(())
548+ Ok(ops)
549+ }
550+ }
551+
552+ #[cfg(test)]
553+ mod test {
554+ use std::{collections::HashMap, path::Path};
555+
556+ use crate::{
557+ config::Image,
558+ graph::{Allocator, Graph},
559+ manifest::{Manifest, Step, Workflow},
560+ runtime::{
561+ WorkflowState,
562+ oci::{
563+ GcBuilder,
564+ gc::{Gc, GcOps},
565+ template::Template,
566+ workspace::RootBuilder,
567+ },
568+ test_utils::FakeRuntime,
569+ },
570+ };
571+
572+ fn setup(test_dir: &Path, workflows: &[Workflow]) -> Gc {
573+ let db_path = test_dir.join("ayllu.sqlite");
574+ let manifest = Manifest {
575+ workflows: workflows.to_vec(),
576+ };
577+ let source = crate::source::Resolved {
578+ manifest,
579+ ..Default::default()
580+ };
581+ let images =
582+ HashMap::<String, Image>::from_iter(vec![(String::from("noop"), Image::default())]);
583+ ayllu_database::migrate(&db_path).unwrap();
584+ let mut db = ayllu_database::Wrapper::new(&db_path).unwrap();
585+ let _ = RootBuilder::default()
586+ .root(test_dir.to_path_buf())
587+ .images(images)
588+ .graph(Graph::new(&source, &mut Allocator::Database(db.call())).unwrap())
589+ .template(Template::default())
590+ .source(source.clone())
591+ .build()
592+ .unwrap()
593+ .initialize()
594+ .unwrap();
595+ GcBuilder::default()
596+ .db_path(db_path.clone())
597+ .timeout_secs(1)
598+ .work_dir(test_dir.to_path_buf())
599+ .build()
600+ .unwrap()
601+ }
602+
603+ #[test]
604+ fn gc_running_no_workspace_or_db_entry() {
605+ let test_dir = tempfile::tempdir().unwrap();
606+ let gc = setup(
607+ test_dir.path(),
608+ &[Workflow {
609+ name: "WF1".to_string(),
610+ image: "noop".to_string(),
611+ steps: vec![Step {
612+ name: "S1".to_string(),
613+ ..Default::default()
614+ }],
615+ ..Default::default()
616+ }],
617+ );
618+ let mut db = ayllu_database::Wrapper::new(&test_dir.path().join("ayllu.sqlite")).unwrap();
619+ db.with(|mut tx| tx.manifest_cancel(1)).unwrap();
620+ let mut proc = std::process::Command::new("/bin/sh")
621+ .args(["sleep", "60"])
622+ .spawn()
623+ .unwrap();
624+ let stats = gc
625+ .run(
626+ &FakeRuntime {
627+ running: vec![WorkflowState {
628+ id: String::from("build-1-1"),
629+ pid: Some(proc.id() as i32),
630+ running: true,
631+ ..Default::default()
632+ }],
633+ ..Default::default()
634+ },
635+ true,
636+ false,
637+ )
638+ .unwrap();
639+ proc.wait().unwrap();
640+ println!("{stats}");
641+ assert!(stats.eq(&GcOps {
642+ n_stopped_processes: 1,
643+ n_deleted_workspaces: 0,
644+ n_cancelled_manifests: 1
645+ }));
646+ }
647+
648+ #[test]
649+ fn gc_cancel_only() {
650+ let test_dir = tempfile::tempdir().unwrap();
651+ let gc = setup(
652+ test_dir.path(),
653+ &[Workflow {
654+ name: "WF1".to_string(),
655+ image: "noop".to_string(),
656+ steps: vec![Step {
657+ name: "S1".to_string(),
658+ ..Default::default()
659+ }],
660+ ..Default::default()
661+ }],
662+ );
663+ let mut db = ayllu_database::Wrapper::new(&test_dir.path().join("ayllu.sqlite")).unwrap();
664+ db.with(|mut tx| tx.manifest_start(1)).unwrap();
665+ let stats = gc.run(&FakeRuntime::default(), false, false).unwrap();
666+ assert!(stats.eq(&GcOps {
667+ n_stopped_processes: 0,
668+ n_deleted_workspaces: 0,
669+ n_cancelled_manifests: 1
670+ }));
671+ }
672+
673+ #[test]
674+ fn gc_cancel_delete_workspace() {
675+ let test_dir = tempfile::tempdir().unwrap();
676+ let gc = setup(
677+ test_dir.path(),
678+ &[Workflow {
679+ name: "WF1".to_string(),
680+ image: "noop".to_string(),
681+ steps: vec![Step {
682+ name: "S1".to_string(),
683+ ..Default::default()
684+ }],
685+ ..Default::default()
686+ }],
687+ );
688+ let mut db = ayllu_database::Wrapper::new(&test_dir.path().join("ayllu.sqlite")).unwrap();
689+ db.with(|mut tx| tx.manifest_start(1)).unwrap();
690+ let stats = gc.run(&FakeRuntime::default(), false, true).unwrap();
691+ assert!(stats.eq(&GcOps {
692+ n_stopped_processes: 0,
693+ n_deleted_workspaces: 1,
694+ n_cancelled_manifests: 1
695+ }));
696+ }
697+
698+ #[test]
699+ fn gc_workspace_bare_no_proc() {
700+ let test_dir = tempfile::tempdir().unwrap();
701+ let gc = setup(
702+ test_dir.path(),
703+ &[Workflow {
704+ name: "WF1".to_string(),
705+ image: "noop".to_string(),
706+ steps: vec![Step {
707+ name: "S1".to_string(),
708+ ..Default::default()
709+ }],
710+ ..Default::default()
711+ }],
712+ );
713+ let stats = gc.run(&FakeRuntime::default(), false, true).unwrap();
714+ assert!(stats.eq(&GcOps {
715+ n_stopped_processes: 0,
716+ n_deleted_workspaces: 1,
717+ n_cancelled_manifests: 0
718+ }));
719 }
720 }
721 diff --git a/ayllu-build/src/runtime/oci/handle.rs b/ayllu-build/src/runtime/oci/handle.rs
722index 1366255..64e88fb 100644
723--- a/ayllu-build/src/runtime/oci/handle.rs
724+++ b/ayllu-build/src/runtime/oci/handle.rs
725 @@ -60,18 +60,24 @@ pub struct OciHandle {
726 }
727
728 impl crate::runtime::Handle for OciHandle {
729- fn shutdown(&self, timeout: Duration) -> Duration {
730+ fn shutdown(&self, timeout: Duration) -> Result<Duration, Error> {
731 if !self.thread.is_finished() {
732 let (tx, rx) = channel::<()>();
733 self.tx.send(tx).unwrap();
734 match rx.recv_timeout(timeout) {
735- Ok(..) => tracing::info!("Process successfully shutdown"),
736- Err(err) => tracing::error!("Timeout: {err}"),
737+ Ok(..) => {
738+ tracing::info!("Process successfully shutdown");
739+ Ok(self.start.elapsed())
740+ }
741+ Err(err) => {
742+ tracing::error!("Timeout: {err}");
743+ Err(Error::ShutdownTimeout(self.start.elapsed()))
744+ }
745 }
746 } else {
747- tracing::warn!("Exec thread already shutdown, nothing to do")
748- };
749- self.start.elapsed()
750+ tracing::warn!("Exec thread already shutdown, nothing to do");
751+ Ok(Duration::default())
752+ }
753 }
754
755 fn sample(&self, timestamp: u64) -> Result<crate::runtime::Sample, Error> {
756 @@ -105,9 +111,9 @@ impl OciHandle {
757 sender.send(()).unwrap();
758 break;
759 }
760- let now = std::time::SystemTime::now();
761+ let now = std::time::Instant::now();
762 tracing::info!("Process {parent_pid} will be reaped in: {timeout:?}");
763- while now.elapsed().unwrap() < timeout {
764+ while now.elapsed() < timeout {
765 if let Err(err) = signal::kill(child_pid, None) {
766 tracing::info!("Process {child_pid} was shutdown gracefully: {err}");
767 sender.send(()).unwrap();
768 @@ -149,7 +155,7 @@ impl OciHandle {
769 if path.exists()
770 && let Ok(pid_string) = std::fs::read_to_string(path)
771 && !pid_string.is_empty()
772- && let Ok(pid_raw) = pid_string.parse::<i32>()
773+ && let Ok(pid_raw) = pid_string.trim_end().parse::<i32>()
774 {
775 return Ok(Pid::from_raw(pid_raw));
776 }
777 @@ -183,11 +189,11 @@ impl OciHandle {
778 pub fn new(
779 container_ready_file: &Path,
780 child_pid_path: &Path,
781- child: Child,
782+ managment_proc: Child,
783 timeout: Duration,
784 ) -> Result<Self, Error> {
785 let child_pid = OciHandle::get_child_pid(child_pid_path, timeout)?;
786- let parent_pid = Pid::from_raw(child.id() as i32);
787+ let parent_pid = Pid::from_raw(managment_proc.id() as i32);
788 let (tx, rx) = channel::<Sender<()>>();
789 let (ready_tx, ready_rx) = channel::<()>();
790 let thread = std::thread::spawn(move || {
791 @@ -220,37 +226,121 @@ impl OciHandle {
792 #[cfg(test)]
793 mod test {
794
795+ use crate::runtime::Handle;
796 use std::process::Command;
797
798- use crate::runtime::Handle;
799+ const FAKE_INIT_MANAGER: &str = r#"
800+ cleanup() {
801+ [ -n "$child_pid" ] && kill "$child_pid" 2>/dev/null
802+ }
803+
804+ on_signal() {
805+ cleanup
806+ wait "$child_pid" 2>/dev/null
807+ exit 1
808+ }
809+
810+ trap on_signal INT TERM HUP
811+
812+ (
813+ echo 1 > $$READY_PATH$$
814+ for i in $(seq 0 12); do
815+ sleep .1;
816+ done
817+ exit 1
818+ ) &
819+
820+ child_pid=$!
821+
822+ echo $child_pid > $$CHILD_PID_PATH$$
823+
824+ while kill -0 "$child_pid" 2>/dev/null; do
825+ sleep .2
826+ done
827+ wait "$child_pid"
828+ "#;
829
830 use super::*;
831 #[test]
832- fn process() {
833+ fn handle_kill_early() {
834 let tmp = tempfile::tempdir().unwrap();
835- let ready_file = tmp.path().join("ready.txt");
836- let ready_file_str = ready_file.to_str().unwrap();
837+ let ready_path = tmp.path().join("ready.txt");
838+ let pid_path = tmp.path().join("init.pid");
839 let mut cmd = Command::new("/bin/sh");
840 cmd.args([
841 "-c",
842- format!("for i in seq 0 5; do sleep .1; done && echo ready > {ready_file_str}")
843- .as_str(),
844+ &FAKE_INIT_MANAGER
845+ .replace("$$CHILD_PID_PATH$$", &pid_path.to_string_lossy())
846+ .replace("$$READY_PATH$$", &ready_path.to_string_lossy()),
847 ]);
848
849- let child = cmd.spawn().unwrap();
850- let pid = child.id();
851+ let init_manager = cmd.spawn().unwrap();
852+ let handle = OciHandle::new(
853+ ready_path.as_path(),
854+ &pid_path,
855+ init_manager,
856+ Duration::from_secs(DEFAULT_INTERRUPT_TIMEOUT_SECS),
857+ )
858+ .unwrap();
859+ assert!(pid_path.exists());
860+ assert!(ready_path.exists());
861+ handle.shutdown(Duration::from_secs(2)).unwrap();
862+ }
863+
864+ #[test]
865+ fn handle_shutdown() {
866+ let tmp = tempfile::tempdir().unwrap();
867+ let ready_path = tmp.path().join("ready.txt");
868 let pid_path = tmp.path().join("init.pid");
869- std::fs::write(&pid_path, pid.to_string()).unwrap();
870+ let mut cmd = Command::new("/bin/sh");
871+ cmd.args([
872+ "-c",
873+ &FAKE_INIT_MANAGER
874+ .replace("$$CHILD_PID_PATH$$", &pid_path.to_string_lossy())
875+ .replace("$$READY_PATH$$", &ready_path.to_string_lossy()),
876+ ]);
877
878+ let init_manager = cmd.spawn().unwrap();
879 let handle = OciHandle::new(
880- ready_file.as_path(),
881+ ready_path.as_path(),
882 &pid_path,
883- child,
884+ init_manager,
885 Duration::from_secs(DEFAULT_INTERRUPT_TIMEOUT_SECS),
886 )
887 .unwrap();
888- std::thread::sleep(Duration::from_millis(750));
889- assert!(ready_file.exists());
890- handle.shutdown(Duration::from_secs(2));
891+ // thread runs for ~1200ms
892+ assert!(pid_path.exists());
893+ assert!(ready_path.exists());
894+ handle.shutdown(Duration::from_secs(2)).unwrap();
895+ }
896+
897+ #[test]
898+ fn handle_manager_fails_to_set_pid() {
899+ let tmp = tempfile::tempdir().unwrap();
900+ let ready_path = tmp.path().join("ready.txt");
901+ let pid_path = tmp.path().join("init.pid");
902+ let mut cmd = Command::new("/bin/sh");
903+ cmd.args([
904+ "-c",
905+ &FAKE_INIT_MANAGER
906+ .replace(
907+ "$$CHILD_PID_PATH$$",
908+ &tmp.path().join("wrong-path.txt").to_string_lossy(),
909+ )
910+ .replace("$$READY_PATH$$", &ready_path.to_string_lossy()),
911+ ]);
912+
913+ let init_manager = cmd.spawn().unwrap();
914+ assert!(
915+ OciHandle::new(
916+ ready_path.as_path(),
917+ &pid_path,
918+ init_manager,
919+ Duration::from_millis(800),
920+ )
921+ .err()
922+ .is_some_and(|e| matches!(e, Error::ContainerTimeout(_)))
923+ );
924+ assert!(ready_path.exists());
925 }
926 }
927 diff --git a/ayllu-build/src/runtime/oci/runtime.rs b/ayllu-build/src/runtime/oci/runtime.rs
928index 4594998..7a75252 100644
929--- a/ayllu-build/src/runtime/oci/runtime.rs
930+++ b/ayllu-build/src/runtime/oci/runtime.rs
931 @@ -16,7 +16,7 @@ use crate::{
932 error::Error,
933 graph::Graph,
934 runtime::{
935- Handle, Sha256Sum, Stream, Update,
936+ Handle, Sha256Sum, Stream, Update, WorkflowState,
937 oci::{
938 DEFAULT_INTERRUPT_TIMEOUT_SECS,
939 handle::{DEFAULT_SAMPLE_RATE, OciHandle},
940 @@ -48,29 +48,6 @@ fn log_stream<R: Read>(on_line: Sender<String>, buf: BufReader<R>) -> Result<(),
941 // && std::fs::read_to_string(comm_path).is_ok_and(|content| content.starts_with("systemd"))
942 // }
943
944- pub struct WorkflowState {
945- pub id: String,
946- pub pid: Option<i32>,
947- pub path: PathBuf,
948- pub status: ContainerState,
949- }
950-
951- impl WorkflowState {
952- pub fn ids(&self) -> (i32, i32) {
953- let mut split = self.id.split("-");
954- let _ = split.next().unwrap();
955- let manifest_id = split
956- .next()
957- .map(|part| part.parse::<i32>().unwrap())
958- .unwrap();
959- let workflow_id = split
960- .next()
961- .map(|part| part.parse::<i32>().unwrap())
962- .unwrap();
963- (manifest_id, workflow_id)
964- }
965- }
966-
967 #[derive(Builder)]
968 pub struct OciRuntime {
969 runtime_dir: PathBuf,
970 @@ -280,11 +257,13 @@ impl crate::runtime::Runtime for OciRuntime {
971 fn cleanup(&self, manifest_id: i32, workflow_id: i32) -> Result<(), Error> {
972 let workspace = Workspace::new(&self.runtime_dir, manifest_id, workflow_id);
973 let container_name = &self.container_name(manifest_id, workflow_id);
974- if self.runtime_dir.join(container_name).exists() {
975- let _ = self.check_output(&["delete", container_name.as_str()])?;
976+ let _ = self.check_output(&["delete", container_name.as_str()])?;
977+ if let Err(err) = workspace.unmount() {
978+ tracing::info!("Tried to unmount workspace overlayfs but failed: {err}");
979+ }
980+ if workspace.exists() {
981+ workspace.destroy()?;
982 }
983- workspace.unmount()?;
984- workspace.destroy()?;
985 Ok(())
986 }
987
988 @@ -318,6 +297,25 @@ impl crate::runtime::Runtime for OciRuntime {
989 })?;
990 Ok(status.code().unwrap_or_default())
991 }
992+
993+ fn kill(&self, manifest_id: i32, workflow_id: i32) -> Result<(), Error> {
994+ let container_name = self.container_name(manifest_id, workflow_id);
995+ let _ = self.check_output(&["kill", container_name.as_str(), "9"])?;
996+ Ok(())
997+ }
998+
999+ fn list(&self) -> Result<Vec<WorkflowState>, Error> {
1000+ Ok(self
1001+ .decode_output::<Vec<oci_spec::runtime::State>>(&["list", "--format=json"])?
1002+ .iter()
1003+ .map(|state| WorkflowState {
1004+ id: state.id().clone(),
1005+ pid: *state.pid(),
1006+ path: state.bundle().to_path_buf(),
1007+ running: matches!(state.status(), ContainerState::Running),
1008+ })
1009+ .collect())
1010+ }
1011 }
1012
1013 impl OciRuntime {
1014 @@ -396,24 +394,4 @@ impl OciRuntime {
1015 let output = self.check_output(others)?;
1016 Ok(serde_json::from_slice::<T>(output.stdout.as_slice())?)
1017 }
1018-
1019- #[allow(dead_code)]
1020- fn kill(&self, manifest_id: i32, workflow_id: i32) -> Result<(), Error> {
1021- let container_name = self.container_name(manifest_id, workflow_id);
1022- let _ = self.check_output(&["kill", container_name.as_str(), "KILL"])?;
1023- Ok(())
1024- }
1025-
1026- pub fn list(&self) -> Result<Vec<WorkflowState>, Error> {
1027- Ok(self
1028- .decode_output::<Vec<oci_spec::runtime::State>>(&["list", "--format=json"])?
1029- .iter()
1030- .map(|state| WorkflowState {
1031- id: state.id().clone(),
1032- pid: *state.pid(),
1033- path: state.bundle().to_path_buf(),
1034- status: *state.status(),
1035- })
1036- .collect())
1037- }
1038 }
1039 diff --git a/ayllu-build/src/runtime/oci/template.rs b/ayllu-build/src/runtime/oci/template.rs
1040index 8751f6d..7310f08 100644
1041--- a/ayllu-build/src/runtime/oci/template.rs
1042+++ b/ayllu-build/src/runtime/oci/template.rs
1043 @@ -42,7 +42,7 @@ impl Default for Step {
1044 /// Initializes config.json and process specs for OCI containers.
1045 /// Template params are system settings while workflow and step
1046 /// data structures allow users to influence the build.
1047- #[derive(Builder, Clone)]
1048+ #[derive(Builder, Clone, Default)]
1049 pub struct Template {
1050 init_script: String,
1051 default_branch: String,
1052 diff --git a/ayllu-build/src/runtime/oci/workspace.rs b/ayllu-build/src/runtime/oci/workspace.rs
1053index c883cfd..9b6be62 100644
1054--- a/ayllu-build/src/runtime/oci/workspace.rs
1055+++ b/ayllu-build/src/runtime/oci/workspace.rs
1056 @@ -314,11 +314,16 @@ impl Workspace {
1057 OpenOptions::new().read(true).open(&source_file)?,
1058 ))
1059 }
1060+
1061+ /// Determine if the workspace actually exists on disk
1062+ pub fn exists(&self) -> bool {
1063+ self.rootfs().exists() && self.child_pid_path().exists() && self.config().exists()
1064+ }
1065 }
1066
1067 pub fn scan_workspaces(root: &Path) -> Result<Vec<Workspace>, Error> {
1068 tracing::info!("Scanning for workspaces in {root:?}");
1069- let files = std::fs::read_dir(root.join("cache"))?;
1070+ let files = std::fs::read_dir(root.join("builds"))?;
1071 files
1072 .into_iter()
1073 .try_fold(Vec::default(), |mut workspaces, file| {