/build/source/nativelink-sim/src/actions.rs

Source
// Copyright 2026 Trace Machina, Inc. All rights reserved.
//
// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    See LICENSE file for details
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Simulated execution. An action is a duration per phase and an outcome
//! per attempt; the worker around it is the production `LocalWorker`.

use core::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use core::time::Duration;
use std::collections::HashMap;
use std::sync::Arc;

use nativelink_error::{Code, Error, make_err};
use nativelink_proto::com::github::trace_machina::nativelink::remote_execution::{
    ActionResourceUsage, StartExecute,
};
use nativelink_util::action_messages::{ActionResult, OperationId};
use nativelink_util::common::DigestInfo;
use nativelink_util::digest_hasher::DigestHasherFunc;
use nativelink_worker::running_actions_manager::{Metrics, RunningAction, RunningActionsManager};
use parking_lot::Mutex;
use tokio::sync::Notify;

use crate::trace::Event;

/// What one attempt at an action does.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[allow(variant_size_differences, reason = "a few bytes")]
pub enum Run {
    /// Exits 0.
    Succeed,
    /// Exits with this code.
    Exit(i32),
    /// The worker fails it with this code, as a lost input or a crash
    /// in the runner would.
    Error(Code),
    /// Never finishes until killed.
    Hang,
    /// The execution task panics.
    Panic,
}

/// One action of a scenario. Its digest comes from `seed`, so two clients
/// that name the same seed ask for the same action.
#[derive(Clone, Debug)]
pub struct Action {
    pub(crate) seed: u32,
    pub(crate) priority: i32,
    pub(crate) cacheable: bool,
    pub(crate) prepare: Duration,
    pub(crate) run: Duration,
    pub(crate) upload: Duration,
    /// One per attempt; the last repeats.
    pub(crate) attempts: Vec<Run>,
    pub(crate) properties: Vec<(String, String)>,
    pub(crate) timeout: Option<Duration>,
}

impl Action {
    /// A cacheable action that runs for one second and succeeds.
    pub fn new(seed: u32) -> Self {
        Self {
            seed,
            priority: 0,
            cacheable: true,
            prepare: Duration::ZERO,
            run: Duration::from_secs(1),
            upload: Duration::ZERO,
            attempts: vec![Run::Succeed],
            properties: Vec::new(),
            timeout: None,
        }
    }

    #[must_use]
    pub const fn priority(mut self, priority: i32) -> Self {
        self.priority = priority;
        self
    }

    /// Sent with `skip_cache_lookup`, so it never coalesces with another.
    #[must_use]
    pub const fn uncacheable(mut self) -> Self {
        self.cacheable = false;
        self
    }

    #[must_use]
    pub const fn run_ms(mut self, ms: u64) -> Self {
        self.run = Duration::from_millis(ms);
        self
    }

    #[must_use]
    pub const fn prepare_ms(mut self, ms: u64) -> Self {
        self.prepare = Duration::from_millis(ms);
        self
    }

    #[must_use]
    pub const fn upload_ms(mut self, ms: u64) -> Self {
        self.upload = Duration::from_millis(ms);
        self
    }

    /// What each attempt does, in order; the last repeats.
    #[must_use]
    pub fn attempts(mut self, attempts: &[Run]) -> Self {
        assert!(!attempts.is_empty(), "an action needs at least one attempt");
        self.attempts = attempts.to_vec();
        self
    }

    #[must_use]
    pub fn property(mut self, name: &str, value: &str) -> Self {
        self.properties.push((name.to_string(), value.to_string()));
        self
    }

    /// `Action.timeout`.
    #[must_use]
    pub const fn timeout_s(mut self, seconds: u64) -> Self {
        self.timeout = Some(Duration::from_secs(seconds));
        self
    }

    pub fn label(&self) -> String {
        format!("a{}", self.seed)
    }

    pub const fn seed(&self) -> u32 {
        self.seed
    }
}

#[derive(Debug)]
pub(crate) struct Entry {
    pub(crate) action: Action,
    pub(crate) attempts: AtomicU32,
}

/// Every action a scenario submitted, by its digest, so the worker side can
/// find what a `StartExecute` should do.
#[derive(Debug, Default)]
pub struct Registry {
    entries: Mutex<HashMap<DigestInfo, Arc<Entry>>>,
}

impl Registry {
    pub(crate) fn insert(&self, digest: DigestInfo, action: Action) {
        self.entries.lock().entry(digest).or_insert_with(|| {15.9k
            Arc::new(Entry {
                action,
                attempts: AtomicU32::new(0),
            })
        });
    }

    pub(crate) fn label(&self, digest: &DigestInfo) -> Option<String> {
        self.entries.lock().get(digest).map(|e| e.action.label())
    }

    pub(crate) fn lookup(&self, digest: &DigestInfo) -> Option<Arc<Entry>> {
        self.entries.lock().get(digest).cloned()
    }
}

pub(crate) fn start_execute_digest(start_execute: &StartExecute) -> Option<DigestInfo> {
    start_execute
        .execute_request
        .as_ref()?0
        .action_digest
        .as_ref()
        .and_then(|digest| DigestInfo::try_from(digest).ok())
}

/// What one attempt at an action does on the worker that got it.
#[derive(Clone, Debug)]
pub struct Plan {
    /// The action's name in the trace.
    pub action: String,
    /// A stable name for the scheduler's operation.
    pub key: String,
    pub prepare: Duration,
    pub run_for: Duration,
    pub upload: Duration,
    pub run: Run,
}

/// Where a simulated worker learns what an action does, and where it
/// reports what it did: the in-process scenario (`Net`), or a seeded
/// distribution in a worker of its own process (the fleet simulator).
pub trait ActionSource: Send + Sync + core::fmt::Debug + 'static {
    fn plan(&self, start_execute: &StartExecute) -> Result<Plan, Error>;

    fn record(&self, event: Event);

    /// The scheduler asked for a kill. The in-process network already
    /// records it.
    fn killed(&self, _operation_id: &OperationId) {}
}

/// The `RunningActionsManager` of one simulated worker.
#[derive(Debug)]
pub struct SimActionsManager {
    worker: String,
    source: Arc<dyn ActionSource>,
    running: Arc<Mutex<HashMap<OperationId, Arc<KillSwitch>>>>,
    metrics: Arc<Metrics>,
}

impl SimActionsManager {
    pub fn new(worker: &str, source: Arc<dyn ActionSource>) -> Self {
        Self {
            worker: worker.to_string(),
            source,
            running: Arc::new(Mutex::new(HashMap::new())),
            metrics: Arc::new(Metrics::default()),
        }
    }

    /// Executions this worker has not finished.
    pub fn running(&self) -> usize {
        self.running.lock().len()
    }
}

#[derive(Debug, Default)]
struct KillSwitch {
    killed: AtomicBool,
    notify: Notify,
}

impl KillSwitch {
    fn kill(&self) {
        self.killed.store(true, Ordering::Release);
        self.notify.notify_waiters();
    }

    async fn wait(&self) {
        loop {
            let notified = self.notify.notified();
            if self.killed.load(Ordering::Acquire) {
                return;
            }
            notified.await;
        }
    }
}

impl RunningActionsManager for SimActionsManager {
    type RunningAction = SimRunningAction;

    async fn create_and_add_action(
        self: &Arc<Self>,
        _worker_id: String,
        start_execute: StartExecute,
    ) -> Result<Arc<SimRunningAction>, Error> {
        let plan = self.source.plan(&start_execute)?0;
        let operation_id = OperationId::from(start_execute.operation_id);
        let kill = Arc::new(KillSwitch::default());
        self.running
            .lock()
            .insert(operation_id.clone(), kill.clone());
        Ok(Arc::new(SimRunningAction {
            worker: self.worker.clone(),
            action: plan.action,
            key: plan.key,
            operation_id,
            prepare: plan.prepare,
            run_for: plan.run_for,
            upload: plan.upload,
            run: plan.run,
            kill,
            running: self.running.clone(),
            source: self.source.clone(),
            work_directory: String::new(),
            started: AtomicBool::new(false),
            ended: AtomicBool::new(false),
        }))
    }

    async fn cache_action_result(
        &self,
        _action_digest: DigestInfo,
        _action_result: &mut ActionResult,
        _hasher: DigestHasherFunc,
    ) -> Result<(), Error> {
        Ok(())
    }

    async fn kill_all(&self) {
        self.source.record(Event::KillAll {
            worker: self.worker.clone(),
        });
        let switches: Vec<_> = self.running.lock().values().cloned().collect();
        for switch277 in switches {
            switch.kill();
        }
    }

    async fn kill_operation(&self, operation_id: &OperationId) -> Result<(), Error> {
        self.source.killed(operation_id);
        let switch = self.running.lock().get(operation_id).cloned();
        if let Some(switch0) = switch {
            switch.kill();
        }
        Ok(())
    }

    fn metrics(&self) -> &Arc<Metrics> {
        &self.metrics
    }
}

#[derive(Debug)]
pub struct SimRunningAction {
    worker: String,
    action: String,
    key: String,
    operation_id: OperationId,
    prepare: Duration,
    run_for: Duration,
    upload: Duration,
    run: Run,
    kill: Arc<KillSwitch>,
    running: Arc<Mutex<HashMap<OperationId, Arc<KillSwitch>>>>,
    source: Arc<dyn ActionSource>,
    work_directory: String,
    started: AtomicBool,
    ended: AtomicBool,
}

/// An execution whose task was aborted (the worker dropped it on a
/// disconnect) still ends in the trace and leaves the running set.
impl Drop for SimRunningAction {
    fn drop(&mut self) {
        if self.started.load(Ordering::Acquire) && !10.6kself.ended10.6k.load(Ordering::Acquire) {
            self.end(true);
        }
        self.running.lock().remove(&self.operation_id);
    }
}

impl SimRunningAction {
    /// Sleeps `duration` unless killed first; `Err` when killed.
    async fn phase(&self, duration: Duration) -> Result<(), Error> {
        if duration.is_zero() {
            return Ok(());
        }
        tokio::select! {
            () = tokio::time::sleep(duration) => Ok(())10.4k,
            () = self.kill.wait() => Err(make_err!(Code::Aborted, "Killed"))0,
        }
    }

    fn end(&self, killed: bool) {
        if self.ended.swap(true, Ordering::AcqRel) {
            return;
        }
        self.source.record(Event::ExecEnd {
            worker: self.worker.clone(),
            action: self.action.clone(),
            key: self.key.clone(),
            killed,
        });
    }
}

impl RunningAction for SimRunningAction {
    fn get_operation_id(&self) -> &OperationId {
        &self.operation_id
    }

    async fn prepare_action(self: Arc<Self>) -> Result<Arc<Self>, Error> {
        self.phase(self.prepare).await?0;
        Ok(self)
    }

    async fn execute(self: Arc<Self>) -> Result<Arc<Self>, Error> {
        self.started.store(true, Ordering::Release);
        self.source.record(Event::ExecStart {
            worker: self.worker.clone(),
            action: self.action.clone(),
            key: self.key.clone(),
        });
        let ran10.3k = match self.run {
            Run::Hang => {
                self.kill.wait().await;
                Err(make_err!(Code::Aborted, "Killed"))
            }
            _ => self.phase(self.run_for).await,
        };
        if let Err(err0) = ran {
            self.end(true);
            return Err(err);
        }
        self.end(false);
        match self.run {
            Run::Panic => panic!("simulated action {} panicked", self.action),
            Run::Error(code) => Err(make_err!(code, "Simulated failure of {}", self.action)),
            Run::Succeed | Run::Exit(_) | Run::Hang => Ok(self),
        }
    }

    async fn upload_results(self: Arc<Self>) -> Result<Arc<Self>, Error> {
        self.phase(self.upload).await?0;
        Ok(self)
    }

    async fn cleanup(self: Arc<Self>) -> Result<Arc<Self>, Error> {
        self.running.lock().remove(&self.operation_id);
        Ok(self)
    }

    async fn get_finished_result(self: Arc<Self>) -> Result<ActionResult, Error> {
        let exit_code = match self.run {
            Run::Exit(code) => code,
            _ => 0,
        };
        Ok(ActionResult {
            exit_code,
            ..ActionResult::default()
        })
    }

    fn resource_usage(&self) -> Option<ActionResourceUsage> {
        None
    }

    fn get_work_directory(&self) -> &String {
        &self.work_directory
    }
}

Coverage Report

Created: 2026-11-13 23:29

Line	Count	Source
1		// Copyright 2026 Trace Machina, Inc. All rights reserved.
2		//
3		// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License");
4		// you may not use this file except in compliance with the License.
5		// You may obtain a copy of the License at
6		//
7		// See LICENSE file for details
8		//
9		// Unless required by applicable law or agreed to in writing, software
10		// distributed under the License is distributed on an "AS IS" BASIS,
11		// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12		// See the License for the specific language governing permissions and
13		// limitations under the License.
14
15		//! Simulated execution. An action is a duration per phase and an outcome
16		//! per attempt; the worker around it is the production `LocalWorker`.
17
18		use core::sync::atomic::{AtomicBool, AtomicU32, Ordering};
19		use core::time::Duration;
20		use std::collections::HashMap;
21		use std::sync::Arc;
22
23		use nativelink_error::{Code, Error, make_err};
24		use nativelink_proto::com::github::trace_machina::nativelink::remote_execution::{
25		ActionResourceUsage, StartExecute,
26		};
27		use nativelink_util::action_messages::{ActionResult, OperationId};
28		use nativelink_util::common::DigestInfo;
29		use nativelink_util::digest_hasher::DigestHasherFunc;
30		use nativelink_worker::running_actions_manager::{Metrics, RunningAction, RunningActionsManager};
31		use parking_lot::Mutex;
32		use tokio::sync::Notify;
33
34		use crate::trace::Event;
35
36		/// What one attempt at an action does.
37		#[derive(Clone, Copy, Debug, PartialEq, Eq)]
38		#[allow(variant_size_differences, reason = "a few bytes")]
39		pub enum Run {
40		/// Exits 0.
41		Succeed,
42		/// Exits with this code.
43		Exit(i32),
44		/// The worker fails it with this code, as a lost input or a crash
45		/// in the runner would.
46		Error(Code),
47		/// Never finishes until killed.
48		Hang,
49		/// The execution task panics.
50		Panic,
51		}
52
53		/// One action of a scenario. Its digest comes from `seed`, so two clients
54		/// that name the same seed ask for the same action.
55		#[derive(Clone, Debug)]
56		pub struct Action {
57		pub(crate) seed: u32,
58		pub(crate) priority: i32,
59		pub(crate) cacheable: bool,
60		pub(crate) prepare: Duration,
61		pub(crate) run: Duration,
62		pub(crate) upload: Duration,
63		/// One per attempt; the last repeats.
64		pub(crate) attempts: Vec<Run>,
65		pub(crate) properties: Vec<(String, String)>,
66		pub(crate) timeout: Option<Duration>,
67		}
68
69		impl Action {
70		/// A cacheable action that runs for one second and succeeds.
71	11.3k	pub fn new(seed: u32) -> Self {
72	11.3k	Self {
73	11.3k	seed,
74	11.3k	priority: 0,
75	11.3k	cacheable: true,
76	11.3k	prepare: Duration::ZERO,
77	11.3k	run: Duration::from_secs(1),
78	11.3k	upload: Duration::ZERO,
79	11.3k	attempts: vec![Run::Succeed],
80	11.3k	properties: Vec::new(),
81	11.3k	timeout: None,
82	11.3k	}
83	11.3k	}
84
85		#[must_use]
86	1.74k	pub const fn priority(mut self, priority: i32) -> Self {
87	1.74k	self.priority = priority;
88	1.74k	self
89	1.74k	}
90
91		/// Sent with `skip_cache_lookup`, so it never coalesces with another.
92		#[must_use]
93	338	pub const fn uncacheable(mut self) -> Self {
94	338	self.cacheable = false;
95	338	self
96	338	}
97
98		#[must_use]
99	9.30k	pub const fn run_ms(mut self, ms: u64) -> Self {
100	9.30k	self.run = Duration::from_millis(ms);
101	9.30k	self
102	9.30k	}
103
104		#[must_use]
105	0	pub const fn prepare_ms(mut self, ms: u64) -> Self {
106	0	self.prepare = Duration::from_millis(ms);
107	0	self
108	0	}
109
110		#[must_use]
111	0	pub const fn upload_ms(mut self, ms: u64) -> Self {
112	0	self.upload = Duration::from_millis(ms);
113	0	self
114	0	}
115
116		/// What each attempt does, in order; the last repeats.
117		#[must_use]
118	262	pub fn attempts(mut self, attempts: &[Run]) -> Self {
119	262	assert!(!attempts.is_empty(), "an action needs at least one attempt");
120	262	self.attempts = attempts.to_vec();
121	262	self
122	262	}
123
124		#[must_use]
125	1	pub fn property(mut self, name: &str, value: &str) -> Self {
126	1	self.properties.push((name.to_string(), value.to_string()));
127	1	self
128	1	}
129
130		/// `Action.timeout`.
131		#[must_use]
132	0	pub const fn timeout_s(mut self, seconds: u64) -> Self {
133	0	self.timeout = Some(Duration::from_secs(seconds));
134	0	self
135	0	}
136
137	53.3k	pub fn label(&self) -> String {
138	53.3k	format!("a{}", self.seed)
139	53.3k	}
140
141	0	pub const fn seed(&self) -> u32 {
142	0	self.seed
143	0	}
144		}
145
146		#[derive(Debug)]
147		pub(crate) struct Entry {
148		pub(crate) action: Action,
149		pub(crate) attempts: AtomicU32,
150		}
151
152		/// Every action a scenario submitted, by its digest, so the worker side can
153		/// find what a `StartExecute` should do.
154		#[derive(Debug, Default)]
155		pub struct Registry {
156		entries: Mutex<HashMap<DigestInfo, Arc<Entry>>>,
157		}
158
159		impl Registry {
160	15.9k	pub(crate) fn insert(&self, digest: DigestInfo, action: Action) {
161	15.9k	self.entries.lock().entry(digest).or_insert_with(\|\| {15.9k
162	15.9k	Arc::new(Entry {
163	15.9k	action,
164	15.9k	attempts: AtomicU32::new(0),
165	15.9k	})
166	15.9k	});
167	15.9k	}
168
169	10.6k	pub(crate) fn label(&self, digest: &DigestInfo) -> Option<String> {
170	10.6k	self.entries.lock().get(digest).map(\|e\| e.action.label())
171	10.6k	}
172
173	10.6k	pub(crate) fn lookup(&self, digest: &DigestInfo) -> Option<Arc<Entry>> {
174	10.6k	self.entries.lock().get(digest).cloned()
175	10.6k	}
176		}
177
178	21.3k	pub(crate) fn start_execute_digest(start_execute: &StartExecute) -> Option<DigestInfo> {
179	21.3k	start_execute
180	21.3k	.execute_request
181	21.3k	.as_ref() ?0
182		.action_digest
183	21.3k	.as_ref()
184	21.3k	.and_then(\|digest\| DigestInfo::try_from(digest).ok())
185	21.3k	}
186
187		/// What one attempt at an action does on the worker that got it.
188		#[derive(Clone, Debug)]
189		pub struct Plan {
190		/// The action's name in the trace.
191		pub action: String,
192		/// A stable name for the scheduler's operation.
193		pub key: String,
194		pub prepare: Duration,
195		pub run_for: Duration,
196		pub upload: Duration,
197		pub run: Run,
198		}
199
200		/// Where a simulated worker learns what an action does, and where it
201		/// reports what it did: the in-process scenario (`Net`), or a seeded
202		/// distribution in a worker of its own process (the fleet simulator).
203		pub trait ActionSource: Send + Sync + core::fmt::Debug + 'static {
204		fn plan(&self, start_execute: &StartExecute) -> Result<Plan, Error>;
205
206		fn record(&self, event: Event);
207
208		/// The scheduler asked for a kill. The in-process network already
209		/// records it.
210	2	fn killed(&self, _operation_id: &OperationId) {}
211		}
212
213		/// The `RunningActionsManager` of one simulated worker.
214		#[derive(Debug)]
215		pub struct SimActionsManager {
216		worker: String,
217		source: Arc<dyn ActionSource>,
218		running: Arc<Mutex<HashMap<OperationId, Arc<KillSwitch>>>>,
219		metrics: Arc<Metrics>,
220		}
221
222		impl SimActionsManager {
223	1.19k	pub fn new(worker: &str, source: Arc<dyn ActionSource>) -> Self {
224	1.19k	Self {
225	1.19k	worker: worker.to_string(),
226	1.19k	source,
227	1.19k	running: Arc::new(Mutex::new(HashMap::new())),
228	1.19k	metrics: Arc::new(Metrics::default()),
229	1.19k	}
230	1.19k	}
231
232		/// Executions this worker has not finished.
233	20.2k	pub fn running(&self) -> usize {
234	20.2k	self.running.lock().len()
235	20.2k	}
236		}
237
238		#[derive(Debug, Default)]
239		struct KillSwitch {
240		killed: AtomicBool,
241		notify: Notify,
242		}
243
244		impl KillSwitch {
245	277	fn kill(&self) {
246	277	self.killed.store(true, Ordering::Release);
247	277	self.notify.notify_waiters();
248	277	}
249
250	10.7k	async fn wait(&self) {
251		loop {
252	10.7k	let notified = self.notify.notified();
253	10.7k	if self.killed.load(Ordering::Acquire) {
254	0	return;
255	10.7k	}
256	10.7k	notified.await;
257		}
258	0	}
259		}
260
261		impl RunningActionsManager for SimActionsManager {
262		type RunningAction = SimRunningAction;
263
264	10.6k	async fn create_and_add_action(
265	10.6k	self: &Arc<Self>,
266	10.6k	_worker_id: String,
267	10.6k	start_execute: StartExecute,
268	10.6k	) -> Result<Arc<SimRunningAction>, Error> {
269	10.6k	let plan = self.source.plan(&start_execute) ?0 ;
270	10.6k	let operation_id = OperationId::from(start_execute.operation_id);
271	10.6k	let kill = Arc::new(KillSwitch::default());
272	10.6k	self.running
273	10.6k	.lock()
274	10.6k	.insert(operation_id.clone(), kill.clone());
275	10.6k	Ok(Arc::new(SimRunningAction {
276	10.6k	worker: self.worker.clone(),
277	10.6k	action: plan.action,
278	10.6k	key: plan.key,
279	10.6k	operation_id,
280	10.6k	prepare: plan.prepare,
281	10.6k	run_for: plan.run_for,
282	10.6k	upload: plan.upload,
283	10.6k	run: plan.run,
284	10.6k	kill,
285	10.6k	running: self.running.clone(),
286	10.6k	source: self.source.clone(),
287	10.6k	work_directory: String::new(),
288	10.6k	started: AtomicBool::new(false),
289	10.6k	ended: AtomicBool::new(false),
290	10.6k	}))
291	10.6k	}
292
293	10.0k	async fn cache_action_result(
294	10.0k	&self,
295	10.0k	_action_digest: DigestInfo,
296	10.0k	_action_result: &mut ActionResult,
297	10.0k	_hasher: DigestHasherFunc,
298	10.0k	) -> Result<(), Error> {
299	10.0k	Ok(())
300	10.0k	}
301
302	646	async fn kill_all(&self) {
303	646	self.source.record(Event::KillAll {
304	646	worker: self.worker.clone(),
305	646	});
306	646	let switches: Vec<_> = self.running.lock().values().cloned().collect();
307	646	for switch277 in switches {
308	277	switch.kill();
309	277	}
310	646	}
311
312	2	async fn kill_operation(&self, operation_id: &OperationId) -> Result<(), Error> {
313	2	self.source.killed(operation_id);
314	2	let switch = self.running.lock().get(operation_id).cloned();
315	2	if let Some( switch0 ) = switch {
316	0	switch.kill();
317	2	}
318	2	Ok(())
319	2	}
320
321	1.19k	fn metrics(&self) -> &Arc<Metrics> {
322	1.19k	&self.metrics
323	1.19k	}
324		}
325
326		#[derive(Debug)]
327		pub struct SimRunningAction {
328		worker: String,
329		action: String,
330		key: String,
331		operation_id: OperationId,
332		prepare: Duration,
333		run_for: Duration,
334		upload: Duration,
335		run: Run,
336		kill: Arc<KillSwitch>,
337		running: Arc<Mutex<HashMap<OperationId, Arc<KillSwitch>>>>,
338		source: Arc<dyn ActionSource>,
339		work_directory: String,
340		started: AtomicBool,
341		ended: AtomicBool,
342		}
343
344		/// An execution whose task was aborted (the worker dropped it on a
345		/// disconnect) still ends in the trace and leaves the running set.
346		impl Drop for SimRunningAction {
347	10.6k	fn drop(&mut self) {
348	10.6k	if self.started.load(Ordering::Acquire) && !10.6k self.ended10.6k .load(Ordering::Acquire) {
349	348	self.end(true);
350	10.3k	}
351	10.6k	self.running.lock().remove(&self.operation_id);
352	10.6k	}
353		}
354
355		impl SimRunningAction {
356		/// Sleeps `duration` unless killed first; `Err` when killed.
357	31.4k	async fn phase(&self, duration: Duration) -> Result<(), Error> {
358	31.4k	if duration.is_zero() {
359	20.6k	return Ok(());
360	10.7k	}
361	10.7k	tokio::select! {
362	10.7k	() = tokio::time::sleep(duration) => Ok(())10.4k ,
363	10.7k	() = self.kill.wait() => Err(make_err!(Code::Aborted, "Killed"))0 ,
364		}
365	31.0k	}
366
367	10.6k	fn end(&self, killed: bool) {
368	10.6k	if self.ended.swap(true, Ordering::AcqRel) {
369	0	return;
370	10.6k	}
371	10.6k	self.source.record(Event::ExecEnd {
372	10.6k	worker: self.worker.clone(),
373	10.6k	action: self.action.clone(),
374	10.6k	key: self.key.clone(),
375	10.6k	killed,
376	10.6k	});
377	10.6k	}
378		}
379
380		impl RunningAction for SimRunningAction {
381	236	fn get_operation_id(&self) -> &OperationId {
382	236	&self.operation_id
383	236	}
384
385	10.6k	async fn prepare_action(self: Arc<Self>) -> Result<Arc<Self>, Error> {
386	10.6k	self.phase(self.prepare).await ?0 ;
387	10.6k	Ok(self)
388	10.6k	}
389
390	10.6k	async fn execute(self: Arc<Self>) -> Result<Arc<Self>, Error> {
391	10.6k	self.started.store(true, Ordering::Release);
392	10.6k	self.source.record(Event::ExecStart {
393	10.6k	worker: self.worker.clone(),
394	10.6k	action: self.action.clone(),
395	10.6k	key: self.key.clone(),
396	10.6k	});
397	10.6k	let ran10.3k = match self.run {
398		Run::Hang => {
399	0	self.kill.wait().await;
400	0	Err(make_err!(Code::Aborted, "Killed"))
401		}
402	10.6k	_ => self.phase(self.run_for).await,
403		};
404	10.3k	if let Err( err0 ) = ran {
405	0	self.end(true);
406	0	return Err(err);
407	10.3k	}
408	10.3k	self.end(false);
409	10.3k	match self.run {
410	0	Run::Panic => panic!("simulated action {} panicked", self.action),
411	258	Run::Error(code) => Err(make_err!(code, "Simulated failure of {}", self.action)),
412	10.0k	Run::Succeed \| Run::Exit(_) \| Run::Hang => Ok(self),
413		}
414	10.3k	}
415
416	10.0k	async fn upload_results(self: Arc<Self>) -> Result<Arc<Self>, Error> {
417	10.0k	self.phase(self.upload).await ?0 ;
418	10.0k	Ok(self)
419	10.0k	}
420
421	10.3k	async fn cleanup(self: Arc<Self>) -> Result<Arc<Self>, Error> {
422	10.3k	self.running.lock().remove(&self.operation_id);
423	10.3k	Ok(self)
424	10.3k	}
425
426	10.0k	async fn get_finished_result(self: Arc<Self>) -> Result<ActionResult, Error> {
427	10.0k	let exit_code = match self.run {
428	0	Run::Exit(code) => code,
429	10.0k	_ => 0,
430		};
431	10.0k	Ok(ActionResult {
432	10.0k	exit_code,
433	10.0k	..ActionResult::default()
434	10.0k	})
435	10.0k	}
436
437	10.0k	fn resource_usage(&self) -> Option<ActionResourceUsage> {
438	10.0k	None
439	10.0k	}
440
441	0	fn get_work_directory(&self) -> &String {
442	0	&self.work_directory
443	0	}
444		}