cufflink_migrate/
exec.rs

1//! What a runner that applies migrations sends and checks, whichever
2//! client it holds: the lock, the history's rows, the time bounds, how a
3//! migration is caught taking the transaction or the role from the
4//! runner, and which invalid indexes an interrupted build left.
5
6use crate::history::QualifiedName;
7use crate::lock::{legacy_table_lock_key, platform_lock_key};
8use cufflink_types::{MigrationLock, SqlMigration};
9use std::time::Duration;
10
11/// How long a run waits for its lock and how long its statements may take.
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub struct Timeouts {
14    pub lock_wait: Duration,
15    pub statement: Duration,
16    pub lock: Duration,
17}
18
19impl Default for Timeouts {
20    fn default() -> Self {
21        Self {
22            lock_wait: Duration::from_secs(600),
23            statement: Duration::from_secs(600),
24            lock: Duration::from_secs(10),
25        }
26    }
27}
28
29/// Takes the run's advisory lock if it is free, answering whether it did.
30pub const TRY_LOCK: &str = "SELECT pg_try_advisory_lock($1)";
31pub const UNLOCK: &str = "SELECT pg_advisory_unlock($1)";
32
33/// The key of the advisory lock a run holds on `history`.
34pub fn lock_key(lock: MigrationLock, history: &QualifiedName) -> i64 {
35    let table = history.dotted();
36    match lock {
37        MigrationLock::Platform => platform_lock_key(&table),
38        MigrationLock::LegacyTableHash => legacy_table_lock_key(&table),
39    }
40}
41
42/// The rows of `history`: `version`, `name`, `applied_on`, `checksum`, as
43/// text but the version, in version order.
44pub fn applied_rows(history: &QualifiedName) -> String {
45    format!(
46        "SELECT version::int4, name::text, applied_on::text, checksum::text \
47         FROM {} ORDER BY version",
48        history.quoted()
49    )
50}
51
52/// Records a migration in `history`, taking `version`, `name`,
53/// `applied_on` and `checksum`.
54pub fn record_statement(history: &QualifiedName) -> String {
55    format!(
56        "INSERT INTO {} (version, name, applied_on, checksum) VALUES ($1, $2, $3, $4)",
57        history.quoted()
58    )
59}
60
61/// The transaction and the role statements run in now, answered as
62/// [`Standing`]'s two fields.
63pub const STANDING: &str = "SELECT txid_current()::text, current_user::text";
64
65/// The transaction and the role statements run in.
66#[derive(Debug, Clone, PartialEq, Eq)]
67pub struct Standing {
68    pub transaction: String,
69    pub role: String,
70}
71
72impl Standing {
73    /// How a migration that began in `self` and ended in `after` escaped.
74    /// The role is the runner's to keep only when the run imposes one.
75    pub fn escape(&self, after: &Self, role_imposed: bool) -> Option<Escape> {
76        if self.transaction != after.transaction {
77            Some(Escape::Transaction)
78        } else if role_imposed && self.role != after.role {
79            Some(Escape::Role)
80        } else {
81            None
82        }
83    }
84}
85
86/// How a migration left what the runner ran it in.
87#[derive(Debug, Clone, Copy, PartialEq, Eq)]
88pub enum Escape {
89    Transaction,
90    Role,
91}
92
93impl std::fmt::Display for Escape {
94    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
95        f.write_str(match self {
96            Escape::Transaction => "ended the transaction it ran in",
97            Escape::Role => "changed the role it ran as",
98        })
99    }
100}
101
102/// Why a migration that escaped is refused rather than recorded.
103pub fn escaped(migration: &SqlMigration, escape: Escape) -> String {
104    format!(
105        "migration V{}__{} {escape}; it is not recorded, and what it committed itself stays. \
106         Transaction and role control belong to the runner: a migration that must run \
107         outside a transaction begins with '{}'",
108        migration.version,
109        migration.name,
110        SqlMigration::NO_TRANSACTION_MARKER
111    )
112}
113
114/// The invalid indexes of a schema, taking its folded name.
115pub const INVALID_INDEXES: &str = "SELECT c.relname::text FROM pg_index i \
116     JOIN pg_class c ON c.oid = i.indexrelid \
117     JOIN pg_namespace n ON n.oid = c.relnamespace \
118     WHERE n.nspname = $1 AND NOT i.indisvalid ORDER BY 1";
119
120/// The invalid indexes of the schema that `sql` names: a concurrent build
121/// the last attempt left unfinished, which would make this attempt fail.
122pub fn interrupted_builds<'a>(invalid: &'a [String], sql: &str) -> Vec<&'a str> {
123    let named = |index: &str| {
124        sql.match_indices(index).any(|(at, _)| {
125            let is_word = |c: char| c.is_ascii_alphanumeric() || c == '_';
126            let before = sql[..at].chars().next_back();
127            let after = sql[at + index.len()..].chars().next();
128            !before.is_some_and(is_word) && !after.is_some_and(is_word)
129        })
130    };
131    invalid
132        .iter()
133        .map(String::as_str)
134        .filter(|index| !index.is_empty() && named(index))
135        .collect()
136}
137
138#[cfg(test)]
139mod tests {
140    use super::*;
141
142    fn history() -> QualifiedName {
143        QualifiedName::parse("schema_history", "loan_service").expect("a history")
144    }
145
146    #[test]
147    fn the_history_is_read_and_written_by_its_quoted_name() {
148        assert_eq!(
149            applied_rows(&history()),
150            "SELECT version::int4, name::text, applied_on::text, checksum::text \
151             FROM \"loan_service\".\"schema_history\" ORDER BY version"
152        );
153        assert_eq!(
154            record_statement(&history()),
155            "INSERT INTO \"loan_service\".\"schema_history\" (version, name, applied_on, \
156             checksum) VALUES ($1, $2, $3, $4)"
157        );
158    }
159
160    #[test]
161    fn the_lock_is_the_one_the_history_declares() {
162        let history = history();
163        assert_eq!(
164            lock_key(MigrationLock::LegacyTableHash, &history),
165            legacy_table_lock_key("loan_service.schema_history")
166        );
167        assert_eq!(
168            lock_key(MigrationLock::Platform, &history),
169            platform_lock_key("loan_service.schema_history")
170        );
171    }
172
173    #[test]
174    fn timeouts_default_to_the_documented_values() {
175        assert_eq!(
176            Timeouts::default(),
177            Timeouts {
178                lock_wait: Duration::from_secs(600),
179                statement: Duration::from_secs(600),
180                lock: Duration::from_secs(10),
181            }
182        );
183    }
184
185    #[test]
186    fn a_migration_may_not_take_the_transaction_or_an_imposed_role() {
187        let standing = |transaction: &str, role: &str| Standing {
188            transaction: transaction.into(),
189            role: role.into(),
190        };
191        let before = standing("811", "loan_service_user");
192        assert_eq!(before.escape(&before.clone(), true), None);
193        assert_eq!(
194            before.escape(&standing("812", "loan_service_user"), false),
195            Some(Escape::Transaction)
196        );
197        assert_eq!(
198            before.escape(&standing("811", "platform"), true),
199            Some(Escape::Role)
200        );
201        assert_eq!(
202            before.escape(&standing("811", "platform"), false),
203            None,
204            "with no role imposed there is none to escape"
205        );
206        let migration = SqlMigration {
207            version: 8,
208            name: "commits".into(),
209            sql: String::new(),
210            no_transaction: false,
211        };
212        assert!(escaped(&migration, Escape::Transaction).starts_with(
213            "migration V8__commits ended the transaction it ran in; it is not recorded"
214        ));
215        assert!(escaped(&migration, Escape::Role)
216            .starts_with("migration V8__commits changed the role it ran as;"));
217    }
218
219    #[test]
220    fn an_invalid_index_is_recovered_only_when_the_migration_names_it() {
221        let invalid = vec![
222            "notes_kind".to_string(),
223            "notes_kind_created".to_string(),
224            "other".to_string(),
225        ];
226        assert_eq!(
227            interrupted_builds(
228                &invalid,
229                "CREATE INDEX CONCURRENTLY notes_kind ON notes (kind);"
230            ),
231            ["notes_kind"]
232        );
233        assert_eq!(
234            interrupted_builds(
235                &invalid,
236                "CREATE INDEX CONCURRENTLY \"notes_kind_created\" ON notes (kind, created_at);"
237            ),
238            ["notes_kind_created"]
239        );
240        assert!(interrupted_builds(&invalid, "CREATE INDEX another ON notes (id);").is_empty());
241        assert!(interrupted_builds(&[String::new()], "SELECT 1;").is_empty());
242    }
243}
cufflink_migrate/exec.rs

cufflink_migrate/
exec.rs
