Skip to main content

max / makenotwork

13.9 KB · 353 lines History Blame Raw
1 //! The migration dry run: prove every pending migration applies to a restored
2 //! copy of what production is actually holding, before a deploy makes that
3 //! irreversible.
4
5 use super::GateCtx;
6 use super::log::GateLog;
7 use super::pg::{pg_create_db, pg_url_with_dbname, reset_scratch, restore_dump, run_migrator};
8 use crate::classify;
9 use crate::domain::{GateKind, GateRunId};
10 use crate::outcome::{GateBlocker, GateFailure, GateOutcome, PassNote};
11 use anyhow::Result;
12 use chrono::Utc;
13
14 pub(super) async fn migration_dry_run(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
15 let log = GateLog::open(ctx, run_id, GateKind::MigrationDryRun).await;
16 let outcome = migration_dry_run_inner(ctx, &log).await;
17 log.close().await;
18 outcome.map(|o| o.with_log_ref(ctx.log_ref(GateKind::MigrationDryRun)))
19 }
20
21 /// The staged interior of [`migration_dry_run`], writing every step through the
22 /// gate's live log. The caller owns the sink so it can flush it on every exit
23 /// path, and attaches the `log_ref` once instead of at each return.
24 /// Runs one configured check per database, in config order, and stops at the
25 /// first that does not pass — a red gate is a red gate, and continuing would
26 /// bury it under a second restore's output.
27 ///
28 /// The server's check runs against `scratch_db_url` itself and is deliberately
29 /// last-writer for it: `cargo_test` reuses that database in migrated state, so
30 /// every other check must name its own `scratch_db` (enforced at config load).
31 async fn migration_dry_run_inner(ctx: &GateCtx, log: &GateLog) -> Result<GateOutcome> {
32 let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() else {
33 log.line("scratch_db_url unset in daemon config\n").await;
34 return Ok(GateOutcome::blocked(GateBlocker::ScratchDbUrlUnset));
35 };
36
37 let mut checked = Vec::new();
38 let mut primary_backup_path = String::new();
39 for check in &ctx.cfg.migration_checks {
40 let label = check.dir.display().to_string();
41 log.line(&format!("==== migration_check: {label} ====\n"))
42 .await;
43 match run_migration_check(ctx, log, scratch_url, check).await? {
44 CheckResult::Passed { backup_path } => {
45 if primary_backup_path.is_empty() {
46 primary_backup_path = backup_path;
47 }
48 checked.push(label);
49 }
50 CheckResult::Stopped(outcome) => return Ok(outcome),
51 }
52 }
53
54 log.line(&format!(
55 "all {} migration check(s) passed: {}",
56 checked.len(),
57 checked.join(", ")
58 ))
59 .await;
60 Ok(GateOutcome::passed(PassNote::Migrated {
61 backup_path: primary_backup_path,
62 checks: checked,
63 }))
64 }
65
66 /// One check's verdict: it passed (against `backup_path`), or it produced the
67 /// outcome the whole gate reports.
68 enum CheckResult {
69 Passed { backup_path: String },
70 Stopped(GateOutcome),
71 }
72
73 /// Restore one database's dump into its scratch DB and run its migrations on top.
74 async fn run_migration_check(
75 ctx: &GateCtx,
76 log: &GateLog,
77 scratch_url: &str,
78 check: &crate::config::MigrationCheck,
79 ) -> Result<CheckResult> {
80 let label = check.dir.display().to_string();
81
82 let backup: Option<(String, String)> = sqlx::query_as(
83 "SELECT local_path, fetched_at FROM backups
84 WHERE app = ? AND name = ? ORDER BY id DESC LIMIT 1",
85 )
86 .bind(&ctx.cfg.id)
87 .bind(&check.backup)
88 .fetch_optional(&ctx.pool)
89 .await?;
90 let Some((backup_path, fetched_at)) = backup else {
91 log.line(&format!(
92 "no {} backup fetched; call /backup/fetch first\n",
93 check.backup
94 ))
95 .await;
96 return Ok(CheckResult::Stopped(GateOutcome::blocked(
97 GateBlocker::NoBackupAvailable {
98 check: label,
99 backup: check.backup.clone(),
100 },
101 )));
102 };
103
104 // Presence is not freshness. A fetch that quietly stopped working leaves this
105 // row in place, and restoring it dry-runs the migrations against a schema prod
106 // has moved past — green, and worthless. Block on age instead. An unparsable
107 // timestamp is treated as stale: this row is daemon-written RFC 3339, so a
108 // value that will not parse means something is wrong, and failing closed on a
109 // freshness check is the whole point.
110 let age_hours = chrono::DateTime::parse_from_rfc3339(&fetched_at).map_or(i64::MAX, |t| {
111 (Utc::now() - t.with_timezone(&Utc)).num_hours()
112 });
113 let max_age_hours = ctx.cfg.backup_max_age_hours;
114 if age_hours > i64::from(max_age_hours) {
115 let msg = format!(
116 "backup {backup_path} was fetched {fetched_at} ({age_hours}h ago, max \
117 {max_age_hours}h); re-run /backup/fetch\n"
118 );
119 log.line(&msg).await;
120 return Ok(CheckResult::Stopped(GateOutcome::blocked(
121 GateBlocker::BackupStale {
122 age_hours,
123 max_age_hours,
124 check: label,
125 },
126 )));
127 }
128
129 // A check with its own `scratch_db` gets that database created here rather
130 // than by a host bootstrap step: adding a `[[migration_check]]` should not
131 // silently depend on someone having remembered to `createdb` on the Sando
132 // host, which is exactly the class of footgun this gate exists to remove.
133 // DROP + CREATE also makes the database sando-owned, so the PG15+ public
134 // schema grants `reset_scratch` applies next are the owner's to give.
135 let db_url = match check.scratch_db.as_deref() {
136 None => scratch_url.to_string(),
137 Some(dbname) => {
138 let maintenance_url = pg_url_with_dbname(scratch_url, "postgres");
139 log.line(&format!("---- create scratch db {dbname} ----\n"))
140 .await;
141 if let Err(e) = pg_create_db(&maintenance_url, dbname).await {
142 let msg = format!("{label}: creating scratch db {dbname}: {e}");
143 log.line(&msg).await;
144 return Ok(CheckResult::Stopped(GateOutcome::failed(
145 GateFailure::RestoreFailed { reason: msg },
146 )));
147 }
148 pg_url_with_dbname(scratch_url, dbname)
149 }
150 };
151
152 let owner_role = check
153 .owner_role
154 .as_deref()
155 .unwrap_or(&ctx.cfg.scratch_owner_role);
156 log.line("---- reset_scratch ----\n").await;
157 if let Err(e) = reset_scratch(&db_url, owner_role).await {
158 let msg = format!("{label}: scratch reset: {e}");
159 log.line(&msg).await;
160 return Ok(CheckResult::Stopped(GateOutcome::failed(
161 GateFailure::RestoreFailed { reason: msg },
162 )));
163 }
164 log.line(&format!("---- restore_dump ({backup_path}) ----\n"))
165 .await;
166 if let Err(e) = restore_dump(&db_url, &backup_path, log).await {
167 let msg = format!("{label}: restore: {e}");
168 log.line(&msg).await;
169 return Ok(CheckResult::Stopped(GateOutcome::failed(
170 GateFailure::RestoreFailed { reason: msg },
171 )));
172 }
173
174 let Some(migrations_dir) = ctx.migrations_dir(&check.dir) else {
175 // Neither the bundle nor a checkout holds them. For an accepted
176 // artifact that means the builder did not ship its migrations, and a
177 // dry run over nothing would report green having proved nothing.
178 let msg = format!(
179 "{label}: no migrations at {} in the bundle or a checkout",
180 check.dir.display()
181 );
182 log.line(&msg).await;
183 return Ok(CheckResult::Stopped(GateOutcome::failed(
184 GateFailure::RestoreFailed { reason: msg },
185 )));
186 };
187 log.line("---- run_migrator ----\n").await;
188 match run_migrator(&db_url, &migrations_dir).await {
189 Ok(()) => {
190 log.line(&format!("{label}: restored {backup_path} + migrated\n"))
191 .await;
192 Ok(CheckResult::Passed { backup_path })
193 }
194 Err(e) => {
195 let err_s = format!("{label}: {e}");
196 log.line(&err_s).await;
197 Ok(CheckResult::Stopped(GateOutcome::failed(
198 classify::classify_migration_error(&err_s, None),
199 )))
200 }
201 }
202 }
203
204 #[cfg(test)]
205 mod tests {
206 use super::*;
207 use crate::gates::testkit::{
208 dry_run_ctx, mt_check, seed_backup, seed_named_backup, with_check,
209 };
210
211 #[tokio::test]
212 async fn migration_dry_run_blocks_when_a_checks_own_dump_was_never_fetched() {
213 // The hazard the check list exists for: multithreaded applies its own
214 // migrations at boot against its own database, so the server's dump must
215 // never stand in for it. A fetched `server` row with no `multithreaded`
216 // row is exactly that substitution, and it has to block.
217 let tmp = tempfile::tempdir().unwrap();
218 let mut ctx = dry_run_ctx(tmp.path(), 48).await;
219 with_check(&mut ctx, mt_check());
220 seed_named_backup(&ctx, "server", 1).await;
221 let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;
222
223 let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
224 log.close().await;
225
226 let crate::outcome::GateStatus::Blocked { blocker } = outcome.status else {
227 panic!("a missing multithreaded dump must block");
228 };
229 let GateBlocker::NoBackupAvailable { check, backup } = blocker else {
230 panic!("expected NoBackupAvailable, got {blocker:?}");
231 };
232 assert_eq!(backup, "multithreaded", "names the dump that is missing");
233 assert!(
234 check.contains("multithreaded/migrations"),
235 "names the check that wanted it, got {check}"
236 );
237 }
238
239 #[tokio::test]
240 async fn migration_dry_run_freshness_is_per_dump() {
241 // A fresh server dump must not make a 45-day-old multithreaded dump look
242 // current: the clock is per-database, or the second check inherits the
243 // first's freshness and the gate is theatre.
244 let tmp = tempfile::tempdir().unwrap();
245 let mut ctx = dry_run_ctx(tmp.path(), 48).await;
246 with_check(&mut ctx, mt_check());
247 seed_named_backup(&ctx, "server", 1).await;
248 seed_named_backup(&ctx, "multithreaded", 24 * 45).await;
249 let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;
250
251 let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
252 log.close().await;
253
254 let crate::outcome::GateStatus::Blocked { blocker } = outcome.status else {
255 panic!("a 45-day-old multithreaded dump must block");
256 };
257 let GateBlocker::BackupStale { check, .. } = blocker else {
258 panic!("expected BackupStale, got {blocker:?}");
259 };
260 assert!(
261 check.contains("multithreaded/migrations"),
262 "names the check whose dump is stale, got {check}"
263 );
264 }
265
266 #[tokio::test]
267 async fn migration_dry_run_blocks_on_a_stale_backup() {
268 // The failure this closes: the gate used to check only that a backups row
269 // existed, so a fetch that silently stopped working left it green against
270 // an ever-older schema. Sando ran 45 days that way.
271 let tmp = tempfile::tempdir().unwrap();
272 let ctx = dry_run_ctx(tmp.path(), 48).await;
273 seed_backup(&ctx, 24 * 45).await;
274 let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;
275
276 let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
277 log.close().await;
278
279 let crate::outcome::GateStatus::Blocked { blocker } = outcome.status else {
280 panic!("a 45-day-old backup must block");
281 };
282 let GateBlocker::BackupStale {
283 age_hours,
284 max_age_hours,
285 ..
286 } = blocker
287 else {
288 panic!("expected BackupStale, got {blocker:?}");
289 };
290 assert_eq!(max_age_hours, 48);
291 assert!(
292 age_hours >= 24 * 45,
293 "reports the real age, got {age_hours}"
294 );
295 }
296
297 #[tokio::test]
298 async fn migration_dry_run_accepts_a_fresh_backup() {
299 // The other side of the boundary: a backup inside the window must not be
300 // blocked on freshness. It fails later (there is no such dump on disk),
301 // which is exactly the proof the age check let it through.
302 let tmp = tempfile::tempdir().unwrap();
303 let ctx = dry_run_ctx(tmp.path(), 48).await;
304 seed_backup(&ctx, 6).await;
305 let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;
306
307 let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
308 log.close().await;
309
310 assert!(
311 !matches!(
312 outcome.status,
313 crate::outcome::GateStatus::Blocked {
314 blocker: GateBlocker::BackupStale { .. }
315 }
316 ),
317 "a 6h-old backup is fresh, got {:?}",
318 outcome.status,
319 );
320 }
321
322 #[tokio::test]
323 async fn migration_dry_run_treats_an_unparsable_fetched_at_as_stale() {
324 // Fail closed: `fetched_at` is daemon-written RFC 3339, so a value that
325 // will not parse means the row is untrustworthy — and a freshness check
326 // that shrugs at a timestamp it cannot read is not a freshness check.
327 let tmp = tempfile::tempdir().unwrap();
328 let ctx = dry_run_ctx(tmp.path(), 48).await;
329 sqlx::query(
330 "INSERT INTO backups (fetched_at, source, local_path, byte_size)
331 VALUES ('not-a-timestamp', 'file:///x.sql.gz', '/tmp/x.sql.gz', 1000000)",
332 )
333 .execute(&ctx.pool)
334 .await
335 .unwrap();
336 let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;
337
338 let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
339 log.close().await;
340
341 assert!(
342 matches!(
343 outcome.status,
344 crate::outcome::GateStatus::Blocked {
345 blocker: GateBlocker::BackupStale { .. }
346 }
347 ),
348 "an unreadable fetched_at must block, got {:?}",
349 outcome.status,
350 );
351 }
352 }
353