From 9b4e585e6fdb32a0b0658c1910d9601f97a5ad49 Mon Sep 17 00:00:00 2001 From: rexjz <54502639+rexjz@users.noreply.github.com> Date: Wed, 28 Oct 2026 22:07:48 +0800 Subject: [PATCH 1/2] feat(persistence): export a Store's changed rows as bundles for replicas A Store marked as an export base logs the key of every row written to its replicated tables. export_changes writes the rows changed after a cursor into a small SQLite bundle, from one read snapshot of the live Store: it takes no write lock and stops no writer. apply_changes installs bundles in order into an offline copy of the base. Each bundle carries a BLAKE3 digest of its rows and deletions and a chained state, so an out-of-order, repeated, changed or diverged bundle is refused before anything is installed. - change_log / change_state tables and AFTER INSERT/UPDATE/DELETE triggers on the 18 replicated tables (plus pack_unit in the group-row layouts). The triggers log nothing until mark_base, so an unmarked Store only pays one EXISTS check per written row. - Additive open-time migration (add_change_log, REBUILD.md kind 2) for Stores written before, under the same conditions as the other additive migrations. - The module (src/replication/) keeps marking, export and apply in their own files. Store connections run the Disposable profile (profile::apply: WAL, synchronous OFF) with no busy handler; a locked or non-WAL Store is refused at once, as ChangeError::Profile with the profile's own error. - An export writes the bundle under a temporary name and hard-links it into place, so the bundle path names a complete file or nothing, also after a process crash. Bundles use 1 KiB pages and carry only the tables with logged changes. - apply_changes parses the bundle's meta row without trusting the bundle's own constraints (one row, value types and lengths, sequence order). - mark_base and apply_changes checkpoint after they commit. A checkpoint that a reader blocks, or that fails, returns ChangeError::NotCheckpointed: the change committed, and the Store file alone does not hold it yet. - New tests/change_export.rs (8 tests: the replica round trip, refused bundles, locked and non-WAL replicas, a reader blocking the checkpoint, all three pack layouts, an unmarked Store). The old-format fixtures and the table count follow the two new tables. Architecture doc 84 lists them. Production LOC: 102763 -> 103792 (delta +1029) engine 61709 -> 62738 (+1029): layerfs-persistence 8422 -> 9451; core 38395 and api 2659 unchanged. Counted with tools/production_loc.py (blob 3341e4d036) --root on git archive extracts of the parent (main 582bc786f, tree e7075fc0ec) and the committed tree f9d55911d4. --- .../architecture/84-global-sqlite-store.md | 15 + engine/crates/layerfs-persistence/Cargo.toml | 4 +- .../sql/sqlite/change_log.sql | 134 ++++ .../sql/sqlite/change_log_units.sql | 2 + .../src/backend/sqlite/mod.rs | 2 +- .../src/backend/sqlite/schema.rs | 64 +- engine/crates/layerfs-persistence/src/lib.rs | 1 + .../src/replication/apply.rs | 207 ++++++ .../src/replication/base.rs | 51 ++ .../src/replication/connection.rs | 61 ++ .../src/replication/digest.rs | 105 +++ .../src/replication/error.rs | 74 +++ .../src/replication/export.rs | 227 +++++++ .../src/replication/lookup.rs | 50 ++ .../src/replication/mod.rs | 50 ++ .../src/replication/records.rs | 43 ++ .../src/replication/tables.rs | 171 +++++ .../layerfs-persistence/src/store/open.rs | 1 + .../tests/change_export.rs | 623 ++++++++++++++++++ .../tests/lineage_backfill.rs | 28 +- .../tests/old_format_open.rs | 29 +- .../tests/sqlite_publication.rs | 5 +- 22 files changed, 1931 insertions(+), 16 deletions(-) create mode 100644 engine/crates/layerfs-persistence/sql/sqlite/change_log.sql create mode 100644 engine/crates/layerfs-persistence/sql/sqlite/change_log_units.sql create mode 100644 engine/crates/layerfs-persistence/src/replication/apply.rs create mode 100644 engine/crates/layerfs-persistence/src/replication/base.rs create mode 100644 engine/crates/layerfs-persistence/src/replication/connection.rs create mode 100644 engine/crates/layerfs-persistence/src/replication/digest.rs create mode 100644 engine/crates/layerfs-persistence/src/replication/error.rs create mode 100644 engine/crates/layerfs-persistence/src/replication/export.rs create mode 100644 engine/crates/layerfs-persistence/src/replication/lookup.rs create mode 100644 engine/crates/layerfs-persistence/src/replication/mod.rs create mode 100644 engine/crates/layerfs-persistence/src/replication/records.rs create mode 100644 engine/crates/layerfs-persistence/src/replication/tables.rs create mode 100644 engine/crates/layerfs-persistence/tests/change_export.rs 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..fc7961459 --- /dev/null +++ b/engine/crates/layerfs-persistence/src/replication/apply.rs @@ -0,0 +1,207 @@ +//! 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> { + 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}"), + [], + )?; + } + // The installed rows logged themselves; 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( + "UPDATE main.change_state SET state=?1,seq=?2 WHERE id=1", + 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..7d28ff255 --- /dev/null +++ b/engine/crates/layerfs-persistence/tests/change_export.rs @@ -0,0 +1,623 @@ +//! 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 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() { From 8993ff8a79df82f87f2d9623794b2e952ccb6fb9 Mon Sep 17 00:00:00 2001 From: rexjz <54502639+rexjz@users.noreply.github.com> Date: Wed, 28 Oct 2026 22:07:48 +0800 Subject: [PATCH 2/2] fix(persistence): do not log the rows a bundle installs on a replica apply_changes installed a bundle's rows while the replica's change_state row was present, so every trigger logged them, and the apply then deleted that log again. The deleted log left free pages in the replica file, and every copy or backup of the replica carried them. - install deletes change_state before it installs rows and inserts it again with the bundle's ending state, in the same transaction, so a failed apply leaves the replica as it was. - New test: applying a 1500-object bundle leaves only the sequence marker in change_log and no free pages. Without the change it leaves 22 free pages. Production LOC: 103792 -> 103793 (delta +1) engine 62738 -> 62739 (+1): layerfs-persistence 9451 -> 9452; core 38395 and api 2659 unchanged. Counted with tools/production_loc.py (blob 3341e4d036) --root on git archive extracts of the parent tree f9d55911d4 and the committed tree 58c86faf55. --- .../src/replication/apply.rs | 9 ++++-- .../tests/change_export.rs | 30 +++++++++++++++++++ 2 files changed, 37 insertions(+), 2 deletions(-) diff --git a/engine/crates/layerfs-persistence/src/replication/apply.rs b/engine/crates/layerfs-persistence/src/replication/apply.rs index fc7961459..e42aa2788 100644 --- a/engine/crates/layerfs-persistence/src/replication/apply.rs +++ b/engine/crates/layerfs-persistence/src/replication/apply.rs @@ -159,6 +159,11 @@ fn verify(c: &Connection, catalog: &[u8], meta: &BundleMeta) -> Result<(), Chang /// 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) @@ -193,14 +198,14 @@ fn install(c: &Connection, meta: &BundleMeta) -> Result<(), ChangeError> { [], )?; } - // The installed rows logged themselves; the replica now stands at `after`. + // 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( - "UPDATE main.change_state SET state=?1,seq=?2 WHERE id=1", + "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/tests/change_export.rs b/engine/crates/layerfs-persistence/tests/change_export.rs index 7d28ff255..530bdbb6a 100644 --- a/engine/crates/layerfs-persistence/tests/change_export.rs +++ b/engine/crates/layerfs-persistence/tests/change_export.rs @@ -318,6 +318,36 @@ fn a_replica_from_the_base_and_its_bundles_holds_the_live_rows() { ); } +#[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");