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