crates/nornir-release-ledger/src/schema.rs
//! The CANONICAL `::iceberg::spec::Schema` builders for the three release
//! tables — the single shape source of truth for the whole constellation.
//!
//! These were forked into two hand-maintained copies (nornir's
//! `warehouse/iceberg_schema.rs` and dwarves' `release_doctor/schema.rs`) and
//! that fork was the sharpest hazard in the seam: both halves write into the
//! SAME warehouse, so a column added to one copy and not the other produces
//! mutually incompatible tables with **nothing failing at compile time**. There
//! is now exactly one definition and both halves re-export it.
//!
//! Field IDs are explicit and stable — append a new field with a fresh, unused
//! ID; never reuse, never renumber (the Iceberg schema-evolution invariant).
use std::sync::Arc;
use anyhow::Result;
use iceberg::spec::{NestedField, PrimitiveType, Schema, Type};
// The KNOWLEDGE-map table names are not this leaf's to define — they belong to
// the `Warehouse` seam itself, where every reader and the owner's scanners can
// take them from one place. Re-exported here only so existing
// `schema::TABLE_SYMBOL_FACTS` paths keep resolving.
pub use nornir_warehouse_trait::{
TABLE_CALL_EDGES, TABLE_SCIP_OCCURRENCES, TABLE_SYMBOL_FACTS,
};
/// The `release_lineage` table name (per-`(release, repo)` outcome ledger).
pub const TABLE_RELEASE_LINEAGE: &str = "release_lineage";
/// The `release_events` table name (the release-op DAG: one row per
/// component×op×phase boundary). Read by `release events` and the release viz
/// panes.
pub const TABLE_RELEASE_EVENTS: &str = "release_events";
/// The `release_changes` table name (the reversible release-change ledger: one
/// row per mutating release step, with enough before-state to undo it —
/// SHA-keyed, grouped by `release_id`).
pub const TABLE_RELEASE_CHANGES: &str = "release_changes";
/// One row per (release, repo). A release's full provenance — which
/// git SHAs of which repos went into it, against which dep-graph
/// snapshot, in which order, and with what gate outcome.
///
/// `gate_status` is per-repo so a partial release that only got
/// through `znippy` before tripping a gate on `holger` is fully
/// reconstructable.
///
/// The time-machine pins let a release attach immutable derived artifacts
/// captured at the same `git_sha`, so the release is *reconstructable* and not
/// merely described.
///
/// `published_versions_csv` is `crate@version,crate@version` for the
/// crates this repo actually shipped (empty for dry-runs / failed
/// gates).
pub fn release_lineage() -> Result<Schema> {
Ok(Schema::builder()
.with_schema_id(0)
.with_fields(vec![
Arc::new(NestedField::required(1, "release_id", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::required(2, "workspace_name", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::required(3, "ts_micros", Type::Primitive(PrimitiveType::Timestamptz))),
Arc::new(NestedField::required(4, "dep_graph_snapshot_id", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::required(5, "build_order_idx", Type::Primitive(PrimitiveType::Int))),
Arc::new(NestedField::required(6, "repo", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::required(7, "git_sha", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::required(8, "git_branch", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::required(9, "git_dirty", Type::Primitive(PrimitiveType::Boolean))),
Arc::new(NestedField::required(10, "gate_status", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::required(11, "tests_passed", Type::Primitive(PrimitiveType::Int))),
Arc::new(NestedField::required(12, "tests_failed", Type::Primitive(PrimitiveType::Int))),
Arc::new(NestedField::required(13, "dry_run", Type::Primitive(PrimitiveType::Boolean))),
Arc::new(NestedField::required(14, "published_versions_csv", Type::Primitive(PrimitiveType::String))),
// Time-machine pins: each release can attach immutable
// derived artifacts captured at the same `git_sha`. Nullable
// because (a) the table predates the time machine and
// (b) not every artifact kind exists for every repo.
Arc::new(NestedField::optional(15, "tantivy_snapshot_id", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::optional(16, "dwarf_snapshot_id", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::optional(17, "gimli_snapshot_id", Type::Primitive(PrimitiveType::String))),
// Cargo-pipeline pins.
Arc::new(NestedField::optional(18, "git_tag", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::optional(19, "git_tag_matches_published", Type::Primitive(PrimitiveType::Boolean))),
Arc::new(NestedField::optional(20, "rustdoc_json_snapshot_id", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::optional(21, "patch_lines_stripped", Type::Primitive(PrimitiveType::Int))),
])
.build()?)
}
/// Release events: a clean row-level DAG of a release run.
///
/// Where `release_lineage` records the *outcome* per `(release, repo)`,
/// this records the *shape* of the operation: one row per
/// `component × op × phase` boundary as the pipeline walks the build
/// order. `depends_on` carries the components this one waited on (the
/// upstream producers from the dep graph), so a reader can draw the live
/// op DAG and see exactly where a partial release stalled — the
/// structural twin of `funnel_events`.
///
/// Grain: one row per emitted boundary. `(run_id, seq)` is the stable
/// ordering key — `seq` is a per-run monotonic counter minted at emit
/// time (Iceberg gives no scan order, so reads sort by `seq`).
///
/// `depends_on` is `list<string>` (the dependency component names);
/// nullable so a no-dep root op writes a null rather than an empty list.
/// `elapsed_ms` is nullable — only `phase = end` rows carry a duration.
pub fn release_events() -> Result<Schema> {
use iceberg::spec::ListType;
let depends_on = Type::List(ListType::new(Arc::new(NestedField::required(
111,
"element",
Type::Primitive(PrimitiveType::String),
))));
Ok(Schema::builder()
.with_schema_id(0)
.with_fields(vec![
Arc::new(NestedField::required(1, "run_id", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::required(2, "seq", Type::Primitive(PrimitiveType::Long))),
Arc::new(NestedField::required(3, "ts_micros", Type::Primitive(PrimitiveType::Timestamptz))),
Arc::new(NestedField::required(4, "component", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::required(5, "repo", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::required(6, "op", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::required(7, "phase", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::required(8, "status", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::required(9, "detail", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::optional(10, "depends_on", depends_on)),
Arc::new(NestedField::optional(11, "elapsed_ms", Type::Primitive(PrimitiveType::Long))),
])
.build()?)
}
/// Reversible release-change ledger (task #32): one row per MUTATING release
/// step, carrying enough BEFORE-STATE to undo it. SHA-keyed (`change_sha` is the
/// sha256 of the canonical payload, for dedup + integrity), grouped by
/// `release_id` (one run's records). `release undo [<id>]` reads a run's
/// rows newest-first (`seq` desc) and reverses each.
///
/// Wide schema — one optional column set per `kind` — so a SQL reader can query
/// the ledger directly. `kind` ∈ `bump | patch_strip | branch | publish`:
/// * `bump` — `crate_name`, `old_version`, `new_version`, `file`,
/// `old_req_lines` (the exact pre-bump manifest text to
/// restore byte-for-byte).
/// * `patch_strip` — `file`, `removed_block` (the stripped `[patch]` text).
/// * `branch` — `repo`, `branch`, `prev_head` (the SHA HEAD was at, or ""
/// when the branch was freshly created → undo deletes it).
/// * `publish` — `crate_name`, `new_version`, `registry`, `immutable`
/// (true for crates.io: undo can't un-publish → yank).
pub fn release_changes() -> Result<Schema> {
Ok(Schema::builder()
.with_schema_id(0)
.with_fields(vec![
Arc::new(NestedField::required(1, "release_id", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::required(2, "seq", Type::Primitive(PrimitiveType::Long))),
Arc::new(NestedField::required(3, "ts_micros", Type::Primitive(PrimitiveType::Timestamptz))),
Arc::new(NestedField::required(4, "kind", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::required(5, "repo", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::required(6, "change_sha", Type::Primitive(PrimitiveType::String))),
// bump
Arc::new(NestedField::optional(7, "crate_name", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::optional(8, "old_version", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::optional(9, "new_version", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::optional(10, "file", Type::Primitive(PrimitiveType::String))),
// The exact pre-bump file text (the whole manifest) so undo restores
// it byte-for-byte. Carried as a string (manifests are UTF-8 TOML).
Arc::new(NestedField::optional(11, "old_req_lines", Type::Primitive(PrimitiveType::String))),
// patch_strip
Arc::new(NestedField::optional(12, "removed_block", Type::Primitive(PrimitiveType::String))),
// branch
Arc::new(NestedField::optional(13, "branch", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::optional(14, "prev_head", Type::Primitive(PrimitiveType::String))),
// publish
Arc::new(NestedField::optional(15, "registry", Type::Primitive(PrimitiveType::String))),
Arc::new(NestedField::optional(16, "immutable", Type::Primitive(PrimitiveType::Boolean))),
])
.build()?)
}
#[cfg(test)]
mod tests {
use super::*;
/// RED-when-broken: the three release tables are the ONE shape both the
/// warehouse owner and the release orchestrator write, so their field
/// count + names + IDs are a contract. Renumbering or dropping a column
/// (the exact accident the fork made invisible) fails here.
#[test]
fn release_table_field_ids_are_stable_and_dense() {
let lineage = release_lineage().unwrap();
let names: Vec<&str> =
lineage.as_struct().fields().iter().map(|f| f.name.as_str()).collect();
assert_eq!(names.len(), 21, "release_lineage column count is a wire contract");
assert_eq!(names[0], "release_id");
assert_eq!(names[13], "published_versions_csv");
assert_eq!(names[20], "patch_lines_stripped");
// IDs are 1..=21 in order — never reused, never renumbered.
let ids: Vec<i32> = lineage.as_struct().fields().iter().map(|f| f.id).collect();
assert_eq!(ids, (1..=21).collect::<Vec<i32>>());
let events = release_events().unwrap();
let ev_names: Vec<&str> =
events.as_struct().fields().iter().map(|f| f.name.as_str()).collect();
assert_eq!(ev_names.len(), 11);
assert_eq!(ev_names[9], "depends_on");
assert_eq!(ev_names[10], "elapsed_ms");
let ev_ids: Vec<i32> = events.as_struct().fields().iter().map(|f| f.id).collect();
assert_eq!(ev_ids, (1..=11).collect::<Vec<i32>>());
let changes = release_changes().unwrap();
let ch_names: Vec<&str> =
changes.as_struct().fields().iter().map(|f| f.name.as_str()).collect();
assert_eq!(ch_names.len(), 16);
assert_eq!(ch_names[5], "change_sha");
assert_eq!(ch_names[15], "immutable");
let ch_ids: Vec<i32> = changes.as_struct().fields().iter().map(|f| f.id).collect();
assert_eq!(ch_ids, (1..=16).collect::<Vec<i32>>());
}
/// The column INDICES the Arrow codecs downcast by position must match the
/// schema field order. This is the assertion that would have caught a
/// divergent-copy column insert.
#[test]
fn codec_column_indices_match_schema_order() {
let events = release_events().unwrap();
let f = events.as_struct().fields();
assert_eq!(f[crate::release_events::COL_RUN_ID].name, "run_id");
assert_eq!(f[crate::release_events::COL_SEQ].name, "seq");
assert_eq!(f[crate::release_events::COL_DEPENDS_ON].name, "depends_on");
assert_eq!(f[crate::release_events::COL_ELAPSED_MS].name, "elapsed_ms");
let changes = release_changes().unwrap();
let g = changes.as_struct().fields();
assert_eq!(g[crate::release_changes::COL_RELEASE_ID].name, "release_id");
assert_eq!(g[crate::release_changes::COL_CHANGE_SHA].name, "change_sha");
assert_eq!(g[crate::release_changes::COL_IMMUTABLE].name, "immutable");
}
}