Skip to main content

max / balanced_breakfast

11.6 KB · 329 lines History Blame Raw
1 //! The schema migration runner.
2 //!
3 //! # Why the ledger table is called `_sqlx_migrations`
4 //!
5 //! Every Balanced Breakfast install in the field carries that table, one row per
6 //! applied migration, written by `sqlx::migrate!`. Renaming it, or writing a
7 //! fresh ledger beside it, makes an upgraded install believe it has applied
8 //! nothing and re-run every migration against a populated database, which fails
9 //! at the first `CREATE TABLE`.
10 //!
11 //! So this runner adopts that ledger verbatim: same table name, same columns,
12 //! same semantics, and the same checksum function (sha384 over the raw file
13 //! bytes, as sqlx 0.9 computes it). An upgraded install reads its own rows,
14 //! finds nothing pending, and does nothing. The checksum algorithm is not a free
15 //! choice.
16
17 use std::time::Instant;
18
19 use rusqlite::{Connection, OptionalExtension};
20 use sha2::{Digest, Sha384};
21
22 include!(concat!(env!("OUT_DIR"), "/migrations.rs"));
23
24 /// The ledger table. Named by sqlx; see the module docs before changing it.
25 const LEDGER: &str = "_sqlx_migrations";
26
27 /// A migration that failed to record itself needs a human, not a retry.
28 #[derive(Debug, thiserror::Error)]
29 pub enum MigrateError {
30 #[error("database error running migrations: {0}")]
31 Db(#[from] rusqlite::Error),
32
33 #[error("could not check out a connection to run migrations: {0}")]
34 Pool(#[from] r2d2::Error),
35
36 #[error(
37 "migration {version} ({description}) was already applied, but its file has changed since. \
38 Applied migrations are immutable -- add a new migration instead of editing a shipped one."
39 )]
40 ChecksumMismatch { version: i64, description: String },
41
42 #[error(
43 "migration {0} is partially applied; fix it and remove its row from the `{LEDGER}` table"
44 )]
45 Dirty(i64),
46
47 #[error("migration {version} ({description}) failed: {source}")]
48 Apply {
49 version: i64,
50 description: String,
51 #[source]
52 source: rusqlite::Error,
53 },
54 }
55
56 /// sha384 of a migration's bytes, matching what sqlx recorded.
57 pub(crate) fn checksum(sql: &str) -> Vec<u8> {
58 Sha384::digest(sql.as_bytes()).to_vec()
59 }
60
61 /// Every migration compiled into this binary, in version order.
62 pub(crate) fn all() -> impl Iterator<Item = (i64, &'static str, &'static str)> {
63 MIGRATIONS.iter().copied()
64 }
65
66 /// Create the ledger if absent. The DDL is sqlx 0.9's verbatim, so this is a
67 /// no-op on an install that sqlx already set up.
68 fn ensure_ledger(conn: &Connection) -> Result<(), rusqlite::Error> {
69 conn.execute_batch(&format!(
70 "CREATE TABLE IF NOT EXISTS {LEDGER} (
71 version BIGINT PRIMARY KEY,
72 description TEXT NOT NULL,
73 installed_on TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
74 success BOOLEAN NOT NULL,
75 checksum BLOB NOT NULL,
76 execution_time BIGINT NOT NULL
77 );"
78 ))
79 }
80
81 /// Apply every migration not yet in the ledger.
82 ///
83 /// Each migration and its ledger row commit together, so a crash mid-run leaves
84 /// the database at a migration boundary rather than half-applied.
85 #[tracing::instrument(skip_all)]
86 pub fn run(conn: &mut Connection) -> Result<(), MigrateError> {
87 ensure_ledger(conn)?;
88
89 // A `success = false` row means a previous run died between applying a
90 // migration and committing its ledger row. sqlx wrote these; this runner
91 // cannot (it commits both together), but an install upgraded from sqlx may
92 // carry one, and it still needs a human.
93 let dirty: Option<i64> = conn
94 .query_row(
95 &format!("SELECT version FROM {LEDGER} WHERE success = false ORDER BY version LIMIT 1"),
96 [],
97 |row| row.get(0),
98 )
99 .optional()?;
100 if let Some(version) = dirty {
101 return Err(MigrateError::Dirty(version));
102 }
103
104 let applied: std::collections::BTreeMap<i64, Vec<u8>> = {
105 let mut stmt = conn.prepare(&format!(
106 "SELECT version, checksum FROM {LEDGER} ORDER BY version"
107 ))?;
108 let rows = stmt.query_map([], |row| {
109 Ok((row.get::<_, i64>(0)?, row.get::<_, Vec<u8>>(1)?))
110 })?;
111 rows.collect::<Result<_, _>>()?
112 };
113
114 for (version, description, sql) in all() {
115 let digest = checksum(sql);
116
117 if let Some(recorded) = applied.get(&version) {
118 // Already applied. The only question left is whether the file still
119 // hashes to what was recorded; if not, a shipped migration was
120 // edited and every install that ran it now disagrees with this one.
121 if *recorded != digest {
122 return Err(MigrateError::ChecksumMismatch {
123 version,
124 description: description.to_owned(),
125 });
126 }
127 continue;
128 }
129
130 tracing::info!(version, description, "applying migration");
131 let started = Instant::now();
132 let tx = conn.transaction()?;
133 tx.execute_batch(sql)
134 .map_err(|source| MigrateError::Apply {
135 version,
136 description: description.to_owned(),
137 source,
138 })?;
139 tx.execute(
140 &format!(
141 "INSERT INTO {LEDGER} (version, description, success, checksum, execution_time)
142 VALUES (?1, ?2, TRUE, ?3, ?4)"
143 ),
144 rusqlite::params![
145 version,
146 description,
147 digest,
148 i64::try_from(started.elapsed().as_nanos()).unwrap_or(i64::MAX),
149 ],
150 )?;
151 tx.commit()?;
152 }
153
154 Ok(())
155 }
156
157 #[cfg(test)]
158 mod tests {
159 use super::*;
160
161 #[test]
162 fn every_migration_is_embedded() {
163 let versions: Vec<i64> = all().map(|(v, _, _)| v).collect();
164 assert!(!versions.is_empty(), "no migrations were embedded");
165 assert!(
166 versions.windows(2).all(|w| w[0] < w[1]),
167 "migrations are not in strictly increasing version order: {versions:?}"
168 );
169 }
170
171 #[test]
172 fn description_matches_sqlx_filename_parsing() {
173 // sqlx turns `004_feed_tags.sql` into the description "feed tags"; the
174 // ledger rows in the field carry that spelling.
175 let (_, description, _) = all().find(|(v, _, _)| *v == 4).expect("migration 004");
176 assert_eq!(description, "feed tags");
177 }
178
179 #[test]
180 fn running_twice_is_a_no_op() {
181 let mut conn = Connection::open_in_memory().unwrap();
182 run(&mut conn).unwrap();
183 let first: i64 = conn
184 .query_row(&format!("SELECT COUNT(*) FROM {LEDGER}"), [], |r| r.get(0))
185 .unwrap();
186 run(&mut conn).unwrap();
187 let second: i64 = conn
188 .query_row(&format!("SELECT COUNT(*) FROM {LEDGER}"), [], |r| r.get(0))
189 .unwrap();
190 assert_eq!(first, second);
191 assert_eq!(first, all().count() as i64);
192 }
193
194 #[test]
195 fn a_ledger_written_by_sqlx_is_adopted_untouched() {
196 // Simulates an upgraded install: sqlx applied everything, so the new
197 // runner must find nothing pending and leave the rows alone.
198 let mut conn = Connection::open_in_memory().unwrap();
199 run(&mut conn).unwrap();
200 conn.execute(&format!("UPDATE {LEDGER} SET execution_time = 12345"), [])
201 .unwrap();
202
203 run(&mut conn).unwrap();
204
205 let untouched: i64 = conn
206 .query_row(
207 &format!("SELECT COUNT(*) FROM {LEDGER} WHERE execution_time = 12345"),
208 [],
209 |r| r.get(0),
210 )
211 .unwrap();
212 assert_eq!(untouched, all().count() as i64);
213 }
214
215 #[test]
216 fn an_edited_migration_is_refused() {
217 let mut conn = Connection::open_in_memory().unwrap();
218 run(&mut conn).unwrap();
219 conn.execute(
220 &format!("UPDATE {LEDGER} SET checksum = X'00' WHERE version = 1"),
221 [],
222 )
223 .unwrap();
224
225 let err = run(&mut conn).unwrap_err();
226 assert!(matches!(
227 err,
228 MigrateError::ChecksumMismatch { version: 1, .. }
229 ));
230 }
231
232 /// What sqlx 0.9 records, read out of a Balanced Breakfast install that sqlx
233 /// migrated. Not recomputed here: these are the bytes in the field, and
234 /// reproducing them is what lets an upgraded install find nothing pending.
235 const SQLX_CHECKSUMS: &[(i64, &str)] = &[
236 (
237 1,
238 "f130b2307cda80086a379ae69723033ece65f0bfba06203354ab6e0d1613613677e2a98a5b7683b94c88e2479ccd4294",
239 ),
240 (
241 2,
242 "601dcfdb67d8fa2d2aa0a34dbde1dbea1bde71477d3c34d8327ea90594ab9db6d04669f0103f9de15aa7591d23b877fc",
243 ),
244 (
245 3,
246 "57b560ae0cea9bf6d2767feb8d113dbc065fde4ffc163ae966da945b8c0f33150e2e5a4de7c6a814e1faf915971b48fe",
247 ),
248 (
249 4,
250 "ab80308289562efbf9c808c37ac0b89f424c67bf5d054f61379f5e955c1aa75effe06d64311e6ec02f1ee3322b9116f8",
251 ),
252 (
253 5,
254 "bbed9b6af1942d85147f9558ab6c71b7018f194f70d3090abbd3a2a23c12073234e599df7aa0007f8c108462f3a45ca2",
255 ),
256 (
257 6,
258 "ca68cbf17892742631940c807f87fb9302c9e14cdf65432ce1c4877aa7d8f621cc551705478aae102640041eaaffe95a",
259 ),
260 (
261 7,
262 "7a8e9e6f930a4b87951cc062f8b9dc2f16e4f958a2772e8694a129d5dbc53d8573f013859e5e3cd31b4ed6b18fc49a63",
263 ),
264 (
265 8,
266 "efd01690547a1e12b964bada2f0eb163ee7946ed76538f70f685e5e17b89c5bfbab5931565aafd80d5b2d536d664f85e",
267 ),
268 (
269 9,
270 "8f5e8a3ac01a1e9251a7960f02d62dbc507ff5bd1978f555e6058a9977ed85ef1a11e699eace98b372626211adb8ebf8",
271 ),
272 (
273 10,
274 "c365b1e57c7722e83130f6d3fd7563abf6e68bd38bcba58ff94c19f64f7fae265b734aab0846e494eff5020ead77343b",
275 ),
276 (
277 11,
278 "5fce7042c81c0e58376b18cd6973708a25a0d4a473e2f89738653ccc390c20add47fa43751667da4278bb6ffb414b76c",
279 ),
280 (
281 12,
282 "44547197a05d3febe98770858d33d207eadfc1e365cba9075932347fff28bc0609b14a3f6adfa3e901dd36c9b46affb5",
283 ),
284 (
285 13,
286 "361c6ac7e31576ee9840a4af66afc7108f6414d6d1ecc8e2926886eb974f381f01706d0f3555d9a05f329b614af28bbd",
287 ),
288 ];
289
290 /// The compatibility test that matters. If this fails, every install in the
291 /// field sees a checksum mismatch on its first launch after upgrading and
292 /// refuses to start -- so a failure here is never "update the constant", it
293 /// is "the migration file was edited, put it back".
294 #[test]
295 fn checksums_match_what_sqlx_recorded() {
296 let embedded: std::collections::BTreeMap<i64, &str> =
297 all().map(|(v, _, sql)| (v, sql)).collect();
298
299 for (version, expected_hex) in SQLX_CHECKSUMS {
300 let sql = embedded
301 .get(version)
302 .unwrap_or_else(|| panic!("migration {version} is no longer embedded"));
303 let actual = checksum(sql).iter().fold(String::new(), |mut out, b| {
304 use std::fmt::Write as _;
305 let _ = write!(out, "{b:02x}");
306 out
307 });
308 assert_eq!(
309 &actual, expected_hex,
310 "migration {version} no longer hashes to what sqlx recorded in the field"
311 );
312 }
313 }
314
315 #[test]
316 fn a_dirty_ledger_row_stops_the_run() {
317 let mut conn = Connection::open_in_memory().unwrap();
318 run(&mut conn).unwrap();
319 conn.execute(
320 &format!("UPDATE {LEDGER} SET success = false WHERE version = 1"),
321 [],
322 )
323 .unwrap();
324
325 let err = run(&mut conn).unwrap_err();
326 assert!(matches!(err, MigrateError::Dirty(1)));
327 }
328 }
329