siderite_migrations/executor/
intents.rs

1//! Fail closed when explicit non-transactional data work is unconfirmed.
2
3use super::catalog::table_exists;
4use super::recovery::{read_index, text};
5use super::{Db, Migration, MigrationError as Error, MigrationRegistry};
6use super::{Value, placeholder, quote_star};
7use serde::Serialize;
8
9const TABLE: &str = "siderite_migration_intents";
10type Result<T> = std::result::Result<T, Error>;
11type Intents = Vec<MigrationIntent>;
12
13/// Explicit SQL/callback work entered before completion was confirmed.
14#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
15pub struct MigrationIntent {
16    /// Migration containing the explicit script or callback.
17    pub migration_id: String,
18    /// File checksum when execution started.
19    pub checksum: String,
20    /// `apply` or `unapply`.
21    pub direction: String,
22    /// Zero-based operation index in this direction.
23    pub operation: usize,
24    /// Callback name; None denotes an explicit RunSQL script.
25    pub callback: Option<String>,
26}
27
28impl MigrationIntent {
29    fn uncertain(&self) -> Error {
30        match &self.callback {
31            Some(callback) => Error::UncertainRustStep {
32                id: self.migration_id.clone(),
33                callback: callback.clone(),
34                operation: self.operation,
35            },
36            None => Error::UncertainSqlStep {
37                id: self.migration_id.clone(),
38                operation: self.operation,
39            },
40        }
41    }
42}
43
44/// Read recorded intents without creating bookkeeping tables.
45///
46/// Args:
47///     db: Database handle used for catalog and bookkeeping reads.
48///     id: Optional migration filter.
49///
50/// Returns:
51///     Validated pending steps, ordered by migration id.
52pub(super) async fn read(db: &Db, id: Option<&str>) -> Result<Intents> {
53    if !table_exists(db, TABLE).await? {
54        return Ok(Vec::new());
55    }
56    let kind = db.capabilities().kind;
57    let mut sql = format!("SELECT * FROM {}", quote_star(kind, TABLE));
58    let mut params = Vec::new();
59    if let Some(id) = id {
60        let bind = placeholder(kind, 1);
61        sql.push_str(&format!(" WHERE migration_id = {bind}"));
62        params.push(Value::Text(id.to_owned()));
63    }
64    sql.push_str(" ORDER BY migration_id");
65    let mut intents = Vec::new();
66    for row in db.raw_sql(&sql, params).await?.rows {
67        let direction = text(&row, "direction")?;
68        if !matches!(direction.as_str(), "apply" | "unapply") {
69            return Err(Error::state("invalid intent direction"));
70        }
71        intents.push(MigrationIntent {
72            migration_id: text(&row, "migration_id")?,
73            checksum: text(&row, "checksum")?,
74            direction,
75            operation: read_index(&row, "operation_index")?,
76            callback: match row.get("callback") {
77                Some(Value::Null) => None,
78                _ => Some(text(&row, "callback")?),
79            },
80        });
81    }
82    Ok(intents)
83}
84
85/// Reject uncertain work unless its callback is declared safe to replay.
86///
87/// Args:
88///     db: Migration session.
89///     migration: Original checksummed migration.
90///     registry: Callback implementations and replay declarations.
91///     direction: Requested apply or unapply direction.
92///
93/// Returns:
94///     Success when the requested execution may proceed.
95pub(super) async fn guard(
96    db: &Db,
97    migration: &Migration,
98    registry: &MigrationRegistry,
99    direction: &str,
100) -> Result<()> {
101    for intent in read(db, Some(&migration.id)).await? {
102        if intent.checksum != migration.checksum {
103            return Err(Error::state(format!(
104                "migration `{}` changed since a partial execution; \
105                 restore the original file before reconciliation",
106                migration.id,
107            )));
108        }
109        if intent.direction != direction
110            || !intent
111                .callback
112                .as_deref()
113                .is_some_and(|name| registry.is_replay_safe(name))
114        {
115            return Err(intent.uncertain());
116        }
117    }
118    Ok(())
119}
120
121/// Persist uncertainty before entering non-transactional explicit work.
122///
123/// Args:
124///     db: Migration session.
125///     migration: Original checksummed migration.
126///     direction: Apply or unapply direction.
127///     operation: Zero-based operation index.
128///     callback: Callback name, or None for an explicit SQL script.
129///
130/// Returns:
131///     Success after the intent is recorded or its identity is verified.
132pub(super) async fn begin(
133    db: &Db,
134    migration: &Migration,
135    direction: &str,
136    operation: usize,
137    callback: Option<&str>,
138) -> Result<()> {
139    let kind = db.capabilities().kind;
140    let table = quote_star(kind, TABLE);
141    db.execute_script(&format!(
142        "CREATE TABLE IF NOT EXISTS {table} (\
143         migration_id VARCHAR(255) PRIMARY KEY, \
144         checksum VARCHAR(255) NOT NULL, \
145         direction VARCHAR(16) NOT NULL, \
146         operation_index INTEGER NOT NULL, callback TEXT)"
147    ))
148    .await?;
149    if let Some(intent) = read(db, Some(&migration.id)).await?.first() {
150        let same_callback = intent.callback.as_deref() == callback;
151        if intent.operation != operation || !same_callback {
152            return Err(intent.uncertain());
153        }
154        return Ok(());
155    }
156    let values = (1..=5)
157        .map(|index| placeholder(kind, index))
158        .collect::<Vec<_>>()
159        .join(", ");
160    let operation = i64::try_from(operation).map_err(|_| index_overflow())?;
161    db.raw_execute(
162        &format!(
163            "INSERT INTO {table} (migration_id, checksum, direction, \
164         operation_index, callback) VALUES ({values})"
165        ),
166        vec![
167            Value::Text(migration.id.clone()),
168            Value::Text(migration.checksum.clone()),
169            Value::Text(direction.to_owned()),
170            Value::Int(operation),
171            callback.map_or(Value::Null, |name| Value::Text(name.to_owned())),
172        ],
173    )
174    .await?;
175    Ok(())
176}
177
178/// Clear an intent after completion and progress are recorded.
179///
180/// Args:
181///     db: Migration session.
182///     migration: Original checksummed migration.
183///
184/// Returns:
185///     Success only when exactly one matching intent was removed.
186pub(super) async fn finish(db: &Db, migration: &Migration) -> Result<()> {
187    let kind = db.capabilities().kind;
188    let affected = db
189        .raw_execute(
190            &format!(
191                "DELETE FROM {} WHERE migration_id = {} AND checksum = {}",
192                quote_star(kind, TABLE),
193                placeholder(kind, 1),
194                placeholder(kind, 2),
195            ),
196            vec![
197                Value::Text(migration.id.clone()),
198                Value::Text(migration.checksum.clone()),
199            ],
200        )
201        .await?;
202    if affected != 1 {
203        return Err(Error::state("migration intent lost before completion"));
204    }
205    Ok(())
206}
207
208fn index_overflow() -> Error {
209    Error::state("migration operation index overflow")
210}
211
212/// Clear a guarded callback intent when durable progress records completion.
213///
214/// Args:
215///     db: Migration session after the replay-safety guard succeeded.
216///     migration: Original checksummed migration.
217///     completed: Validated number of completed operations.
218///
219/// Returns:
220///     Success after clearing an already-recorded callback completion.
221pub(super) async fn finish_recorded_callback(
222    db: &Db,
223    migration: &Migration,
224    completed: usize,
225) -> Result<()> {
226    if completed == 0 {
227        return Ok(());
228    }
229    for intent in read(db, Some(&migration.id)).await? {
230        if intent.callback.is_some() && intent.operation < completed {
231            finish(db, migration).await?;
232        }
233    }
234    Ok(())
235}
siderite_migrations/executor/intents.rs

siderite_migrations/executor/
intents.rs
