#[path = "common/harness.rs"]
mod harness;
const ARCHIVED_AT: &str = "2026-07-30T12:00:00.000000Z";
const CTS: &str = "2026-01-01T00:00:00.000000Z";
use harness::TestHarness;
use macrame::graph::AttributeMode;
use macrame::integrity::audit_current;
use macrame::temporal::{
archive, hydrate_attributes, load_snapshot, reconstruct, save_snapshot, AsOf, Interval,
MaterializedState,
};
use std::path::Path;
#[test]
fn test_interval_containment_and_overlap() {
let open_interval = Interval::new("2026-01-01T00:00:00.000000Z", "9999-12-31T23:59:59.999999Z");
assert!(open_interval.is_open());
assert!(open_interval.contains("2026-06-01T00:00:00.000000Z"));
let closed_interval =
Interval::new("2026-01-01T00:00:00.000000Z", "2026-06-01T00:00:00.000000Z");
assert!(!closed_interval.is_open());
assert!(closed_interval.contains("2026-03-01T00:00:00.000000Z"));
assert!(!closed_interval.contains("2026-07-01T00:00:00.000000Z"));
let overlapping = Interval::new("2026-05-01T00:00:00.000000Z", "2026-08-01T00:00:00.000000Z");
let non_overlapping =
Interval::new("2026-07-01T00:00:00.000000Z", "2026-09-01T00:00:00.000000Z");
assert!(closed_interval.overlaps(&overlapping));
assert!(!closed_interval.overlaps(&non_overlapping));
}
#[tokio::test]
async fn test_monday_wednesday_friday_scenario() {
let harness = TestHarness::new();
let db = libsql::Builder::new_local(&harness.db_path)
.build()
.await
.unwrap();
let conn = db.connect().unwrap();
macrame::schema::run_migrations(&conn).await.unwrap();
conn.execute(
"INSERT INTO concepts (id, title, content, valid_from, recorded_at) \
VALUES ('c1', 'Monday Title', 'Content', '2026-01-05T00:00:00.000000Z', '2026-01-05T00:00:00.000000Z')",
(),
)
.await
.unwrap();
conn.execute(
"UPDATE concepts SET title = 'Wednesday Title', recorded_at = '2026-01-07T00:00:00.000000Z' WHERE id = 'c1'",
(),
)
.await
.unwrap();
let node_ids = vec!["c1".to_string()];
let tuesday_ts = "2026-01-06T00:00:00.000000Z";
let current_attrs = hydrate_attributes(&conn, &node_ids, &AsOf::now(), AttributeMode::Current)
.await
.unwrap();
assert_eq!(current_attrs[0].title, "Wednesday Title");
let at_time_attrs = hydrate_attributes(
&conn,
&node_ids,
&AsOf::recorded_at(tuesday_ts),
AttributeMode::AtTime,
)
.await
.unwrap();
assert_eq!(at_time_attrs[0].title, "Monday Title");
let omit_attrs = hydrate_attributes(&conn, &node_ids, &AsOf::now(), AttributeMode::Omit)
.await
.unwrap();
assert!(omit_attrs.is_empty());
}
#[tokio::test]
async fn test_reconstruct_now_equals_live_table() {
let harness = TestHarness::new();
let db = libsql::Builder::new_local(&harness.db_path)
.build()
.await
.unwrap();
let conn = db.connect().unwrap();
macrame::schema::run_migrations(&conn).await.unwrap();
conn.execute(
"INSERT INTO concepts (id, title, valid_from, recorded_at) VALUES ('c1', 'Node 1', '2026-01-01T00:00:00.000000Z', '2026-01-01T00:00:00.000000Z')",
(),
)
.await
.unwrap();
conn.execute(
"INSERT INTO concepts (id, title, valid_from, recorded_at) VALUES ('c2', 'Node 2', '2026-01-01T00:00:00.000000Z', '2026-01-01T00:00:00.000000Z')",
(),
)
.await
.unwrap();
conn.execute(
"INSERT INTO links (source_id, target_id, edge_type, valid_from, valid_to, weight, properties, recorded_at) \
VALUES ('c1', 'c2', 'KNOWS', '2026-01-01T00:00:00.000000Z', '9999-12-31T23:59:59.999999Z', 1.0, '{}', '2026-01-01T00:00:00.000000Z')",
(),
)
.await
.unwrap();
let state = reconstruct(&conn, "2026-01-10T00:00:00.000000Z", None, None)
.await
.unwrap();
assert!(state.concepts.contains_key("c1"));
assert!(state.concepts.contains_key("c2"));
assert_eq!(state.edges.len(), 1);
assert_eq!(state.edges[0].source_id, "c1");
assert_eq!(state.edges[0].target_id, "c2");
}
#[test]
fn test_snapshot_save_load_roundtrip() {
let harness = TestHarness::new();
let snapshots_dir = harness.temp_dir.path().join("snapshots");
let mut state = MaterializedState::empty("2026-01-01T00:00:00.000000Z");
state.seq_anchor = 100;
state.edges = vec![macrame::temporal::EdgeBelief::new(
"c1",
"c2",
"KNOWS",
"2026-01-01T00:00:00.000000Z",
"9999-12-31T23:59:59.999999Z",
)];
let path = save_snapshot(&snapshots_dir, &state).unwrap();
assert!(path.exists());
let loaded = load_snapshot(&path).unwrap();
assert_eq!(loaded.seq_anchor, 100);
assert_eq!(loaded.edges.len(), 1);
assert_eq!(loaded.edges[0].source_id, "c1");
}
#[tokio::test]
async fn test_archive_moves_closed_intervals_and_leaves_no_drift() {
let harness = TestHarness::new();
let db = libsql::Builder::new_local(&harness.db_path)
.build()
.await
.unwrap();
let conn = db.connect().unwrap();
macrame::schema::run_migrations(&conn).await.unwrap();
for id in ["c1", "c2"] {
conn.execute(
&format!("INSERT INTO concepts (id, title, valid_from, recorded_at) VALUES ('{id}', 'N', '2026-01-01T00:00:00.000000Z', '2026-01-01T00:00:00.000000Z')"),
(),
).await.unwrap();
}
conn.execute(
"INSERT INTO links (source_id, target_id, edge_type, valid_from, valid_to, weight, properties, recorded_at) \
VALUES ('c1', 'c2', 'OLD', '2026-01-01T00:00:00.000000Z', '2026-02-01T00:00:00.000000Z', 1.0, '{}', '2026-01-01T00:00:00.000000Z')",
(),
).await.unwrap();
conn.execute(
"INSERT INTO links (source_id, target_id, edge_type, valid_from, valid_to, weight, properties, recorded_at) \
VALUES ('c1', 'c2', 'LIVE', '2026-01-01T00:00:00.000000Z', '9999-12-31T23:59:59.999999Z', 1.0, '{}', '2026-01-01T00:00:00.000000Z')",
(),
).await.unwrap();
let archive_path = harness.temp_dir.path().join("test_macrame_archive.db");
let report = archive(
&conn,
"2026-06-01T00:00:00.000000Z",
ARCHIVED_AT,
&archive_path,
)
.await
.expect("archive session should succeed");
assert!(
archive_path.exists(),
"cold database file should be created"
);
assert_eq!(
report.links_archived, 1,
"only the closed interval is archivable"
);
let remaining: i64 = conn
.query("SELECT COUNT(*) FROM links", ())
.await
.unwrap()
.next()
.await
.unwrap()
.unwrap()
.get(0)
.unwrap();
assert_eq!(remaining, 1, "the open interval must stay hot");
let live: i64 = conn
.query(
"SELECT COUNT(*) FROM links_current WHERE edge_type = 'LIVE'",
(),
)
.await
.unwrap()
.next()
.await
.unwrap()
.unwrap()
.get(0)
.unwrap();
assert_eq!(live, 1);
assert_eq!(
audit_current(&conn).await.unwrap(),
0,
"archive must not induce drift"
);
assert!(conn.execute("DELETE FROM links", ()).await.is_err());
archive(
&conn,
"2026-06-01T00:00:00.000000Z",
ARCHIVED_AT,
&archive_path,
)
.await
.expect("second archive session should succeed (cold DB detached)");
}
#[tokio::test]
async fn a_missing_archive_is_an_error_when_rows_were_actually_archived() {
use macrame::prelude::*;
let harness = TestHarness::new();
let db = macrame::Database::open(&harness.db_path).await.unwrap();
for title in ["first", "second", "third"] {
db.upsert_concept(ConceptUpsert::new("c1", title).valid_from(CTS))
.await
.unwrap();
}
let report = db.archive("2030-01-01T00:00:00.000000Z").await.unwrap();
assert!(
report.log_entries_archived >= 2,
"the fixture needs log rows to have actually moved: {report:?}"
);
db.close().await.unwrap();
let archive_path = harness.db_path.with_file_name("kb_archive.db");
let archive_path = if archive_path.exists() {
archive_path
} else {
std::fs::read_dir(harness.temp_dir.path())
.unwrap()
.flatten()
.map(|e| e.path())
.find(|p| p.to_string_lossy().contains("archive"))
.expect("archive() must have created a cold file")
};
std::fs::remove_file(&archive_path).unwrap();
let conn = libsql::Builder::new_local(&harness.db_path)
.build()
.await
.unwrap()
.connect()
.unwrap();
macrame::schema::run_migrations(&conn).await.unwrap();
let res = reconstruct(
&conn,
"2020-01-01T00:00:00.000000Z",
Some(&archive_path),
None,
)
.await;
match res {
Err(DbError::ReplayCorrupt { reason, .. }) => {
assert!(
reason.contains("does not exist"),
"the error should name the missing file: {reason}"
);
}
other => panic!("expected ReplayCorrupt for a missing archive, got {other:?}"),
}
}
#[tokio::test]
async fn reconstructing_below_the_log_floor_is_not_a_corruption() {
use macrame::prelude::*;
let harness = TestHarness::new();
let db = macrame::Database::open(&harness.db_path).await.unwrap();
let before_any_data = db.reconstruct("2020-01-01T00:00:00.000000Z").await.unwrap();
assert!(before_any_data.concepts.is_empty());
assert!(before_any_data.predates_recorded_history);
db.upsert_concept(ConceptUpsert::new("c0", "Title").valid_from(CTS))
.await
.unwrap();
db.upsert_concept(ConceptUpsert::new("c1", "Other").valid_from(CTS))
.await
.unwrap();
db.assert_edge(EdgeAssertion::new("c0", "c1", "KNOWS").valid_from(CTS))
.await
.unwrap();
let after = db.reconstruct("2020-01-01T00:00:00.000000Z").await.unwrap();
assert!(
after.concepts.is_empty() && after.edges.is_empty(),
"nothing had been recorded by 2020"
);
assert!(
after.predates_recorded_history,
"the caller cannot otherwise tell this from a fully retired ledger"
);
let inside_valid_time = db.reconstruct("2026-02-01T00:00:00.000000Z").await.unwrap();
assert!(
inside_valid_time.predates_recorded_history,
"the floor is MIN(recorded_at), not MIN(valid_from)"
);
let now = db.reconstruct(&max_recorded_at(&db).await).await.unwrap();
assert_eq!(now.concepts.len(), 2, "the ordinary fold still works");
assert!(
!now.predates_recorded_history,
"a fold with rows in it must not claim the ledger had not started"
);
db.close().await.unwrap();
}
async fn max_recorded_at(db: ¯ame::Database) -> String {
db.read_conn()
.query("SELECT MAX(recorded_at) FROM transaction_log", ())
.await
.unwrap()
.next()
.await
.unwrap()
.unwrap()
.get(0)
.unwrap()
}
const VIII_JAN: &str = "2026-01-01T00:00:00.000000Z";
const VIII_FEB: &str = "2026-02-01T00:00:00.000000Z";
const VIII_MAR: &str = "2026-03-01T00:00:00.000000Z";
const VIII_OPEN: &str = "9999-12-31T23:59:59.999999Z";
async fn newest_stamp(conn: &libsql::Connection) -> String {
conn.query("SELECT MAX(recorded_at) FROM transaction_log", ())
.await
.unwrap()
.next()
.await
.unwrap()
.unwrap()
.get(0)
.unwrap()
}
fn at_valid_time(
state: &MaterializedState,
t: &str,
) -> std::collections::BTreeSet<(String, String, String, String, String)> {
state
.edges
.iter()
.filter(|e| e.valid_from.as_str() <= t && t < e.valid_to.as_str())
.map(|e| {
(
e.source_id.clone(),
e.target_id.clone(),
e.edge_type.clone(),
e.valid_from.clone(),
e.valid_to.clone(),
)
})
.collect()
}
async fn as_of_set(
conn: &libsql::Connection,
t: &str,
) -> std::collections::BTreeSet<(String, String, String, String, String)> {
macrame::temporal::query_as_of_edges(conn, t)
.await
.unwrap()
.into_iter()
.collect()
}
#[tokio::test]
async fn a_retroactive_retirement_makes_as_of_and_reconstruct_diverge() {
let harness = TestHarness::new();
let db = macrame::prelude::Database::open(&harness.db_path)
.await
.unwrap();
for id in ["c1", "c2"] {
db.upsert_concept(macrame::prelude::ConceptUpsert::new(id, "N").valid_from(VIII_JAN))
.await
.unwrap();
}
db.assert_edge(
macrame::prelude::EdgeAssertion::new("c1", "c2", "REL")
.valid_from(VIII_JAN)
.valid_to(VIII_OPEN),
)
.await
.unwrap();
let believed_open = newest_stamp(db.read_conn()).await;
db.retire_edge("c1", "c2", "REL", VIII_JAN, VIII_FEB)
.await
.unwrap();
let believed_closed = newest_stamp(db.read_conn()).await;
assert!(
believed_open < believed_closed,
"the correction must be recorded strictly later, or there is no gap to test"
);
let live = as_of_set(db.read_conn(), VIII_MAR).await;
let then = at_valid_time(&db.reconstruct(&believed_open).await.unwrap(), VIII_MAR);
let now = at_valid_time(&db.reconstruct(&believed_closed).await.unwrap(), VIII_MAR);
assert!(
live.is_empty(),
"current belief has the interval closed in February, so March is outside it: {live:?}"
);
assert_eq!(
then.len(),
1,
"belief at the first stamp had the interval open, so March is inside it: {then:?}"
);
assert_ne!(
live, then,
"as_of and reconstruct agreed across a retroactive retirement; \
Doctrine VIII requires the gap to be visible, not smoothed over"
);
assert_eq!(
live, now,
"once the correction is itself in the past, the two questions have the same answer"
);
db.close().await.unwrap();
}
#[tokio::test]
async fn a_retroactive_assertion_is_invisible_to_the_earlier_belief() {
let harness = TestHarness::new();
let db = macrame::prelude::Database::open(&harness.db_path)
.await
.unwrap();
for id in ["c1", "c2"] {
db.upsert_concept(macrame::prelude::ConceptUpsert::new(id, "N").valid_from(VIII_JAN))
.await
.unwrap();
}
db.assert_edge(
macrame::prelude::EdgeAssertion::new("c1", "c2", "EARLY")
.valid_from(VIII_JAN)
.valid_to(VIII_OPEN),
)
.await
.unwrap();
let before_the_late_news = newest_stamp(db.read_conn()).await;
db.assert_edge(
macrame::prelude::EdgeAssertion::new("c1", "c2", "LATE")
.valid_from(VIII_JAN)
.valid_to(VIII_OPEN),
)
.await
.unwrap();
let live = as_of_set(db.read_conn(), VIII_MAR).await;
let then = at_valid_time(
&db.reconstruct(&before_the_late_news).await.unwrap(),
VIII_MAR,
);
let types = |s: &std::collections::BTreeSet<(String, String, String, String, String)>| {
s.iter()
.map(|e| e.2.clone())
.collect::<std::collections::BTreeSet<_>>()
};
assert!(
types(&live).contains("LATE"),
"current belief must include the retroactively asserted edge: {live:?}"
);
assert!(
!types(&then).contains("LATE"),
"belief at the earlier stamp cannot contain something recorded after it: {then:?}"
);
assert_ne!(
live, then,
"as_of and reconstruct agreed across a retroactive assertion; \
reconstruct is reading current state rather than folding the ledger"
);
assert!(
types(&then).contains("EARLY"),
"the earlier belief must still hold what it did know: {then:?}"
);
db.close().await.unwrap();
}
async fn row_set(
conn: &libsql::Connection,
expr: &str,
table: &str,
) -> std::collections::BTreeSet<String> {
let mut rows = conn
.query(&format!("SELECT {expr} FROM {table}"), ())
.await
.unwrap();
let mut out = std::collections::BTreeSet::new();
while let Some(r) = rows.next().await.unwrap() {
out.insert(r.get::<String>(0).unwrap());
}
out
}
const HOT_LINK_ROW: &str = "source_id||'|'||target_id||'|'||edge_type||'|'||valid_from||'|'||\
valid_to||'|'||weight||'|'||properties||'|'||recorded_at";
const HOT_LOG_ROW: &str = "seq_id||'|'||table_name||'|'||entity_id||'|'||operation||'|'||\
payload||'|'||recorded_at";
const HOT_CURRENT_ROW: &str = "source_id||'|'||target_id||'|'||edge_type||'|'||valid_from||'|'||\
valid_to||'|'||recorded_at";
async fn archivable_fixture(path: &Path) -> (libsql::Database, libsql::Connection) {
let db = libsql::Builder::new_local(path).build().await.unwrap();
let conn = db.connect().unwrap();
macrame::schema::run_migrations(&conn).await.unwrap();
for id in ["c1", "c2"] {
conn.execute(
&format!(
"INSERT INTO concepts (id, title, valid_from, recorded_at) \
VALUES ('{id}', 'N', '2026-01-01T00:00:00.000000Z', '2026-01-01T00:00:00.000000Z')"
),
(),
)
.await
.unwrap();
}
for (etype, valid_to) in [
("OLD", "2026-02-01T00:00:00.000000Z"),
("LIVE", "9999-12-31T23:59:59.999999Z"),
] {
conn.execute(
&format!(
"INSERT INTO links (source_id, target_id, edge_type, valid_from, valid_to, \
weight, properties, recorded_at) VALUES \
('c1', 'c2', '{etype}', '2026-01-01T00:00:00.000000Z', '{valid_to}', \
1.0, '{{}}', '2026-01-01T00:00:00.000000Z')"
),
(),
)
.await
.unwrap();
}
(db, conn)
}
#[tokio::test]
async fn a_failed_archive_session_leaves_the_hot_database_untouched_and_the_guards_armed() {
let cases: [(&str, &str, &str); 2] = [
(
"links",
"CREATE TABLE links (not_the_right_column TEXT)",
"source_id",
),
(
"transaction_log",
"CREATE TABLE transaction_log (seq_id INTEGER PRIMARY KEY, table_name TEXT, \
entity_id TEXT, operation TEXT, recorded_at TEXT)",
"payload",
),
];
for (broken_table, broken_ddl, missing_column) in cases {
let harness = TestHarness::new();
let (_db, conn) = archivable_fixture(&harness.db_path).await;
let cold = harness
.temp_dir
.path()
.join(format!("incompatible_{broken_table}.db"));
{
let d = libsql::Builder::new_local(&cold).build().await.unwrap();
let c = d.connect().unwrap();
c.execute(broken_ddl, ()).await.unwrap();
}
let before_links = row_set(&conn, HOT_LINK_ROW, "links").await;
let before_log = row_set(&conn, HOT_LOG_ROW, "transaction_log").await;
let before_current = row_set(&conn, HOT_CURRENT_ROW, "links_current").await;
assert!(
!before_links.is_empty(),
"the fixture must have rows to lose"
);
let cutoff = "2026-06-01T00:00:00.000000Z";
let err = archive(&conn, cutoff, ARCHIVED_AT, &cold)
.await
.expect_err(
"the session was supposed to fail on the incompatible cold table; \
if it succeeded this case tests nothing",
)
.to_string();
assert!(
err.contains(missing_column),
"[{broken_table}] expected the failure to be the copy missing {missing_column:?}, \
so the session dies at the phase this case exists to cover; got: {err}"
);
assert_eq!(
row_set(&conn, HOT_LINK_ROW, "links").await,
before_links,
"[{broken_table}] a failed archive session moved or dropped rows from links"
);
assert_eq!(
row_set(&conn, HOT_LOG_ROW, "transaction_log").await,
before_log,
"[{broken_table}] a failed archive session moved or dropped ledger rows"
);
assert_eq!(
row_set(&conn, HOT_CURRENT_ROW, "links_current").await,
before_current,
"[{broken_table}] a failed archive session left the materialization re-derived"
);
assert_eq!(
audit_current(&conn).await.unwrap(),
0,
"[{broken_table}] a failed archive session left links_current drifted from links"
);
let marker: i64 = conn
.query(
"SELECT COUNT(*) FROM sqlite_master \
WHERE type='table' AND name='macrame_archive_session'",
(),
)
.await
.unwrap()
.next()
.await
.unwrap()
.unwrap()
.get(0)
.unwrap();
assert_eq!(
marker, 0,
"[{broken_table}] the archive-session marker outlived the session: the \
delete guards are disarmed and will stay that way"
);
assert!(
conn.execute("DELETE FROM links", ()).await.is_err(),
"[{broken_table}] an ad-hoc DELETE succeeded after a failed archive \
session (Doctrine V)"
);
let good = harness.temp_dir.path().join("good_cold.db");
let report = archive(&conn, cutoff, ARCHIVED_AT, &good)
.await
.expect("a later archive session must still succeed after a failed one");
assert_eq!(
report.links_archived, 1,
"[{broken_table}] the closed interval is still archivable"
);
}
}
#[tokio::test]
async fn the_unreachable_archive_error_says_how_much_went_and_how_far_back_the_log_reaches() {
use macrame::prelude::*;
let harness = TestHarness::new();
let db = macrame::Database::open(&harness.db_path).await.unwrap();
for title in ["first", "second", "third", "fourth"] {
db.upsert_concept(ConceptUpsert::new("c1", title).valid_from(CTS))
.await
.unwrap();
}
let report = db.archive("2030-01-01T00:00:00.000000Z").await.unwrap();
assert!(
report.log_entries_archived >= 2,
"the fixture needs log rows to have actually moved: {report:?}"
);
let horizon = report
.horizon
.expect("an archive that moved rows has a horizon");
db.close().await.unwrap();
let archive_path = std::fs::read_dir(harness.temp_dir.path())
.unwrap()
.flatten()
.map(|e| e.path())
.find(|p| p.to_string_lossy().contains("archive"))
.expect("archive() must have created a cold file");
std::fs::remove_file(&archive_path).unwrap();
let conn = libsql::Builder::new_local(&harness.db_path)
.build()
.await
.unwrap()
.connect()
.unwrap();
macrame::schema::run_migrations(&conn).await.unwrap();
let res = reconstruct(
&conn,
"2020-01-01T00:00:00.000000Z",
Some(&archive_path),
None,
)
.await;
let Err(DbError::ReplayCorrupt { reason, .. }) = res else {
panic!("expected ReplayCorrupt for a missing archive, got {res:?}");
};
assert!(
reason.contains(&format!(
"{} log rows have been archived",
report.log_entries_archived
)),
"the hint's count disagrees with what the archive reported moving \
({}): {reason}",
report.log_entries_archived
);
assert!(
reason.contains(&format!("begins at seq_id {horizon}")),
"the hint should name the horizon the archive recorded ({horizon}): {reason}"
);
}
#[tokio::test]
async fn a_rehydration_leaves_the_archive_hint_saying_exactly_what_it_said_before() {
use macrame::prelude::*;
let harness = TestHarness::new();
let db = macrame::Database::open(&harness.db_path).await.unwrap();
for title in ["first", "second"] {
db.upsert_concept(ConceptUpsert::new("c1", title).valid_from(CTS))
.await
.unwrap();
}
db.upsert_concept(
ConceptUpsert::new("c1", "last")
.valid_from(CTS)
.valid_to("2027-01-01T00:00:00.000000Z")
.retired(true),
)
.await
.unwrap();
let report = db.archive("2030-01-01T00:00:00.000000Z").await.unwrap();
assert_eq!(
report.concepts_archived, 1,
"the fixture needs the concept itself to have gone cold: {report:?}"
);
assert!(report.log_entries_archived >= 2, "{report:?}");
let horizon = report
.horizon
.expect("an archive that moved rows has a horizon");
let rehydrated = db.rehydrate(&["c1"]).await.unwrap();
assert_eq!(rehydrated.concepts_rehydrated, 1);
db.close().await.unwrap();
let archive_path = std::fs::read_dir(harness.temp_dir.path())
.unwrap()
.flatten()
.map(|e| e.path())
.find(|p| p.to_string_lossy().contains("archive"))
.expect("archive() must have created a cold file");
std::fs::remove_file(&archive_path).unwrap();
let conn = libsql::Builder::new_local(&harness.db_path)
.build()
.await
.unwrap()
.connect()
.unwrap();
macrame::schema::run_migrations(&conn).await.unwrap();
let res = reconstruct(
&conn,
"2020-01-01T00:00:00.000000Z",
Some(&archive_path),
None,
)
.await;
let Err(DbError::ReplayCorrupt { reason, .. }) = res else {
panic!("expected ReplayCorrupt for a missing archive, got {res:?}");
};
assert!(
reason.contains(&format!(
"{} log rows have been archived",
report.log_entries_archived
)),
"the round trip changed the archived-row count the hint reports \
(expected {}): {reason}",
report.log_entries_archived
);
assert!(
reason.contains(&format!("begins at seq_id {horizon}")),
"the round trip moved the horizon the hint reports (expected {horizon}): {reason}"
);
}
const REACH_VALID_FROM: &str = "1970-01-01T00:00:00.000000Z";
const REACH_EARLY: &str = "1970-01-01T00:30:00.000000Z";
const REACH_CUTOFF: &str = "1970-01-01T02:30:00.000000Z";
const REACH_NOW: &str = "1970-01-01T03:00:00.000000Z";
async fn newest_recorded_at(conn: &libsql::Connection) -> String {
conn.query("SELECT MAX(recorded_at) FROM transaction_log", ())
.await
.unwrap()
.next()
.await
.unwrap()
.unwrap()
.get(0)
.unwrap()
}
async fn a_ledger_archived_mid_history(harness: &TestHarness) -> macrame::Database {
use macrame::prelude::*;
let db = harness.db_with_fake_clock().await;
let hour = std::time::Duration::from_secs(3_600);
db.upsert_concept(ConceptUpsert::new("c1", "first").valid_from(REACH_VALID_FROM))
.await
.unwrap();
db.upsert_concept(ConceptUpsert::new("c2", "other").valid_from(REACH_VALID_FROM))
.await
.unwrap();
db.assert_edge(
macrame::graph::EdgeAssertion::new("c1", "c2", "KNOWS").valid_from(REACH_VALID_FROM),
)
.await
.unwrap();
harness.advance(hour);
for title in ["second", "third"] {
db.upsert_concept(ConceptUpsert::new("c1", title).valid_from(REACH_VALID_FROM))
.await
.unwrap();
harness.advance(hour);
}
db
}
#[tokio::test]
async fn a_recorded_instant_is_refused_once_rows_have_been_archived() {
use macrame::graph::TraversalBuilder;
let harness = TestHarness::new();
let db = a_ledger_archived_mid_history(&harness).await;
let before = TraversalBuilder::new("c1")
.max_depth(1)
.as_of_recorded(REACH_EARLY)
.execute_ids(db.read_conn(), REACH_NOW)
.await
.expect("an intact hot log answers for any instant");
assert_eq!(before, vec!["c1".to_string(), "c2".to_string()]);
let report = db.archive(REACH_CUTOFF).await.unwrap();
assert!(
report.log_entries_archived >= 2,
"the fixture needs log rows to have actually moved: {report:?}"
);
let err = TraversalBuilder::new("c1")
.max_depth(1)
.as_of_recorded(REACH_EARLY)
.execute_ids(db.read_conn(), REACH_NOW)
.await
.expect_err("a short hot log must refuse rather than fold what is left");
match &err {
macrame::DbError::RecordedInstantUnreachable { ts } => assert_eq!(ts, REACH_EARLY),
other => panic!("got {other:?}"),
}
assert!(err.to_string().contains("reconstruct"), "{err}");
let by_valid = TraversalBuilder::new("c1")
.max_depth(1)
.as_of_valid(REACH_NOW)
.execute_ids(db.read_conn(), REACH_NOW)
.await
.expect("valid time does not read the log and must be unaffected");
assert_eq!(by_valid, vec!["c1".to_string(), "c2".to_string()]);
let err = db
.edges(
macrame::ReadPlan::new()
.valid_at(REACH_NOW)
.recorded_at(REACH_EARLY),
)
.await
.expect_err("a plan reader that skipped the guard would fold a short log");
match &err {
macrame::DbError::RecordedInstantUnreachable { ts } => assert_eq!(ts, REACH_EARLY),
other => panic!("got {other:?}"),
}
db.edges(macrame::ReadPlan::new().valid_at(REACH_NOW))
.await
.expect("current belief never reads the log");
db.close().await.unwrap();
}
#[tokio::test]
async fn an_instant_at_or_after_the_newest_hot_row_survives_the_archive() {
use macrame::graph::TraversalBuilder;
let harness = TestHarness::new();
let db = a_ledger_archived_mid_history(&harness).await;
let ids = vec!["c1".to_string()];
let late = newest_recorded_at(db.read_conn()).await;
let topology = TraversalBuilder::new("c1")
.max_depth(1)
.as_of_recorded(&late)
.execute_ids(db.read_conn(), REACH_NOW)
.await
.expect("an intact hot log answers for any instant");
let text = hydrate_attributes(
db.read_conn(),
&ids,
&AsOf::recorded_at(&late),
AttributeMode::AtTime,
)
.await
.expect("an intact hot log answers for any instant");
let report = db.archive(REACH_CUTOFF).await.unwrap();
assert!(
report.log_entries_archived >= 2,
"the fixture needs log rows to have actually moved: {report:?}"
);
assert_eq!(
newest_recorded_at(db.read_conn()).await,
late,
"the newest row per entity is never archivable, and the whole rule rests on it"
);
assert_eq!(
TraversalBuilder::new("c1")
.max_depth(1)
.as_of_recorded(&late)
.execute_ids(db.read_conn(), REACH_NOW)
.await
.expect("the newest row per entity is hot, so this instant is answerable"),
topology,
"the archive moved superseded rows, it did not change the topology at {late}"
);
assert_eq!(
hydrate_attributes(
db.read_conn(),
&ids,
&AsOf::recorded_at(&late),
AttributeMode::AtTime,
)
.await
.expect("the second reader takes the same verdict"),
text,
"the archive moved superseded rows, it did not change the text at {late}"
);
db.close().await.unwrap();
}
#[tokio::test]
async fn reconstructing_without_the_archive_path_refuses_rather_than_folding_a_gap() {
let harness = TestHarness::new();
let db = a_ledger_archived_mid_history(&harness).await;
let truth = reconstruct(db.read_conn(), REACH_EARLY, None, None)
.await
.expect("an intact hot log needs no archive");
assert!(
truth.concepts.contains_key("c1"),
"the fixture must have something at {REACH_EARLY} for the archive to take"
);
db.archive(REACH_CUTOFF).await.unwrap();
let floor: String = db
.read_conn()
.query("SELECT MIN(recorded_at) FROM transaction_log", ())
.await
.unwrap()
.next()
.await
.unwrap()
.unwrap()
.get(0)
.unwrap();
assert!(
floor.as_str() <= REACH_EARLY,
"the hot log must still stretch back past {REACH_EARLY} for this to be the case \
a `MIN(recorded_at)` reach test gets wrong; floor is {floor}"
);
let err = reconstruct(db.read_conn(), REACH_EARLY, None, None)
.await
.expect_err("the log reaches back to the instant but no longer completes it");
assert!(
matches!(err, macrame::DbError::ReplayCorrupt { .. }),
"got {err:?}"
);
assert!(
err.to_string().contains("no archive path was given"),
"the refusal must name the missing ingredient: {err}"
);
assert_eq!(
db.reconstruct(REACH_EARLY)
.await
.expect("reconstruct takes the archive path and must answer")
.concepts
.get("c1"),
truth.concepts.get("c1"),
"the archive moved the row, it did not change what was believed"
);
db.close().await.unwrap();
}
const HORIZON_VALID_FROM: &str = "1970-01-01T00:00:00.000000Z";
const HORIZON_INSTANT: &str = "1970-01-01T00:30:00.000000Z";
const HORIZON_CUTOFF: &str = "1970-01-01T01:30:00.000000Z";
async fn a_ledger_archived_across_a_horizon(harness: &TestHarness) -> macrame::Database {
use macrame::prelude::*;
let db = harness.db_with_fake_clock().await;
for title in ["first", "second", "third"] {
db.upsert_concept(ConceptUpsert::new("c1", title).valid_from(HORIZON_VALID_FROM))
.await
.unwrap();
harness.advance(std::time::Duration::from_secs(3_600));
}
db
}
#[tokio::test]
async fn hydrating_at_a_recorded_instant_refuses_once_rows_have_been_archived() {
const INSTANT: &str = HORIZON_INSTANT;
let harness = TestHarness::new();
let db = a_ledger_archived_across_a_horizon(&harness).await;
let ids = vec!["c1".to_string()];
let before = hydrate_attributes(
db.read_conn(),
&ids,
&AsOf::recorded_at(INSTANT),
AttributeMode::AtTime,
)
.await
.expect("an intact hot log answers for any instant");
assert_eq!(
before.iter().map(|a| a.title.as_str()).collect::<Vec<_>>(),
vec!["first"],
"at {INSTANT} the ledger held the first generation"
);
let report = db.archive(HORIZON_CUTOFF).await.unwrap();
assert!(
report.log_entries_archived >= 2,
"the fixture needs the superseded rows to have actually moved: {report:?}"
);
let err = hydrate_attributes(
db.read_conn(),
&ids,
&AsOf::recorded_at(INSTANT),
AttributeMode::AtTime,
)
.await
.expect_err("a short hot log must refuse rather than return a shorter Vec");
match &err {
macrame::DbError::RecordedInstantUnreachable { ts } => assert_eq!(ts, INSTANT),
other => panic!("got {other:?}"),
}
assert!(err.to_string().contains("reconstruct"), "{err}");
for as_of in [AsOf::now(), AsOf::valid_at(INSTANT)] {
let live = hydrate_attributes(db.read_conn(), &ids, &as_of, AttributeMode::AtTime)
.await
.expect("the live arms do not read the log and must be unaffected");
assert_eq!(
live.iter().map(|a| a.title.as_str()).collect::<Vec<_>>(),
vec!["third"],
"live text, whichever instant the valid axis names"
);
}
db.close().await.unwrap();
}
#[tokio::test]
async fn what_the_hot_log_refuses_the_archive_path_still_answers() {
let harness = TestHarness::new();
let db = a_ledger_archived_across_a_horizon(&harness).await;
let ids = vec!["c1".to_string()];
let truth = hydrate_attributes(
db.read_conn(),
&ids,
&AsOf::recorded_at(HORIZON_INSTANT),
AttributeMode::AtTime,
)
.await
.expect("an intact hot log answers for any instant")
.pop()
.expect("the concept existed at the instant");
let before = reconstruct(db.read_conn(), HORIZON_INSTANT, None, None)
.await
.expect("an intact hot log needs no archive");
assert_eq!(
before.concepts.get("c1"),
Some(&truth),
"the two readers must already agree before the archive splits them"
);
let report = db.archive(HORIZON_CUTOFF).await.unwrap();
assert!(
report.log_entries_archived >= 2,
"the fixture needs the superseded rows to have actually moved: {report:?}"
);
let err = hydrate_attributes(
db.read_conn(),
&ids,
&AsOf::recorded_at(HORIZON_INSTANT),
AttributeMode::AtTime,
)
.await
.expect_err("the instant is past the horizon now");
assert!(err.to_string().contains("reconstruct"), "{err}");
let after = db
.reconstruct(HORIZON_INSTANT)
.await
.expect("reconstruct takes the archive path and must answer");
assert_eq!(
after.concepts.get("c1"),
Some(&truth),
"the archive moved the row, it did not change what was believed"
);
let err = reconstruct(db.read_conn(), HORIZON_INSTANT, None, None)
.await
.expect_err("a short hot log and no archive path is not answerable");
assert!(
matches!(err, macrame::DbError::ReplayCorrupt { .. }),
"got {err:?}"
);
db.close().await.unwrap();
}