crates/nornir-release-ledger/src/release_changes.rs
//! Arrow writer + reader for the **reversible release-change ledger**
//! (`release_changes`, task #32).
//!
//! Every MUTATING step of a release run records one [`ReleaseChange`] row WITH
//! the before-state needed to reverse it: a version bump carries the exact
//! pre-bump manifest text; a patch-strip carries the removed `[patch]` block; a
//! branch cut carries the SHA HEAD was at; a publish carries whether it landed on
//! an immutable registry (crates.io → can only be yanked, never un-published).
//!
//! A `release_id` groups one run's records. [`ChangeRecorder::record`] appends one
//! row (best-effort, but here it IS the safety net so callers should surface a
//! write error). [`query_release_changes`] reads a run's rows; `release undo`
//! reverses them newest-first (`seq` descending).
//!
//! Everything here reasons over the engine-agnostic
//! [`nornir_warehouse_trait::Warehouse`] seam (SYNC `append_arrow`/`scan_arrow`)
//! — never a concrete `IcebergWarehouse`. Schema:
//! [`crate::schema::release_changes`].
use std::sync::Arc;
use std::sync::atomic::{AtomicI64, Ordering};
use anyhow::{anyhow, Result};
use arrow::array::{
Array, BooleanArray, Int64Array, RecordBatch, StringArray, TimestampMicrosecondArray,
};
use iceberg::arrow::schema_to_arrow_schema;
use sha2::{Digest, Sha256};
use uuid::Uuid;
use nornir_warehouse_trait::Warehouse;
use crate::schema::TABLE_RELEASE_CHANGES;
// Column indices match `schema::release_changes` field order. `pub(crate)` so
// the schema module can assert index↔name agreement.
pub(crate) const COL_RELEASE_ID: usize = 0;
pub(crate) const COL_SEQ: usize = 1;
pub(crate) const COL_TS_MICROS: usize = 2;
pub(crate) const COL_KIND: usize = 3;
pub(crate) const COL_REPO: usize = 4;
pub(crate) const COL_CHANGE_SHA: usize = 5;
pub(crate) const COL_CRATE_NAME: usize = 6;
pub(crate) const COL_OLD_VERSION: usize = 7;
pub(crate) const COL_NEW_VERSION: usize = 8;
pub(crate) const COL_FILE: usize = 9;
pub(crate) const COL_OLD_REQ_LINES: usize = 10;
pub(crate) const COL_REMOVED_BLOCK: usize = 11;
pub(crate) const COL_BRANCH: usize = 12;
pub(crate) const COL_PREV_HEAD: usize = 13;
pub(crate) const COL_REGISTRY: usize = 14;
pub(crate) const COL_IMMUTABLE: usize = 15;
/// Canonical `kind` tags.
pub mod kind {
pub const BUMP: &str = "bump";
pub const PATCH_STRIP: &str = "patch_strip";
pub const BRANCH: &str = "branch";
pub const PUBLISH: &str = "publish";
}
/// The reversible payload of one mutating release step. Only the fields relevant
/// to a `kind` are populated; the rest are `None`.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum ChangeKind {
/// A coordinated version bump in one manifest. `old_file_text` is the WHOLE
/// pre-bump manifest so undo restores it byte-for-byte.
Bump {
crate_name: String,
old_version: String,
new_version: String,
file: String,
old_file_text: String,
},
/// A `[patch.crates-io]` strip from one manifest. `removed_block` is the
/// stripped text (the lost lines) — undo re-appends it.
PatchStrip { file: String, removed_block: String },
/// A release-branch cut. `prev_head` is the SHA HEAD pointed at before (empty
/// when the branch was freshly created → undo deletes it).
Branch { repo: String, branch: String, prev_head: String },
/// A publish. `immutable` is true for crates.io (undo → yank, never
/// un-publish); false for a /sparring rehearsal.
Publish { crate_name: String, version: String, registry: String, immutable: bool },
}
/// One row of the `release_changes` ledger.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct ReleaseChange {
/// The release run these records group under.
pub release_id: String,
/// Per-run monotonic sequence — the durable ordering key (undo reverses by
/// `seq` descending).
pub seq: i64,
/// Event time, microseconds since the Unix epoch (UTC).
pub ts_micros: i64,
/// The repo this change touched (the workspace member / checkout name).
pub repo: String,
/// sha256 of the canonical payload (integrity + dedup).
pub change_sha: String,
/// The reversible payload.
pub change: ChangeKind,
}
impl ReleaseChange {
/// `(release_id, seq)` sort key.
fn key(&self) -> (String, i64) {
(self.release_id.clone(), self.seq)
}
}
/// Mints monotonic `seq` for one release run's change records.
#[derive(Clone)]
pub struct ChangeRecorder {
release_id: String,
seq: Arc<AtomicI64>,
}
impl ChangeRecorder {
pub fn new(release_id: impl Into<String>) -> Self {
Self { release_id: release_id.into(), seq: Arc::new(AtomicI64::new(0)) }
}
pub fn release_id(&self) -> &str {
&self.release_id
}
/// Build the next [`ReleaseChange`] (stamped now, fresh `seq`, computed
/// `change_sha`) WITHOUT writing it.
pub fn build(&self, repo: &str, change: ChangeKind) -> ReleaseChange {
let seq = self.seq.fetch_add(1, Ordering::SeqCst);
let change_sha = change_sha(&change);
ReleaseChange {
release_id: self.release_id.clone(),
seq,
ts_micros: chrono::Utc::now().timestamp_micros(),
repo: repo.to_string(),
change_sha,
change,
}
}
/// Build + append one change record to the warehouse. Sync: routes through
/// the backend-neutral [`Warehouse::append_arrow`] (which blocks internally),
/// so no `block_on` here.
pub fn record(&self, wh: &dyn Warehouse, repo: &str, change: ChangeKind) -> Result<()> {
let row = self.build(repo, change);
append_release_changes(wh, std::slice::from_ref(&row))
}
}
/// sha256 of a change's canonical payload — `kind|field|field|…`.
///
/// `pub` (it was private in both forked copies) so a consumer can verify a row's
/// integrity without re-deriving the canonicalisation — re-deriving it is exactly
/// how the two copies would drift apart again.
pub fn change_sha(change: &ChangeKind) -> String {
let mut h = Sha256::new();
match change {
ChangeKind::Bump { crate_name, old_version, new_version, file, old_file_text } => {
h.update(b"bump\0");
for p in [crate_name, old_version, new_version, file, old_file_text] {
h.update(p.as_bytes());
h.update(b"\0");
}
}
ChangeKind::PatchStrip { file, removed_block } => {
h.update(b"patch_strip\0");
h.update(file.as_bytes());
h.update(b"\0");
h.update(removed_block.as_bytes());
}
ChangeKind::Branch { repo, branch, prev_head } => {
h.update(b"branch\0");
for p in [repo, branch, prev_head] {
h.update(p.as_bytes());
h.update(b"\0");
}
}
ChangeKind::Publish { crate_name, version, registry, immutable } => {
h.update(b"publish\0");
for p in [crate_name, version, registry] {
h.update(p.as_bytes());
h.update(b"\0");
}
h.update(if *immutable { b"1" } else { b"0" });
}
}
h.finalize().iter().map(|b| format!("{b:02x}")).collect::<String>()
}
/// Append change records (one Iceberg snapshot). Sync — routes through the
/// backend-neutral [`Warehouse::append_arrow`]. The batch is built against the
/// CANONICAL `release_changes` schema; the backend recasts to the stored schema.
pub fn append_release_changes(wh: &dyn Warehouse, rows: &[ReleaseChange]) -> Result<()> {
if rows.is_empty() {
return Ok(());
}
let arrow_schema = Arc::new(schema_to_arrow_schema(&crate::schema::release_changes()?)?);
// Project each row's optional fields by kind.
let kind_of = |c: &ChangeKind| match c {
ChangeKind::Bump { .. } => kind::BUMP,
ChangeKind::PatchStrip { .. } => kind::PATCH_STRIP,
ChangeKind::Branch { .. } => kind::BRANCH,
ChangeKind::Publish { .. } => kind::PUBLISH,
};
let crate_name = |c: &ChangeKind| match c {
ChangeKind::Bump { crate_name, .. } | ChangeKind::Publish { crate_name, .. } => {
Some(crate_name.clone())
}
_ => None,
};
let old_version = |c: &ChangeKind| match c {
ChangeKind::Bump { old_version, .. } => Some(old_version.clone()),
_ => None,
};
let new_version = |c: &ChangeKind| match c {
ChangeKind::Bump { new_version, .. } => Some(new_version.clone()),
ChangeKind::Publish { version, .. } => Some(version.clone()),
_ => None,
};
let file = |c: &ChangeKind| match c {
ChangeKind::Bump { file, .. } | ChangeKind::PatchStrip { file, .. } => Some(file.clone()),
_ => None,
};
let old_req_lines = |c: &ChangeKind| match c {
ChangeKind::Bump { old_file_text, .. } => Some(old_file_text.clone()),
_ => None,
};
let removed_block = |c: &ChangeKind| match c {
ChangeKind::PatchStrip { removed_block, .. } => Some(removed_block.clone()),
_ => None,
};
let branch = |c: &ChangeKind| match c {
ChangeKind::Branch { branch, .. } => Some(branch.clone()),
_ => None,
};
let prev_head = |c: &ChangeKind| match c {
ChangeKind::Branch { prev_head, .. } => Some(prev_head.clone()),
_ => None,
};
let registry = |c: &ChangeKind| match c {
ChangeKind::Publish { registry, .. } => Some(registry.clone()),
_ => None,
};
let immutable = |c: &ChangeKind| match c {
ChangeKind::Publish { immutable, .. } => Some(*immutable),
_ => None,
};
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(rows.iter().map(|r| r.release_id.clone()).collect::<Vec<_>>())),
Arc::new(Int64Array::from(rows.iter().map(|r| r.seq).collect::<Vec<_>>())),
Arc::new(
TimestampMicrosecondArray::from(rows.iter().map(|r| r.ts_micros).collect::<Vec<_>>())
.with_timezone("+00:00"),
),
Arc::new(StringArray::from(rows.iter().map(|r| kind_of(&r.change).to_string()).collect::<Vec<_>>())),
Arc::new(StringArray::from(rows.iter().map(|r| r.repo.clone()).collect::<Vec<_>>())),
Arc::new(StringArray::from(rows.iter().map(|r| r.change_sha.clone()).collect::<Vec<_>>())),
Arc::new(StringArray::from(rows.iter().map(|r| crate_name(&r.change)).collect::<Vec<_>>())),
Arc::new(StringArray::from(rows.iter().map(|r| old_version(&r.change)).collect::<Vec<_>>())),
Arc::new(StringArray::from(rows.iter().map(|r| new_version(&r.change)).collect::<Vec<_>>())),
Arc::new(StringArray::from(rows.iter().map(|r| file(&r.change)).collect::<Vec<_>>())),
Arc::new(StringArray::from(rows.iter().map(|r| old_req_lines(&r.change)).collect::<Vec<_>>())),
Arc::new(StringArray::from(rows.iter().map(|r| removed_block(&r.change)).collect::<Vec<_>>())),
Arc::new(StringArray::from(rows.iter().map(|r| branch(&r.change)).collect::<Vec<_>>())),
Arc::new(StringArray::from(rows.iter().map(|r| prev_head(&r.change)).collect::<Vec<_>>())),
Arc::new(StringArray::from(rows.iter().map(|r| registry(&r.change)).collect::<Vec<_>>())),
Arc::new(BooleanArray::from(rows.iter().map(|r| immutable(&r.change)).collect::<Vec<_>>())),
];
let batch = RecordBatch::try_new(arrow_schema, cols)?;
wh.append_arrow(TABLE_RELEASE_CHANGES, batch)?;
Ok(())
}
/// Read every change record for `release_id`, sorted by `(release_id, seq)`.
/// Pass `None` to read every run, then pick the most recent by max `ts_micros`.
/// Sync — routes through the backend-neutral [`Warehouse::scan_arrow`].
pub fn query_release_changes(
wh: &dyn Warehouse,
release_id: Option<&str>,
) -> Result<Vec<ReleaseChange>> {
// Blank/torn Iceberg metadata JSON → 0 rows: `scan_arrow` degrades a torn
// snapshot to an empty scan (never a JSON-EOF crash).
let batches: Vec<RecordBatch> = wh.scan_arrow(TABLE_RELEASE_CHANGES)?;
let mut out: Vec<ReleaseChange> = Vec::new();
for b in &batches {
let rid = col_str(b, COL_RELEASE_ID)?;
let seq = col_i64(b, COL_SEQ)?;
let ts = col_ts(b, COL_TS_MICROS)?;
let kind = col_str(b, COL_KIND)?;
let repo = col_str(b, COL_REPO)?;
let sha = col_str(b, COL_CHANGE_SHA)?;
let crate_name = col_str(b, COL_CRATE_NAME)?;
let old_version = col_str(b, COL_OLD_VERSION)?;
let new_version = col_str(b, COL_NEW_VERSION)?;
let file = col_str(b, COL_FILE)?;
let old_req = col_str(b, COL_OLD_REQ_LINES)?;
let removed = col_str(b, COL_REMOVED_BLOCK)?;
let branch = col_str(b, COL_BRANCH)?;
let prev_head = col_str(b, COL_PREV_HEAD)?;
let registry = col_str(b, COL_REGISTRY)?;
let immutable = col_bool(b, COL_IMMUTABLE)?;
let s = |a: &StringArray, i: usize| if a.is_null(i) { String::new() } else { a.value(i).to_string() };
for i in 0..b.num_rows() {
let change = match kind.value(i) {
kind::BUMP => ChangeKind::Bump {
crate_name: s(crate_name, i),
old_version: s(old_version, i),
new_version: s(new_version, i),
file: s(file, i),
old_file_text: s(old_req, i),
},
kind::PATCH_STRIP => ChangeKind::PatchStrip {
file: s(file, i),
removed_block: s(removed, i),
},
kind::BRANCH => ChangeKind::Branch {
repo: s(repo, i),
branch: s(branch, i),
prev_head: s(prev_head, i),
},
kind::PUBLISH => ChangeKind::Publish {
crate_name: s(crate_name, i),
version: s(new_version, i),
registry: s(registry, i),
immutable: !immutable.is_null(i) && immutable.value(i),
},
other => return Err(anyhow!("unknown release_change kind `{other}`")),
};
let row = ReleaseChange {
release_id: rid.value(i).to_string(),
seq: seq.value(i),
ts_micros: ts.value(i),
repo: repo.value(i).to_string(),
change_sha: sha.value(i).to_string(),
change,
};
if release_id.map(|id| row.release_id == id).unwrap_or(true) {
out.push(row);
}
}
}
out.sort_by_key(|r| r.key());
Ok(out)
}
/// The `release_id` of the most recent run in the ledger (max `ts_micros`), or
/// `None` if the ledger is empty.
pub fn latest_release_id(wh: &dyn Warehouse) -> Result<Option<String>> {
let all = query_release_changes(wh, None)?;
Ok(all.iter().max_by_key(|r| r.ts_micros).map(|r| r.release_id.clone()))
}
/// A fresh release id (UUIDv4).
pub fn new_release_id() -> String {
Uuid::new_v4().to_string()
}
// ─── column helpers ─────────────────────────────────────────────────────
fn col_str<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a StringArray> {
b.column(idx)
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| anyhow!("release_changes col {idx} is not StringArray"))
}
fn col_i64<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a Int64Array> {
b.column(idx)
.as_any()
.downcast_ref::<Int64Array>()
.ok_or_else(|| anyhow!("release_changes col {idx} is not Int64Array"))
}
fn col_ts<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a TimestampMicrosecondArray> {
b.column(idx)
.as_any()
.downcast_ref::<TimestampMicrosecondArray>()
.ok_or_else(|| anyhow!("release_changes col {idx} is not TimestampMicrosecondArray"))
}
fn col_bool<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a BooleanArray> {
b.column(idx)
.as_any()
.downcast_ref::<BooleanArray>()
.ok_or_else(|| anyhow!("release_changes col {idx} is not BooleanArray"))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::testkit::MockWarehouse;
#[test]
fn append_and_query_round_trips_each_kind() {
let wh = MockWarehouse::new();
let rec = ChangeRecorder::new("run-xyz");
let bump = rec.build("skade", ChangeKind::Bump {
crate_name: "arrow".into(),
old_version: "57".into(),
new_version: "58".into(),
file: "Cargo.toml".into(),
old_file_text: "[dependencies]\narrow = \"57\"\n".into(),
});
let strip = rec.build("nornir", ChangeKind::PatchStrip {
file: "Cargo.toml".into(),
removed_block: "[patch.crates-io]\niceberg = { path = \"../fork\" }\n".into(),
});
let branch = rec.build("nornir", ChangeKind::Branch {
repo: "nornir".into(),
branch: "release/staging".into(),
prev_head: "deadbeef".into(),
});
let publish = rec.build("nornir", ChangeKind::Publish {
crate_name: "nornir".into(),
version: "0.2.0".into(),
registry: "crates.io".into(),
immutable: true,
});
append_release_changes(&wh, &[bump.clone(), strip.clone(), branch.clone(), publish.clone()])
.unwrap();
let rows = query_release_changes(&wh, Some("run-xyz")).unwrap();
assert_eq!(rows.len(), 4);
assert_eq!(rows.iter().map(|r| r.seq).collect::<Vec<_>>(), vec![0, 1, 2, 3]);
assert_eq!(rows[0].change, bump.change);
assert_eq!(rows[1].change, strip.change);
assert_eq!(rows[2].change, branch.change);
assert_eq!(rows[3].change, publish.change);
// change_sha is stable + non-empty.
assert!(!rows[0].change_sha.is_empty());
assert_eq!(rows[0].change_sha, change_sha(&bump.change));
// Latest-run resolution.
let latest = latest_release_id(&wh).unwrap();
assert_eq!(latest.as_deref(), Some("run-xyz"));
}
/// RED-when-broken: the canonical payload is domain-separated per kind, so
/// two DIFFERENT changes can never collide on `change_sha` (dedup would
/// silently drop one), and the same change always hashes the same.
#[test]
fn change_sha_is_deterministic_and_kind_separated() {
let a = ChangeKind::Branch {
repo: "nornir".into(),
branch: "release/staging".into(),
prev_head: String::new(),
};
assert_eq!(change_sha(&a), change_sha(&a.clone()), "same payload ⇒ same sha");
let b = ChangeKind::Branch {
repo: "nornir".into(),
branch: "release/staging".into(),
prev_head: "deadbeef".into(),
};
assert_ne!(change_sha(&a), change_sha(&b), "prev_head is part of the payload");
// Same field text, different kind ⇒ different sha (the `kind\0` prefix).
let strip = ChangeKind::PatchStrip { file: "x".into(), removed_block: "y".into() };
let bump = ChangeKind::Bump {
crate_name: "x".into(),
old_version: "y".into(),
new_version: String::new(),
file: String::new(),
old_file_text: String::new(),
};
assert_ne!(change_sha(&strip), change_sha(&bump));
}
}