choir_node/queue.rs
1//! The node's own merge queue (D68).
2//!
3//! The bridge queues pull requests from somebody else's forge. A node
4//! has no pull requests; it has proposals, which are pushes to
5//! `refs/for/<branch>/<user>/<topic>` admitted by the `propose` grant
6//! (D53, D60). This is the reading that turns those refs into a round,
7//! and the landing that writes the round's result back where the log
8//! decides.
9//!
10//! Two things differ from the bridge's composition, and both come from
11//! the same fact: here the log *is* the source of truth.
12//!
13//! - A landing is real. It moves `refs/heads/<branch>`, so it is an
14//! [`OpKind::SetRef`] carrying the value the round started from as
15//! its compare-and-swap `prev`. A push that beat the round makes the
16//! landing fail rather than clobber.
17//! - The landing itself moves no git ref. The log leads and git
18//! follows, which is the direction D21 fixed: `/api/queue/run` writes
19//! git's ref after the round, CAS'd on the round's base, and
20//! [`crate::Platform::reconcile_git_refs`] repairs anything that
21//! still lags at the next startup.
22
23use std::sync::Arc;
24
25use choir_hash::ContentHash;
26use choir_identity::ActorKey;
27use choir_queue::landing::Landing;
28use choir_queue::Change;
29use choir_sequencer::SequencerHandle;
30use choir_view::{OpKind, View, ViewOp};
31
32/// The channel the node's queue authors under.
33///
34/// Its own, not the proposer's: the ordering of a round is the node's
35/// statement, and attributing it to whoever happened to be first in the
36/// train would put a landing decision in somebody else's name.
37pub const QUEUE_CHANNEL: &str = "node/queue";
38
39/// One proposal in a round.
40#[derive(Debug, Clone, PartialEq, Eq)]
41pub struct Proposal {
42 /// Round-local address, the position of [`Proposal::refname`] in the
43 /// round's sorted order. Not an identity: the queue's
44 /// cross-round identity is the patch identity
45 /// [`choir_queue::speculate::Speculator::identity`] computes, which
46 /// survives a rebase and a rename and this does not.
47 pub id: u64,
48 /// The proposal's full name in the view, `<repo>:refs/for/...`.
49 pub refname: String,
50 /// The commit the proposal points at.
51 pub head: String,
52}
53
54/// The proposals aimed at one branch of one repository, plus the base
55/// they are aimed at.
56#[derive(Debug, Clone, PartialEq, Eq)]
57pub struct ProposalRound {
58 /// The repository, as the view names it (`<owner>/<name>.git`).
59 pub repo: String,
60 /// The branch proposals are aimed at, short form (`main`).
61 pub branch: String,
62 /// The branch's commit when the round was read, and the `prev` every
63 /// landing in it is CAS'd against.
64 pub base: String,
65 /// The round's proposals, in refname order.
66 pub proposals: Vec<Proposal>,
67}
68
69impl ProposalRound {
70 /// Reads the round for `repo` and `branch` out of `view`.
71 ///
72 /// `None` when the branch does not exist: there is nothing to
73 /// propose to, and a queue that invented a base would be deciding
74 /// what a repository's history starts from.
75 ///
76 /// Ordering is by refname, and deliberately not by when the ref was
77 /// pushed. The view records no push time, and taking the order from
78 /// the op log's sequence numbers would make a round's behaviour
79 /// depend on how far the log had been compacted.
80 #[must_use]
81 pub fn from_view(view: &View, repo: &str, branch: &str) -> Option<Self> {
82 let base = view.refs.get(&format!("{repo}:refs/heads/{branch}"))?;
83 let base = base.git_oid()?;
84 let prefix = format!("{repo}:refs/for/{branch}/");
85 let proposals = view
86 .refs
87 .range(prefix.clone()..)
88 .take_while(|(name, _)| name.starts_with(&prefix))
89 .filter_map(|(name, head)| Some((name.clone(), head.git_oid()?)))
90 .enumerate()
91 .map(|(i, (refname, head))| Proposal {
92 // One-based, so a report never addresses a change as 0
93 // and leaves a reader wondering whether it means "none".
94 id: i as u64 + 1,
95 refname,
96 head,
97 })
98 .collect();
99 Some(Self {
100 repo: repo.to_string(),
101 branch: branch.to_string(),
102 base,
103 proposals,
104 })
105 }
106
107 /// The round as queue input.
108 ///
109 /// The workspace is the proposal's own refname, so a landing
110 /// recorded against a workspace names the thing that was proposed
111 /// rather than a number only this round knows.
112 #[must_use]
113 pub fn changes(&self) -> Vec<Change> {
114 self.proposals
115 .iter()
116 .map(|p| Change {
117 id: p.id,
118 workspace: p.refname.clone(),
119 base: self.base.clone(),
120 proposed: p.head.clone(),
121 depends: Vec::new(),
122 })
123 .collect()
124 }
125
126 /// The view key of the branch this round lands on.
127 #[must_use]
128 pub fn target_ref(&self) -> String {
129 format!("{}:refs/heads/{}", self.repo, self.branch)
130 }
131
132 /// The proposals in this round whose head is already an ancestor of
133 /// the branch: landed by an earlier round, or merged some other way.
134 ///
135 /// Proposal refs survive their landing on purpose (sweeping one
136 /// would decide on the author's behalf that the change is done), and
137 /// the queue's own already-landed memory lives in a `MergeQueue`
138 /// that a node builds fresh per round. Without this, every landed
139 /// proposal re-entered every later round as a no-op merge, costing a
140 /// CI run and a landing op each time. Ancestry is read from git in
141 /// `workdir`, a worktree of the served repository.
142 #[must_use]
143 pub fn already_integrated(&self, workdir: &std::path::Path) -> Vec<u64> {
144 self.proposals
145 .iter()
146 .filter(|p| {
147 std::process::Command::new("git")
148 .args(["merge-base", "--is-ancestor", &p.head, &self.base])
149 .current_dir(workdir)
150 .status()
151 .is_ok_and(|status| status.success())
152 })
153 .map(|p| p.id)
154 .collect()
155 }
156}
157
158/// Where a landing reads the scope it signs: this node's id and a head
159/// its window still holds.
160///
161/// Read per landing rather than once per round, because a round appends
162/// as it goes and a node running with `--require-scope` admits only a
163/// *recent* head. A scope captured at the start would go stale in
164/// exactly the rounds that land the most.
165pub type ScopeSource = Box<dyn Fn() -> (ContentHash, Option<ContentHash>) + Send>;
166
167/// Lands by moving the branch the proposals were aimed at (D68).
168pub struct RefLanding {
169 key: Arc<ActorKey>,
170 refname: String,
171 scope: ScopeSource,
172 at: Option<ContentHash>,
173}
174
175impl RefLanding {
176 /// Lands on `refname`, CAS'd from `base` and moving forward from
177 /// each landing to the next.
178 ///
179 /// `base` is the value the round was read at. It is the first
180 /// landing's `prev`, which is what makes a push that beat the round
181 /// win: the op is refused, and the queue stops rather than
182 /// overwriting somebody's work with a merge computed against a
183 /// branch that has moved.
184 #[must_use]
185 pub fn new(key: Arc<ActorKey>, refname: String, base: &str, scope: ScopeSource) -> Self {
186 Self {
187 key,
188 refname,
189 scope,
190 at: ContentHash::from_git_oid(base),
191 }
192 }
193}
194
195impl Landing for RefLanding {
196 fn name(&self) -> &'static str {
197 "ref"
198 }
199
200 fn record(
201 &mut self,
202 handle: &SequencerHandle,
203 _change: &Change,
204 commit: ContentHash,
205 ) -> Result<(), String> {
206 let (node, head) = (self.scope)();
207 let payload = ViewOp::new(OpKind::SetRef {
208 name: self.refname.clone(),
209 commit: commit.clone(),
210 prev: self.at.clone(),
211 })
212 .in_scope(node, head)
213 .to_payload();
214 let sig = self.key.sign_submission(QUEUE_CHANNEL, &payload);
215 handle.try_submit(QUEUE_CHANNEL, payload, Some(sig))?;
216 // Only after the sequencer took it. Advancing on a refusal would
217 // make the next landing CAS against a value the log never held,
218 // turning one lost race into a round that can never land.
219 self.at = Some(commit);
220 Ok(())
221 }
222}