From 44b6217e3b18b2b15767761fa8d0e29d9cb8f5ea Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 14:30:21 -0800 Subject: [PATCH 01/14] fix(sqlite): shift clone postings without TEMP copy Refs #4152 --- .../projection/artifact/increment/clones.rs | 177 ++++++++++++++++-- 1 file changed, 164 insertions(+), 13 deletions(-) diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs index 9c229319776..5729a5b59ff 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs @@ -52,19 +52,11 @@ pub(super) fn carry_clone_rows( [], ) .map_err(sqlite_error)?; - if let Some(first) = clones.first_affected().filter(|_| clones.shifts()) { - // The posting's ordinal is part of its key, so shifted rows are - // rewritten through a temporary copy. - transaction - .execute_batch(&format!( - "CREATE TEMP TABLE carried_clone_exact AS SELECT class, normalization_revision, digest, {image} AS occurrence_ordinal FROM clone_exact_postings WHERE occurrence_ordinal >= {first}; - DELETE FROM clone_exact_postings WHERE occurrence_ordinal >= {first}; - INSERT INTO clone_exact_postings(class, normalization_revision, digest, occurrence_ordinal) SELECT class, normalization_revision, digest, occurrence_ordinal FROM temp.carried_clone_exact; - DROP TABLE temp.carried_clone_exact;", - image = clones.sql_image("occurrence_ordinal") - )) - .map_err(sqlite_error)?; - } + // A TEMP copy of every surviving exact posting retains unlinked pager pages + // after DROP until this long-lived builder resets its TEMP schema. Move the + // composite primary keys through a disjoint negative ordinal namespace, + // just as the carried occurrence rows do, without materializing a TEMP b-tree. + shift_clone_posting_keys(transaction, clones)?; { let mut insert = transaction .prepare_cached( @@ -140,6 +132,21 @@ pub(super) fn carry_clone_rows( renumber_clone_payloads(transaction, &parent_payloads, &successor_payloads, control) } +/// Shift the occurrence component of the WITHOUT ROWID posting primary key. +/// Staging the new keys as negative ordinals prevents an in-place update from +/// colliding with other positive parent keys, including on shrinking ranges. +fn shift_clone_posting_keys( + transaction: &Transaction<'_>, + clones: &ShiftV1, +) -> Result<(), CodeLexicalArtifactErrorV1> { + shift_rowid_keys( + transaction, + "clone_exact_postings", + "occurrence_ordinal", + clones, + ) +} + /// Where a run's first successor occurrence lands: its parent start moved /// by the growth of the runs before it. `index` pairs the run with its /// replaced range, since two runs without clone bodies can replace the same @@ -278,3 +285,147 @@ fn renumber_clone_payloads( } Ok(()) } + +#[cfg(test)] +mod tests { + use rusqlite::{Connection, params}; + + use super::{ShiftV1, shift_clone_posting_keys}; + + type Posting = (i64, i64, Vec, i64); + + fn postings(connection: &Connection) -> Vec { + let mut statement = connection + .prepare( + "SELECT class, normalization_revision, digest, occurrence_ordinal + FROM clone_exact_postings + ORDER BY class, normalization_revision, digest, occurrence_ordinal", + ) + .unwrap(); + let rows = statement + .query_map([], |row| { + Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)) + }) + .unwrap() + .collect::>>() + .unwrap(); + rows + } + + fn integrity(connection: &Connection, schema: &str) { + let result: String = connection + .query_row(&format!("PRAGMA {schema}.integrity_check"), [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(result, "ok", "{schema} page integrity"); + } + + #[test] + fn carried_clone_keys_shift_without_retained_file_backed_temp_pages() { + let mut connection = Connection::open_in_memory().unwrap(); + connection + .execute_batch( + "PRAGMA temp_store = FILE; + PRAGMA temp.cache_size = 8; + CREATE TABLE clone_exact_postings ( + class INTEGER NOT NULL, + normalization_revision INTEGER NOT NULL, + digest BLOB NOT NULL, + occurrence_ordinal INTEGER NOT NULL, + PRIMARY KEY(class, normalization_revision, digest, occurrence_ordinal) + ) WITHOUT ROWID; + CREATE TABLE independent(value TEXT NOT NULL); + INSERT INTO independent VALUES ('survivor');", + ) + .unwrap(); + assert_eq!( + connection + .pragma_query_value(Some("temp"), "auto_vacuum", |row| row.get::<_, i64>(0)) + .unwrap(), + 0, + "the default TEMP pager does not truncate pages on DROP" + ); + { + let mut insert = connection + .prepare( + "INSERT INTO clone_exact_postings + (class, normalization_revision, digest, occurrence_ordinal) + VALUES (?1, ?2, ?3, ?4)", + ) + .unwrap(); + for ordinal in 1_i64..=16_000 { + let digest = (ordinal % 17).to_be_bytes().repeat(4); + for class in 0_i64..3 { + insert + .execute(params![class, class + 1, &digest, ordinal]) + .unwrap(); + } + } + } + let original = postings(&connection); + assert_eq!(original.len(), 48_000); + for (index, shift) in [ + ShiftV1::new([((11, 20), 5), ((36, 39), 6)]).unwrap(), + ShiftV1::new([((11, 20), 12), ((36, 39), 1)]).unwrap(), + ] + .iter() + .enumerate() + { + let mut expected: Vec = original + .iter() + .filter_map(|(class, revision, digest, ordinal)| { + shift + .map(u64::try_from(*ordinal).unwrap()) + .unwrap() + .map(|mapped| { + ( + *class, + *revision, + digest.clone(), + i64::try_from(mapped).unwrap(), + ) + }) + }) + .collect(); + expected.sort(); + assert!(expected.len() > 47_000); + assert_ne!(expected, original, "the shift must move real keys"); + + let transaction = connection.transaction().unwrap(); + transaction + .execute( + &format!( + "DELETE FROM clone_exact_postings WHERE {}", + shift.sql_replaced("occurrence_ordinal") + ), + [], + ) + .unwrap(); + shift_clone_posting_keys(&transaction, shift).unwrap(); + assert_eq!(postings(&transaction), expected); + assert_eq!( + transaction + .pragma_query_value(Some("temp"), "page_count", |row| row.get::<_, i64>(0)) + .unwrap(), + 0, + "shifting a large WITHOUT ROWID index must not retain TEMP pager pages" + ); + integrity(&transaction, "main"); + integrity(&transaction, "temp"); + let survivor: String = transaction + .query_row("SELECT value FROM independent", [], |row| row.get(0)) + .unwrap(); + assert_eq!(survivor, "survivor"); + + if index == 0 { + transaction.rollback().unwrap(); + assert_eq!(postings(&connection), original, "rollback restores every key"); + } else { + transaction.commit().unwrap(); + assert_eq!(postings(&connection), expected, "commit preserves every key"); + } + integrity(&connection, "main"); + } + } +} From 4a9ff682802e16add7599f0e207755c0e9302e39 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 14:33:51 -0800 Subject: [PATCH 02/14] test(sqlite): prove TEMP copy spills with a positive control Refs #4152 --- .../projection/artifact/increment/clones.rs | 35 +++++++++++++++---- 1 file changed, 29 insertions(+), 6 deletions(-) diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs index 5729a5b59ff..3e3b7224ab7 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs @@ -139,12 +139,7 @@ fn shift_clone_posting_keys( transaction: &Transaction<'_>, clones: &ShiftV1, ) -> Result<(), CodeLexicalArtifactErrorV1> { - shift_rowid_keys( - transaction, - "clone_exact_postings", - "occurrence_ordinal", - clones, - ) + shift_rowid_keys(transaction, "clone_exact_postings", "occurrence_ordinal", clones) } /// Where a run's first successor occurrence lands: its parent start moved @@ -339,6 +334,13 @@ mod tests { INSERT INTO independent VALUES ('survivor');", ) .unwrap(); + assert_eq!( + connection + .pragma_query_value(None, "temp_store", |row| row.get::<_, i64>(0)) + .unwrap(), + 1, + "the probe must use file-backed TEMP storage" + ); assert_eq!( connection .pragma_query_value(Some("temp"), "auto_vacuum", |row| row.get::<_, i64>(0)) @@ -427,5 +429,26 @@ mod tests { } integrity(&connection, "main"); } + + // Positive control: the removed CREATE TEMP / DROP pattern really + // allocates retained pages on this SQLite build. Without this check, + // a zero-page assertion could pass on a build that never spills. + connection + .execute_batch( + "CREATE TEMP TABLE old_style_clone_copy AS + SELECT class, normalization_revision, digest, occurrence_ordinal + FROM clone_exact_postings; + DROP TABLE temp.old_style_clone_copy;", + ) + .unwrap(); + let retained_pages: i64 = connection + .pragma_query_value(Some("temp"), "page_count", |row| row.get(0)) + .unwrap(); + assert!( + retained_pages > 32, + "the old large TEMP copy must retain physical pager pages: {retained_pages}" + ); + integrity(&connection, "main"); + integrity(&connection, "temp"); } } From e96a95c1ef4ec3c4211884661a55e21a156ef2a1 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 15:25:39 -0800 Subject: [PATCH 03/14] fix(sqlite): bound clone posting key updates to avoid statement spills Refs #4152 --- .../projection/artifact/increment/clones.rs | 140 +++++++++++++++++- 1 file changed, 132 insertions(+), 8 deletions(-) diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs index 3e3b7224ab7..8a66bc95e14 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs @@ -4,6 +4,7 @@ use std::collections::BTreeMap; +use rusqlite::types::Value; use rusqlite::{Transaction, params}; use tracedecay_code_index::production::CodeIndexExecutionControlV1; @@ -13,6 +14,7 @@ use super::super::format::contract_number; use super::super::prepared::PreparedCloneBodyV1; use super::super::row_codec::stored_symbol_key; use super::super::{CodeLexicalArtifactErrorV1, checkpoint, sqlite_error}; +use super::postings::scan_window; use super::{ReplacedFilesV1, ShiftV1, shift_rowid_keys, shifted}; /// One successor clone body with the occurrence ordinal it takes. @@ -54,8 +56,9 @@ pub(super) fn carry_clone_rows( .map_err(sqlite_error)?; // A TEMP copy of every surviving exact posting retains unlinked pager pages // after DROP until this long-lived builder resets its TEMP schema. Move the - // composite primary keys through a disjoint negative ordinal namespace, - // just as the carried occurrence rows do, without materializing a TEMP b-tree. + // composite primary keys through bounded, ordered windows and a negative + // ordinal namespace. A single bulk UPDATE can itself spill an unlinked + // statement journal, even when the TEMP pager has zero pages. shift_clone_posting_keys(transaction, clones)?; { let mut insert = transaction @@ -132,14 +135,113 @@ pub(super) fn carry_clone_rows( renumber_clone_payloads(transaction, &parent_payloads, &successor_payloads, control) } +/// A key-ordered window, so SQLite can seek the WITHOUT ROWID primary key +/// without sorting the entire posting index into an external TEMP file. +type ClonePostingKey = (i64, i64, Vec, i64); + +fn clone_posting_window( + transaction: &Transaction<'_>, + predicate: &str, + last: &Option, +) -> Result, CodeLexicalArtifactErrorV1> { + let columns = "class, normalization_revision, digest, occurrence_ordinal"; + scan_window( + transaction, + &format!( + "SELECT {columns} FROM clone_exact_postings WHERE {predicate} ORDER BY {columns} LIMIT ?1" + ), + &format!( + "SELECT {columns} FROM clone_exact_postings WHERE {predicate} AND ({columns}) > (?1, ?2, ?3, ?4) ORDER BY {columns} LIMIT ?5" + ), + last, + |(class, revision, digest, ordinal)| { + vec![ + Value::Integer(*class), + Value::Integer(*revision), + Value::Blob(digest.clone()), + Value::Integer(*ordinal), + ] + }, + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + ) +} + /// Shift the occurrence component of the WITHOUT ROWID posting primary key. -/// Staging the new keys as negative ordinals prevents an in-place update from -/// colliding with other positive parent keys, including on shrinking ranges. +/// Both passes seek bounded primary-key windows and update one keyed row at a +/// time: SQLite's single bulk UPDATE can allocate an unlinked statement +/// journal independently of the TEMP b-tree pager. The negative namespace +/// prevents collisions with positive keys while a window advances. fn shift_clone_posting_keys( transaction: &Transaction<'_>, clones: &ShiftV1, ) -> Result<(), CodeLexicalArtifactErrorV1> { - shift_rowid_keys(transaction, "clone_exact_postings", "occurrence_ordinal", clones) + let Some(first) = clones.first_affected().filter(|_| clones.shifts()) else { + return Ok(()); + }; + let mut update = transaction + .prepare_cached( + "UPDATE clone_exact_postings SET occurrence_ordinal = ?5 + WHERE class = ?1 AND normalization_revision = ?2 + AND digest = ?3 AND occurrence_ordinal = ?4", + ) + .map_err(sqlite_error)?; + let mut last = None; + loop { + let window = clone_posting_window( + transaction, + &format!("occurrence_ordinal >= {first}"), + &last, + )?; + let Some(final_key) = window.last() else { + break; + }; + last = Some(final_key.clone()); + for (class, revision, digest, ordinal) in window { + let mapped = clones + .map(u64::try_from(ordinal).map_err(contract_number)?)? + .ok_or_else(|| { + CodeLexicalArtifactErrorV1::Corrupt( + "carried clone posting belongs to a replaced range".to_owned(), + ) + })?; + let mapped = i64::try_from(mapped).map_err(contract_number)?; + // -mapped - 1 fits even when mapped == i64::MAX. + let staged = -mapped - 1; + if update + .execute(params![class, revision, digest, ordinal, staged]) + .map_err(sqlite_error)? + != 1 + { + return Err(CodeLexicalArtifactErrorV1::Corrupt( + "carried clone posting changed during key shift".to_owned(), + )); + } + } + } + // All moved keys are now negative and disjoint from every unchanged key. + // Keyset pagination remains stable as restored positive keys leave the + // scan; compute -(ordinal + 1) to handle an i64::MIN staged key. + let mut last = None; + loop { + let window = clone_posting_window(transaction, "occurrence_ordinal < 0", &last)?; + let Some(final_key) = window.last() else { + break; + }; + last = Some(final_key.clone()); + for (class, revision, digest, ordinal) in window { + let restored = -(ordinal + 1); + if update + .execute(params![class, revision, digest, ordinal, restored]) + .map_err(sqlite_error)? + != 1 + { + return Err(CodeLexicalArtifactErrorV1::Corrupt( + "staged clone posting changed during key restoration".to_owned(), + )); + } + } + } + Ok(()) } /// Where a run's first successor occurrence lands: its parent start moved @@ -318,7 +420,9 @@ mod tests { #[test] fn carried_clone_keys_shift_without_retained_file_backed_temp_pages() { - let mut connection = Connection::open_in_memory().unwrap(); + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("clone-posting-carry.sqlite"); + let mut connection = Connection::open(&path).unwrap(); connection .execute_batch( "PRAGMA temp_store = FILE; @@ -367,6 +471,7 @@ mod tests { } let original = postings(&connection); assert_eq!(original.len(), 48_000); + let mut committed = original.clone(); for (index, shift) in [ ShiftV1::new([((11, 20), 5), ((36, 39), 6)]).unwrap(), ShiftV1::new([((11, 20), 12), ((36, 39), 1)]).unwrap(), @@ -422,10 +527,19 @@ mod tests { if index == 0 { transaction.rollback().unwrap(); - assert_eq!(postings(&connection), original, "rollback restores every key"); + assert_eq!( + postings(&connection), + original, + "rollback restores every key" + ); } else { transaction.commit().unwrap(); - assert_eq!(postings(&connection), expected, "commit preserves every key"); + committed = expected; + assert_eq!( + postings(&connection), + committed, + "commit preserves every key" + ); } integrity(&connection, "main"); } @@ -450,5 +564,15 @@ mod tests { ); integrity(&connection, "main"); integrity(&connection, "temp"); + // Main storage must actually survive closing the writer and reopening + // its file: checking only an in-memory connection masks durability bugs. + drop(connection); + let reopened = Connection::open(&path).unwrap(); + assert_eq!(postings(&reopened), committed); + integrity(&reopened, "main"); + let survivor: String = reopened + .query_row("SELECT value FROM independent", [], |row| row.get(0)) + .unwrap(); + assert_eq!(survivor, "survivor"); } } From 9eb32c4394f4461c2e303f159fa1d518af679fbb Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 15:29:00 -0800 Subject: [PATCH 04/14] fix(sqlite): make bounded clone remaps cancellable Refs #4152 Preserve the concurrent bounded exact-key remap and its row-count guards. Check the existing execution control before each window and keyed update in both phases so cancellation can roll back partially staged keys. Exercise real adjacent-key collisions, interruption during both phases, and reopening the durable database after rollback and commit. Sample live Linux SQLite spill files from the progress callback, with the bulk update as a positive control: TEMP page_count after execution misses it. Populate the disk fixture in one transaction rather than 48000 fsyncs. --- .../projection/artifact/increment/clones.rs | 206 +--------- .../artifact/increment/clones/tests.rs | 377 ++++++++++++++++++ 2 files changed, 387 insertions(+), 196 deletions(-) create mode 100644 crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs index 8a66bc95e14..6b7408ab7dc 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs @@ -58,8 +58,8 @@ pub(super) fn carry_clone_rows( // after DROP until this long-lived builder resets its TEMP schema. Move the // composite primary keys through bounded, ordered windows and a negative // ordinal namespace. A single bulk UPDATE can itself spill an unlinked - // statement journal, even when the TEMP pager has zero pages. - shift_clone_posting_keys(transaction, clones)?; + // statement-local key index, even when the TEMP pager has zero pages. + shift_clone_posting_keys(transaction, clones, control)?; { let mut insert = transaction .prepare_cached( @@ -168,12 +168,13 @@ fn clone_posting_window( /// Shift the occurrence component of the WITHOUT ROWID posting primary key. /// Both passes seek bounded primary-key windows and update one keyed row at a -/// time: SQLite's single bulk UPDATE can allocate an unlinked statement -/// journal independently of the TEMP b-tree pager. The negative namespace +/// time: SQLite's single bulk UPDATE can allocate an unlinked statement-local +/// key index independently of the TEMP b-tree pager. The negative namespace /// prevents collisions with positive keys while a window advances. fn shift_clone_posting_keys( transaction: &Transaction<'_>, clones: &ShiftV1, + control: &dyn CodeIndexExecutionControlV1, ) -> Result<(), CodeLexicalArtifactErrorV1> { let Some(first) = clones.first_affected().filter(|_| clones.shifts()) else { return Ok(()); @@ -187,6 +188,7 @@ fn shift_clone_posting_keys( .map_err(sqlite_error)?; let mut last = None; loop { + checkpoint(control)?; let window = clone_posting_window( transaction, &format!("occurrence_ordinal >= {first}"), @@ -197,6 +199,7 @@ fn shift_clone_posting_keys( }; last = Some(final_key.clone()); for (class, revision, digest, ordinal) in window { + checkpoint(control)?; let mapped = clones .map(u64::try_from(ordinal).map_err(contract_number)?)? .ok_or_else(|| { @@ -223,12 +226,14 @@ fn shift_clone_posting_keys( // scan; compute -(ordinal + 1) to handle an i64::MIN staged key. let mut last = None; loop { + checkpoint(control)?; let window = clone_posting_window(transaction, "occurrence_ordinal < 0", &last)?; let Some(final_key) = window.last() else { break; }; last = Some(final_key.clone()); for (class, revision, digest, ordinal) in window { + checkpoint(control)?; let restored = -(ordinal + 1); if update .execute(params![class, revision, digest, ordinal, restored]) @@ -384,195 +389,4 @@ fn renumber_clone_payloads( } #[cfg(test)] -mod tests { - use rusqlite::{Connection, params}; - - use super::{ShiftV1, shift_clone_posting_keys}; - - type Posting = (i64, i64, Vec, i64); - - fn postings(connection: &Connection) -> Vec { - let mut statement = connection - .prepare( - "SELECT class, normalization_revision, digest, occurrence_ordinal - FROM clone_exact_postings - ORDER BY class, normalization_revision, digest, occurrence_ordinal", - ) - .unwrap(); - let rows = statement - .query_map([], |row| { - Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)) - }) - .unwrap() - .collect::>>() - .unwrap(); - rows - } - - fn integrity(connection: &Connection, schema: &str) { - let result: String = connection - .query_row(&format!("PRAGMA {schema}.integrity_check"), [], |row| { - row.get(0) - }) - .unwrap(); - assert_eq!(result, "ok", "{schema} page integrity"); - } - - #[test] - fn carried_clone_keys_shift_without_retained_file_backed_temp_pages() { - let directory = tempfile::tempdir().unwrap(); - let path = directory.path().join("clone-posting-carry.sqlite"); - let mut connection = Connection::open(&path).unwrap(); - connection - .execute_batch( - "PRAGMA temp_store = FILE; - PRAGMA temp.cache_size = 8; - CREATE TABLE clone_exact_postings ( - class INTEGER NOT NULL, - normalization_revision INTEGER NOT NULL, - digest BLOB NOT NULL, - occurrence_ordinal INTEGER NOT NULL, - PRIMARY KEY(class, normalization_revision, digest, occurrence_ordinal) - ) WITHOUT ROWID; - CREATE TABLE independent(value TEXT NOT NULL); - INSERT INTO independent VALUES ('survivor');", - ) - .unwrap(); - assert_eq!( - connection - .pragma_query_value(None, "temp_store", |row| row.get::<_, i64>(0)) - .unwrap(), - 1, - "the probe must use file-backed TEMP storage" - ); - assert_eq!( - connection - .pragma_query_value(Some("temp"), "auto_vacuum", |row| row.get::<_, i64>(0)) - .unwrap(), - 0, - "the default TEMP pager does not truncate pages on DROP" - ); - { - let mut insert = connection - .prepare( - "INSERT INTO clone_exact_postings - (class, normalization_revision, digest, occurrence_ordinal) - VALUES (?1, ?2, ?3, ?4)", - ) - .unwrap(); - for ordinal in 1_i64..=16_000 { - let digest = (ordinal % 17).to_be_bytes().repeat(4); - for class in 0_i64..3 { - insert - .execute(params![class, class + 1, &digest, ordinal]) - .unwrap(); - } - } - } - let original = postings(&connection); - assert_eq!(original.len(), 48_000); - let mut committed = original.clone(); - for (index, shift) in [ - ShiftV1::new([((11, 20), 5), ((36, 39), 6)]).unwrap(), - ShiftV1::new([((11, 20), 12), ((36, 39), 1)]).unwrap(), - ] - .iter() - .enumerate() - { - let mut expected: Vec = original - .iter() - .filter_map(|(class, revision, digest, ordinal)| { - shift - .map(u64::try_from(*ordinal).unwrap()) - .unwrap() - .map(|mapped| { - ( - *class, - *revision, - digest.clone(), - i64::try_from(mapped).unwrap(), - ) - }) - }) - .collect(); - expected.sort(); - assert!(expected.len() > 47_000); - assert_ne!(expected, original, "the shift must move real keys"); - - let transaction = connection.transaction().unwrap(); - transaction - .execute( - &format!( - "DELETE FROM clone_exact_postings WHERE {}", - shift.sql_replaced("occurrence_ordinal") - ), - [], - ) - .unwrap(); - shift_clone_posting_keys(&transaction, shift).unwrap(); - assert_eq!(postings(&transaction), expected); - assert_eq!( - transaction - .pragma_query_value(Some("temp"), "page_count", |row| row.get::<_, i64>(0)) - .unwrap(), - 0, - "shifting a large WITHOUT ROWID index must not retain TEMP pager pages" - ); - integrity(&transaction, "main"); - integrity(&transaction, "temp"); - let survivor: String = transaction - .query_row("SELECT value FROM independent", [], |row| row.get(0)) - .unwrap(); - assert_eq!(survivor, "survivor"); - - if index == 0 { - transaction.rollback().unwrap(); - assert_eq!( - postings(&connection), - original, - "rollback restores every key" - ); - } else { - transaction.commit().unwrap(); - committed = expected; - assert_eq!( - postings(&connection), - committed, - "commit preserves every key" - ); - } - integrity(&connection, "main"); - } - - // Positive control: the removed CREATE TEMP / DROP pattern really - // allocates retained pages on this SQLite build. Without this check, - // a zero-page assertion could pass on a build that never spills. - connection - .execute_batch( - "CREATE TEMP TABLE old_style_clone_copy AS - SELECT class, normalization_revision, digest, occurrence_ordinal - FROM clone_exact_postings; - DROP TABLE temp.old_style_clone_copy;", - ) - .unwrap(); - let retained_pages: i64 = connection - .pragma_query_value(Some("temp"), "page_count", |row| row.get(0)) - .unwrap(); - assert!( - retained_pages > 32, - "the old large TEMP copy must retain physical pager pages: {retained_pages}" - ); - integrity(&connection, "main"); - integrity(&connection, "temp"); - // Main storage must actually survive closing the writer and reopening - // its file: checking only an in-memory connection masks durability bugs. - drop(connection); - let reopened = Connection::open(&path).unwrap(); - assert_eq!(postings(&reopened), committed); - integrity(&reopened, "main"); - let survivor: String = reopened - .query_row("SELECT value FROM independent", [], |row| row.get(0)) - .unwrap(); - assert_eq!(survivor, "survivor"); - } -} +mod tests; diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs new file mode 100644 index 00000000000..7e365e59574 --- /dev/null +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs @@ -0,0 +1,377 @@ +use std::path::Path; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use rusqlite::{Connection, ErrorCode, params}; +use tempfile::{TempDir, tempdir}; +use tracedecay_code_index::production::CodeIndexExecutionControlV1; + +use super::super::CARRY_SCAN_ROWS; +use super::{ClonePostingKey, CodeLexicalArtifactErrorV1, ShiftV1, shift_clone_posting_keys}; + +struct ActiveControl; + +impl CodeIndexExecutionControlV1 for ActiveControl { + fn is_cancelled(&self) -> bool { + false + } + + fn is_deadline_exceeded(&self) -> bool { + false + } +} + +struct CancelAfter { + checks: AtomicUsize, + limit: usize, +} + +impl CodeIndexExecutionControlV1 for CancelAfter { + fn is_cancelled(&self) -> bool { + self.checks.fetch_add(1, Ordering::Relaxed) >= self.limit + } + + fn is_deadline_exceeded(&self) -> bool { + false + } +} + +fn open_fixture(root: &Path) -> Connection { + let connection = Connection::open(root.join("clones.sqlite")).unwrap(); + connection + .execute_batch( + "PRAGMA temp_store = FILE; + PRAGMA cache_size = 8; + PRAGMA temp.cache_size = 8;", + ) + .unwrap(); + connection +} + +fn fixture() -> (TempDir, Connection) { + let root = tempdir().unwrap(); + let mut connection = open_fixture(root.path()); + connection + .execute_batch( + "CREATE TABLE clone_exact_postings ( + class INTEGER NOT NULL, + normalization_revision INTEGER NOT NULL, + digest BLOB NOT NULL, + occurrence_ordinal INTEGER NOT NULL, + PRIMARY KEY(class, normalization_revision, digest, occurrence_ordinal) + ) WITHOUT ROWID; + CREATE TABLE independent(value TEXT NOT NULL); + INSERT INTO independent VALUES ('survivor');", + ) + .unwrap(); + let transaction = connection.transaction().unwrap(); + { + let mut insert = transaction + .prepare( + "INSERT INTO clone_exact_postings + (class, normalization_revision, digest, occurrence_ordinal) + VALUES (?1, ?2, ?3, ?4)", + ) + .unwrap(); + // Consecutive ordinals share a key prefix. Expansion really collides + // without staging, and each family crosses multiple scan windows. + let digest = [7_u8; 32]; + for ordinal in 1_i64..=16_000 { + for class in 0_i64..3 { + insert + .execute(params![class, class + 1, digest.as_slice(), ordinal]) + .unwrap(); + } + } + } + transaction.commit().unwrap(); + (root, connection) +} + +fn postings(connection: &Connection) -> Vec { + let mut statement = connection + .prepare( + "SELECT class, normalization_revision, digest, occurrence_ordinal + FROM clone_exact_postings + ORDER BY class, normalization_revision, digest, occurrence_ordinal", + ) + .unwrap(); + statement + .query_map([], |row| { + Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)) + }) + .unwrap() + .collect::>>() + .unwrap() +} + +fn integrity(connection: &Connection, schema: &str) { + let result: String = connection + .query_row(&format!("PRAGMA {schema}.integrity_check"), [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(result, "ok", "{schema} page integrity"); +} + +fn expected_postings(original: &[ClonePostingKey], shift: &ShiftV1) -> Vec { + let mut expected: Vec<_> = original + .iter() + .filter_map(|(class, revision, digest, ordinal)| { + shift + .map(u64::try_from(*ordinal).unwrap()) + .unwrap() + .map(|mapped| { + ( + *class, + *revision, + digest.clone(), + i64::try_from(mapped).unwrap(), + ) + }) + }) + .collect(); + expected.sort(); + expected +} + +#[test] +fn carried_clone_keys_shift_without_retained_file_backed_temp_pages() { + let (root, mut connection) = fixture(); + let original = postings(&connection); + assert_eq!(original.len(), 48_000); + assert_eq!( + connection + .pragma_query_value(None, "temp_store", |row| row.get::<_, i64>(0)) + .unwrap(), + 1 + ); + assert_eq!( + connection + .pragma_query_value(Some("temp"), "auto_vacuum", |row| row.get::<_, i64>(0)) + .unwrap(), + 0 + ); + for (index, shift) in [ + ShiftV1::new([((11, 20), 5), ((36, 39), 6)]).unwrap(), + ShiftV1::new([((11, 20), 12), ((36, 39), 1)]).unwrap(), + ] + .iter() + .enumerate() + { + let expected = expected_postings(&original, shift); + assert_eq!(expected.len(), 47_964); + assert_ne!(expected, original); + let transaction = connection.transaction().unwrap(); + transaction + .execute( + &format!( + "DELETE FROM clone_exact_postings WHERE {}", + shift.sql_replaced("occurrence_ordinal") + ), + [], + ) + .unwrap(); + shift_clone_posting_keys(&transaction, shift, &ActiveControl).unwrap(); + assert_eq!(postings(&transaction), expected); + assert_eq!( + transaction + .pragma_query_value(Some("temp"), "page_count", |row| row.get::<_, i64>(0)) + .unwrap(), + 0, + "the remap must not retain TEMP pager pages" + ); + integrity(&transaction, "main"); + integrity(&transaction, "temp"); + if index == 0 { + transaction.rollback().unwrap(); + } else { + transaction.commit().unwrap(); + } + drop(connection); + connection = open_fixture(root.path()); + let committed = if index == 0 { &original } else { &expected }; + assert_eq!( + &postings(&connection), + committed, + "a fresh on-disk connection must see exactly the committed keys" + ); + let survivor: String = connection + .query_row("SELECT value FROM independent", [], |row| row.get(0)) + .unwrap(); + assert_eq!(survivor, "survivor"); + integrity(&connection, "main"); + } + connection + .execute_batch( + "CREATE TEMP TABLE old_style_clone_copy AS + SELECT class, normalization_revision, digest, occurrence_ordinal + FROM clone_exact_postings; + DROP TABLE temp.old_style_clone_copy;", + ) + .unwrap(); + let retained_pages: i64 = connection + .pragma_query_value(Some("temp"), "page_count", |row| row.get(0)) + .unwrap(); + assert!(retained_pages > 32, "positive TEMP-copy control: {retained_pages}"); + integrity(&connection, "main"); + integrity(&connection, "temp"); +} + +#[test] +fn clone_remap_stages_colliding_keys_and_rolls_back_interrupted_windows() { + let (root, mut connection) = fixture(); + let original = postings(&connection); + let shift = ShiftV1::new([((1, 1), 1)]).unwrap(); + { + let transaction = connection.transaction().unwrap(); + let error = transaction + .execute( + "UPDATE clone_exact_postings SET occurrence_ordinal = occurrence_ordinal + 1", + [], + ) + .unwrap_err(); + assert_eq!( + error.sqlite_error_code(), + Some(ErrorCode::ConstraintViolation) + ); + transaction.rollback().unwrap(); + } + // Interrupt after at least one complete window in each staging direction. + for limit in [CARRY_SCAN_ROWS + 2, original.len() + CARRY_SCAN_ROWS + 32] { + let control = CancelAfter { + checks: AtomicUsize::new(0), + limit, + }; + { + let transaction = connection.transaction().unwrap(); + let error = shift_clone_posting_keys(&transaction, &shift, &control).unwrap_err(); + assert!(matches!(error, CodeLexicalArtifactErrorV1::Interrupted(_))); + let negative: i64 = transaction + .query_row( + "SELECT COUNT(*) FROM clone_exact_postings WHERE occurrence_ordinal < 0", + [], + |row| row.get(0), + ) + .unwrap(); + assert!(negative > 0 && negative < 48_000); + integrity(&transaction, "main"); + // Drop must roll back partially staged/restored keys, not publish them. + } + drop(connection); + connection = open_fixture(root.path()); + assert_eq!(postings(&connection), original); + integrity(&connection, "main"); + } + let transaction = connection.transaction().unwrap(); + shift_clone_posting_keys(&transaction, &shift, &ActiveControl).unwrap(); + assert_eq!(postings(&transaction), expected_postings(&original, &shift)); + transaction.commit().unwrap(); +} + +#[cfg(target_os = "linux")] +mod physical_spill { + use std::io; + use std::process::Command; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + + use super::super::super::shift_rowid_keys; + use super::{ + ActiveControl, ShiftV1, expected_postings, fixture, postings, shift_clone_posting_keys, + }; + + fn spill_bytes() -> io::Result { + let mut bytes = 0; + for entry in std::fs::read_dir("/proc/self/fd")? { + let path = entry?.path(); + let target = match std::fs::read_link(&path) { + Ok(target) => target, + Err(error) if error.kind() == io::ErrorKind::NotFound => continue, + Err(error) => return Err(error), + }; + if target + .file_name() + .is_some_and(|name| name.to_string_lossy().starts_with("etilqs_")) + { + bytes += std::fs::metadata(path)?.len(); + } + } + Ok(bytes) + } + + #[test] + fn clone_posting_remap_has_no_implicit_file_spill() { + const CHILD: &str = "TRACEDECAY_CLONE_SPILL_TEST_CHILD"; + if std::env::var_os(CHILD).is_none() { + // Other unit tests may spill concurrently; only measure our child. + let output = Command::new(std::env::current_exe().unwrap()) + .arg("clone_posting_remap_has_no_implicit_file_spill") + .arg("--test-threads=1") + .arg("--nocapture") + .env(CHILD, "1") + .output() + .unwrap(); + let stdout = String::from_utf8_lossy(&output.stdout); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(output.status.success(), "{stdout}\n{stderr}"); + assert!(stdout.contains("1 passed"), "child did not execute: {stdout}"); + return; + } + let (_root, mut connection) = fixture(); + let original = postings(&connection); + let shift = ShiftV1::new([((1, 1), 1)]).unwrap(); + let peak = Arc::new(AtomicU64::new(0)); + let unavailable = Arc::new(AtomicBool::new(false)); + let observed_peak = Arc::clone(&peak); + let failed_probe = Arc::clone(&unavailable); + connection.progress_handler( + 4096, + Some(move || match spill_bytes() { + Ok(bytes) => { + observed_peak.fetch_max(bytes, Ordering::Relaxed); + false + } + Err(_) => { + failed_probe.store(true, Ordering::Relaxed); + true + } + }), + ); + { + let transaction = connection.transaction().unwrap(); + shift_rowid_keys( + &transaction, + "clone_exact_postings", + "occurrence_ordinal", + &shift, + ) + .unwrap(); + transaction.rollback().unwrap(); + } + assert!(!unavailable.load(Ordering::Relaxed)); + assert!( + peak.load(Ordering::Relaxed) > 0, + "bulk UPDATE must really spill" + ); + assert_eq!( + connection + .pragma_query_value(Some("temp"), "page_count", |row| row.get::<_, i64>(0)) + .unwrap(), + 0, + "TEMP page_count alone misses the just-freed statement spill" + ); + assert_eq!(postings(&connection), original); + assert_eq!(spill_bytes().unwrap(), 0); + peak.store(0, Ordering::Relaxed); + { + let transaction = connection.transaction().unwrap(); + shift_clone_posting_keys(&transaction, &shift, &ActiveControl).unwrap(); + transaction.commit().unwrap(); + } + connection.progress_handler(0, None:: bool>); + assert!(!unavailable.load(Ordering::Relaxed)); + assert_eq!(peak.load(Ordering::Relaxed), 0, "no implicit remap spill"); + assert_eq!(spill_bytes().unwrap(), 0); + assert_eq!(postings(&connection), expected_postings(&original, &shift)); + } +} From f5358c4179a0b3c461823bcedbc506a563070041 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 15:47:25 -0800 Subject: [PATCH 05/14] fix(sqlite): bound lexical row carry updates and TEMP scratch Refs #4152 Bulk rowid rewrites and non-key chunk ordinal rewrites allocate statement-local ephemeral SQLite indexes, invisible to TEMP page_count after execution. Use bounded ordered key scans and exact-key updates, retain two-phase collision-safe rowid staging, and checkpoint every row. Add real on-disk key/durability/rollback/cancellation coverage and a Linux live-unlinked-FD positive control for the old bulk rewrite. No allocator, TEMP reset, durable schema or CI policy changes. --- .../lexical/projection/artifact/increment.rs | 148 +++++++-- .../projection/artifact/increment/clones.rs | 2 +- .../projection/artifact/increment/tests.rs | 312 ++++++++++++++++++ 3 files changed, 435 insertions(+), 27 deletions(-) create mode 100644 crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs index fc1b39374a2..c1f9ea584de 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs @@ -15,6 +15,7 @@ use std::cell::RefCell; use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; +use rusqlite::types::Value; use rusqlite::{OptionalExtension, Transaction, params}; use tracedecay_code_index::production::{ CodeIndexExecutionControlV1, SealedLexicalFileReplacementV1, VerifiedSealedLexicalPageV1, @@ -636,25 +637,68 @@ fn shift_rowid_keys( table: &str, column: &str, shift: &ShiftV1, + control: &dyn CodeIndexExecutionControlV1, ) -> Result<(), CodeLexicalArtifactErrorV1> { let Some(first) = shift.first_affected().filter(|_| shift.shifts()) else { return Ok(()); }; - transaction - .execute( - &format!( - "UPDATE {table} SET {column} = -({image}) - 1 WHERE {column} >= {first}", - image = shift.sql_image(column) - ), - [], - ) - .map_err(sqlite_error)?; - transaction - .execute( - &format!("UPDATE {table} SET {column} = -{column} - 1 WHERE {column} < 0"), - [], - ) + // A bulk UPDATE of INTEGER PRIMARY KEY values builds a statement-local + // ephemeral key index that can spill despite an empty TEMP pager afterward. + // Use bounded key-ordered reads and exact updates in both staging passes. + let mut update = transaction + .prepare_cached(&format!( + "UPDATE {table} SET {column} = ?2 WHERE {column} = ?1" + )) .map_err(sqlite_error)?; + for staging in [true, false] { + let predicate = if staging { + format!("{column} >= {first}") + } else { + format!("{column} < 0") + }; + let first_sql = format!( + "SELECT {column} FROM {table} WHERE {predicate} ORDER BY {column} LIMIT ?1" + ); + let after_sql = format!( + "SELECT {column} FROM {table} WHERE {predicate} AND {column} > ?1 ORDER BY {column} LIMIT ?2" + ); + let mut last: Option = None; + loop { + checkpoint(control)?; + let window: Vec = postings::scan_window( + transaction, + &first_sql, + &after_sql, + &last, + |key| vec![Value::Integer(*key)], + |row| row.get(0), + )?; + let Some(final_key) = window.last() else { + break; + }; + last = Some(*final_key); + for key in window { + checkpoint(control)?; + let target = if staging { + let mapped = shift + .map(u64::try_from(key).map_err(contract_number)?)? + .ok_or_else(|| { + CodeLexicalArtifactErrorV1::Corrupt( + "replaced lexical row survived deletion".to_owned(), + ) + })?; + -i64::try_from(mapped).map_err(contract_number)? - 1 + } else { + -(key + 1) + }; + if update.execute(params![key, target]).map_err(sqlite_error)? != 1 { + return Err(CodeLexicalArtifactErrorV1::Corrupt( + "carried lexical row changed during key shift".to_owned(), + )); + } + } + } + } Ok(()) } @@ -673,7 +717,7 @@ fn carry_row_blocks( [], ) .map_err(sqlite_error)?; - shift_rowid_keys(transaction, "row_blocks", "first_document", documents)?; + shift_rowid_keys(transaction, "row_blocks", "first_document", documents, control)?; let mut insert = transaction .prepare_cached("INSERT INTO row_blocks(first_document, payload) VALUES (?1, ?2)") .map_err(sqlite_error)?; @@ -703,17 +747,7 @@ fn carry_row_chunks( [], ) .map_err(sqlite_error)?; - if let Some(first) = documents.first_affected().filter(|_| documents.shifts()) { - transaction - .execute( - &format!( - "UPDATE row_chunks SET document_id = {} WHERE document_id >= {first}", - documents.sql_image("document_id") - ), - [], - ) - .map_err(sqlite_error)?; - } + shift_row_chunk_documents(transaction, documents, control)?; let mut insert = transaction .prepare_cached("INSERT INTO row_chunks(chunk_id, document_id) VALUES (?1, ?2)") .map_err(sqlite_error)?; @@ -731,6 +765,65 @@ fn carry_row_chunks( Ok(()) } +/// Row chunks are keyed by chunk_id, not document_id. A bulk UPDATE of their +/// document ordinals can spill SQLite's statement-local key index. The +/// primary-key window remains stable because chunk_id never changes. +fn shift_row_chunk_documents( + transaction: &Transaction<'_>, + documents: &ShiftV1, + control: &dyn CodeIndexExecutionControlV1, +) -> Result<(), CodeLexicalArtifactErrorV1> { + let Some(first) = documents.first_affected().filter(|_| documents.shifts()) else { + return Ok(()); + }; + let first_sql = format!( + "SELECT chunk_id, document_id FROM row_chunks WHERE document_id >= {first} ORDER BY chunk_id LIMIT ?1" + ); + let after_sql = format!( + "SELECT chunk_id, document_id FROM row_chunks WHERE document_id >= {first} AND chunk_id > ?1 ORDER BY chunk_id LIMIT ?2" + ); + let mut update = transaction + .prepare_cached("UPDATE row_chunks SET document_id = ?2 WHERE chunk_id = ?1") + .map_err(sqlite_error)?; + let mut last: Option> = None; + loop { + checkpoint(control)?; + let window: Vec<(Vec, i64)> = postings::scan_window( + transaction, + &first_sql, + &after_sql, + &last, + |key| vec![Value::Blob(key.clone())], + |row| Ok((row.get(0)?, row.get(1)?)), + )?; + let Some((final_key, _)) = window.last() else { + break; + }; + last = Some(final_key.clone()); + for (chunk_id, ordinal) in window { + checkpoint(control)?; + let mapped = documents + .map(u64::try_from(ordinal).map_err(contract_number)?)? + .ok_or_else(|| { + CodeLexicalArtifactErrorV1::Corrupt( + "replaced lexical chunk survived deletion".to_owned(), + ) + })?; + let mapped = i64::try_from(mapped).map_err(contract_number)?; + if update + .execute(params![chunk_id, mapped]) + .map_err(sqlite_error)? + != 1 + { + return Err(CodeLexicalArtifactErrorV1::Corrupt( + "carried lexical chunk changed during key shift".to_owned(), + )); + } + } + } + Ok(()) +} + /// `field_totals` holds each field's net change; a field nothing indexes /// any more has no row, as in a cold build. fn apply_field_totals( @@ -819,3 +912,6 @@ fn write_page_rows( } Ok(rows.len() as u64) } + +#[cfg(test)] +mod tests; diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs index 6b7408ab7dc..a0e48692d4b 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs @@ -89,7 +89,7 @@ pub(super) fn carry_clone_rows( [], ) .map_err(sqlite_error)?; - shift_rowid_keys(transaction, "clone_occurrences", "ordinal", clones)?; + shift_rowid_keys(transaction, "clone_occurrences", "ordinal", clones, control)?; let mut parent_payloads = BTreeMap::>::new(); { let mut statement = transaction diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs new file mode 100644 index 00000000000..f0f5bde6ea9 --- /dev/null +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs @@ -0,0 +1,312 @@ +use rusqlite::{Connection, ErrorCode, params}; +use tracedecay_code_index::production::CodeIndexExecutionControlV1; + +use super::{CodeLexicalArtifactErrorV1, ShiftV1, shift_row_chunk_documents, shift_rowid_keys}; + +struct Active; +impl CodeIndexExecutionControlV1 for Active { + fn is_cancelled(&self) -> bool { + false + } + + fn is_deadline_exceeded(&self) -> bool { + false + } +} + +fn fixture() -> (tempfile::TempDir, Connection) { + let directory = tempfile::tempdir().unwrap(); + let mut connection = Connection::open(directory.path().join("carry.sqlite")).unwrap(); + connection + .execute_batch( + "PRAGMA temp_store=FILE; + PRAGMA cache_size=8; + PRAGMA temp.cache_size=8; + CREATE TABLE row_blocks( + first_document INTEGER PRIMARY KEY, payload BLOB NOT NULL + ); + CREATE TABLE row_chunks( + chunk_id BLOB PRIMARY KEY, document_id INTEGER NOT NULL + ) WITHOUT ROWID; + CREATE TABLE independent(value TEXT NOT NULL); + INSERT INTO independent VALUES ('live');", + ) + .unwrap(); + let transaction = connection.transaction().unwrap(); + { + let mut block = transaction + .prepare("INSERT INTO row_blocks(first_document,payload) VALUES(?1,?2)") + .unwrap(); + let mut chunk = transaction + .prepare("INSERT INTO row_chunks(chunk_id,document_id) VALUES(?1,?2)") + .unwrap(); + for n in 1_i64..=16_000 { + block.execute(params![n, [7_u8; 32].as_slice()]).unwrap(); + chunk.execute(params![n.to_be_bytes().as_slice(), n]).unwrap(); + } + } + transaction.commit().unwrap(); + (directory, connection) +} + +fn blocks(c: &Connection) -> Vec<(i64, Vec)> { + let mut query = c + .prepare("SELECT first_document,payload FROM row_blocks ORDER BY first_document") + .unwrap(); + query + .query_map([], |r| Ok((r.get(0)?, r.get(1)?))) + .unwrap() + .collect::>() + .unwrap() +} + +fn chunks(c: &Connection) -> Vec<(Vec, i64)> { + let mut query = c + .prepare("SELECT chunk_id,document_id FROM row_chunks ORDER BY chunk_id") + .unwrap(); + query + .query_map([], |r| Ok((r.get(0)?, r.get(1)?))) + .unwrap() + .collect::>() + .unwrap() +} + +fn integrity(c: &Connection) { + for schema in ["main", "temp"] { + let actual: String = c + .query_row(&format!("PRAGMA {schema}.integrity_check"), [], |r| r.get(0)) + .unwrap(); + assert_eq!(actual, "ok", "{schema} remains intact"); + } +} + +#[test] +fn lexical_row_carry_is_disk_durable_and_collision_safe() { + let (directory, mut connection) = fixture(); + let original_blocks = blocks(&connection); + let original_chunks = chunks(&connection); + assert_eq!(original_blocks.len(), 16_000); + assert_eq!(original_chunks.len(), 16_000); + let shifts = [ + ShiftV1::new([((11, 20), 5), ((36, 39), 6)]).unwrap(), + ShiftV1::new([((11, 20), 12), ((36, 39), 1)]).unwrap(), + ]; + // The old positive-key update really collides without negative staging. + { + let tx = connection.transaction().unwrap(); + let error = tx + .execute("UPDATE row_blocks SET first_document=first_document+1", []) + .unwrap_err(); + assert_eq!(error.sqlite_error_code(), Some(ErrorCode::ConstraintViolation)); + tx.rollback().unwrap(); + } + for (index, shift) in shifts.iter().enumerate() { + let expected_blocks: Vec<_> = original_blocks + .iter() + .filter_map(|(key, blob)| { + shift + .map(u64::try_from(*key).unwrap()) + .unwrap() + .map(|mapped| (i64::try_from(mapped).unwrap(), blob.clone())) + }) + .collect(); + let expected_chunks: Vec<_> = original_chunks + .iter() + .filter_map(|(key, document)| { + shift + .map(u64::try_from(*document).unwrap()) + .unwrap() + .map(|mapped| (key.clone(), i64::try_from(mapped).unwrap())) + }) + .collect(); + let tx = connection.transaction().unwrap(); + for (table, column) in [("row_blocks", "first_document"), ("row_chunks", "document_id")] { + tx.execute( + &format!("DELETE FROM {table} WHERE {}", shift.sql_replaced(column)), + [], + ) + .unwrap(); + } + shift_rowid_keys(&tx, "row_blocks", "first_document", shift, &Active).unwrap(); + shift_row_chunk_documents(&tx, shift, &Active).unwrap(); + assert_eq!(blocks(&tx), expected_blocks); + assert_eq!(chunks(&tx), expected_chunks); + assert_eq!( + tx.pragma_query_value(Some("temp"), "page_count", |r| r.get::<_, i64>(0)) + .unwrap(), + 0 + ); + integrity(&tx); + if index == 0 { + tx.rollback().unwrap(); + } else { + tx.commit().unwrap(); + } + drop(connection); + connection = Connection::open(directory.path().join("carry.sqlite")).unwrap(); + assert_eq!( + blocks(&connection), + if index == 0 { + original_blocks.clone() + } else { + expected_blocks + } + ); + assert_eq!( + chunks(&connection), + if index == 0 { + original_chunks.clone() + } else { + expected_chunks + } + ); + let live: String = connection + .query_row("SELECT value FROM independent", [], |r| r.get(0)) + .unwrap(); + assert_eq!(live, "live"); + integrity(&connection); + } +} + +#[test] +fn interrupted_row_carry_rolls_back_both_key_families() { + let (directory, mut connection) = fixture(); + let original_blocks = blocks(&connection); + let original_chunks = chunks(&connection); + let shift = ShiftV1::new([((1, 1), 1)]).unwrap(); + struct CancelAfter { + probes: std::sync::atomic::AtomicUsize, + limit: usize, + } + impl CodeIndexExecutionControlV1 for CancelAfter { + fn is_cancelled(&self) -> bool { + self.probes.fetch_add(1, std::sync::atomic::Ordering::Relaxed) >= self.limit + } + + fn is_deadline_exceeded(&self) -> bool { + false + } + } + for chunk_family in [false, true] { + let tx = connection.transaction().unwrap(); + let cancel = CancelAfter { + probes: std::sync::atomic::AtomicUsize::new(0), + limit: 4_100, + }; + let error = if chunk_family { + shift_row_chunk_documents(&tx, &shift, &cancel).unwrap_err() + } else { + shift_rowid_keys(&tx, "row_blocks", "first_document", &shift, &cancel).unwrap_err() + }; + assert!(matches!(error, CodeLexicalArtifactErrorV1::Interrupted(_))); + if chunk_family { + assert_ne!(chunks(&tx), original_chunks); + } else { + assert_ne!(blocks(&tx), original_blocks); + } + integrity(&tx); + drop(tx); + drop(connection); + connection = Connection::open(directory.path().join("carry.sqlite")).unwrap(); + assert_eq!(blocks(&connection), original_blocks); + assert_eq!(chunks(&connection), original_chunks); + integrity(&connection); + } +} + +#[cfg(target_os = "linux")] +#[test] +fn lexical_row_carry_avoids_live_unlinked_sqlite_spills() { + use std::io; + use std::process::Command; + use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + use std::sync::Arc; + + fn spill_bytes() -> io::Result { + let mut total = 0; + for entry in std::fs::read_dir("/proc/self/fd")? { + let path = entry?.path(); + let target = match std::fs::read_link(&path) { + Ok(target) => target, + Err(error) if error.kind() == io::ErrorKind::NotFound => continue, + Err(error) => return Err(error), + }; + if target + .file_name() + .is_some_and(|name| name.to_string_lossy().starts_with("etilqs_")) + { + total += std::fs::metadata(path)?.len(); + } + } + Ok(total) + } + + const CHILD: &str = "TRACEDECAY_ROW_CARRY_SPILL_CHILD"; + if std::env::var_os(CHILD).is_none() { + let output = Command::new(std::env::current_exe().unwrap()) + .arg("lexical_row_carry_avoids_live_unlinked_sqlite_spills") + .arg("--test-threads=1") + .arg("--nocapture") + .env(CHILD, "1") + .output() + .unwrap(); + let stdout = String::from_utf8_lossy(&output.stdout); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(output.status.success(), "{stdout}\n{stderr}"); + assert!(stdout.contains("1 passed"), "child test was vacuous: {stdout}"); + return; + } + let (_directory, mut connection) = fixture(); + let shift = ShiftV1::new([((1, 1), 1)]).unwrap(); + let original_blocks = blocks(&connection); + let original_chunks = chunks(&connection); + let peak = Arc::new(AtomicU64::new(0)); + let unreadable = Arc::new(AtomicBool::new(false)); + let observer_peak = Arc::clone(&peak); + let observer_failed = Arc::clone(&unreadable); + connection.progress_handler( + 1_024, + Some(move || match spill_bytes() { + Ok(bytes) => { + observer_peak.fetch_max(bytes, Ordering::Relaxed); + false + } + Err(_) => { + observer_failed.store(true, Ordering::Relaxed); + true + } + }), + ); + { + let tx = connection.transaction().unwrap(); + tx.execute( + "UPDATE row_blocks SET first_document=-first_document-1 WHERE first_document>0", + [], + ) + .unwrap(); + tx.rollback().unwrap(); + } + assert!(!unreadable.load(Ordering::Relaxed)); + assert!(peak.load(Ordering::Relaxed) > 0, "old bulk update must spill"); + assert_eq!( + connection.pragma_query_value(Some("temp"), "page_count", |r| r.get::<_, i64>(0)).unwrap(), + 0, + "TEMP pager counts cannot observe the transient statement spill" + ); + assert_eq!(blocks(&connection), original_blocks); + assert_eq!(spill_bytes().unwrap(), 0); + peak.store(0, Ordering::Relaxed); + { + let tx = connection.transaction().unwrap(); + shift_rowid_keys(&tx, "row_blocks", "first_document", &shift, &Active).unwrap(); + shift_row_chunk_documents(&tx, &shift, &Active).unwrap(); + tx.commit().unwrap(); + } + connection.progress_handler(0, None:: bool>); + assert!(!unreadable.load(Ordering::Relaxed)); + assert_eq!(peak.load(Ordering::Relaxed), 0, "bounded remap must not spill"); + assert_eq!(spill_bytes().unwrap(), 0); + assert_ne!(blocks(&connection), original_blocks); + assert_ne!(chunks(&connection), original_chunks); + integrity(&connection); +} From 2eed865ff2defb40e05d9c368689e3f55cf32b62 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 15:51:27 -0800 Subject: [PATCH 06/14] fix(sqlite): isolate measured rowid spill repair Refs #4152 Independent FILE-backed SQLite probes show the bulk two-phase rowid key rewrite spills an unlinked statement-local index, whereas the non-key chunk ordinal update in production does not. Keep the proven rowid fix but restore row_chunks to its existing optimized update. Exercise both real rowid tables, cancellation, rollback and reopen. Correct the Linux spill positive control to execute both bulk passes; the transient spill occurs during the second pass. --- .../lexical/projection/artifact/increment.rs | 71 ++---------- .../projection/artifact/increment/tests.rs | 108 ++++++++++-------- 2 files changed, 73 insertions(+), 106 deletions(-) diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs index c1f9ea584de..bffacaee720 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs @@ -747,7 +747,17 @@ fn carry_row_chunks( [], ) .map_err(sqlite_error)?; - shift_row_chunk_documents(transaction, documents, control)?; + if let Some(first) = documents.first_affected().filter(|_| documents.shifts()) { + transaction + .execute( + &format!( + "UPDATE row_chunks SET document_id = {} WHERE document_id >= {first}", + documents.sql_image("document_id") + ), + [], + ) + .map_err(sqlite_error)?; + } let mut insert = transaction .prepare_cached("INSERT INTO row_chunks(chunk_id, document_id) VALUES (?1, ?2)") .map_err(sqlite_error)?; @@ -765,65 +775,6 @@ fn carry_row_chunks( Ok(()) } -/// Row chunks are keyed by chunk_id, not document_id. A bulk UPDATE of their -/// document ordinals can spill SQLite's statement-local key index. The -/// primary-key window remains stable because chunk_id never changes. -fn shift_row_chunk_documents( - transaction: &Transaction<'_>, - documents: &ShiftV1, - control: &dyn CodeIndexExecutionControlV1, -) -> Result<(), CodeLexicalArtifactErrorV1> { - let Some(first) = documents.first_affected().filter(|_| documents.shifts()) else { - return Ok(()); - }; - let first_sql = format!( - "SELECT chunk_id, document_id FROM row_chunks WHERE document_id >= {first} ORDER BY chunk_id LIMIT ?1" - ); - let after_sql = format!( - "SELECT chunk_id, document_id FROM row_chunks WHERE document_id >= {first} AND chunk_id > ?1 ORDER BY chunk_id LIMIT ?2" - ); - let mut update = transaction - .prepare_cached("UPDATE row_chunks SET document_id = ?2 WHERE chunk_id = ?1") - .map_err(sqlite_error)?; - let mut last: Option> = None; - loop { - checkpoint(control)?; - let window: Vec<(Vec, i64)> = postings::scan_window( - transaction, - &first_sql, - &after_sql, - &last, - |key| vec![Value::Blob(key.clone())], - |row| Ok((row.get(0)?, row.get(1)?)), - )?; - let Some((final_key, _)) = window.last() else { - break; - }; - last = Some(final_key.clone()); - for (chunk_id, ordinal) in window { - checkpoint(control)?; - let mapped = documents - .map(u64::try_from(ordinal).map_err(contract_number)?)? - .ok_or_else(|| { - CodeLexicalArtifactErrorV1::Corrupt( - "replaced lexical chunk survived deletion".to_owned(), - ) - })?; - let mapped = i64::try_from(mapped).map_err(contract_number)?; - if update - .execute(params![chunk_id, mapped]) - .map_err(sqlite_error)? - != 1 - { - return Err(CodeLexicalArtifactErrorV1::Corrupt( - "carried lexical chunk changed during key shift".to_owned(), - )); - } - } - } - Ok(()) -} - /// `field_totals` holds each field's net change; a field nothing indexes /// any more has no row, as in a cold build. fn apply_field_totals( diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs index f0f5bde6ea9..f1a0dc849d0 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs @@ -1,7 +1,7 @@ use rusqlite::{Connection, ErrorCode, params}; use tracedecay_code_index::production::CodeIndexExecutionControlV1; -use super::{CodeLexicalArtifactErrorV1, ShiftV1, shift_row_chunk_documents, shift_rowid_keys}; +use super::{CodeLexicalArtifactErrorV1, ShiftV1, shift_rowid_keys}; struct Active; impl CodeIndexExecutionControlV1 for Active { @@ -25,9 +25,11 @@ fn fixture() -> (tempfile::TempDir, Connection) { CREATE TABLE row_blocks( first_document INTEGER PRIMARY KEY, payload BLOB NOT NULL ); - CREATE TABLE row_chunks( - chunk_id BLOB PRIMARY KEY, document_id INTEGER NOT NULL - ) WITHOUT ROWID; + CREATE TABLE clone_occurrences( + ordinal INTEGER PRIMARY KEY, + symbol_key BLOB NOT NULL UNIQUE, + payload_ordinal INTEGER NOT NULL + ); CREATE TABLE independent(value TEXT NOT NULL); INSERT INTO independent VALUES ('live');", ) @@ -37,12 +39,17 @@ fn fixture() -> (tempfile::TempDir, Connection) { let mut block = transaction .prepare("INSERT INTO row_blocks(first_document,payload) VALUES(?1,?2)") .unwrap(); - let mut chunk = transaction - .prepare("INSERT INTO row_chunks(chunk_id,document_id) VALUES(?1,?2)") + let mut occurrence = transaction + .prepare( + "INSERT INTO clone_occurrences(ordinal,symbol_key,payload_ordinal) + VALUES(?1,?2,?3)", + ) .unwrap(); for n in 1_i64..=16_000 { block.execute(params![n, [7_u8; 32].as_slice()]).unwrap(); - chunk.execute(params![n.to_be_bytes().as_slice(), n]).unwrap(); + occurrence + .execute(params![n, n.to_be_bytes().as_slice(), n % 31]) + .unwrap(); } } transaction.commit().unwrap(); @@ -60,12 +67,15 @@ fn blocks(c: &Connection) -> Vec<(i64, Vec)> { .unwrap() } -fn chunks(c: &Connection) -> Vec<(Vec, i64)> { +fn occurrences(c: &Connection) -> Vec<(i64, Vec, i64)> { let mut query = c - .prepare("SELECT chunk_id,document_id FROM row_chunks ORDER BY chunk_id") + .prepare( + "SELECT ordinal,symbol_key,payload_ordinal + FROM clone_occurrences ORDER BY ordinal", + ) .unwrap(); query - .query_map([], |r| Ok((r.get(0)?, r.get(1)?))) + .query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?))) .unwrap() .collect::>() .unwrap() @@ -81,17 +91,17 @@ fn integrity(c: &Connection) { } #[test] -fn lexical_row_carry_is_disk_durable_and_collision_safe() { +fn lexical_rowid_carry_preserves_disk_rows_across_replacements() { let (directory, mut connection) = fixture(); let original_blocks = blocks(&connection); - let original_chunks = chunks(&connection); + let original_occurrences = occurrences(&connection); assert_eq!(original_blocks.len(), 16_000); - assert_eq!(original_chunks.len(), 16_000); + assert_eq!(original_occurrences.len(), 16_000); let shifts = [ ShiftV1::new([((11, 20), 5), ((36, 39), 6)]).unwrap(), ShiftV1::new([((11, 20), 12), ((36, 39), 1)]).unwrap(), ]; - // The old positive-key update really collides without negative staging. + // Positive collision control: old in-place positive keys cannot just move. { let tx = connection.transaction().unwrap(); let error = tx @@ -110,27 +120,29 @@ fn lexical_row_carry_is_disk_durable_and_collision_safe() { .map(|mapped| (i64::try_from(mapped).unwrap(), blob.clone())) }) .collect(); - let expected_chunks: Vec<_> = original_chunks + let expected_occurrences: Vec<_> = original_occurrences .iter() - .filter_map(|(key, document)| { + .filter_map(|(key, symbol, payload)| { shift - .map(u64::try_from(*document).unwrap()) + .map(u64::try_from(*key).unwrap()) .unwrap() - .map(|mapped| (key.clone(), i64::try_from(mapped).unwrap())) + .map(|mapped| (i64::try_from(mapped).unwrap(), symbol.clone(), *payload)) }) .collect(); let tx = connection.transaction().unwrap(); - for (table, column) in [("row_blocks", "first_document"), ("row_chunks", "document_id")] { + for (table, column) in [ + ("row_blocks", "first_document"), + ("clone_occurrences", "ordinal"), + ] { tx.execute( &format!("DELETE FROM {table} WHERE {}", shift.sql_replaced(column)), [], ) .unwrap(); + shift_rowid_keys(&tx, table, column, shift, &Active).unwrap(); } - shift_rowid_keys(&tx, "row_blocks", "first_document", shift, &Active).unwrap(); - shift_row_chunk_documents(&tx, shift, &Active).unwrap(); assert_eq!(blocks(&tx), expected_blocks); - assert_eq!(chunks(&tx), expected_chunks); + assert_eq!(occurrences(&tx), expected_occurrences); assert_eq!( tx.pragma_query_value(Some("temp"), "page_count", |r| r.get::<_, i64>(0)) .unwrap(), @@ -153,11 +165,11 @@ fn lexical_row_carry_is_disk_durable_and_collision_safe() { } ); assert_eq!( - chunks(&connection), + occurrences(&connection), if index == 0 { - original_chunks.clone() + original_occurrences.clone() } else { - expected_chunks + expected_occurrences } ); let live: String = connection @@ -169,10 +181,10 @@ fn lexical_row_carry_is_disk_durable_and_collision_safe() { } #[test] -fn interrupted_row_carry_rolls_back_both_key_families() { +fn interrupted_rowid_carry_rolls_back_partial_key_staging() { let (directory, mut connection) = fixture(); let original_blocks = blocks(&connection); - let original_chunks = chunks(&connection); + let original_occurrences = occurrences(&connection); let shift = ShiftV1::new([((1, 1), 1)]).unwrap(); struct CancelAfter { probes: std::sync::atomic::AtomicUsize, @@ -187,20 +199,21 @@ fn interrupted_row_carry_rolls_back_both_key_families() { false } } - for chunk_family in [false, true] { + for occurrences_family in [false, true] { let tx = connection.transaction().unwrap(); let cancel = CancelAfter { probes: std::sync::atomic::AtomicUsize::new(0), limit: 4_100, }; - let error = if chunk_family { - shift_row_chunk_documents(&tx, &shift, &cancel).unwrap_err() + let (table, column) = if occurrences_family { + ("clone_occurrences", "ordinal") } else { - shift_rowid_keys(&tx, "row_blocks", "first_document", &shift, &cancel).unwrap_err() + ("row_blocks", "first_document") }; + let error = shift_rowid_keys(&tx, table, column, &shift, &cancel).unwrap_err(); assert!(matches!(error, CodeLexicalArtifactErrorV1::Interrupted(_))); - if chunk_family { - assert_ne!(chunks(&tx), original_chunks); + if occurrences_family { + assert_ne!(occurrences(&tx), original_occurrences); } else { assert_ne!(blocks(&tx), original_blocks); } @@ -209,18 +222,18 @@ fn interrupted_row_carry_rolls_back_both_key_families() { drop(connection); connection = Connection::open(directory.path().join("carry.sqlite")).unwrap(); assert_eq!(blocks(&connection), original_blocks); - assert_eq!(chunks(&connection), original_chunks); + assert_eq!(occurrences(&connection), original_occurrences); integrity(&connection); } } #[cfg(target_os = "linux")] #[test] -fn lexical_row_carry_avoids_live_unlinked_sqlite_spills() { +fn lexical_rowid_shift_does_not_spill_unlinked_sqlite_files() { use std::io; use std::process::Command; - use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::Arc; + use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; fn spill_bytes() -> io::Result { let mut total = 0; @@ -241,10 +254,10 @@ fn lexical_row_carry_avoids_live_unlinked_sqlite_spills() { Ok(total) } - const CHILD: &str = "TRACEDECAY_ROW_CARRY_SPILL_CHILD"; + const CHILD: &str = "TRACEDECAY_ROWID_CARRY_SPILL_CHILD"; if std::env::var_os(CHILD).is_none() { let output = Command::new(std::env::current_exe().unwrap()) - .arg("lexical_row_carry_avoids_live_unlinked_sqlite_spills") + .arg("lexical_rowid_shift_does_not_spill_unlinked_sqlite_files") .arg("--test-threads=1") .arg("--nocapture") .env(CHILD, "1") @@ -258,8 +271,7 @@ fn lexical_row_carry_avoids_live_unlinked_sqlite_spills() { } let (_directory, mut connection) = fixture(); let shift = ShiftV1::new([((1, 1), 1)]).unwrap(); - let original_blocks = blocks(&connection); - let original_chunks = chunks(&connection); + let original = blocks(&connection); let peak = Arc::new(AtomicU64::new(0)); let unreadable = Arc::new(AtomicBool::new(false)); let observer_peak = Arc::clone(&peak); @@ -284,29 +296,33 @@ fn lexical_row_carry_avoids_live_unlinked_sqlite_spills() { [], ) .unwrap(); + // The second bulk UPDATE materializes the transient rowid set. + tx.execute( + "UPDATE row_blocks SET first_document=-first_document-1 WHERE first_document<0", + [], + ) + .unwrap(); tx.rollback().unwrap(); } assert!(!unreadable.load(Ordering::Relaxed)); - assert!(peak.load(Ordering::Relaxed) > 0, "old bulk update must spill"); + assert!(peak.load(Ordering::Relaxed) > 0, "old bulk rowid rewrite must spill"); assert_eq!( connection.pragma_query_value(Some("temp"), "page_count", |r| r.get::<_, i64>(0)).unwrap(), 0, "TEMP pager counts cannot observe the transient statement spill" ); - assert_eq!(blocks(&connection), original_blocks); + assert_eq!(blocks(&connection), original); assert_eq!(spill_bytes().unwrap(), 0); peak.store(0, Ordering::Relaxed); { let tx = connection.transaction().unwrap(); shift_rowid_keys(&tx, "row_blocks", "first_document", &shift, &Active).unwrap(); - shift_row_chunk_documents(&tx, &shift, &Active).unwrap(); tx.commit().unwrap(); } connection.progress_handler(0, None:: bool>); assert!(!unreadable.load(Ordering::Relaxed)); - assert_eq!(peak.load(Ordering::Relaxed), 0, "bounded remap must not spill"); + assert_eq!(peak.load(Ordering::Relaxed), 0, "bounded rowid shift must not spill"); assert_eq!(spill_bytes().unwrap(), 0); - assert_ne!(blocks(&connection), original_blocks); - assert_ne!(chunks(&connection), original_chunks); + assert_ne!(blocks(&connection), original); integrity(&connection); } From 21b2fd47a3616bf8436d6cd02f447130500e48ca Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 16:40:06 -0800 Subject: [PATCH 07/14] fix(sqlite): repair clone TEMP regression and Bazel formatting Refs #4152 Apply the exact Rustfmt hunks reported by the hosted Bazel gate. Restore the real pre-fix bulk SQL in the clone spill positive control: the shared rowid helper now takes execution control and no longer performs bulk updates. This keeps the Linux physical-spill assertion nonvacuous without weakening any tests. --- .../lexical/projection/artifact/increment.rs | 13 ++++--- .../artifact/increment/clones/tests.rs | 36 +++++++++++++------ .../projection/artifact/increment/tests.rs | 33 +++++++++++++---- 3 files changed, 61 insertions(+), 21 deletions(-) diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs index bffacaee720..286d49aff12 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs @@ -656,9 +656,8 @@ fn shift_rowid_keys( } else { format!("{column} < 0") }; - let first_sql = format!( - "SELECT {column} FROM {table} WHERE {predicate} ORDER BY {column} LIMIT ?1" - ); + let first_sql = + format!("SELECT {column} FROM {table} WHERE {predicate} ORDER BY {column} LIMIT ?1"); let after_sql = format!( "SELECT {column} FROM {table} WHERE {predicate} AND {column} > ?1 ORDER BY {column} LIMIT ?2" ); @@ -717,7 +716,13 @@ fn carry_row_blocks( [], ) .map_err(sqlite_error)?; - shift_rowid_keys(transaction, "row_blocks", "first_document", documents, control)?; + shift_rowid_keys( + transaction, + "row_blocks", + "first_document", + documents, + control, + )?; let mut insert = transaction .prepare_cached("INSERT INTO row_blocks(first_document, payload) VALUES (?1, ?2)") .map_err(sqlite_error)?; diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs index 7e365e59574..d95929c37c0 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs @@ -212,7 +212,10 @@ fn carried_clone_keys_shift_without_retained_file_backed_temp_pages() { let retained_pages: i64 = connection .pragma_query_value(Some("temp"), "page_count", |row| row.get(0)) .unwrap(); - assert!(retained_pages > 32, "positive TEMP-copy control: {retained_pages}"); + assert!( + retained_pages > 32, + "positive TEMP-copy control: {retained_pages}" + ); integrity(&connection, "main"); integrity(&connection, "temp"); } @@ -275,7 +278,6 @@ mod physical_spill { use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; - use super::super::super::shift_rowid_keys; use super::{ ActiveControl, ShiftV1, expected_postings, fixture, postings, shift_clone_posting_keys, }; @@ -314,7 +316,10 @@ mod physical_spill { let stdout = String::from_utf8_lossy(&output.stdout); let stderr = String::from_utf8_lossy(&output.stderr); assert!(output.status.success(), "{stdout}\n{stderr}"); - assert!(stdout.contains("1 passed"), "child did not execute: {stdout}"); + assert!( + stdout.contains("1 passed"), + "child did not execute: {stdout}" + ); return; } let (_root, mut connection) = fixture(); @@ -339,13 +344,24 @@ mod physical_spill { ); { let transaction = connection.transaction().unwrap(); - shift_rowid_keys( - &transaction, - "clone_exact_postings", - "occurrence_ordinal", - &shift, - ) - .unwrap(); + // Execute the actual pre-fix, bulk two-phase SQL as a positive + // control. The production rowid helper is intentionally bounded now. + transaction + .execute( + "UPDATE clone_exact_postings + SET occurrence_ordinal = -occurrence_ordinal - 2 + WHERE occurrence_ordinal > 0", + [], + ) + .unwrap(); + transaction + .execute( + "UPDATE clone_exact_postings + SET occurrence_ordinal = -occurrence_ordinal - 1 + WHERE occurrence_ordinal < 0", + [], + ) + .unwrap(); transaction.rollback().unwrap(); } assert!(!unavailable.load(Ordering::Relaxed)); diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs index f1a0dc849d0..7c98f93237f 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs @@ -84,7 +84,9 @@ fn occurrences(c: &Connection) -> Vec<(i64, Vec, i64)> { fn integrity(c: &Connection) { for schema in ["main", "temp"] { let actual: String = c - .query_row(&format!("PRAGMA {schema}.integrity_check"), [], |r| r.get(0)) + .query_row(&format!("PRAGMA {schema}.integrity_check"), [], |r| { + r.get(0) + }) .unwrap(); assert_eq!(actual, "ok", "{schema} remains intact"); } @@ -107,7 +109,10 @@ fn lexical_rowid_carry_preserves_disk_rows_across_replacements() { let error = tx .execute("UPDATE row_blocks SET first_document=first_document+1", []) .unwrap_err(); - assert_eq!(error.sqlite_error_code(), Some(ErrorCode::ConstraintViolation)); + assert_eq!( + error.sqlite_error_code(), + Some(ErrorCode::ConstraintViolation) + ); tx.rollback().unwrap(); } for (index, shift) in shifts.iter().enumerate() { @@ -192,7 +197,9 @@ fn interrupted_rowid_carry_rolls_back_partial_key_staging() { } impl CodeIndexExecutionControlV1 for CancelAfter { fn is_cancelled(&self) -> bool { - self.probes.fetch_add(1, std::sync::atomic::Ordering::Relaxed) >= self.limit + self.probes + .fetch_add(1, std::sync::atomic::Ordering::Relaxed) + >= self.limit } fn is_deadline_exceeded(&self) -> bool { @@ -266,7 +273,10 @@ fn lexical_rowid_shift_does_not_spill_unlinked_sqlite_files() { let stdout = String::from_utf8_lossy(&output.stdout); let stderr = String::from_utf8_lossy(&output.stderr); assert!(output.status.success(), "{stdout}\n{stderr}"); - assert!(stdout.contains("1 passed"), "child test was vacuous: {stdout}"); + assert!( + stdout.contains("1 passed"), + "child test was vacuous: {stdout}" + ); return; } let (_directory, mut connection) = fixture(); @@ -305,9 +315,14 @@ fn lexical_rowid_shift_does_not_spill_unlinked_sqlite_files() { tx.rollback().unwrap(); } assert!(!unreadable.load(Ordering::Relaxed)); - assert!(peak.load(Ordering::Relaxed) > 0, "old bulk rowid rewrite must spill"); + assert!( + peak.load(Ordering::Relaxed) > 0, + "old bulk rowid rewrite must spill" + ); assert_eq!( - connection.pragma_query_value(Some("temp"), "page_count", |r| r.get::<_, i64>(0)).unwrap(), + connection + .pragma_query_value(Some("temp"), "page_count", |r| r.get::<_, i64>(0)) + .unwrap(), 0, "TEMP pager counts cannot observe the transient statement spill" ); @@ -321,7 +336,11 @@ fn lexical_rowid_shift_does_not_spill_unlinked_sqlite_files() { } connection.progress_handler(0, None:: bool>); assert!(!unreadable.load(Ordering::Relaxed)); - assert_eq!(peak.load(Ordering::Relaxed), 0, "bounded rowid shift must not spill"); + assert_eq!( + peak.load(Ordering::Relaxed), + 0, + "bounded rowid shift must not spill" + ); assert_eq!(spill_bytes().unwrap(), 0); assert_ne!(blocks(&connection), original); integrity(&connection); From 45d908d39881450ccc768481605bbc7f1356608a Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 17:33:26 -0800 Subject: [PATCH 08/14] fix(sqlite): bound cancellable row chunk carry updates Refs #4152 Traverse stable chunk primary keys in bounded windows with explicit cancellation and deadline checkpoints. Preserve the canonical document shift and skip unchanged values to avoid unnecessary writes. Exercise 12000 real on-disk rows across mixed shift ranges; interrupt a partly completed suffix remap for both stop reasons, prove rollback and fresh-connection commit integrity, and preserve foreign live data. --- .../lexical/projection/artifact/increment.rs | 58 +++++++-- .../projection/artifact/increment/tests.rs | 113 +++++++++++++++++- 2 files changed, 163 insertions(+), 8 deletions(-) diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs index 286d49aff12..9a41193b289 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs @@ -753,15 +753,59 @@ fn carry_row_chunks( ) .map_err(sqlite_error)?; if let Some(first) = documents.first_affected().filter(|_| documents.shifts()) { - transaction - .execute( - &format!( - "UPDATE row_chunks SET document_id = {} WHERE document_id >= {first}", - documents.sql_image("document_id") - ), - [], + // The chunk ID is the stable primary key; the document ID is only a + // value. Seek primary-key windows so a large carry can observe + // cancellation between updates without creating a sorted TEMP copy. + let mut update = transaction + .prepare_cached( + "UPDATE row_chunks SET document_id = ?3 WHERE chunk_id = ?1 AND document_id = ?2", ) .map_err(sqlite_error)?; + let first_sql = format!( + "SELECT chunk_id, document_id FROM row_chunks WHERE document_id >= {first} ORDER BY chunk_id LIMIT ?1" + ); + let after_sql = format!( + "SELECT chunk_id, document_id FROM row_chunks WHERE document_id >= {first} AND chunk_id > ?1 ORDER BY chunk_id LIMIT ?2" + ); + let mut last: Option> = None; + loop { + checkpoint(control)?; + let window: Vec<(Vec, i64)> = postings::scan_window( + transaction, + &first_sql, + &after_sql, + &last, + |chunk_id| vec![Value::Blob(chunk_id.clone())], + |row| Ok((row.get(0)?, row.get(1)?)), + )?; + let Some((final_chunk, _)) = window.last() else { + break; + }; + last = Some(final_chunk.clone()); + for (chunk_id, old_document) in window { + checkpoint(control)?; + let mapped = documents + .map(u64::try_from(old_document).map_err(contract_number)?)? + .ok_or_else(|| { + CodeLexicalArtifactErrorV1::Corrupt( + "replaced lexical chunk survived deletion".to_owned(), + ) + })?; + let mapped = i64::try_from(mapped).map_err(contract_number)?; + if mapped == old_document { + continue; + } + if update + .execute(params![chunk_id, old_document, mapped]) + .map_err(sqlite_error)? + != 1 + { + return Err(CodeLexicalArtifactErrorV1::Corrupt( + "carried lexical chunk changed during remap".to_owned(), + )); + } + } + } } let mut insert = transaction .prepare_cached("INSERT INTO row_chunks(chunk_id, document_id) VALUES (?1, ?2)") diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs index 7c98f93237f..5fda9b21a80 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs @@ -1,7 +1,7 @@ use rusqlite::{Connection, ErrorCode, params}; use tracedecay_code_index::production::CodeIndexExecutionControlV1; -use super::{CodeLexicalArtifactErrorV1, ShiftV1, shift_rowid_keys}; +use super::{CodeLexicalArtifactErrorV1, ShiftV1, carry_row_chunks, shift_rowid_keys}; struct Active; impl CodeIndexExecutionControlV1 for Active { @@ -345,3 +345,114 @@ fn lexical_rowid_shift_does_not_spill_unlinked_sqlite_files() { assert_ne!(blocks(&connection), original); integrity(&connection); } + +#[test] +fn carried_row_chunks_check_cancellation_and_preserve_disk_rows() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + struct StopAfter { + checks: AtomicUsize, + deadline: bool, + } + impl CodeIndexExecutionControlV1 for StopAfter { + fn is_cancelled(&self) -> bool { + !self.deadline && self.checks.fetch_add(1, Ordering::Relaxed) >= 4_100 + } + + fn is_deadline_exceeded(&self) -> bool { + self.deadline && self.checks.fetch_add(1, Ordering::Relaxed) >= 4_100 + } + } + + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("row-chunks.sqlite"); + let mut connection = Connection::open(&path).unwrap(); + connection + .execute_batch( + "PRAGMA temp_store=FILE; + PRAGMA cache_size=8; + CREATE TABLE row_chunks( + chunk_id BLOB NOT NULL PRIMARY KEY, + document_id INTEGER NOT NULL + ) WITHOUT ROWID; + CREATE TABLE independent(value TEXT NOT NULL); + INSERT INTO independent VALUES ('live');", + ) + .unwrap(); + { + let tx = connection.transaction().unwrap(); + let mut insert = tx + .prepare("INSERT INTO row_chunks(chunk_id, document_id) VALUES (?1, ?2)") + .unwrap(); + for document in 1_i64..=12_000 { + insert + .execute(params![document.to_be_bytes().repeat(4), document]) + .unwrap(); + } + drop(insert); + tx.commit().unwrap(); + } + + let chunks = |db: &Connection| -> Vec<(Vec, i64)> { + let mut query = db + .prepare("SELECT chunk_id, document_id FROM row_chunks ORDER BY chunk_id") + .unwrap(); + query + .query_map([], |row| Ok((row.get(0)?, row.get(1)?))) + .unwrap() + .collect::>() + .unwrap() + }; + let original = chunks(&connection); + assert_eq!(original.len(), 12_000); + let shift = ShiftV1::new([((8, 14), 4), ((45, 52), 10)]).unwrap(); + let expected: Vec<_> = original + .iter() + .filter_map(|(chunk, document)| { + shift + .map(u64::try_from(*document).unwrap()) + .unwrap() + .map(|mapped| (chunk.clone(), i64::try_from(mapped).unwrap())) + }) + .collect(); + assert_eq!(expected.len(), 11_987); + assert_ne!(expected, original); + + for deadline in [false, true] { + let tx = connection.transaction().unwrap(); + let control = StopAfter { + checks: AtomicUsize::new(0), + deadline, + }; + let error = carry_row_chunks(&tx, &shift, &[], &control).unwrap_err(); + assert!( + matches!(error, CodeLexicalArtifactErrorV1::Interrupted(_)), + "both cancellation and deadline must interrupt the real remap: {error:?}" + ); + let partial = chunks(&tx); + assert_ne!(partial, original, "the interrupted carry did actual writes"); + assert_ne!(partial, expected, "it stopped before completing the remap"); + integrity(&tx); + tx.rollback().unwrap(); + drop(connection); + connection = Connection::open(&path).unwrap(); + assert_eq!(chunks(&connection), original, "rollback must restore every chunk"); + integrity(&connection); + } + + let tx = connection.transaction().unwrap(); + carry_row_chunks(&tx, &shift, &[], &Active).unwrap(); + assert_eq!(chunks(&tx), expected); + integrity(&tx); + tx.commit().unwrap(); + drop(connection); + let reopened = Connection::open(&path).unwrap(); + assert_eq!(chunks(&reopened), expected, "commit must survive a fresh connection"); + assert_eq!( + reopened + .query_row("SELECT value FROM independent", [], |row| row.get::<_, String>(0)) + .unwrap(), + "live" + ); + integrity(&reopened); +} From 407e488ca02e490598a29db75cb702c9794ca683 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 17:35:29 -0800 Subject: [PATCH 09/14] test(sqlite): independently verify chunk remap boundaries Refs #4152 Calculate expected durable chunk positions from the two concrete replacement ranges instead of calling the same ShiftV1 mapping as the production path. Keep the cancellation, rollback and disk-reopen checks. --- .../lexical/projection/artifact/increment/tests.rs | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs index 5fda9b21a80..bcb0f0644d0 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs @@ -406,13 +406,19 @@ fn carried_row_chunks_check_cancellation_and_preserve_disk_rows() { let original = chunks(&connection); assert_eq!(original.len(), 12_000); let shift = ShiftV1::new([((8, 14), 4), ((45, 52), 10)]).unwrap(); + // Reference mapping is deliberately independent of ShiftV1::map: + // two replaced ranges remove 13 old chunks, shifting the intervening + // documents down by two and the trailing suffix up by one. let expected: Vec<_> = original .iter() .filter_map(|(chunk, document)| { - shift - .map(u64::try_from(*document).unwrap()) - .unwrap() - .map(|mapped| (chunk.clone(), i64::try_from(mapped).unwrap())) + let mapped = match *document { + 8..=13 | 45..=51 => None, + 14..=44 => Some(*document - 2), + 52.. => Some(*document + 1), + _ => Some(*document), + }; + mapped.map(|mapped| (chunk.clone(), mapped)) }) .collect(); assert_eq!(expected.len(), 11_987); From 8631a6331935e5e997b24f679f4d7fbb2f5213be Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 18:27:52 -0800 Subject: [PATCH 10/14] fix(sqlite): preserve literal chunk keys during bounded carry Refs #4152 A bounded row_chunks scan decoded every chunk_id as Vec, but valid literal chunk IDs are SQLite TEXT, not BLOB. Preserve the exact SQLite storage class in seek cursors and point-update bindings so incremental rebuilds do not abort or change key identity. Cover 6,000 TEXT and 6,000 BLOB keys in the same on-disk rollback, cancellation, deadline, commit, and reopen regression. Correct the reviewed inline test import and the transient-spill control comment. --- .../lexical/projection/artifact/increment.rs | 6 ++--- .../artifact/increment/clones/tests.rs | 4 +-- .../projection/artifact/increment/tests.rs | 27 ++++++++++++++----- 3 files changed, 26 insertions(+), 11 deletions(-) diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs index 9a41193b289..2444a3d224d 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs @@ -767,15 +767,15 @@ fn carry_row_chunks( let after_sql = format!( "SELECT chunk_id, document_id FROM row_chunks WHERE document_id >= {first} AND chunk_id > ?1 ORDER BY chunk_id LIMIT ?2" ); - let mut last: Option> = None; + let mut last: Option = None; loop { checkpoint(control)?; - let window: Vec<(Vec, i64)> = postings::scan_window( + let window: Vec<(Value, i64)> = postings::scan_window( transaction, &first_sql, &after_sql, &last, - |chunk_id| vec![Value::Blob(chunk_id.clone())], + |chunk_id| vec![chunk_id.clone()], |row| Ok((row.get(0)?, row.get(1)?)), )?; let Some((final_chunk, _)) = window.last() else { diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs index d95929c37c0..0336f2c7d06 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs @@ -344,8 +344,8 @@ mod physical_spill { ); { let transaction = connection.transaction().unwrap(); - // Execute the actual pre-fix, bulk two-phase SQL as a positive - // control. The production rowid helper is intentionally bounded now. + // Bulk key updates are the positive control for statement-local + // spills that bounded production key updates avoid. transaction .execute( "UPDATE clone_exact_postings diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs index bcb0f0644d0..ec6d0695b04 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs @@ -1,3 +1,6 @@ +use std::sync::atomic::{AtomicUsize, Ordering}; + +use rusqlite::types::Value; use rusqlite::{Connection, ErrorCode, params}; use tracedecay_code_index::production::CodeIndexExecutionControlV1; @@ -348,8 +351,6 @@ fn lexical_rowid_shift_does_not_spill_unlinked_sqlite_files() { #[test] fn carried_row_chunks_check_cancellation_and_preserve_disk_rows() { - use std::sync::atomic::{AtomicUsize, Ordering}; - struct StopAfter { checks: AtomicUsize, deadline: bool, @@ -385,15 +386,20 @@ fn carried_row_chunks_check_cancellation_and_preserve_disk_rows() { .prepare("INSERT INTO row_chunks(chunk_id, document_id) VALUES (?1, ?2)") .unwrap(); for document in 1_i64..=12_000 { - insert - .execute(params![document.to_be_bytes().repeat(4), document]) - .unwrap(); + // Canonical chunk digests are BLOB keys; other valid IDs retain + // TEXT storage. Both must survive the same mixed-key scan. + let chunk_id = if document % 2 == 0 { + Value::Text(format!("literal-chunk:{document:05}")) + } else { + Value::Blob(document.to_be_bytes().repeat(4)) + }; + insert.execute(params![chunk_id, document]).unwrap(); } drop(insert); tx.commit().unwrap(); } - let chunks = |db: &Connection| -> Vec<(Vec, i64)> { + let chunks = |db: &Connection| -> Vec<(Value, i64)> { let mut query = db .prepare("SELECT chunk_id, document_id FROM row_chunks ORDER BY chunk_id") .unwrap(); @@ -405,6 +411,15 @@ fn carried_row_chunks_check_cancellation_and_preserve_disk_rows() { }; let original = chunks(&connection); assert_eq!(original.len(), 12_000); + let (blob_keys, text_keys): (i64, i64) = connection + .query_row( + "SELECT SUM(typeof(chunk_id) = 'blob'), SUM(typeof(chunk_id) = 'text') + FROM row_chunks", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .unwrap(); + assert_eq!((blob_keys, text_keys), (6_000, 6_000)); let shift = ShiftV1::new([((8, 14), 4), ((45, 52), 10)]).unwrap(); // Reference mapping is deliberately independent of ShiftV1::map: // two replaced ranges remove 13 old chunks, shifting the intervening From 96bc655124719bb87aef5d1e3317511623857ec4 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 19:22:01 -0800 Subject: [PATCH 11/14] style(sqlite): format mixed-key carry rollback assertions Refs #4152 --- .../projection/artifact/increment/tests.rs | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs index ec6d0695b04..1212a405f41 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs @@ -457,7 +457,11 @@ fn carried_row_chunks_check_cancellation_and_preserve_disk_rows() { tx.rollback().unwrap(); drop(connection); connection = Connection::open(&path).unwrap(); - assert_eq!(chunks(&connection), original, "rollback must restore every chunk"); + assert_eq!( + chunks(&connection), + original, + "rollback must restore every chunk" + ); integrity(&connection); } @@ -468,10 +472,15 @@ fn carried_row_chunks_check_cancellation_and_preserve_disk_rows() { tx.commit().unwrap(); drop(connection); let reopened = Connection::open(&path).unwrap(); - assert_eq!(chunks(&reopened), expected, "commit must survive a fresh connection"); + assert_eq!( + chunks(&reopened), + expected, + "commit must survive a fresh connection" + ); assert_eq!( reopened - .query_row("SELECT value FROM independent", [], |row| row.get::<_, String>(0)) + .query_row("SELECT value FROM independent", [], |row| row + .get::<_, String>(0)) .unwrap(), "live" ); From f835a9e2b5117e83cad20681e7b24df1a52255fb Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 19:45:11 -0800 Subject: [PATCH 12/14] fix(sqlite): bound clone payload ordinal restoration Refs #4152 The lexical carry still bulk-restored negative clone_body_payloads INTEGER PRIMARY KEY values, allocating a transient unlinked SQLite statement index despite temp.page_count returning zero. Restore through bounded key-ordered exact updates with cancellation and row-count guards. Cover real 16000-payload reversal, rollback during restoration, durable reopen and Linux live-FD spill detection with the old SQL as a control. --- .../projection/artifact/increment/clones.rs | 51 +++- .../artifact/increment/clones/tests.rs | 266 +++++++++++++++++- 2 files changed, 309 insertions(+), 8 deletions(-) diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs index a0e48692d4b..1f66e77c207 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs @@ -270,6 +270,49 @@ fn clone_range_image( shifted(start, growth_before) } +/// Restore staged negative payload ordinals without the statement-local +/// key index that a bulk INTEGER PRIMARY KEY UPDATE can create. Restored +/// positive keys cannot re-enter the negative-key scan. +fn restore_clone_payload_ordinals( + transaction: &Transaction<'_>, + control: &dyn CodeIndexExecutionControlV1, +) -> Result<(), CodeLexicalArtifactErrorV1> { + let mut update = transaction + .prepare_cached("UPDATE clone_body_payloads SET ordinal = ?2 WHERE ordinal = ?1") + .map_err(sqlite_error)?; + let mut last: Option = None; + loop { + checkpoint(control)?; + let window: Vec = scan_window( + transaction, + "SELECT ordinal FROM clone_body_payloads WHERE ordinal < 0 ORDER BY ordinal LIMIT ?1", + "SELECT ordinal FROM clone_body_payloads WHERE ordinal < 0 AND ordinal > ?1 ORDER BY ordinal LIMIT ?2", + &last, + |ordinal| vec![Value::Integer(*ordinal)], + |row| row.get(0), + )?; + let Some(final_ordinal) = window.last() else { + break; + }; + last = Some(*final_ordinal); + for ordinal in window { + checkpoint(control)?; + // Reversing -number-1 as -(ordinal+1) also handles i64::MIN. + let restored = -(ordinal + 1); + if update + .execute(params![ordinal, restored]) + .map_err(sqlite_error)? + != 1 + { + return Err(CodeLexicalArtifactErrorV1::Corrupt( + "staged clone payload changed during key restoration".to_owned(), + )); + } + } + } + Ok(()) +} + /// Number payloads by first use over the occurrence sequence, retire the /// ones no occurrence uses, and point every occurrence at its number. fn renumber_clone_payloads( @@ -339,6 +382,7 @@ fn renumber_clone_payloads( } } for (ordinal, digest) in parent_payloads { + checkpoint(control)?; match numbered.get(digest) { None => { transaction @@ -359,12 +403,7 @@ fn renumber_clone_payloads( Some(_) => {} } } - transaction - .execute( - "UPDATE clone_body_payloads SET ordinal = -ordinal - 1 WHERE ordinal < 0", - [], - ) - .map_err(sqlite_error)?; + restore_clone_payload_ordinals(transaction, control)?; { let mut insert = transaction .prepare_cached( diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs index 0336f2c7d06..3dfc9b22f1e 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs @@ -1,3 +1,4 @@ +use std::collections::BTreeMap; use std::path::Path; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -6,7 +7,10 @@ use tempfile::{TempDir, tempdir}; use tracedecay_code_index::production::CodeIndexExecutionControlV1; use super::super::CARRY_SCAN_ROWS; -use super::{ClonePostingKey, CodeLexicalArtifactErrorV1, ShiftV1, shift_clone_posting_keys}; +use super::{ + ClonePostingKey, CodeLexicalArtifactErrorV1, ShiftV1, renumber_clone_payloads, + shift_clone_posting_keys, +}; struct ActiveControl; @@ -271,6 +275,170 @@ fn clone_remap_stages_colliding_keys_and_rolls_back_interrupted_windows() { transaction.commit().unwrap(); } + +fn clone_payload_fixture() -> (TempDir, Connection, BTreeMap>) { + let root = tempdir().unwrap(); + let mut connection = open_fixture(root.path()); + connection + .execute_batch( + "CREATE TABLE clone_body_payloads( + ordinal INTEGER PRIMARY KEY, + payload_digest BLOB NOT NULL, + payload BLOB NOT NULL + ); + CREATE TABLE clone_occurrences( + ordinal INTEGER PRIMARY KEY, + payload_ordinal INTEGER NOT NULL + ); + CREATE TABLE independent_payload(value TEXT NOT NULL); + INSERT INTO independent_payload VALUES ('survivor');", + ) + .unwrap(); + let transaction = connection.transaction().unwrap(); + let mut parent = BTreeMap::new(); + { + let mut payload_insert = transaction + .prepare( + "INSERT INTO clone_body_payloads + (ordinal, payload_digest, payload) VALUES (?1, ?2, ?3)", + ) + .unwrap(); + let mut occurrence_insert = transaction + .prepare( + "INSERT INTO clone_occurrences(ordinal, payload_ordinal) VALUES (?1, ?2)", + ) + .unwrap(); + for ordinal in 1_i64..=16_000 { + let digest = ordinal.to_be_bytes().repeat(4); + let payload = ordinal.to_be_bytes().repeat(32); + payload_insert + .execute(params![ordinal, digest.as_slice(), payload.as_slice()]) + .unwrap(); + // First use is deliberately reverse-ordered relative to the + // existing payload keys, so every old ordinal must be rekeyed. + occurrence_insert + .execute(params![ordinal, 16_001 - ordinal]) + .unwrap(); + parent.insert(ordinal, digest); + } + } + transaction.commit().unwrap(); + (root, connection, parent) +} + +fn payload_rows(connection: &Connection) -> Vec<(i64, Vec, Vec)> { + let mut query = connection + .prepare( + "SELECT ordinal, payload_digest, payload + FROM clone_body_payloads ORDER BY ordinal", + ) + .unwrap(); + query + .query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?))) + .unwrap() + .collect::>>() + .unwrap() +} + +fn payload_occurrences(connection: &Connection) -> Vec<(i64, i64)> { + let mut query = connection + .prepare("SELECT ordinal, payload_ordinal FROM clone_occurrences ORDER BY ordinal") + .unwrap(); + query + .query_map([], |row| Ok((row.get(0)?, row.get(1)?))) + .unwrap() + .collect::>>() + .unwrap() +} + +#[test] +fn clone_payload_renumber_preserves_durable_rows_and_interrupts_restoration() { + let (root, mut connection, parent) = clone_payload_fixture(); + let original = payload_rows(&connection); + let original_occurrences = payload_occurrences(&connection); + assert_eq!(original.len(), 16_000); + assert_eq!(original_occurrences.len(), 16_000); + let mut expected: Vec<_> = original + .iter() + .map(|(ordinal, digest, payload)| { + (16_001 - ordinal, digest.clone(), payload.clone()) + }) + .collect(); + expected.sort_by_key(|row| row.0); + assert_ne!(expected, original); + let empty = BTreeMap::::new(); + + // A direct positive-key renumber would collide with an occupied key. + { + let transaction = connection.transaction().unwrap(); + let error = transaction + .execute( + "UPDATE clone_body_payloads SET ordinal = 16001 - ordinal", + [], + ) + .unwrap_err(); + assert_eq!( + error.sqlite_error_code(), + Some(ErrorCode::ConstraintViolation) + ); + transaction.rollback().unwrap(); + } + + // The operation checkpoints during both the scan and each old-key write. + // Stop after staging all 16,000 keys and restoring one bounded window. + { + let transaction = connection.transaction().unwrap(); + let stop = CancelAfter { + checks: AtomicUsize::new(0), + limit: 16_000 * 2 + CARRY_SCAN_ROWS + 8, + }; + let error = + renumber_clone_payloads(&transaction, &parent, &empty, &stop).unwrap_err(); + assert!(matches!(error, CodeLexicalArtifactErrorV1::Interrupted(_))); + let negatives: i64 = transaction + .query_row( + "SELECT COUNT(*) FROM clone_body_payloads WHERE ordinal < 0", + [], + |row| row.get(0), + ) + .unwrap(); + assert!( + negatives > 0 && negatives < 16_000, + "interruption must happen after partial restoration: {negatives}" + ); + integrity(&transaction, "main"); + transaction.rollback().unwrap(); + } + drop(connection); + connection = open_fixture(root.path()); + assert_eq!(payload_rows(&connection), original); + assert_eq!(payload_occurrences(&connection), original_occurrences); + integrity(&connection, "main"); + + let transaction = connection.transaction().unwrap(); + renumber_clone_payloads(&transaction, &parent, &empty, &ActiveControl).unwrap(); + assert_eq!(payload_rows(&transaction), expected); + assert_eq!( + payload_occurrences(&transaction), + (1_i64..=16_000).map(|ordinal| (ordinal, ordinal)).collect::>() + ); + integrity(&transaction, "main"); + transaction.commit().unwrap(); + drop(connection); + let reopened = open_fixture(root.path()); + assert_eq!(payload_rows(&reopened), expected); + assert_eq!( + payload_occurrences(&reopened), + (1_i64..=16_000).map(|ordinal| (ordinal, ordinal)).collect::>() + ); + let survivor: String = reopened + .query_row("SELECT value FROM independent_payload", [], |row| row.get(0)) + .unwrap(); + assert_eq!(survivor, "survivor"); + integrity(&reopened, "main"); + integrity(&reopened, "temp"); +} + #[cfg(target_os = "linux")] mod physical_spill { use std::io; @@ -278,8 +446,11 @@ mod physical_spill { use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + use rusqlite::params; + use super::{ - ActiveControl, ShiftV1, expected_postings, fixture, postings, shift_clone_posting_keys, + ActiveControl, ShiftV1, clone_payload_fixture, expected_postings, fixture, + payload_rows, postings, renumber_clone_payloads, shift_clone_posting_keys, }; fn spill_bytes() -> io::Result { @@ -301,6 +472,97 @@ mod physical_spill { Ok(bytes) } + + #[test] + fn clone_payload_renumber_has_no_unlinked_statement_spill() { + const CHILD: &str = "TRACEDECAY_PAYLOAD_SPILL_TEST_CHILD"; + if std::env::var_os(CHILD).is_none() { + let output = Command::new(std::env::current_exe().unwrap()) + .arg("clone_payload_renumber_has_no_unlinked_statement_spill") + .arg("--test-threads=1") + .arg("--nocapture") + .env(CHILD, "1") + .output() + .unwrap(); + let stdout = String::from_utf8_lossy(&output.stdout); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(output.status.success(), "{stdout}\n{stderr}"); + assert!(stdout.contains("1 passed"), "child test was vacuous: {stdout}"); + return; + } + + let (_root, mut connection, parent) = clone_payload_fixture(); + let original = payload_rows(&connection); + let peak = Arc::new(AtomicU64::new(0)); + let unavailable = Arc::new(AtomicBool::new(false)); + let observed = Arc::clone(&peak); + let unavailable_probe = Arc::clone(&unavailable); + connection.progress_handler( + 1_024, + Some(move || match spill_bytes() { + Ok(bytes) => { + observed.fetch_max(bytes, Ordering::Relaxed); + false + } + Err(_) => { + unavailable_probe.store(true, Ordering::Relaxed); + true + } + }), + ); + + { + let transaction = connection.transaction().unwrap(); + for ordinal in 1_i64..=16_000 { + transaction + .execute( + "UPDATE clone_body_payloads SET ordinal = ?2 WHERE ordinal = ?1", + params![ordinal, -(16_001 - ordinal) - 1], + ) + .unwrap(); + } + // Old bulk restoration, with the actual production key pattern. + transaction + .execute( + "UPDATE clone_body_payloads SET ordinal = -ordinal - 1 WHERE ordinal < 0", + [], + ) + .unwrap(); + transaction.rollback().unwrap(); + } + assert!(!unavailable.load(Ordering::Relaxed)); + assert!( + peak.load(Ordering::Relaxed) > 0, + "bulk payload restoration must produce an observable spill" + ); + assert_eq!(spill_bytes().unwrap(), 0); + assert_eq!( + connection + .pragma_query_value(Some("temp"), "page_count", |row| row.get::<_, i64>(0)) + .unwrap(), + 0, + "TEMP pager counts miss the transient statement-local spill" + ); + assert_eq!(payload_rows(&connection), original); + + peak.store(0, Ordering::Relaxed); + let empty = std::collections::BTreeMap::::new(); + { + let transaction = connection.transaction().unwrap(); + renumber_clone_payloads(&transaction, &parent, &empty, &ActiveControl).unwrap(); + transaction.commit().unwrap(); + } + connection.progress_handler(0, None:: bool>); + assert!(!unavailable.load(Ordering::Relaxed)); + assert_eq!( + peak.load(Ordering::Relaxed), + 0, + "bounded payload key restoration must not create a file spill" + ); + assert_eq!(spill_bytes().unwrap(), 0); + assert_ne!(payload_rows(&connection), original); + } + #[test] fn clone_posting_remap_has_no_implicit_file_spill() { const CHILD: &str = "TRACEDECAY_CLONE_SPILL_TEST_CHILD"; From f91b68a22b97ab1697cd15e8d0f90e304b97deea Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 21:32:09 -0800 Subject: [PATCH 13/14] fix(sqlite): satisfy Bazel formatting and strict Clippy Refs #4152 Remove the obsolete SQL range-image API after the bounded carry migration. Apply exact Rustfmt changes from CI and propagate errors from all SQLite progress-handler install/removal operations in live-spill regressions so installation failures cannot silently pass. --- .../lexical/projection/artifact/increment.rs | 13 ------ .../artifact/increment/clones/tests.rs | 44 ++++++++++--------- .../projection/artifact/increment/tests.rs | 5 ++- 3 files changed, 27 insertions(+), 35 deletions(-) diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs index 2444a3d224d..e1e079b2c91 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs @@ -239,19 +239,6 @@ impl ShiftV1 { }) } - /// An SQL expression giving `column`'s successor value for a carried - /// row; it reads only the parent value, so one `UPDATE` applies it. - fn sql_image(&self, column: &str) -> String { - let mut expression = String::from("CASE"); - for ((_, end), growth) in self.replaced.iter().zip(&self.growth).rev() { - expression.push_str(&format!( - " WHEN {column} >= {end} THEN {column} + ({growth})" - )); - } - expression.push_str(&format!(" ELSE {column} END")); - expression - } - fn sql_replaced(&self, column: &str) -> String { if self.replaced.is_empty() { return "0".to_owned(); diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs index 3dfc9b22f1e..a12b9a2287a 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs @@ -275,7 +275,6 @@ fn clone_remap_stages_colliding_keys_and_rolls_back_interrupted_windows() { transaction.commit().unwrap(); } - fn clone_payload_fixture() -> (TempDir, Connection, BTreeMap>) { let root = tempdir().unwrap(); let mut connection = open_fixture(root.path()); @@ -304,9 +303,7 @@ fn clone_payload_fixture() -> (TempDir, Connection, BTreeMap>) { ) .unwrap(); let mut occurrence_insert = transaction - .prepare( - "INSERT INTO clone_occurrences(ordinal, payload_ordinal) VALUES (?1, ?2)", - ) + .prepare("INSERT INTO clone_occurrences(ordinal, payload_ordinal) VALUES (?1, ?2)") .unwrap(); for ordinal in 1_i64..=16_000 { let digest = ordinal.to_be_bytes().repeat(4); @@ -360,9 +357,7 @@ fn clone_payload_renumber_preserves_durable_rows_and_interrupts_restoration() { assert_eq!(original_occurrences.len(), 16_000); let mut expected: Vec<_> = original .iter() - .map(|(ordinal, digest, payload)| { - (16_001 - ordinal, digest.clone(), payload.clone()) - }) + .map(|(ordinal, digest, payload)| (16_001 - ordinal, digest.clone(), payload.clone())) .collect(); expected.sort_by_key(|row| row.0); assert_ne!(expected, original); @@ -392,8 +387,7 @@ fn clone_payload_renumber_preserves_durable_rows_and_interrupts_restoration() { checks: AtomicUsize::new(0), limit: 16_000 * 2 + CARRY_SCAN_ROWS + 8, }; - let error = - renumber_clone_payloads(&transaction, &parent, &empty, &stop).unwrap_err(); + let error = renumber_clone_payloads(&transaction, &parent, &empty, &stop).unwrap_err(); assert!(matches!(error, CodeLexicalArtifactErrorV1::Interrupted(_))); let negatives: i64 = transaction .query_row( @@ -420,7 +414,9 @@ fn clone_payload_renumber_preserves_durable_rows_and_interrupts_restoration() { assert_eq!(payload_rows(&transaction), expected); assert_eq!( payload_occurrences(&transaction), - (1_i64..=16_000).map(|ordinal| (ordinal, ordinal)).collect::>() + (1_i64..=16_000) + .map(|ordinal| (ordinal, ordinal)) + .collect::>() ); integrity(&transaction, "main"); transaction.commit().unwrap(); @@ -429,10 +425,14 @@ fn clone_payload_renumber_preserves_durable_rows_and_interrupts_restoration() { assert_eq!(payload_rows(&reopened), expected); assert_eq!( payload_occurrences(&reopened), - (1_i64..=16_000).map(|ordinal| (ordinal, ordinal)).collect::>() + (1_i64..=16_000) + .map(|ordinal| (ordinal, ordinal)) + .collect::>() ); let survivor: String = reopened - .query_row("SELECT value FROM independent_payload", [], |row| row.get(0)) + .query_row("SELECT value FROM independent_payload", [], |row| { + row.get(0) + }) .unwrap(); assert_eq!(survivor, "survivor"); integrity(&reopened, "main"); @@ -449,8 +449,8 @@ mod physical_spill { use rusqlite::params; use super::{ - ActiveControl, ShiftV1, clone_payload_fixture, expected_postings, fixture, - payload_rows, postings, renumber_clone_payloads, shift_clone_posting_keys, + ActiveControl, ShiftV1, clone_payload_fixture, expected_postings, fixture, payload_rows, + postings, renumber_clone_payloads, shift_clone_posting_keys, }; fn spill_bytes() -> io::Result { @@ -472,7 +472,6 @@ mod physical_spill { Ok(bytes) } - #[test] fn clone_payload_renumber_has_no_unlinked_statement_spill() { const CHILD: &str = "TRACEDECAY_PAYLOAD_SPILL_TEST_CHILD"; @@ -487,7 +486,10 @@ mod physical_spill { let stdout = String::from_utf8_lossy(&output.stdout); let stderr = String::from_utf8_lossy(&output.stderr); assert!(output.status.success(), "{stdout}\n{stderr}"); - assert!(stdout.contains("1 passed"), "child test was vacuous: {stdout}"); + assert!( + stdout.contains("1 passed"), + "child test was vacuous: {stdout}" + ); return; } @@ -509,7 +511,8 @@ mod physical_spill { true } }), - ); + ) + .unwrap(); { let transaction = connection.transaction().unwrap(); @@ -552,7 +555,7 @@ mod physical_spill { renumber_clone_payloads(&transaction, &parent, &empty, &ActiveControl).unwrap(); transaction.commit().unwrap(); } - connection.progress_handler(0, None:: bool>); + connection.progress_handler(0, None:: bool>).unwrap(); assert!(!unavailable.load(Ordering::Relaxed)); assert_eq!( peak.load(Ordering::Relaxed), @@ -603,7 +606,8 @@ mod physical_spill { true } }), - ); + ) + .unwrap(); { let transaction = connection.transaction().unwrap(); // Bulk key updates are the positive control for statement-local @@ -646,7 +650,7 @@ mod physical_spill { shift_clone_posting_keys(&transaction, &shift, &ActiveControl).unwrap(); transaction.commit().unwrap(); } - connection.progress_handler(0, None:: bool>); + connection.progress_handler(0, None:: bool>).unwrap(); assert!(!unavailable.load(Ordering::Relaxed)); assert_eq!(peak.load(Ordering::Relaxed), 0, "no implicit remap spill"); assert_eq!(spill_bytes().unwrap(), 0); diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs index 1212a405f41..63f4a3c255a 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs @@ -301,7 +301,8 @@ fn lexical_rowid_shift_does_not_spill_unlinked_sqlite_files() { true } }), - ); + ) + .unwrap(); { let tx = connection.transaction().unwrap(); tx.execute( @@ -337,7 +338,7 @@ fn lexical_rowid_shift_does_not_spill_unlinked_sqlite_files() { shift_rowid_keys(&tx, "row_blocks", "first_document", &shift, &Active).unwrap(); tx.commit().unwrap(); } - connection.progress_handler(0, None:: bool>); + connection.progress_handler(0, None:: bool>).unwrap(); assert!(!unreadable.load(Ordering::Relaxed)); assert_eq!( peak.load(Ordering::Relaxed), From 18cc1a415202df9f4870318918a68f3e5f9df3fc Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 21:33:45 -0800 Subject: [PATCH 14/14] style(sqlite): align progress-handler error chains with Rustfmt Refs #4152 Match the repository's canonical chained SQLite progress-handler format for installation and teardown in live spill regressions. --- .../artifact/increment/clones/tests.rs | 66 ++++++++++--------- .../projection/artifact/increment/tests.rs | 33 +++++----- 2 files changed, 54 insertions(+), 45 deletions(-) diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs index a12b9a2287a..2c2b2a08ea7 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs @@ -499,20 +499,21 @@ mod physical_spill { let unavailable = Arc::new(AtomicBool::new(false)); let observed = Arc::clone(&peak); let unavailable_probe = Arc::clone(&unavailable); - connection.progress_handler( - 1_024, - Some(move || match spill_bytes() { - Ok(bytes) => { - observed.fetch_max(bytes, Ordering::Relaxed); - false - } - Err(_) => { - unavailable_probe.store(true, Ordering::Relaxed); - true - } - }), - ) - .unwrap(); + connection + .progress_handler( + 1_024, + Some(move || match spill_bytes() { + Ok(bytes) => { + observed.fetch_max(bytes, Ordering::Relaxed); + false + } + Err(_) => { + unavailable_probe.store(true, Ordering::Relaxed); + true + } + }), + ) + .unwrap(); { let transaction = connection.transaction().unwrap(); @@ -555,7 +556,9 @@ mod physical_spill { renumber_clone_payloads(&transaction, &parent, &empty, &ActiveControl).unwrap(); transaction.commit().unwrap(); } - connection.progress_handler(0, None:: bool>).unwrap(); + connection + .progress_handler(0, None:: bool>) + .unwrap(); assert!(!unavailable.load(Ordering::Relaxed)); assert_eq!( peak.load(Ordering::Relaxed), @@ -594,20 +597,21 @@ mod physical_spill { let unavailable = Arc::new(AtomicBool::new(false)); let observed_peak = Arc::clone(&peak); let failed_probe = Arc::clone(&unavailable); - connection.progress_handler( - 4096, - Some(move || match spill_bytes() { - Ok(bytes) => { - observed_peak.fetch_max(bytes, Ordering::Relaxed); - false - } - Err(_) => { - failed_probe.store(true, Ordering::Relaxed); - true - } - }), - ) - .unwrap(); + connection + .progress_handler( + 4096, + Some(move || match spill_bytes() { + Ok(bytes) => { + observed_peak.fetch_max(bytes, Ordering::Relaxed); + false + } + Err(_) => { + failed_probe.store(true, Ordering::Relaxed); + true + } + }), + ) + .unwrap(); { let transaction = connection.transaction().unwrap(); // Bulk key updates are the positive control for statement-local @@ -650,7 +654,9 @@ mod physical_spill { shift_clone_posting_keys(&transaction, &shift, &ActiveControl).unwrap(); transaction.commit().unwrap(); } - connection.progress_handler(0, None:: bool>).unwrap(); + connection + .progress_handler(0, None:: bool>) + .unwrap(); assert!(!unavailable.load(Ordering::Relaxed)); assert_eq!(peak.load(Ordering::Relaxed), 0, "no implicit remap spill"); assert_eq!(spill_bytes().unwrap(), 0); diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs index 63f4a3c255a..af1b5d10462 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs @@ -289,20 +289,21 @@ fn lexical_rowid_shift_does_not_spill_unlinked_sqlite_files() { let unreadable = Arc::new(AtomicBool::new(false)); let observer_peak = Arc::clone(&peak); let observer_failed = Arc::clone(&unreadable); - connection.progress_handler( - 1_024, - Some(move || match spill_bytes() { - Ok(bytes) => { - observer_peak.fetch_max(bytes, Ordering::Relaxed); - false - } - Err(_) => { - observer_failed.store(true, Ordering::Relaxed); - true - } - }), - ) - .unwrap(); + connection + .progress_handler( + 1_024, + Some(move || match spill_bytes() { + Ok(bytes) => { + observer_peak.fetch_max(bytes, Ordering::Relaxed); + false + } + Err(_) => { + observer_failed.store(true, Ordering::Relaxed); + true + } + }), + ) + .unwrap(); { let tx = connection.transaction().unwrap(); tx.execute( @@ -338,7 +339,9 @@ fn lexical_rowid_shift_does_not_spill_unlinked_sqlite_files() { shift_rowid_keys(&tx, "row_blocks", "first_document", &shift, &Active).unwrap(); tx.commit().unwrap(); } - connection.progress_handler(0, None:: bool>).unwrap(); + connection + .progress_handler(0, None:: bool>) + .unwrap(); assert!(!unreadable.load(Ordering::Relaxed)); assert_eq!( peak.load(Ordering::Relaxed),