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