Skip to main content

max / quasi

13.8 KB · 386 lines History Blame Raw
1 //! Applying what the ledger says is missing.
2 //!
3 //! Lifted from goingson's runner, which was written when that app left
4 //! `sqlx-sqlite` and needed to keep reading a ledger `sqlx` had written. Two
5 //! things carried over unchanged and neither is a free choice there: the table
6 //! shape and the checksum function (sha384 over the raw file bytes, verified
7 //! against `sqlx` 0.9). They are kept here so that an app with `sqlx` history
8 //! can adopt this runner by naming its old ledger and nothing else — see
9 //! [`Migrator::ledger`].
10 //!
11 //! What did not carry over is the *default* name. A generated app has no
12 //! installs in the field and no `sqlx` past, and calling its ledger
13 //! `_sqlx_migrations` would be a new app inheriting a compatibility note about
14 //! a library it never linked.
15
16 use std::time::Instant;
17
18 use rusqlite::{Connection, OptionalExtension};
19 use sha2::{Digest, Sha384};
20
21 /// One migration, as [`crate::embed::from_dir`] emits it.
22 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
23 pub struct Migration {
24 /// The integer prefix of the filename. Applied in this order.
25 pub version: i64,
26 /// The rest of the filename, underscores as spaces.
27 pub description: &'static str,
28 /// The file, verbatim. Hashed as-is.
29 pub sql: &'static str,
30 }
31
32 /// A migration that could not be applied.
33 ///
34 /// Every member needs a person. There is no variant meaning "retry": a run that
35 /// stops has either found a database it does not recognise or a migration that
36 /// does not apply, and running it again produces the same answer.
37 #[derive(Debug, thiserror::Error)]
38 pub enum MigrateError {
39 #[error("database error running migrations: {0}")]
40 Db(#[from] rusqlite::Error),
41
42 #[error(
43 "migration {version} ({description}) was already applied, but its file has changed since. \
44 Applied migrations are immutable — add a new migration instead of editing a shipped one."
45 )]
46 ChecksumMismatch { version: i64, description: String },
47
48 #[error("migration {0} is partially applied; fix it and remove its row from the ledger")]
49 Dirty(i64),
50
51 #[error("migration {version} ({description}) failed: {source}")]
52 Apply {
53 version: i64,
54 description: String,
55 #[source]
56 source: rusqlite::Error,
57 },
58 }
59
60 /// sha384 of a migration's bytes.
61 ///
62 /// `sqlx` 0.9's function, so a ledger it wrote verifies against this one. Not a
63 /// free choice for an adopting app, and not worth changing for a new one.
64 #[must_use]
65 pub fn checksum(sql: &str) -> Vec<u8> {
66 Sha384::digest(sql.as_bytes()).to_vec()
67 }
68
69 /// The ledger a fresh app writes.
70 const DEFAULT_LEDGER: &str = "_quasi_migrations";
71
72 /// A set of migrations and the ledger recording which of them ran.
73 pub struct Migrator {
74 migrations: &'static [Migration],
75 ledger: &'static str,
76 }
77
78 impl Migrator {
79 /// A migrator over this table, recording to the default ledger.
80 #[must_use]
81 pub fn new(migrations: &'static [Migration]) -> Self {
82 Self {
83 migrations,
84 ledger: DEFAULT_LEDGER,
85 }
86 }
87
88 /// Record to a ledger of this name instead.
89 ///
90 /// For an app that has `sqlx` history: `.ledger("_sqlx_migrations")` makes
91 /// this runner read the rows already there, find nothing pending, and do
92 /// nothing. Writing a fresh ledger beside an existing one would instead
93 /// make an upgraded install believe it had applied nothing and re-run every
94 /// migration against a populated database.
95 ///
96 /// # Panics
97 ///
98 /// If the name is not a bare identifier. It is interpolated into SQL, since
99 /// a table name cannot be bound as a parameter, and a startup literal is the
100 /// right place to be strict about that.
101 #[must_use]
102 pub fn ledger(mut self, name: &'static str) -> Self {
103 assert!(
104 !name.is_empty()
105 && name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')
106 && !name.starts_with(|c: char| c.is_ascii_digit()),
107 "ledger name `{name}` is not a bare identifier"
108 );
109 self.ledger = name;
110 self
111 }
112
113 /// Create the ledger if absent.
114 ///
115 /// The DDL is `sqlx` 0.9's verbatim, which is what makes adopting an
116 /// existing ledger a no-op rather than a conflicting `CREATE TABLE`.
117 fn ensure_ledger(&self, conn: &Connection) -> Result<(), rusqlite::Error> {
118 conn.execute_batch(&format!(
119 "CREATE TABLE IF NOT EXISTS {} (
120 version BIGINT PRIMARY KEY,
121 description TEXT NOT NULL,
122 installed_on TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
123 success BOOLEAN NOT NULL,
124 checksum BLOB NOT NULL,
125 execution_time BIGINT NOT NULL
126 );",
127 self.ledger
128 ))
129 }
130
131 /// Apply every migration not yet in the ledger.
132 ///
133 /// Each migration and its ledger row commit together, so a crash mid-run
134 /// leaves the database at a migration boundary rather than half-applied.
135 ///
136 /// # Errors
137 ///
138 /// If a migration fails, if an applied migration's file has changed since,
139 /// or if the ledger carries a failed row from a previous run.
140 #[tracing::instrument(skip_all)]
141 pub fn run(&self, conn: &mut Connection) -> Result<(), MigrateError> {
142 self.ensure_ledger(conn)?;
143
144 // A `success = false` row means a previous run died between applying a
145 // migration and committing its ledger row. This runner cannot write
146 // one, since it commits both together, but an install upgraded from
147 // `sqlx` may carry one and it still needs a person.
148 let dirty: Option<i64> = conn
149 .query_row(
150 &format!(
151 "SELECT version FROM {} WHERE success = false ORDER BY version LIMIT 1",
152 self.ledger
153 ),
154 [],
155 |row| row.get(0),
156 )
157 .optional()?;
158 if let Some(version) = dirty {
159 return Err(MigrateError::Dirty(version));
160 }
161
162 let applied: std::collections::BTreeMap<i64, Vec<u8>> = {
163 let mut stmt = conn.prepare(&format!(
164 "SELECT version, checksum FROM {} ORDER BY version",
165 self.ledger
166 ))?;
167 let rows = stmt.query_map([], |row| {
168 Ok((row.get::<_, i64>(0)?, row.get::<_, Vec<u8>>(1)?))
169 })?;
170 rows.collect::<Result<_, _>>()?
171 };
172
173 for migration in self.migrations {
174 let digest = checksum(migration.sql);
175
176 if let Some(recorded) = applied.get(&migration.version) {
177 // Already applied. The only question left is whether the file
178 // still hashes to what was recorded; if not, a shipped
179 // migration was edited and every install that ran it now
180 // disagrees with this one.
181 if *recorded != digest {
182 return Err(MigrateError::ChecksumMismatch {
183 version: migration.version,
184 description: migration.description.to_owned(),
185 });
186 }
187 continue;
188 }
189
190 tracing::info!(
191 version = migration.version,
192 description = migration.description,
193 "applying migration"
194 );
195 let started = Instant::now();
196 let tx = conn.transaction()?;
197 tx.execute_batch(migration.sql)
198 .map_err(|source| MigrateError::Apply {
199 version: migration.version,
200 description: migration.description.to_owned(),
201 source,
202 })?;
203 tx.execute(
204 &format!(
205 "INSERT INTO {} (version, description, success, checksum, execution_time)
206 VALUES (?1, ?2, TRUE, ?3, ?4)",
207 self.ledger
208 ),
209 rusqlite::params![
210 migration.version,
211 migration.description,
212 digest,
213 i64::try_from(started.elapsed().as_nanos()).unwrap_or(i64::MAX),
214 ],
215 )?;
216 tx.commit()?;
217 }
218
219 Ok(())
220 }
221 }
222
223 #[cfg(test)]
224 mod tests {
225 use super::*;
226
227 const FIRST: Migration = Migration {
228 version: 1,
229 description: "initial schema",
230 sql: "CREATE TABLE note (id INTEGER PRIMARY KEY, body TEXT NOT NULL);",
231 };
232
233 const SECOND: Migration = Migration {
234 version: 2,
235 description: "archived flag",
236 sql: "ALTER TABLE note ADD COLUMN archived INTEGER NOT NULL DEFAULT 0;",
237 };
238
239 static ONE: &[Migration] = &[FIRST];
240 static BOTH: &[Migration] = &[FIRST, SECOND];
241
242 fn columns(conn: &Connection) -> Vec<String> {
243 let mut stmt = conn
244 .prepare("SELECT name FROM pragma_table_info('note')")
245 .unwrap();
246 let rows = stmt.query_map([], |row| row.get::<_, String>(0)).unwrap();
247 rows.collect::<Result<_, _>>().unwrap()
248 }
249
250 fn applied(conn: &Connection, ledger: &str) -> Vec<i64> {
251 let mut stmt = conn
252 .prepare(&format!("SELECT version FROM {ledger} ORDER BY version"))
253 .unwrap();
254 let rows = stmt.query_map([], |row| row.get::<_, i64>(0)).unwrap();
255 rows.collect::<Result<_, _>>().unwrap()
256 }
257
258 #[test]
259 fn an_empty_database_gets_every_migration() {
260 let mut conn = Connection::open_in_memory().unwrap();
261 Migrator::new(BOTH).run(&mut conn).unwrap();
262 assert_eq!(columns(&conn), ["id", "body", "archived"]);
263 assert_eq!(applied(&conn, DEFAULT_LEDGER), [1, 2]);
264 }
265
266 #[test]
267 fn a_second_run_applies_nothing() {
268 let mut conn = Connection::open_in_memory().unwrap();
269 Migrator::new(BOTH).run(&mut conn).unwrap();
270 // The second run is the one that would fail loudly if it re-applied:
271 // `CREATE TABLE` on a table that exists.
272 Migrator::new(BOTH).run(&mut conn).unwrap();
273 assert_eq!(applied(&conn, DEFAULT_LEDGER), [1, 2]);
274 }
275
276 #[test]
277 fn a_new_migration_applies_over_an_existing_database() {
278 let mut conn = Connection::open_in_memory().unwrap();
279 Migrator::new(ONE).run(&mut conn).unwrap();
280 assert_eq!(columns(&conn), ["id", "body"]);
281
282 Migrator::new(BOTH).run(&mut conn).unwrap();
283 assert_eq!(columns(&conn), ["id", "body", "archived"]);
284 }
285
286 #[test]
287 fn editing_a_shipped_migration_is_refused() {
288 let mut conn = Connection::open_in_memory().unwrap();
289 Migrator::new(ONE).run(&mut conn).unwrap();
290
291 // Same version and description, one character of SQL different.
292 static EDITED: &[Migration] = &[Migration {
293 version: 1,
294 description: "initial schema",
295 sql: "CREATE TABLE note (id INTEGER PRIMARY KEY, body TEXT);",
296 }];
297 let refused = Migrator::new(EDITED).run(&mut conn).unwrap_err();
298 assert!(matches!(
299 refused,
300 MigrateError::ChecksumMismatch { version: 1, .. }
301 ));
302 }
303
304 #[test]
305 fn a_failed_row_from_a_previous_run_stops_everything() {
306 let mut conn = Connection::open_in_memory().unwrap();
307 Migrator::new(ONE).run(&mut conn).unwrap();
308 conn.execute(
309 &format!(
310 "INSERT INTO {DEFAULT_LEDGER} (version, description, success, checksum, execution_time)
311 VALUES (9, 'half done', FALSE, X'00', 0)"
312 ),
313 [],
314 )
315 .unwrap();
316
317 let stopped = Migrator::new(BOTH).run(&mut conn).unwrap_err();
318 assert!(matches!(stopped, MigrateError::Dirty(9)));
319 }
320
321 #[test]
322 fn a_broken_migration_names_itself() {
323 let mut conn = Connection::open_in_memory().unwrap();
324 static BROKEN: &[Migration] = &[Migration {
325 version: 1,
326 description: "not sql",
327 sql: "CREATE TABL note (id INTEGER);",
328 }];
329
330 let failed = Migrator::new(BROKEN).run(&mut conn).unwrap_err();
331 let MigrateError::Apply {
332 version,
333 description,
334 ..
335 } = failed
336 else {
337 panic!("expected an apply failure");
338 };
339 assert_eq!(version, 1);
340 assert_eq!(description, "not sql");
341
342 // And nothing was recorded, so fixing the file and re-running works.
343 assert_eq!(applied(&conn, DEFAULT_LEDGER), Vec::<i64>::new());
344 }
345
346 #[test]
347 fn an_sqlx_ledger_is_adopted_rather_than_duplicated() {
348 let mut conn = Connection::open_in_memory().unwrap();
349 // What an install upgraded from sqlx looks like: the schema is there
350 // and so are the rows, written by a library this app no longer links.
351 conn.execute_batch(FIRST.sql).unwrap();
352 conn.execute_batch(
353 "CREATE TABLE _sqlx_migrations (
354 version BIGINT PRIMARY KEY,
355 description TEXT NOT NULL,
356 installed_on TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
357 success BOOLEAN NOT NULL,
358 checksum BLOB NOT NULL,
359 execution_time BIGINT NOT NULL
360 );",
361 )
362 .unwrap();
363 conn.execute(
364 "INSERT INTO _sqlx_migrations (version, description, success, checksum, execution_time)
365 VALUES (1, 'initial schema', TRUE, ?1, 0)",
366 rusqlite::params![checksum(FIRST.sql)],
367 )
368 .unwrap();
369
370 Migrator::new(BOTH)
371 .ledger("_sqlx_migrations")
372 .run(&mut conn)
373 .unwrap();
374
375 // Migration 1 was recognised as done and only 2 ran.
376 assert_eq!(applied(&conn, "_sqlx_migrations"), [1, 2]);
377 assert_eq!(columns(&conn), ["id", "body", "archived"]);
378 }
379
380 #[test]
381 #[should_panic(expected = "not a bare identifier")]
382 fn a_ledger_name_that_is_not_an_identifier_is_a_bug() {
383 let _ = Migrator::new(ONE).ledger("ledger; DROP TABLE note");
384 }
385 }
386