src/migrate.rs
//! CONFIG MIGRATION — the stamped, ordered **chain** a [`crate::install::refresh_product`]
//! replays between installing a new bundle and flipping `current`.
//!
//! The doctrine (super-plan §🔄 REFRESH): *migration code lives IN THE PRODUCT, the
//! orchestration lives in Skidbladnir.* Skidbladnir owns none of a product's schema
//! knowledge — it owns only:
//!
//! 1. **The stamp.** Every install writes the product's *current schema version* into
//! a tiny [`MigrationStamp`](crate::migrate::MigrationStamp) file **beside the data** ([`stamp_path`](crate::migrate::stamp_path)). The stamp
//! is the single source of truth for "what schema does the data on disk speak?".
//! 2. **The chain.** The product declares its ordered list of schema versions (oldest
//! → head). [`migration_steps`](crate::migrate::migration_steps) diffs the on-disk stamp against that chain and
//! yields the exact ordered `(from → to)` hops to replay. `stamped == head` ⇒ **no
//! hops** (this is what makes `refresh` idempotent); a stamp not in the chain is a
//! hard error (never silently "catch up").
//! 3. **The replay.** [`replay_migrations`](crate::migrate::replay_migrations) runs each hop through a [`MigrateRunner`](crate::migrate::MigrateRunner)
//! — and the runner is the ONLY thing that knows how to actually migrate: it is the
//! product's own binary, invoked as `--migrate-config from=X to=Y`. On success the
//! new head is stamped; a failed hop leaves the stamp untouched (so a retry replays
//! from the same point).
//!
//! ZERO-SHELL: the production runner ([`ProductBinRunner`](crate::migrate::ProductBinRunner)) `exec`s the product's own
//! CLI verb directly (argv, never `sh -c`) — the same direct-exec a systemd unit's
//! `ExecStart=` does. It is not shelling a system tool; it is the product running its
//! own migration. Tests inject a [`RecordingRunner`](crate::migrate::RecordingRunner) and assert the exact hops without
//! spawning anything.
use std::path::{Path, PathBuf};
use anyhow::{Context, Result, bail};
use serde::{Deserialize, Serialize};
/// The schema-version stamp written beside a product's data dir. It records the
/// schema version the **data on disk** currently speaks, so a later `refresh` knows
/// where in the product's migration chain to resume.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct MigrationStamp {
/// The product this stamp belongs to (guards a mis-placed stamp).
pub product: String,
/// The schema version the data currently speaks — an entry in the product's
/// ordered migration chain.
pub schema_version: String,
/// RFC3339 UTC instant the stamp was last written.
pub stamped_at: String,
}
impl MigrationStamp {
/// A fresh stamp for `product` at `schema_version`, timestamped now.
pub fn now(product: impl Into<String>, schema_version: impl Into<String>) -> Self {
MigrationStamp {
product: product.into(),
schema_version: schema_version.into(),
stamped_at: now_rfc3339(),
}
}
}
/// The stamp file path beside a product's `data` dir: `<data_dir>/.migration-stamp.json`.
/// A dotfile so it never collides with product data, and INSIDE the data root so the
/// uninstall law erases it with the data (a wiped product must forget its schema too).
pub fn stamp_path(data_dir: &Path) -> PathBuf {
data_dir.join(".migration-stamp.json")
}
/// Read the stamp beside `data_dir`, or `None` if none has ever been written.
pub fn read_stamp(data_dir: &Path) -> Result<Option<MigrationStamp>> {
let p = stamp_path(data_dir);
if !p.exists() {
return Ok(None);
}
let bytes =
std::fs::read(&p).with_context(|| format!("read migration stamp {}", p.display()))?;
let stamp: MigrationStamp = serde_json::from_slice(&bytes)
.with_context(|| format!("parse migration stamp {}", p.display()))?;
Ok(Some(stamp))
}
/// Write (create-or-replace) the stamp beside `data_dir`. The `data_dir` is created
/// if absent so a first install can stamp before any data is written.
pub fn write_stamp(data_dir: &Path, stamp: &MigrationStamp) -> Result<PathBuf> {
std::fs::create_dir_all(data_dir)
.with_context(|| format!("ensure data dir {} for stamp", data_dir.display()))?;
let p = stamp_path(data_dir);
let json = serde_json::to_vec_pretty(stamp).context("serialize migration stamp")?;
std::fs::write(&p, json).with_context(|| format!("write migration stamp {}", p.display()))?;
Ok(p)
}
/// Diff the on-disk `stamped` schema version against a product's ordered `chain`
/// (oldest → head) and return the ordered `(from, to)` hops to replay.
///
/// - `chain` must be non-empty; its LAST entry is the head (current) schema.
/// - `stamped == head` ⇒ `[]` (nothing to do — the idempotency guarantee).
/// - `stamped` earlier in the chain ⇒ each consecutive hop from `stamped` to head.
/// - `stamped` NOT in the chain ⇒ **error** (a stamp the product does not recognize —
/// e.g. a downgrade, or corruption — must never be silently "migrated forward").
pub fn migration_steps(stamped: &str, chain: &[String]) -> Result<Vec<(String, String)>> {
let head = chain.last().ok_or_else(|| {
anyhow::anyhow!(
"migration chain is empty — a product must declare at least its head schema"
)
})?;
if stamped == head {
return Ok(Vec::new());
}
let from_idx = chain.iter().position(|v| v == stamped).ok_or_else(|| {
anyhow::anyhow!(
"stamped schema `{stamped}` is not in the product's migration chain {chain:?} — refusing to migrate (downgrade or corruption)"
)
})?;
// Consecutive hops from the stamped index to the head.
let hops = chain[from_idx..]
.windows(2)
.map(|w| (w[0].clone(), w[1].clone()))
.collect();
Ok(hops)
}
/// The seam that actually performs ONE migration hop. Production is [`ProductBinRunner`]
/// (execs the product's own `--migrate-config from=X to=Y`); tests use [`RecordingRunner`].
pub trait MigrateRunner {
/// Migrate the product's config/data from schema `from` to schema `to`. The
/// implementation owns the how; a hop that returns `Err` aborts the replay with
/// the stamp left at its pre-hop value (so a retry resumes from the same point).
fn migrate(&self, product: &str, from: &str, to: &str) -> Result<()>;
}
/// PRODUCTION runner — invokes the freshly-installed product binary's own migration
/// verb: `<bin> --migrate-config from=<from> to=<to>`. Direct argv exec (NOT a shell):
/// the product runs its own Rust migration, Skidbladnir only orchestrates the order.
/// `extra_args` are prepended (e.g. a `--user`/`--system` scope the product needs to
/// resolve its data root the same way the running service does).
pub struct ProductBinRunner {
/// Absolute path to the product binary to run the migration (the NEW generation's
/// binary, in its version dir — migrations are a property of the incoming code).
pub bin: PathBuf,
/// Scope/context args the product needs to find the same data root as its service.
pub extra_args: Vec<String>,
}
impl ProductBinRunner {
pub fn new(bin: impl Into<PathBuf>) -> Self {
ProductBinRunner {
bin: bin.into(),
extra_args: Vec::new(),
}
}
pub fn with_args(mut self, args: Vec<String>) -> Self {
self.extra_args = args;
self
}
}
impl MigrateRunner for ProductBinRunner {
fn migrate(&self, product: &str, from: &str, to: &str) -> Result<()> {
// Direct argv exec of the product's own CLI verb (zero-shell — no `sh -c`,
// no interpreter): the product forward-migrates its own config.
let mut cmd = std::process::Command::new(&self.bin);
cmd.args(&self.extra_args);
cmd.arg("--migrate-config");
cmd.arg(format!("from={from}"));
cmd.arg(format!("to={to}"));
let status = cmd.status().with_context(|| {
format!(
"exec `{} --migrate-config from={from} to={to}` for `{product}`",
self.bin.display()
)
})?;
if !status.success() {
bail!(
"`{} --migrate-config from={from} to={to}` for `{product}` exited with {status}",
self.bin.display()
);
}
Ok(())
}
}
/// TEST runner — records every hop it is asked to run (and can be told to FAIL at a
/// chosen `to`, to exercise the abort-leaves-stamp-untouched path).
#[derive(Debug, Default)]
pub struct RecordingRunner {
/// The `(product, from, to)` hops seen, in order.
pub seen: std::cell::RefCell<Vec<(String, String, String)>>,
/// If `Some(to)`, the hop whose `to` equals this fails.
pub fail_at: Option<String>,
}
impl RecordingRunner {
pub fn new() -> Self {
Self::default()
}
pub fn failing_at(to: impl Into<String>) -> Self {
RecordingRunner {
seen: Default::default(),
fail_at: Some(to.into()),
}
}
/// The recorded hops as `(from, to)` pairs.
pub fn hops(&self) -> Vec<(String, String)> {
self.seen
.borrow()
.iter()
.map(|(_, f, t)| (f.clone(), t.clone()))
.collect()
}
}
impl MigrateRunner for RecordingRunner {
fn migrate(&self, product: &str, from: &str, to: &str) -> Result<()> {
self.seen
.borrow_mut()
.push((product.into(), from.into(), to.into()));
if self.fail_at.as_deref() == Some(to) {
bail!("RecordingRunner: injected failure at hop {from}→{to}");
}
Ok(())
}
}
/// What a [`replay_migrations`] run did — fully inspectable (the inject-assert law).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MigrateReport {
pub product: String,
/// The schema the data spoke before this replay (`None` = never stamped).
pub from_stamp: Option<String>,
/// The head schema the chain targets.
pub head: String,
/// The hops actually replayed, in order.
pub replayed: Vec<(String, String)>,
/// The schema now stamped after this replay (head on success; the last good hop's
/// `to` if a hop failed; the prior stamp if nothing ran).
pub now_stamped: String,
/// `true` when the data was already at head and nothing ran (idempotent no-op).
pub was_noop: bool,
}
/// Replay the migration chain for `product` against the data at `data_dir`, using
/// `runner` to perform each hop, then stamp the new head. THE orchestration:
///
/// 1. Read the stamp beside `data_dir`. **Absent** ⇒ treat as a brand-new data root:
/// stamp the head and run nothing (a fresh install has nothing to forward-migrate).
/// A product introducing the stamp onto a pre-existing deployment must backfill the
/// stamp first — this is stated honestly, not papered over.
/// 2. Compute [`migration_steps`] (stamp → head). Empty ⇒ idempotent no-op.
/// 3. Run each hop through `runner`. A hop error aborts the replay and stamps the LAST
/// SUCCESSFUL schema (so a retry resumes precisely), then returns the error.
/// 4. On full success, stamp the head.
pub fn replay_migrations(
product: &str,
data_dir: &Path,
chain: &[String],
runner: &dyn MigrateRunner,
) -> Result<MigrateReport> {
let head = chain
.last()
.ok_or_else(|| anyhow::anyhow!("migration chain for `{product}` is empty"))?
.clone();
let from_stamp = read_stamp(data_dir)?.map(|s| s.schema_version);
// Brand-new data root: stamp head, migrate nothing.
let Some(stamped) = from_stamp.clone() else {
write_stamp(data_dir, &MigrationStamp::now(product, &head))?;
return Ok(MigrateReport {
product: product.into(),
from_stamp: None,
head: head.clone(),
replayed: Vec::new(),
now_stamped: head,
was_noop: true,
});
};
let hops = migration_steps(&stamped, chain)?;
if hops.is_empty() {
return Ok(MigrateReport {
product: product.into(),
from_stamp: Some(stamped.clone()),
head: head.clone(),
replayed: Vec::new(),
now_stamped: stamped,
was_noop: true,
});
}
let mut replayed = Vec::new();
let mut good = stamped.clone();
for (from, to) in &hops {
match runner.migrate(product, from, to) {
Ok(()) => {
replayed.push((from.clone(), to.clone()));
good = to.clone();
// Stamp after every successful hop so an abort resumes precisely.
write_stamp(data_dir, &MigrationStamp::now(product, &good))?;
}
Err(e) => {
// Leave the stamp at the last good schema (already written above).
return Err(e.context(format!(
"migration `{product}` failed at hop {from}→{to} (data left stamped at `{good}`)"
)));
}
}
}
Ok(MigrateReport {
product: product.into(),
from_stamp: Some(stamped),
head: head.clone(),
replayed,
now_stamped: good,
was_noop: false,
})
}
/// RFC3339 UTC now, with a std-only fallback (no chrono dep pulled just for this —
/// the crate has chrono behind other paths, but the stamp must work in the lean core).
fn now_rfc3339() -> String {
// chrono is already a dep of the crate (layout date-stamping); reuse it.
chrono::Utc::now().to_rfc3339()
}
#[cfg(test)]
mod tests {
use super::*;
fn chain(v: &[&str]) -> Vec<String> {
v.iter().map(|s| s.to_string()).collect()
}
/// PURE step computation — the heart of the chain model.
#[test]
fn steps_from_stamp_to_head() {
let c = chain(&["v0", "v1", "v2"]);
// stamped at the oldest → replay every hop to head
assert_eq!(
migration_steps("v0", &c).unwrap(),
vec![("v0".into(), "v1".into()), ("v1".into(), "v2".into())]
);
// stamped mid-chain → only the remaining hops
assert_eq!(
migration_steps("v1", &c).unwrap(),
vec![("v1".into(), "v2".into())]
);
// stamped AT head → nothing (idempotency)
assert!(migration_steps("v2", &c).unwrap().is_empty());
}
/// A stamp outside the chain (downgrade / corruption) is a hard error.
#[test]
fn unknown_stamp_is_refused() {
let c = chain(&["v0", "v1"]);
assert!(migration_steps("v9", &c).is_err());
// empty chain is refused too
assert!(migration_steps("v0", &[]).is_err());
}
/// Absent stamp ⇒ brand-new: head is stamped, nothing replays.
#[test]
fn absent_stamp_is_fresh_install() {
let dir = tempfile::tempdir().unwrap();
let c = chain(&["v0", "v1", "v2"]);
let runner = RecordingRunner::new();
let rep = replay_migrations("korp", dir.path(), &c, &runner).unwrap();
assert!(rep.was_noop);
assert_eq!(rep.now_stamped, "v2");
assert!(runner.hops().is_empty(), "a fresh install migrates nothing");
// …and the stamp is now on disk at head.
assert_eq!(
read_stamp(dir.path()).unwrap().unwrap().schema_version,
"v2"
);
}
/// The full replay: a data root stamped at an old schema is walked to head, in
/// order, and re-stamped — then a SECOND replay is a proven no-op (idempotent).
#[test]
fn replay_walks_the_chain_then_is_idempotent() {
let dir = tempfile::tempdir().unwrap();
let c = chain(&["v0", "v1", "v2"]);
// Pretend an older install left the data at v0.
write_stamp(dir.path(), &MigrationStamp::now("korp", "v0")).unwrap();
let runner = RecordingRunner::new();
let rep = replay_migrations("korp", dir.path(), &c, &runner).unwrap();
assert!(!rep.was_noop);
assert_eq!(
rep.replayed,
vec![("v0".into(), "v1".into()), ("v1".into(), "v2".into())]
);
assert_eq!(runner.hops(), rep.replayed);
assert_eq!(rep.now_stamped, "v2");
assert_eq!(
read_stamp(dir.path()).unwrap().unwrap().schema_version,
"v2"
);
// Refresh again against the SAME head → zero hops, stamp unchanged.
let runner2 = RecordingRunner::new();
let rep2 = replay_migrations("korp", dir.path(), &c, &runner2).unwrap();
assert!(rep2.was_noop);
assert!(runner2.hops().is_empty());
assert_eq!(rep2.now_stamped, "v2");
}
/// A failing hop aborts the replay and LEAVES THE STAMP at the last good schema,
/// so a retry resumes from exactly there (no double-apply, no skip).
#[test]
fn failed_hop_leaves_resumable_stamp() {
let dir = tempfile::tempdir().unwrap();
let c = chain(&["v0", "v1", "v2", "v3"]);
write_stamp(dir.path(), &MigrationStamp::now("korp", "v0")).unwrap();
// Fail on the hop INTO v2 → v0→v1 succeeds, v1→v2 fails.
let runner = RecordingRunner::failing_at("v2");
let err = replay_migrations("korp", dir.path(), &c, &runner).unwrap_err();
assert!(format!("{err:#}").contains("v1→v2"), "err: {err:#}");
// The good hop was stamped; the failed one was not.
assert_eq!(
read_stamp(dir.path()).unwrap().unwrap().schema_version,
"v1"
);
// Retry with a healthy runner → resumes at v1, finishes to v3.
let runner2 = RecordingRunner::new();
let rep = replay_migrations("korp", dir.path(), &c, &runner2).unwrap();
assert_eq!(
rep.replayed,
vec![("v1".into(), "v2".into()), ("v2".into(), "v3".into())]
);
assert_eq!(rep.now_stamped, "v3");
}
}