siderite_migrations/executor/
holder.rs

1//! Read-only identification of the session holding the migration lock.
2//!
3//! Only the lock-timeout error path calls this: successful runs never pay
4//! for the catalog reads.
5
6use super::locks::{MYSQL_LOCK_NAME, PG_MIGRATE_LOCK};
7use super::{BackendKind, MigrationError as Error, Migrator, Value};
8use serde::Serialize;
9use siderite_orm::Row;
10use std::fmt;
11
12/// The session that holds the migration lock, as far as the server shows
13/// it to the inspecting role. Fields other than `session` are `None` when
14/// the server hides them (PostgreSQL without `pg_read_all_stats`, MySQL
15/// without `PROCESS`) or the session ended between the two reads.
16#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
17pub struct LockHolder {
18    /// PostgreSQL backend pid or MySQL connection id.
19    pub session: i64,
20    /// Login role (PostgreSQL) or user (MySQL).
21    pub user: Option<String>,
22    /// Client address (PostgreSQL) or `host:port` (MySQL).
23    pub client: Option<String>,
24    /// PostgreSQL `application_name`; MySQL `COMMAND`.
25    pub application: Option<String>,
26    /// Session state (`active`, `idle in transaction`, MySQL `STATE`).
27    pub state: Option<String>,
28    /// Seconds in the current state (PostgreSQL) or command (MySQL).
29    pub seconds: Option<i64>,
30}
31
32impl fmt::Display for LockHolder {
33    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
34        write!(f, "session {}", self.session)?;
35        let parts = [
36            ("user", &self.user),
37            ("client", &self.client),
38            ("application", &self.application),
39            ("state", &self.state),
40        ];
41        for (label, value) in parts {
42            if let Some(value) = value {
43                write!(f, ", {label} {value}")?;
44            }
45        }
46        if let Some(seconds) = self.seconds {
47            write!(f, ", for {seconds}s")?;
48        }
49        Ok(())
50    }
51}
52
53/// Advisory locks taken with one `bigint` key store its high and low
54/// halves in `classid` / `objid` and `objsubid = 1`.
55const PG_HOLDER: &str = "SELECT l.pid::bigint AS session, \
56    a.usename::text AS usr, a.client_addr::text AS client, \
57    a.application_name::text AS app, a.state::text AS state, \
58    CAST(EXTRACT(EPOCH FROM now() - a.state_change) AS bigint) AS seconds \
59    FROM pg_catalog.pg_locks l \
60    LEFT JOIN pg_catalog.pg_stat_activity a ON a.pid = l.pid \
61    WHERE l.locktype = 'advisory' AND l.granted AND l.objsubid = 1 \
62    AND l.classid::bigint = $1 AND l.objid::bigint = $2 \
63    AND l.database = (SELECT oid FROM pg_catalog.pg_database \
64    WHERE datname = current_database()) LIMIT 1";
65
66const MYSQL_SESSION: &str = "SELECT CAST(IS_USED_LOCK(?) AS SIGNED) \
67    AS session";
68
69const MYSQL_DETAILS: &str = "SELECT CAST(USER AS CHAR) AS usr, \
70    CAST(HOST AS CHAR) AS client, CAST(COMMAND AS CHAR) AS app, \
71    CAST(STATE AS CHAR) AS state, CAST(TIME AS SIGNED) AS seconds \
72    FROM information_schema.processlist WHERE ID = ?";
73
74impl Migrator<'_> {
75    /// Identify the session currently holding the migration lock.
76    ///
77    /// Intended for the [`MigrationError::LockTimeout`](Error::LockTimeout)
78    /// path: it takes no lock and changes nothing, and the holder may
79    /// finish between the timeout and this read.
80    ///
81    /// Returns:
82    ///   The holder, or `None` when the lock is free (or on SQLite, which
83    ///   has no migration lock).
84    ///
85    /// # Errors
86    /// Unsupported backend or unreadable catalogs.
87    pub async fn lock_holder(&self) -> Result<Option<LockHolder>, Error> {
88        match self.kind() {
89            BackendKind::Sqlite => Ok(None),
90            BackendKind::Postgres => self.pg_holder().await,
91            BackendKind::MySql => self.mysql_holder().await,
92            other => Err(Error::UnsupportedBackend(other)),
93        }
94    }
95
96    async fn pg_holder(&self) -> Result<Option<LockHolder>, Error> {
97        let params = vec![
98            Value::Int(PG_MIGRATE_LOCK >> 32),
99            Value::Int(PG_MIGRATE_LOCK & 0xffff_ffff),
100        ];
101        let rows = self.db.raw_sql(PG_HOLDER, params).await?.rows;
102        Ok(rows.first().and_then(|row| {
103            let session = int(row, "session")?;
104            Some(details(session, Some(row)))
105        }))
106    }
107
108    async fn mysql_holder(&self) -> Result<Option<LockHolder>, Error> {
109        let name = vec![Value::Text(MYSQL_LOCK_NAME.into())];
110        let rows = self.db.raw_sql(MYSQL_SESSION, name).await?.rows;
111        let Some(session) = rows.first().and_then(|r| int(r, "session")) else {
112            return Ok(None);
113        };
114        let id = vec![Value::Int(session)];
115        let rows = self.db.raw_sql(MYSQL_DETAILS, id).await?.rows;
116        Ok(Some(details(session, rows.first())))
117    }
118}
119
120fn details(session: i64, row: Option<&Row>) -> LockHolder {
121    let text = |key| row.and_then(|row| text(row, key));
122    LockHolder {
123        session,
124        user: text("usr"),
125        client: text("client"),
126        application: text("app"),
127        state: text("state"),
128        seconds: row.and_then(|row| int(row, "seconds")),
129    }
130}
131
132fn int(row: &Row, key: &str) -> Option<i64> {
133    match row.get(key) {
134        Some(Value::Int(n)) => Some(*n),
135        _ => None,
136    }
137}
138
139fn text(row: &Row, key: &str) -> Option<String> {
140    match row.get(key) {
141        Some(Value::Text(s)) if !s.is_empty() => Some(s.clone()),
142        _ => None,
143    }
144}
145
146#[cfg(test)]
147mod tests {
148    use super::LockHolder;
149
150    #[test]
151    fn display_lists_visible_fields_only() {
152        let mut holder = LockHolder {
153            session: 42,
154            user: Some("app".into()),
155            client: None,
156            application: Some("deploy".into()),
157            state: Some("active".into()),
158            seconds: Some(7),
159        };
160        assert_eq!(
161            holder.to_string(),
162            "session 42, user app, application deploy, state active, for 7s"
163        );
164        holder = LockHolder {
165            user: None,
166            application: None,
167            state: None,
168            seconds: None,
169            ..holder
170        };
171        assert_eq!(holder.to_string(), "session 42");
172    }
173}
siderite_migrations/executor/holder.rs

siderite_migrations/executor/
holder.rs
