git-ents.gitmain
⌘K
foforge

crates/kernel/ents-effect/src/run.rs

run.rs625 lines · 21.8 KB · rusthistorycomment on this file
1//! The run loop (`effect.execution`, `effect.local-run`): materialize an
2//! effect's declared toolchains and the tested commit's tree, hand both to
3//! an [`Executor`], and write the outcome back through
4//! [`crate::write_result`].
5//!
6//! [`run_one`] is the one code path a hosted worker and `git effect run`
7//! both call (`effect.local-run`: "the identical code path a hosted worker
8//! uses"); only what surrounds it differs — a durable queue feeding a
9//! worker's loop of [`run_one`] calls, versus [`run_effect`] deriving the
10//! same obligations directly from [`ents_query::Evaluator::outstanding`]
11//! and calling [`run_one`] once per commit, with no queue at all
12//! (`effect.local-run`: "only the durable queue MUST be skipped").
13
14use std::path::{Path, PathBuf};
15
16use ents_model::Effect;
17use ents_query::{Evaluator, Query};
18use ents_receive::{EventSink, Mode, Outcome};
19use gix::refs::FullName;
20use gix_hash::ObjectId;
21use gix_object::{CommitRef, Find, Kind, Write};
22use gix_ref_store::RefStore;
23
24use crate::error::{Error, Result};
25use crate::executor::{Executor, RunStatus, SandboxInputs};
26use crate::results::write_result;
27
28/// The tree of the commit at `oid`.
29fn commit_tree(objects: &impl Find, oid: ObjectId) -> Result<ObjectId> {
30 let mut buf = Vec::new();
31 let data = objects
32 .try_find(&oid, &mut buf)
33 .map_err(|source| Error::Decode {
34 oid,
35 detail: source.to_string(),
36 })?
37 .ok_or(Error::Missing { oid })?;
38 if data.kind != Kind::Commit {
39 return Err(Error::Decode {
40 oid,
41 detail: "expected a commit".to_owned(),
42 });
43 }
44 let commit = CommitRef::from_bytes(data.data, oid.kind()).map_err(|e| Error::Decode {
45 oid,
46 detail: e.to_string(),
47 })?;
48 Ok(commit.tree())
49}
50
51/// The short-oid segment convention every results refname uses:
52/// `refs/meta/results/<effect>/<short-oid>` (`effect.results-writeback`) —
53/// the first 12 hex characters, long enough to stay unambiguous within one
54/// effect's results namespace while keeping refnames short.
55///
56/// # Examples
57///
58/// ```
59/// use ents_effect::run::short_oid;
60///
61/// let oid = gix_hash::ObjectId::null(gix_hash::Kind::Sha1);
62/// assert_eq!(short_oid(oid), "000000000000");
63/// ```
64#[must_use]
65pub fn short_oid(oid: ObjectId) -> String {
66 let hex = oid.to_string();
67 hex.get(..12).unwrap_or(&hex).to_owned()
68}
69
70/// Run `effect` against the single commit `oid`: check out `oid`'s tree,
71/// execute via `executor` against it and the already-materialized
72/// `toolchains`, and write the outcome to `results_ref` — the one code
73/// path `effect.local-run` names.
74///
75/// `toolchains` is each of `effect`'s declared toolchains, already resolved
76/// to a host `bin/` directory, in the effect's declared order
77/// (`crate::executor::activate`'s PATH-collision tiebreak depends on this
78/// order surviving) — this crate no longer resolves toolchain names
79/// itself (`ents-kiln` owns that; see this crate's own module doc), so the
80/// caller (a composition root that can depend on both `ents-effect` and
81/// `ents-kiln`) must resolve and materialize them first.
82///
83/// `results_ref` is the caller's choice (`effect.self-run`,
84/// `effect.official`): the canonical results ref for a designated worker,
85/// or a self-run mirror for any other member. `scratch` holds the
86/// per-run, never-cached tree checkout — the per-commit workdir under it
87/// is wiped and re-checked-out on every run, so a re-run over a persistent
88/// scratch directory never inherits a previous run's artifacts (a Docker
89/// container is thrown away per run; a Sprite's `sync_dir` re-syncs it
90/// every time too, so nothing here needs it to survive).
91///
92/// # Errors
93///
94/// Any [`Error`] from checking out `oid`'s tree, the executor itself, or
95/// [`crate::write_result`].
96///
97/// # Examples
98///
99/// ```
100/// use ents_effect::run::run_one;
101/// use ents_effect::{Executor, RunOutput, RunStatus, SandboxInputs};
102/// use ents_model::{Effect, Provenance, namespace};
103/// use ents_receive::{Mode, NullEventSink};
104/// use ents_testutil::{Keypair, MemRefStore, ObjectStore, advance_ref, enroll_member};
105///
106/// struct AlwaysPass;
107/// impl Executor for AlwaysPass {
108/// fn run(&self, _inputs: &SandboxInputs<'_>) -> ents_effect::Result<RunOutput> {
109/// Ok(RunOutput { status: RunStatus::Pass, log: String::new() })
110/// }
111/// }
112///
113/// let refs = MemRefStore::default();
114/// let objects = ObjectStore::default();
115/// let worker = Keypair::from_seed(1);
116/// enroll_member(&refs, &objects, "worker", &worker, Provenance::AdminRegistered, 100);
117/// let commits = advance_ref(&refs, &objects, "refs/heads/main", 1, 200);
118///
119/// let effect = Effect { name: "unit".into(), trigger: "rev(refs/heads/main)".into(), toolchains: vec![], run: "true".into() };
120/// let results_ref = namespace::result_ref("unit", "abcabcabcabc").expect("valid");
121/// let author = gix::actor::Signature {
122/// name: "worker".into(), email: "worker@ents.test".into(),
123/// time: gix::date::Time { seconds: 300, offset: 0 },
124/// };
125/// let scratch = tempfile::tempdir().expect("tempdir");
126///
127/// let outcome = run_one(
128/// &refs, &objects, &NullEventSink, &AlwaysPass, scratch.path(), &[],
129/// commits[0], &effect, results_ref, &author, |p| worker.sign(p), Mode::Advisory,
130/// ).expect("runs");
131/// assert_eq!(outcome.result, ents_receive::TxResult::Applied);
132/// ```
133// @relation(effect.execution, effect.local-run, effect.toolchains, scope=function)
134#[expect(
135 clippy::too_many_arguments,
136 reason = "one input per materialization step, mirrors pre-redo's engine::run shape"
137)]
138pub fn run_one(
139 refs: &dyn RefStore,
140 objects: &(impl Find + Write),
141 events: &dyn EventSink,
142 executor: &dyn Executor,
143 scratch: &Path,
144 toolchains: &[(String, PathBuf)],
145 oid: ObjectId,
146 effect: &Effect,
147 results_ref: FullName,
148 author: &gix::actor::Signature,
149 sign: impl FnOnce(&[u8]) -> String,
150 mode: Mode,
151) -> Result<Outcome> {
152 // A fresh checkout per run: over a persistent scratch directory, a
153 // re-run of the same commit must not inherit the previous run's
154 // artifacts (build output, files the last command left behind).
155 let workdir = scratch.join(oid.to_string());
156 if workdir.exists() {
157 std::fs::remove_dir_all(&workdir).map_err(|source| Error::Io {
158 path: workdir.clone(),
159 source,
160 })?;
161 }
162 std::fs::create_dir_all(&workdir).map_err(|source| Error::Io {
163 path: workdir.clone(),
164 source,
165 })?;
166 let tree = commit_tree(objects, oid)?;
167 crate::materialize::checkout(objects, tree, &workdir)?;
168
169 let inputs = SandboxInputs {
170 workdir: &workdir,
171 toolchains,
172 command: &effect.run,
173 // An ordinary effect run has no per-member credential to inject
174 // (`roots.config-isolation`'s BYOK injection seam is a
175 // composition-root concern); every backend still accepts the
176 // field uniformly, it is simply empty here.
177 env: &[],
178 };
179 let output = executor.run(&inputs)?;
180 let status = match output.status {
181 RunStatus::Pass => ents_model::Status::Pass,
182 RunStatus::Fail => ents_model::Status::Fail,
183 };
184
185 write_result(
186 refs,
187 objects,
188 events,
189 results_ref,
190 &effect.name,
191 oid,
192 status,
193 author,
194 sign,
195 mode,
196 )
197}
198
199/// Run `effect` against every commit currently owed a result
200/// (`ents_query::Evaluator::outstanding`, `query.workset`), or against the
201/// single commit `at` when given — the boot-time/on-demand form
202/// [`run_one`]'s doc names, and the shape `git effect run [--at <commit>]`
203/// (a future frontend) calls.
204///
205/// `results_ref` builds each run's target refname from its short oid
206/// (`crate::run::short_oid`) — pass `ents_model::namespace::result_ref` for
207/// a canonical worker or `ents_model::namespace::self_result_ref` curried
208/// to one member for a self-run (`effect.self-run`); this function makes
209/// no canonical-vs-self decision itself.
210///
211/// `toolchains` is `effect`'s declared toolchains, already resolved once
212/// for the whole batch (an effect's declared toolchain list is fixed for
213/// the whole call, so there is no need to re-resolve per commit) — passed
214/// straight through to every [`run_one`] call.
215///
216/// # Errors
217///
218/// [`Error::Trigger`] or [`Error::Eval`] if the trigger cannot be parsed
219/// or the work set computed; otherwise [`Error::Run`], carrying the id of
220/// the commit whose run failed with the underlying failure as its source
221/// — later commits in the (oid-sorted) set are not attempted once one
222/// fails. Outcomes for commits that completed before the failure were
223/// already durably recorded through `receive` (each run writes back
224/// immediately), so a caller applying its own retry policy
225/// (`effect.deployment-property`) resumes from the reported commit, and a
226/// plain retry of the whole batch re-runs only what is still owed
227/// (`query.workset`).
228// @relation(effect.local-run, query.workset, scope=function)
229#[expect(
230 clippy::too_many_arguments,
231 reason = "one input per materialization step plus the target-ref builder"
232)]
233pub fn run_effect(
234 refs: &dyn RefStore,
235 objects: &(impl Find + Write),
236 events: &dyn EventSink,
237 executor: &dyn Executor,
238 scratch: &Path,
239 toolchains: &[(String, PathBuf)],
240 effect_name: &str,
241 effect: &Effect,
242 at: Option<ObjectId>,
243 results_ref: impl Fn(&str) -> Result<FullName>,
244 author: &gix::actor::Signature,
245 sign: &impl Fn(&[u8]) -> String,
246 mode: Mode,
247) -> Result<Vec<(ObjectId, Outcome)>> {
248 let trigger: Query = effect.trigger.parse()?;
249 let oids: Vec<ObjectId> = match at {
250 Some(oid) => vec![oid],
251 None => {
252 // `query.workset`'s dedup marker is always the effect's own
253 // *canonical* results namespace, regardless of which ref this
254 // particular run's outcome ends up targeting
255 // (`results_ref`) — a self-run mirror never discharges the
256 // canonical obligation, by construction.
257 let evaluator = Evaluator::new(refs, objects);
258 evaluator
259 .outstanding(effect_name, &trigger)?
260 .into_iter()
261 .collect()
262 }
263 };
264
265 let mut outcomes = Vec::with_capacity(oids.len());
266 for oid in oids {
267 let one = results_ref(&short_oid(oid)).and_then(|target| {
268 run_one(
269 refs, objects, events, executor, scratch, toolchains, oid, effect, target, author,
270 sign, mode,
271 )
272 });
273 match one {
274 Ok(outcome) => outcomes.push((oid, outcome)),
275 Err(source) => {
276 return Err(Error::Run {
277 oid,
278 source: Box::new(source),
279 });
280 }
281 }
282 }
283 Ok(outcomes)
284}
285
286#[cfg(test)]
287mod tests {
288 #![allow(
289 clippy::expect_used,
290 clippy::panic,
291 reason = "unit test; the panic is an assertion on the error's variant"
292 )]
293
294 use ents_model::{Provenance, namespace};
295 use ents_receive::{NullEventSink, TxResult};
296 use ents_testutil::{Keypair, MemRefStore, ObjectStore, advance_ref, enroll_member};
297 use gix_ref_store::RefStoreRead as _;
298 use rstest::rstest;
299
300 use super::*;
301 use crate::executor::{RunOutput, RunStatus, SandboxInputs};
302
303 struct AlwaysPass;
304 impl Executor for AlwaysPass {
305 fn run(&self, _inputs: &SandboxInputs<'_>) -> Result<RunOutput> {
306 Ok(RunOutput {
307 status: RunStatus::Pass,
308 log: String::new(),
309 })
310 }
311 }
312
313 /// Passes `passes` runs, then reports an infrastructure failure on
314 /// every run after them.
315 struct FailsAfter {
316 passes: usize,
317 calls: std::sync::atomic::AtomicUsize,
318 }
319 impl Executor for FailsAfter {
320 fn run(&self, _inputs: &SandboxInputs<'_>) -> Result<RunOutput> {
321 let call = self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
322 if call < self.passes {
323 Ok(RunOutput {
324 status: RunStatus::Pass,
325 log: String::new(),
326 })
327 } else {
328 Err(Error::Sandbox("the sandbox never started".to_owned()))
329 }
330 }
331 }
332
333 fn author() -> gix::actor::Signature {
334 gix::actor::Signature {
335 name: "worker".into(),
336 email: "worker@ents.test".into(),
337 time: gix::date::Time {
338 seconds: 500,
339 offset: 0,
340 },
341 }
342 }
343
344 #[rstest]
345 // @relation(effect.local-run, query.workset, scope=function, role=Verifies)
346 fn run_effect_derives_the_full_outstanding_set_with_no_queue_at_all() {
347 let refs = MemRefStore::default();
348 let objects = ObjectStore::default();
349 let worker = Keypair::from_seed(1);
350 enroll_member(
351 &refs,
352 &objects,
353 "worker",
354 &worker,
355 Provenance::AdminRegistered,
356 100,
357 );
358 let commits = advance_ref(&refs, &objects, "refs/heads/main", 2, 200);
359
360 let effect = Effect {
361 name: "unit".into(),
362 trigger: "rev(refs/heads/main)".into(),
363 toolchains: vec![],
364 run: "true".into(),
365 };
366 let scratch = tempfile::tempdir().expect("tempdir");
367
368 // `NullEventSink`: the only component `effect.local-run` says this
369 // path skips is the durable queue, and this run derives its work
370 // set directly from `query.workset` instead of draining one.
371 let outcomes = run_effect(
372 &refs,
373 &objects,
374 &NullEventSink,
375 &AlwaysPass,
376 scratch.path(),
377 &[],
378 "unit",
379 &effect,
380 None,
381 |short| Ok(namespace::result_ref("unit", short).expect("valid")),
382 &author(),
383 &|payload| worker.sign(payload),
384 Mode::Advisory,
385 )
386 .expect("runs");
387
388 assert_eq!(outcomes.len(), 2);
389 let mut ran: Vec<_> = outcomes.iter().map(|(oid, _)| *oid).collect();
390 ran.sort();
391 let mut expected = commits.clone();
392 expected.sort();
393 assert_eq!(ran, expected);
394 for (_, outcome) in &outcomes {
395 assert_eq!(outcome.result, TxResult::Applied);
396 }
397 }
398
399 #[rstest]
400 // @relation(effect.local-run, scope=function, role=Verifies)
401 fn run_effect_at_a_single_commit_skips_the_work_set_scan() {
402 let refs = MemRefStore::default();
403 let objects = ObjectStore::default();
404 let worker = Keypair::from_seed(1);
405 enroll_member(
406 &refs,
407 &objects,
408 "worker",
409 &worker,
410 Provenance::AdminRegistered,
411 100,
412 );
413 let commits = advance_ref(&refs, &objects, "refs/heads/main", 3, 200);
414
415 let effect = Effect {
416 name: "unit".into(),
417 trigger: "rev(refs/heads/main)".into(),
418 toolchains: vec![],
419 run: "true".into(),
420 };
421 let scratch = tempfile::tempdir().expect("tempdir");
422
423 let first = *commits.first().expect("advance_ref produced a commit");
424 let outcomes = run_effect(
425 &refs,
426 &objects,
427 &NullEventSink,
428 &AlwaysPass,
429 scratch.path(),
430 &[],
431 "unit",
432 &effect,
433 Some(first),
434 |short| Ok(namespace::result_ref("unit", short).expect("valid")),
435 &author(),
436 &|payload| worker.sign(payload),
437 Mode::Advisory,
438 )
439 .expect("runs");
440
441 assert_eq!(outcomes.len(), 1);
442 let (oid, _) = outcomes.first().expect("one outcome");
443 assert_eq!(*oid, first);
444 }
445
446 #[rstest]
447 // @relation(effect.self-run, effect.local-run, scope=function, role=Verifies)
448 fn run_effect_can_target_the_self_run_namespace_via_its_results_ref_closure() {
449 let refs = MemRefStore::default();
450 let objects = ObjectStore::default();
451 let bob = Keypair::from_seed(2);
452 enroll_member(
453 &refs,
454 &objects,
455 "bob",
456 &bob,
457 Provenance::AdminRegistered,
458 100,
459 );
460 let commits = advance_ref(&refs, &objects, "refs/heads/main", 1, 200);
461
462 let effect = Effect {
463 name: "unit".into(),
464 trigger: "rev(refs/heads/main)".into(),
465 toolchains: vec![],
466 run: "true".into(),
467 };
468 let scratch = tempfile::tempdir().expect("tempdir");
469 let member = ents_model::MemberId::new("bob");
470
471 let outcomes = run_effect(
472 &refs,
473 &objects,
474 &NullEventSink,
475 &AlwaysPass,
476 scratch.path(),
477 &[],
478 "unit",
479 &effect,
480 None,
481 |short| Ok(namespace::self_result_ref(&member, "unit", short).expect("valid")),
482 &author(),
483 &|payload| bob.sign(payload),
484 Mode::Advisory,
485 )
486 .expect("runs");
487
488 assert_eq!(outcomes.len(), 1);
489 let first = *commits.first().expect("advance_ref produced a commit");
490 let name = namespace::self_result_ref(&member, "unit", &short_oid(first)).expect("valid");
491 assert!(refs.get(name.as_ref()).expect("readable").is_some());
492 }
493
494 #[rstest]
495 // @relation(effect.local-run, effect.deployment-property, scope=function, role=Verifies)
496 fn run_effect_reports_which_commit_stopped_the_batch() {
497 let refs = MemRefStore::default();
498 let objects = ObjectStore::default();
499 let worker = Keypair::from_seed(1);
500 enroll_member(
501 &refs,
502 &objects,
503 "worker",
504 &worker,
505 Provenance::AdminRegistered,
506 100,
507 );
508 advance_ref(&refs, &objects, "refs/heads/main", 2, 200);
509
510 let effect = Effect {
511 name: "unit".into(),
512 trigger: "rev(refs/heads/main)".into(),
513 toolchains: vec![],
514 run: "true".into(),
515 };
516 let scratch = tempfile::tempdir().expect("tempdir");
517
518 // The first run completes and writes back; the second's sandbox
519 // never starts.
520 let executor = FailsAfter {
521 passes: 1,
522 calls: std::sync::atomic::AtomicUsize::new(0),
523 };
524 let err = run_effect(
525 &refs,
526 &objects,
527 &NullEventSink,
528 &executor,
529 scratch.path(),
530 &[],
531 "unit",
532 &effect,
533 None,
534 |short| Ok(namespace::result_ref("unit", short).expect("valid")),
535 &author(),
536 &|payload| worker.sign(payload),
537 Mode::Advisory,
538 )
539 .expect_err("the second run's infrastructure failure stops the batch");
540
541 // The batch runs the work set in oid-sorted order, so the failing
542 // commit is the second-sorted one — exactly what the error names.
543 let evaluator = Evaluator::new(&refs, &objects);
544 let sorted: Vec<ObjectId> = evaluator
545 .eval(&"rev(refs/heads/main)".parse().expect("valid"))
546 .expect("evaluates")
547 .into_iter()
548 .collect();
549 let Error::Run { oid, source } = err else {
550 panic!("expected Error::Run, got {err:?}");
551 };
552 assert_eq!(Some(&oid), sorted.get(1), "the error names the stopper");
553 assert!(matches!(*source, Error::Sandbox(_)));
554
555 // The completed first run was durably recorded before the failure:
556 // a retry re-runs only what is still owed (query.workset).
557 let first = *sorted.first().expect("two commits");
558 let name = namespace::result_ref("unit", &short_oid(first)).expect("valid");
559 assert!(
560 refs.get(name.as_ref()).expect("readable").is_some(),
561 "the completed run's result must survive the batch failure"
562 );
563 let outstanding = evaluator
564 .outstanding("unit", &"rev(refs/heads/main)".parse().expect("valid"))
565 .expect("evaluates");
566 assert_eq!(
567 outstanding.into_iter().collect::<Vec<_>>(),
568 vec![oid],
569 "only the failed commit is still owed a result"
570 );
571 }
572
573 #[rstest]
574 // @relation(effect.local-run, scope=function, role=Verifies)
575 fn run_one_wipes_a_reused_workdir_before_checkout() {
576 let refs = MemRefStore::default();
577 let objects = ObjectStore::default();
578 let worker = Keypair::from_seed(1);
579 enroll_member(
580 &refs,
581 &objects,
582 "worker",
583 &worker,
584 Provenance::AdminRegistered,
585 100,
586 );
587 let commits = advance_ref(&refs, &objects, "refs/heads/main", 1, 200);
588 let first = *commits.first().expect("advance_ref produced a commit");
589
590 let effect = Effect {
591 name: "unit".into(),
592 trigger: "rev(refs/heads/main)".into(),
593 toolchains: vec![],
594 run: "true".into(),
595 };
596 let scratch = tempfile::tempdir().expect("tempdir");
597
598 // A previous run over the same persistent scratch left artifacts
599 // in this commit's workdir.
600 let workdir = scratch.path().join(first.to_string());
601 std::fs::create_dir_all(&workdir).expect("mkdir");
602 std::fs::write(workdir.join("stale-artifact.txt"), b"left behind").expect("write");
603
604 run_one(
605 &refs,
606 &objects,
607 &NullEventSink,
608 &AlwaysPass,
609 scratch.path(),
610 &[],
611 first,
612 &effect,
613 namespace::result_ref("unit", &short_oid(first)).expect("valid"),
614 &author(),
615 |p| worker.sign(p),
616 Mode::Advisory,
617 )
618 .expect("runs");
619
620 assert!(
621 !workdir.join("stale-artifact.txt").exists(),
622 "a re-run must not inherit the previous run's artifacts"
623 );
624 }
625}