use chrono::{DateTime, SubsecRound, Utc};
use serde::{Deserialize, Serialize};
use crate::version_history::{ColumnChange, VersionOp};
pub const LEDGER_TABLE: &str = "_autumn_ledger_revisions";
pub const LEDGER_MAX_CLOCK_SKEW_SECS: i64 = 3_600;
pub const LEDGER_HIGH_WATER_TABLE: &str = "_autumn_ledger_high_water";
const HASH_DOMAIN: &str = "autumn.ledger.revision.v1";
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LedgerRevision {
pub id: i64,
pub table_name: String,
pub tenant_id: Option<String>,
pub record_id: i64,
pub seq: i64,
pub op: VersionOp,
pub actor: String,
pub request_id: Option<String>,
pub snapshot: serde_json::Value,
pub valid_from: DateTime<Utc>,
pub recorded_at: DateTime<Utc>,
pub prev_hash: Option<String>,
pub hash: String,
}
impl LedgerRevision {
#[must_use]
pub fn compute_hash(&self) -> String {
revision_hash(&RevisionHashInput {
prev_hash: self.prev_hash.as_deref(),
table_name: &self.table_name,
tenant_id: self.tenant_id.as_deref(),
record_id: self.record_id,
seq: self.seq,
op: self.op,
actor: &self.actor,
request_id: self.request_id.as_deref(),
snapshot: &self.snapshot,
valid_from: self.valid_from,
recorded_at: self.recorded_at,
})
}
}
#[derive(Debug, Clone, Copy)]
pub struct RevisionHashInput<'a> {
pub prev_hash: Option<&'a str>,
pub table_name: &'a str,
pub tenant_id: Option<&'a str>,
pub record_id: i64,
pub seq: i64,
pub op: VersionOp,
pub actor: &'a str,
pub request_id: Option<&'a str>,
pub snapshot: &'a serde_json::Value,
pub valid_from: DateTime<Utc>,
pub recorded_at: DateTime<Utc>,
}
#[must_use]
pub fn canonical_json(value: &serde_json::Value) -> String {
let mut out = String::new();
write_canonical(value, &mut out);
out
}
fn write_canonical(value: &serde_json::Value, out: &mut String) {
match value {
serde_json::Value::Object(map) => {
let mut keys: Vec<&String> = map.keys().collect();
keys.sort_unstable();
out.push('{');
for (idx, key) in keys.into_iter().enumerate() {
if idx > 0 {
out.push(',');
}
out.push_str(&serde_json::Value::String(key.clone()).to_string());
out.push(':');
if let Some(child) = map.get(key) {
write_canonical(child, out);
}
}
out.push('}');
}
serde_json::Value::Array(items) => {
out.push('[');
for (idx, item) in items.iter().enumerate() {
if idx > 0 {
out.push(',');
}
write_canonical(item, out);
}
out.push(']');
}
other => out.push_str(&other.to_string()),
}
}
#[must_use]
pub fn truncate_to_micros(at: DateTime<Utc>) -> DateTime<Utc> {
at.trunc_subsecs(6)
}
#[must_use]
pub fn revision_hash(input: &RevisionHashInput<'_>) -> String {
use sha2::{Digest, Sha256};
fn field(hasher: &mut Sha256, bytes: &[u8]) {
hasher.update(u64::try_from(bytes.len()).unwrap_or(u64::MAX).to_be_bytes());
hasher.update(bytes);
}
fn optional_field(hasher: &mut Sha256, value: Option<&str>) {
field(hasher, if value.is_some() { b"\x01" } else { b"\x00" });
field(hasher, value.unwrap_or("").as_bytes());
}
let mut hasher = Sha256::new();
let hasher = &mut hasher;
field(hasher, HASH_DOMAIN.as_bytes());
optional_field(hasher, input.prev_hash);
field(hasher, input.table_name.as_bytes());
optional_field(hasher, input.tenant_id);
field(hasher, input.record_id.to_string().as_bytes());
field(hasher, input.seq.to_string().as_bytes());
field(hasher, input.op.as_str().as_bytes());
field(hasher, input.actor.as_bytes());
optional_field(hasher, input.request_id);
field(hasher, format_instant(input.valid_from).as_bytes());
field(hasher, format_instant(input.recorded_at).as_bytes());
field(hasher, canonical_json(input.snapshot).as_bytes());
hex::encode(hasher.finalize_reset())
}
fn format_instant(at: DateTime<Utc>) -> String {
truncate_to_micros(at).to_rfc3339_opts(chrono::SecondsFormat::Micros, true)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum LedgerBreak {
HashMismatch,
PrevHashMismatch,
MissingRevision,
DuplicateSeq,
BrokenChainStart,
UnusableSeq,
LiveStateMismatch,
RecordedAtRegression,
HighWaterMissing,
HighWaterBehind,
HighWaterMismatch,
SnapshotNotCanonical,
}
impl LedgerBreak {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::HashMismatch => "hash_mismatch",
Self::PrevHashMismatch => "prev_hash_mismatch",
Self::MissingRevision => "missing_revision",
Self::DuplicateSeq => "duplicate_seq",
Self::BrokenChainStart => "broken_chain_start",
Self::UnusableSeq => "unusable_seq",
Self::LiveStateMismatch => "live_state_mismatch",
Self::RecordedAtRegression => "recorded_at_regression",
Self::SnapshotNotCanonical => "snapshot_not_canonical",
Self::HighWaterMissing => "high_water_missing",
Self::HighWaterBehind => "high_water_behind",
Self::HighWaterMismatch => "high_water_mismatch",
}
}
}
impl std::fmt::Display for LedgerBreak {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.as_str())
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LedgerBreakReport {
pub seq: i64,
pub revision_id: Option<i64>,
pub kind: LedgerBreak,
pub detail: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LedgerVerification {
pub record_id: i64,
pub revisions_checked: usize,
pub head_hash: Option<String>,
pub broken: Option<LedgerBreakReport>,
}
impl LedgerVerification {
#[must_use]
pub const fn is_intact(&self) -> bool {
self.broken.is_none()
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LedgerHead {
pub record_id: i64,
pub seq: i64,
pub hash: String,
pub recorded_at: DateTime<Utc>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LedgerHighWater {
pub record_id: i64,
pub seq: i64,
pub hash: String,
pub recorded_at: DateTime<Utc>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LedgerPin {
pub head: Option<LedgerHead>,
pub high_water: Option<LedgerHighWater>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LedgerHighWaterState<'a> {
NotChecked,
Absent,
Present(&'a LedgerHighWater),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LedgerLiveState {
NotChecked,
Absent,
Matches,
Diverged,
}
#[must_use]
pub fn verify_chain(record_id: i64, revisions: &[LedgerRevision]) -> LedgerVerification {
verify_chain_against(record_id, revisions, LedgerLiveState::NotChecked)
}
#[must_use]
pub fn verify_chain_against(
record_id: i64,
revisions: &[LedgerRevision],
live: LedgerLiveState,
) -> LedgerVerification {
verify_chain_with_high_water(record_id, revisions, live, LedgerHighWaterState::NotChecked)
}
#[must_use]
pub fn verify_chain_with_high_water(
record_id: i64,
revisions: &[LedgerRevision],
live: LedgerLiveState,
high_water: LedgerHighWaterState<'_>,
) -> LedgerVerification {
let mut verifier = LedgerChainVerifier::new(record_id);
for revision in revisions {
verifier.push(revision);
}
verifier.finish(live, high_water)
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct ChainTip {
id: i64,
seq: i64,
hash: String,
recorded_at: DateTime<Utc>,
}
impl ChainTip {
fn of(revision: &LedgerRevision) -> Self {
Self {
id: revision.id,
seq: revision.seq,
hash: revision.hash.clone(),
recorded_at: revision.recorded_at,
}
}
}
#[derive(Debug, Clone)]
pub struct LedgerChainVerifier {
record_id: i64,
checked: usize,
expected_seq: i64,
head: Option<ChainTip>,
chain_break: Option<LedgerBreakReport>,
time_break: Option<LedgerBreakReport>,
}
impl LedgerChainVerifier {
#[must_use]
pub const fn new(record_id: i64) -> Self {
Self {
record_id,
checked: 0,
expected_seq: 1,
head: None,
chain_break: None,
time_break: None,
}
}
pub fn push(&mut self, revision: &LedgerRevision) {
self.checked += 1;
if self.chain_break.is_none() {
self.chain_break = self.structural_break(revision);
}
if self.time_break.is_none() {
self.time_break = self
.head
.as_ref()
.and_then(|previous| recorded_at_break(previous, revision));
}
self.head = Some(ChainTip::of(revision));
}
#[must_use]
pub fn finish(
self,
live: LedgerLiveState,
high_water: LedgerHighWaterState<'_>,
) -> LedgerVerification {
let Self {
record_id,
checked,
head,
chain_break,
time_break,
..
} = self;
let broken = chain_break
.or_else(|| high_water_break(head.as_ref(), high_water))
.or_else(|| live_state_break(head.as_ref(), live))
.or(time_break);
let head_hash = if broken.is_none() {
head.map(|tip| tip.hash)
} else {
None
};
LedgerVerification {
record_id,
revisions_checked: checked,
head_hash,
broken,
}
}
fn structural_break(&mut self, revision: &LedgerRevision) -> Option<LedgerBreakReport> {
if revision.seq != self.expected_seq {
return Some(sequence_break(
revision,
self.expected_seq,
self.head.is_none(),
));
}
if let Some(report) = link_break(revision, self.head.as_ref()) {
return Some(report);
}
let Some(next_seq) = revision.seq.checked_add(1) else {
return Some(LedgerBreakReport {
seq: revision.seq,
revision_id: Some(revision.id),
kind: LedgerBreak::UnusableSeq,
detail: "sequence number is at i64::MAX and cannot be followed; \
the chain cannot be continued or checked past this revision"
.to_owned(),
});
};
self.expected_seq = next_seq;
None
}
}
fn high_water_break(
head: Option<&ChainTip>,
high_water: LedgerHighWaterState<'_>,
) -> Option<LedgerBreakReport> {
let high_water = match high_water {
LedgerHighWaterState::NotChecked => return None,
LedgerHighWaterState::Absent => None,
LedgerHighWaterState::Present(mark) => Some(mark),
};
match (head, high_water) {
(None, None) => None,
(None, Some(mark)) => Some(LedgerBreakReport {
seq: 1,
revision_id: None,
kind: LedgerBreak::MissingRevision,
detail: format!(
"the high-water mark reaches revision {} but the record has no \
revisions at all; the whole chain was erased",
mark.seq
),
}),
(Some(head), None) => Some(LedgerBreakReport {
seq: head.seq,
revision_id: Some(head.id),
kind: LedgerBreak::HighWaterMissing,
detail: format!(
"the chain reaches revision {} but the record has no high-water \
mark; the mark that would expose a deleted tail was itself removed",
head.seq
),
}),
(Some(head), Some(mark)) => head_versus_mark(head, mark),
}
}
fn head_versus_mark(head: &ChainTip, mark: &LedgerHighWater) -> Option<LedgerBreakReport> {
if mark.seq > head.seq {
let absent = head.seq.saturating_add(1);
return Some(LedgerBreakReport {
seq: absent,
revision_id: None,
kind: LedgerBreak::MissingRevision,
detail: format!(
"the high-water mark reaches revision {} but the chain stops at {}; \
revision {absent} onwards was deleted from the end",
mark.seq, head.seq
),
});
}
if mark.seq < head.seq {
return Some(LedgerBreakReport {
seq: head.seq,
revision_id: Some(head.id),
kind: LedgerBreak::HighWaterBehind,
detail: format!(
"the chain reaches revision {} but its high-water mark stops at {}; \
the mark was rolled back, or revisions were appended without it",
head.seq, mark.seq
),
});
}
if mark.hash != head.hash || mark.recorded_at != head.recorded_at {
return Some(LedgerBreakReport {
seq: head.seq,
revision_id: Some(head.id),
kind: LedgerBreak::HighWaterMismatch,
detail: format!(
"the high-water mark names revision {} but does not describe the \
revision stored there; the head revision was replaced, or the mark \
was rewritten to match a replacement",
head.seq
),
});
}
None
}
fn live_state_break(head: Option<&ChainTip>, live: LedgerLiveState) -> Option<LedgerBreakReport> {
let head = head?;
match live {
LedgerLiveState::NotChecked | LedgerLiveState::Matches => None,
LedgerLiveState::Absent => Some(LedgerBreakReport {
seq: head.seq,
revision_id: Some(head.id),
kind: LedgerBreak::LiveStateMismatch,
detail: format!(
"the ledger's newest revision is {}, but the record no longer \
exists in the table; the row was erased outside the repository",
head.seq
),
}),
LedgerLiveState::Diverged => Some(LedgerBreakReport {
seq: head.seq,
revision_id: Some(head.id),
kind: LedgerBreak::LiveStateMismatch,
detail: format!(
"revision {} is the newest in the ledger but does not describe \
the row the table holds; either the tail of the history was \
deleted or a write reached the table without appending a revision",
head.seq
),
}),
}
}
fn recorded_at_break(previous: &ChainTip, revision: &LedgerRevision) -> Option<LedgerBreakReport> {
(revision.recorded_at < previous.recorded_at).then(|| LedgerBreakReport {
seq: revision.seq,
revision_id: Some(revision.id),
kind: LedgerBreak::RecordedAtRegression,
detail: format!(
"revision {} was recorded before revision {}, which precedes it in the \
chain; transaction time cannot move backwards along a chain a post-#2323 \
writer produced",
revision.seq, previous.seq
),
})
}
fn sequence_break(
revision: &LedgerRevision,
expected_seq: i64,
at_chain_start: bool,
) -> LedgerBreakReport {
if revision.seq < expected_seq {
return LedgerBreakReport {
seq: revision.seq,
revision_id: Some(revision.id),
kind: LedgerBreak::DuplicateSeq,
detail: format!(
"revision {} repeats or precedes sequence {expected_seq}; \
a revision was inserted into the chain",
revision.seq
),
};
}
let detail = if at_chain_start {
format!(
"chain starts at sequence {} instead of 1; revision {expected_seq} was deleted",
revision.seq
)
} else {
format!(
"sequence {expected_seq} is absent between {} and {}; a revision was deleted",
expected_seq.saturating_sub(1),
revision.seq
)
};
LedgerBreakReport {
seq: expected_seq,
revision_id: None,
kind: LedgerBreak::MissingRevision,
detail,
}
}
fn link_break(revision: &LedgerRevision, prev: Option<&ChainTip>) -> Option<LedgerBreakReport> {
if revision.compute_hash() != revision.hash {
return Some(LedgerBreakReport {
seq: revision.seq,
revision_id: Some(revision.id),
kind: LedgerBreak::HashMismatch,
detail: format!(
"revision {} no longer hashes to its stored digest; \
the stored revision was mutated out of band",
revision.seq
),
});
}
match prev {
None => {
if revision.prev_hash.is_some() {
return Some(LedgerBreakReport {
seq: revision.seq,
revision_id: Some(revision.id),
kind: LedgerBreak::BrokenChainStart,
detail: "the first revision carries a prev_hash; \
the revision it chains to is missing"
.to_owned(),
});
}
}
Some(previous) => {
if revision.prev_hash.as_deref() != Some(previous.hash.as_str()) {
return Some(LedgerBreakReport {
seq: revision.seq,
revision_id: Some(revision.id),
kind: LedgerBreak::PrevHashMismatch,
detail: format!(
"revision {}'s prev_hash does not match revision {}'s hash; \
the chain was re-linked",
revision.seq, previous.seq
),
});
}
}
}
None
}
#[must_use]
pub fn monotonic_recorded_at(
db_now: DateTime<Utc>,
floor: Option<DateTime<Utc>>,
) -> Option<DateTime<Utc>> {
let now = truncate_to_micros(db_now);
let Some(floor) = floor.map(truncate_to_micros) else {
return Some(now);
};
if floor > now + chrono::TimeDelta::seconds(LEDGER_MAX_CLOCK_SKEW_SECS) {
return None;
}
Some(now.max(floor))
}
#[doc(hidden)]
#[must_use]
pub fn parse_sqlite_instant(raw: &str) -> Option<DateTime<Utc>> {
chrono::NaiveDateTime::parse_from_str(raw.trim(), "%Y-%m-%d %H:%M:%S%.f")
.ok()
.map(|naive| naive.and_utc())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct LedgerAsOf {
pub transaction: Option<DateTime<Utc>>,
pub valid: Option<DateTime<Utc>>,
}
impl LedgerAsOf {
#[must_use]
pub const fn transaction(at: DateTime<Utc>) -> Self {
Self {
transaction: Some(at),
valid: None,
}
}
#[must_use]
pub const fn valid(at: DateTime<Utc>) -> Self {
Self {
transaction: None,
valid: Some(at),
}
}
#[must_use]
pub const fn bitemporal(transaction: DateTime<Utc>, valid: DateTime<Utc>) -> Self {
Self {
transaction: Some(transaction),
valid: Some(valid),
}
}
}
#[must_use]
pub fn snapshot_as_of(revisions: &[LedgerRevision], as_of: LedgerAsOf) -> Option<&LedgerRevision> {
revisions
.iter()
.filter(|r| as_of.transaction.is_none_or(|t| r.recorded_at <= t))
.filter(|r| as_of.valid.is_none_or(|v| r.valid_from <= v))
.max_by_key(|r| r.seq)
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LedgerDiff {
pub record_id: i64,
pub from_seq: Option<i64>,
pub to_seq: Option<i64>,
pub changes: Vec<ColumnChange>,
}
impl LedgerDiff {
#[must_use]
pub const fn is_empty(&self) -> bool {
self.changes.is_empty()
}
}
#[must_use]
pub fn diff_snapshots(before: &serde_json::Value, after: &serde_json::Value) -> Vec<ColumnChange> {
let empty = serde_json::Map::new();
let before_obj = before.as_object().unwrap_or(&empty);
let after_obj = after.as_object().unwrap_or(&empty);
let mut columns: Vec<&String> = before_obj.keys().chain(after_obj.keys()).collect();
columns.sort_unstable();
columns.dedup();
columns
.into_iter()
.filter_map(|column| {
let before_val = before_obj.get(column);
let after_val = after_obj.get(column);
if before_val == after_val {
return None;
}
Some(ColumnChange::new(
column.clone(),
before_val.cloned(),
after_val.cloned(),
))
})
.collect()
}
#[must_use]
pub fn diff_as_of(
record_id: i64,
revisions: &[LedgerRevision],
from: LedgerAsOf,
to: LedgerAsOf,
) -> LedgerDiff {
let from_rev = snapshot_as_of(revisions, from);
let to_rev = snapshot_as_of(revisions, to);
let empty = serde_json::Value::Object(serde_json::Map::new());
let before = from_rev.map_or(&empty, |r| &r.snapshot);
let after = to_rev.map_or(&empty, |r| &r.snapshot);
LedgerDiff {
record_id,
from_seq: from_rev.map(|r| r.seq),
to_seq: to_rev.map(|r| r.seq),
changes: diff_snapshots(before, after),
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct LedgerCursor {
pub seq: i64,
pub id: i64,
}
impl LedgerCursor {
#[must_use]
pub const fn of(revision: &LedgerRevision) -> Self {
Self {
seq: revision.seq,
id: revision.id,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct LedgerPageRequest {
pub after: Option<LedgerCursor>,
pub limit: u32,
}
impl LedgerPageRequest {
pub const DEFAULT_LIMIT: u32 = 100;
pub const MAX_LIMIT: u32 = 1000;
pub const VERIFY_LIMIT: u32 = 256;
#[must_use]
pub const fn first(limit: u32) -> Self {
Self { after: None, limit }
}
#[must_use]
pub const fn after(cursor: LedgerCursor, limit: u32) -> Self {
Self {
after: Some(cursor),
limit,
}
}
#[must_use]
pub fn effective_limit(&self) -> u32 {
self.limit.clamp(1, Self::MAX_LIMIT)
}
}
impl Default for LedgerPageRequest {
fn default() -> Self {
Self::first(Self::DEFAULT_LIMIT)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[non_exhaustive]
pub struct LedgerRevisionPage {
pub revisions: Vec<LedgerRevision>,
pub next: Option<LedgerCursor>,
}
impl LedgerRevisionPage {
pub(crate) fn from_rows(mut rows: Vec<LedgerRevision>, limit: u32) -> Self {
let limit = usize::try_from(limit).unwrap_or(usize::MAX);
let next = if rows.len() > limit {
rows.truncate(limit);
rows.last().map(LedgerCursor::of)
} else {
None
};
Self {
revisions: rows,
next,
}
}
#[must_use]
pub fn next_request(&self, limit: u32) -> Option<LedgerPageRequest> {
self.next
.map(|cursor| LedgerPageRequest::after(cursor, limit))
}
}
#[cfg(feature = "db")]
const fn revisions_page_sql(first: bool) -> &'static str {
macro_rules! page_sql {
($tenant:literal, $id_cmp:literal) => {
concat!(
"SELECT id, table_name, tenant_id, record_id, seq, op, actor, \
request_id, snapshot, valid_from, recorded_at, prev_hash, hash \
FROM _autumn_ledger_revisions \
WHERE table_name = $1 AND record_id = $2 AND (",
$tenant,
" IS NULL OR tenant_id = $3) AND seq >= $4 AND (seq > $4 OR id ",
$id_cmp,
" $5) ORDER BY seq ASC, id ASC LIMIT $6"
)
};
}
#[cfg(not(feature = "sqlite"))]
let sql = if first {
page_sql!("$3::text", ">=")
} else {
page_sql!("$3::text", ">")
};
#[cfg(feature = "sqlite")]
let sql = if first {
page_sql!("$3", ">=")
} else {
page_sql!("$3", ">")
};
sql
}
#[cfg(feature = "db")]
#[doc(hidden)]
pub async fn read_revisions_page(
conn: &mut crate::db::RuntimeConnection,
table_name: &str,
tenant_id: Option<&str>,
record_id: i64,
page: LedgerPageRequest,
) -> crate::AutumnResult<LedgerRevisionPage> {
use diesel::sql_types::{BigInt, Nullable, Text};
use diesel_async::RunQueryDsl;
#[derive(diesel::QueryableByName)]
struct Row {
#[diesel(sql_type = BigInt)]
id: i64,
#[diesel(sql_type = Text)]
table_name: String,
#[diesel(sql_type = Nullable<Text>)]
tenant_id: Option<String>,
#[diesel(sql_type = BigInt)]
record_id: i64,
#[diesel(sql_type = BigInt)]
seq: i64,
#[diesel(sql_type = Text)]
op: String,
#[diesel(sql_type = Text)]
actor: String,
#[diesel(sql_type = Nullable<Text>)]
request_id: Option<String>,
#[diesel(sql_type = Text)]
snapshot: String,
#[diesel(sql_type = Instant)]
valid_from: DateTime<Utc>,
#[diesel(sql_type = Instant)]
recorded_at: DateTime<Utc>,
#[diesel(sql_type = Nullable<Text>)]
prev_hash: Option<String>,
#[diesel(sql_type = Text)]
hash: String,
}
let limit = page.effective_limit();
let (after_seq, after_id) = page.after.map_or((i64::MIN, i64::MIN), |c| (c.seq, c.id));
let fetch = i64::from(limit) + 1;
let rows: Vec<Row> = diesel::sql_query(revisions_page_sql(page.after.is_none()))
.bind::<Text, _>(table_name)
.bind::<BigInt, _>(record_id)
.bind::<Nullable<Text>, _>(tenant_id)
.bind::<BigInt, _>(after_seq)
.bind::<BigInt, _>(after_id)
.bind::<BigInt, _>(fetch)
.get_results(conn)
.await
.map_err(crate::AutumnError::from)?;
let mut revisions = Vec::with_capacity(rows.len());
for row in rows {
let op = match row.op.as_str() {
"insert" => VersionOp::Insert,
"delete" => VersionOp::Delete,
_ => VersionOp::Update,
};
let snapshot = parse_stored_snapshot(table_name, record_id, row.seq, &row.snapshot)?;
revisions.push(LedgerRevision {
id: row.id,
table_name: row.table_name,
tenant_id: row.tenant_id,
record_id: row.record_id,
seq: row.seq,
op,
actor: row.actor,
request_id: row.request_id,
snapshot,
valid_from: row.valid_from,
recorded_at: row.recorded_at,
prev_hash: row.prev_hash,
hash: row.hash,
});
}
Ok(LedgerRevisionPage::from_rows(revisions, limit))
}
#[diagnostic::on_unimplemented(
message = "`{Self}` is a ledgered entity, so it cannot be hard-deleted",
label = "hard delete would erase the row this entity's ledger reconstructs",
note = "ledgered repositories soft-delete: call `delete_by_id` (which records a revision) \
and `restore` instead of `purge`"
)]
pub trait LedgerHardDeleteAllowed {}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
#[non_exhaustive]
pub enum LedgerError {
#[error(
"ledger chain for {table}#{record_id} is unreadable: {detail}; \
a ledgered write cannot proceed without appending a revision"
)]
ChainUnreadable {
table: String,
record_id: i64,
detail: String,
},
#[error(
"cannot cascade a hard delete into ledgered {table}#{record_id}: erasing it \
would destroy the record its ledger reconstructs, and keeping it would leave \
a foreign key pointing at a deleted parent. Make the parent repository \
`soft_delete` (a soft parent delete soft-deletes this child and records a \
revision), or remove the `dependent(...)` clause"
)]
HardDeleteCascade {
table: String,
record_id: i64,
},
#[error(
"{path} cannot write to ledgered table {table}. The write records no \
revision, so the ledger would disagree with the table. Write through the \
ledgered repository of {table} (for a dependent, use \
`on_delete = destroy`), or stop ledgering {table}"
)]
OutOfBandWrite {
table: String,
path: &'static str,
},
#[error(
"cannot ledger {table}#{record_id}: the write changes its tenant. A revision \
chain belongs to one tenant, so the move would split it. Create a new record \
in the other tenant instead"
)]
TenantChange {
table: String,
record_id: i64,
},
#[error(
"cannot ledger {table}#{record_id}: column `{column}` holds NaN or an infinite \
float, which a snapshot cannot store. Write a finite value, or stop ledgering \
{table}"
)]
NonFiniteValue {
table: String,
record_id: i64,
column: String,
},
#[error(
"ledger revision {seq} of {table}#{record_id} is not in canonical form: the stored \
snapshot text was re-encoded, reordered or holds a duplicate key"
)]
SnapshotNotCanonical {
table: String,
record_id: i64,
seq: i64,
},
#[error(
"ledger revision {seq} of {table}#{record_id} does not match the current model \
schema: {detail}. The model changed after this revision was written. Add \
`#[serde(default)]` to a new field. To convert a renamed or retyped field, \
set `ledgered(schema_version = N, upcast = path)`. To check for an edit, \
run `ledger_verify`"
)]
SnapshotSchemaMismatch {
table: String,
record_id: i64,
seq: i64,
detail: String,
},
#[error("ledger chain for {table}#{record_id} is broken at revision {seq}: {detail}")]
ChainBroken {
table: String,
record_id: i64,
seq: i64,
detail: String,
},
}
pub trait LedgeredRecord: crate::version_history::VersionedRecord {
fn ledger_valid_from(&self) -> Option<DateTime<Utc>> {
None
}
fn ledger_tenant_id(&self) -> Option<&str> {
None
}
const LEDGER_SCHEMA_VERSION: u32 = 1;
fn ledger_upcast(_from_version: u32, _snapshot: &mut serde_json::Value) -> Result<(), String> {
Ok(())
}
}
#[diagnostic::on_unimplemented(
message = "a ledgered `tenant_scoped` repository needs `tenant_id: String`, not `{Self}`",
label = "the tenant column of a ledgered model must not be nullable",
note = "a revision with a NULL tenant is not visible to a tenant-scoped read, and a \
cross-tenant ledger read is refused, so no read can reach its chain. Make the \
field `String` and the column `NOT NULL`, or remove `ledgered`"
)]
pub trait LedgerTenantColumn {
fn ledger_tenant_id(&self) -> &str;
}
impl LedgerTenantColumn for String {
fn ledger_tenant_id(&self) -> &str {
self.as_str()
}
}
#[doc(hidden)]
pub fn refuse_tenant_change(
table: &str,
record_id: i64,
before: Option<&str>,
after: Option<&str>,
) -> crate::AutumnResult<()> {
if before == after {
return Ok(());
}
Err(crate::AutumnError::conflict(LedgerError::TenantChange {
table: table.to_string(),
record_id,
}))
}
#[doc(hidden)]
pub trait LedgerValidTimeValue {
fn ledger_valid_from(&self) -> Option<DateTime<Utc>>;
}
impl LedgerValidTimeValue for DateTime<Utc> {
fn ledger_valid_from(&self) -> Option<DateTime<Utc>> {
Some(*self)
}
}
impl LedgerValidTimeValue for Option<DateTime<Utc>> {
fn ledger_valid_from(&self) -> Option<DateTime<Utc>> {
*self
}
}
impl LedgerValidTimeValue for chrono::NaiveDateTime {
fn ledger_valid_from(&self) -> Option<DateTime<Utc>> {
Some(self.and_utc())
}
}
impl LedgerValidTimeValue for Option<chrono::NaiveDateTime> {
fn ledger_valid_from(&self) -> Option<DateTime<Utc>> {
self.as_ref().map(chrono::NaiveDateTime::and_utc)
}
}
#[doc(hidden)]
pub struct NonFiniteProbe<'a, T: ?Sized>(pub &'a T);
#[doc(hidden)]
pub trait FloatContent {
fn has_non_finite(&self) -> bool;
}
impl FloatContent for f64 {
fn has_non_finite(&self) -> bool {
!self.is_finite()
}
}
impl FloatContent for f32 {
fn has_non_finite(&self) -> bool {
!self.is_finite()
}
}
impl<T: FloatContent> FloatContent for Option<T> {
fn has_non_finite(&self) -> bool {
self.as_ref().is_some_and(FloatContent::has_non_finite)
}
}
impl<T: FloatContent> FloatContent for Vec<T> {
fn has_non_finite(&self) -> bool {
self.iter().any(FloatContent::has_non_finite)
}
}
#[doc(hidden)]
pub trait NonFiniteViaFloat {
fn autumn_has_non_finite(&self) -> bool;
}
impl<T: FloatContent> NonFiniteViaFloat for NonFiniteProbe<'_, T> {
fn autumn_has_non_finite(&self) -> bool {
self.0.has_non_finite()
}
}
#[doc(hidden)]
pub trait NonFiniteFallback {
fn autumn_has_non_finite(&self) -> bool;
}
impl<T: ?Sized> NonFiniteFallback for &NonFiniteProbe<'_, T> {
fn autumn_has_non_finite(&self) -> bool {
false
}
}
#[doc(hidden)]
pub fn refuse_non_finite(
table: &str,
record_id: i64,
column: Option<&'static str>,
) -> crate::AutumnResult<()> {
column.map_or(Ok(()), |column| {
Err(crate::AutumnError::internal_server_error(
LedgerError::NonFiniteValue {
table: table.to_string(),
record_id,
column: column.to_string(),
},
))
})
}
#[doc(hidden)]
pub fn parse_stored_snapshot(
table: &str,
record_id: i64,
seq: i64,
text: &str,
) -> crate::AutumnResult<serde_json::Value> {
let value: serde_json::Value = serde_json::from_str(text).map_err(|err| {
crate::AutumnError::internal_server_error(LedgerError::ChainUnreadable {
table: table.to_string(),
record_id,
detail: format!("revision {seq} has an unreadable snapshot: {err}"),
})
})?;
if canonical_json(&value) != text {
return Err(crate::AutumnError::internal_server_error(
LedgerError::SnapshotNotCanonical {
table: table.to_string(),
record_id,
seq,
},
));
}
Ok(value)
}
#[doc(hidden)]
#[must_use]
pub fn redact_quoted(detail: &str) -> String {
let mut out = String::with_capacity(detail.len());
let mut in_quote = false;
let mut escaped = false;
for ch in detail.chars() {
if in_quote {
if escaped {
escaped = false;
} else if ch == '\\' {
escaped = true;
} else if ch == '"' {
in_quote = false;
out.push_str("…\"");
}
} else {
out.push(ch);
in_quote = ch == '"';
}
}
out
}
#[doc(hidden)]
#[must_use]
pub fn schema_mismatch(table: &str, record_id: i64, seq: i64, detail: &str) -> crate::AutumnError {
crate::AutumnError::internal_server_error(LedgerError::SnapshotSchemaMismatch {
table: table.to_string(),
record_id,
seq,
detail: redact_quoted(detail),
})
}
#[doc(hidden)]
pub fn verification_from_read_error(
record_id: i64,
err: crate::AutumnError,
) -> crate::AutumnResult<LedgerVerification> {
let Some(LedgerError::SnapshotNotCanonical { seq, .. }) = err.downcast_ref::<LedgerError>()
else {
return Err(err);
};
Ok(LedgerVerification {
record_id,
revisions_checked: 0,
head_hash: None,
broken: Some(LedgerBreakReport {
seq: *seq,
revision_id: None,
kind: LedgerBreak::SnapshotNotCanonical,
detail: format!("revision {seq} is stored in a form other than its canonical text"),
}),
})
}
pub const SNAPSHOT_SCHEMA_KEY: &str = "$schema_version";
#[doc(hidden)]
pub type SnapshotUpcast = fn(u32, &mut serde_json::Value) -> Result<(), String>;
#[doc(hidden)]
pub fn stamp_schema_version(snapshot: &mut serde_json::Value, current: u32) {
if current > 1
&& let Some(object) = snapshot.as_object_mut()
{
object.insert(SNAPSHOT_SCHEMA_KEY.to_string(), serde_json::json!(current));
}
}
#[doc(hidden)]
pub fn upcast_snapshot(
table: &str,
record_id: i64,
seq: i64,
snapshot: &serde_json::Value,
current: u32,
hook: SnapshotUpcast,
) -> crate::AutumnResult<(serde_json::Value, u32)> {
let mut value = snapshot.clone();
let Some(object) = value.as_object_mut() else {
return Ok((value, 1));
};
let stored = match object.remove(SNAPSHOT_SCHEMA_KEY) {
None => 1,
Some(marker) => marker
.as_u64()
.and_then(|version| u32::try_from(version).ok())
.filter(|version| *version >= 1)
.ok_or_else(|| {
schema_mismatch(
table,
record_id,
seq,
"the snapshot has a damaged schema version marker",
)
})?,
};
if stored < current {
let ran =
std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| hook(stored, &mut value)))
.unwrap_or_else(|_| Err("the hook panicked".to_string()));
ran.map_err(|message| {
schema_mismatch(
table,
record_id,
seq,
&format!("the upcast hook failed for schema version {stored}: {message}"),
)
})?;
}
Ok((value, stored))
}
#[doc(hidden)]
#[must_use]
pub fn schema_mismatch_at(
table: &str,
record_id: i64,
seq: i64,
stored: u32,
current: u32,
detail: &str,
) -> crate::AutumnError {
let origin = if stored > current {
" This revision was written by a newer model."
} else {
""
};
schema_mismatch(
table,
record_id,
seq,
&format!(
"{detail} (the revision is at schema version {stored}, the model is at \
{current}.{origin})"
),
)
}
#[doc(hidden)]
#[must_use]
pub fn snapshots_agree(
head: &serde_json::Value,
live: &serde_json::Value,
tolerate_new_columns: bool,
) -> bool {
match (head.as_object(), live.as_object()) {
(Some(head), Some(live)) if tolerate_new_columns => head.iter().all(|(column, value)| {
live.get(column)
.is_none_or(|other| canonical_json(value) == canonical_json(other))
}),
_ => canonical_json(head) == canonical_json(live),
}
}
#[doc(hidden)]
pub fn note_live_check_skipped(table: &str, record_id: i64, error: &crate::AutumnError) {
tracing::warn!(
table,
record_id,
%error,
"ledger_verify skipped the live-row check: the head revision could not be converted"
);
}
#[doc(hidden)]
pub fn ensure_snapshot_recoverable(
table: &str,
record_id: i64,
seq: i64,
snapshot: &serde_json::Value,
) -> crate::AutumnResult<()> {
if !crate::encryption::registered_encrypted_columns()
.iter()
.any(|d| d.table == table)
{
return Ok(());
}
let mut probe = snapshot.clone();
let lost = crate::encryption::decrypt_snapshot_columns(table, &mut probe);
if lost.is_empty() {
return Ok(());
}
Err(crate::AutumnError::internal_server_error(
LedgerError::ChainUnreadable {
table: table.to_string(),
record_id,
detail: format!(
"revision {seq}: encrypted column(s) `{}` cannot be decrypted. The key \
is missing, the key ring is not set, or the envelope is damaged",
lost.join("`, `")
),
},
))
}
#[cfg(feature = "db")]
struct ChainState {
db_now: DateTime<Utc>,
head_seq: Option<i64>,
head_hash: Option<String>,
head_recorded_at: Option<DateTime<Utc>>,
mark_seq: Option<i64>,
mark_hash: Option<String>,
mark_recorded_at: Option<DateTime<Utc>>,
}
#[cfg(feature = "db")]
#[doc(hidden)]
#[derive(Debug)]
pub struct LedgerAppend<'a> {
pub table_name: &'static str,
pub tenant_id: Option<&'a str>,
pub record_id: i64,
pub op: VersionOp,
pub actor: &'a str,
pub request_id: Option<&'a str>,
pub snapshot: serde_json::Value,
pub valid_from: Option<DateTime<Utc>>,
}
#[cfg(feature = "db")]
#[allow(
clippy::too_many_lines,
reason = "one backend-forked statement: each arm carries its own \
`QueryableByName` row because the clock and the two instants come \
back as different SQL types per tier. Splitting it would separate \
the SQL from the row shape it decodes into"
)]
async fn read_chain_state(
conn: &mut crate::db::RuntimeConnection,
table_name: &str,
record_id: i64,
tenant_id: Option<&str>,
) -> crate::AutumnResult<ChainState> {
use diesel::sql_types::{BigInt, Nullable, Text};
use diesel_async::RunQueryDsl;
let unreadable = |detail: String| {
crate::AutumnError::internal_server_error(LedgerError::ChainUnreadable {
table: table_name.to_string(),
record_id,
detail,
})
};
#[cfg(not(feature = "sqlite"))]
{
use diesel::sql_types::Timestamptz;
#[derive(diesel::QueryableByName)]
struct Row {
#[diesel(sql_type = Timestamptz)]
db_now: DateTime<Utc>,
#[diesel(sql_type = Nullable<BigInt>)]
head_seq: Option<i64>,
#[diesel(sql_type = Nullable<Text>)]
head_hash: Option<String>,
#[diesel(sql_type = Nullable<Timestamptz>)]
head_recorded_at: Option<DateTime<Utc>>,
#[diesel(sql_type = Nullable<BigInt>)]
mark_seq: Option<i64>,
#[diesel(sql_type = Nullable<Text>)]
mark_hash: Option<String>,
#[diesel(sql_type = Nullable<Timestamptz>)]
mark_recorded_at: Option<DateTime<Utc>>,
}
let row: Row = diesel::sql_query(
"SELECT clock_timestamp() AS db_now, \
__head.seq AS head_seq, \
__head.hash AS head_hash, \
__head.recorded_at AS head_recorded_at, \
__mark.high_seq AS mark_seq, \
__mark.head_hash AS mark_hash, \
__mark.recorded_at AS mark_recorded_at \
FROM (SELECT 1 AS anchor) AS __anchor \
LEFT JOIN ( \
SELECT seq, hash, recorded_at FROM _autumn_ledger_revisions \
WHERE table_name = $1 AND record_id = $2 \
AND COALESCE(tenant_id, '') = COALESCE($3::text, '') \
ORDER BY seq DESC LIMIT 1 \
) AS __head ON 1 = 1 \
LEFT JOIN _autumn_ledger_high_water AS __mark \
ON __mark.table_name = $1 AND __mark.record_id = $2 \
AND __mark.tenant_key = COALESCE($3::text, '')",
)
.bind::<Text, _>(table_name)
.bind::<BigInt, _>(record_id)
.bind::<Nullable<Text>, _>(tenant_id)
.get_results::<Row>(conn)
.await
.map_err(crate::AutumnError::from)?
.into_iter()
.next()
.ok_or_else(|| unreadable("the chain-state read returned no row".to_string()))?;
Ok(ChainState {
db_now: row.db_now,
head_seq: row.head_seq,
head_hash: row.head_hash,
head_recorded_at: row.head_recorded_at,
mark_seq: row.mark_seq,
mark_hash: row.mark_hash,
mark_recorded_at: row.mark_recorded_at,
})
}
#[cfg(feature = "sqlite")]
{
use diesel::sql_types::TimestamptzSqlite;
#[derive(diesel::QueryableByName)]
struct Row {
#[diesel(sql_type = Text)]
db_now: String,
#[diesel(sql_type = Nullable<BigInt>)]
head_seq: Option<i64>,
#[diesel(sql_type = Nullable<Text>)]
head_hash: Option<String>,
#[diesel(sql_type = Nullable<TimestamptzSqlite>)]
head_recorded_at: Option<DateTime<Utc>>,
#[diesel(sql_type = Nullable<BigInt>)]
mark_seq: Option<i64>,
#[diesel(sql_type = Nullable<Text>)]
mark_hash: Option<String>,
#[diesel(sql_type = Nullable<TimestamptzSqlite>)]
mark_recorded_at: Option<DateTime<Utc>>,
}
let row: Row = diesel::sql_query(
"SELECT strftime('%Y-%m-%d %H:%M:%f', 'now') AS db_now, \
__head.seq AS head_seq, \
__head.hash AS head_hash, \
__head.recorded_at AS head_recorded_at, \
__mark.high_seq AS mark_seq, \
__mark.head_hash AS mark_hash, \
__mark.recorded_at AS mark_recorded_at \
FROM (SELECT 1 AS anchor) AS __anchor \
LEFT JOIN ( \
SELECT seq, hash, recorded_at FROM _autumn_ledger_revisions \
WHERE table_name = $1 AND record_id = $2 \
AND COALESCE(tenant_id, '') = COALESCE($3, '') \
ORDER BY seq DESC LIMIT 1 \
) AS __head ON 1 = 1 \
LEFT JOIN _autumn_ledger_high_water AS __mark \
ON __mark.table_name = $1 AND __mark.record_id = $2 \
AND __mark.tenant_key = COALESCE($3, '')",
)
.bind::<Text, _>(table_name)
.bind::<BigInt, _>(record_id)
.bind::<Nullable<Text>, _>(tenant_id)
.get_results::<Row>(conn)
.await
.map_err(crate::AutumnError::from)?
.into_iter()
.next()
.ok_or_else(|| unreadable("the chain-state read returned no row".to_string()))?;
let db_now = parse_sqlite_instant(&row.db_now).ok_or_else(|| {
unreadable(format!(
"the database clock read back as {:?}, which is not the encoding the \
ledger expects",
row.db_now,
))
})?;
Ok(ChainState {
db_now,
head_seq: row.head_seq,
head_hash: row.head_hash,
head_recorded_at: row.head_recorded_at,
mark_seq: row.mark_seq,
mark_hash: row.mark_hash,
mark_recorded_at: row.mark_recorded_at,
})
}
}
#[cfg(feature = "db")]
pub(crate) const REVISION_INSERT_SQL: &str = "INSERT INTO _autumn_ledger_revisions \
(table_name, tenant_id, record_id, seq, op, actor, request_id, \
snapshot, valid_from, recorded_at, prev_hash, hash) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)";
#[cfg(all(feature = "db", not(feature = "sqlite")))]
pub(crate) const HIGH_WATER_UPSERT_SQL: &str = "INSERT INTO _autumn_ledger_high_water \
(table_name, tenant_key, record_id, high_seq, head_hash, recorded_at) \
VALUES ($1, COALESCE($2::text, ''), $3, $4, $5, $6) \
ON CONFLICT (table_name, tenant_key, record_id) DO UPDATE SET \
high_seq = EXCLUDED.high_seq, \
head_hash = EXCLUDED.head_hash, \
recorded_at = EXCLUDED.recorded_at \
WHERE _autumn_ledger_high_water.high_seq < EXCLUDED.high_seq";
#[cfg(all(feature = "db", feature = "sqlite"))]
pub(crate) const HIGH_WATER_UPSERT_SQL: &str = "INSERT INTO _autumn_ledger_high_water \
(table_name, tenant_key, record_id, high_seq, head_hash, recorded_at) \
VALUES ($1, COALESCE($2, ''), $3, $4, $5, $6) \
ON CONFLICT (table_name, tenant_key, record_id) DO UPDATE SET \
high_seq = excluded.high_seq, \
head_hash = excluded.head_hash, \
recorded_at = excluded.recorded_at \
WHERE _autumn_ledger_high_water.high_seq < excluded.high_seq";
#[cfg(all(feature = "db", not(feature = "sqlite")))]
type Instant = diesel::sql_types::Timestamptz;
#[cfg(all(feature = "db", feature = "sqlite"))]
type Instant = diesel::sql_types::TimestamptzSqlite;
#[cfg(feature = "db")]
#[doc(hidden)]
#[allow(
clippy::too_many_lines,
reason = "the #2323 high-water cross-checks, the sequence allocation and the \
transaction-time clamp are one indivisible decision: each rule's \
rationale is stated where it is enforced, and splitting them would \
scatter the argument for why an append is allowed to proceed"
)]
pub async fn append_revision(
conn: &mut crate::db::RuntimeConnection,
append: LedgerAppend<'_>,
) -> crate::AutumnResult<()> {
use diesel::sql_types::{BigInt, Nullable, Text};
use diesel_async::RunQueryDsl;
let LedgerAppend {
table_name,
tenant_id,
record_id,
op,
actor,
request_id,
snapshot,
valid_from,
} = append;
let unreadable = |detail: String| {
crate::AutumnError::internal_server_error(LedgerError::ChainUnreadable {
table: table_name.to_string(),
record_id,
detail,
})
};
let state = read_chain_state(conn, table_name, record_id, tenant_id).await?;
if state.head_seq.is_some() && state.mark_seq.is_none() {
return Err(unreadable(
"the record has revisions but no high-water mark; the mark that makes a \
deleted revision permanent evidence is itself missing. Appending here \
would re-create it over whatever is left. Run `ledger_verify` and \
re-apply the high-water migration's backfill once you know what happened"
.to_string(),
));
}
if state.mark_seq.is_some_and(|seq| seq < 1) {
return Err(unreadable(
"the record's high-water mark carries a sequence number below 1, which \
names no revision. Appending would allocate as though the mark were \
absent and then overwrite it — run `ledger_verify`, and check whether \
the mark table's CHECK constraint is still in place"
.to_string(),
));
}
if state.mark_seq == state.head_seq
&& (state.mark_hash.as_deref() != state.head_hash.as_deref()
|| state.mark_recorded_at != state.head_recorded_at)
{
return Err(unreadable(
"the high-water mark and the chain head carry the same sequence number \
but describe different revisions; one of the two was rewritten out of \
band. Appending here would settle the disagreement in the writer's \
favour and erase it — run `ledger_verify` instead"
.to_string(),
));
}
let seq_base: i64 = state.head_seq.unwrap_or(0).max(state.mark_seq.unwrap_or(0));
if seq_base == i64::MAX {
return Err(unreadable(
"the record's chain is at i64::MAX and cannot be continued; no further \
revision can be numbered"
.to_string(),
));
}
let seq: i64 = seq_base.saturating_add(1);
let prev_hash: Option<String> = state.head_hash;
let floor = if state.mark_seq > state.head_seq {
state.head_recorded_at.max(state.mark_recorded_at)
} else {
state.head_recorded_at
};
let recorded_at = monotonic_recorded_at(state.db_now, floor).ok_or_else(|| {
unreadable(format!(
"the record's transaction-time floor is more than {LEDGER_MAX_CLOCK_SKEW_SECS}s \
ahead of the database's own clock; writing behind it would hash that instant \
into the chain and make every as-of query miss the revision. Refusing rather \
than ratcheting",
))
})?;
let valid_from = valid_from.map_or(recorded_at, truncate_to_micros);
let hash: String = revision_hash(&RevisionHashInput {
prev_hash: prev_hash.as_deref(),
table_name,
tenant_id,
record_id,
seq,
op,
actor,
request_id,
snapshot: &snapshot,
valid_from,
recorded_at,
});
let snapshot_str: String = canonical_json(&snapshot);
diesel::sql_query(REVISION_INSERT_SQL)
.bind::<Text, _>(table_name)
.bind::<Nullable<Text>, _>(tenant_id)
.bind::<BigInt, _>(record_id)
.bind::<BigInt, _>(seq)
.bind::<Text, _>(op.as_str())
.bind::<Text, _>(actor)
.bind::<Nullable<Text>, _>(request_id)
.bind::<Text, _>(snapshot_str)
.bind::<Instant, _>(valid_from)
.bind::<Instant, _>(recorded_at)
.bind::<Nullable<Text>, _>(prev_hash.as_deref())
.bind::<Text, _>(hash.as_str())
.execute(conn)
.await
.map_err(crate::AutumnError::from)?;
diesel::sql_query(HIGH_WATER_UPSERT_SQL)
.bind::<Text, _>(table_name)
.bind::<Nullable<Text>, _>(tenant_id)
.bind::<BigInt, _>(record_id)
.bind::<BigInt, _>(seq)
.bind::<Text, _>(hash.as_str())
.bind::<Instant, _>(recorded_at)
.execute(conn)
.await
.map_err(crate::AutumnError::from)?;
Ok(())
}
#[doc(hidden)]
pub struct LedgeredTableDescriptor {
pub table: &'static str,
}
inventory::collect!(LedgeredTableDescriptor);
#[doc(hidden)]
#[must_use]
pub fn is_ledgered_table(table: &str) -> bool {
static TABLES: std::sync::OnceLock<std::collections::HashSet<&'static str>> =
std::sync::OnceLock::new();
TABLES
.get_or_init(|| {
inventory::iter::<LedgeredTableDescriptor>
.into_iter()
.map(|descriptor| descriptor.table)
.collect()
})
.contains(table)
}
#[doc(hidden)]
pub fn refuse_out_of_band_write(table: &str, path: &'static str) -> crate::AutumnResult<()> {
if is_ledgered_table(table) {
return Err(crate::AutumnError::conflict(LedgerError::OutOfBandWrite {
table: table.to_string(),
path,
}));
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[cfg(feature = "db")]
#[test]
fn ledger_append_statements_keep_their_contract() {
assert!(
!REVISION_INSERT_SQL.contains("::jsonb"),
"the snapshot must not be cast to jsonb: {REVISION_INSERT_SQL}"
);
assert!(
REVISION_INSERT_SQL.contains(
"(table_name, tenant_id, record_id, seq, op, actor, request_id, \
snapshot, valid_from, recorded_at, prev_hash, hash)"
),
"the revision insert must name all twelve columns: {REVISION_INSERT_SQL}"
);
assert!(
HIGH_WATER_UPSERT_SQL.contains("ON CONFLICT (table_name, tenant_key, record_id)"),
"the mark is keyed per (table, tenant, record): {HIGH_WATER_UPSERT_SQL}"
);
assert!(
HIGH_WATER_UPSERT_SQL
.to_ascii_lowercase()
.contains("where _autumn_ledger_high_water.high_seq < excluded.high_seq"),
"the mark must refuse to move backwards: {HIGH_WATER_UPSERT_SQL}"
);
}
use chrono::TimeZone as _;
use serde_json::json;
fn at(secs: i64) -> DateTime<Utc> {
Utc.timestamp_opt(1_800_000_000 + secs, 0)
.single()
.expect("valid instant")
}
fn chain(n: i64) -> Vec<LedgerRevision> {
let mut out: Vec<LedgerRevision> = Vec::new();
let mut prev_hash: Option<String> = None;
for seq in 1..=n {
let snapshot = json!({ "id": 7, "title": format!("v{seq}"), "deleted_at": null });
let op = if seq == 1 {
VersionOp::Insert
} else {
VersionOp::Update
};
let recorded_at = at(seq * 10);
let hash = revision_hash(&RevisionHashInput {
prev_hash: prev_hash.as_deref(),
table_name: "widgets",
tenant_id: None,
record_id: 7,
seq,
op,
actor: "alice",
request_id: None,
snapshot: &snapshot,
valid_from: recorded_at,
recorded_at,
});
out.push(LedgerRevision {
id: seq,
table_name: "widgets".to_owned(),
tenant_id: None,
record_id: 7,
seq,
op,
actor: "alice".to_owned(),
request_id: None,
snapshot,
valid_from: recorded_at,
recorded_at,
prev_hash: prev_hash.clone(),
hash: hash.clone(),
});
prev_hash = Some(hash);
}
out
}
#[test]
fn canonical_json_sorts_object_keys() {
let value = json!({ "b": 1, "a": 2, "c": 3 });
assert_eq!(canonical_json(&value), r#"{"a":2,"b":1,"c":3}"#);
}
#[test]
fn canonical_json_is_insertion_order_independent() {
let mut first = serde_json::Map::new();
first.insert("z".to_owned(), json!(1));
first.insert("a".to_owned(), json!(2));
let mut second = serde_json::Map::new();
second.insert("a".to_owned(), json!(2));
second.insert("z".to_owned(), json!(1));
assert_eq!(
canonical_json(&serde_json::Value::Object(first)),
canonical_json(&serde_json::Value::Object(second)),
);
}
#[test]
fn canonical_json_sorts_nested_objects_and_preserves_array_order() {
let value = json!({ "outer": { "b": [3, 1, 2], "a": { "y": 1, "x": 2 } } });
assert_eq!(
canonical_json(&value),
r#"{"outer":{"a":{"x":2,"y":1},"b":[3,1,2]}}"#
);
}
#[test]
fn canonical_json_escapes_keys_and_values() {
let value = json!({ "a\"b": "c\nd" });
assert_eq!(canonical_json(&value), r#"{"a\"b":"c\nd"}"#);
}
#[test]
fn canonical_json_handles_scalars_and_nulls() {
assert_eq!(canonical_json(&json!(null)), "null");
assert_eq!(canonical_json(&json!(true)), "true");
assert_eq!(canonical_json(&json!(12)), "12");
assert_eq!(canonical_json(&json!("s")), "\"s\"");
assert_eq!(canonical_json(&json!([])), "[]");
assert_eq!(canonical_json(&json!({})), "{}");
}
fn base_input(snapshot: &serde_json::Value) -> RevisionHashInput<'_> {
RevisionHashInput {
prev_hash: None,
table_name: "widgets",
tenant_id: None,
record_id: 7,
seq: 1,
op: VersionOp::Insert,
actor: "alice",
request_id: None,
snapshot,
valid_from: at(0),
recorded_at: at(0),
}
}
#[test]
fn revision_hash_is_deterministic() {
let snapshot = json!({ "id": 7, "title": "hello" });
let input = base_input(&snapshot);
assert_eq!(revision_hash(&input), revision_hash(&input));
}
#[test]
fn revision_hash_is_hex_sha256() {
let snapshot = json!({});
let hash = revision_hash(&base_input(&snapshot));
assert_eq!(hash.len(), 64);
assert!(hash.chars().all(|c| c.is_ascii_hexdigit()));
}
#[test]
fn revision_hash_ignores_snapshot_key_order() {
let a = json!({ "id": 7, "title": "hello" });
let mut map = serde_json::Map::new();
map.insert("title".to_owned(), json!("hello"));
map.insert("id".to_owned(), json!(7));
let b = serde_json::Value::Object(map);
assert_eq!(
revision_hash(&base_input(&a)),
revision_hash(&base_input(&b))
);
}
#[test]
fn revision_hash_changes_with_every_covered_field() {
let snapshot = json!({ "id": 7, "title": "hello" });
let other_snapshot = json!({ "id": 7, "title": "changed" });
let baseline = revision_hash(&base_input(&snapshot));
let mutations: Vec<RevisionHashInput<'_>> = vec![
RevisionHashInput {
prev_hash: Some("aa"),
..base_input(&snapshot)
},
RevisionHashInput {
table_name: "gadgets",
..base_input(&snapshot)
},
RevisionHashInput {
tenant_id: Some("t1"),
..base_input(&snapshot)
},
RevisionHashInput {
record_id: 8,
..base_input(&snapshot)
},
RevisionHashInput {
seq: 2,
..base_input(&snapshot)
},
RevisionHashInput {
op: VersionOp::Update,
..base_input(&snapshot)
},
RevisionHashInput {
actor: "mallory",
..base_input(&snapshot)
},
RevisionHashInput {
request_id: Some("r1"),
..base_input(&snapshot)
},
RevisionHashInput {
snapshot: &other_snapshot,
..base_input(&snapshot)
},
RevisionHashInput {
valid_from: at(1),
..base_input(&snapshot)
},
RevisionHashInput {
recorded_at: at(1),
..base_input(&snapshot)
},
];
for (idx, mutated) in mutations.iter().enumerate() {
assert_ne!(
revision_hash(mutated),
baseline,
"mutation {idx} must change the hash"
);
}
}
#[test]
fn revision_hash_length_prefixes_defeat_field_smuggling() {
let snapshot = json!({});
let a = RevisionHashInput {
actor: "ab",
request_id: Some("c"),
..base_input(&snapshot)
};
let b = RevisionHashInput {
actor: "a",
request_id: Some("bc"),
..base_input(&snapshot)
};
assert_ne!(revision_hash(&a), revision_hash(&b));
}
#[test]
fn revision_hash_truncates_sub_microsecond_precision() {
let snapshot = json!({});
let micros = Utc.timestamp_nanos(1_800_000_000_000_123_000);
let nanos = Utc.timestamp_nanos(1_800_000_000_000_123_456);
let a = RevisionHashInput {
recorded_at: micros,
..base_input(&snapshot)
};
let b = RevisionHashInput {
recorded_at: nanos,
..base_input(&snapshot)
};
assert_eq!(revision_hash(&a), revision_hash(&b));
}
#[test]
fn compute_hash_matches_the_write_path_hash() {
let revisions = chain(3);
for revision in &revisions {
assert_eq!(revision.compute_hash(), revision.hash);
}
}
#[test]
fn verify_chain_accepts_an_intact_chain() {
let revisions = chain(3);
let report = verify_chain(7, &revisions);
assert!(report.is_intact(), "{report:?}");
assert_eq!(report.revisions_checked, 3);
assert_eq!(
report.head_hash.as_deref(),
Some(revisions[2].hash.as_str())
);
}
#[test]
fn verify_chain_accepts_an_empty_chain() {
let report = verify_chain(7, &[]);
assert!(report.is_intact());
assert_eq!(report.revisions_checked, 0);
assert_eq!(report.head_hash, None);
}
#[test]
fn verify_chain_has_no_false_positives_across_repeated_runs() {
let revisions = chain(12);
for _ in 0..5 {
assert!(verify_chain(7, &revisions).is_intact());
}
}
#[test]
fn verify_chain_detects_a_mutated_revision() {
let mut revisions = chain(3);
revisions[1].snapshot = json!({ "id": 7, "title": "tampered", "deleted_at": null });
let report = verify_chain(7, &revisions);
let broken = report.broken.expect("mutation must be detected");
assert_eq!(broken.kind, LedgerBreak::HashMismatch);
assert_eq!(broken.seq, 2);
assert_eq!(broken.revision_id, Some(2));
assert_eq!(report.head_hash, None);
}
#[test]
fn verify_chain_detects_a_re_hashed_mutation_at_the_next_link() {
let mut revisions = chain(3);
revisions[1].snapshot = json!({ "id": 7, "title": "tampered", "deleted_at": null });
revisions[1].hash = revisions[1].compute_hash();
let broken = verify_chain(7, &revisions)
.broken
.expect("a re-hashed mutation must break the next link");
assert_eq!(broken.kind, LedgerBreak::PrevHashMismatch);
assert_eq!(broken.seq, 3);
}
#[test]
fn verify_chain_detects_a_deleted_revision() {
let mut revisions = chain(4);
revisions.remove(1);
let broken = verify_chain(7, &revisions)
.broken
.expect("deletion detected");
assert_eq!(broken.kind, LedgerBreak::MissingRevision);
assert_eq!(broken.seq, 2);
assert_eq!(broken.revision_id, None);
}
#[test]
fn verify_chain_detects_a_deleted_head_revision() {
let mut revisions = chain(3);
revisions.remove(0);
let broken = verify_chain(7, &revisions)
.broken
.expect("head deletion detected");
assert_eq!(broken.kind, LedgerBreak::MissingRevision);
assert_eq!(broken.seq, 1);
}
#[test]
fn verify_chain_detects_an_inserted_revision() {
let mut revisions = chain(3);
let forged = LedgerRevision {
id: 99,
seq: 2,
snapshot: json!({ "id": 7, "title": "forged", "deleted_at": null }),
..revisions[1].clone()
};
revisions.insert(2, forged);
let broken = verify_chain(7, &revisions)
.broken
.expect("insertion detected");
assert_eq!(broken.kind, LedgerBreak::DuplicateSeq);
assert_eq!(broken.seq, 2);
assert_eq!(broken.revision_id, Some(99));
}
#[test]
fn verify_chain_detects_an_appended_forgery() {
let mut revisions = chain(3);
let mut forged = revisions[2].clone();
forged.id = 99;
forged.seq = 4;
forged.snapshot = json!({ "id": 7, "title": "forged", "deleted_at": null });
revisions.push(forged);
let broken = verify_chain(7, &revisions).broken.expect("append detected");
assert_eq!(broken.kind, LedgerBreak::HashMismatch);
assert_eq!(broken.seq, 4);
}
#[test]
fn verify_chain_detects_a_dangling_chain_start() {
let mut revisions = chain(2);
revisions[0].prev_hash = Some("deadbeef".to_owned());
revisions[0].hash = revisions[0].compute_hash();
revisions[1].prev_hash = Some(revisions[0].hash.clone());
revisions[1].hash = revisions[1].compute_hash();
let broken = verify_chain(7, &revisions)
.broken
.expect("dangling start detected");
assert_eq!(broken.kind, LedgerBreak::BrokenChainStart);
assert_eq!(broken.seq, 1);
}
#[test]
fn verify_chain_reports_the_first_break_when_several_exist() {
let mut revisions = chain(4);
revisions[1].snapshot = json!({ "id": 7, "title": "first tamper" });
revisions[3].snapshot = json!({ "id": 7, "title": "second tamper" });
let broken = verify_chain(7, &revisions).broken.expect("break detected");
assert_eq!(broken.seq, 2, "the earliest break must be reported");
}
#[test]
fn snapshot_as_of_returns_none_before_the_first_revision() {
let revisions = chain(3);
assert!(snapshot_as_of(&revisions, LedgerAsOf::transaction(at(0))).is_none());
}
#[test]
fn snapshot_as_of_selects_the_revision_in_force() {
let revisions = chain(3); let picked = snapshot_as_of(&revisions, LedgerAsOf::transaction(at(25)))
.expect("a revision is in force at t=25");
assert_eq!(picked.seq, 2);
assert_eq!(picked.snapshot["title"], json!("v2"));
}
#[test]
fn snapshot_as_of_is_inclusive_of_the_recording_instant() {
let revisions = chain(3);
let picked = snapshot_as_of(&revisions, LedgerAsOf::transaction(at(20)))
.expect("the revision recorded exactly at t=20 is in force");
assert_eq!(picked.seq, 2);
}
#[test]
fn snapshot_as_of_with_no_bounds_returns_the_head() {
let revisions = chain(3);
let picked = snapshot_as_of(&revisions, LedgerAsOf::default()).expect("head");
assert_eq!(picked.seq, 3);
}
#[test]
fn snapshot_as_of_separates_valid_time_from_transaction_time() {
let mut revisions = chain(2);
revisions[1].valid_from = at(5);
revisions[1].recorded_at = at(40);
revisions[1].hash = revisions[1].compute_hash();
let known_then = snapshot_as_of(&revisions, LedgerAsOf::transaction(at(30))).expect("rev");
assert_eq!(known_then.seq, 1);
let true_then = snapshot_as_of(&revisions, LedgerAsOf::valid(at(6))).expect("rev");
assert_eq!(true_then.seq, 2);
assert!(snapshot_as_of(&revisions, LedgerAsOf::bitemporal(at(30), at(6))).is_none());
let corrected =
snapshot_as_of(&revisions, LedgerAsOf::bitemporal(at(50), at(6))).expect("rev");
assert_eq!(corrected.seq, 2);
let later =
snapshot_as_of(&revisions, LedgerAsOf::bitemporal(at(50), at(20))).expect("rev");
assert_eq!(later.seq, 2);
}
#[test]
fn snapshot_as_of_hides_a_revision_that_is_not_yet_in_effect() {
let mut revisions = chain(1);
revisions[0].valid_from = at(500);
revisions[0].hash = revisions[0].compute_hash();
assert!(snapshot_as_of(&revisions, LedgerAsOf::valid(at(100))).is_none());
assert_eq!(
snapshot_as_of(&revisions, LedgerAsOf::valid(at(500)))
.expect("in effect")
.seq,
1
);
}
#[test]
fn snapshot_as_of_breaks_valid_time_ties_by_sequence() {
let mut revisions = chain(3);
for revision in &mut revisions {
revision.valid_from = at(5);
revision.hash = revision.compute_hash();
}
let picked = snapshot_as_of(&revisions, LedgerAsOf::valid(at(5))).expect("rev");
assert_eq!(picked.seq, 3, "the latest correction wins a valid-time tie");
}
#[test]
fn diff_snapshots_reports_changed_added_and_removed_columns() {
let before = json!({ "a": 1, "b": 2, "gone": true });
let after = json!({ "a": 1, "b": 3, "added": "x" });
let changes = diff_snapshots(&before, &after);
assert_eq!(changes.len(), 3);
assert_eq!(changes[0].column, "added");
assert_eq!(changes[0].before, None);
assert_eq!(changes[0].after, Some(json!("x")));
assert_eq!(changes[1].column, "b");
assert_eq!(changes[1].before, Some(json!(2)));
assert_eq!(changes[1].after, Some(json!(3)));
assert_eq!(changes[2].column, "gone");
assert_eq!(changes[2].before, Some(json!(true)));
assert_eq!(changes[2].after, None);
}
#[test]
fn diff_snapshots_of_identical_snapshots_is_empty() {
let value = json!({ "a": 1, "b": [1, 2] });
assert!(diff_snapshots(&value, &value).is_empty());
}
#[test]
fn diff_snapshots_is_sorted_regardless_of_key_order() {
let mut before = serde_json::Map::new();
before.insert("z".to_owned(), json!(1));
before.insert("a".to_owned(), json!(1));
let after = json!({ "a": 2, "z": 2 });
let changes = diff_snapshots(&serde_json::Value::Object(before), &after);
let columns: Vec<&str> = changes.iter().map(|c| c.column.as_str()).collect();
assert_eq!(columns, vec!["a", "z"]);
}
#[test]
fn diff_as_of_between_two_instants() {
let revisions = chain(3);
let diff = diff_as_of(
7,
&revisions,
LedgerAsOf::transaction(at(15)),
LedgerAsOf::transaction(at(35)),
);
assert_eq!(diff.from_seq, Some(1));
assert_eq!(diff.to_seq, Some(3));
assert_eq!(diff.changes.len(), 1);
assert_eq!(diff.changes[0].column, "title");
assert_eq!(diff.changes[0].before, Some(json!("v1")));
assert_eq!(diff.changes[0].after, Some(json!("v3")));
}
#[test]
fn diff_as_of_before_creation_reports_every_column_as_added() {
let revisions = chain(2);
let diff = diff_as_of(
7,
&revisions,
LedgerAsOf::transaction(at(0)),
LedgerAsOf::transaction(at(15)),
);
assert_eq!(diff.from_seq, None);
assert_eq!(diff.to_seq, Some(1));
assert_eq!(diff.changes.len(), 3);
assert!(diff.changes.iter().all(|c| c.before.is_none()));
}
#[test]
fn diff_as_of_over_an_unchanged_window_is_empty() {
let revisions = chain(3);
let diff = diff_as_of(
7,
&revisions,
LedgerAsOf::transaction(at(21)),
LedgerAsOf::transaction(at(29)),
);
assert!(diff.is_empty());
assert_eq!(diff.from_seq, Some(2));
assert_eq!(diff.to_seq, Some(2));
}
#[test]
fn verify_accepts_a_head_that_matches_the_live_row() {
let report = verify_chain_against(7, &chain(3), LedgerLiveState::Matches);
assert!(report.is_intact(), "{report:?}");
}
#[test]
fn verify_detects_a_truncated_tail_against_the_live_row() {
let full = chain(5);
let truncated = &full[..3];
assert!(
verify_chain(7, truncated).is_intact(),
"a truncated chain is internally consistent by construction"
);
let broken = verify_chain_against(7, truncated, LedgerLiveState::Diverged)
.broken
.expect("the live row must expose the truncation");
assert_eq!(broken.kind, LedgerBreak::LiveStateMismatch);
assert_eq!(broken.seq, 3, "the break is reported at the surviving head");
}
#[test]
fn verify_detects_a_write_that_never_appended_a_revision() {
let broken = verify_chain_against(7, &chain(2), LedgerLiveState::Diverged)
.broken
.expect("a ledger-bypassing write must be detected");
assert_eq!(broken.kind, LedgerBreak::LiveStateMismatch);
assert_eq!(broken.seq, 2);
assert!(
broken.detail.contains("without appending a revision"),
"{}",
broken.detail
);
}
#[test]
fn verify_does_not_accuse_a_row_that_predates_ledgering() {
let report = verify_chain_against(7, &[], LedgerLiveState::Diverged);
assert!(report.is_intact(), "{report:?}");
assert_eq!(report.revisions_checked, 0);
assert_eq!(report.head_hash, None);
}
#[test]
fn verify_detects_a_row_erased_out_of_band() {
let revisions = chain(2);
let broken = verify_chain_against(7, &revisions, LedgerLiveState::Absent)
.broken
.expect("a chain whose record no longer exists must be detected");
assert_eq!(broken.kind, LedgerBreak::LiveStateMismatch);
assert_eq!(broken.seq, 2);
}
#[test]
fn verify_accepts_an_absent_record_with_no_revisions() {
assert!(verify_chain_against(7, &[], LedgerLiveState::Absent).is_intact());
}
#[test]
fn a_chain_internal_break_outranks_a_live_state_mismatch() {
let mut revisions = chain(3);
revisions[1].snapshot = json!({ "id": 7, "title": "tampered" });
let broken = verify_chain_against(7, &revisions, LedgerLiveState::Diverged)
.broken
.expect("break detected");
assert_eq!(broken.kind, LedgerBreak::HashMismatch);
assert_eq!(broken.seq, 2);
}
#[test]
fn verify_skips_the_cross_check_when_the_head_moved_under_the_reader() {
let report = verify_chain_against(7, &chain(2), LedgerLiveState::NotChecked);
assert!(report.is_intact(), "{report:?}");
assert_eq!(report.head_hash.as_deref(), Some(chain(2)[1].hash.as_str()));
}
#[test]
fn the_ledger_postgres_migrations_are_identical_in_both_sets() {
for (control, shard_required) in [
(
include_str!("../migrations/20260826000000_create_ledger_revisions/up.sql"),
include_str!(
"../version_history_migrations/20260826000000_create_ledger_revisions/up.sql"
),
),
(
include_str!("../migrations/20260826000000_create_ledger_revisions/down.sql"),
include_str!(
"../version_history_migrations/20260826000000_create_ledger_revisions/down.sql"
),
),
(
include_str!("../migrations/20260901213107_create_ledger_high_water/up.sql"),
include_str!(
"../version_history_migrations/20260901213107_create_ledger_high_water/up.sql"
),
),
(
include_str!("../migrations/20260901213107_create_ledger_high_water/down.sql"),
include_str!(
"../version_history_migrations/20260901213107_create_ledger_high_water/down.sql"
),
),
] {
assert_eq!(
control, shard_required,
"the control-plane and shard-required copies of a ledger migration must \
stay byte-identical; they share a version string, so a drift is silent"
);
}
}
#[test]
fn both_tiers_declare_the_same_high_water_columns() {
let pg = include_str!(
"../version_history_migrations/20260901213107_create_ledger_high_water/up.sql"
);
let sqlite = include_str!(
"../version_history_migrations_sqlite/20260901213107_create_ledger_high_water/up.sql"
);
for fragment in [
LEDGER_HIGH_WATER_TABLE,
"table_name",
"tenant_key",
"record_id",
"high_seq",
"head_hash",
"recorded_at",
"PRIMARY KEY (table_name, tenant_key, record_id)",
] {
assert!(pg.contains(fragment), "postgres migration lacks {fragment}");
assert!(
sqlite.contains(fragment),
"sqlite migration lacks {fragment}"
);
}
}
fn mark_for(revisions: &[LedgerRevision]) -> LedgerHighWater {
let head = revisions.last().expect("a non-empty chain");
LedgerHighWater {
record_id: head.record_id,
seq: head.seq,
hash: head.hash.clone(),
recorded_at: head.recorded_at,
}
}
#[test]
fn verify_accepts_a_chain_its_high_water_mark_agrees_with() {
let revisions = chain(3);
let report = verify_chain_with_high_water(
7,
&revisions,
LedgerLiveState::Matches,
LedgerHighWaterState::Present(&mark_for(&revisions)),
);
assert!(report.is_intact(), "{report:?}");
assert_eq!(
report.head_hash.as_deref(),
Some(revisions[2].hash.as_str())
);
}
#[test]
fn verify_reports_a_truncated_tail_the_high_water_mark_remembers() {
let full = chain(3);
let mark = mark_for(&full);
let truncated = &full[..2];
let broken = verify_chain_with_high_water(
7,
truncated,
LedgerLiveState::Matches,
LedgerHighWaterState::Present(&mark),
)
.broken
.expect("the mark must expose the deleted head revision");
assert_eq!(broken.kind, LedgerBreak::MissingRevision);
assert_eq!(
broken.seq, 3,
"the absent sequence number, not the survivor"
);
assert_eq!(broken.revision_id, None);
}
#[test]
fn verify_reports_the_gap_a_post_truncation_append_leaves() {
let full = chain(4);
let mut surviving: Vec<LedgerRevision> = full[..2].to_vec();
surviving.push(full[3].clone());
let broken = verify_chain_with_high_water(
7,
&surviving,
LedgerLiveState::Matches,
LedgerHighWaterState::Present(&mark_for(&surviving)),
)
.broken
.expect("the gap left by the re-numbered append must be reported");
assert_eq!(broken.kind, LedgerBreak::MissingRevision);
assert_eq!(broken.seq, 3);
}
#[test]
fn verify_reports_a_wholly_erased_chain_against_its_mark() {
let mark = mark_for(&chain(3));
let broken = verify_chain_with_high_water(
7,
&[],
LedgerLiveState::Absent,
LedgerHighWaterState::Present(&mark),
)
.broken
.expect("an erased chain whose mark survives must be reported");
assert_eq!(broken.kind, LedgerBreak::MissingRevision);
assert_eq!(broken.seq, 1);
}
#[test]
fn verify_still_accepts_a_row_that_predates_ledgering() {
let report = verify_chain_with_high_water(
7,
&[],
LedgerLiveState::Diverged,
LedgerHighWaterState::Absent,
);
assert!(report.is_intact(), "{report:?}");
assert_eq!(report.revisions_checked, 0);
}
#[test]
fn verify_reports_a_deleted_high_water_row() {
let broken = verify_chain_with_high_water(
7,
&chain(3),
LedgerLiveState::Matches,
LedgerHighWaterState::Absent,
)
.broken
.expect("a chain with no mark at all must be reported");
assert_eq!(broken.kind, LedgerBreak::HighWaterMissing);
assert_eq!(broken.seq, 3);
}
#[test]
fn verify_reports_a_high_water_mark_rolled_backwards() {
let revisions = chain(3);
let mut mark = mark_for(&revisions);
mark.seq = 1;
mark.hash = revisions[0].hash.clone();
mark.recorded_at = revisions[0].recorded_at;
let broken = verify_chain_with_high_water(
7,
&revisions,
LedgerLiveState::Matches,
LedgerHighWaterState::Present(&mark),
)
.broken
.expect("a rolled-back mark must be reported");
assert_eq!(broken.kind, LedgerBreak::HighWaterBehind);
assert_eq!(broken.seq, 3);
}
#[test]
fn verify_reports_a_rewritten_high_water_hash() {
let revisions = chain(3);
let mut mark = mark_for(&revisions);
mark.hash = "0".repeat(64);
let broken = verify_chain_with_high_water(
7,
&revisions,
LedgerLiveState::Matches,
LedgerHighWaterState::Present(&mark),
)
.broken
.expect("a mark that no longer names the head revision must be reported");
assert_eq!(broken.kind, LedgerBreak::HighWaterMismatch);
assert_eq!(broken.seq, 3);
}
#[test]
fn verify_reports_a_high_water_mark_whose_instant_was_rewritten() {
let revisions = chain(3);
let mut mark = mark_for(&revisions);
mark.recorded_at = at(9_999);
let broken = verify_chain_with_high_water(
7,
&revisions,
LedgerLiveState::Matches,
LedgerHighWaterState::Present(&mark),
)
.broken
.expect("a mark whose instant disagrees with the head must be reported");
assert_eq!(broken.kind, LedgerBreak::HighWaterMismatch);
}
#[test]
fn a_chain_internal_break_outranks_a_high_water_break() {
let mut revisions = chain(3);
let mark = mark_for(&revisions);
revisions[1].snapshot = json!({ "id": 7, "title": "tampered" });
let broken = verify_chain_with_high_water(
7,
&revisions,
LedgerLiveState::Matches,
LedgerHighWaterState::Present(&mark),
)
.broken
.expect("break detected");
assert_eq!(broken.kind, LedgerBreak::HashMismatch);
assert_eq!(broken.seq, 2);
}
#[test]
fn a_high_water_break_outranks_a_live_state_mismatch() {
let full = chain(3);
let mark = mark_for(&full);
let broken = verify_chain_with_high_water(
7,
&full[..2],
LedgerLiveState::Diverged,
LedgerHighWaterState::Present(&mark),
)
.broken
.expect("break detected");
assert_eq!(broken.kind, LedgerBreak::MissingRevision);
assert_eq!(broken.seq, 3);
}
#[test]
fn verify_chain_against_still_checks_without_a_mark() {
let report = verify_chain_against(7, &chain(3), LedgerLiveState::Matches);
assert!(report.is_intact(), "{report:?}");
}
#[test]
fn verify_reports_a_transaction_time_that_moves_backwards() {
let mut revisions = chain(3);
revisions[2].recorded_at = revisions[1].recorded_at - chrono::Duration::seconds(1);
revisions[2].hash = revisions[2].compute_hash();
let broken = verify_chain(7, &revisions)
.broken
.expect("a backwards transaction time must be reported");
assert_eq!(broken.kind, LedgerBreak::RecordedAtRegression);
assert_eq!(broken.seq, 3);
}
#[test]
fn a_transaction_time_regression_never_masks_a_real_break() {
let full = chain(4);
let mark = mark_for(&full);
let mut truncated = full[..3].to_vec();
truncated[2].recorded_at = truncated[1].recorded_at - chrono::Duration::seconds(1);
truncated[2].hash = truncated[2].compute_hash();
let mut fixed = truncated.clone();
fixed[2].prev_hash = Some(fixed[1].hash.clone());
fixed[2].hash = fixed[2].compute_hash();
let broken = verify_chain_with_high_water(
7,
&fixed,
LedgerLiveState::Matches,
LedgerHighWaterState::Present(&mark),
)
.broken
.expect("break detected");
assert_eq!(
broken.kind,
LedgerBreak::MissingRevision,
"the deleted revision outranks the backwards clock: {broken:?}"
);
assert_eq!(broken.seq, 4);
let broken = verify_chain(7, &fixed)
.broken
.expect("the regression is still reported on its own");
assert_eq!(broken.kind, LedgerBreak::RecordedAtRegression);
assert_eq!(broken.seq, 3);
}
#[test]
fn equal_transaction_times_are_not_a_regression() {
let mut revisions = chain(3);
revisions[2].recorded_at = revisions[1].recorded_at;
revisions[2].hash = revisions[2].compute_hash();
assert!(verify_chain(7, &revisions).is_intact());
}
#[test]
fn valid_time_may_move_backwards_without_being_a_break() {
let mut revisions = chain(3);
revisions[2].valid_from = revisions[0].valid_from - chrono::Duration::days(30);
revisions[2].hash = revisions[2].compute_hash();
assert!(verify_chain(7, &revisions).is_intact());
}
#[test]
fn monotonic_recorded_at_never_precedes_its_floor() {
let now = at(100);
assert_eq!(monotonic_recorded_at(now, None), Some(now));
assert_eq!(monotonic_recorded_at(now, Some(at(50))), Some(now));
assert_eq!(monotonic_recorded_at(now, Some(at(150))), Some(at(150)));
assert_eq!(monotonic_recorded_at(now, Some(now)), Some(now));
}
#[test]
fn monotonic_recorded_at_truncates_both_sides_to_micros() {
let now = at(100) + chrono::Duration::nanoseconds(1_500);
assert_eq!(
monotonic_recorded_at(now, None),
Some(at(100) + chrono::Duration::microseconds(1))
);
let floor = at(200) + chrono::Duration::nanoseconds(1_500);
assert_eq!(
monotonic_recorded_at(at(100), Some(floor)),
Some(at(200) + chrono::Duration::microseconds(1))
);
}
#[test]
fn monotonic_recorded_at_refuses_a_floor_far_ahead_of_the_database_clock() {
let now = at(0);
let inside = now + chrono::Duration::seconds(LEDGER_MAX_CLOCK_SKEW_SECS);
assert_eq!(monotonic_recorded_at(now, Some(inside)), Some(inside));
let outside = inside + chrono::Duration::microseconds(1);
assert_eq!(
monotonic_recorded_at(now, Some(outside)),
None,
"a floor past the tolerated skew must refuse the write, not honour it"
);
assert_eq!(
monotonic_recorded_at(now, Some(now + chrono::Duration::days(365_000))),
None
);
}
#[test]
fn sqlite_instants_round_trip_from_the_database_clock() {
let parsed = parse_sqlite_instant("2026-09-01 18:08:53.597")
.expect("the documented SQLite clock encoding parses");
assert_eq!(
parsed.to_rfc3339_opts(chrono::SecondsFormat::Micros, true),
"2026-09-01T18:08:53.597000Z"
);
assert_eq!(
parse_sqlite_instant("2026-09-01 18:08:53")
.expect("a fractionless instant parses")
.to_rfc3339_opts(chrono::SecondsFormat::Micros, true),
"2026-09-01T18:08:53.000000Z"
);
assert_eq!(parse_sqlite_instant("not an instant"), None);
assert_eq!(parse_sqlite_instant(""), None);
}
#[test]
fn verify_refuses_to_follow_a_saturated_sequence_number() {
let mut revisions = chain(1);
revisions[0].seq = i64::MAX;
revisions[0].hash = revisions[0].compute_hash();
let broken = verify_chain(7, &revisions)
.broken
.expect("a chain that starts at i64::MAX cannot be checked");
assert_eq!(broken.kind, LedgerBreak::MissingRevision);
let mut revisions = chain(1);
revisions[0].seq = 1;
let mut tail = revisions[0].clone();
tail.id = 2;
tail.seq = i64::MAX;
tail.prev_hash = Some(revisions[0].hash.clone());
tail.hash = tail.compute_hash();
revisions.push(tail);
let broken = verify_chain(7, &revisions).broken.expect("gap detected");
assert_eq!(broken.kind, LedgerBreak::MissingRevision);
assert_eq!(broken.seq, 2);
}
#[test]
fn revision_hash_distinguishes_absent_from_empty_optionals() {
let snapshot = json!({});
for (absent, empty) in [
(
RevisionHashInput {
tenant_id: None,
..base_input(&snapshot)
},
RevisionHashInput {
tenant_id: Some(""),
..base_input(&snapshot)
},
),
(
RevisionHashInput {
request_id: None,
..base_input(&snapshot)
},
RevisionHashInput {
request_id: Some(""),
..base_input(&snapshot)
},
),
(
RevisionHashInput {
prev_hash: None,
..base_input(&snapshot)
},
RevisionHashInput {
prev_hash: Some(""),
..base_input(&snapshot)
},
),
] {
assert_ne!(
revision_hash(&absent),
revision_hash(&empty),
"a NULL optional must not hash like an empty string"
);
}
}
#[test]
fn as_of_transaction_only_returns_the_newest_revision_the_database_held() {
let mut revisions = chain(2);
revisions[1].valid_from = at(5);
revisions[1].hash = revisions[1].compute_hash();
let now = snapshot_as_of(&revisions, LedgerAsOf::transaction(at(100)))
.expect("a revision is in force");
assert_eq!(
now.seq, 2,
"a transaction-time query must return the latest state, not the \
one with the greatest valid_from"
);
let head = snapshot_as_of(&revisions, LedgerAsOf::default()).expect("head");
assert_eq!(head.seq, 2, "the unbounded query is the live state");
}
#[test]
fn a_valid_time_bound_filters_eligibility_rather_than_reordering_the_chain() {
let mut revisions = chain(2);
revisions[1].valid_from = at(5);
revisions[1].hash = revisions[1].compute_hash();
assert_eq!(
snapshot_as_of(&revisions, LedgerAsOf::valid(at(6)))
.expect("rev")
.seq,
2
);
assert_eq!(
snapshot_as_of(&revisions, LedgerAsOf::valid(at(20)))
.expect("rev")
.seq,
2
);
assert!(snapshot_as_of(&revisions, LedgerAsOf::valid(at(1))).is_none());
}
#[test]
fn ledger_break_strings_are_stable() {
assert_eq!(LedgerBreak::HashMismatch.as_str(), "hash_mismatch");
assert_eq!(LedgerBreak::PrevHashMismatch.as_str(), "prev_hash_mismatch");
assert_eq!(LedgerBreak::MissingRevision.as_str(), "missing_revision");
assert_eq!(LedgerBreak::DuplicateSeq.as_str(), "duplicate_seq");
assert_eq!(LedgerBreak::BrokenChainStart.as_str(), "broken_chain_start");
assert_eq!(format!("{}", LedgerBreak::HashMismatch), "hash_mismatch");
}
#[test]
fn verification_serde_roundtrip() {
let report = verify_chain(7, &chain(2));
let json = serde_json::to_string(&report).expect("serialize");
let back: LedgerVerification = serde_json::from_str(&json).expect("deserialize");
assert_eq!(back, report);
}
#[test]
fn ledger_error_messages_name_the_record() {
let err = LedgerError::ChainBroken {
table: "widgets".to_owned(),
record_id: 7,
seq: 2,
detail: "hash_mismatch".to_owned(),
};
let rendered = err.to_string();
assert!(rendered.contains("widgets#7"), "{rendered}");
assert!(rendered.contains("revision 2"), "{rendered}");
}
#[test]
fn valid_time_value_shapes() {
let instant = at(3);
assert_eq!(
LedgerValidTimeValue::ledger_valid_from(&instant),
Some(instant)
);
assert_eq!(
LedgerValidTimeValue::ledger_valid_from(&Some(instant)),
Some(instant)
);
assert_eq!(
LedgerValidTimeValue::ledger_valid_from(&None::<DateTime<Utc>>),
None
);
let naive = instant.naive_utc();
assert_eq!(
LedgerValidTimeValue::ledger_valid_from(&naive),
Some(instant)
);
assert_eq!(
LedgerValidTimeValue::ledger_valid_from(&None::<chrono::NaiveDateTime>),
None
);
}
inventory::submit! {
crate::encryption::EncryptedColumnDescriptor {
model: "LedgerFidelityModel",
table: "ledger_fidelity_table",
column: "secret",
deterministic: false,
admin_visible: false,
versioned_ciphertext: false,
}
}
#[test]
#[allow(
clippy::needless_borrow,
reason = "the extra borrow selects the fallback impl"
)]
fn non_finite_probe_flags_floats_the_snapshot_would_null() {
for bad in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] {
assert!((&NonFiniteProbe(&bad)).autumn_has_non_finite());
assert!((&NonFiniteProbe(&Some(bad))).autumn_has_non_finite());
assert!((&NonFiniteProbe(&vec![1.0, bad])).autumn_has_non_finite());
}
assert!((&NonFiniteProbe(&f32::NAN)).autumn_has_non_finite());
}
#[test]
#[allow(
clippy::needless_borrow,
reason = "the extra borrow selects the fallback impl"
)]
fn non_finite_probe_accepts_finite_and_foreign_types() {
assert!(!(&NonFiniteProbe(&1.5_f64)).autumn_has_non_finite());
assert!(!(&NonFiniteProbe(&None::<f64>)).autumn_has_non_finite());
assert!(!(&NonFiniteProbe(&"NaN".to_string())).autumn_has_non_finite());
assert!(!(&NonFiniteProbe(&7_i64)).autumn_has_non_finite());
}
#[test]
fn unrecovered_envelope_is_chain_unreadable_and_names_column_and_revision() {
let snapshot = serde_json::json!({"id": 1, "secret": "not-a-real-envelope"});
let err = ensure_snapshot_recoverable("ledger_fidelity_table", 1, 4, &snapshot)
.expect_err("an unrecovered envelope must not reconstruct");
let text = err.to_string();
assert!(text.contains("secret"), "names the column: {text}");
assert!(text.contains("revision 4"), "names the revision: {text}");
}
#[test]
fn snapshot_without_encrypted_columns_is_recoverable() {
let snapshot = serde_json::json!({"id": 1, "secret": "plain"});
ensure_snapshot_recoverable("some_other_table", 1, 1, &snapshot).unwrap();
let null_secret = serde_json::json!({"id": 1, "secret": null});
ensure_snapshot_recoverable("ledger_fidelity_table", 1, 1, &null_secret).unwrap();
}
#[test]
fn refuse_non_finite_passes_none_and_refuses_a_column() {
refuse_non_finite("t", 1, None).unwrap();
let err = refuse_non_finite("t", 9, Some("score")).expect_err("must refuse");
let text = err.to_string();
assert!(text.contains("t#9") && text.contains("score"), "{text}");
}
#[test]
fn non_finite_error_names_table_record_and_column() {
let text = LedgerError::NonFiniteValue {
table: "t".into(),
record_id: 9,
column: "score".into(),
}
.to_string();
assert!(text.contains("t#9") && text.contains("score"), "{text}");
}
fn tamper_cases() -> Vec<Vec<LedgerRevision>> {
let mut cases = vec![Vec::new(), chain(1), chain(5), chain(6)];
let mut mutated = chain(5);
mutated[2].snapshot = json!({ "id": 7, "title": "tampered" });
cases.push(mutated);
let mut relinked = chain(5);
relinked[2].snapshot = json!({ "id": 7, "title": "tampered" });
relinked[2].hash = relinked[2].compute_hash();
cases.push(relinked);
let mut gap = chain(5);
gap.remove(2);
cases.push(gap);
let mut no_start = chain(5);
no_start.remove(0);
cases.push(no_start);
let mut duplicate = chain(5);
let forged = LedgerRevision {
id: 99,
..duplicate[1].clone()
};
duplicate.insert(2, forged);
cases.push(duplicate);
let mut regression = chain(5);
regression[3].recorded_at = regression[2].recorded_at - chrono::Duration::seconds(1);
regression[3].hash = regression[3].compute_hash();
regression[4].prev_hash = Some(regression[3].hash.clone());
regression[4].hash = regression[4].compute_hash();
cases.push(regression);
let mut saturated = chain(2);
saturated[1].seq = i64::MAX;
saturated[1].hash = saturated[1].compute_hash();
cases.push(saturated);
cases
}
fn oracle_verify(
record_id: i64,
revisions: &[LedgerRevision],
live: LedgerLiveState,
high_water: LedgerHighWaterState<'_>,
) -> LedgerVerification {
fn first_break(revisions: &[LedgerRevision]) -> Option<LedgerBreakReport> {
let mut expected_seq: i64 = 1;
let mut prev: Option<ChainTip> = None;
for revision in revisions {
if revision.seq != expected_seq {
return Some(sequence_break(revision, expected_seq, prev.is_none()));
}
if let Some(report) = link_break(revision, prev.as_ref()) {
return Some(report);
}
let Some(next_seq) = revision.seq.checked_add(1) else {
return Some(LedgerBreakReport {
seq: revision.seq,
revision_id: Some(revision.id),
kind: LedgerBreak::UnusableSeq,
detail: "sequence number is at i64::MAX and cannot be followed; \
the chain cannot be continued or checked past this revision"
.to_owned(),
});
};
expected_seq = next_seq;
prev = Some(ChainTip::of(revision));
}
None
}
let head = revisions.last().map(ChainTip::of);
let broken = first_break(revisions)
.or_else(|| high_water_break(head.as_ref(), high_water))
.or_else(|| live_state_break(head.as_ref(), live))
.or_else(|| {
revisions
.windows(2)
.find_map(|pair| recorded_at_break(&ChainTip::of(&pair[0]), &pair[1]))
});
LedgerVerification {
record_id,
revisions_checked: revisions.len(),
head_hash: if broken.is_none() {
head.map(|tip| tip.hash)
} else {
None
},
broken,
}
}
#[test]
fn the_streaming_verifier_matches_the_slice_oracle() {
let full = chain(6);
let marks = [None, Some(mark_for(&full[..5])), Some(mark_for(&full))];
let lives = [
LedgerLiveState::NotChecked,
LedgerLiveState::Absent,
LedgerLiveState::Matches,
LedgerLiveState::Diverged,
];
let mut broken_kinds = std::collections::BTreeSet::new();
for revisions in tamper_cases() {
for live in lives {
for mark in &marks {
let states = [
LedgerHighWaterState::NotChecked,
mark.as_ref()
.map_or(LedgerHighWaterState::Absent, LedgerHighWaterState::Present),
];
for high_water in states {
let expected = oracle_verify(7, &revisions, live, high_water);
let mut verifier = LedgerChainVerifier::new(7);
for revision in &revisions {
verifier.push(revision);
}
assert_eq!(verifier.finish(live, high_water), expected);
if let Some(broken) = &expected.broken {
broken_kinds.insert(broken.kind.as_str());
}
}
}
}
}
for kind in [
"hash_mismatch",
"prev_hash_mismatch",
"missing_revision",
"duplicate_seq",
"recorded_at_regression",
"live_state_mismatch",
"high_water_missing",
"high_water_behind",
] {
assert!(
broken_kinds.contains(kind),
"no case reaches {kind}: {broken_kinds:?}"
);
}
}
#[test]
fn a_page_request_clamps_its_limit() {
assert_eq!(LedgerPageRequest::first(0).effective_limit(), 1);
assert_eq!(LedgerPageRequest::first(10).effective_limit(), 10);
assert_eq!(
LedgerPageRequest::first(u32::MAX).effective_limit(),
LedgerPageRequest::MAX_LIMIT
);
assert_eq!(
LedgerPageRequest::default().effective_limit(),
LedgerPageRequest::DEFAULT_LIMIT
);
let cursor = LedgerCursor { seq: 3, id: 30 };
assert_eq!(LedgerPageRequest::after(cursor, 5).after, Some(cursor));
}
#[test]
fn a_page_carries_a_cursor_only_when_more_rows_exist() {
let revisions = chain(3);
let full = LedgerRevisionPage::from_rows(revisions.clone(), 3);
assert!(full.next.is_none());
assert_eq!(full.revisions.len(), 3);
let more = LedgerRevisionPage::from_rows(revisions, 2);
assert_eq!(more.revisions.len(), 2);
assert_eq!(more.next, Some(LedgerCursor { seq: 2, id: 2 }));
assert_eq!(
more.next_request(5),
Some(LedgerPageRequest::after(LedgerCursor { seq: 2, id: 2 }, 5))
);
assert_eq!(full.next_request(5), None);
}
#[test]
fn a_tenant_change_is_refused() {
refuse_tenant_change("t", 1, Some("tenant-alpha"), Some("tenant-alpha")).unwrap();
refuse_tenant_change("t", 1, None, None).unwrap();
let err = refuse_tenant_change("t", 9, Some("tenant-alpha"), Some("tenant-beta"))
.expect_err("refused");
let text = err.to_string();
assert!(text.contains("t#9"), "{text}");
assert!(
!text.contains("tenant-alpha") && !text.contains("tenant-beta"),
"{text}"
);
}
#[test]
fn a_string_tenant_column_is_a_ledger_tenant() {
let tenant = String::from("acme");
assert_eq!(LedgerTenantColumn::ledger_tenant_id(&tenant), "acme");
}
#[test]
fn stored_snapshot_accepts_canonical_text() {
let text = r#"{"a":1,"b":[1,2]}"#;
let value = parse_stored_snapshot("t", 1, 1, text).unwrap();
assert_eq!(value, json!({"a": 1, "b": [1, 2]}));
}
#[test]
fn stored_snapshot_refuses_re_encoded_text() {
for text in [r#"{ "a": 1 }"#, r#"{"b":1,"a":2}"#, r#"{"a":1,"a":2}"#] {
let err = parse_stored_snapshot("t", 1, 3, text).expect_err(text);
let shown = err.to_string();
assert!(
shown.contains("canonical") && shown.contains('3'),
"{shown}"
);
}
}
#[test]
fn stored_snapshot_refuses_invalid_json() {
let err = parse_stored_snapshot("t", 1, 2, "{").expect_err("invalid");
assert!(err.to_string().contains("unreadable"), "{err}");
}
#[test]
fn stored_snapshot_accepts_floats_the_writer_produced() {
for x in [4.536_315_785_538_721e-181_f64, 4.138_526_855_143_091e260] {
let text = canonical_json(&json!({ "x": x }));
parse_stored_snapshot("t", 1, 1, &text).expect(&text);
}
}
#[test]
fn schema_detail_hides_quoted_values() {
let shown = redact_quoted(r#"invalid type: string "sk-secret", expected i64"#);
assert!(!shown.contains("sk-secret"), "{shown}");
assert!(shown.contains("expected i64"), "{shown}");
}
#[test]
fn a_non_canonical_read_error_becomes_a_break() {
let err = crate::AutumnError::internal_server_error(LedgerError::SnapshotNotCanonical {
table: "t".into(),
record_id: 4,
seq: 2,
});
let report = verification_from_read_error(4, err).unwrap();
let broken = report.broken.expect("a break");
assert_eq!(broken.kind, LedgerBreak::SnapshotNotCanonical);
assert_eq!(broken.seq, 2);
assert_eq!(
LedgerBreak::SnapshotNotCanonical.as_str(),
"snapshot_not_canonical"
);
}
#[test]
fn other_read_errors_pass_through() {
let err = crate::AutumnError::internal_server_error(LedgerError::ChainUnreadable {
table: "t".into(),
record_id: 4,
detail: "x".into(),
});
assert!(verification_from_read_error(4, err).is_err());
}
fn rename_subject(from: u32, snapshot: &mut serde_json::Value) -> Result<(), String> {
if from < 2 {
let object = snapshot.as_object_mut().ok_or("not an object")?;
if let Some(subject) = object.remove("subject") {
object.insert("title".into(), subject);
}
}
Ok(())
}
fn refuse_upcast(_: u32, _: &mut serde_json::Value) -> Result<(), String> {
Err("no rule for this shape".into())
}
#[allow(
clippy::unnecessary_wraps,
reason = "the signature must match `SnapshotUpcast`"
)]
fn note_from_version(from: u32, snapshot: &mut serde_json::Value) -> Result<(), String> {
snapshot["seen"] = json!(from);
Ok(())
}
fn mismatch(err: &crate::AutumnError) -> (i64, i64, String) {
match err.downcast_ref::<LedgerError>() {
Some(LedgerError::SnapshotSchemaMismatch {
record_id,
seq,
detail,
..
}) => (*record_id, *seq, detail.clone()),
other => panic!("expected SnapshotSchemaMismatch, got {other:?}"),
}
}
#[test]
fn a_version_one_model_adds_no_marker() {
let mut snapshot = json!({ "title": "a" });
stamp_schema_version(&mut snapshot, 1);
assert_eq!(snapshot, json!({ "title": "a" }));
}
#[test]
fn a_later_schema_version_is_stamped_into_the_snapshot() {
let mut snapshot = json!({ "title": "a" });
stamp_schema_version(&mut snapshot, 3);
assert_eq!(snapshot[SNAPSHOT_SCHEMA_KEY], json!(3));
assert!(canonical_json(&snapshot).contains(SNAPSHOT_SCHEMA_KEY));
}
#[test]
fn an_unmarked_snapshot_is_version_one() {
let snapshot = json!({ "title": "a" });
let (value, stored) = upcast_snapshot("t", 1, 1, &snapshot, 1, refuse_upcast).unwrap();
assert_eq!((value, stored), (snapshot, 1));
}
#[test]
fn an_older_snapshot_is_converted_by_the_hook() {
let snapshot = json!({ "subject": "a", "amount": 1 });
let (value, stored) = upcast_snapshot("t", 1, 1, &snapshot, 2, rename_subject).unwrap();
assert_eq!(stored, 1);
assert_eq!(value, json!({ "title": "a", "amount": 1 }));
}
#[test]
fn the_marker_is_removed_before_the_model_reads_the_snapshot() {
let mut snapshot = json!({ "title": "a" });
stamp_schema_version(&mut snapshot, 2);
let (value, stored) = upcast_snapshot("t", 1, 1, &snapshot, 2, refuse_upcast).unwrap();
assert_eq!(stored, 2);
assert_eq!(value, json!({ "title": "a" }));
}
#[test]
fn a_snapshot_at_the_current_version_skips_the_hook() {
let mut snapshot = json!({ "title": "a" });
stamp_schema_version(&mut snapshot, 2);
upcast_snapshot("t", 1, 1, &snapshot, 2, refuse_upcast).expect("the hook is not called");
}
#[test]
fn the_hook_receives_the_stored_version() {
let mut snapshot = json!({ "title": "a" });
stamp_schema_version(&mut snapshot, 2);
let (value, _) = upcast_snapshot("t", 1, 1, &snapshot, 3, note_from_version).unwrap();
assert_eq!(value["seen"], json!(2));
}
#[test]
fn a_failing_hook_is_a_schema_error_and_not_tamper_evidence() {
let err =
upcast_snapshot("t", 7, 4, &json!({ "subject": "a" }), 2, refuse_upcast).unwrap_err();
let (record_id, seq, detail) = mismatch(&err);
assert_eq!((record_id, seq), (7, 4));
assert!(detail.contains("no rule for this shape"), "{detail}");
assert!(detail.contains("schema version 1"), "{detail}");
}
#[test]
fn a_snapshot_from_a_newer_model_is_flagged_and_not_converted() {
let mut snapshot = json!({ "title": "a" });
stamp_schema_version(&mut snapshot, 5);
let (value, stored) = upcast_snapshot("t", 1, 1, &snapshot, 2, refuse_upcast).unwrap();
assert_eq!(stored, 5);
assert_eq!(value, json!({ "title": "a" }));
let shown = schema_mismatch_at("t", 1, 1, stored, 2, "cannot decode").to_string();
assert!(shown.contains("schema version 5"), "{shown}");
assert!(shown.contains("newer"), "{shown}");
}
#[test]
fn a_damaged_marker_is_a_schema_error() {
for bad in [json!("x"), json!(0), json!(-1), json!(1.5), json!(null)] {
let snapshot = json!({ "title": "a", SNAPSHOT_SCHEMA_KEY: bad });
let err = upcast_snapshot("t", 1, 1, &snapshot, 2, rename_subject).unwrap_err();
let (_, _, detail) = mismatch(&err);
assert!(detail.contains("schema version marker"), "{detail}");
}
}
#[test]
fn a_snapshot_that_is_not_an_object_passes_through() {
let (value, stored) = upcast_snapshot("t", 1, 1, &json!([1, 2]), 2, refuse_upcast).unwrap();
assert_eq!((value, stored), (json!([1, 2]), 1));
}
#[test]
fn the_schema_error_names_the_versions_and_the_remedies() {
let shown = schema_mismatch_at("t", 1, 2, 1, 3, "cannot decode").to_string();
assert!(shown.contains("schema version 1"), "{shown}");
assert!(shown.contains("model is at 3"), "{shown}");
assert!(shown.contains("upcast"), "{shown}");
assert!(shown.contains("serde(default)"), "{shown}");
}
#[test]
fn the_live_check_ignores_columns_only_one_side_has_after_a_schema_change() {
let head = json!({ "title": "a" });
let live = json!({ "title": "a", "region": "eu" });
assert!(snapshots_agree(&head, &live, true));
assert!(snapshots_agree(&live, &head, true));
}
#[test]
fn the_live_check_is_strict_when_no_schema_change_is_declared() {
let head = json!({ "title": "a" });
let live = json!({ "title": "a", "role": "admin" });
assert!(!snapshots_agree(&head, &live, false));
assert!(!snapshots_agree(&live, &head, false));
assert!(snapshots_agree(&live, &live, false));
}
#[test]
fn the_live_check_still_sees_a_changed_value() {
let head = json!({ "title": "a", "amount": 1 });
let live = json!({ "title": "a", "amount": 2, "region": "eu" });
assert!(!snapshots_agree(&head, &live, true));
assert!(!snapshots_agree(&head, &live, false));
}
#[test]
fn the_live_check_compares_non_objects_whole() {
assert!(snapshots_agree(&json!(1), &json!(1), true));
assert!(!snapshots_agree(&json!(1), &json!(2), true));
assert!(!snapshots_agree(&json!({}), &json!(2), true));
}
fn panic_upcast(_: u32, _: &mut serde_json::Value) -> Result<(), String> {
panic!("the hook has a bug")
}
#[test]
fn a_hook_that_panics_is_a_schema_error() {
let err =
upcast_snapshot("t", 3, 2, &json!({ "subject": "a" }), 2, panic_upcast).unwrap_err();
let (record_id, seq, detail) = mismatch(&err);
assert_eq!((record_id, seq), (3, 2));
assert!(detail.contains("the hook panicked"), "{detail}");
}
#[test]
fn a_marked_snapshot_read_by_a_version_one_model_loses_the_marker() {
let mut snapshot = json!({ "title": "a" });
stamp_schema_version(&mut snapshot, 2);
let (value, stored) = upcast_snapshot("t", 1, 1, &snapshot, 1, refuse_upcast).unwrap();
assert_eq!((value, stored), (json!({ "title": "a" }), 2));
let shown = schema_mismatch_at("t", 1, 1, stored, 1, "cannot decode").to_string();
assert!(shown.contains("newer"), "{shown}");
}
}