#![cfg(feature = "sqlite")]
#![allow(clippy::must_use_candidate, clippy::missing_const_for_fn)]
#![allow(clippy::float_cmp)]
use autumn_web::config::DatabaseConfig;
use autumn_web::current::with_actor;
use autumn_web::db::{RuntimeConnection, create_pool};
use autumn_web::hooks::Patch;
use autumn_web::ledger::{LedgerAsOf, LedgerBreak};
use autumn_web::reexports::{chrono, diesel, diesel_async};
use autumn_web::tenancy::with_tenant;
use autumn_web::version_history::VersionOp;
use chrono::{DateTime, Duration, Utc};
use diesel_async::pooled_connection::deadpool::Pool;
use diesel_async::{RunQueryDsl as _, SimpleAsyncConnection as _};
type SqlitePool = Pool<RuntimeConnection>;
mod schema {
autumn_web::reexports::diesel::table! {
lg_invoices (id) {
id -> Int8,
reference -> Text,
amount_cents -> Int8,
amount_rate -> Double,
metadata -> Text,
deleted_at -> Nullable<Timestamp>,
}
}
autumn_web::reexports::diesel::table! {
lg_effective_notes (id) {
id -> Int8,
body -> Text,
effective_at -> Timestamp,
deleted_at -> Nullable<Timestamp>,
}
}
autumn_web::reexports::diesel::table! {
lg_vault_notes (id) {
id -> Int8,
body -> Text,
secret -> Text,
deleted_at -> Nullable<Timestamp>,
}
}
autumn_web::reexports::diesel::table! {
lg_cascade_parents (id) {
id -> Int8,
name -> Text,
}
}
autumn_web::reexports::diesel::table! {
lg_cascade_children (id) {
id -> Int8,
parent_id -> Int8,
label -> Text,
deleted_at -> Nullable<Timestamp>,
}
}
autumn_web::reexports::diesel::table! {
lg_secret_notes (id) {
id -> Int8,
body -> Text,
internal_note -> Text,
deleted_at -> Nullable<Timestamp>,
}
}
autumn_web::reexports::diesel::table! {
lg_tenant_invoices (id) {
id -> Int8,
reference -> Text,
tenant_id -> Text,
deleted_at -> Nullable<Timestamp>,
}
}
}
use schema::{
lg_cascade_children, lg_cascade_parents, lg_effective_notes, lg_invoices, lg_secret_notes,
lg_tenant_invoices, lg_vault_notes,
};
#[autumn_web::model(table = "lg_invoices")]
pub struct LgInvoice {
#[id]
pub id: i64,
pub reference: String,
pub amount_cents: i64,
pub amount_rate: f64,
pub metadata: String,
#[default]
pub deleted_at: Option<chrono::NaiveDateTime>,
}
#[autumn_web::repository(LgInvoice, table = "lg_invoices", soft_delete, ledgered = true)]
pub trait LgInvoiceRepository {}
#[autumn_web::model(table = "lg_tenant_invoices")]
#[derive(PartialEq, Eq)]
pub struct LgTenantInvoice {
#[id]
pub id: i64,
pub reference: String,
#[default]
pub tenant_id: String,
#[default]
pub deleted_at: Option<chrono::NaiveDateTime>,
}
#[autumn_web::repository(
LgTenantInvoice,
table = "lg_tenant_invoices",
tenant_scoped,
soft_delete,
ledgered = true
)]
pub trait LgTenantInvoiceRepository {}
#[autumn_web::model(table = "lg_effective_notes")]
#[derive(PartialEq, Eq)]
pub struct LgEffectiveNote {
#[id]
pub id: i64,
pub body: String,
pub effective_at: chrono::NaiveDateTime,
#[default]
pub deleted_at: Option<chrono::NaiveDateTime>,
}
#[autumn_web::repository(
LgEffectiveNote,
table = "lg_effective_notes",
soft_delete,
ledgered(valid_time = "effective_at")
)]
pub trait LgEffectiveNoteRepository {}
#[autumn_web::model(table = "lg_cascade_children")]
pub struct LgCascadeChild {
#[id]
pub id: i64,
pub parent_id: i64,
pub label: String,
#[default]
pub deleted_at: Option<chrono::NaiveDateTime>,
}
#[autumn_web::repository(
LgCascadeChild,
table = "lg_cascade_children",
soft_delete,
ledgered = true
)]
pub trait LgCascadeChildRepository {}
#[autumn_web::model(table = "lg_cascade_parents")]
pub struct LgCascadeParent {
#[id]
pub id: i64,
pub name: String,
}
#[autumn_web::repository(
LgCascadeParent,
table = "lg_cascade_parents",
dependent(PgLgCascadeChildRepository, fk = "parent_id", on_delete = destroy)
)]
pub trait LgCascadeParentRepository {}
#[autumn_web::model(table = "lg_vault_notes")]
pub struct LgVaultNote {
#[id]
pub id: i64,
pub body: String,
#[encrypted]
pub secret: String,
#[default]
pub deleted_at: Option<chrono::NaiveDateTime>,
}
#[autumn_web::repository(LgVaultNote, table = "lg_vault_notes", soft_delete, ledgered = true)]
pub trait LgVaultNoteRepository {}
#[autumn_web::model(table = "lg_secret_notes")]
#[derive(PartialEq, Eq)]
pub struct LgSecretNote {
#[id]
pub id: i64,
pub body: String,
#[private]
pub internal_note: String,
#[default]
pub deleted_at: Option<chrono::NaiveDateTime>,
}
#[autumn_web::repository(LgSecretNote, table = "lg_secret_notes", soft_delete, ledgered = true)]
pub trait LgSecretNoteRepository {}
const LEDGER_UP: &str = include_str!(
"../version_history_migrations_sqlite/20260826000000_create_ledger_revisions/up.sql"
);
const LEDGER_HIGH_WATER_UP: &str = include_str!(
"../version_history_migrations_sqlite/20260901213107_create_ledger_high_water/up.sql"
);
const VERSION_HISTORY_UP: &str = include_str!(
"../version_history_migrations_sqlite/20260526000000_create_version_history/up.sql"
);
fn install_encryption_keys() {
use std::sync::OnceLock;
static INSTALLED: OnceLock<()> = OnceLock::new();
INSTALLED.get_or_init(|| {
const PRIMARY: &str = "1111111111111111111111111111111111111111111111111111111111111111";
const DETERMINISTIC: &str =
"3333333333333333333333333333333333333333333333333333333333333333";
let ring = autumn_web::encryption::KeyRing::from_master_hex(
PRIMARY,
&[],
Some(DETERMINISTIC),
b"ledger-suite-salt",
)
.expect("fixture key material derives");
autumn_web::encryption::install_key_ring(ring);
});
}
async fn boot_pool(db_name: &str) -> SqlitePool {
let pool = boot_pool_without_high_water(db_name).await;
{
let mut conn = pool.get().await.expect("checkout a sqlite connection");
conn.batch_execute(LEDGER_HIGH_WATER_UP)
.await
.expect("apply the high-water migration");
}
pool
}
async fn boot_pool_without_high_water(db_name: &str) -> SqlitePool {
install_encryption_keys();
let config = DatabaseConfig {
url: Some(format!("sqlite://file:{db_name}?mode=memory&cache=shared")),
primary_pool_size: Some(1),
..Default::default()
};
let pool: SqlitePool = create_pool(&config)
.expect("sqlite pool builds via build_sqlite_pool")
.expect("a url is configured");
{
let mut conn = pool.get().await.expect("checkout a sqlite connection");
for ddl in [
"CREATE TABLE lg_invoices (\
id INTEGER PRIMARY KEY AUTOINCREMENT, \
reference TEXT NOT NULL, \
amount_cents BIGINT NOT NULL, \
amount_rate DOUBLE PRECISION NOT NULL, \
metadata TEXT NOT NULL, \
deleted_at TIMESTAMP\
)",
"CREATE TABLE lg_tenant_invoices (\
id INTEGER PRIMARY KEY AUTOINCREMENT, \
reference TEXT NOT NULL, \
tenant_id TEXT NOT NULL, \
deleted_at TIMESTAMP\
)",
"CREATE TABLE lg_vault_notes (\
id INTEGER PRIMARY KEY AUTOINCREMENT, \
body TEXT NOT NULL, \
secret TEXT NOT NULL, \
deleted_at TIMESTAMP\
)",
"CREATE TABLE lg_cascade_parents (\
id INTEGER PRIMARY KEY AUTOINCREMENT, \
name TEXT NOT NULL\
)",
"CREATE TABLE lg_cascade_children (\
id INTEGER PRIMARY KEY AUTOINCREMENT, \
parent_id BIGINT NOT NULL REFERENCES lg_cascade_parents(id), \
label TEXT NOT NULL, \
deleted_at TIMESTAMP\
)",
"CREATE TABLE lg_secret_notes (\
id INTEGER PRIMARY KEY AUTOINCREMENT, \
body TEXT NOT NULL, \
internal_note TEXT NOT NULL, \
deleted_at TIMESTAMP\
)",
"CREATE TABLE lg_effective_notes (\
id INTEGER PRIMARY KEY AUTOINCREMENT, \
body TEXT NOT NULL, \
effective_at TIMESTAMP NOT NULL, \
deleted_at TIMESTAMP\
)",
VERSION_HISTORY_UP,
LEDGER_UP,
] {
conn.batch_execute(ddl)
.await
.unwrap_or_else(|err| panic!("apply DDL: {err}\n{ddl}"));
}
}
pool
}
fn assert_same_record<T: serde::Serialize>(left: &T, right: &T, context: &str) {
assert_eq!(
serde_json::to_string(left).expect("serialize left"),
serde_json::to_string(right).expect("serialize right"),
"{context}"
);
}
async fn resync_high_water_mark(pool: &SqlitePool, table: &str, record_id: i64) {
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"DELETE FROM _autumn_ledger_high_water WHERE table_name = ? AND record_id = ?",
)
.bind::<diesel::sql_types::Text, _>(table)
.bind::<diesel::sql_types::BigInt, _>(record_id)
.execute(&mut *conn)
.await
.expect("clear the mark");
diesel::sql_query(
"INSERT INTO _autumn_ledger_high_water \
(table_name, tenant_key, record_id, high_seq, head_hash, recorded_at) \
SELECT table_name, COALESCE(tenant_id, ''), record_id, seq, hash, recorded_at \
FROM _autumn_ledger_revisions \
WHERE table_name = ? AND record_id = ? ORDER BY seq DESC LIMIT 1",
)
.bind::<diesel::sql_types::Text, _>(table)
.bind::<diesel::sql_types::BigInt, _>(record_id)
.execute(&mut *conn)
.await
.expect("re-establish the mark over the surviving chain");
}
#[allow(
clippy::disallowed_methods,
reason = "test asserts against real recorded_at values"
)]
fn now() -> DateTime<Utc> {
Utc::now()
}
async fn tick() {
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
}
#[tokio::test]
#[allow(clippy::too_many_lines)]
async fn golden_as_of_reconstruction_matches_the_oracle_and_tampering_is_detected() {
let pool = boot_pool("lg_golden").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let mut oracle: Vec<(DateTime<Utc>, LgInvoice)> = Vec::new();
let created = with_actor("alice", async {
repo.save(&NewLgInvoice {
reference: "INV-1".to_string(),
amount_cents: 1000,
amount_rate: 1e16,
metadata: r#"{"b":1,"a":{"y":2,"x":3}}"#.to_string(),
})
.await
.expect("insert")
})
.await;
let id = created.id;
tick().await;
oracle.push((now(), created.clone()));
tick().await;
let updated_once = with_actor("alice", async {
repo.update(
id,
&UpdateLgInvoice {
amount_cents: Patch::Set(2000),
..Default::default()
},
)
.await
.expect("first update")
})
.await;
tick().await;
oracle.push((now(), updated_once.clone()));
tick().await;
let updated_twice = with_actor("bob", async {
repo.update(
id,
&UpdateLgInvoice {
reference: Patch::Set("INV-1-REV".to_string()),
amount_cents: Patch::Set(3000),
..Default::default()
},
)
.await
.expect("second update")
})
.await;
tick().await;
oracle.push((now(), updated_twice.clone()));
for (instant, expected) in &oracle {
let reconstructed = repo
.ledger_as_of(id, *instant)
.await
.expect("as-of read")
.expect("the record existed at this instant");
assert_same_record(
&reconstructed,
expected,
&format!("as-of state at {instant} must equal what a live query returned then"),
);
}
let before_creation = oracle[0].0 - Duration::seconds(60);
assert!(
repo.ledger_as_of(id, before_creation)
.await
.expect("as-of read")
.is_none(),
"as-of before creation must report that the record did not exist"
);
let report = repo.ledger_verify(id).await.expect("verify");
assert!(
report.is_intact(),
"an untouched chain must verify: {report:?}"
);
assert_eq!(report.revisions_checked, 3, "insert + two updates");
let head = repo
.ledger_head(id)
.await
.expect("head read")
.expect("a written record has a head");
assert_eq!(head.seq, 3);
assert_eq!(report.head_hash.as_deref(), Some(head.hash.as_str()));
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"UPDATE _autumn_ledger_revisions \
SET snapshot = json_set(snapshot, '$.amount_cents', 999999) \
WHERE table_name = 'lg_invoices' AND record_id = ? AND seq = 2",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("tamper with revision 2");
}
let tampered = repo
.ledger_verify(id)
.await
.expect("verify after tampering");
let broken = tampered
.broken
.expect("an out-of-band mutation must be detected");
assert_eq!(broken.kind, LedgerBreak::HashMismatch);
assert_eq!(broken.seq, 2, "verify must report the first broken link");
assert!(
tampered.head_hash.is_none(),
"a broken chain exports no head"
);
}
#[tokio::test]
async fn every_write_appends_a_revision_with_both_time_axes() {
let pool = boot_pool("lg_writes").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool);
let id = with_actor("alice", async {
let created = repo
.save(&NewLgInvoice {
reference: "INV-2".to_string(),
amount_cents: 500,
amount_rate: 1e16,
metadata: r#"{"b":1,"a":{"y":2,"x":3}}"#.to_string(),
})
.await
.expect("insert");
repo.update(
created.id,
&UpdateLgInvoice {
amount_cents: Patch::Set(750),
..Default::default()
},
)
.await
.expect("update");
repo.delete_by_id(created.id).await.expect("soft delete");
created.id
})
.await;
let revisions = repo.ledger_revisions(id).await.expect("revisions");
assert_eq!(
revisions.iter().map(|r| r.op).collect::<Vec<_>>(),
vec![VersionOp::Insert, VersionOp::Update, VersionOp::Delete],
"insert, update and soft-delete each append exactly one revision"
);
assert_eq!(
revisions.iter().map(|r| r.seq).collect::<Vec<_>>(),
vec![1, 2, 3],
"sequence numbers are contiguous from 1"
);
for revision in &revisions {
assert_eq!(revision.actor, "alice", "the ambient actor is attributed");
assert_eq!(revision.table_name, "lg_invoices");
assert_eq!(revision.record_id, id);
assert!(
revision.snapshot.get("reference").is_some(),
"a revision carries a FULL snapshot, not just the changed columns"
);
assert!(
revision.snapshot.get("amount_cents").is_some(),
"a revision carries a FULL snapshot, not just the changed columns"
);
assert_eq!(
revision.valid_from, revision.recorded_at,
"valid time defaults to transaction time"
);
}
assert_eq!(
revisions[0].prev_hash, None,
"the chain head opens with no parent"
);
assert_eq!(
revisions[1].prev_hash.as_deref(),
Some(revisions[0].hash.as_str())
);
assert_eq!(
revisions[2].prev_hash.as_deref(),
Some(revisions[1].hash.as_str())
);
let after_delete = repo
.ledger_as_of_at(id, LedgerAsOf::default())
.await
.expect("as-of read")
.expect("a soft-deleted record still has state");
assert!(
after_delete.deleted_at.is_some(),
"the delete revision snapshots the soft-deleted row"
);
assert!(repo.ledger_verify(id).await.expect("verify").is_intact());
}
#[tokio::test]
async fn diff_reports_the_field_level_delta_between_two_instants() {
let pool = boot_pool("lg_diff").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool);
let created = repo
.save(&NewLgInvoice {
reference: "INV-3".to_string(),
amount_cents: 100,
amount_rate: 1e16,
metadata: r#"{"b":1,"a":{"y":2,"x":3}}"#.to_string(),
})
.await
.expect("insert");
let id = created.id;
tick().await;
let after_create = now();
tick().await;
repo.update(
id,
&UpdateLgInvoice {
amount_cents: Patch::Set(250),
..Default::default()
},
)
.await
.expect("update");
tick().await;
let after_update = now();
let diff = repo
.ledger_diff(id, after_create, after_update)
.await
.expect("diff");
assert_eq!(diff.from_seq, Some(1));
assert_eq!(diff.to_seq, Some(2));
assert_eq!(
diff.changes.len(),
1,
"only amount_cents changed: {:?}",
diff.changes
);
let change = &diff.changes[0];
assert_eq!(change.column, "amount_cents");
assert_eq!(
change.before.as_ref().and_then(serde_json::Value::as_i64),
Some(100)
);
assert_eq!(
change.after.as_ref().and_then(serde_json::Value::as_i64),
Some(250)
);
let quiet = repo
.ledger_diff(id, after_update, after_update + Duration::seconds(5))
.await
.expect("diff");
assert!(quiet.is_empty(), "{quiet:?}");
}
#[tokio::test]
async fn verify_detects_deletion_of_a_stored_revision() {
let pool = boot_pool("lg_tamper_delete").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
assert!(repo.ledger_verify(id).await.expect("verify").is_intact());
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"DELETE FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_invoices' AND record_id = ? AND seq = 2",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("delete revision 2");
}
let broken = repo
.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("a deleted revision must be detected");
assert_eq!(broken.kind, LedgerBreak::MissingRevision);
assert_eq!(broken.seq, 2);
}
#[tokio::test]
async fn verify_detects_insertion_of_a_forged_revision() {
let pool = boot_pool("lg_tamper_insert").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"INSERT INTO _autumn_ledger_revisions \
(table_name, tenant_id, record_id, seq, op, actor, request_id, snapshot, \
valid_from, recorded_at, prev_hash, hash) \
SELECT table_name, tenant_id, record_id, 4, 'update', 'mallory', NULL, \
json_set(snapshot, '$.amount_cents', 1), valid_from, recorded_at, \
hash, 'f0f0f0f0' \
FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_invoices' AND record_id = ? AND seq = 3",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("insert a forged revision");
}
let broken = repo
.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("an inserted revision must be detected");
assert_eq!(broken.kind, LedgerBreak::HashMismatch);
assert_eq!(broken.seq, 4);
}
#[tokio::test]
async fn verify_has_no_false_positives_on_an_untouched_chain() {
let pool = boot_pool("lg_no_false_positives").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool);
let mut ids = Vec::new();
for n in 0..5 {
let created = repo
.save(&NewLgInvoice {
reference: format!("INV-{n}"),
amount_cents: i64::from(n),
amount_rate: 1e16,
metadata: r#"{"b":1,"a":{"y":2,"x":3}}"#.to_string(),
})
.await
.expect("insert");
for step in 1..=3 {
repo.update(
created.id,
&UpdateLgInvoice {
amount_cents: Patch::Set(i64::from(n) * 100 + step),
..Default::default()
},
)
.await
.expect("update");
}
ids.push(created.id);
}
for _ in 0..3 {
for id in &ids {
let report = repo.ledger_verify(*id).await.expect("verify");
assert!(report.is_intact(), "record {id} must verify: {report:?}");
assert_eq!(report.revisions_checked, 4);
}
}
}
#[tokio::test]
async fn duplicate_sequence_numbers_are_rejected_by_the_database() {
let pool = boot_pool("lg_unique_chain").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
let mut conn = pool.get().await.expect("conn");
let result = diesel::sql_query(
"INSERT INTO _autumn_ledger_revisions \
(table_name, tenant_id, record_id, seq, op, actor, snapshot, valid_from, \
recorded_at, prev_hash, hash) \
SELECT table_name, tenant_id, record_id, seq, op, actor, snapshot, valid_from, \
recorded_at, prev_hash, 'duplicate' \
FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_invoices' AND record_id = ? AND seq = 2",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await;
assert!(
result.is_err(),
"the (table, tenant, record, seq) unique index must refuse a forked chain"
);
}
#[tokio::test]
async fn the_high_water_mark_upsert_refuses_to_move_backwards() {
let pool = boot_pool("lg_mark_monotonic").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"INSERT INTO _autumn_ledger_high_water \
(table_name, tenant_key, record_id, high_seq, head_hash, recorded_at) \
VALUES ('lg_invoices', '', ?, 1, 'stale', ?) \
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",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.bind::<diesel::sql_types::TimestamptzSqlite, _>(now())
.execute(&mut *conn)
.await
.expect("the upsert itself must succeed — it simply must not lower the mark");
}
let mark = repo
.ledger_high_water(id)
.await
.expect("mark")
.expect("mark");
assert_eq!(
mark.seq, 3,
"a stale writer must not be able to lower the mark"
);
assert_ne!(mark.hash, "stale");
assert!(repo.ledger_verify(id).await.expect("verify").is_intact());
}
#[tokio::test]
async fn a_failed_mark_upsert_rolls_the_whole_append_back() {
let pool = boot_pool("lg_mark_rollback").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
{
let mut conn = pool.get().await.expect("conn");
conn.batch_execute(
"CREATE TRIGGER lg_freeze_mark BEFORE UPDATE ON _autumn_ledger_high_water \
BEGIN SELECT RAISE(ABORT, 'mark frozen'); END",
)
.await
.expect("freeze the mark");
}
let result = repo
.update(
id,
&UpdateLgInvoice {
amount_cents: Patch::Set(99),
..Default::default()
},
)
.await;
assert!(
result.is_err(),
"an append whose mark cannot be raised must fail rather than proceed"
);
let revisions = repo.ledger_revisions(id).await.expect("revisions");
assert_eq!(
revisions.len(),
3,
"the revision must roll back with the mark, not survive it: {:?}",
revisions.iter().map(|r| r.seq).collect::<Vec<_>>()
);
let mark = repo
.ledger_high_water(id)
.await
.expect("mark")
.expect("mark");
assert_eq!(mark.seq, 3);
assert!(repo.ledger_verify(id).await.expect("verify").is_intact());
{
let mut conn = pool.get().await.expect("conn");
conn.batch_execute("DROP TRIGGER lg_freeze_mark")
.await
.expect("unfreeze the mark");
}
repo.update(
id,
&UpdateLgInvoice {
amount_cents: Patch::Set(99),
..Default::default()
},
)
.await
.expect("update");
let mark = repo
.ledger_high_water(id)
.await
.expect("mark")
.expect("mark");
assert_eq!(
mark.seq, 4,
"the rolled-back attempt burned no sequence number"
);
assert!(repo.ledger_verify(id).await.expect("verify").is_intact());
}
#[tokio::test]
async fn ledger_reads_fail_closed_across_tenants() {
let pool = boot_pool("lg_tenant").await;
let repo = PgLgTenantInvoiceRepository::with_pool_untracked(pool);
let id = with_tenant("tenant-a".to_string(), async {
repo.save(&NewLgTenantInvoice {
reference: "A-1".to_string(),
})
.await
.expect("insert as tenant-a")
.id
})
.await;
let mine = with_tenant("tenant-a".to_string(), async {
repo.ledger_revisions(id).await.expect("read as tenant-a")
})
.await;
assert_eq!(mine.len(), 1);
assert_eq!(mine[0].tenant_id.as_deref(), Some("tenant-a"));
let theirs = with_tenant("tenant-b".to_string(), async {
repo.ledger_revisions(id).await.expect("read as tenant-b")
})
.await;
assert!(
theirs.is_empty(),
"a ledgered read must not leak another tenant's revisions: {theirs:?}"
);
let reconstructed = with_tenant("tenant-b".to_string(), async {
repo.ledger_as_of_at(id, LedgerAsOf::default())
.await
.expect("as-of as tenant-b")
})
.await;
assert!(
reconstructed.is_none(),
"as-of must fail closed across tenants"
);
}
#[tokio::test]
async fn restore_records_a_revision_and_keeps_as_of_true() {
let pool = boot_pool("lg_restore").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool);
let id = write_three_revisions(&repo).await;
repo.delete_by_id(id).await.expect("soft delete");
repo.restore(id).await.expect("restore");
let revisions = repo.ledger_revisions(id).await.expect("revisions");
assert_eq!(
revisions.iter().map(|r| r.op).collect::<Vec<_>>(),
vec![
VersionOp::Insert,
VersionOp::Update,
VersionOp::Update,
VersionOp::Delete,
VersionOp::Update,
],
"restore appends an update revision"
);
let reconstructed = repo
.ledger_as_of_at(id, LedgerAsOf::default())
.await
.expect("as-of")
.expect("state exists");
assert!(
reconstructed.deleted_at.is_none(),
"the restore revision must snapshot the undeleted row"
);
let live = repo
.find_by_id(id)
.await
.expect("live read")
.expect("live row");
assert_same_record(&reconstructed, &live, "as-of(now) must equal the live row");
assert!(repo.ledger_verify(id).await.expect("verify").is_intact());
}
#[tokio::test]
async fn verify_detects_a_truncated_tail() {
let pool = boot_pool("lg_truncate").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"DELETE FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_invoices' AND record_id = ? AND seq = 3",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("lop off the newest revision");
}
let broken = repo
.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("a truncated tail must be detected");
assert_eq!(
broken.kind,
LedgerBreak::MissingRevision,
"the mark outlives the row it names, so the break is the absent revision \
rather than merely a live row that disagrees"
);
assert_eq!(
broken.seq, 3,
"reported at the sequence number that is gone"
);
resync_high_water_mark(&pool, "lg_invoices", id).await;
let broken = repo
.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("the live row must still expose the truncation");
assert_eq!(broken.kind, LedgerBreak::LiveStateMismatch);
assert_eq!(broken.seq, 2, "reported at the surviving head");
}
#[tokio::test]
async fn a_post_truncation_append_leaves_a_gap_verify_reports() {
let pool = boot_pool("lg_truncate_then_append").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"DELETE FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_invoices' AND record_id = ? AND seq = 3",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("lop off the newest revision");
}
repo.update(
id,
&UpdateLgInvoice {
amount_cents: Patch::Set(4_242),
..Default::default()
},
)
.await
.expect("an ordinary write lands after the truncation");
let revisions = repo.ledger_revisions(id).await.expect("revisions");
assert_eq!(
revisions.iter().map(|r| r.seq).collect::<Vec<_>>(),
vec![1, 2, 4],
"the append must allocate past the deleted sequence number, not re-use it"
);
let broken = repo
.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("the gap the append left must be reported");
assert_eq!(broken.kind, LedgerBreak::MissingRevision);
assert_eq!(broken.seq, 3);
repo.update(
id,
&UpdateLgInvoice {
amount_cents: Patch::Set(4_343),
..Default::default()
},
)
.await
.expect("more ordinary traffic");
let broken = repo
.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("the gap must not heal");
assert_eq!(broken.kind, LedgerBreak::MissingRevision);
assert_eq!(broken.seq, 3);
}
#[tokio::test]
async fn ledger_pin_returns_the_head_and_the_mark_from_one_snapshot() {
let pool = boot_pool("lg_pin").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let empty = repo.ledger_pin(9_999).await.expect("pin");
assert!(empty.head.is_none() && empty.high_water.is_none());
let id = write_three_revisions(&repo).await;
let pin = repo.ledger_pin(id).await.expect("pin");
let head = pin.head.expect("a written record has a head");
let mark = pin.high_water.expect("and a mark");
assert_eq!(head.seq, 3);
assert_eq!(
mark.seq, head.seq,
"the pair agrees because it is one snapshot"
);
assert_eq!(mark.hash, head.hash);
assert_eq!(mark.recorded_at, head.recorded_at);
assert_eq!(repo.ledger_head(id).await.expect("head"), Some(head));
assert_eq!(repo.ledger_high_water(id).await.expect("mark"), Some(mark));
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"DELETE FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_invoices' AND record_id = ? AND seq = 3",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("lop off the newest revision");
}
let pin = repo.ledger_pin(id).await.expect("pin");
assert_eq!(pin.head.expect("head").seq, 2);
assert_eq!(
pin.high_water.expect("mark").seq,
3,
"the mark outlives the revision it names, and the pair shows it"
);
}
#[tokio::test]
async fn tampering_with_the_high_water_mark_is_itself_detected() {
let pool = boot_pool("lg_mark_tamper").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
let mark = repo
.ledger_high_water(id)
.await
.expect("mark")
.expect("a written record has a mark");
assert_eq!(mark.seq, 3, "the mark tracks every append");
let head = repo
.ledger_head(id)
.await
.expect("head")
.expect("a written record has a head");
assert_eq!(mark.hash, head.hash, "the mark names the head revision");
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"UPDATE _autumn_ledger_high_water SET high_seq = 1 \
WHERE table_name = 'lg_invoices' AND record_id = ?",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("roll the mark back");
}
let broken = repo
.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("a rolled-back mark must be detected");
assert_eq!(broken.kind, LedgerBreak::HighWaterBehind);
resync_high_water_mark(&pool, "lg_invoices", id).await;
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"UPDATE _autumn_ledger_high_water SET head_hash = 'deadbeef' \
WHERE table_name = 'lg_invoices' AND record_id = ?",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("rewrite the mark's hash");
}
let broken = repo
.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("a rewritten mark must be detected");
assert_eq!(broken.kind, LedgerBreak::HighWaterMismatch);
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"DELETE FROM _autumn_ledger_high_water \
WHERE table_name = 'lg_invoices' AND record_id = ?",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("delete the mark");
}
let broken = repo
.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("a deleted mark must be detected");
assert_eq!(broken.kind, LedgerBreak::HighWaterMissing);
resync_high_water_mark(&pool, "lg_invoices", id).await;
assert!(repo.ledger_verify(id).await.expect("verify").is_intact());
}
#[tokio::test]
async fn a_wholly_erased_chain_is_reported_against_its_surviving_mark() {
let pool = boot_pool("lg_erased_marked").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"DELETE FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_invoices' AND record_id = ?",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("erase the whole chain");
}
let report = repo.ledger_verify(id).await.expect("verify");
assert_eq!(report.revisions_checked, 0);
let broken = report.broken.expect("an erased chain must be reported");
assert_eq!(broken.kind, LedgerBreak::MissingRevision);
assert_eq!(broken.seq, 1);
}
#[tokio::test]
async fn a_row_with_no_chain_is_not_reported_as_tampering() {
let pool = boot_pool("lg_erased").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"DELETE FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_invoices' AND record_id = ?",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("leave the record with no chain");
}
resync_high_water_mark(&pool, "lg_invoices", id).await;
let report = repo.ledger_verify(id).await.expect("verify");
assert!(report.is_intact(), "{report:?}");
assert_eq!(
report.revisions_checked, 0,
"the empty chain is what the caller inspects, not an accusation"
);
repo.update(
id,
&UpdateLgInvoice {
amount_cents: Patch::Set(42),
..Default::default()
},
)
.await
.expect("update");
let report = repo.ledger_verify(id).await.expect("verify");
assert!(report.is_intact(), "{report:?}");
assert_eq!(report.revisions_checked, 1);
}
#[tokio::test]
async fn a_pinned_head_detects_a_correctly_hashed_forgery() {
let pool = boot_pool("lg_pinned_head").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
let pinned = repo
.ledger_head(id)
.await
.expect("head")
.expect("a written record has a head");
let head_revision = repo
.ledger_revisions(id)
.await
.expect("revisions")
.pop()
.expect("head revision");
let mut snapshot = head_revision.snapshot.clone();
snapshot["amount_cents"] = serde_json::json!(999_999);
let recorded_at = autumn_web::ledger::truncate_to_micros(now());
let forged_hash = autumn_web::ledger::revision_hash(&autumn_web::ledger::RevisionHashInput {
prev_hash: Some(head_revision.hash.as_str()),
table_name: "lg_invoices",
tenant_id: None,
record_id: id,
seq: head_revision.seq + 1,
op: VersionOp::Update,
actor: "mallory",
request_id: None,
snapshot: &snapshot,
valid_from: recorded_at,
recorded_at,
});
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"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 ('lg_invoices', NULL, ?, ?, 'update', 'mallory', NULL, ?, ?, ?, ?, ?)",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.bind::<diesel::sql_types::BigInt, _>(head_revision.seq + 1)
.bind::<diesel::sql_types::Text, _>(autumn_web::ledger::canonical_json(&snapshot))
.bind::<diesel::sql_types::TimestamptzSqlite, _>(recorded_at)
.bind::<diesel::sql_types::TimestamptzSqlite, _>(recorded_at)
.bind::<diesel::sql_types::Text, _>(head_revision.hash.clone())
.bind::<diesel::sql_types::Text, _>(forged_hash)
.execute(&mut *conn)
.await
.expect("append a well-formed forgery");
diesel::sql_query("UPDATE lg_invoices SET amount_cents = 999999 WHERE id = ?")
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("rewrite the live row to match");
}
let broken = repo
.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("an append the mark did not authorise must be detected");
assert_eq!(broken.kind, LedgerBreak::HighWaterBehind);
assert_eq!(broken.seq, head_revision.seq + 1);
resync_high_water_mark(&pool, "lg_invoices", id).await;
assert!(
repo.ledger_verify(id).await.expect("verify").is_intact(),
"a correctly-hashed, live-consistent, mark-consistent append is invisible \
from inside the database"
);
let now_head = repo.ledger_head(id).await.expect("head").expect("head");
assert_ne!(
now_head.hash, pinned.hash,
"a head pinned outside the database must disagree after a forgery"
);
assert_eq!(now_head.seq, pinned.seq + 1);
}
#[tokio::test]
async fn transaction_time_never_moves_backwards_along_a_chain() {
let pool = boot_pool("lg_txn_time_monotonic").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
let recorded: Vec<DateTime<Utc>> = repo
.ledger_revisions(id)
.await
.expect("revisions")
.iter()
.map(|r| r.recorded_at)
.collect();
assert!(
recorded.windows(2).all(|w| w[0] <= w[1]),
"transaction time must be non-decreasing along a chain: {recorded:?}"
);
let ahead = autumn_web::ledger::truncate_to_micros(now() + Duration::minutes(30));
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"DELETE FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_invoices' AND record_id = ? AND seq = 3",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("lop off the newest revision");
diesel::sql_query(
"UPDATE _autumn_ledger_high_water SET recorded_at = ? \
WHERE table_name = 'lg_invoices' AND record_id = ?",
)
.bind::<diesel::sql_types::TimestamptzSqlite, _>(ahead)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("move the record's transaction-time floor forward");
}
repo.update(
id,
&UpdateLgInvoice {
amount_cents: Patch::Set(7),
..Default::default()
},
)
.await
.expect("update");
let revisions = repo.ledger_revisions(id).await.expect("revisions");
let head = revisions.last().expect("head");
assert!(
head.recorded_at >= ahead,
"a write behind the chain's floor must be clamped up to it, not written \
behind it: {} < {ahead}",
head.recorded_at
);
let recorded: Vec<DateTime<Utc>> = revisions.iter().map(|r| r.recorded_at).collect();
assert!(
recorded.windows(2).all(|w| w[0] <= w[1]),
"still non-decreasing across the gap the deleted revision left: {recorded:?}"
);
assert_eq!(
head.compute_hash(),
head.hash,
"the stored transaction time must be the value that was hashed"
);
}
#[tokio::test]
async fn a_transaction_time_floor_far_ahead_of_the_database_refuses_the_write() {
let pool = boot_pool("lg_txn_time_ratchet").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
let far = autumn_web::ledger::truncate_to_micros(now() + Duration::days(365 * 900));
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"DELETE FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_invoices' AND record_id = ? AND seq = 3",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("lop off the newest revision");
diesel::sql_query(
"UPDATE _autumn_ledger_high_water SET recorded_at = ? \
WHERE table_name = 'lg_invoices' AND record_id = ?",
)
.bind::<diesel::sql_types::TimestamptzSqlite, _>(far)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("poison the record's transaction-time floor");
}
let result = repo
.update(
id,
&UpdateLgInvoice {
amount_cents: Patch::Set(7),
..Default::default()
},
)
.await;
assert!(
result.is_err(),
"a poisoned floor must refuse the write rather than hash a far-future \
instant into the chain"
);
let revisions = repo.ledger_revisions(id).await.expect("revisions");
assert_eq!(revisions.len(), 2);
assert!(revisions.iter().all(|r| r.recorded_at < far));
let broken = repo
.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("the truncation is still reported");
assert_eq!(broken.kind, LedgerBreak::MissingRevision);
}
#[tokio::test]
async fn a_rewritten_mark_instant_is_refused_like_any_other_mark_rewrite() {
let pool = boot_pool("lg_txn_time_mark_rewrite").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
let ahead = autumn_web::ledger::truncate_to_micros(now() + Duration::minutes(45));
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"UPDATE _autumn_ledger_high_water SET recorded_at = ? \
WHERE table_name = 'lg_invoices' AND record_id = ?",
)
.bind::<diesel::sql_types::TimestamptzSqlite, _>(ahead)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("rewrite the mark's instant, leaving seq and hash alone");
}
assert_eq!(
repo.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("reported")
.kind,
LedgerBreak::HighWaterMismatch
);
assert!(
repo.update(
id,
&UpdateLgInvoice {
amount_cents: Patch::Set(11),
..Default::default()
},
)
.await
.is_err(),
"the append must refuse rather than overwrite the mark and launder the \
accusation away"
);
assert_eq!(repo.ledger_revisions(id).await.expect("revisions").len(), 3);
assert_eq!(
repo.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("still reported")
.kind,
LedgerBreak::HighWaterMismatch
);
}
#[tokio::test]
async fn a_mark_behind_the_head_cannot_steer_the_next_revisions_time() {
let pool = boot_pool("lg_txn_time_mark_ignored").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
let ahead = autumn_web::ledger::truncate_to_micros(now() + Duration::minutes(45));
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"UPDATE _autumn_ledger_high_water SET high_seq = 2, \
head_hash = (SELECT hash FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_invoices' AND record_id = ? AND seq = 2), \
recorded_at = ? \
WHERE table_name = 'lg_invoices' AND record_id = ?",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.bind::<diesel::sql_types::TimestamptzSqlite, _>(ahead)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("leave the mark behind, carrying a bogus instant");
}
repo.update(
id,
&UpdateLgInvoice {
amount_cents: Patch::Set(11),
..Default::default()
},
)
.await
.expect("a mark behind the head must not block writes");
let head = repo
.ledger_revisions(id)
.await
.expect("revisions")
.pop()
.expect("head");
assert!(
head.recorded_at < ahead,
"the mark's unhashed instant must not become the floor while the head \
revision that carries a hashed one is still there: {} >= {ahead}",
head.recorded_at
);
assert!(repo.ledger_verify(id).await.expect("verify").is_intact());
}
#[tokio::test]
async fn an_as_of_query_never_returns_a_revision_recorded_after_the_instant() {
let pool = boot_pool("lg_txn_time_as_of").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let created = repo
.save(&NewLgInvoice {
reference: "INV-AS-OF".to_string(),
amount_cents: 1,
amount_rate: 1.5,
metadata: "{}".to_string(),
})
.await
.expect("insert");
for step in 2..=4 {
tick().await;
repo.update(
created.id,
&UpdateLgInvoice {
amount_cents: Patch::Set(step),
..Default::default()
},
)
.await
.expect("update");
}
let revisions = repo.ledger_revisions(created.id).await.expect("revisions");
assert_eq!(revisions.len(), 4);
assert!(
revisions
.windows(2)
.all(|w| w[0].recorded_at < w[1].recorded_at),
"the writes are spaced, so each revision has its own instant"
);
for (index, revision) in revisions.iter().enumerate() {
let at_instant = repo
.ledger_as_of(created.id, revision.recorded_at)
.await
.expect("as-of")
.expect("the record existed");
assert_eq!(
serde_json::json!(at_instant.amount_cents),
revision.snapshot["amount_cents"],
"as-of at revision {}'s own instant must return revision {}",
revision.seq,
revision.seq,
);
let just_before = revision.recorded_at - Duration::microseconds(1);
let earlier = repo
.ledger_as_of(created.id, just_before)
.await
.expect("as-of");
match index.checked_sub(1).and_then(|prev| revisions.get(prev)) {
None => assert!(
earlier.is_none(),
"before the insert the record did not exist yet"
),
Some(previous) => {
let earlier = earlier.expect("the record existed");
assert_eq!(
serde_json::json!(earlier.amount_cents),
previous.snapshot["amount_cents"],
"as-of just before revision {} must return revision {}, never a \
revision recorded after the instant asked about",
revision.seq,
previous.seq,
);
}
}
}
}
#[tokio::test]
async fn verify_detects_a_transaction_time_that_moves_backwards() {
let pool = boot_pool("lg_txn_time_regression").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
let revisions = repo.ledger_revisions(id).await.expect("revisions");
let head = revisions.last().expect("head");
let backdated = revisions[1].recorded_at - Duration::seconds(1);
let rehashed = autumn_web::ledger::revision_hash(&autumn_web::ledger::RevisionHashInput {
prev_hash: head.prev_hash.as_deref(),
table_name: "lg_invoices",
tenant_id: None,
record_id: id,
seq: head.seq,
op: head.op,
actor: head.actor.as_str(),
request_id: head.request_id.as_deref(),
snapshot: &head.snapshot,
valid_from: head.valid_from,
recorded_at: backdated,
});
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"UPDATE _autumn_ledger_revisions SET recorded_at = ?, hash = ? \
WHERE table_name = 'lg_invoices' AND record_id = ? AND seq = ?",
)
.bind::<diesel::sql_types::TimestamptzSqlite, _>(backdated)
.bind::<diesel::sql_types::Text, _>(rehashed)
.bind::<diesel::sql_types::BigInt, _>(id)
.bind::<diesel::sql_types::BigInt, _>(head.seq)
.execute(&mut *conn)
.await
.expect("back-date the newest revision");
}
resync_high_water_mark(&pool, "lg_invoices", id).await;
let broken = repo
.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("a backwards transaction time must be detected");
assert_eq!(broken.kind, LedgerBreak::RecordedAtRegression);
assert_eq!(broken.seq, head.seq);
}
#[tokio::test]
async fn an_append_refuses_to_re_create_a_deleted_high_water_mark() {
let pool = boot_pool("lg_mark_recreate").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
{
let mut conn = pool.get().await.expect("conn");
for sql in [
"DELETE FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_invoices' AND record_id = ? AND seq = 3",
"DELETE FROM _autumn_ledger_high_water \
WHERE table_name = 'lg_invoices' AND record_id = ?",
] {
diesel::sql_query(sql)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("cover both tables");
}
}
let result = repo
.update(
id,
&UpdateLgInvoice {
amount_cents: Patch::Set(1),
..Default::default()
},
)
.await;
assert!(
result.is_err(),
"an append over a chain whose mark is gone must refuse rather than \
re-create the mark and launder the truncation away"
);
let revisions = repo.ledger_revisions(id).await.expect("revisions");
assert_eq!(revisions.len(), 2);
let broken = repo
.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("the missing mark is still reported");
assert_eq!(broken.kind, LedgerBreak::HighWaterMissing);
}
#[tokio::test]
async fn an_append_refuses_when_the_mark_and_the_head_disagree() {
let pool = boot_pool("lg_mark_disagree").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"UPDATE _autumn_ledger_high_water SET head_hash = 'deadbeef' \
WHERE table_name = 'lg_invoices' AND record_id = ?",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("rewrite the mark");
}
let result = repo
.update(
id,
&UpdateLgInvoice {
amount_cents: Patch::Set(1),
..Default::default()
},
)
.await;
assert!(result.is_err(), "the disagreement must stop the write");
assert_eq!(
repo.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("still reported")
.kind,
LedgerBreak::HighWaterMismatch
);
}
#[tokio::test]
async fn a_write_after_a_wholly_erased_chain_leaves_permanent_evidence() {
let pool = boot_pool("lg_erased_then_write").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"DELETE FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_invoices' AND record_id = ?",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("erase the whole chain");
}
repo.update(
id,
&UpdateLgInvoice {
amount_cents: Patch::Set(1),
..Default::default()
},
)
.await
.expect("ordinary traffic continues");
let revisions = repo.ledger_revisions(id).await.expect("revisions");
assert_eq!(
revisions.iter().map(|r| r.seq).collect::<Vec<_>>(),
vec![4],
"the append allocates above the surviving mark, not from 1"
);
let broken = repo
.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("the erasure must still be reported after the write");
assert_eq!(broken.kind, LedgerBreak::MissingRevision);
assert_eq!(broken.seq, 1);
}
#[tokio::test]
async fn a_non_positive_high_water_mark_is_refused_rather_than_overwritten() {
let pool = boot_pool("lg_mark_non_positive").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
{
let mut conn = pool.get().await.expect("conn");
let refused = diesel::sql_query(
"INSERT INTO _autumn_ledger_high_water \
(table_name, tenant_key, record_id, high_seq, head_hash, recorded_at) \
VALUES ('lg_invoices', '', 424242, 0, 'planted', '2026-01-01T00:00:00Z')",
)
.execute(&mut *conn)
.await;
assert!(
refused.is_err(),
"CHECK (high_seq >= 1) must make a non-positive mark unrepresentable"
);
}
{
let mut conn = pool.get().await.expect("conn");
conn.batch_execute(
"ALTER TABLE _autumn_ledger_high_water RENAME TO _hw_old; \
CREATE TABLE _autumn_ledger_high_water ( \
table_name TEXT NOT NULL, tenant_key TEXT NOT NULL, \
record_id BIGINT NOT NULL, high_seq BIGINT NOT NULL, \
head_hash TEXT NOT NULL, recorded_at TEXT NOT NULL, \
PRIMARY KEY (table_name, tenant_key, record_id)); \
INSERT INTO _autumn_ledger_high_water SELECT * FROM _hw_old; \
DROP TABLE _hw_old;",
)
.await
.expect("rebuild the mark table without its CHECK");
diesel::sql_query(
"UPDATE _autumn_ledger_high_water SET high_seq = 0 \
WHERE table_name = 'lg_invoices' AND record_id = ?",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("plant a non-positive mark");
}
assert!(
repo.ledger_verify(id)
.await
.expect("verify")
.broken
.is_some(),
"a mark below the chain head is reported"
);
assert!(
repo.update(
id,
&UpdateLgInvoice {
amount_cents: Patch::Set(1),
..Default::default()
},
)
.await
.is_err(),
"the append must refuse rather than allocate past the mark and raise it"
);
assert_eq!(
repo.ledger_revisions(id).await.expect("revisions").len(),
3,
"nothing was written"
);
assert!(
repo.ledger_verify(id)
.await
.expect("verify")
.broken
.is_some(),
"and the accusation survives"
);
}
#[tokio::test]
async fn a_mark_behind_the_head_is_reported_but_still_writable() {
let pool = boot_pool("lg_mark_behind").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"UPDATE _autumn_ledger_high_water SET high_seq = 2, \
head_hash = (SELECT hash FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_invoices' AND record_id = ? AND seq = 2), \
recorded_at = (SELECT recorded_at FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_invoices' AND record_id = ? AND seq = 2) \
WHERE table_name = 'lg_invoices' AND record_id = ?",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.bind::<diesel::sql_types::BigInt, _>(id)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("leave the mark one behind, as an old writer would");
}
assert_eq!(
repo.ledger_verify(id)
.await
.expect("verify")
.broken
.expect("reported")
.kind,
LedgerBreak::HighWaterBehind
);
repo.update(
id,
&UpdateLgInvoice {
amount_cents: Patch::Set(1),
..Default::default()
},
)
.await
.expect("a mark behind the head must not block writes");
assert!(
repo.ledger_verify(id).await.expect("verify").is_intact(),
"the record's next write re-establishes the mark above both"
);
}
#[tokio::test]
async fn a_truncation_covered_up_in_both_tables_is_only_caught_by_a_pinned_head() {
let pool = boot_pool("lg_residual").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let id = write_three_revisions(&repo).await;
let pinned = repo
.ledger_head(id)
.await
.expect("head")
.expect("head")
.hash;
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"DELETE FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_invoices' AND record_id = ? AND seq = 3",
)
.bind::<diesel::sql_types::BigInt, _>(id)
.execute(&mut *conn)
.await
.expect("delete the newest revision");
}
resync_high_water_mark(&pool, "lg_invoices", id).await;
repo.update(
id,
&UpdateLgInvoice {
amount_cents: Patch::Set(1),
..Default::default()
},
)
.await
.expect("ordinary traffic continues");
assert!(
repo.ledger_verify(id).await.expect("verify").is_intact(),
"documented residual: consistent tampering across both tables is not \
visible from inside the database"
);
let now_head = repo.ledger_head(id).await.expect("head").expect("head");
assert_ne!(
now_head.hash, pinned,
"a head pinned outside the database is what still disagrees"
);
}
#[tokio::test]
async fn the_migration_backfills_a_mark_for_every_chain_that_already_exists() {
let pool = boot_pool("lg_backfill").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool.clone());
let mut ids = Vec::new();
for n in 0..2 {
let created = repo
.save(&NewLgInvoice {
reference: format!("PRE-{n}"),
amount_cents: 1,
amount_rate: 1.0,
metadata: "{}".to_string(),
})
.await
.expect("insert");
for step in 0..=n {
repo.update(
created.id,
&UpdateLgInvoice {
amount_cents: Patch::Set(i64::from(step) + 2),
..Default::default()
},
)
.await
.expect("update");
}
ids.push(created.id);
}
{
let mut conn = pool.get().await.expect("conn");
conn.batch_execute("DELETE FROM _autumn_ledger_high_water")
.await
.expect("un-mark every chain");
conn.batch_execute(LEDGER_HIGH_WATER_UP)
.await
.expect("apply the high-water migration over an existing ledger");
}
for (n, id) in ids.iter().enumerate() {
let mark = repo
.ledger_high_water(*id)
.await
.expect("mark")
.expect("the backfill must have marked this chain");
let head = repo.ledger_head(*id).await.expect("head").expect("head");
assert_eq!(mark.seq, head.seq, "record {id} ({n})");
assert_eq!(mark.hash, head.hash);
assert_eq!(mark.recorded_at, head.recorded_at);
assert!(
repo.ledger_verify(*id).await.expect("verify").is_intact(),
"a backfilled chain must not be accused"
);
}
repo.update(
ids[0],
&UpdateLgInvoice {
amount_cents: Patch::Set(99),
..Default::default()
},
)
.await
.expect("update after the backfill");
assert!(
repo.ledger_verify(ids[0])
.await
.expect("verify")
.is_intact()
);
}
#[tokio::test]
async fn high_water_marks_are_per_tenant() {
let pool = boot_pool("lg_tenant_mark").await;
let repo = PgLgTenantInvoiceRepository::with_pool_untracked(pool.clone());
let a_id = with_tenant("tenant-a".to_string(), async {
let created = repo
.save(&NewLgTenantInvoice {
reference: "A-1".to_string(),
})
.await
.expect("insert as tenant-a");
for step in 0..2 {
repo.update(
created.id,
&UpdateLgTenantInvoice {
reference: Patch::Set(format!("A-1-{step}")),
..Default::default()
},
)
.await
.expect("update as tenant-a");
}
created.id
})
.await;
let b_id = with_tenant("tenant-b".to_string(), async {
repo.save(&NewLgTenantInvoice {
reference: "B-1".to_string(),
})
.await
.expect("insert as tenant-b")
.id
})
.await;
with_tenant("tenant-a".to_string(), async {
let mark = repo
.ledger_high_water(a_id)
.await
.expect("mark")
.expect("tenant-a has a mark");
assert_eq!(mark.seq, 3, "tenant-a's own chain length, not tenant-b's");
assert!(repo.ledger_verify(a_id).await.expect("verify").is_intact());
})
.await;
with_tenant("tenant-b".to_string(), async {
let mark = repo
.ledger_high_water(b_id)
.await
.expect("mark")
.expect("tenant-b has a mark");
assert_eq!(mark.seq, 1);
assert!(repo.ledger_verify(b_id).await.expect("verify").is_intact());
if a_id == b_id {
assert_eq!(
repo.ledger_high_water(a_id)
.await
.expect("mark")
.unwrap()
.seq,
1
);
}
})
.await;
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"DELETE FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_tenant_invoices' AND tenant_id = 'tenant-a' \
AND record_id = ? AND seq = 3",
)
.bind::<diesel::sql_types::BigInt, _>(a_id)
.execute(&mut *conn)
.await
.expect("truncate tenant-a's chain");
}
with_tenant("tenant-a".to_string(), async {
let broken = repo
.ledger_verify(a_id)
.await
.expect("verify")
.broken
.expect("tenant-a's truncation must be reported");
assert_eq!(broken.kind, LedgerBreak::MissingRevision);
assert_eq!(broken.seq, 3);
})
.await;
with_tenant("tenant-b".to_string(), async {
assert!(
repo.ledger_verify(b_id).await.expect("verify").is_intact(),
"tenant-b must not inherit tenant-a's accusation"
);
})
.await;
}
#[tokio::test]
async fn bulk_writes_chain_every_row_independently() {
let pool = boot_pool("lg_bulk").await;
let repo = PgLgInvoiceRepository::with_pool_untracked(pool);
let created = repo
.save_many(&[
NewLgInvoice {
reference: "BULK-1".to_string(),
amount_cents: 1,
amount_rate: 1e16,
metadata: "{}".to_string(),
},
NewLgInvoice {
reference: "BULK-2".to_string(),
amount_cents: 2,
amount_rate: 2.5,
metadata: "{}".to_string(),
},
])
.await
.expect("save_many");
assert_eq!(created.len(), 2);
let ids: Vec<i64> = created.iter().map(|r| r.id).collect();
for id in &ids {
let revisions = repo.ledger_revisions(*id).await.expect("revisions");
assert_eq!(
revisions.len(),
1,
"each bulk-inserted row opens its own chain"
);
assert_eq!(revisions[0].seq, 1);
assert_eq!(revisions[0].prev_hash, None);
assert!(repo.ledger_verify(*id).await.expect("verify").is_intact());
}
repo.delete_many(&ids).await.expect("delete_many");
for id in &ids {
let revisions = repo.ledger_revisions(*id).await.expect("revisions");
assert_eq!(
revisions.iter().map(|r| r.op).collect::<Vec<_>>(),
vec![VersionOp::Insert, VersionOp::Delete],
"a bulk soft-delete appends one revision per row"
);
assert_eq!(revisions[1].seq, 2);
assert_eq!(
revisions[1].prev_hash.as_deref(),
Some(revisions[0].hash.as_str())
);
let report = repo.ledger_verify(*id).await.expect("verify");
assert!(report.is_intact(), "record {id}: {report:?}");
}
}
#[tokio::test]
async fn a_truncated_tail_is_detected_when_only_an_encrypted_column_changed() {
let pool = boot_pool("lg_vault_tail").await;
let repo = PgLgVaultNoteRepository::with_pool_untracked(pool.clone());
let created = repo
.save(&NewLgVaultNote {
body: "public".to_string(),
secret: "before".to_string(),
})
.await
.expect("insert");
repo.update(
created.id,
&UpdateLgVaultNote {
secret: Patch::Set("after".to_string()),
..Default::default()
},
)
.await
.expect("update an encrypted column only");
let revisions = repo.ledger_revisions(created.id).await.expect("revisions");
assert_eq!(revisions.len(), 2);
let stored_secret = revisions[1].snapshot["secret"]
.as_str()
.expect("the snapshot carries the encrypted column");
assert_ne!(
stored_secret, "after",
"the ledger must not store the plaintext of an encrypted column"
);
assert!(
repo.ledger_verify(created.id)
.await
.expect("verify")
.is_intact(),
"no false positive on an intact chain with an encrypted column"
);
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"DELETE FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_vault_notes' AND record_id = ? AND seq = 2",
)
.bind::<diesel::sql_types::BigInt, _>(created.id)
.execute(&mut *conn)
.await
.expect("lop off the revision that changed only the encrypted column");
}
resync_high_water_mark(&pool, "lg_vault_notes", created.id).await;
let broken = repo
.ledger_verify(created.id)
.await
.expect("verify")
.broken
.expect("an encrypted-column-only truncation must still be detected");
assert_eq!(broken.kind, LedgerBreak::LiveStateMismatch);
assert_eq!(broken.seq, 1);
}
#[tokio::test]
async fn an_encrypted_column_does_not_cause_a_false_positive() {
let pool = boot_pool("lg_vault_intact").await;
let repo = PgLgVaultNoteRepository::with_pool_untracked(pool);
let created = repo
.save(&NewLgVaultNote {
body: "public".to_string(),
secret: "s1".to_string(),
})
.await
.expect("insert");
for secret in ["s2", "s3"] {
repo.update(
created.id,
&UpdateLgVaultNote {
secret: Patch::Set(secret.to_string()),
..Default::default()
},
)
.await
.expect("update");
}
repo.delete_by_id(created.id).await.expect("soft delete");
repo.restore(created.id).await.expect("restore");
for _ in 0..3 {
let report = repo.ledger_verify(created.id).await.expect("verify");
assert!(report.is_intact(), "{report:?}");
assert_eq!(report.revisions_checked, 5);
}
let reconstructed = repo
.ledger_as_of_at(created.id, LedgerAsOf::default())
.await
.expect("as-of")
.expect("state");
assert_eq!(reconstructed.secret, "s3");
}
#[tokio::test]
async fn a_hard_parent_delete_is_refused_rather_than_erasing_a_ledgered_child() {
let pool = boot_pool("lg_cascade").await;
let parents = PgLgCascadeParentRepository::with_pool_untracked(pool.clone());
let children = PgLgCascadeChildRepository::with_pool_untracked(pool);
let parent = parents
.save(&NewLgCascadeParent {
name: "p".to_string(),
})
.await
.expect("insert parent");
let child = children
.save(&NewLgCascadeChild {
parent_id: parent.id,
label: "c".to_string(),
})
.await
.expect("insert ledgered child");
let err = parents
.delete_by_id(parent.id)
.await
.expect_err("a hard parent delete must not erase a ledgered child");
let typed = err
.downcast_chain_ref::<autumn_web::ledger::LedgerError>()
.expect("the refusal is a typed LedgerError, not a foreign-key violation");
assert!(
matches!(
typed,
autumn_web::ledger::LedgerError::HardDeleteCascade { record_id, .. }
if *record_id == child.id
),
"{typed:?}"
);
let rendered = typed.to_string();
assert!(rendered.contains("soft_delete"), "{rendered}");
let live = children
.find_by_id(child.id)
.await
.expect("read child")
.expect("the child survives a refused cascade");
assert!(live.deleted_at.is_none());
let report = children.ledger_verify(child.id).await.expect("verify");
assert!(report.is_intact(), "{report:?}");
assert_eq!(report.revisions_checked, 1);
}
#[tokio::test]
async fn a_truncated_tail_is_detected_when_only_a_hidden_column_changed() {
let pool = boot_pool("lg_hidden_tail").await;
let repo = PgLgSecretNoteRepository::with_pool_untracked(pool.clone());
let created = repo
.save(&NewLgSecretNote {
body: "public".to_string(),
internal_note: "before".to_string(),
})
.await
.expect("insert");
repo.update(
created.id,
&UpdateLgSecretNote {
internal_note: Patch::Set("after".to_string()),
..Default::default()
},
)
.await
.expect("update a hidden column only");
let revisions = repo.ledger_revisions(created.id).await.expect("revisions");
assert_eq!(revisions.len(), 2);
assert_eq!(
revisions[0].snapshot["body"], revisions[1].snapshot["body"],
"the public column is unchanged across the two revisions"
);
assert_ne!(
revisions[0].snapshot["internal_note"], revisions[1].snapshot["internal_note"],
"the hidden column is what changed — and the snapshot carries it"
);
assert!(
repo.ledger_verify(created.id)
.await
.expect("verify")
.is_intact(),
"no false positive on an intact chain with a hidden column"
);
{
let mut conn = pool.get().await.expect("conn");
diesel::sql_query(
"DELETE FROM _autumn_ledger_revisions \
WHERE table_name = 'lg_secret_notes' AND record_id = ? AND seq = 2",
)
.bind::<diesel::sql_types::BigInt, _>(created.id)
.execute(&mut *conn)
.await
.expect("lop off the revision that changed only the hidden column");
}
resync_high_water_mark(&pool, "lg_secret_notes", created.id).await;
let broken = repo
.ledger_verify(created.id)
.await
.expect("verify")
.broken
.expect("a hidden-column-only truncation must still be detected");
assert_eq!(broken.kind, LedgerBreak::LiveStateMismatch);
assert_eq!(broken.seq, 1);
}
#[tokio::test]
async fn a_hidden_column_does_not_cause_a_false_positive() {
let pool = boot_pool("lg_hidden_intact").await;
let repo = PgLgSecretNoteRepository::with_pool_untracked(pool);
let created = repo
.save(&NewLgSecretNote {
body: "public".to_string(),
internal_note: "secret".to_string(),
})
.await
.expect("insert");
for note in ["secret-2", "secret-3"] {
repo.update(
created.id,
&UpdateLgSecretNote {
internal_note: Patch::Set(note.to_string()),
..Default::default()
},
)
.await
.expect("update");
}
repo.delete_by_id(created.id).await.expect("soft delete");
repo.restore(created.id).await.expect("restore");
for _ in 0..3 {
let report = repo.ledger_verify(created.id).await.expect("verify");
assert!(report.is_intact(), "{report:?}");
assert_eq!(report.revisions_checked, 5);
}
let reconstructed = repo
.ledger_as_of_at(created.id, LedgerAsOf::default())
.await
.expect("as-of")
.expect("state");
assert_eq!(reconstructed.internal_note, "secret-3");
}
#[tokio::test]
async fn a_declared_valid_time_column_separates_the_two_axes() {
let pool = boot_pool("lg_valid_time").await;
let repo = PgLgEffectiveNoteRepository::with_pool_untracked(pool);
let effective = chrono::NaiveDate::from_ymd_opt(2026, 1, 1)
.expect("date")
.and_hms_opt(0, 0, 0)
.expect("time");
let created = repo
.save(&NewLgEffectiveNote {
body: "v1".to_string(),
effective_at: effective,
})
.await
.expect("insert");
let revisions = repo.ledger_revisions(created.id).await.expect("revisions");
assert_eq!(revisions.len(), 1);
assert_eq!(
revisions[0].valid_from,
effective.and_utc(),
"valid time comes from the declared column, not the clock"
);
assert!(
revisions[0].recorded_at > revisions[0].valid_from,
"transaction time is now; valid time is back in January"
);
let earlier = chrono::NaiveDate::from_ymd_opt(2025, 6, 1)
.expect("date")
.and_hms_opt(0, 0, 0)
.expect("time");
repo.update(
created.id,
&UpdateLgEffectiveNote {
body: Patch::Set("v2".to_string()),
effective_at: Patch::Set(earlier),
..Default::default()
},
)
.await
.expect("correction");
let live = repo
.find_by_id(created.id)
.await
.expect("live read")
.expect("live row");
let latest = repo
.ledger_as_of(created.id, now() + Duration::seconds(1))
.await
.expect("as-of")
.expect("state");
assert_eq!(latest, live, "as-of(now) must equal the live row");
assert_eq!(latest.body, "v2");
let mid_2025 = chrono::NaiveDate::from_ymd_opt(2025, 8, 1)
.expect("date")
.and_hms_opt(0, 0, 0)
.expect("time")
.and_utc();
let then = repo
.ledger_as_of_at(created.id, LedgerAsOf::valid(mid_2025))
.await
.expect("as-of")
.expect("state");
assert_eq!(
then.body, "v2",
"the back-dated correction governs August 2025"
);
let mid_2026 = chrono::NaiveDate::from_ymd_opt(2026, 6, 1)
.expect("date")
.and_hms_opt(0, 0, 0)
.expect("time")
.and_utc();
let later = repo
.ledger_as_of_at(created.id, LedgerAsOf::valid(mid_2026))
.await
.expect("as-of")
.expect("state");
assert_eq!(
later.body, "v2",
"a correction supersedes what it corrects from its own effective_at onward"
);
assert!(
repo.ledger_verify(created.id)
.await
.expect("verify")
.is_intact()
);
}
async fn write_three_revisions(repo: &PgLgInvoiceRepository) -> i64 {
let created = repo
.save(&NewLgInvoice {
reference: "INV-X".to_string(),
amount_cents: 1,
amount_rate: 1e16,
metadata: r#"{"b":1,"a":{"y":2,"x":3}}"#.to_string(),
})
.await
.expect("insert");
for amount in [2, 3] {
repo.update(
created.id,
&UpdateLgInvoice {
amount_cents: Patch::Set(amount),
..Default::default()
},
)
.await
.expect("update");
}
created.id
}