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

Commit

Catch up on work.migrated pages that never arrived

Search, the inbox, integrations and security rewrite old issue numbers when they hear work.migrated. A page that went to the dead-letter queue was lost; each now also pages legacy_item_map once the move is done, applies the same rewrite and remembers where it got to.

G1T-Session:
ses_01m56adwzvenhby748vvbn96c5
syntaqxcommitted Parent96fc50bBrowse files
12 files+210−140/12 viewed
+74−0
632632 Ok(true)
633633 }
634634 }
635+
636+/// Catching up on `work.migrated` (issues became work items, g1t #57).
637+/// A service that rewrites what it kept by issue number when it hears
638+/// `work.migrated` also pages through the work service's
639+/// `legacy_item_map` on its cron, applying each page the same way, so a
640+/// page that never arrived (one left in the dead-letter queue) is not lost.
641+/// Where it got to is kept in the service's `work_catch_up` table (made by
642+/// its migration, one row named `work.migrated`), and once the last page is
643+/// applied it does nothing more. Every rewrite leaves what it already
644+/// rewrote alone, so a page applied twice changes nothing.
645+pub mod work_migrated {
646+ use std::future::Future;
647+
648+ use g1t_contracts::events::WorkMigrated;
649+ use g1t_contracts::time::rfc3339;
650+ use g1t_contracts::work::{LegacyItemMapArgs, LegacyItemPage};
651+ use serde::Deserialize;
652+ use worker::wasm_bindgen::JsValue;
653+ use worker::{D1Database, Fetcher, Result};
654+
655+ /// Moves asked for at a time, and pages applied in one run.
656+ pub const PAGE: u32 = 250;
657+ pub const PAGES_PER_RUN: u32 = 20;
658+
659+ #[derive(Deserialize)]
660+ struct Row {
661+ #[serde(default)]
662+ cursor: Option<String>,
663+ page: u32,
664+ #[serde(default)]
665+ done_at: Option<String>,
666+ }
667+
668+ /// One more stretch of the catch-up: each page read is handed to
669+ /// `apply` as the `work.migrated` it would have been. Whether it is done.
670+ /// It waits while the work service is still moving issues.
671+ pub async fn catch_up<F, Fut>(db: &D1Database, work: &Fetcher, mut apply: F) -> Result<bool>
672+ where
673+ F: FnMut(WorkMigrated) -> Fut,
674+ Fut: Future<Output = Result<()>>,
675+ {
676+ let row = db
677+ .prepare("SELECT cursor, page, done_at FROM work_catch_up WHERE name = 'work.migrated'")
678+ .first::<Row>(None)
679+ .await?;
680+ let Some(row) = row else { return Ok(true) };
681+ if row.done_at.is_some() {
682+ return Ok(true);
683+ }
684+ let (mut cursor, mut page) = (row.cursor, row.page);
685+ for _ in 0..PAGES_PER_RUN {
686+ let found: LegacyItemPage =
687+ crate::call(work, "legacy_item_map", &LegacyItemMapArgs { after: cursor.clone(), limit: PAGE }).await?;
688+ if !found.moved {
689+ return Ok(false);
690+ }
691+ page += 1;
692+ let done = found.next.is_none();
693+ if !found.items.is_empty() {
694+ apply(WorkMigrated { page, items: found.items, done }).await?;
695+ }
696+ cursor = found.next;
697+ let done_at = if done { JsValue::from(rfc3339(crate::now_ms())) } else { JsValue::NULL };
698+ db.prepare("UPDATE work_catch_up SET cursor = ?1, page = ?2, done_at = ?3 WHERE name = 'work.migrated'")
699+ .bind(&[cursor.as_deref().map_or(JsValue::NULL, JsValue::from), page.into(), done_at])?
700+ .run()
701+ .await?;
702+ if done {
703+ return Ok(true);
704+ }
705+ }
706+ Ok(false)
707+ }
708+}
+10−0
1+-- Where catching up on `work.migrated` got to (g1t_kit::work_migrated): the
2+-- cron pages through the work service's `legacy_item_map` and applies each
3+-- page as the event would have, until the last.
4+CREATE TABLE work_catch_up (
5+ name TEXT PRIMARY KEY,
6+ cursor TEXT,
7+ page INTEGER NOT NULL DEFAULT 0,
8+ done_at TEXT
9+);
10+INSERT INTO work_catch_up (name) VALUES ('work.migrated');
+5−1
388388 /// any workspace keeps (`AUDIT_MAX_DAYS`, 400 days), then each workspace's
389389 /// entries older than its own plan keeps, as billing says (7 days free, 90
390390 /// on the plan, or what staff set). Workspaces with nothing older than the
391−/// shortest (`AUDIT_MIN_DAYS`, 7) are left alone.
391+/// shortest (`AUDIT_MIN_DAYS`, 7) are left alone. Each cron also catches up
392+/// on pages of `work.migrated` that never arrived (work_items.rs).
392393 #[event(scheduled)]
393394 async fn scheduled(event: worker::ScheduledEvent, env: Env, _ctx: worker::ScheduleContext) {
395+ if let (Ok(db), Ok(work)) = (env.d1("DB"), env.service("WORK")) {
396+ work_items::catch_up(&db, &work).await;
397+ }
394398 // The morning cron sends the security digests that are due, and nothing else.
395399 if event.cron() == security_digest::DIGEST_CRON {
396400 let (Ok(db), Ok(identity)) = (env.d1("DB"), env.service("IDENTITY")) else { return };
+34−3
1414 //! and pull request events) name the work item and its number.
1515 //!
1616 //! Every statement leaves what it already rewrote alone, so a page
17−//! delivered again, or out of order, changes nothing more.
17+//! delivered again, or out of order, changes nothing more. The cron
18+//! catches up on pages that never arrived ([`catch_up`]).
1819
1920 use g1t_contracts::events::{Event, WorkMigrated};
20−use worker::{D1Database, D1PreparedStatement, Result};
21+use worker::{D1Database, D1PreparedStatement, Fetcher, Result};
2122
2223 /// The issue events of the log, by their old names.
2324 pub const ISSUE_EVENTS: [&str; 9] = [
128129 if event.data["truncated"].as_bool() == Some(true) {
129130 worker::console_error!("work.migrated {} (page {}) was cut short: some issue numbers are left as they were", event.id, page.page);
130131 }
132+ page_statements(db, &event.id, &page)
133+}
134+
135+/// The statements for one page, recorded as `id`'s.
136+fn page_statements(db: &D1Database, id: &str, page: &WorkMigrated) -> Result<Vec<D1PreparedStatement>> {
131137 if page.items.is_empty() {
132138 return Ok(Vec::new());
133139 }
134140 let items = serde_json::to_string(&page.items)?;
135141 let issue_events = serde_json::to_string(&ISSUE_EVENTS)?;
136− let values: [worker::wasm_bindgen::JsValue; 3] = [event.id.as_str().into(), items.as_str().into(), issue_events.as_str().into()];
142+ let values: [worker::wasm_bindgen::JsValue; 3] = [id.into(), items.as_str().into(), issue_events.as_str().into()];
137143 MOVED.iter().map(|sql| db.prepare(*sql).bind(&values[..bound(sql)])).collect()
138144 }
139145
146+/// The cron: pages of `work.migrated` that never arrived, read from the
147+/// work service (`g1t_kit::work_migrated`) and applied as the event's
148+/// would have been, each recorded as `catch-up:<page>`.
149+pub async fn catch_up(db: &D1Database, work: &Fetcher) {
150+ let caught = g1t_kit::work_migrated::catch_up(db, work, |page| async move {
151+ let statements = page_statements(db, &format!("catch-up:{}", page.page), &page)?;
152+ if !statements.is_empty() {
153+ db.batch(statements).await?;
154+ }
155+ Ok(())
156+ })
157+ .await;
158+ if let Err(error) = caught {
159+ worker::console_error!("catching up on work.migrated: {error}");
160+ }
161+}
162+
140163 #[cfg(test)]
141164 mod tests {
142165 use super::*;
247270 let first = items(&db);
248271 run(&db, "evt_9", &page());
249272 assert_eq!(items(&db), first);
273+ // And again as the cron's catch-up reads it: nothing more either.
274+ let log = |db: &Connection| -> Vec<(String, serde_json::Value)> {
275+ ["evt_1", "evt_2", "evt_3", "evt_5", "evt_6", "evt_7"].iter().map(|id| logged(db, id)).collect()
276+ };
277+ let logged_once = log(&db);
278+ run(&db, "catch-up:1", &page());
279+ assert_eq!(items(&db), first);
280+ assert_eq!(log(&db), logged_once);
250281 assert_eq!(
251282 first,
252283 vec![
+10−0
1+-- Where catching up on `work.migrated` got to (g1t_kit::work_migrated): the
2+-- cron pages through the work service's `legacy_item_map` and applies each
3+-- page as the event would have, until the last.
4+CREATE TABLE work_catch_up (
5+ name TEXT PRIMARY KEY,
6+ cursor TEXT,
7+ page INTEGER NOT NULL DEFAULT 0,
8+ done_at TEXT
9+);
10+INSERT INTO work_catch_up (name) VALUES ('work.migrated');
+3−1
11261126 }
11271127
11281128 /// Every minute: the sessions started on workspaces' Anthropic accounts
1129−/// that are still going are followed (agent_sessions.rs).
1129+/// that are still going are followed (agent_sessions.rs), and pages of
1130+/// `work.migrated` that never arrived are caught up on (work_migrated.rs).
11301131 #[event(scheduled)]
11311132 async fn scheduled(_event: worker::ScheduledEvent, env: Env, _ctx: worker::ScheduleContext) {
11321133 if let Ok(service) = Integrations::new(&env) {
11331134 agent_sessions::scheduled(&service).await;
11341135 summon::forget_old(&service).await;
1136+ work_migrated::catch_up(&service.db, &service.work).await;
11351137 }
11361138 }
11371139
+32−8
33 //! (`legacy_repo_id`, `legacy_number`, from migrations/0013_work_items.sql)
44 //! take the work item's id, workspace and number, and lose the old pair.
55 //! Each page is applied on its own, and a page seen again changes nothing.
6+//! The cron catches up on pages that never arrived ([`catch_up`]).
67
78 use g1t_contracts::events::{Event, WorkMigrated};
8−use worker::{D1Database, Result};
9+use g1t_contracts::work::MigratedItem;
10+use worker::{D1Database, Fetcher, Result};
911
1012 /// The page's moves, from `?1` (its `items` as JSON), as rows.
1113 const MOVES: &str = "SELECT json_extract(value, '$.repoId') AS repo_id, json_extract(value, '$.legacyNumber') AS legacy_number,
3739 let Ok(page) = serde_json::from_value::<WorkMigrated>(event.data.clone()) else {
3840 return Ok(true);
3941 };
40− if page.items.is_empty() {
41− return Ok(true);
42+ apply(db, &page.items).await?;
43+ Ok(true)
44+}
45+
46+/// Applies one page's moves.
47+async fn apply(db: &D1Database, moves: &[MigratedItem]) -> Result<()> {
48+ if moves.is_empty() {
49+ return Ok(());
4250 }
43− let items = serde_json::to_string(&page.items)?;
51+ let items = serde_json::to_string(moves)?;
4452 let mut batch = Vec::new();
4553 for sql in statements() {
4654 batch.push(db.prepare(&sql).bind(&[items.as_str().into()])?);
4755 }
4856 db.batch(batch).await?;
49− Ok(true)
57+ Ok(())
58+}
59+
60+/// The cron: pages of `work.migrated` that never arrived, read from the
61+/// work service (`g1t_kit::work_migrated`), until the last.
62+pub async fn catch_up(db: &D1Database, work: &Fetcher) {
63+ if let Err(error) = g1t_kit::work_migrated::catch_up(db, work, |page| async move { apply(db, &page.items).await }).await {
64+ worker::console_error!("catching up on work.migrated: {error}");
65+ }
5066 }
5167
5268 #[cfg(test)]
123139 MigratedItem { repo_id: "rep_api".into(), legacy_number: 4, workspace: "acme".into(), number: 32, item_id: "iss_api4".into() },
124140 MigratedItem { repo_id: "rep_web".into(), legacy_number: 5, workspace: "acme".into(), number: 33, item_id: "iss_web5".into() },
125141 ];
126− // Twice: a page seen again changes nothing.
127− for _ in 0..2 {
142+ // Part of it as the event's page, then all of it twice as the
143+ // catch-up's pages: a move applied again changes nothing.
144+ for moves in [&moved[..1], &moved[..], &moved[..]] {
128145 for sql in statements() {
129− db.execute(&sql, params![page(&moved)]).unwrap();
146+ db.execute(&sql, params![page(moves)]).unwrap();
130147 }
131148 }
132149 let link = |id: &str| -> (String, Option<String>, Option<u32>, Option<String>, Option<u32>) {
150167 )
151168 .unwrap();
152169 assert_eq!(left, 0);
170+ // Where the catch-up starts.
171+ let start: (Option<String>, u32, Option<String>) = db
172+ .query_row("SELECT cursor, page, done_at FROM work_catch_up WHERE name = 'work.migrated'", [], |row| {
173+ Ok((row.get(0)?, row.get(1)?, row.get(2)?))
174+ })
175+ .unwrap();
176+ assert_eq!(start, (None, 0, None));
153177 }
154178 }
+10−0
1+-- Where catching up on `work.migrated` got to (g1t_kit::work_migrated): the
2+-- cron pages through the work service's `legacy_item_map` and applies each
3+-- page as the event would have, until the last.
4+CREATE TABLE work_catch_up (
5+ name TEXT PRIMARY KEY,
6+ cursor TEXT,
7+ page INTEGER NOT NULL DEFAULT 0,
8+ done_at TEXT
9+);
10+INSERT INTO work_catch_up (name) VALUES ('work.migrated');
+7−0
280280 g1t_kit::call(&self.repos, "path_by_id", &PathByIdArgs { id: repo_id.to_owned() }).await
281281 }
282282
283+ /// Pages of `work.migrated` that never arrived, read from the work
284+ /// service (`g1t_kit::work_migrated`): each page's work items are read
285+ /// again, as the event's would have been. Whether it is done.
286+ pub async fn catch_up_on_moves(&self) -> Result<bool> {
287+ g1t_kit::work_migrated::catch_up(&self.db, &self.work, |page| async move { self.enqueue(migrated_jobs(&page)).await }).await
288+ }
289+
283290 pub async fn enqueue(&self, jobs: Vec<Job>) -> Result<()> {
284291 let queue = self.env.queue("JOBS")?;
285292 for batch in jobs.chunks(100) {
+5−1
122122 if let Err(error) = search.ensure_backfill().await {
123123 worker::console_error!("search: could not start the backfill: {error}");
124124 }
125− // Pages parked while paused, looked for at most every few minutes.
125+ // Pages parked while paused, looked for at most every few minutes,
126+ // and pages of `work.migrated` that never arrived.
126127 if resume_due(g1t_kit::now_ms()) {
127128 match search.resume_parked().await {
128129 Ok(0) => {}
129130 Ok(found) => worker::console_log!("search: resumed {found} parked backfill pages"),
130131 Err(error) => worker::console_error!("search: could not resume parked backfill pages: {error}"),
131132 }
133+ if let Err(error) = search.catch_up_on_moves().await {
134+ worker::console_error!("search: catching up on work.migrated failed: {error}");
135+ }
132136 }
133137 }
134138 for message in batch.messages()? {
+10−0
1+-- Where catching up on `work.migrated` got to (g1t_kit::work_migrated): the
2+-- cron pages through the work service's `legacy_item_map` and applies each
3+-- page as the event would have, until the last.
4+CREATE TABLE work_catch_up (
5+ name TEXT PRIMARY KEY,
6+ cursor TEXT,
7+ page INTEGER NOT NULL DEFAULT 0,
8+ done_at TEXT
9+);
10+INSERT INTO work_catch_up (name) VALUES ('work.migrated');
+10−0
655655 Ok(())
656656 }
657657
658+ /// Pages of `work.migrated` that never arrived, read from the work
659+ /// service (`g1t_kit::work_migrated`), until the last. Whether it is done.
660+ async fn catch_up_on_moves(&self) -> Result<bool> {
661+ let store = &self.store;
662+ g1t_kit::work_migrated::catch_up(&store.db, &self.work, |page| async move { store.work_migrated(&page.items).await }).await
663+ }
664+
658665 /// The sweep: continues history scans and scans of pushes that landed
659666 /// unscanned, catches security updates whose sandbox never pushed, and
660667 /// reads dependencies that have not been read for a day.
804811 if let Err(error) = security.sweep().await {
805812 worker::console_error!("security: the sweep failed: {error}");
806813 }
814+ if let Err(error) = security.catch_up_on_moves().await {
815+ worker::console_error!("security: catching up on work.migrated failed: {error}");
816+ }
807817 }
808818 Err(error) => worker::console_error!("security: could not start: {error}"),
809819 }

Comments

Sign in to comment.