diff --git a/core/docs/architecture/84-global-sqlite-store.md b/core/docs/architecture/84-global-sqlite-store.md index d320f7109..590818fa5 100644 --- a/core/docs/architecture/84-global-sqlite-store.md +++ b/core/docs/architecture/84-global-sqlite-store.md @@ -80,6 +80,21 @@ adding a table or index is a schema-version change with an additive migration | Accounting | `store_totals` (kept by `commit_logical` and `layer_logical` triggers), `store_policy`, `scope_allocator`, `history_meta` | logical sizes, the frozen storage policy, inode-serial allocation, schema metadata | | Hints and rings | `binding_hint` (#352, additive), `content_signature` (bounded ring, ≤ 8,192 rows) | parent bindings for O(changed) diffs; similarity signatures for delta bases | | Init | `init_entry`, `init_host_file`, `init_operation` | Project Init's resumable records | +| Change export | `change_state`, `change_log` (additive; AFTER INSERT/UPDATE/DELETE triggers on the replicated tables) | the export cursor and the key of every row written since the Store was marked as an export base | + +**Change export** (`replication/`) [impl]. The triggers log only while +`change_state` has its row, so a Store that was never marked with `mark_base` +logs nothing. `export_changes` reads one snapshot of a live Store and writes the +rows changed after a cursor into a bundle file; it takes no writer and stops no +writer. It writes the bundle under a temporary name and links it into place, so +the bundle path names a complete file or nothing. `apply_changes` installs a +bundle into an offline replica in one transaction, and refuses a bundle whose +meta, starting state, digest or ending state does not match. A bundle is a +transfer file, not a second database: the Store never reads it, and the caller +deletes it once it is applied. `mark_base` and `apply_changes` checkpoint after +they commit; when another connection's read blocks the checkpoint, they report +`NotCheckpointed`, and the Store file holds the change only together with its +write-ahead log. ## 5. Objects inside the file diff --git a/engine/crates/layerfs-persistence/Cargo.toml b/engine/crates/layerfs-persistence/Cargo.toml index 747630e70..f744a102d 100644 --- a/engine/crates/layerfs-persistence/Cargo.toml +++ b/engine/crates/layerfs-persistence/Cargo.toml @@ -11,10 +11,8 @@ layerfs-storage = { path = "../../../core/crates/layerfs-storage" } layerfs-history = { path = "../../../core/crates/layerfs-history", default-features = false } layerfs-content = { path = "../../../core/crates/layerfs-content" } layerfs-telemetry = { path = "../../../core/crates/layerfs-telemetry" } -rusqlite = { version = "=0.40.2", features = ["cache", "limits", "blob"] } - -[dev-dependencies] blake3 = { version = "=1.8.5", default-features = false } +rusqlite = { version = "=0.40.2", features = ["cache", "limits", "blob"] } [target.'cfg(any(target_os = "macos", target_os = "linux"))'.dependencies] nix = { version = "=0.31.3", default-features = false, features = ["fs"] } diff --git a/engine/crates/layerfs-persistence/sql/sqlite/change_log.sql b/engine/crates/layerfs-persistence/sql/sqlite/change_log.sql new file mode 100644 index 000000000..5d7881138 --- /dev/null +++ b/engine/crates/layerfs-persistence/sql/sqlite/change_log.sql @@ -0,0 +1,134 @@ +CREATE TABLE change_state ( + id INTEGER PRIMARY KEY CHECK(id=1), + state BLOB NOT NULL CHECK(length(state)=32), + seq INTEGER NOT NULL CHECK(seq>=0) +) STRICT; +CREATE TABLE change_log ( + seq INTEGER PRIMARY KEY CHECK(seq>0), + tbl INTEGER NOT NULL CHECK(tbl BETWEEN 0 AND 19), + k1 ANY, + k2 ANY, + k3 ANY, + CHECK((tbl=0)=(k1 IS NULL)) +) STRICT; +CREATE TRIGGER change_pack_insert AFTER INSERT ON pack WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(1,new.pack_id); END; +CREATE TRIGGER change_object_location_insert AFTER INSERT ON object_location WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(3,new.object_id); END; +CREATE TRIGGER change_object_location_update AFTER UPDATE ON object_location WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(3,new.object_id); + INSERT INTO change_log(tbl,k1) SELECT 3,old.object_id WHERE old.object_id IS NOT new.object_id; END; +CREATE TRIGGER change_object_location_delete AFTER DELETE ON object_location WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(3,old.object_id); END; +CREATE TRIGGER change_inode_value_insert AFTER INSERT ON inode_value WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(4,new.ordinal); END; +CREATE TRIGGER change_inode_value_update AFTER UPDATE ON inode_value WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(4,new.ordinal); + INSERT INTO change_log(tbl,k1) SELECT 4,old.ordinal WHERE old.ordinal IS NOT new.ordinal; END; +CREATE TRIGGER change_inode_value_delete AFTER DELETE ON inode_value WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(4,old.ordinal); END; +CREATE TRIGGER change_content_signature_insert AFTER INSERT ON content_signature WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(5,new.slot); END; +CREATE TRIGGER change_content_signature_update AFTER UPDATE ON content_signature WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(5,new.slot); + INSERT INTO change_log(tbl,k1) SELECT 5,old.slot WHERE old.slot IS NOT new.slot; END; +CREATE TRIGGER change_content_signature_delete AFTER DELETE ON content_signature WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(5,old.slot); END; +CREATE TRIGGER change_store_policy_insert AFTER INSERT ON store_policy WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(6,new.id); END; +CREATE TRIGGER change_store_policy_update AFTER UPDATE ON store_policy WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(6,new.id); + INSERT INTO change_log(tbl,k1) SELECT 6,old.id WHERE old.id IS NOT new.id; END; +CREATE TRIGGER change_store_policy_delete AFTER DELETE ON store_policy WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(6,old.id); END; +CREATE TRIGGER change_store_totals_insert AFTER INSERT ON store_totals WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(7,new.id); END; +CREATE TRIGGER change_store_totals_update AFTER UPDATE ON store_totals WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(7,new.id); + INSERT INTO change_log(tbl,k1) SELECT 7,old.id WHERE old.id IS NOT new.id; END; +CREATE TRIGGER change_store_totals_delete AFTER DELETE ON store_totals WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(7,old.id); END; +CREATE TRIGGER change_history_meta_insert AFTER INSERT ON history_meta WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(8,new.id); END; +CREATE TRIGGER change_history_meta_update AFTER UPDATE ON history_meta WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(8,new.id); + INSERT INTO change_log(tbl,k1) SELECT 8,old.id WHERE old.id IS NOT new.id; END; +CREATE TRIGGER change_history_meta_delete AFTER DELETE ON history_meta WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(8,old.id); END; +CREATE TRIGGER change_layer_stack_insert AFTER INSERT ON layer_stack WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(9,new.layer_stack_id); END; +CREATE TRIGGER change_layer_stack_update AFTER UPDATE ON layer_stack WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(9,new.layer_stack_id); + INSERT INTO change_log(tbl,k1) SELECT 9,old.layer_stack_id WHERE old.layer_stack_id IS NOT new.layer_stack_id; END; +CREATE TRIGGER change_layer_stack_delete AFTER DELETE ON layer_stack WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(9,old.layer_stack_id); END; +CREATE TRIGGER change_layer_insert AFTER INSERT ON layer WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(10,new.layer_id); END; +CREATE TRIGGER change_layer_update AFTER UPDATE ON layer WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(10,new.layer_id); + INSERT INTO change_log(tbl,k1) SELECT 10,old.layer_id WHERE old.layer_id IS NOT new.layer_id; END; +CREATE TRIGGER change_layer_delete AFTER DELETE ON layer WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(10,old.layer_id); END; +CREATE TRIGGER change_commit_insert AFTER INSERT ON "commit" WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(11,new.commit_id); END; +CREATE TRIGGER change_commit_update AFTER UPDATE ON "commit" WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(11,new.commit_id); + INSERT INTO change_log(tbl,k1) SELECT 11,old.commit_id WHERE old.commit_id IS NOT new.commit_id; END; +CREATE TRIGGER change_commit_delete AFTER DELETE ON "commit" WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(11,old.commit_id); END; +CREATE TRIGGER change_branch_insert AFTER INSERT ON branch WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(12,new.branch_id); END; +CREATE TRIGGER change_branch_update AFTER UPDATE ON branch WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(12,new.branch_id); + INSERT INTO change_log(tbl,k1) SELECT 12,old.branch_id WHERE old.branch_id IS NOT new.branch_id; END; +CREATE TRIGGER change_branch_delete AFTER DELETE ON branch WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(12,old.branch_id); END; +CREATE TRIGGER change_scope_allocator_insert AFTER INSERT ON scope_allocator WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(13,new.scope_id); END; +CREATE TRIGGER change_scope_allocator_update AFTER UPDATE ON scope_allocator WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(13,new.scope_id); + INSERT INTO change_log(tbl,k1) SELECT 13,old.scope_id WHERE old.scope_id IS NOT new.scope_id; END; +CREATE TRIGGER change_scope_allocator_delete AFTER DELETE ON scope_allocator WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(13,old.scope_id); END; +CREATE TRIGGER change_binding_hint_insert AFTER INSERT ON binding_hint WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1,k2,k3) VALUES(14,new.child,new.parent,new.name); END; +CREATE TRIGGER change_binding_hint_update AFTER UPDATE ON binding_hint WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1,k2,k3) VALUES(14,new.child,new.parent,new.name); + INSERT INTO change_log(tbl,k1,k2,k3) SELECT 14,old.child,old.parent,old.name WHERE old.child IS NOT new.child OR old.parent IS NOT new.parent OR old.name IS NOT new.name; END; +CREATE TRIGGER change_binding_hint_delete AFTER DELETE ON binding_hint WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1,k2,k3) VALUES(14,old.child,old.parent,old.name); END; +CREATE TRIGGER change_layer_lineage_insert AFTER INSERT ON layer_lineage WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(15,new.layer_id); END; +CREATE TRIGGER change_layer_lineage_update AFTER UPDATE ON layer_lineage WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(15,new.layer_id); + INSERT INTO change_log(tbl,k1) SELECT 15,old.layer_id WHERE old.layer_id IS NOT new.layer_id; END; +CREATE TRIGGER change_layer_lineage_delete AFTER DELETE ON layer_lineage WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(15,old.layer_id); END; +CREATE TRIGGER change_commit_lineage_insert AFTER INSERT ON commit_lineage WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(16,new.commit_id); END; +CREATE TRIGGER change_commit_lineage_update AFTER UPDATE ON commit_lineage WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(16,new.commit_id); + INSERT INTO change_log(tbl,k1) SELECT 16,old.commit_id WHERE old.commit_id IS NOT new.commit_id; END; +CREATE TRIGGER change_commit_lineage_delete AFTER DELETE ON commit_lineage WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(16,old.commit_id); END; +CREATE TRIGGER change_absorbed_insert AFTER INSERT ON absorbed WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1,k2) VALUES(17,new.run,new.depth); END; +CREATE TRIGGER change_absorbed_update AFTER UPDATE ON absorbed WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1,k2) VALUES(17,new.run,new.depth); + INSERT INTO change_log(tbl,k1,k2) SELECT 17,old.run,old.depth WHERE old.run IS NOT new.run OR old.depth IS NOT new.depth; END; +CREATE TRIGGER change_absorbed_delete AFTER DELETE ON absorbed WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1,k2) VALUES(17,old.run,old.depth); END; +CREATE TRIGGER change_reach_insert AFTER INSERT ON reach WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1,k2) VALUES(18,new.key,new.x); END; +CREATE TRIGGER change_reach_update AFTER UPDATE ON reach WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1,k2) VALUES(18,new.key,new.x); + INSERT INTO change_log(tbl,k1,k2) SELECT 18,old.key,old.x WHERE old.key IS NOT new.key OR old.x IS NOT new.x; END; +CREATE TRIGGER change_reach_delete AFTER DELETE ON reach WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1,k2) VALUES(18,old.key,old.x); END; +CREATE TRIGGER change_file_digest_insert AFTER INSERT ON file_digest WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1,k2) VALUES(19,new.digest,new.length); END; +CREATE TRIGGER change_file_digest_update AFTER UPDATE ON file_digest WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1,k2) VALUES(19,new.digest,new.length); + INSERT INTO change_log(tbl,k1,k2) SELECT 19,old.digest,old.length WHERE old.digest IS NOT new.digest OR old.length IS NOT new.length; END; +CREATE TRIGGER change_file_digest_delete AFTER DELETE ON file_digest WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1,k2) VALUES(19,old.digest,old.length); END; diff --git a/engine/crates/layerfs-persistence/sql/sqlite/change_log_units.sql b/engine/crates/layerfs-persistence/sql/sqlite/change_log_units.sql new file mode 100644 index 000000000..73db9f4b1 --- /dev/null +++ b/engine/crates/layerfs-persistence/sql/sqlite/change_log_units.sql @@ -0,0 +1,2 @@ +CREATE TRIGGER change_pack_unit_insert AFTER INSERT ON pack_unit WHEN EXISTS(SELECT 1 FROM change_state) BEGIN + INSERT INTO change_log(tbl,k1) VALUES(2,new.unit_id); END; diff --git a/engine/crates/layerfs-persistence/src/backend/sqlite/mod.rs b/engine/crates/layerfs-persistence/src/backend/sqlite/mod.rs index fce18a22d..c143f0a35 100644 --- a/engine/crates/layerfs-persistence/src/backend/sqlite/mod.rs +++ b/engine/crates/layerfs-persistence/src/backend/sqlite/mod.rs @@ -16,7 +16,7 @@ pub(crate) mod metadata_values; pub(crate) mod objects_read; pub(crate) mod objects_selection; pub(crate) mod prepared; -mod profile; +pub(crate) mod profile; pub(crate) mod publish; pub(crate) mod query; pub(crate) mod reclamation; diff --git a/engine/crates/layerfs-persistence/src/backend/sqlite/schema.rs b/engine/crates/layerfs-persistence/src/backend/sqlite/schema.rs index 88fb63097..0c3165207 100644 --- a/engine/crates/layerfs-persistence/src/backend/sqlite/schema.rs +++ b/engine/crates/layerfs-persistence/src/backend/sqlite/schema.rs @@ -13,10 +13,22 @@ pub(crate) const LINEAGE: &str = include_str!("../../../sql/sqlite/lineage.sql") pub(crate) const SCRATCH: &str = include_str!("../../../sql/sqlite/merge_scratch.sql"); /// Whole-content file digests (#364); a Store written before them gains them at open. pub(crate) const DIGESTS: &str = include_str!("../../../sql/sqlite/file_digest.sql"); +/// Change log for incremental export; a Store written before it gains it at open. +pub(crate) const CHANGES: &str = include_str!("../../../sql/sqlite/change_log.sql"); +/// Change log trigger for the group-row layouts' `pack_unit` table. +pub(crate) const CHANGE_UNITS: &str = include_str!("../../../sql/sqlite/change_log_units.sql"); const UNIT_INDEX: &str = include_str!("../../../sql/sqlite/objects_units_index.sql"); const UNITS: &str = include_str!("../../../sql/sqlite/objects_units.sql"); const ACQUISITION: &str = include_str!("../../../sql/sqlite/acquisition/schema.sql"); type Selection = (crate::SqlitePackLayout, crate::SqliteAcquisitionSchema); +/// The change log script for `layout`: its tables and every logging trigger. +fn changes(layout: crate::SqlitePackLayout) -> String { + if layout.uses_units() { + format!("{CHANGES}{CHANGE_UNITS}") + } else { + CHANGES.to_owned() + } +} fn scripts((layout, acquisition): Selection) -> Vec { let objects = match layout { crate::SqlitePackLayout::Monolithic => OBJECTS.to_owned(), @@ -32,6 +44,7 @@ fn scripts((layout, acquisition): Selection) -> Vec { LINEAGE.to_owned(), SCRATCH.to_owned(), DIGESTS.to_owned(), + changes(layout), ]; if acquisition == crate::SqliteAcquisitionSchema::Tables { scripts.push(ACQUISITION.to_owned()); @@ -41,6 +54,10 @@ fn scripts((layout, acquisition): Selection) -> Vec { pub(crate) fn source(selection: Selection) -> String { scripts(selection).concat() } +/// The recorded schema of a Store written before the change log existed. +fn before_changes(selection: Selection) -> String { + source(selection).replacen(&changes(selection.0), "", 1) +} pub(crate) fn definition(c: &Connection, w: &RefCell) -> Result { let rows: Vec = query::run( c, @@ -85,6 +102,8 @@ pub(crate) fn check( "absorbed", "binding_hint", "branch", + "change_log", + "change_state", "commit", "commit_lineage", "content_signature", @@ -166,8 +185,8 @@ impl super::connection::Session { return Ok(()); }; // The schema this migration records is the one before the - // lineage and merge tables; later migrations add those. - let current = source((self.layout, self.acquisition)) + // lineage, merge, digest and change tables; later migrations add those. + let current = before_changes((self.layout, self.acquisition)) .replacen(DIGESTS, "", 1) .replacen(SCRATCH, "", 1) .replacen(LINEAGE, "", 1); @@ -214,7 +233,7 @@ impl super::connection::Session { let Some(row) = rows.first() else { return Ok(()); }; - let current = source((self.layout, self.acquisition)) + let current = before_changes((self.layout, self.acquisition)) .replacen(DIGESTS, "", 1) .replacen(SCRATCH, "", 1); if row.get::(0)? != current.replacen(LINEAGE, "", 1) @@ -251,7 +270,7 @@ impl super::connection::Session { let Some(row) = rows.first() else { return Ok(()); }; - let current = source((self.layout, self.acquisition)).replacen(DIGESTS, "", 1); + let current = before_changes((self.layout, self.acquisition)).replacen(DIGESTS, "", 1); if row.get::(0)? != current.replacen(SCRATCH, "", 1) || row.get::(1)? != definition(tx.connection, tx.work)? { @@ -286,7 +305,7 @@ impl super::connection::Session { let Some(row) = rows.first() else { return Ok(()); }; - let current = source((self.layout, self.acquisition)); + let current = before_changes((self.layout, self.acquisition)); if row.get::(0)? != current.replacen(DIGESTS, "", 1) || row.get::(1)? != definition(tx.connection, tx.work)? { @@ -301,6 +320,41 @@ impl super::connection::Session { Ok(()) }) } + /// Adds the change log tables and triggers to a Store written before them + /// and records the current schema: an additive migration (REBUILD.md kind + /// 2) under the same conditions as [`Self::add_binding_hints`]. Logging + /// stays off until the Store is marked as an export base. + pub(crate) fn add_change_log(&self) -> Result<(), BackendError> { + self.run::<_, BackendError>(true, |tx| { + let present = tx.query( + "SELECT 1 FROM sqlite_schema WHERE type='table' AND name='change_log'", + vec![], + )?; + if !present.is_empty() { + return Ok(()); + } + let rows = tx.query( + "SELECT schema_source,schema_definition FROM history_meta WHERE id=1", + vec![], + )?; + let Some(row) = rows.first() else { + return Ok(()); + }; + let selection = (self.layout, self.acquisition); + if row.get::(0)? != before_changes(selection) + || row.get::(1)? != definition(tx.connection, tx.work)? + { + return Ok(()); + } + tx.bootstrap(&changes(self.layout))?; + let definition = definition(tx.connection, tx.work)?; + tx.query( + "UPDATE history_meta SET schema_source=?1,schema_definition=?2 WHERE id=1", + vec![Param::Text(source(selection)), Param::Text(definition)], + )?; + Ok(()) + }) + } pub(crate) fn validate( &self, catalog_id: layerfs_history::CatalogId, diff --git a/engine/crates/layerfs-persistence/src/lib.rs b/engine/crates/layerfs-persistence/src/lib.rs index 9a2553b4e..bccd392c4 100644 --- a/engine/crates/layerfs-persistence/src/lib.rs +++ b/engine/crates/layerfs-persistence/src/lib.rs @@ -7,6 +7,7 @@ mod store; pub(crate) use store::config; pub(crate) use store::handles; mod history; +pub mod replication; pub use backend::sqlite::connection::{ConnectionProfile, SqlWork}; #[cfg(any(target_os = "macos", target_os = "linux"))] pub use backend::sqlite::headroom::STORE_HEADROOM_BYTES; diff --git a/engine/crates/layerfs-persistence/src/replication/apply.rs b/engine/crates/layerfs-persistence/src/replication/apply.rs new file mode 100644 index 000000000..e42aa2788 --- /dev/null +++ b/engine/crates/layerfs-persistence/src/replication/apply.rs @@ -0,0 +1,212 @@ +//! Installing a bundle into an offline replica: parse its meta, verify it +//! against the replica, then install it. +use super::connection::{finish, open_store}; +use super::digest::{bundle_digest, next_state}; +use super::lookup::{catalog_of, last_seq, state_of, table_exists}; +use super::tables::{key_columns, quoted, BUNDLE_FORMAT, LOGGED, TOTALS}; +use super::{ApplyReceipt, ChangeCursor, ChangeError}; +use rusqlite::{params, Connection, OpenFlags}; +use std::path::Path; + +/// Installs the bundle at `bundle` into the offline replica at `store`, which +/// must be at the bundle's starting cursor and unwritten since. Verifies the +/// bundle's digest and state, installs every row and deletion in one +/// transaction with foreign keys checked at commit, and records the bundle's +/// ending cursor. A refused bundle leaves the replica unchanged. +/// +/// When another connection reads the replica, the bundle still commits but the +/// call returns [`ChangeError::NotCheckpointed`]. +pub fn apply_changes(store: &Path, bundle: &Path) -> Result { + let c = open_store(store, OpenFlags::SQLITE_OPEN_READ_WRITE)?; + c.execute_batch("PRAGMA foreign_keys=ON")?; + let name = bundle + .to_str() + .ok_or(ChangeError::Integrity("the bundle path is not UTF-8"))?; + c.execute("ATTACH DATABASE ?1 AS b", [name])?; + c.execute_batch("BEGIN IMMEDIATE; PRAGMA defer_foreign_keys=ON;")?; + let applied = apply_in(&c); + let finished = finish(&c, applied); + let _ = c.execute_batch("DETACH DATABASE b"); + finished +} + +fn apply_in(c: &Connection) -> Result { + let catalog = catalog_of(c, "main")?; + let meta = read_meta(c, &catalog)?; + verify(c, &catalog, &meta)?; + install(c, &meta)?; + Ok(ApplyReceipt { + before: meta.before, + after: meta.after, + rows: meta.rows, + deletes: meta.deletes, + }) +} + +/// The `bundle_meta` row of an attached bundle, parsed. +struct BundleMeta { + before: ChangeCursor, + after: ChangeCursor, + rows: u64, + deletes: u64, + digest: [u8; 32], +} + +/// Parses the attached bundle's one `bundle_meta` row. A bundle comes from +/// outside the replica, so its own constraints are not trusted: the row count, +/// format, catalog, value types and lengths are checked before a value is +/// read, and the sequences after. +fn read_meta(c: &Connection, catalog: &[u8]) -> Result { + let (count, format, same_catalog, shaped): (i64, bool, bool, bool) = c.query_row( + "SELECT count(*), coalesce(min(format IS ?1),0), coalesce(min(catalog_id IS ?2),0), \ + coalesce(min(typeof(before_state)='blob' AND length(before_state)=32 \ + AND typeof(after_state)='blob' AND length(after_state)=32 \ + AND typeof(digest)='blob' AND length(digest)=32 \ + AND typeof(before_seq)='integer' AND typeof(after_seq)='integer' \ + AND typeof(rows)='integer' AND typeof(deletes)='integer'),0) \ + FROM b.bundle_meta", + params![BUNDLE_FORMAT, catalog], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)), + )?; + if count != 1 { + return Err(ChangeError::Integrity("the bundle meta is not one row")); + } + if !format { + return Err(ChangeError::Integrity("unknown bundle format")); + } + if !same_catalog { + return Err(ChangeError::Integrity( + "the bundle belongs to another catalog", + )); + } + if !shaped { + return Err(ChangeError::Integrity("the bundle meta is malformed")); + } + let (before_state, before_seq, after_state, after_seq, rows, deletes, digest) = c.query_row( + "SELECT before_state,before_seq,after_state,after_seq,rows,deletes,digest \ + FROM b.bundle_meta", + [], + |r| { + let number = |i| r.get::<_, i64>(i); + Ok(( + r.get(0)?, + number(1)?, + r.get(2)?, + number(3)?, + number(4)?, + number(5)?, + r.get(6)?, + )) + }, + )?; + let count = |n: i64| { + u64::try_from(n).map_err(|_| ChangeError::Integrity("the bundle meta is malformed")) + }; + let before = ChangeCursor { + state: before_state, + seq: count(before_seq)?, + }; + let after = ChangeCursor { + state: after_state, + seq: count(after_seq)?, + }; + if after.seq <= before.seq { + return Err(ChangeError::Integrity( + "the bundle's sequences are out of order", + )); + } + Ok(BundleMeta { + before, + after, + rows: count(rows)?, + deletes: count(deletes)?, + digest, + }) +} + +/// Refuses a bundle that does not apply to this replica or is not intact, +/// before anything is installed. +fn verify(c: &Connection, catalog: &[u8], meta: &BundleMeta) -> Result<(), ChangeError> { + let current = state_of(c, "main")?.ok_or(ChangeError::NotBase)?; + if current != meta.before { + return Err(ChangeError::Cursor( + "the replica is not at the bundle's starting state", + )); + } + if last_seq(c, "main")?.is_some_and(|seq| seq != meta.before.seq) { + return Err(ChangeError::Diverged); + } + for t in LOGGED.iter().filter(|t| t.immutable) { + let deleted: bool = c.query_row( + "SELECT EXISTS(SELECT 1 FROM b.bundle_delete WHERE tbl=?1)", + [t.code], + |r| r.get(0), + )?; + if deleted { + return Err(ChangeError::Integrity("a bundle deletes an immutable row")); + } + } + let (computed, rows, deletes) = bundle_digest(c, "b")?; + if computed != meta.digest || rows != meta.rows || deletes != meta.deletes { + return Err(ChangeError::Integrity("bundle digest")); + } + if next_state(catalog, &meta.before, meta.after.seq, &computed) != meta.after.state { + return Err(ChangeError::Integrity("bundle ending state")); + } + Ok(()) +} + +/// Installs a verified bundle's rows and deletions and records its ending +/// cursor. +fn install(c: &Connection, meta: &BundleMeta) -> Result<(), ChangeError> { + // Pause logging for the installed rows: every trigger fires only while + // `change_state` has its row. Logging them and deleting the log afterwards + // would leave the replica with free pages as large as the log (about + // 0.6 MB per 14k objects). + c.execute("DELETE FROM main.change_state", [])?; + let order = LOGGED + .iter() + .filter(|t| t.code != TOTALS) + .chain(LOGGED.iter().filter(|t| t.code == TOTALS)); + for t in order { + if !table_exists(c, "b", t.table)? { + continue; + } + if !table_exists(c, "main", t.table)? { + return Err(ChangeError::Integrity( + "a bundled table is absent from the replica", + )); + } + let table = quoted(t.table); + if !t.immutable { + let key = t.keys.join(","); + let columns = key_columns(t.keys.len()); + c.execute( + &format!("DELETE FROM main.{table} WHERE ({key}) IN (SELECT {key} FROM b.{table})"), + [], + )?; + c.execute( + &format!( + "DELETE FROM main.{table} WHERE ({key}) IN \ + (SELECT {columns} FROM b.bundle_delete WHERE tbl=?1)" + ), + [t.code], + )?; + } + c.execute( + &format!("INSERT INTO main.{table} SELECT * FROM b.{table}"), + [], + )?; + } + // Resume logging; the replica now stands at `after`. + c.execute("DELETE FROM main.change_log", [])?; + c.execute( + "INSERT INTO main.change_log(seq,tbl) VALUES(?1,0)", + [meta.after.seq as i64], + )?; + c.execute( + "INSERT INTO main.change_state(id,state,seq) VALUES(1,?1,?2)", + params![&meta.after.state[..], meta.after.seq as i64], + )?; + Ok(()) +} diff --git a/engine/crates/layerfs-persistence/src/replication/base.rs b/engine/crates/layerfs-persistence/src/replication/base.rs new file mode 100644 index 000000000..96ee389eb --- /dev/null +++ b/engine/crates/layerfs-persistence/src/replication/base.rs @@ -0,0 +1,51 @@ +//! Marking an export base and reading a Store's change state. +use super::connection::{finish, open_store}; +use super::digest::base_state; +use super::lookup::{catalog_of, last_seq, state_of, table_exists}; +use super::{ChangeCursor, ChangeError}; +use rusqlite::{params, OpenFlags}; +use std::path::Path; + +/// Makes the offline Store at `path` the start of an export chain and returns +/// its cursor. Clears the change log down to one sequence marker and enables +/// logging. Marking again starts a new chain at the current sequence. `token` +/// makes the base identity unique; supply fresh random bytes. +/// +/// A copy of the closed Store file is then a replica at the returned cursor. +/// When another connection reads the Store, the mark still commits but returns +/// [`ChangeError::NotCheckpointed`], and the file alone is not yet that replica. +pub fn mark_base(path: &Path, token: [u8; 32]) -> Result { + let c = open_store(path, OpenFlags::SQLITE_OPEN_READ_WRITE)?; + c.execute_batch("BEGIN IMMEDIATE")?; + let marked = (|| { + if !table_exists(&c, "main", "change_log")? { + return Err(ChangeError::Integrity( + "the Store has no change log; open it writable once to migrate it", + )); + } + let catalog = catalog_of(&c, "main")?; + let recorded = state_of(&c, "main")?.map_or(0, |s| s.seq); + let seq = last_seq(&c, "main")?.unwrap_or(0).max(recorded); + c.execute("DELETE FROM change_log", [])?; + if seq > 0 { + c.execute("INSERT INTO change_log(seq,tbl) VALUES(?1,0)", [seq as i64])?; + } + let state = base_state(&catalog, seq, &token); + c.execute( + "INSERT INTO change_state(id,state,seq) VALUES(1,?1,?2) \ + ON CONFLICT(id) DO UPDATE SET state=excluded.state,seq=excluded.seq", + params![&state[..], seq as i64], + )?; + Ok(ChangeCursor { state, seq }) + })(); + finish(&c, marked) +} + +/// The change state recorded in the Store at `path`, if it is an export base. +pub fn read_state(path: &Path) -> Result, ChangeError> { + let c = open_store(path, OpenFlags::SQLITE_OPEN_READ_ONLY)?; + if !table_exists(&c, "main", "change_state")? { + return Ok(None); + } + state_of(&c, "main") +} diff --git a/engine/crates/layerfs-persistence/src/replication/connection.rs b/engine/crates/layerfs-persistence/src/replication/connection.rs new file mode 100644 index 000000000..6285bd330 --- /dev/null +++ b/engine/crates/layerfs-persistence/src/replication/connection.rs @@ -0,0 +1,61 @@ +//! Opening the Store and a bundle, and ending a write transaction. +use super::ChangeError; +use rusqlite::{Connection, OpenFlags}; +use std::path::Path; + +/// Opens `path` with no busy handler, so a locked file is refused at once: one +/// attempt, never a wait (rusqlite installs a 5 s busy handler by default). +pub(super) fn open(path: &Path, flags: OpenFlags) -> Result { + let c = Connection::open_with_flags(path, flags | OpenFlags::SQLITE_OPEN_NO_MUTEX)?; + c.busy_timeout(std::time::Duration::ZERO)?; + Ok(c) +} + +/// Opens the Store at `path` with the Store's own connection settings under +/// the Disposable profile, the only one that executes: WAL, synchronous OFF. +/// A Store that is not in WAL mode is refused. +pub(super) fn open_store(path: &Path, flags: OpenFlags) -> Result { + let c = open(path, flags)?; + let work = std::cell::RefCell::new(crate::SqlWork::default()); + // The acquisition schema is read only when a Store is created. + crate::backend::sqlite::profile::apply( + &c, + false, + crate::SqlitePersistenceProfile::Disposable, + crate::SqliteAcquisitionSchema::Absent, + &work, + ) + .map_err(|error| ChangeError::Profile(error.into()))?; + Ok(c) +} + +/// Commits `outcome`'s transaction or rolls it back, then checkpoints so the +/// Store file alone holds every change. A checkpoint that another reader +/// blocks, or that fails, is reported as [`ChangeError::NotCheckpointed`], +/// because the change has already committed. +pub(super) fn finish(c: &Connection, outcome: Result) -> Result { + let value = match outcome { + Ok(value) => value, + Err(error) => { + let _ = c.execute_batch("ROLLBACK"); + return Err(error); + } + }; + if let Err(error) = c.execute_batch("COMMIT") { + let _ = c.execute_batch("ROLLBACK"); + return Err(match error.sqlite_error_code() { + Some(rusqlite::ErrorCode::ConstraintViolation) => { + ChangeError::Integrity("a foreign key is unsatisfied at commit") + } + _ => error.into(), + }); + } + // The first column is 1 when a reader kept the checkpoint from completing. + match c.query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |r| { + r.get::<_, i64>(0) + }) { + Ok(0) => Ok(value), + Ok(_) => Err(ChangeError::NotCheckpointed(None)), + Err(error) => Err(ChangeError::NotCheckpointed(Some(error))), + } +} diff --git a/engine/crates/layerfs-persistence/src/replication/digest.rs b/engine/crates/layerfs-persistence/src/replication/digest.rs new file mode 100644 index 000000000..47b76a60d --- /dev/null +++ b/engine/crates/layerfs-persistence/src/replication/digest.rs @@ -0,0 +1,105 @@ +//! State identities and the bundle digest. +use super::lookup::table_exists; +use super::tables::{key_columns, quoted, LOGGED}; +use super::{ChangeCursor, ChangeError}; +use rusqlite::{types::ValueRef, Connection}; + +pub(super) fn base_state(catalog: &[u8], seq: u64, token: &[u8; 32]) -> [u8; 32] { + let mut h = blake3::Hasher::new(); + h.update(b"layerfs-change-base-v1\0"); + h.update(&(catalog.len() as u64).to_le_bytes()); + h.update(catalog); + h.update(&seq.to_le_bytes()); + h.update(token); + h.finalize().into() +} + +pub(super) fn next_state( + catalog: &[u8], + before: &ChangeCursor, + after_seq: u64, + digest: &[u8; 32], +) -> [u8; 32] { + let mut h = blake3::Hasher::new(); + h.update(b"layerfs-change-v1\0"); + h.update(&(catalog.len() as u64).to_le_bytes()); + h.update(catalog); + h.update(&before.state); + h.update(&before.seq.to_le_bytes()); + h.update(&after_seq.to_le_bytes()); + h.update(digest); + h.finalize().into() +} + +/// Feeds one value with its type tag and length. +fn feed(h: &mut blake3::Hasher, value: ValueRef<'_>) { + match value { + ValueRef::Null => { + h.update(&[0]); + } + ValueRef::Integer(i) => { + h.update(&[1]); + h.update(&i.to_le_bytes()); + } + ValueRef::Real(r) => { + h.update(&[2]); + h.update(&r.to_bits().to_le_bytes()); + } + ValueRef::Text(t) => { + h.update(&[3]); + h.update(&(t.len() as u64).to_le_bytes()); + h.update(t); + } + ValueRef::Blob(b) => { + h.update(&[4]); + h.update(&(b.len() as u64).to_le_bytes()); + h.update(b); + } + } +} + +/// The digest of the bundle tables in `schema`: for each logged table in code +/// order, its rows ordered by primary key, then its deleted keys in order. +/// Export and apply both compute it from the bundle file itself. +pub(super) fn bundle_digest( + c: &Connection, + schema: &str, +) -> Result<([u8; 32], u64, u64), ChangeError> { + let mut h = blake3::Hasher::new(); + h.update(b"layerfs-change-bundle-v1\0"); + let (mut rows, mut deletes) = (0u64, 0u64); + for t in &LOGGED { + if !table_exists(c, schema, t.table)? { + continue; + } + h.update(&t.code.to_le_bytes()); + let mut stmt = c.prepare(&format!( + "SELECT * FROM {schema}.{} ORDER BY {}", + quoted(t.table), + t.keys.join(",") + ))?; + let width = stmt.column_count(); + let mut cursor = stmt.query([])?; + while let Some(row) = cursor.next()? { + h.update(b"r"); + h.update(&(width as u32).to_le_bytes()); + for i in 0..width { + feed(&mut h, row.get_ref(i)?); + } + rows += 1; + } + let columns = key_columns(t.keys.len()); + let mut stmt = c.prepare(&format!( + "SELECT {columns} FROM {schema}.bundle_delete WHERE tbl=?1 ORDER BY {columns}" + ))?; + let mut cursor = stmt.query([t.code])?; + while let Some(row) = cursor.next()? { + h.update(b"d"); + for i in 0..t.keys.len() { + feed(&mut h, row.get_ref(i)?); + } + deletes += 1; + } + } + Ok((h.finalize().into(), rows, deletes)) +} diff --git a/engine/crates/layerfs-persistence/src/replication/error.rs b/engine/crates/layerfs-persistence/src/replication/error.rs new file mode 100644 index 000000000..d23b58ee9 --- /dev/null +++ b/engine/crates/layerfs-persistence/src/replication/error.rs @@ -0,0 +1,74 @@ +//! Why an export, apply or base mark was refused or failed. +use layerfs_storage::port::PersistenceError; +use std::{error::Error, fmt, io}; + +/// Why an export, apply or base mark was refused or failed. +#[derive(Debug)] +pub enum ChangeError { + /// The Store is not an export base: it records no change state. + NotBase, + /// The cursor does not name a state this Store or bundle can use. + Cursor(&'static str), + /// The replica was written after its recorded state. + Diverged, + /// The Store or bundle is not consistent. + Integrity(&'static str), + /// The bundle path already exists. + Exists, + /// The Store refused its connection settings: it is not in WAL mode + /// (`Malformed`), another connection holds it (`Busy`), or SQLite refused + /// a setting. + Profile(PersistenceError), + /// Unlike every other variant, the change committed. SQLite then did not + /// copy the write-ahead log into the Store file: another connection was + /// reading the Store, or the checkpoint failed (the source). Until a later + /// checkpoint, the Store file holds the change only together with its + /// write-ahead log. + NotCheckpointed(Option), + /// SQLite failed. + Sqlite(rusqlite::Error), + /// The filesystem failed. + Io(io::Error), +} + +impl fmt::Display for ChangeError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::NotBase => write!(f, "the Store is not an export base"), + Self::Cursor(why) => write!(f, "cursor refused: {why}"), + Self::Diverged => write!(f, "the replica was written after its recorded state"), + Self::Integrity(why) => write!(f, "integrity: {why}"), + Self::Exists => write!(f, "the bundle path already exists"), + Self::Profile(_) => write!(f, "the Store refused its connection settings"), + Self::NotCheckpointed(_) => write!( + f, + "the change committed, but the Store file holds it only with its write-ahead log" + ), + Self::Sqlite(_) => write!(f, "SQLite failed"), + Self::Io(_) => write!(f, "the filesystem failed"), + } + } +} + +impl Error for ChangeError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + Self::Profile(error) => Some(error), + Self::NotCheckpointed(Some(error)) | Self::Sqlite(error) => Some(error), + Self::Io(error) => Some(error), + _ => None, + } + } +} + +impl From for ChangeError { + fn from(error: rusqlite::Error) -> Self { + Self::Sqlite(error) + } +} + +impl From for ChangeError { + fn from(error: io::Error) -> Self { + Self::Io(error) + } +} diff --git a/engine/crates/layerfs-persistence/src/replication/export.rs b/engine/crates/layerfs-persistence/src/replication/export.rs new file mode 100644 index 000000000..826171387 --- /dev/null +++ b/engine/crates/layerfs-persistence/src/replication/export.rs @@ -0,0 +1,227 @@ +//! Exporting a live Store's changed rows into a new bundle file. +use super::connection::{open, open_store}; +use super::digest::{bundle_digest, next_state}; +use super::lookup::{catalog_of, last_seq, state_of, table_exists}; +use super::tables::{key_columns, quoted, BUNDLE_FORMAT, BUNDLE_SCHEMA, LOGGED}; +use super::{ChangeCursor, ChangeError, ExportReceipt}; +use rusqlite::{params, Connection, OpenFlags}; +use std::ffi::OsString; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::{fs, io}; + +/// Writes every row of the live Store at `store` changed after `since` into a +/// new bundle file at `bundle`, from one read snapshot. Takes no write lock +/// and does not stop writers. When nothing changed, writes no file and +/// returns `after == before`. +/// +/// `since` is the replica's cursor: the one `mark_base` returned, or the +/// `after` of the bundle the replica applied last. Above the base, the Store +/// checks the cursor's sequence, not its state, because it keeps no record of +/// the states it exported. A cursor from a separately written copy of this +/// Store is therefore not detected here. +/// +/// The bundle is written under a temporary name beside `bundle` and linked +/// into place, so `bundle` names a complete file or nothing. A failed or +/// interrupted export can leave a `..-.partial` file there, +/// which nothing reads. +pub fn export_changes( + store: &Path, + since: ChangeCursor, + bundle: &Path, +) -> Result { + let source = open_store(store, OpenFlags::SQLITE_OPEN_READ_ONLY)?; + source.execute_batch("BEGIN")?; + let exported = export_in(&source, since, bundle); + // Ends the read snapshot; a read transaction has nothing to commit. + let _ = source.execute_batch("COMMIT"); + exported +} + +fn export_in( + source: &Connection, + since: ChangeCursor, + bundle: &Path, +) -> Result { + let base = state_of(source, "main")?.ok_or(ChangeError::NotBase)?; + if since.seq < base.seq || (since.seq == base.seq && since.state != base.state) { + return Err(ChangeError::Cursor( + "the cursor precedes or differs from the Store's recorded state", + )); + } + let last = last_seq(source, "main")?.unwrap_or(0).max(base.seq); + if since.seq > last { + return Err(ChangeError::Cursor("the cursor is ahead of the Store")); + } + if since.seq == last { + return Ok(ExportReceipt { + before: since, + after: since, + rows: 0, + deletes: 0, + log_entries: 0, + bytes: 0, + digest: [0; 32], + }); + } + // Refuses an existing path before the work; the link below decides. + if bundle.try_exists()? { + return Err(ChangeError::Exists); + } + let catalog = catalog_of(source, "main")?; + let log_entries: i64 = source.query_row( + "SELECT count(*) FROM change_log WHERE tbl>0 AND seq>?1 AND seq<=?2", + [since.seq as i64, last as i64], + |r| r.get(0), + )?; + let partial = partial_path(bundle)?; + fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&partial)?; + let published = write_bundle(source, since, last, &catalog, &partial).and_then(|written| { + let bytes = fs::metadata(&partial)?.len(); + match fs::hard_link(&partial, bundle) { + Ok(()) => Ok((written, bytes)), + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => Err(ChangeError::Exists), + Err(error) => Err(error.into()), + } + }); + // The outcome above is the result: the bundle is published or the export + // failed. A temporary file that cannot be removed is never read. + let _ = fs::remove_file(&partial); + let ((after, digest, rows, deletes), bytes) = published?; + Ok(ExportReceipt { + before: since, + after, + rows, + deletes, + log_entries: log_entries as u64, + bytes, + digest, + }) +} + +/// `..-.partial` beside `bundle`, unique within this process. +fn partial_path(bundle: &Path) -> Result { + static NEXT: AtomicU64 = AtomicU64::new(0); + let name = bundle + .file_name() + .ok_or(ChangeError::Integrity("the bundle path names no file"))?; + let mut partial = OsString::from("."); + partial.push(name); + partial.push(format!( + ".{}-{}.partial", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + Ok(bundle.with_file_name(partial)) +} + +fn write_bundle( + source: &Connection, + since: ChangeCursor, + last: u64, + catalog: &[u8], + bundle: &Path, +) -> Result<(ChangeCursor, [u8; 32], u64, u64), ChangeError> { + let out = open(bundle, OpenFlags::SQLITE_OPEN_READ_WRITE)?; + // The bundle holds copies of some rows; the replica checks references + // when the bundle is applied. The bundled SQLite enables them by default. + // Small pages keep a bundle of a few rows small: every table it carries + // takes at least one page. Large payloads lose 4 bytes per overflow page. + out.execute_batch( + "PRAGMA page_size=1024; PRAGMA foreign_keys=OFF; PRAGMA journal_mode=OFF; \ + PRAGMA synchronous=OFF; BEGIN;", + )?; + out.execute_batch(BUNDLE_SCHEMA)?; + let mut logged = source + .prepare("SELECT EXISTS(SELECT 1 FROM change_log WHERE tbl=?1 AND seq>?2 AND seq<=?3)")?; + for t in &LOGGED { + if !table_exists(source, "main", t.table)? { + continue; + } + // A table carries its changed rows and its deleted keys; a table + // with neither is left out of the bundle. + let changed: bool = + logged.query_row(params![t.code, since.seq as i64, last as i64], |r| r.get(0))?; + if !changed { + continue; + } + let definition: String = source.query_row( + "SELECT sql FROM sqlite_schema WHERE type='table' AND name=?1", + [t.table], + |r| r.get(0), + )?; + out.execute_batch(&definition)?; + let columns = key_columns(t.keys.len()); + let join = t + .keys + .iter() + .enumerate() + .map(|(i, key)| format!("r.{key}=c.k{}", i + 1)) + .collect::>() + .join(" AND "); + let mut read = source.prepare(&format!( + "SELECT {columns},(r.{} IS NOT NULL),r.* FROM \ + (SELECT DISTINCT {columns} FROM change_log WHERE tbl=?1 AND seq>?2 AND seq<=?3) c \ + LEFT JOIN {} r ON {join}", + t.keys[0], + quoted(t.table) + ))?; + let offset = t.keys.len() + 1; + let width = read.column_count() - offset; + let mut insert = out.prepare(&format!( + "INSERT INTO {} VALUES({})", + quoted(t.table), + vec!["?"; width].join(",") + ))?; + let mut delete = out.prepare(&format!( + "INSERT INTO bundle_delete(tbl,{columns}) VALUES(?1,{})", + (2..=t.keys.len() + 1) + .map(|i| format!("?{i}")) + .collect::>() + .join(",") + ))?; + let mut rows = read.query([t.code, since.seq as i64, last as i64])?; + while let Some(row) = rows.next()? { + let present: bool = row.get(t.keys.len())?; + if present { + let values = (offset..offset + width) + .map(|i| row.get_ref(i).map(rusqlite::types::ToSqlOutput::Borrowed)) + .collect::, _>>()?; + insert.execute(rusqlite::params_from_iter(values))?; + } else if t.immutable { + return Err(ChangeError::Integrity("a logged immutable row is absent")); + } else { + let mut values = vec![rusqlite::types::Value::Integer(t.code)]; + for i in 0..t.keys.len() { + values.push(row.get(i)?); + } + delete.execute(rusqlite::params_from_iter(values))?; + } + } + } + let (digest, rows, deletes) = bundle_digest(&out, "main")?; + let after = ChangeCursor { + state: next_state(catalog, &since, last, &digest), + seq: last, + }; + out.execute( + "INSERT INTO bundle_meta(id,format,catalog_id,before_state,before_seq,after_state,after_seq,rows,deletes,digest) \ + VALUES(1,?1,?2,?3,?4,?5,?6,?7,?8,?9)", + params![ + BUNDLE_FORMAT, + catalog, + &since.state[..], + since.seq as i64, + &after.state[..], + after.seq as i64, + rows as i64, + deletes as i64, + &digest[..] + ], + )?; + out.execute_batch("COMMIT")?; + Ok((after, digest, rows, deletes)) +} diff --git a/engine/crates/layerfs-persistence/src/replication/lookup.rs b/engine/crates/layerfs-persistence/src/replication/lookup.rs new file mode 100644 index 000000000..e589d571f --- /dev/null +++ b/engine/crates/layerfs-persistence/src/replication/lookup.rs @@ -0,0 +1,50 @@ +//! Reads of one attached schema: `main` (the Store) or `b` (a bundle). +use super::{ChangeCursor, ChangeError}; +use rusqlite::{Connection, OptionalExtension}; + +pub(super) fn table_exists(c: &Connection, schema: &str, name: &str) -> Result { + Ok(c.query_row( + &format!("SELECT 1 FROM {schema}.sqlite_schema WHERE type='table' AND name=?1"), + [name], + |_| Ok(()), + ) + .optional()? + .is_some()) +} + +pub(super) fn state_of(c: &Connection, schema: &str) -> Result, ChangeError> { + let row: Option<(Vec, i64)> = c + .query_row( + &format!("SELECT state,seq FROM {schema}.change_state WHERE id=1"), + [], + |r| Ok((r.get(0)?, r.get(1)?)), + ) + .optional()?; + row.map(|(state, seq)| { + Ok(ChangeCursor { + state: state + .try_into() + .map_err(|_| ChangeError::Integrity("change state length"))?, + seq: u64::try_from(seq).map_err(|_| ChangeError::Integrity("change state sequence"))?, + }) + }) + .transpose() +} + +pub(super) fn catalog_of(c: &Connection, schema: &str) -> Result, ChangeError> { + Ok(c.query_row( + &format!("SELECT catalog_id FROM {schema}.history_meta WHERE id=1"), + [], + |r| r.get(0), + )?) +} + +pub(super) fn last_seq(c: &Connection, schema: &str) -> Result, ChangeError> { + let seq: Option = c.query_row( + &format!("SELECT max(seq) FROM {schema}.change_log"), + [], + |r| r.get(0), + )?; + seq.map(|s| u64::try_from(s).map_err(|_| ChangeError::Integrity("change log sequence"))) + .transpose() +} diff --git a/engine/crates/layerfs-persistence/src/replication/mod.rs b/engine/crates/layerfs-persistence/src/replication/mod.rs new file mode 100644 index 000000000..add444868 --- /dev/null +++ b/engine/crates/layerfs-persistence/src/replication/mod.rs @@ -0,0 +1,50 @@ +//! Incremental change export between replicas of one Store. +//! +//! A Store marked as an export base logs the primary key of every row its +//! writers insert, update or delete in the logged tables. Schema triggers fill +//! `change_log`, so every opener's writes are logged without a code path of +//! its own, in the single writer's commit order. A Store that was never marked +//! logs nothing. +//! +//! [`export_changes`] reads one snapshot of a live Store and writes every row +//! changed after a cursor, at its current value, into a new bundle file: a +//! SQLite file with the logged tables' exact definitions, the deleted keys and +//! the bundle's two cursors. [`apply_changes`] installs a bundle into an +//! offline replica at the bundle's starting cursor, in one transaction. +//! [`mark_base`] makes an offline Store (normally a sealed file) the start of a +//! chain. +//! +//! A cursor names one replica state: a 32-byte state identity and the change +//! sequence it reaches. Each bundle's ending state derives from its starting +//! state and the digest of its contents, so two replicas that diverge from one +//! state never share a later state, and a bundle applies only to the state it +//! was exported from. The digest and states detect accidental damage and +//! misordering; they are not a signature, so a caller that moves bundles over +//! an untrusted path authenticates them itself. +//! +//! An export writes the bundle under a temporary name beside the bundle path +//! and links it into place, so the path names a complete bundle or nothing, +//! also after a process crash. Like the Store under the Disposable profile, a +//! bundle claims process-crash survival only. +//! +//! [`mark_base`] and [`apply_changes`] checkpoint the Store after they commit, +//! so that the Store file alone holds the change. When another connection's +//! read blocks that checkpoint, they return [`ChangeError::NotCheckpointed`]: +//! the change committed, and the file holds it only together with its +//! write-ahead log. +mod apply; +mod base; +mod connection; +mod digest; +mod error; +mod export; +mod lookup; +mod records; +mod tables; + +pub use apply::apply_changes; +pub use base::{mark_base, read_state}; +pub use error::ChangeError; +pub use export::export_changes; +pub use records::{ApplyReceipt, ChangeCursor, ExportReceipt}; +pub use tables::BUNDLE_FORMAT; diff --git a/engine/crates/layerfs-persistence/src/replication/records.rs b/engine/crates/layerfs-persistence/src/replication/records.rs new file mode 100644 index 000000000..e1100aebc --- /dev/null +++ b/engine/crates/layerfs-persistence/src/replication/records.rs @@ -0,0 +1,43 @@ +//! The cursor that names a replica state and the receipts of an export and an +//! apply. + +/// One replica state: its identity and the change sequence it reaches. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ChangeCursor { + /// State identity. + pub state: [u8; 32], + /// Last change sequence this state includes. + pub seq: u64, +} + +/// What one export read and wrote. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ExportReceipt { + /// The cursor the bundle starts from. + pub before: ChangeCursor, + /// The cursor the bundle reaches; equal to `before` when nothing changed. + pub after: ChangeCursor, + /// Rows written at their current value. + pub rows: u64, + /// Keys written as deleted. + pub deletes: u64, + /// Change log entries read in the cursor range. + pub log_entries: u64, + /// Bundle file length; 0 when nothing changed and no file was written. + pub bytes: u64, + /// Digest of the bundle's rows and deletions. + pub digest: [u8; 32], +} + +/// What one apply installed. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ApplyReceipt { + /// The replica's cursor before the bundle. + pub before: ChangeCursor, + /// The replica's cursor after the bundle. + pub after: ChangeCursor, + /// Rows installed. + pub rows: u64, + /// Keys deleted. + pub deletes: u64, +} diff --git a/engine/crates/layerfs-persistence/src/replication/tables.rs b/engine/crates/layerfs-persistence/src/replication/tables.rs new file mode 100644 index 000000000..e3058bb68 --- /dev/null +++ b/engine/crates/layerfs-persistence/src/replication/tables.rs @@ -0,0 +1,171 @@ +//! The logged tables and the bundle's own tables: both are part of the bundle +//! format. + +/// Bundle format written and read by this module. +pub const BUNDLE_FORMAT: i64 = 1; + +/// One logged table: its stable code, name and primary key. +pub(super) struct Logged { + pub(super) code: i64, + pub(super) table: &'static str, + pub(super) keys: &'static [&'static str], + /// Rows are never updated or deleted (published packs). + pub(super) immutable: bool, +} + +/// Logged tables in code order. Codes are part of the bundle format and of +/// `change_log.sql`; code 0 is the sequence marker. +pub(super) const LOGGED: [Logged; 19] = [ + Logged { + code: 1, + table: "pack", + keys: &["pack_id"], + immutable: true, + }, + Logged { + code: 2, + table: "pack_unit", + keys: &["unit_id"], + immutable: true, + }, + Logged { + code: 3, + table: "object_location", + keys: &["object_id"], + immutable: false, + }, + Logged { + code: 4, + table: "inode_value", + keys: &["ordinal"], + immutable: false, + }, + Logged { + code: 5, + table: "content_signature", + keys: &["slot"], + immutable: false, + }, + Logged { + code: 6, + table: "store_policy", + keys: &["id"], + immutable: false, + }, + Logged { + code: 7, + table: "store_totals", + keys: &["id"], + immutable: false, + }, + Logged { + code: 8, + table: "history_meta", + keys: &["id"], + immutable: false, + }, + Logged { + code: 9, + table: "layer_stack", + keys: &["layer_stack_id"], + immutable: false, + }, + Logged { + code: 10, + table: "layer", + keys: &["layer_id"], + immutable: false, + }, + Logged { + code: 11, + table: "commit", + keys: &["commit_id"], + immutable: false, + }, + Logged { + code: 12, + table: "branch", + keys: &["branch_id"], + immutable: false, + }, + Logged { + code: 13, + table: "scope_allocator", + keys: &["scope_id"], + immutable: false, + }, + Logged { + code: 14, + table: "binding_hint", + keys: &["child", "parent", "name"], + immutable: false, + }, + Logged { + code: 15, + table: "layer_lineage", + keys: &["layer_id"], + immutable: false, + }, + Logged { + code: 16, + table: "commit_lineage", + keys: &["commit_id"], + immutable: false, + }, + Logged { + code: 17, + table: "absorbed", + keys: &["run", "depth"], + immutable: false, + }, + Logged { + code: 18, + table: "reach", + keys: &["key", "x"], + immutable: false, + }, + Logged { + code: 19, + table: "file_digest", + keys: &["digest", "length"], + immutable: false, + }, +]; + +/// `store_totals`' code. Inserting Commits and Layers adjusts the replica's +/// totals through its own triggers, so the bundled totals are installed last. +pub(super) const TOTALS: i64 = 7; + +/// The bundle's own tables, next to copies of the logged tables. +pub(super) const BUNDLE_SCHEMA: &str = " +CREATE TABLE bundle_meta ( + id INTEGER PRIMARY KEY CHECK(id=1), + format INTEGER NOT NULL, + catalog_id BLOB NOT NULL, + before_state BLOB NOT NULL CHECK(length(before_state)=32), + before_seq INTEGER NOT NULL CHECK(before_seq>=0), + after_state BLOB NOT NULL CHECK(length(after_state)=32), + after_seq INTEGER NOT NULL CHECK(after_seq>before_seq), + rows INTEGER NOT NULL CHECK(rows>=0), + deletes INTEGER NOT NULL CHECK(deletes>=0), + digest BLOB NOT NULL CHECK(length(digest)=32) +) STRICT; +CREATE TABLE bundle_delete ( + tbl INTEGER NOT NULL CHECK(tbl BETWEEN 1 AND 19), + k1 ANY NOT NULL, + k2 ANY, + k3 ANY +) STRICT; +"; + +pub(super) fn quoted(name: &str) -> String { + format!("\"{name}\"") +} + +/// `k1,...,kn` for a table's key width. +pub(super) fn key_columns(width: usize) -> String { + (1..=width) + .map(|i| format!("k{i}")) + .collect::>() + .join(",") +} diff --git a/engine/crates/layerfs-persistence/src/store/open.rs b/engine/crates/layerfs-persistence/src/store/open.rs index 2cb9e9827..8eb68a7dc 100644 --- a/engine/crates/layerfs-persistence/src/store/open.rs +++ b/engine/crates/layerfs-persistence/src/store/open.rs @@ -88,6 +88,7 @@ impl Handles { session.add_lineage(crate::history::backfill_lineage)?; session.add_merge_scratch()?; session.add_file_digests()?; + session.add_change_log()?; } let handles = Self::validated(session, directory, binding, cursor_key)?; if writable { diff --git a/engine/crates/layerfs-persistence/tests/change_export.rs b/engine/crates/layerfs-persistence/tests/change_export.rs new file mode 100644 index 000000000..530bdbb6a --- /dev/null +++ b/engine/crates/layerfs-persistence/tests/change_export.rs @@ -0,0 +1,653 @@ +//! Incremental change export: a replica built from a marked base and the +//! bundles exported from the live Store holds exactly the live Store's rows, +//! opens through the normal open path and answers the same history. Bundles +//! apply only in order, only to an unwritten replica, and only intact. +mod support; +use layerfs_content::{construct_bytes, ObjectId}; +use layerfs_history::records::{ + BindingHint, CommitMeta, CommitOutcome, CommitRequest, CommitVia, ForkRequest, ForkSource, + StackInitialization, +}; +use layerfs_history::{BranchId, HistoryCatalog, HistoryCatalogConfig}; +use layerfs_persistence::replication::{ + apply_changes, export_changes, mark_base, read_state, ChangeError, +}; +use layerfs_persistence::{Handles, PersistenceConfig, SqlitePackLayout, SqlitePersistenceProfile}; +use layerfs_storage::port::PersistenceError; +use layerfs_storage::{Storage, StoragePolicy}; +use layerfs_telemetry::timer::Timing; +use support::*; + +/// Every logged table; `pack_unit` exists only in the group-row layouts. +const TABLES: [&str; 19] = [ + "pack", + "pack_unit", + "object_location", + "inode_value", + "content_signature", + "store_policy", + "store_totals", + "history_meta", + "layer_stack", + "layer", + "commit", + "branch", + "scope_allocator", + "binding_hint", + "layer_lineage", + "commit_lineage", + "absorbed", + "reach", + "file_digest", +]; + +fn config(path: &std::path::Path, layout: SqlitePackLayout) -> PersistenceConfig { + PersistenceConfig::sqlite(path) + .with_sqlite_profile(SqlitePersistenceProfile::Disposable) + .with_sqlite_pack_layout(layout) +} + +fn created(path: &std::path::Path, layout: SqlitePackLayout) -> Handles { + Handles::create( + config(path, layout), + StoragePolicy::frozen_default(), + &HistoryCatalogConfig { + cursor_key: [71; 32], + binding_key: BINDING.to_vec(), + incarnation: 7, + }, + ) + .unwrap() +} + +fn writable(path: &std::path::Path) -> Handles { + writable_in(path, SqlitePackLayout::Monolithic) +} + +fn writable_in(path: &std::path::Path, layout: SqlitePackLayout) -> Handles { + Handles::open_writable(config(path, layout), BINDING, [71; 32]).unwrap() +} + +fn save(storage: &Storage, bytes: &[u8]) -> ObjectId { + let save = storage.begin_save().unwrap(); + let result = Timing::disabled("construct", |scope| { + construct_bytes( + storage.policy().construction(), + &storage.policy().construction().capacities(), + bytes, + &mut save.sink(), + scope.child("content"), + ) + }) + .0 + .unwrap(); + save.finish().unwrap(); + result.root +} + +fn commit(history: &dyn HistoryCatalog, branch: BranchId, tag: u8) { + let record = history.branch(branch).unwrap().unwrap(); + let current = match record.head_commit { + Some(head) => history.commit(head).unwrap().unwrap().root, + None => history.layer(record.base_layer).unwrap().unwrap().root, + }; + let outcome = history + .commit_branch(&CommitRequest { + branch, + expected_head: record.head_commit, + expected_base: record.base_layer, + expected_root: current, + candidate_root: root(tag), + profile: root(201), + scope: root(200), + second_parent: None, + meta: CommitMeta::new(CommitVia::Files), + force: false, + }) + .unwrap(); + assert!( + matches!(outcome, CommitOutcome::Committed { .. }), + "{outcome:?}" + ); +} + +/// Every row of every logged table the Store has, each rendered as text, +/// sorted. +fn rows(path: &std::path::Path) -> Vec<(&'static str, Vec)> { + let sql = + rusqlite::Connection::open_with_flags(path, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY) + .unwrap(); + let exists = |table: &str| { + sql.query_row( + "SELECT count(*) FROM sqlite_schema WHERE type='table' AND name=?1", + [table], + |r| r.get::<_, i64>(0), + ) + .unwrap() + == 1 + }; + TABLES + .iter() + .filter(|table| exists(table)) + .map(|table| { + let mut statement = sql.prepare(&format!("SELECT * FROM \"{table}\"")).unwrap(); + let width = statement.column_count(); + let mut out: Vec = statement + .query_map([], |r| { + (0..width) + .map(|i| r.get::<_, rusqlite::types::Value>(i)) + .collect::, _>>() + }) + .unwrap() + .map(|row| format!("{:?}", row.unwrap())) + .collect(); + out.sort(); + (*table, out) + }) + .collect() +} + +fn log_entries(path: &std::path::Path) -> i64 { + let sql = rusqlite::Connection::open(path).unwrap(); + sql.query_row("SELECT count(*) FROM change_log WHERE tbl>0", [], |r| { + r.get(0) + }) + .unwrap() +} + +/// Creates a Store with one LayerStack and one saved object, closes it and +/// returns the open-ready path; the base is not yet marked. +fn initialized(t: &Temp, file: &str) -> std::path::PathBuf { + initialized_in(t, file, SqlitePackLayout::Monolithic) +} + +fn initialized_in(t: &Temp, file: &str, layout: SqlitePackLayout) -> std::path::PathBuf { + let path = t.join(file); + let handles = created(&path, layout); + handles + .history + .initialize_layerstack(&StackInitialization { + logical: 0, + stack: stack(1), + name: name("main"), + scope: root(200), + profile: root(201), + genesis_root: root(0), + }) + .unwrap(); + save( + &Storage::new(handles.storage.clone()).unwrap(), + b"genesis bytes", + ); + drop(handles); + path +} + +#[test] +fn a_replica_from_the_base_and_its_bundles_holds_the_live_rows() { + let t = Temp::new("change-export"); + let live = initialized(&t, "live"); + let base = mark_base(&live, [5; 32]).unwrap(); + assert_eq!(read_state(&live).unwrap(), Some(base)); + assert_eq!( + log_entries(&live), + 0, + "marking leaves only the sequence marker" + ); + let replica = t.join("replica"); + std::fs::copy(&live, &replica).unwrap(); + + // The live Store keeps its writer open across both exports. + let handles = writable(&live); + let storage = Storage::new(handles.storage.clone()).unwrap(); + let history: &dyn HistoryCatalog = &handles.history; + let genesis = history.layer_stack(stack(1)).unwrap().unwrap().head_layer; + history + .fork(&ForkRequest { + stack: stack(1), + branch: branch_id(1), + name: name("b1"), + source: ForkSource::Layer(genesis), + }) + .unwrap(); + commit(history, branch_id(1), 1); + commit(history, branch_id(1), 2); + let small = save(&storage, b"small file"); + let large = save(&storage, &vec![7u8; 400_000]); + history + .record_bindings(&[BindingHint { + child: 7, + parent: 1, + name: b"a.txt".to_vec(), + }]) + .unwrap(); + let first = t.join("bundle-1"); + let one = export_changes(&live, base, &first).unwrap(); + assert_eq!(one.before, base); + assert!( + one.after.seq > base.seq && one.rows > 0 && one.bytes > 0, + "{one:?}" + ); + assert!(one.log_entries >= one.rows + one.deletes, "{one:?}"); + + commit(history, branch_id(1), 3); + let more = save(&storage, &vec![9u8; 300_000]); + let second = t.join("bundle-2"); + let two = export_changes(&live, one.after, &second).unwrap(); + assert_eq!(two.before, one.after); + assert!(two.after.seq > one.after.seq && two.rows > 0, "{two:?}"); + + // Nothing changed: no file, the same cursor. + let none = t.join("bundle-3"); + let three = export_changes(&live, two.after, &none).unwrap(); + assert_eq!((three.after, three.bytes), (two.after, 0)); + assert!(!none.exists()); + // A cursor ahead of the Store, or an existing bundle path, is refused. + let ahead = layerfs_persistence::replication::ChangeCursor { + seq: two.after.seq + 1, + ..two.after + }; + assert!(matches!( + export_changes(&live, ahead, &none), + Err(ChangeError::Cursor(_)) + )); + let kept = std::fs::read(&second).unwrap(); + assert!(matches!( + export_changes(&live, one.after, &second), + Err(ChangeError::Exists) + )); + assert_eq!(std::fs::read(&second).unwrap(), kept); + // A deletion alone: the bundle carries the table with its deleted key + // and no rows; no other table changed. + rusqlite::Connection::open(&live) + .unwrap() + .execute("DELETE FROM binding_hint WHERE child=7", []) + .unwrap(); + let third = t.join("bundle-4"); + let four = export_changes(&live, two.after, &third).unwrap(); + assert_eq!((four.rows, four.deletes), (0, 1), "{four:?}"); + assert_eq!(std::fs::metadata(&third).unwrap().len(), four.bytes); + // Every export linked its bundle into place and removed its temporary + // file. + let partial = std::fs::read_dir(t.join("")) + .unwrap() + .filter(|entry| { + entry + .as_ref() + .unwrap() + .file_name() + .to_string_lossy() + .ends_with(".partial") + }) + .count(); + assert_eq!(partial, 0); + drop(handles); + + // Out of order is refused; in order reaches the live rows exactly. + assert!(matches!( + apply_changes(&replica, &second), + Err(ChangeError::Cursor(_)) + )); + assert_eq!(apply_changes(&replica, &first).unwrap().after, one.after); + assert!(matches!( + apply_changes(&replica, &first), + Err(ChangeError::Cursor(_)) + )); + let applied = apply_changes(&replica, &second).unwrap(); + assert_eq!((applied.before, applied.after), (one.after, two.after)); + apply_changes(&replica, &third).unwrap(); + assert_eq!(rows(&replica), rows(&live)); + assert_eq!(read_state(&replica).unwrap(), Some(four.after)); + + // The replica opens through the normal path and answers the same. + let opened = writable(&replica); + let reader = Storage::new(opened.storage.clone()).unwrap(); + let reader = reader.reader().unwrap(); + assert_eq!( + reader.file_lengths(&[small, large, more]).unwrap(), + [10, 400_000, 300_000] + ); + let source = writable(&live); + assert_eq!( + opened.history.branch(branch_id(1)).unwrap(), + source.history.branch(branch_id(1)).unwrap() + ); + assert_eq!( + opened.history.bindings_of(7).unwrap(), + source.history.bindings_of(7).unwrap() + ); +} + +#[test] +fn applying_a_bundle_logs_nothing_and_leaves_no_free_pages_on_the_replica() { + let t = Temp::new("change-export-quiet"); + let live = initialized(&t, "live"); + let base = mark_base(&live, [6; 32]).unwrap(); + let replica = t.join("replica"); + std::fs::copy(&live, &replica).unwrap(); + let handles = writable(&live); + let storage = Storage::new(handles.storage.clone()).unwrap(); + // Enough rows for the log to span many pages. + for i in 0..1500u32 { + save(&storage, format!("object {i}").as_bytes()); + } + drop(handles); + let bundle = t.join("bundle"); + let one = export_changes(&live, base, &bundle).unwrap(); + assert!(one.log_entries >= 1500, "{one:?}"); + apply_changes(&replica, &bundle).unwrap(); + assert_eq!(rows(&replica), rows(&live)); + assert_eq!(read_state(&replica).unwrap(), Some(one.after)); + let sql = rusqlite::Connection::open(&replica).unwrap(); + let count = |q: &str| -> i64 { sql.query_row(q, [], |r| r.get(0)).unwrap() }; + assert_eq!( + count("SELECT count(*) FROM change_log"), + 1, + "only the marker" + ); + assert_eq!(count("PRAGMA freelist_count"), 0, "no pages freed by a log"); +} + +#[test] +fn a_locked_replica_is_refused_at_once_and_left_unchanged() { + let t = Temp::new("change-locked"); + let live = initialized(&t, "live"); + let base = mark_base(&live, [6; 32]).unwrap(); + let replica = t.join("replica"); + std::fs::copy(&live, &replica).unwrap(); + { + let handles = writable(&live); + save(&Storage::new(handles.storage.clone()).unwrap(), b"locked"); + } + let bundle = t.join("bundle"); + export_changes(&live, base, &bundle).unwrap(); + // Another connection holds the replica's write lock. + let holder = rusqlite::Connection::open(&replica).unwrap(); + holder.execute_batch("BEGIN IMMEDIATE").unwrap(); + let started = std::time::Instant::now(); + let refused = apply_changes(&replica, &bundle); + let waited = started.elapsed(); + assert!( + matches!(&refused, Err(ChangeError::Sqlite(e)) + if e.sqlite_error_code() == Some(rusqlite::ErrorCode::DatabaseBusy)), + "{refused:?}" + ); + // One attempt: rusqlite's default busy handler would wait 5 s first. + assert!(waited < std::time::Duration::from_secs(1), "{waited:?}"); + holder.execute_batch("ROLLBACK").unwrap(); + drop(holder); + assert_eq!(read_state(&replica).unwrap(), Some(base)); + apply_changes(&replica, &bundle).unwrap(); + assert_eq!(rows(&replica), rows(&live)); +} + +#[test] +fn a_replica_outside_the_disposable_profile_is_refused_and_left_unchanged() { + let t = Temp::new("change-profile"); + let live = initialized(&t, "live"); + let base = mark_base(&live, [6; 32]).unwrap(); + let replica = t.join("replica"); + std::fs::copy(&live, &replica).unwrap(); + { + let handles = writable(&live); + save(&Storage::new(handles.storage.clone()).unwrap(), b"profile"); + } + let bundle = t.join("bundle"); + export_changes(&live, base, &bundle).unwrap(); + let journal = |mode: &str| { + rusqlite::Connection::open(&replica) + .unwrap() + .query_row(&format!("PRAGMA journal_mode={mode}"), [], |r| { + r.get::<_, String>(0) + }) + .unwrap() + }; + assert_eq!(journal("DELETE"), "delete"); + let refused = apply_changes(&replica, &bundle); + assert!( + matches!( + &refused, + Err(ChangeError::Profile(PersistenceError::Malformed)) + ), + "{refused:?}" + ); + let source = std::error::Error::source(refused.as_ref().unwrap_err()); + assert_eq!( + source.map(ToString::to_string), + Some(PersistenceError::Malformed.to_string()) + ); + assert_eq!(journal("WAL"), "wal"); + assert_eq!(read_state(&replica).unwrap(), Some(base)); + apply_changes(&replica, &bundle).unwrap(); + assert_eq!(rows(&replica), rows(&live)); +} + +#[test] +fn a_written_replica_and_a_changed_bundle_are_refused() { + let t = Temp::new("change-refusal"); + let live = initialized(&t, "live"); + let base = mark_base(&live, [6; 32]).unwrap(); + let replica = t.join("replica"); + std::fs::copy(&live, &replica).unwrap(); + { + let handles = writable(&live); + let genesis = handles + .history + .layer_stack(stack(1)) + .unwrap() + .unwrap() + .head_layer; + handles + .history + .fork(&ForkRequest { + stack: stack(1), + branch: branch_id(1), + name: name("b1"), + source: ForkSource::Layer(genesis), + }) + .unwrap(); + commit(&handles.history, branch_id(1), 1); + } + let bundle = t.join("bundle"); + export_changes(&live, base, &bundle).unwrap(); + + // A changed bundle row fails the digest; nothing is installed. + let changed = t.join("changed"); + std::fs::copy(&bundle, &changed).unwrap(); + rusqlite::Connection::open(&changed) + .unwrap() + .execute("UPDATE store_totals SET logical=logical+1", []) + .unwrap(); + assert!(matches!( + apply_changes(&replica, &changed), + Err(ChangeError::Integrity("bundle digest")) + )); + assert_eq!(read_state(&replica).unwrap(), Some(base)); + + // A replica written after its state has diverged from the chain. + { + let handles = writable(&replica); + handles + .history + .record_bindings(&[BindingHint { + child: 9, + parent: 1, + name: b"local".to_vec(), + }]) + .unwrap(); + } + assert!(matches!( + apply_changes(&replica, &bundle), + Err(ChangeError::Diverged) + )); +} + +#[test] +fn an_unmarked_store_logs_nothing_and_exports_nothing() { + let t = Temp::new("change-unmarked"); + let live = initialized(&t, "live"); + { + let handles = writable(&live); + save( + &Storage::new(handles.storage.clone()).unwrap(), + &vec![3u8; 200_000], + ); + } + assert_eq!(log_entries(&live), 0); + assert_eq!(read_state(&live).unwrap(), None); + let cursor = layerfs_persistence::replication::ChangeCursor { + state: [0; 32], + seq: 0, + }; + assert!(matches!( + export_changes(&live, cursor, &t.join("bundle")), + Err(ChangeError::NotBase) + )); +} + +#[test] +fn a_base_marked_under_a_reader_commits_and_reports_the_blocked_checkpoint() { + let t = Temp::new("change-reader"); + let live = initialized(&t, "live"); + // A reader's snapshot keeps the checkpoint from copying the mark into the + // Store file. + let reader = rusqlite::Connection::open(&live).unwrap(); + reader.execute_batch("BEGIN").unwrap(); + let _: i64 = reader + .query_row("SELECT count(*) FROM history_meta", [], |r| r.get(0)) + .unwrap(); + let marked = mark_base(&live, [8; 32]); + assert!( + matches!(marked, Err(ChangeError::NotCheckpointed(None))), + "{marked:?}" + ); + reader.execute_batch("COMMIT").unwrap(); + drop(reader); + // The mark committed; without a reader, marking again checkpoints, and a + // copy of the file alone is a replica. + assert!(read_state(&live).unwrap().is_some()); + let base = mark_base(&live, [9; 32]).unwrap(); + let replica = t.join("replica"); + std::fs::copy(&live, &replica).unwrap(); + assert_eq!(read_state(&replica).unwrap(), Some(base)); +} + +/// Rebuilds `bundle_meta` without its constraints, so the cases below reach +/// apply's own checks. +const UNCHECKED_META: &str = "CREATE TABLE m AS SELECT * FROM bundle_meta; \ + DROP TABLE bundle_meta; ALTER TABLE m RENAME TO bundle_meta;"; + +#[test] +fn a_refused_bundle_leaves_the_replica_unchanged() { + let t = Temp::new("change-refused"); + let live = initialized(&t, "live"); + let unmarked = t.join("unmarked"); + std::fs::copy(&live, &unmarked).unwrap(); + let base = mark_base(&live, [6; 32]).unwrap(); + let replica = t.join("replica"); + std::fs::copy(&live, &replica).unwrap(); + { + let handles = writable(&live); + save(&Storage::new(handles.storage.clone()).unwrap(), b"refused"); + } + let bundle = t.join("bundle"); + export_changes(&live, base, &bundle).unwrap(); + + let before = rows(&unmarked); + let refused = apply_changes(&unmarked, &bundle); + assert!(matches!(refused, Err(ChangeError::NotBase)), "{refused:?}"); + assert_eq!(rows(&unmarked), before); + assert_eq!(read_state(&unmarked).unwrap(), None); + + let cases = [ + ("UPDATE bundle_meta SET format=2", "unknown bundle format"), + ( + "UPDATE bundle_meta SET catalog_id=x'00'", + "the bundle belongs to another catalog", + ), + ( + "UPDATE bundle_meta SET after_state=zeroblob(32)", + "bundle ending state", + ), + ( + "INSERT INTO bundle_delete(tbl,k1) VALUES(1,1)", + "a bundle deletes an immutable row", + ), + ( + "INSERT INTO bundle_meta SELECT * FROM bundle_meta", + "the bundle meta is not one row", + ), + ( + "UPDATE bundle_meta SET after_seq=before_seq", + "the bundle's sequences are out of order", + ), + ( + "UPDATE bundle_meta SET before_seq=-1", + "the bundle meta is malformed", + ), + ( + "UPDATE bundle_meta SET digest=x'00'", + "the bundle meta is malformed", + ), + ]; + let before = rows(&replica); + for (i, (change, expected)) in cases.into_iter().enumerate() { + let changed = t.join(&format!("changed-{i}")); + std::fs::copy(&bundle, &changed).unwrap(); + rusqlite::Connection::open(&changed) + .unwrap() + .execute_batch(&format!("{UNCHECKED_META} {change};")) + .unwrap(); + let refused = apply_changes(&replica, &changed); + assert!( + matches!(&refused, Err(ChangeError::Integrity(why)) if *why == expected), + "{change}: {refused:?}" + ); + assert_eq!(rows(&replica), before, "{change}"); + assert_eq!(read_state(&replica).unwrap(), Some(base), "{change}"); + } + apply_changes(&replica, &bundle).unwrap(); + assert_eq!(rows(&replica), rows(&live)); +} + +#[test] +fn every_pack_layout_replicates_its_pack_rows() { + for layout in [ + SqlitePackLayout::Monolithic, + SqlitePackLayout::GroupRows, + SqlitePackLayout::GroupRowsIndexed, + ] { + let t = Temp::new("change-layout"); + let live = initialized_in(&t, "live", layout); + let base = mark_base(&live, [6; 32]).unwrap(); + let replica = t.join("replica"); + std::fs::copy(&live, &replica).unwrap(); + let (small, large) = { + let handles = writable_in(&live, layout); + let storage = Storage::new(handles.storage.clone()).unwrap(); + ( + save(&storage, b"small"), + save(&storage, &vec![5u8; 400_000]), + ) + }; + let bundle = t.join("bundle"); + export_changes(&live, base, &bundle).unwrap(); + apply_changes(&replica, &bundle).unwrap(); + let live_rows = rows(&live); + let units = live_rows.iter().find(|(table, _)| *table == "pack_unit"); + assert_eq!( + units.is_some_and(|(_, units)| !units.is_empty()), + layout != SqlitePackLayout::Monolithic, + "{layout:?}" + ); + assert_eq!(rows(&replica), live_rows, "{layout:?}"); + let opened = writable_in(&replica, layout); + let reader = Storage::new(opened.storage.clone()).unwrap(); + let reader = reader.reader().unwrap(); + assert_eq!( + reader.file_lengths(&[small, large]).unwrap(), + [5, 400_000], + "{layout:?}" + ); + } +} diff --git a/engine/crates/layerfs-persistence/tests/lineage_backfill.rs b/engine/crates/layerfs-persistence/tests/lineage_backfill.rs index 7fe9465b2..c5a90536b 100644 --- a/engine/crates/layerfs-persistence/tests/lineage_backfill.rs +++ b/engine/crates/layerfs-persistence/tests/lineage_backfill.rs @@ -13,6 +13,26 @@ use support::*; const LINEAGE: &str = include_str!("../sql/sqlite/lineage.sql"); const SCRATCH: &str = include_str!("../sql/sqlite/merge_scratch.sql"); const DIGESTS: &str = include_str!("../sql/sqlite/file_digest.sql"); +const CHANGES: &str = include_str!("../sql/sqlite/change_log.sql"); +const CHANGE_UNITS: &str = include_str!("../sql/sqlite/change_log_units.sql"); + +/// Drops the change log tables and every logging trigger. +fn drop_change_log(sql: &rusqlite::Connection) { + let triggers: Vec = sql + .prepare( + "SELECT name FROM sqlite_schema WHERE type='trigger' AND substr(name,1,7)='change_'", + ) + .unwrap() + .query_map([], |r| r.get(0)) + .unwrap() + .map(Result::unwrap) + .collect(); + for name in triggers { + sql.execute_batch(&format!("DROP TRIGGER {name}")).unwrap(); + } + sql.execute_batch("DROP TABLE change_log; DROP TABLE change_state;") + .unwrap(); +} fn definition(sql: &rusqlite::Connection) -> String { let mut statement = sql @@ -26,7 +46,8 @@ fn definition(sql: &rusqlite::Connection) -> String { } /// Rewrites a Store into the format written before #353: each Layer lists -/// the Layers it overwrote, newest first, and the lineage tables are absent. +/// the Layers it overwrote, newest first, and the lineage tables (and the +/// later merge, digest and change log tables) are absent. fn make_old(path: &std::path::Path) { let sql = rusqlite::Connection::open(path).unwrap(); let layers: Vec<(Vec, Vec, i64, i64)> = sql @@ -54,6 +75,7 @@ fn make_old(path: &std::path::Path) { ) .unwrap(); } + drop_change_log(&sql); sql.execute_batch( "DROP TABLE layer_lineage; DROP TABLE commit_lineage; DROP TABLE absorbed; DROP TABLE reach; \ DROP TABLE merge_operation; DROP TABLE merge_record; DROP TABLE file_digest;", @@ -65,7 +87,9 @@ fn make_old(path: &std::path::Path) { let old = source .replacen(LINEAGE, "", 1) .replacen(SCRATCH, "", 1) - .replacen(DIGESTS, "", 1); + .replacen(DIGESTS, "", 1) + .replacen(CHANGES, "", 1) + .replacen(CHANGE_UNITS, "", 1); assert_ne!(old, source, "the schema holds the lineage tables"); sql.execute( "UPDATE history_meta SET schema_source=?1, schema_definition=?2", diff --git a/engine/crates/layerfs-persistence/tests/old_format_open.rs b/engine/crates/layerfs-persistence/tests/old_format_open.rs index db7f8c0d4..359bbaa7a 100644 --- a/engine/crates/layerfs-persistence/tests/old_format_open.rs +++ b/engine/crates/layerfs-persistence/tests/old_format_open.rs @@ -10,6 +10,26 @@ const HISTORY: &str = include_str!("../sql/sqlite/history.sql"); const LINEAGE: &str = include_str!("../sql/sqlite/lineage.sql"); const SCRATCH: &str = include_str!("../sql/sqlite/merge_scratch.sql"); const DIGESTS: &str = include_str!("../sql/sqlite/file_digest.sql"); +const CHANGES: &str = include_str!("../sql/sqlite/change_log.sql"); +const CHANGE_UNITS: &str = include_str!("../sql/sqlite/change_log_units.sql"); + +/// Drops the change log tables and every logging trigger. +fn drop_change_log(sql: &rusqlite::Connection) { + let triggers: Vec = sql + .prepare( + "SELECT name FROM sqlite_schema WHERE type='trigger' AND substr(name,1,7)='change_'", + ) + .unwrap() + .query_map([], |r| r.get(0)) + .unwrap() + .map(Result::unwrap) + .collect(); + for name in triggers { + sql.execute_batch(&format!("DROP TRIGGER {name}")).unwrap(); + } + sql.execute_batch("DROP TABLE change_log; DROP TABLE change_state;") + .unwrap(); +} /// The table's definition as the history schema writes it, with its trailing /// blank line: the text an old Store's recorded schema source lacks. @@ -31,11 +51,12 @@ fn definition(sql: &rusqlite::Connection) -> String { } /// Rewrites a fresh Store into the format the daemons before `49499abac` -/// wrote: no `binding_hint` table (nor the later #353 lineage and #355 merge -/// tables), and the +/// wrote: no `binding_hint` table (nor the later #353 lineage, #355 merge, +/// #364 digest and change log tables), and the /// schema source and definition they recorded. fn make_old(path: &std::path::Path) { let sql = rusqlite::Connection::open(path).unwrap(); + drop_change_log(&sql); sql.execute_batch( "DROP TABLE binding_hint; DROP TABLE layer_lineage; DROP TABLE commit_lineage; \ DROP TABLE absorbed; DROP TABLE reach; DROP TABLE merge_operation; \ @@ -49,7 +70,9 @@ fn make_old(path: &std::path::Path) { .replacen(&hint_table(), "", 1) .replacen(LINEAGE, "", 1) .replacen(SCRATCH, "", 1) - .replacen(DIGESTS, "", 1); + .replacen(DIGESTS, "", 1) + .replacen(CHANGES, "", 1) + .replacen(CHANGE_UNITS, "", 1); assert_ne!(old, source, "the history schema holds the table"); sql.execute( "UPDATE history_meta SET schema_source=?1, schema_definition=?2", diff --git a/engine/crates/layerfs-persistence/tests/sqlite_publication.rs b/engine/crates/layerfs-persistence/tests/sqlite_publication.rs index 106c60634..6ac080763 100644 --- a/engine/crates/layerfs-persistence/tests/sqlite_publication.rs +++ b/engine/crates/layerfs-persistence/tests/sqlite_publication.rs @@ -103,8 +103,9 @@ fn wal_full_settings_and_unavailable_engine_are_explicit() { .unwrap(); // #327 dropped workspace_stage and added store_totals; #352 added binding_hint; // #353 added layer_lineage, commit_lineage, absorbed and reach; #355 - // added merge_operation and merge_record; #364 added file_digest. - assert_eq!(names, 20); + // added merge_operation and merge_record; #364 added file_digest; the + // incremental change export added change_log and change_state. + assert_eq!(names, 22); } #[test] fn final_body_multirow_locators_and_first_wins_have_one_atomic_ack() {