cufflink_migrate/
plan.rs

1//! What a history leaves to apply.
2
3use crate::history::{history_version, Applied};
4use cufflink_types::manifest::validate::MAX_MIGRATION_VERSION;
5use cufflink_types::SqlMigration;
6use siphasher::sip::SipHasher13;
7use std::collections::BTreeMap;
8use std::hash::{Hash, Hasher};
9
10/// refinery's checksum of a migration: SipHash-1-3 over its name, its
11/// version as an `i32` and its SQL, so a history either runner wrote
12/// verifies under the other.
13pub fn checksum(migration: &SqlMigration) -> u64 {
14    let mut hasher = SipHasher13::new();
15    migration.name.hash(&mut hasher);
16    history_version(migration.version).hash(&mut hasher);
17    migration.sql.as_str().hash(&mut hasher);
18    hasher.finish()
19}
20
21#[derive(Debug, Clone, PartialEq, Eq)]
22pub struct Plan {
23    /// To apply, ascending.
24    pub pending: Vec<SqlMigration>,
25    /// How many applied migrations match their local file.
26    pub verified: usize,
27    /// Applied versions accepted under the baseline without a matching
28    /// local file.
29    pub baselined: Vec<i32>,
30}
31
32#[derive(Debug, Clone, thiserror::Error, PartialEq, Eq)]
33pub enum PlanError {
34    #[error(
35        "migration V{version} diverges from the one applied: applied '{applied_name}' \
36         (checksum {applied_checksum}), local '{local_name}' (checksum {local_checksum})"
37    )]
38    Divergent {
39        version: i32,
40        applied_name: String,
41        local_name: String,
42        applied_checksum: u64,
43        local_checksum: u64,
44    },
45    #[error("applied migration V{version}__{name} has no local file")]
46    Missing { version: i32, name: String },
47    #[error(
48        "migration V{version} is not applied and precedes V{max_applied}, which is; \
49         a new migration takes a version above every applied one"
50    )]
51    OutOfOrder { version: u32, max_applied: i32 },
52    #[error("migration version {version} is declared twice")]
53    Duplicate { version: u32 },
54    #[error("migration version {version} is above {MAX_MIGRATION_VERSION}")]
55    VersionOutOfRange { version: u32 },
56    #[error("history row of version {version} is unreadable: {reason}")]
57    BadHistoryRow { version: i32, reason: String },
58}
59
60fn by_version(local: &[SqlMigration]) -> Result<BTreeMap<i64, &SqlMigration>, PlanError> {
61    local.iter().try_fold(BTreeMap::new(), |mut found, m| {
62        if m.version > MAX_MIGRATION_VERSION {
63            return Err(PlanError::VersionOutOfRange { version: m.version });
64        }
65        match found.insert(i64::from(m.version), m) {
66            Some(_) => Err(PlanError::Duplicate { version: m.version }),
67            None => Ok(found),
68        }
69    })
70}
71
72enum Standing {
73    Verified,
74    Baselined,
75}
76
77fn standing(
78    applied: &Applied,
79    local: Option<&SqlMigration>,
80    baseline: Option<u32>,
81) -> Result<Standing, PlanError> {
82    let under_baseline =
83        baseline.is_some_and(|baseline| i64::from(applied.version) <= i64::from(baseline));
84    match local.map(|local| (local, checksum(local))) {
85        Some((local, sum)) if local.name == applied.name && sum == applied.checksum => {
86            Ok(Standing::Verified)
87        }
88        _ if under_baseline => Ok(Standing::Baselined),
89        Some((local, sum)) => Err(PlanError::Divergent {
90            version: applied.version,
91            applied_name: applied.name.clone(),
92            local_name: local.name.clone(),
93            applied_checksum: applied.checksum,
94            local_checksum: sum,
95        }),
96        None => Err(PlanError::Missing {
97            version: applied.version,
98            name: applied.name.clone(),
99        }),
100    }
101}
102
103/// Check `applied` against the `local` files and list what is left to
104/// apply. An applied version above `baseline` must have its local file,
105/// unchanged; one at or under it is accepted as it is.
106pub fn plan(
107    applied: &[Applied],
108    local: &[SqlMigration],
109    baseline: Option<u32>,
110) -> Result<Plan, PlanError> {
111    let files = by_version(local)?;
112    let (verified, baselined) =
113        applied
114            .iter()
115            .try_fold((0usize, Vec::new()), |(verified, mut baselined), row| {
116                let file = files.get(&i64::from(row.version)).copied();
117                standing(row, file, baseline).map(|standing| match standing {
118                    Standing::Verified => (verified + 1, baselined),
119                    Standing::Baselined => {
120                        baselined.push(row.version);
121                        (verified, baselined)
122                    }
123                })
124            })?;
125    let max_applied = applied.iter().map(|row| row.version).max();
126    let is_applied = |version: i64| applied.iter().any(|row| i64::from(row.version) == version);
127    let pending: Vec<SqlMigration> = files
128        .iter()
129        .filter(|(version, _)| !is_applied(**version))
130        .map(|(version, file)| match max_applied {
131            Some(max_applied) if *version < i64::from(max_applied) => Err(PlanError::OutOfOrder {
132                version: file.version,
133                max_applied,
134            }),
135            _ => Ok((*file).clone()),
136        })
137        .collect::<Result<_, _>>()?;
138    Ok(Plan {
139        pending,
140        verified,
141        baselined,
142    })
143}
144
145#[cfg(test)]
146mod tests {
147    use super::*;
148
149    fn file(version: u32, name: &str, sql: &str) -> SqlMigration {
150        SqlMigration {
151            version,
152            name: name.into(),
153            sql: sql.into(),
154            no_transaction: false,
155        }
156    }
157
158    fn row(file: &SqlMigration) -> Applied {
159        Applied {
160            version: file.version as i32,
161            name: file.name.clone(),
162            checksum: checksum(file),
163            applied_on: "2024-05-01T09:30:00Z".into(),
164        }
165    }
166
167    fn foreign(version: i32) -> Applied {
168        Applied {
169            version,
170            name: format!("deferred_{version}"),
171            checksum: 7,
172            applied_on: "2024-05-01T09:30:00Z".into(),
173        }
174    }
175
176    fn note_service() -> Vec<SqlMigration> {
177        vec![
178            file(
179                1,
180                "initial_tables",
181                "CREATE TABLE notes (id UUID PRIMARY KEY);\n",
182            ),
183            file(
184                6,
185                "add_note_kind",
186                "ALTER TABLE notes ADD COLUMN kind TEXT;\r\n",
187            ),
188            file(
189                7,
190                "callback_index",
191                "CREATE INDEX notes_callback ON notes (id);",
192            ),
193        ]
194    }
195
196    fn versions(plan: &Plan) -> Vec<u32> {
197        plan.pending.iter().map(|m| m.version).collect()
198    }
199
200    #[test]
201    fn the_checksum_is_refinerys() {
202        for migration in note_service().into_iter().chain([
203            file(2_147_483_647, "last", "SELECT 1;"),
204            file(0, "zero", "\u{feff}-- cufflink:no-transaction\nSELECT 'é';"),
205            file(
206                8,
207                "windows_line_endings",
208                "CREATE TABLE notes (\r\n  id uuid PRIMARY KEY\r\n);\r\n",
209            ),
210        ]) {
211            let refinery =
212                refinery_core::Migration::unapplied(&migration.file_stem(), &migration.sql)
213                    .expect("a refinery migration");
214            assert_eq!(checksum(&migration), refinery.checksum(), "{migration:?}");
215        }
216    }
217
218    #[test]
219    fn a_fresh_history_applies_everything_in_order() {
220        let mut local = note_service();
221        local.reverse();
222        let plan = plan(&[], &local, None).expect("plan");
223        assert_eq!(versions(&plan), [1, 6, 7]);
224        assert_eq!((plan.verified, plan.baselined), (0, vec![]));
225    }
226
227    #[test]
228    fn a_history_is_continued_after_a_gap() {
229        let local = note_service();
230        let applied = [row(&local[0]), row(&local[1])];
231        let plan = plan(&applied, &local, None).expect("plan");
232        assert_eq!(versions(&plan), [7]);
233        assert_eq!((plan.verified, plan.baselined), (2, vec![]));
234        let done = [row(&local[0]), row(&local[1]), row(&local[2])];
235        assert_eq!(
236            super::plan(&done, &local, None),
237            Ok(Plan {
238                pending: vec![],
239                verified: 3,
240                baselined: vec![]
241            })
242        );
243    }
244
245    #[test]
246    fn a_changed_name_or_checksum_diverges() {
247        let local = note_service();
248        let renamed = Applied {
249            name: "add_kind".into(),
250            ..row(&local[1])
251        };
252        assert_eq!(
253            plan(&[row(&local[0]), renamed], &local, None),
254            Err(PlanError::Divergent {
255                version: 6,
256                applied_name: "add_kind".into(),
257                local_name: "add_note_kind".into(),
258                applied_checksum: checksum(&local[1]),
259                local_checksum: checksum(&local[1]),
260            })
261        );
262        let edited = Applied {
263            checksum: 42,
264            ..row(&local[1])
265        };
266        let error = plan(&[row(&local[0]), edited], &local, None).expect_err("diverges");
267        assert_eq!(
268            error.to_string(),
269            format!(
270                "migration V6 diverges from the one applied: applied 'add_note_kind' \
271                 (checksum 42), local 'add_note_kind' (checksum {})",
272                checksum(&local[1])
273            )
274        );
275    }
276
277    #[test]
278    fn an_applied_migration_needs_its_file_above_the_baseline() {
279        let local = note_service();
280        let applied = [row(&local[0]), foreign(3), row(&local[1])];
281        assert_eq!(
282            plan(&applied, &local, None),
283            Err(PlanError::Missing {
284                version: 3,
285                name: "deferred_3".into()
286            })
287        );
288        assert_eq!(
289            plan(&applied, &local, Some(2)),
290            Err(PlanError::Missing {
291                version: 3,
292                name: "deferred_3".into()
293            })
294        );
295        let plan = plan(&applied, &local, Some(3)).expect("plan");
296        assert_eq!(versions(&plan), [7]);
297        assert_eq!((plan.verified, plan.baselined), (2, vec![3]));
298    }
299
300    #[test]
301    fn a_history_under_the_baseline_is_accepted_unverified() {
302        let applied: Vec<Applied> = (1..=15).map(foreign).collect();
303        let local: Vec<SqlMigration> = (14..=16)
304            .map(|v| file(v, &format!("report_{v}"), "SELECT 1;"))
305            .collect();
306        let plan = plan(&applied, &local, Some(15)).expect("plan");
307        assert_eq!(versions(&plan), [16]);
308        assert_eq!(plan.verified, 0);
309        assert_eq!(plan.baselined, (1..=15).collect::<Vec<i32>>());
310    }
311
312    #[test]
313    fn a_new_file_under_an_applied_version_is_out_of_order() {
314        let mut local = note_service();
315        let applied = [row(&local[0]), row(&local[1])];
316        local.push(file(4, "late", "SELECT 1;"));
317        assert_eq!(
318            plan(&applied, &local, None),
319            Err(PlanError::OutOfOrder {
320                version: 4,
321                max_applied: 6
322            })
323        );
324        assert_eq!(
325            plan(&applied, &local, Some(6)),
326            Err(PlanError::OutOfOrder {
327                version: 4,
328                max_applied: 6
329            })
330        );
331    }
332
333    #[test]
334    fn line_endings_are_part_of_what_is_checksummed() {
335        assert_ne!(
336            checksum(&file(8, "notes", "SELECT 1;\r\nSELECT 2;\r\n")),
337            checksum(&file(8, "notes", "SELECT 1;\nSELECT 2;\n")),
338            "a checkout that rewrites line endings diverges, as it does under refinery"
339        );
340    }
341
342    #[test]
343    fn a_version_the_history_cannot_hold_is_refused() {
344        let mut local = note_service();
345        local.push(file(2_147_483_648, "too_far", "SELECT 1;"));
346        assert_eq!(
347            plan(&[], &local, None),
348            Err(PlanError::VersionOutOfRange {
349                version: 2_147_483_648
350            })
351        );
352    }
353
354    #[test]
355    fn a_version_declared_twice_is_refused() {
356        let mut local = note_service();
357        local.push(file(6, "again", "SELECT 1;"));
358        assert_eq!(
359            plan(&[], &local, None),
360            Err(PlanError::Duplicate { version: 6 })
361        );
362    }
363}
cufflink_migrate/plan.rs

cufflink_migrate/
plan.rs
