git-ents.gitmain
⌘K
foforge

crates/kernel/ents-receive/src/reconcile.rs

reconcile.rs161 lines · 6.0 KB · rusthistorycomment on this file
1//! The boot-time reconciliation scan (`receive.reconstructible`): the
2//! reference proof that the obligation queue needs no state `receive`
3//! itself did not already have available in the repository.
4
5use ents_model::Effect;
6use ents_query::{Evaluator, Query};
7use gix::refs::FullName;
8use gix_hash::ObjectId;
9use gix_object::{CommitRef, Find, Kind};
10use gix_ref_store::RefStoreRead;
11
12use crate::error::{Error, Result};
13use crate::sink::EventSink;
14
15const EFFECTS_PREFIX: &str = "refs/meta/effects/";
16
17/// Every effect definition currently readable under `refs/meta/effects/*`,
18/// as `(name, parsed trigger)`.
19///
20/// An effect whose tree cannot be read, or whose `trigger` fails to parse,
21/// is skipped rather than failing the scan: `receive.validation`
22/// (`ents-model`, `ents-query`) is what keeps a *newly written* effect
23/// well-formed, but a scan reused across every future push must stay
24/// resilient to a pre-existing malformed one rather than let it take down
25/// every push after it (the same spirit as `receive.never-blocks`, applied
26/// to a scan that runs on `receive`'s own hot path).
27// @relation(receive.event-sink, scope=function)
28fn known_effects(refs: &dyn RefStoreRead, objects: &impl Find) -> Result<Vec<(String, Query)>> {
29 let mut effects = Vec::new();
30 for entry in refs.iter_prefix(EFFECTS_PREFIX)? {
31 let (name, tip) = entry?;
32 let Some(short) = short_effect_name(&name) else {
33 continue;
34 };
35 let Some(tree) = commit_tree(objects, tip)? else {
36 continue;
37 };
38 let Ok(effect) = facet_git_tree::deserialize::<Effect>(&tree, objects) else {
39 continue;
40 };
41 let Ok(trigger) = effect.trigger.parse::<Query>() else {
42 continue;
43 };
44 effects.push((short, trigger));
45 }
46 Ok(effects)
47}
48
49/// The effect name segment of a `refs/meta/effects/<name>` refname, or
50/// `None` for anything deeper or shallower (mirrors the results-namespace
51/// scan in `ents-query`'s evaluator).
52fn short_effect_name(name: &FullName) -> Option<String> {
53 let path = name.as_bstr().to_string();
54 let short = path.strip_prefix(EFFECTS_PREFIX)?;
55 (!short.is_empty() && !short.contains('/')).then(|| short.to_owned())
56}
57
58/// The tree of the commit at `oid`, or `None` if `oid` is missing or not a
59/// commit — treated as "this ref is unreadable", never a hard failure of
60/// the whole scan. Shared with [`crate::receive`]'s redaction-target scan.
61pub(crate) fn commit_tree(objects: &impl Find, oid: ObjectId) -> Result<Option<ObjectId>> {
62 let mut buf = Vec::new();
63 let Some(data) = objects
64 .try_find(&oid, &mut buf)
65 .map_err(|source| Error::Decode {
66 oid,
67 detail: source.to_string(),
68 })?
69 else {
70 return Ok(None);
71 };
72 if data.kind != Kind::Commit {
73 return Ok(None);
74 }
75 let Ok(commit) = CommitRef::from_bytes(data.data, oid.kind()) else {
76 return Ok(None);
77 };
78 Ok(Some(commit.tree()))
79}
80
81/// The full, reconciliation-grade obligation scan
82/// (`receive.reconstructible`): for every effect currently defined, compute
83/// its outstanding work set (`trigger − results(self, any)`,
84/// `query.workset`) against current ref state and enqueue every commit
85/// still owed a result.
86///
87/// A composition root calls this once at startup, before serving further
88/// pushes, so an `EventSink` that lost its queued events on crash (the null
89/// sink always; the in-memory reference sink after a restart) recovers
90/// exactly the same obligations incremental `receive` calls would have
91/// enqueued — the queue is reconstructible from repository state alone,
92/// with the dedup key (`receive.dedup`) unchanged by reconciliation.
93///
94/// # Errors
95///
96/// Fails only on a ref-store or object-store read failure, or a sink
97/// failure; a malformed individual effect definition is skipped, not an
98/// error (this module's private effect-scan helper treats an unreadable
99/// tree or an unparsable trigger as "no match", never a hard failure).
100///
101/// # Examples
102///
103/// ```
104/// use ents_model::Effect;
105/// use ents_receive::{MemoryEventSink, reconcile};
106/// use ents_testutil::{MemRefStore, ObjectStore, advance_ref, write_meta_entity};
107///
108/// let refs = MemRefStore::default();
109/// let objects = ObjectStore::default();
110/// let commits = advance_ref(&refs, &objects, "refs/heads/main", 1, 100);
111///
112/// let effect = Effect {
113/// name: "unit".to_owned(),
114/// trigger: "rev(refs/heads/main)".to_owned(),
115/// toolchains: vec![],
116/// run: "true".to_owned(),
117/// };
118/// let name: gix::refs::FullName = "refs/meta/effects/unit".try_into().expect("valid");
119/// write_meta_entity(&refs, &objects, name, &effect, None, 200);
120///
121/// let sink = MemoryEventSink::default();
122/// reconcile(&refs, &objects, &sink).expect("reconciles");
123/// assert_eq!(sink.pending(), vec![("unit".to_owned(), commits[0])]);
124/// ```
125// @relation(receive.reconstructible, query.workset, scope=function)
126pub fn reconcile(
127 refs: &dyn RefStoreRead,
128 objects: &impl Find,
129 events: &dyn EventSink,
130) -> Result<()> {
131 let evaluator = Evaluator::new(refs, objects);
132 for (name, trigger) in known_effects(refs, objects)? {
133 for oid in evaluator.outstanding(&name, &trigger)? {
134 enqueue(events, &name, oid)?;
135 }
136 }
137 Ok(())
138}
139
140/// Enqueue matches for `transition` against every known effect
141/// (`receive.event-sink`): the incremental counterpart to [`reconcile`],
142/// called by [`crate::receive`] once per successfully applied transition.
143// @relation(receive.event-sink, query.workset, scope=function)
144pub(crate) fn enqueue_matches(
145 refs: &dyn RefStoreRead,
146 objects: &impl Find,
147 events: &dyn EventSink,
148 transition: &ents_query::Transition,
149) -> Result<()> {
150 let evaluator = Evaluator::new(refs, objects);
151 for (name, trigger) in known_effects(refs, objects)? {
152 for oid in evaluator.work_set(&name, &trigger, transition)? {
153 enqueue(events, &name, oid)?;
154 }
155 }
156 Ok(())
157}
158
159fn enqueue(events: &dyn EventSink, effect: &str, oid: ObjectId) -> Result<()> {
160 events.enqueue(effect, oid)
161}