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,
}
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::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 checked = revisions.len();
let broken = first_break(revisions)
.or_else(|| high_water_break(revisions, high_water))
.or_else(|| live_state_break(revisions, live))
.or_else(|| recorded_at_break(revisions));
let head_hash = if broken.is_none() {
revisions.last().map(|r| r.hash.clone())
} else {
None
};
LedgerVerification {
record_id,
revisions_checked: checked,
head_hash,
broken,
}
}
fn high_water_break(
revisions: &[LedgerRevision],
high_water: LedgerHighWaterState<'_>,
) -> Option<LedgerBreakReport> {
let high_water = match high_water {
LedgerHighWaterState::NotChecked => return None,
LedgerHighWaterState::Absent => None,
LedgerHighWaterState::Present(mark) => Some(mark),
};
match (revisions.last(), 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: &LedgerRevision, 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(
revisions: &[LedgerRevision],
live: LedgerLiveState,
) -> Option<LedgerBreakReport> {
let head = revisions.last()?;
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(revisions: &[LedgerRevision]) -> Option<LedgerBreakReport> {
revisions.windows(2).find_map(|pair| {
let [previous, revision] = pair else {
return None;
};
(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 first_break(revisions: &[LedgerRevision]) -> Option<LedgerBreakReport> {
let mut expected_seq: i64 = 1;
let mut prev: Option<&LedgerRevision> = 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) {
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(revision);
}
None
}
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<&LedgerRevision>,
) -> 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),
}
}
#[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 change the association to `on_delete = nullify`"
)]
HardDeleteCascade {
table: String,
record_id: i64,
},
#[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
}
}
#[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)
}
}
#[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(())
}
#[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
);
}
}