Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp
256 linesCodeBlameRaw
1//! Each session's inbox of coordination updates: what other sessions and
2//! the merge queue did that concerns it, kept until the session says it
3//! considered each one (reading never clears them).
4//!
5//! An update is written in the same D1 batch (one transaction) as the
6//! change it announces, guarded by an SQL condition that holds only once
7//! that change is made: a change is never kept without its updates, and
8//! updates never announce a change that was not made. Its id is derived
9//! from the change (an entity and its transition), so a retried or
10//! concurrent request that makes the same change delivers it once.
11
12use super::*;
13use crate::rows::NumberRow;
14use sha2::{Digest, Sha256};
15use worker::wasm_bindgen::JsValue;
16use worker::D1PreparedStatement;
17
18/// How long an acknowledged update is kept as a receipt, in days.
19pub(crate) const RECEIPT_DAYS: u64 = 30;
20
21/// Puts a `?2` event (data `?3`, at `?4`) of the repositories `?5` (a JSON
22/// list) in the inbox of each active session there it concerns. `?6` is a
23/// JSON list of the work item and pull request numbers it is about (one
24/// sequence of numbers in a workspace): a session attached or tied to one
25/// of them hears of it, and so does a session not on anything yet, which
26/// is still finding its way; an event about nothing in particular reaches
27/// every session. An overlap between two sessions names its sole
28/// recipient. A session is not told what it did itself (its own start or
29/// end, or the claims it made, released or handed off), only what others
30/// did and what happened to it. `guard`, numbered from `?7`, holds only
31/// once the change announced is made.
32pub(crate) fn deliver_sql(guard: &str) -> String {
33 format!(
34 "INSERT OR IGNORE INTO session_updates (id, session_id, kind, data, at)
35 SELECT ?1, s.id, ?2, ?3, ?4 FROM sessions s WHERE s.repo_id IN (SELECT value FROM json_each(?5)) AND s.state = 'active'
36 AND (CASE WHEN ?2 = 'overlap.detected' AND json_extract(?3, '$.sessionId') IS NOT NULL
37 THEN s.id = json_extract(?3, '$.sessionId')
38 ELSE json_array_length(?6) = 0
39 OR (s.item_id IS NULL AND s.pull IS NULL AND NOT EXISTS (SELECT 1 FROM session_links l WHERE l.session_id = s.id))
40 OR s.pull IN (SELECT value FROM json_each(?6))
41 OR s.item_id IN (SELECT w.id FROM work_items w
42 WHERE w.workspace = (SELECT rw.workspace FROM repo_workspaces rw WHERE rw.repo_id = s.repo_id)
43 AND w.number IN (SELECT value FROM json_each(?6)))
44 OR EXISTS (SELECT 1 FROM session_links l
45 WHERE l.session_id = s.id AND l.kind = 'pull' AND l.number IN (SELECT value FROM json_each(?6)))
46 OR EXISTS (SELECT 1 FROM session_links l JOIN work_items w ON w.id = l.item_id
47 WHERE l.session_id = s.id AND l.kind = 'item' AND w.number IN (SELECT value FROM json_each(?6)))
48 END)
49 AND (?2 NOT IN ('session.started', 'session.ended') OR s.id != COALESCE(json_extract(?3, '$.session.id'), ''))
50 AND (?2 NOT IN ('work.claimed', 'work.released', 'work.handoff_requested') OR s.id != COALESCE(json_extract(?3, '$.claim.sessionId'), ''))
51 AND ({guard})"
52 )
53}
54pub(super) const PENDING: &str = "SELECT id, kind, data, at FROM session_updates
55 WHERE session_id = ? AND acknowledged_at IS NULL ORDER BY at, id LIMIT 100";
56pub(super) const ACK: &str = "UPDATE session_updates SET acknowledged_at = COALESCE(acknowledged_at, ?)
57 WHERE session_id = ? AND id = ?";
58/// How many of the IDs (`?2`, a JSON list) name an update of session `?1`.
59pub(super) const KNOWN: &str = "SELECT COUNT(*) AS n FROM session_updates
60 WHERE session_id = ?1 AND id IN (SELECT value FROM json_each(?2))";
61/// Session `?1`'s receipts acknowledged before `?2`.
62pub(super) const PRUNE: &str = "DELETE FROM session_updates
63 WHERE session_id = ?1 AND acknowledged_at IS NOT NULL AND acknowledged_at < ?2";
64/// The payload of update `?1` as the change was finally made (`?2`), where
65/// it was delivered as predicted (`?3`).
66const CORRECT: &str = "UPDATE session_updates SET data = ?2 WHERE id = ?1 AND data = ?3";
67
68/// The id of the update announcing `kind` for the change `key` names.
69pub(crate) fn update_id(kind: &str, key: &str) -> String {
70 let digest = Sha256::digest(format!("{kind}\n{key}").as_bytes());
71 format!("upd_{}", &hex::encode(digest)[..26])
72}
73
74/// The work item and pull request numbers an event's `data` is about: its
75/// `number` and `item`, the work items a dependency change newly waits on,
76/// and the pull requests an integration failure names as suspects.
77pub(crate) fn subjects(data: &serde_json::Value) -> Vec<u32> {
78 let mut numbers: Vec<u32> = Vec::new();
79 for key in ["number", "item", "pull"] {
80 if let Some(n) = data.get(key).and_then(serde_json::Value::as_u64) {
81 numbers.push(n as u32);
82 }
83 }
84 for key in ["added", "suspects"] {
85 if let Some(list) = data.get(key).and_then(serde_json::Value::as_array) {
86 numbers.extend(list.iter().filter_map(serde_json::Value::as_u64).map(|n| n as u32));
87 }
88 }
89 numbers.retain(|n| *n > 0);
90 numbers.sort_unstable();
91 numbers.dedup();
92 numbers
93}
94
95/// An update written with the change it announces: which, and the payload
96/// it was written with.
97pub(crate) struct Sent {
98 id: String,
99 data: serde_json::Value,
100}
101
102#[derive(Deserialize)]
103struct UpdateRow { id: String, kind: String, data: String, at: String }
104
105impl Work {
106 /// The update announcing `kind` (`data`, as the change is expected to
107 /// leave things) for the change `key` names, to run in the same batch
108 /// as that change, after it. `guard` (with `binds` from `?7`) holds
109 /// only once the change is made. None for an event no session hears.
110 pub(crate) fn delivery(
111 &self,
112 kind: &'static str,
113 repo: &str,
114 data: serde_json::Value,
115 key: &str,
116 guard: &str,
117 binds: Vec<JsValue>,
118 ) -> Result<Option<(D1PreparedStatement, Sent)>> {
119 self.delivery_to(kind, &[repo.to_owned()], data, key, guard, binds)
120 }
121
122 /// [`delivery`](Work::delivery), to the sessions of every one of
123 /// `repos`: a work item's are all those it is linked to.
124 pub(crate) fn delivery_to(
125 &self,
126 kind: &'static str,
127 repos: &[String],
128 data: serde_json::Value,
129 key: &str,
130 guard: &str,
131 binds: Vec<JsValue>,
132 ) -> Result<Option<(D1PreparedStatement, Sent)>> {
133 if !COORDINATION_EVENTS.contains(&kind) {
134 return Ok(None);
135 }
136 let id = update_id(kind, key);
137 let mut all: Vec<JsValue> = vec![
138 id.as_str().into(),
139 kind.into(),
140 serde_json::to_string(&data)?.into(),
141 rfc3339(now_ms()).into(),
142 serde_json::to_string(repos)?.into(),
143 serde_json::to_string(&subjects(&data))?.into(),
144 ];
145 all.extend(binds);
146 let statement = self.db.prepare(deliver_sql(guard)).bind(&all)?;
147 Ok(Some((statement, Sent { id, data })))
148 }
149
150 /// After the batch that made the change: its event as it was finally
151 /// made (`actual`), published, and the inbox copy corrected where the
152 /// change turned out other than expected.
153 pub(crate) async fn delivered(
154 &self,
155 sent: Option<Sent>,
156 kind: &'static str,
157 repo: &str,
158 actor: Option<String>,
159 actual: serde_json::Value,
160 ) -> Result<()> {
161 if let Some(sent) = sent
162 && sent.data != actual
163 {
164 self.db
165 .prepare(CORRECT)
166 .bind(&[
167 sent.id.as_str().into(),
168 serde_json::to_string(&actual)?.into(),
169 serde_json::to_string(&sent.data)?.into(),
170 ])?
171 .run()
172 .await?;
173 }
174 self.publish_value(kind, repo, actor, actual).await
175 }
176
177 /// Publishes `kind` for a change already made elsewhere (a merge queue
178 /// attempt's result, a blocker that closed), its updates written under
179 /// `key`, which names that change: telling of the same change again,
180 /// as a retried report would, reaches each session once.
181 pub(crate) async fn publish_keyed(
182 &self,
183 kind: &'static str,
184 repo: &str,
185 actor: Option<String>,
186 data: serde_json::Value,
187 key: &str,
188 ) -> Result<()> {
189 if let Some((statement, _)) = self.delivery(kind, repo, data.clone(), key, "1", Vec::new())? {
190 statement.run().await?;
191 }
192 self.publish_value(kind, repo, actor, data).await
193 }
194
195 /// [`publish_keyed`](Work::publish_keyed) for the sessions of every one
196 /// of `repos`, the event filed under the first.
197 pub(crate) async fn publish_keyed_to(
198 &self,
199 kind: &'static str,
200 repos: &[String],
201 actor: Option<String>,
202 data: serde_json::Value,
203 key: &str,
204 ) -> Result<()> {
205 if let Some((statement, _)) = self.delivery_to(kind, repos, data.clone(), key, "1", Vec::new())? {
206 statement.run().await?;
207 }
208 self.publish_on(kind, repos.first().map(String::as_str), actor, data).await
209 }
210
211 /// An update for a change already made, written on its own: for the
212 /// changes whose updates are not yet written with them.
213 pub(crate) async fn deliver_update(&self, kind: &'static str, repo: &str, data: &serde_json::Value) -> Result<()> {
214 let key = g1t_contracts::new_id("upd", now_ms());
215 if let Some((statement, _)) = self.delivery(kind, repo, data.clone(), &key, "1", Vec::new())? {
216 statement.run().await?;
217 }
218 Ok(())
219 }
220
221 pub(super) async fn pending_updates(&self, session: &str) -> Result<Vec<SessionUpdate>> {
222 self.db.prepare(PENDING).bind(&[session.into()])?.all().await?.results::<UpdateRow>()?
223 .into_iter().map(|row| Ok(SessionUpdate {
224 id: row.id, kind: row.kind, data: serde_json::from_str(&row.data)?, at: row.at,
225 })).collect()
226 }
227
228 /// Acknowledges `ids` for `session` and returns how many of them name
229 /// one of its updates, so a caller can tell a mistyped ID from a
230 /// receipt. Receipts older than [`RECEIPT_DAYS`] go then.
231 pub(super) async fn acknowledge_updates(&self, session: &str, ids: &[String]) -> Result<u32> {
232 if ids.is_empty() { return Ok(0); }
233 let now = now_ms();
234 let at = rfc3339(now);
235 let mut statements = ids.iter().map(|id| self.db.prepare(ACK)
236 .bind(&[at.as_str().into(), session.into(), id.as_str().into()])).collect::<Result<Vec<_>>>()?;
237 let old = rfc3339(now.saturating_sub(RECEIPT_DAYS * 24 * 60 * 60 * 1000));
238 statements.push(self.db.prepare(PRUNE).bind(&[session.into(), old.into()])?);
239 self.db.batch(statements).await?;
240 let mut unique = ids.to_vec();
241 unique.sort();
242 unique.dedup();
243 let known = self.db.prepare(KNOWN).bind(&[session.into(), serde_json::to_string(&unique)?.into()])?
244 .first::<NumberRow>(None).await?;
245 Ok(known.map_or(0, |row| row.n))
246 }
247}
248
249/// Adds a delivery's statement to `statements`, the batch of the change it
250/// announces, and returns what was sent.
251pub(crate) fn push_delivery(statements: &mut Vec<D1PreparedStatement>, delivery: Option<(D1PreparedStatement, Sent)>) -> Option<Sent> {
252 delivery.map(|(statement, sent)| {
253 statements.push(statement);
254 sent
255 })
256}