From 1bdded9c76efab083e1d74f7c3a3500d936ee32a Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Mon, 28 Sep 2026 18:28:52 +0000 Subject: [PATCH 1/2] fix(memory): retain cgroup pressure after allocator trim Co-authored-by: Zack Jackson --- .../tracedecay-application/src/code_index.rs | 17 +++- .../src/resident_memory.rs | 34 +++----- .../src/resident_memory/tests.rs | 83 ++++++++++++++++--- .../src/session_registry/code_graph.rs | 14 ++-- 4 files changed, 101 insertions(+), 47 deletions(-) diff --git a/crates/tracedecay-application/src/code_index.rs b/crates/tracedecay-application/src/code_index.rs index 24fb956a7e..9eede81032 100644 --- a/crates/tracedecay-application/src/code_index.rs +++ b/crates/tracedecay-application/src/code_index.rs @@ -209,9 +209,7 @@ impl CodeIndexExecutionControlV1 for DaemonCodeIndexControlV1 { // the last published state here is how a second linked worktree kept // allocating until the cgroup killer. Reclaimers stay off this thread: // the indexing pool's stack cannot host them. - let Some(state) = pressure.sample_for_checkpoint() else { - return false; - }; + let state = pressure.sample_for_checkpoint(); if matches!(state, ResidentMemoryPressureStateV1::OverBudget { .. }) { self.tripped.store(true, Ordering::Release); return true; @@ -306,4 +304,17 @@ mod tests { assert!(!shutting_down.load(Ordering::Acquire)); assert!(!control.is_deadline_exceeded()); } + + #[test] + fn checkpoint_sampling_failure_preserves_known_pressure() { + let limit = NonZeroU64::new(100 * 1024 * 1024).expect("limit"); + let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( + limit, + Arc::new(|| None), + )); + pressure.publish_observed_resident_bytes(limit.get()); + let control = DaemonCodeIndexControlV1::default().with_resident_memory(pressure); + assert!(control.is_cancelled()); + assert!(control.refused_by_resident_memory()); + } } diff --git a/crates/tracedecay-runtime-core/src/resident_memory.rs b/crates/tracedecay-runtime-core/src/resident_memory.rs index df7db21c75..af90cb3e4e 100644 --- a/crates/tracedecay-runtime-core/src/resident_memory.rs +++ b/crates/tracedecay-runtime-core/src/resident_memory.rs @@ -698,8 +698,9 @@ impl ResidentMemoryPressureV1 { /// /// At most one checkpoint per [`RESIDENT_MEMORY_CHECKPOINT_SAMPLE_INTERVAL_V1`] /// reads the process; the others, and any checkpoint racing that read, - /// answer the standing observation. - pub fn sample_for_checkpoint(&self) -> Option { + /// answer the standing observation. If sampling fails, the last published + /// pressure verdict remains authoritative. + pub fn sample_for_checkpoint(&self) -> ResidentMemoryPressureStateV1 { let now = self.checkpoint_micros(); let next = self.next_checkpoint_sample_micros.load(Ordering::Acquire); if now < next @@ -708,7 +709,7 @@ impl ResidentMemoryPressureV1 { .compare_exchange(next, u64::MAX, Ordering::AcqRel, Ordering::Acquire) .is_err() { - return Some(self.state()); + return self.state(); } let sample = (self.sampler)(); hotpath::gauge!("daemon.memory.checkpoint_samples_total").inc(1_u64); @@ -718,9 +719,11 @@ impl ResidentMemoryPressureV1 { self.checkpoint_micros().saturating_add(interval), Ordering::Release, ); - self.publish_observation(sample?.admission_bytes()); - self.publish_over_budget_gauge(); - Some(self.state()) + if let Some(sample) = sample { + self.publish_observation(sample.admission_bytes()); + self.publish_over_budget_gauge(); + } + self.state() } fn checkpoint_micros(&self) -> u64 { @@ -775,20 +778,6 @@ impl ResidentMemoryPressureV1 { self.state() } - /// Publish RSS measured by a reclaimer after it released memory. - /// - /// This updates the canonical admission observation without running the - /// reclaimer registry again. Reclaimers that can measure their process - /// effect use this path from inside the original pressure pass. - fn publish_post_reclaim_observed_resident_bytes( - &self, - observed_bytes: u64, - ) -> ResidentMemoryPressureStateV1 { - self.publish_observation(observed_bytes); - self.publish_over_budget_gauge(); - self.state() - } - fn publish_observation(&self, observed_bytes: u64) { self.observed_bytes.store(observed_bytes, Ordering::Release); self.observed.store(true, Ordering::Release); @@ -1066,9 +1055,8 @@ pub fn register_process_allocator_pressure_reclaimer_v1( PROCESS_ALLOCATOR_TRIM_PRESSURE_PRIORITY_V1, Arc::new(move |request| { let trim = release_process_allocator_memory_v1(); - if let (Some(after_bytes), Some(pressure)) = (trim.after_bytes, pressure_weak.upgrade()) - { - pressure.publish_post_reclaim_observed_resident_bytes(after_bytes); + if let Some(pressure) = pressure_weak.upgrade() { + pressure.sample_for_checkpoint(); } tracing::info!( event = "process_allocator_trimmed", diff --git a/crates/tracedecay-runtime-core/src/resident_memory/tests.rs b/crates/tracedecay-runtime-core/src/resident_memory/tests.rs index d38c7682ef..af033c2f1d 100644 --- a/crates/tracedecay-runtime-core/src/resident_memory/tests.rs +++ b/crates/tracedecay-runtime-core/src/resident_memory/tests.rs @@ -1054,11 +1054,26 @@ fn reaching_the_high_watermark_runs_pressure_reclaimers_with_the_measurement() { #[test] fn post_reclaim_observation_replaces_pressure_state_without_reentering_reclaimers() { - let (authority, pressure) = pressure_authority(); + let limit = bytes(PRESSURE_TEST_LIMIT_BYTES); + let after_reclaim = 1; + let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( + limit, + Arc::new(move || { + Some(ProcessResidentSampleV1 { + resident_bytes: after_reclaim, + unreclaimable_bytes: after_reclaim, + swapped_bytes: 0, + cgroup_committed_bytes: None, + }) + }), + )); + let authority = Arc::new(ProcessResidentMemoryV1::with_pressure( + limit, + Arc::clone(&pressure), + )); let calls = Arc::new(Mutex::new(0_u64)); let callback_calls = Arc::clone(&calls); let callback_pressure = Arc::downgrade(&pressure); - let after_reclaim = pressure.low_watermark_bytes(); let _registration = pressure .register_pressure_reclaimer( 10, @@ -1067,7 +1082,7 @@ fn post_reclaim_observation_replaces_pressure_state_without_reentering_reclaimer callback_pressure .upgrade() .expect("pressure authority remains live") - .publish_post_reclaim_observed_resident_bytes(after_reclaim); + .sample_for_checkpoint(); 4096 }), ) @@ -1415,12 +1430,7 @@ fn checkpoints_read_the_process_once_per_interval_and_then_see_growth() { let started = Instant::now(); for _ in 0..200_000 { - assert!( - !pressure - .sample_for_checkpoint() - .expect("checkpoint state") - .is_over_budget() - ); + assert!(!pressure.sample_for_checkpoint().is_over_budget()); } let elapsed = started.elapsed(); let reads_taken = reads.load(Ordering::Acquire); @@ -1434,10 +1444,7 @@ fn checkpoints_read_the_process_once_per_interval_and_then_see_growth() { observed.store(limit.get(), Ordering::Release); std::thread::sleep(RESIDENT_MEMORY_CHECKPOINT_SAMPLE_INTERVAL_V1); assert!( - pressure - .sample_for_checkpoint() - .expect("checkpoint state") - .is_over_budget(), + pressure.sample_for_checkpoint().is_over_budget(), "a checkpoint past the interval must read the grown process" ); assert_eq!(reads.load(Ordering::Acquire), reads_taken + 1); @@ -1509,3 +1516,53 @@ fn headroom_is_the_ceiling_less_the_larger_of_ledger_and_measurement() { ); drop((charged, in_flight)); } + +#[test] +fn allocator_reclaim_preserves_cgroup_pressure_until_sample_recovers() { + let limit = bytes(1024 * 1024 * 1024 * 1024); + let sample = Arc::new(Mutex::new(ProcessResidentSampleV1 { + resident_bytes: 1, + unreclaimable_bytes: 1, + swapped_bytes: 0, + cgroup_committed_bytes: Some(limit.get()), + })); + let sampled = Arc::clone(&sample); + let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( + limit, + Arc::new(move || Some(*sampled.lock().expect("sample"))), + )); + let _reclaimer = super::register_process_allocator_pressure_reclaimer_v1(&pressure) + .expect("register allocator reclaimer"); + let authority = Arc::new(ProcessResidentMemoryV1::with_pressure( + limit, + Arc::clone(&pressure), + )); + + assert_eq!(pressure.measure_admission_bytes(), limit.get()); + assert!(matches!( + pressure.state(), + ResidentMemoryPressureStateV1::OverBudget { .. } + )); + assert!(matches!( + authority.reserve( + key("project", "worktree", "generation", "graph"), + growth_request() + ), + Err(ResidentMemoryAdmissionFailureV1::ObservedOverBudget { .. }) + )); + + sample.lock().expect("sample").cgroup_committed_bytes = Some(1); + assert_eq!(pressure.measure_admission_bytes(), 1); + assert!(matches!( + pressure.state(), + ResidentMemoryPressureStateV1::Nominal { .. } + )); + assert!( + authority + .reserve( + key("project", "worktree", "generation", "graph"), + growth_request() + ) + .is_ok() + ); +} diff --git a/crates/tracedecay-store-runtime/src/session_registry/code_graph.rs b/crates/tracedecay-store-runtime/src/session_registry/code_graph.rs index 09b8b5f39f..759449dbd5 100644 --- a/crates/tracedecay-store-runtime/src/session_registry/code_graph.rs +++ b/crates/tracedecay-store-runtime/src/session_registry/code_graph.rs @@ -247,14 +247,12 @@ impl GraphCancellation for ResidentMemoryGuardedGraphCancellationV1 { // build. A checkpoint that trusts the last published state keeps // allocating until the cgroup kill line. Read the process here, // without reclaiming: this checkpoint also runs on the indexing pool. - let Some( - tracedecay_runtime_core::resident_memory::ResidentMemoryPressureStateV1::OverBudget { - observed_bytes, - limit_bytes, - high_watermark_bytes, - .. - }, - ) = self.pressure.sample_for_checkpoint() + let tracedecay_runtime_core::resident_memory::ResidentMemoryPressureStateV1::OverBudget { + observed_bytes, + limit_bytes, + high_watermark_bytes, + .. + } = self.pressure.sample_for_checkpoint() else { return false; }; From 2d27d53ceb8ec9ef996c393c7db4d1b7ea3f9109 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Mon, 28 Sep 2026 18:29:01 +0000 Subject: [PATCH 2/2] fix(code-index): withdraw unsafe carried lexical artifacts Co-authored-by: Zack Jackson --- .../src/code_index_scheduler/serving.rs | 112 +- .../src/production/lexical_page_source.rs | 41 +- .../src/production/partitioned_codec.rs | 12 - .../retrieval/lexical/projection/artifact.rs | 1 - .../lexical/projection/artifact/builder.rs | 306 +--- .../lexical/projection/artifact/delta.rs | 1465 ----------------- .../lexical/projection/artifact/format.rs | 16 +- .../lexical/projection/artifact/prepared.rs | 2 - .../lexical/projection/artifact/row_codec.rs | 14 - .../daemon_suite/indexing_lifecycle_test.rs | 202 --- 10 files changed, 51 insertions(+), 2120 deletions(-) delete mode 100644 crates/tracedecay-query/src/retrieval/lexical/projection/artifact/delta.rs diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/serving.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/serving.rs index 3b01860149..d486f7dc5f 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/serving.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/serving.rs @@ -2385,27 +2385,6 @@ impl LatestCodeTextGenerationV1 { }) } - /// The published text artifact of this generation's parent, when one is - /// still on disk. A missing parent is a full build, not a failure. - fn published_parent_text_artifact(&self) -> Result, RetrievalPortError> { - let Some(parent_id) = self.metadata.manifest().parent_generation.clone() else { - return Ok(None); - }; - let Some(descriptor) = self.text_artifact_store.published_descriptor(&parent_id)? else { - return Ok(None); - }; - let path = code_text_artifact_path(self.text_artifact_store.store_root(), &descriptor) - .map_err(text_artifact_unavailable)?; - match path.symlink_metadata() { - Ok(metadata) if metadata.file_type().is_file() => Ok(Some(path)), - Ok(_) => Err(RetrievalPortError::Contract( - "parent code text artifact is not a regular file".to_owned(), - )), - Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None), - Err(error) => Err(text_artifact_unavailable(error)), - } - } - fn publish_text_progress_boundary( &self, build: &CodeTextArtifactBuildV1, @@ -2979,35 +2958,24 @@ impl LatestCodeTextGenerationV1 { let builder_budget = text_artifact_builder_budget(build_memory_budget, source.staging_window_bytes())?; let metadata = self.text_projection_metadata()?; + let content_key = text_artifact_content_key(&source, &metadata)?; let mut builder = if staging_path.exists() { - open_or_replace_incompatible_text_staging( - store, + match CodeLexicalArtifactBuilderV1::open_or_resume_with_memory_budget_and_control( &staging_path, - &metadata, + metadata.clone(), builder_budget, control, - ) - } else if let Some(parent_artifact) = self.published_parent_text_artifact()? { - match CodeLexicalArtifactBuilderV1::stage_carried_parent( - &parent_artifact, - &staging_path, - &metadata, - control, ) { - Ok(true) => open_or_replace_incompatible_text_staging( - store, - &staging_path, - &metadata, - builder_budget, - control, - ), - Ok(false) => CodeLexicalArtifactBuilderV1::create_with_memory_budget( - &staging_path, - metadata.clone(), - builder_budget, - ) - .map_err(map_text_artifact_error), - Err(error) => Err(map_text_artifact_error(error)), + Ok(builder) => Ok(builder), + Err(CodeLexicalArtifactErrorV1::Incompatible(_)) => { + store.discard_incompatible_staging(&staging_path, control)?; + CodeLexicalArtifactBuilderV1::create_with_memory_budget( + &staging_path, + metadata.clone(), + builder_budget, + ) + } + Err(error) => Err(error), } } else { CodeLexicalArtifactBuilderV1::create_with_memory_budget( @@ -3015,20 +2983,8 @@ impl LatestCodeTextGenerationV1 { metadata.clone(), builder_budget, ) - .map_err(map_text_artifact_error) - }?; - // A carried staging file already holds the parent's unchanged rows. - // Restrict the sealed source to the occurrences still missing so the - // content key, the receipt, and reported progress describe that work. - if let Some(rebuild) = builder - .carried_rebuild_occurrences() - .map_err(map_text_artifact_error)? - { - source - .restrict_to_file_occurrences(&rebuild) - .map_err(map_sealed_page_source_error)?; } - let mut content_key = text_artifact_content_key(&source, &metadata)?; + .map_err(map_text_artifact_error)?; // A staging file sealed before its publication was interrupted keeps // no source cursor; it is complete and is published as it stands. if builder @@ -3053,8 +3009,6 @@ impl LatestCodeTextGenerationV1 { Err(VerifiedSealedLexicalCursorRestoreErrorV1::IncompatiblePosition) => { drop(builder); store.discard_incompatible_staging(&staging_path, control)?; - source = store.open_sealed_source(&sealed_identity, control)?; - content_key = text_artifact_content_key(&source, &metadata)?; builder = CodeLexicalArtifactBuilderV1::create_with_memory_budget( &staging_path, metadata, @@ -3636,34 +3590,6 @@ pub(super) fn sha256_private_file_and_size( Ok((hasher.finalize().into(), file_metadata.len())) } -/// Resume `staging_path`, or replace an incompatible file with a fresh build. -fn open_or_replace_incompatible_text_staging( - store: &DaemonCodeTextArtifactStoreV1, - staging_path: &Path, - metadata: &CodeLexicalProjectionMetadataV1, - builder_budget: usize, - control: &dyn CodeIndexExecutionControlV1, -) -> Result { - match CodeLexicalArtifactBuilderV1::open_or_resume_with_memory_budget_and_control( - staging_path, - metadata.clone(), - builder_budget, - control, - ) { - Ok(builder) => Ok(builder), - Err(CodeLexicalArtifactErrorV1::Incompatible(_)) => { - store.discard_incompatible_staging(staging_path, control)?; - CodeLexicalArtifactBuilderV1::create_with_memory_budget( - staging_path, - metadata.clone(), - builder_budget, - ) - .map_err(map_text_artifact_error) - } - Err(error) => Err(map_text_artifact_error(error)), - } -} - /// The content key the text artifact of `source` under `metadata` is /// published with. fn text_artifact_content_key( @@ -3769,7 +3695,6 @@ fn clear_text_artifact_staging_sidecars(staging_path: &Path) -> std::io::Result< }; // The compacted rewrite's own rollback journal goes before the rewrite, // for the same reason the staging journal goes before the staging file. - // A parent carry killed before its rename leaves the same pair. for suffix in [ "-journal", "-wal", @@ -3849,14 +3774,9 @@ mod staging_sidecar_tests { assert!(!compacting_journal.exists()); let carrying = root.path().join(".text-artifact-ab.staging-carrying"); - let carrying_journal = root - .path() - .join(".text-artifact-ab.staging-carrying-journal"); - std::fs::write(&carrying, b"torn parent copy").expect("plant torn carry"); - std::fs::write(&carrying_journal, b"carry rollback").expect("plant carry journal"); - prepare_absent_text_artifact_staging(&staging).expect("clear torn carry"); + std::fs::write(&carrying, b"obsolete parent carry").expect("plant carry"); + prepare_absent_text_artifact_staging(&staging).expect("clear obsolete carry"); assert!(!carrying.exists()); - assert!(!carrying_journal.exists()); std::fs::write(root.path().join(".text-artifact-ab.staging-wal"), b"wal") .expect("plant wal"); diff --git a/crates/tracedecay-code-index/src/production/lexical_page_source.rs b/crates/tracedecay-code-index/src/production/lexical_page_source.rs index b5f080b397..8ccf1c3e4c 100644 --- a/crates/tracedecay-code-index/src/production/lexical_page_source.rs +++ b/crates/tracedecay-code-index/src/production/lexical_page_source.rs @@ -1,13 +1,9 @@ -use std::{ - collections::{BTreeMap, BTreeSet}, - num::NonZeroUsize, - sync::Arc, -}; +use std::{collections::BTreeMap, num::NonZeroUsize, sync::Arc}; use sha2::{Digest, Sha256}; use tracedecay_domain::{ CodeGenerationSourceCommitmentsV1, CodeSearchChunkGrainV1, CodeSearchChunkV1, - ExactTechnicalTermV1, FileOccurrenceId, + ExactTechnicalTermV1, }; use crate::{ @@ -1307,39 +1303,6 @@ impl VerifiedSealedLexicalPageSourceV1 { &self.metadata } - /// Emit pages only for `occurrences`, in their existing sealed order. - /// - /// The cursor restarts at the first retained file. Unchanged files stay - /// out of this source because their rows are already durable in the - /// parent text artifact the builder carries forward. - pub fn restrict_to_file_occurrences( - &mut self, - occurrences: &BTreeSet, - ) -> Result<(), CodeIndexProductionErrorV1> { - self.file_source.retain_occurrences(occurrences); - self.admitted_window.clear(); - let file_count = u64::try_from(self.file_source.len()).map_err(|_| { - CodeIndexProductionErrorV1::Contract( - "partitioned sealed generation file count exceeds u64".to_owned(), - ) - })?; - self.file_count = file_count; - self.first_file_offset = 0; - self.files_end_offset = file_count; - self.file_ranges = (0..file_count) - .map(|file| (file, file.saturating_add(1))) - .collect(); - self.lexical_byte_offsets = self.file_source.lexical_byte_offsets()?; - self.total_lexical_units = self.lexical_byte_offsets.last().copied().ok_or_else(|| { - CodeIndexProductionErrorV1::Contract( - "partitioned lexical byte offsets are empty".to_owned(), - ) - })?; - self.maximum_file_bytes = self.file_source.maximum_file_bytes(); - self.cursor = VerifiedSealedLexicalCursorV1::initial(self.source_state_digest.clone(), 0)?; - Ok(()) - } - pub fn format_revision(&self) -> u32 { self.format_revision } diff --git a/crates/tracedecay-code-index/src/production/partitioned_codec.rs b/crates/tracedecay-code-index/src/production/partitioned_codec.rs index 9619a6c8dd..2429dd1f42 100644 --- a/crates/tracedecay-code-index/src/production/partitioned_codec.rs +++ b/crates/tracedecay-code-index/src/production/partitioned_codec.rs @@ -1498,18 +1498,6 @@ impl PartitionedLexicalFileSourceV1 { self.descriptors.len() } - /// Keep file segments whose occurrence is in `keep`, preserving order. - /// - /// Callers use this to feed a successor text build only the files that - /// are not already durable in the parent artifact. - pub(super) fn retain_occurrences( - &mut self, - keep: &std::collections::BTreeSet, - ) { - self.descriptors - .retain(|descriptor| keep.contains(&descriptor.file_occurrence_id)); - } - /// Every input the content of the pages this source emits depends on /// apart from the generation's route identity: each file segment in /// order (its key, occurrence, content address, and symbol identities). diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact.rs index e20085d599..02afd6d224 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact.rs @@ -14,7 +14,6 @@ use tracedecay_runtime_core::path_safety::{SqliteDatabasePathError, sqlite_datab mod builder; mod clone_census; mod clone_codec; -mod delta; mod fingerprints; mod format; mod postings; diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/builder.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/builder.rs index b14e386c90..c2b22793be 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/builder.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/builder.rs @@ -1028,10 +1028,6 @@ pub struct CodeLexicalArtifactBuilderV1 { metadata_digest: ManifestDigest, memory_budget_bytes: usize, fixed_ledger_charge_bytes: usize, - /// Added to source chunk ordinals when this staging file already holds a - /// parent artifact's documents. Zero keeps a full build's document ids - /// identical to source ordinals. - document_id_shift: u64, } /// One source-prefix decision whose fresh relational values were prepared @@ -1098,22 +1094,9 @@ impl CodeLexicalArtifactBuilderV1 { metadata_digest, memory_budget_bytes, fixed_ledger_charge_bytes, - document_id_shift: 0, }) } - /// Copy a published parent artifact into `staging` when the child still - /// shares occurrence ids with it. Returns false when nothing can be - /// carried; `staging` is absent in that case. - pub fn stage_carried_parent( - parent_artifact: &Path, - staging: &Path, - metadata: &CodeLexicalProjectionMetadataV1, - control: &dyn CodeIndexExecutionControlV1, - ) -> Result { - super::delta::stage_carried_parent(parent_artifact, staging, metadata, control) - } - /// Reopen only the staged artifact authority while applying the caller's /// scheduler epoch/deadline control to integrity, metadata, receipt, and /// contiguous-cursor verification. @@ -1183,7 +1166,6 @@ impl CodeLexicalArtifactBuilderV1 { )); } validate_contiguous_pages(&connection, control)?; - let document_id_shift = super::delta::read_carried_document_shift(&connection)?; checkpoint(control)?; crate::hotpath_metrics::Residency::Rebuilding.record("query.artifact.residency"); Ok(Self { @@ -1196,19 +1178,9 @@ impl CodeLexicalArtifactBuilderV1 { metadata_digest: expected_digest, memory_budget_bytes, fixed_ledger_charge_bytes, - document_id_shift, }) } - /// Occurrences this carried staging still has to append. `None` when the - /// staging file is a full build. - pub fn carried_rebuild_occurrences( - &self, - ) -> Result>, CodeLexicalArtifactErrorV1> - { - super::delta::read_carried_rebuild_occurrences(&self.connection) - } - #[hotpath::skip] pub fn progress( &self, @@ -1508,7 +1480,6 @@ impl CodeLexicalArtifactBuilderV1 { .map(|page| page_transient_peak_bytes(&self.metadata, page, usize::MAX)) .collect::, _>>()?; let metadata = &self.metadata; - let document_id_shift = self.document_id_shift; let prepared = hotpath::measure_block!("query.artifact.batch.parallel_prepare", { tracedecay_code_index::parallelism::install(|| { fresh_pages @@ -1524,7 +1495,6 @@ impl CodeLexicalArtifactBuilderV1 { page, previous_cursor, scratch_bytes, - document_id_shift, control, ) })) @@ -1582,7 +1552,7 @@ impl CodeLexicalArtifactBuilderV1 { record_artifact_progress(¤t); return Ok(current); } - validate_prepared_page_batch(¤t, pages, self.document_id_shift)?; + validate_prepared_page_batch(¤t, pages)?; admit_prepared_page_batch( self.fixed_ledger_charge_bytes, self.memory_budget_bytes, @@ -1722,9 +1692,6 @@ impl CodeLexicalArtifactBuilderV1 { record_finalization_step(&step); return Ok(step); } - if super::delta::carried_delta_pending(&self.connection)? { - return self.advance_carried_finalization(source, control); - } enter_resumable_journal(&self.connection)?; let pending_state = load_finalization_state(&self.connection)?; @@ -1947,93 +1914,6 @@ impl CodeLexicalArtifactBuilderV1 { Ok(step) } - /// Seal a staging file that already holds its parent's unchanged rows. - /// - /// The consumed source is only the changed files. Section digests still - /// fold that source's page receipts; the carried rows stay authenticated - /// by the artifact file digest written at publication. - fn advance_carried_finalization( - &mut self, - source: &VerifiedSealedLexicalSourceReceiptV1, - control: &dyn CodeIndexExecutionControlV1, - ) -> Result { - checkpoint(control)?; - enter_resumable_journal(&self.connection)?; - let transaction = self.connection.transaction().map_err(sqlite_error)?; - let mut transaction_metrics = FinalizationTransactionMetricsV1::new(); - super::delta::merge_carried_staging(&transaction, &self.mutation_gate, control)?; - transaction - .execute_batch( - "DROP TRIGGER IF EXISTS immutable_ngram_postings_update; - DROP TRIGGER IF EXISTS immutable_ngram_postings_delete;", - ) - .map_err(sqlite_error)?; - if super::delta::carried_ngram_patch_fits(&transaction)? { - let _guard = BuilderMutationGuardV1::enter(&self.mutation_gate)?; - super::delta::patch_carried_ngram_postings( - &transaction, - &self.metadata.generation, - control, - )?; - transaction - .execute_batch( - "CREATE TRIGGER frozen_ngram_postings_insert BEFORE INSERT ON ngram_postings BEGIN SELECT RAISE(ABORT, 'frozen lexical ngram postings'); END;", - ) - .map_err(sqlite_error)?; - } else { - transaction - .execute_batch("DELETE FROM ngram_postings;") - .map_err(sqlite_error)?; - derive_ngram_postings( - &transaction, - &ServingIndexStepAuthorityV1 { - mutation_gate: &self.mutation_gate, - generation: &self.metadata.generation, - ngram_memory_bytes: self.memory_budget_bytes / 4, - }, - control, - )?; - } - super::delta::discard_carried_ngram_plan(&transaction)?; - ensure_carried_builder_triggers(&transaction)?; - let sections = compute_section_digests(&transaction, control)?; - verify_final_sections_against_source(§ions, source)?; - let clone_index_census = read_clone_index_census( - &transaction, - CLONE_FINGERPRINT_HOT_POSTING_THRESHOLD_V1, - control, - )?; - release_free_pages(&transaction, control)?; - let file_size_bytes = sqlite_file_size(&transaction)?; - let receipt = new_verified_receipt( - self.metadata_digest.clone(), - source, - sections, - clone_index_census, - file_size_bytes, - )?; - transaction - .execute( - "UPDATE artifact_state SET receipt = ?1 WHERE singleton = 1", - params![padded_receipt(&receipt)?], - ) - .map_err(sqlite_error)?; - checkpoint(control)?; - commit_finalization_transaction(transaction, &mut transaction_metrics)?; - self.canonicalize_sealed_header()?; - verify_finalized_artifact( - &self.connection, - &self.path, - &self.metadata_digest, - source, - &receipt, - control, - )?; - let step = CodeLexicalArtifactFinalizationStepV1::Ready(Box::new(receipt)); - record_finalization_step(&step); - Ok(step) - } - /// Rewrite the staging file from its content alone. /// /// Under the builder's rollback journal an in-place `VACUUM` writes the @@ -2236,10 +2116,7 @@ fn compacting_staging_sibling(path: &Path) -> Result Result { +fn staging_sibling(path: &Path, suffix: &str) -> Result { let mut name = path .file_name() .ok_or_else(|| { @@ -3443,7 +3320,6 @@ fn anchor_owned_bytes(anchor: &CodeSearchChunkAnchorV1) -> usize { fn validate_prepared_page_batch( current: &CodeLexicalArtifactBuildProgressV1, pages: &[PreparedCodeLexicalArtifactPageV1], - document_id_shift: u64, ) -> Result<(), CodeLexicalArtifactErrorV1> { let mut expected_ordinal = current.next_page_ordinal; let mut expected_document = current.completed_chunks; @@ -3474,14 +3350,7 @@ fn validate_prepared_page_batch( })?; let mut next_document = expected_document; for document in &page.documents { - let document_id = u64::try_from(document.document_id) - .map_err(contract_number)? - .checked_sub(document_id_shift) - .ok_or_else(|| { - CodeLexicalArtifactErrorV1::Corrupt( - "prepared lexical document id is below its carried shift".to_owned(), - ) - })?; + let document_id = u64::try_from(document.document_id).map_err(contract_number)?; if document_id < next_document || document_id >= page_end { return Err(CodeLexicalArtifactErrorV1::Corrupt( "prepared lexical document ids leave their page or repeat".to_owned(), @@ -4533,159 +4402,6 @@ type GateTriggerLayoutV1 = (&'static str, &'static str, &'static str); /// trigger. type ImmutableTriggerLayoutV1 = (&'static str, &'static str, &'static str, &'static str); -pub(super) fn drop_seal_triggers( - connection: &Connection, -) -> Result<(), CodeLexicalArtifactErrorV1> { - let mut statement = connection - .prepare( - "SELECT name FROM sqlite_schema WHERE type = 'trigger' AND (name LIKE 'frozen_%' OR name LIKE 'immutable_%')", - ) - .map_err(sqlite_error)?; - let names = statement - .query_map([], |row| row.get::<_, String>(0)) - .map_err(sqlite_error)? - .collect::, _>>() - .map_err(sqlite_error)?; - drop(statement); - for name in names { - if !name - .chars() - .all(|character| character.is_ascii_alphanumeric() || character == '_') - { - return Err(CodeLexicalArtifactErrorV1::Corrupt(format!( - "lexical artifact trigger name {name} is not a seal trigger" - ))); - } - connection - .execute_batch(&format!("DROP TRIGGER {name}")) - .map_err(sqlite_error)?; - } - Ok(()) -} - -pub(super) fn ensure_carried_builder_triggers( - connection: &Connection, -) -> Result<(), CodeLexicalArtifactErrorV1> { - let has_exact_vocabulary = table_exists(connection, "exact_vocabulary")?; - let has_row_dictionary_pages = table_exists(connection, "row_dictionary_pages")?; - let has_field_stats_staging = table_exists(connection, "field_stats_staging")?; - let has_row_chunk_pages = table_exists(connection, "row_chunk_pages")?; - let has_term_posting_runs = table_exists(connection, "term_posting_runs")?; - let has_exact_posting_runs = table_exists(connection, "exact_posting_runs")?; - let has_clone_index = table_exists(connection, "clone_body_payloads")?; - let has_clone_fingerprints = table_exists(connection, "clone_fingerprint_postings")?; - let has_clone_fingerprint_pages = table_exists(connection, "clone_fingerprint_postings_pages")?; - let has_source_page_cursors = table_exists(connection, "source_page_cursors")?; - let gated_layouts: [(bool, &[GateTriggerLayoutV1]); 10] = [ - ( - has_source_page_cursors, - &SOURCE_PAGE_CURSORS_BUILDER_GATE_TRIGGER_LAYOUT, - ), - ( - has_row_chunk_pages, - &ROW_CHUNK_PAGES_BUILDER_GATE_TRIGGER_LAYOUT, - ), - ( - has_exact_vocabulary, - &EXACT_VOCABULARY_BUILDER_GATE_TRIGGER_LAYOUT, - ), - ( - has_row_dictionary_pages, - &ROW_DICTIONARY_PAGES_BUILDER_GATE_TRIGGER_LAYOUT, - ), - ( - has_field_stats_staging, - &FIELD_STATS_STAGING_BUILDER_GATE_TRIGGER_LAYOUT, - ), - ( - has_term_posting_runs, - &TERM_POSTING_RUNS_BUILDER_GATE_TRIGGER_LAYOUT, - ), - ( - has_exact_posting_runs, - &EXACT_POSTING_RUNS_BUILDER_GATE_TRIGGER_LAYOUT, - ), - (has_clone_index, &CLONE_BUILDER_GATE_TRIGGER_LAYOUT), - ( - has_clone_fingerprints, - &CLONE_FINGERPRINT_BUILDER_GATE_TRIGGER_LAYOUT, - ), - ( - has_clone_fingerprint_pages, - &CLONE_FINGERPRINT_PAGES_BUILDER_GATE_TRIGGER_LAYOUT, - ), - ]; - let immutable_layouts: [(bool, &[ImmutableTriggerLayoutV1]); 7] = [ - (true, &IMMUTABLE_TRIGGER_LAYOUT), - ( - has_source_page_cursors, - &SOURCE_PAGE_CURSORS_IMMUTABLE_TRIGGER_LAYOUT, - ), - ( - has_exact_vocabulary, - &EXACT_VOCABULARY_IMMUTABLE_TRIGGER_LAYOUT, - ), - ( - has_row_dictionary_pages, - &ROW_DICTIONARY_PAGES_IMMUTABLE_TRIGGER_LAYOUT, - ), - (has_clone_index, &CLONE_IMMUTABLE_TRIGGER_LAYOUT), - ( - has_clone_fingerprints, - &CLONE_FINGERPRINT_IMMUTABLE_TRIGGER_LAYOUT, - ), - ( - has_clone_fingerprint_pages, - &CLONE_FINGERPRINT_PAGES_IMMUTABLE_TRIGGER_LAYOUT, - ), - ]; - for (present, layout) in gated_layouts { - if !present { - continue; - } - for (name, table, operation) in layout { - ensure_trigger(connection, name, table, operation, None)?; - } - } - for (present, layout) in immutable_layouts { - if !present { - continue; - } - for (name, table, operation, message) in layout { - ensure_trigger(connection, name, table, operation, Some(message))?; - } - } - Ok(()) -} - -fn ensure_trigger( - connection: &Connection, - name: &str, - table: &str, - operation: &str, - message: Option<&str>, -) -> Result<(), CodeLexicalArtifactErrorV1> { - let exists: bool = connection - .query_row( - "SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE type = 'trigger' AND name = ?1)", - [name], - |row| row.get(0), - ) - .map_err(sqlite_error)?; - if exists { - return Ok(()); - } - let sql = match message { - Some(message) => format!( - "CREATE TRIGGER {name} BEFORE {operation} ON {table} BEGIN SELECT RAISE(ABORT, '{message}'); END" - ), - None => format!( - "CREATE TRIGGER {name} BEFORE {operation} ON {table} WHEN {BUILDER_MUTATION_GATE_FUNCTION}() != 1 BEGIN SELECT RAISE(ABORT, 'private lexical builder mutation required'); END" - ), - }; - connection.execute_batch(&sql).map_err(sqlite_error) -} - fn verify_trigger_schema( connection: &Connection, name: &str, @@ -5375,7 +5091,7 @@ fn derive_exact_postings( /// Append one staged run to its key's list; a run must be non-empty and /// continue strictly after the documents already merged. -pub(super) fn append_staged_run( +fn append_staged_run( encoder: &mut PostingListEncoderV1, staged: &[u8], frequencies: bool, @@ -7113,6 +6829,20 @@ mod tests { struct ActiveControl; + #[test] + fn carried_artifact_is_incompatible_even_at_the_current_revision() { + let connection = Connection::open_in_memory().expect("SQLite"); + create_schema(&connection).expect("current artifact schema"); + verify_artifact_table_layout(&connection).expect("ordinary artifact is compatible"); + connection + .execute_batch("CREATE TABLE carried_document_shift(singleton INTEGER PRIMARY KEY, shift INTEGER NOT NULL)") + .expect("carried marker"); + assert!(matches!( + verify_artifact_table_layout(&connection), + Err(CodeLexicalArtifactErrorV1::Incompatible(_)) + )); + } + impl CodeIndexExecutionControlV1 for ActiveControl { fn is_cancelled(&self) -> bool { false diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/delta.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/delta.rs deleted file mode 100644 index 2cd27c8c38..0000000000 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/delta.rs +++ /dev/null @@ -1,1465 +0,0 @@ -//! Carry a published parent text artifact into a successor staging file. -//! -//! A successor generation reuses unchanged file occurrence ids. Those rows, -//! postings, and clone occurrences stay in the copied artifact. Only files -//! whose occurrence id is new are appended, and only occurrences the child -//! dropped are removed. A restart therefore rebuilds changed and unfinished -//! files instead of replaying the sealed corpus. - -use std::collections::{BTreeMap, BTreeSet, HashSet}; -use std::path::Path; - -use roaring::RoaringBitmap; -use rusqlite::{Connection, OptionalExtension, Params, Transaction, params}; -use serde::Deserialize; -use tracedecay_domain::{CodeGenerationId, FileOccurrenceId}; -use tracedecay_private_fs::create_private_file_retained; - -use super::builder::{ - BuilderMutationGuardV1, append_staged_run, drop_seal_triggers, ensure_carried_builder_triggers, - register_builder_mutation_gate, staging_sibling, -}; -use super::format::{ - PostingListDecoderV1, PostingListEncoderV1, RECEIPT_RESERVATION_BYTES, content_metadata_bytes, - decode_document_set, decode_fingerprint_postings, decode_term_lists, - encode_fingerprint_postings, encode_term_lists, metadata_digest, term_lists_bytes, -}; -use super::prepared::document_ngram_keys; -use super::row_codec::{ - BlockRowV1, ConnectionRowDictionaryV1, RowDictionaryEntryV1, RowDictionaryV1, - decode_artifact_row, decode_row_block, encode_row_blocks, row_file_reference, - scoring_preface_rows, -}; -use super::schema::field_code; -use super::{checkpoint, sqlite_error, sqlite_open_path}; -use crate::retrieval::lexical::LexicalFieldV1; -use crate::retrieval::lexical::projection::CodeLexicalProjectionMetadataV1; -use tracedecay_code_index::production::CodeIndexExecutionControlV1; - -use super::CodeLexicalArtifactErrorV1; - -const CARRIED_SHIFT_TABLE: &str = "carried_document_shift"; -const CARRIED_REBUILD_TABLE: &str = "carried_rebuild_occurrence"; -const CARRIED_RETIRED_NGRAM_TABLE: &str = "carried_retired_ngram"; -/// Staging-name suffix of a carry still being copied and planned. Retention -/// and the staging sweep treat it as a sidecar of its staging database. -const CARRYING_SUFFIX: &str = "-carrying"; -/// Past this multiple of changed documents, rewriting every n-gram list is -/// smaller than patching the keys those documents touch. -const CARRIED_NGRAM_PATCH_DOCUMENT_FACTOR: i64 = 4; - -#[derive(Deserialize)] -struct StoredContentPathsV1 { - logical_paths: BTreeMap, -} - -pub(super) fn carried_delta_pending( - connection: &Connection, -) -> Result { - table_exists(connection, CARRIED_SHIFT_TABLE) -} - -pub(super) fn read_carried_document_shift( - connection: &Connection, -) -> Result { - if !table_exists(connection, CARRIED_SHIFT_TABLE)? { - return Ok(0); - } - let shift: i64 = connection - .query_row( - "SELECT shift FROM carried_document_shift WHERE singleton = 1", - [], - |row| row.get(0), - ) - .map_err(sqlite_error)?; - u64::try_from(shift).map_err(|_| { - CodeLexicalArtifactErrorV1::Corrupt( - "carried lexical artifact document shift is negative".to_owned(), - ) - }) -} - -pub(super) fn read_carried_rebuild_occurrences( - connection: &Connection, -) -> Result>, CodeLexicalArtifactErrorV1> { - if !table_exists(connection, CARRIED_SHIFT_TABLE)? { - return Ok(None); - } - if !table_exists(connection, CARRIED_REBUILD_TABLE)? { - return Err(CodeLexicalArtifactErrorV1::Corrupt( - "carried lexical artifact is missing its rebuild occurrence roster".to_owned(), - )); - } - let mut statement = connection - .prepare(&format!( - "SELECT occurrence FROM {CARRIED_REBUILD_TABLE} ORDER BY occurrence" - )) - .map_err(sqlite_error)?; - let mut rows = statement.query([]).map_err(sqlite_error)?; - let mut occurrences = BTreeSet::new(); - while let Some(row) = rows.next().map_err(sqlite_error)? { - let occurrence: String = row.get(0).map_err(sqlite_error)?; - let occurrence = FileOccurrenceId::new(occurrence).map_err(|error| { - CodeLexicalArtifactErrorV1::Corrupt(format!( - "carried lexical rebuild occurrence is invalid: {error}" - )) - })?; - occurrences.insert(occurrence); - } - Ok(Some(occurrences)) -} - -/// Copy `parent` onto `staging` and retire occurrences the child metadata no -/// longer names. Returns false when no unchanged occurrence can be carried; -/// the staging path is absent in that case. -/// -/// The copy and its carry plan are written beside `staging` and renamed onto -/// it only after the plan commits. A process killed mid-copy leaves a torn -/// file under that sibling name, which the next carry or staging sweep -/// removes, never a torn staging database a restart would try to resume. -pub(super) fn stage_carried_parent( - parent: &Path, - staging: &Path, - metadata: &CodeLexicalProjectionMetadataV1, - control: &dyn CodeIndexExecutionControlV1, -) -> Result { - checkpoint(control)?; - let carrying = staging_sibling(staging, CARRYING_SUFFIX)?; - remove_sqlite_family(&carrying)?; - copy_private_file(parent, &carrying)?; - let mut connection = Connection::open(sqlite_open_path(&carrying)?).map_err(sqlite_error)?; - connection - .pragma_update(None, "journal_mode", "DELETE") - .map_err(sqlite_error)?; - let carried = (|| { - let parent_paths = read_stored_paths(&connection)?; - let unchanged = metadata - .logical_paths - .keys() - .filter(|occurrence| parent_paths.contains_key(*occurrence)) - .count(); - if unchanged == 0 { - return Ok(false); - } - let retired: BTreeSet = parent_paths - .keys() - .filter(|occurrence| !metadata.logical_paths.contains_key(*occurrence)) - .cloned() - .collect(); - let rebuild: BTreeSet = metadata - .logical_paths - .keys() - .filter(|occurrence| !parent_paths.contains_key(*occurrence)) - .cloned() - .collect(); - let gate = register_builder_mutation_gate(&connection)?; - let transaction = connection.transaction().map_err(sqlite_error)?; - let _guard = BuilderMutationGuardV1::enter(&gate)?; - drop_seal_triggers(&transaction)?; - let shift = retire_occurrences(&transaction, &metadata.generation, &retired, control)?; - clear_source_receipts(&transaction)?; - ensure_staging_tables(&transaction)?; - ensure_carried_builder_triggers(&transaction)?; - write_carried_plan(&transaction, metadata, shift, &rebuild)?; - transaction.commit().map_err(sqlite_error)?; - Ok(true) - })(); - drop(connection); - match carried { - Ok(true) => { - std::fs::rename(&carrying, staging).map_err(|error| { - CodeLexicalArtifactErrorV1::Io(format!( - "install carried lexical staging {}: {error}", - staging.display() - )) - })?; - Ok(true) - } - Ok(false) => { - remove_sqlite_family(&carrying)?; - Ok(false) - } - Err(error) => { - remove_sqlite_family(&carrying)?; - Err(error) - } - } -} - -/// Fold staged pages of a carried artifact into the sealed tables copied -/// from its parent. The caller then patches or rebuilds n-gram lists, which -/// are derived from row blocks rather than runs. -pub(super) fn merge_carried_staging( - transaction: &Transaction<'_>, - gate: &std::sync::Arc, - control: &dyn CodeIndexExecutionControlV1, -) -> Result<(), CodeLexicalArtifactErrorV1> { - let _guard = BuilderMutationGuardV1::enter(gate)?; - checkpoint(control)?; - // Staging reinstalls seal triggers so append cannot rewrite the parent - // tables. Folding the new runs has to drop them again. - drop_seal_triggers(transaction)?; - merge_term_runs(transaction, control)?; - merge_exact_runs(transaction, control)?; - merge_row_chunks(transaction)?; - merge_row_dictionary(transaction)?; - merge_field_stats(transaction)?; - merge_clone_fingerprints(transaction, control)?; - transaction - .execute_batch( - "DROP TABLE IF EXISTS term_posting_runs; - DROP TABLE IF EXISTS exact_posting_runs; - DROP TABLE IF EXISTS row_chunk_pages; - DROP TABLE IF EXISTS row_dictionary_pages; - DROP TABLE IF EXISTS field_stats_staging; - DROP TABLE IF EXISTS clone_fingerprint_postings_pages;", - ) - .map_err(sqlite_error)?; - Ok(()) -} - -fn copy_private_file(parent: &Path, staging: &Path) -> Result<(), CodeLexicalArtifactErrorV1> { - let mut source = std::fs::File::open(parent).map_err(|error| { - CodeLexicalArtifactErrorV1::Io(format!( - "open parent lexical artifact {}: {error}", - parent.display() - )) - })?; - let mut destination = create_private_file_retained(staging) - .map_err(|failure| CodeLexicalArtifactErrorV1::Io(failure.to_string()))?; - std::io::copy(&mut source, &mut destination).map_err(|error| { - CodeLexicalArtifactErrorV1::Io(format!("copy parent lexical artifact: {error}")) - })?; - destination.sync_all().map_err(|error| { - CodeLexicalArtifactErrorV1::Io(format!("sync carried lexical artifact: {error}")) - })?; - Ok(()) -} - -fn remove_sqlite_family(path: &Path) -> Result<(), CodeLexicalArtifactErrorV1> { - for suffix in ["", "-journal", "-wal", "-shm"] { - let candidate = staging_sibling(path, suffix)?; - match std::fs::remove_file(&candidate) { - Ok(()) => {} - Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} - Err(error) => { - return Err(CodeLexicalArtifactErrorV1::Io(format!( - "remove carried lexical staging {}: {error}", - candidate.display() - ))); - } - } - } - Ok(()) -} - -fn read_stored_paths( - connection: &Connection, -) -> Result, CodeLexicalArtifactErrorV1> { - let bytes: Vec = connection - .query_row( - "SELECT metadata FROM artifact_state WHERE singleton = 1", - [], - |row| row.get(0), - ) - .map_err(sqlite_error)?; - let stored: StoredContentPathsV1 = serde_json::from_slice(&bytes).map_err(|error| { - CodeLexicalArtifactErrorV1::Corrupt(format!( - "parent lexical artifact metadata is unreadable: {error}" - )) - })?; - Ok(stored.logical_paths) -} - -struct CarriedKeptRowV1 { - document_id: u32, - chunk_id: String, - row: Vec, - text: String, - field_lengths: BTreeMap, - trimmed_normalized_len: usize, -} - -fn retire_occurrences( - transaction: &Transaction<'_>, - generation: &CodeGenerationId, - retired: &BTreeSet, - control: &dyn CodeIndexExecutionControlV1, -) -> Result { - transaction - .execute_batch(&format!( - "CREATE TABLE {CARRIED_RETIRED_NGRAM_TABLE}( - kind INTEGER NOT NULL, - ngram INTEGER NOT NULL, - document_id INTEGER NOT NULL, - PRIMARY KEY(kind, ngram, document_id) - ) WITHOUT ROWID;" - )) - .map_err(sqlite_error)?; - let mut retired_documents = HashSet::new(); - let mut retired_paths = BTreeSet::new(); - let mut field_deltas = BTreeMap::::new(); - let mut retired_ngrams = Vec::new(); - if !retired.is_empty() { - let mut statement = transaction - .prepare("SELECT first_document, payload FROM row_blocks ORDER BY first_document") - .map_err(sqlite_error)?; - let blocks = statement - .query_map([], |row| { - Ok((row.get::<_, i64>(0)?, row.get::<_, Vec>(1)?)) - }) - .map_err(sqlite_error)? - .collect::, _>>() - .map_err(sqlite_error)?; - drop(statement); - let dictionary = ConnectionRowDictionaryV1::new(transaction); - for (first_document, payload) in blocks { - checkpoint(control)?; - let preface = scoring_preface_rows(first_document, &payload)?; - let rows = decode_row_block(first_document, &payload)?; - if preface.len() != rows.len() { - return Err(CodeLexicalArtifactErrorV1::Corrupt( - "lexical artifact scoring preface does not match its row block".to_owned(), - )); - } - let mut kept = Vec::new(); - let mut retired_in_block = false; - for (preface_row, row) in preface.into_iter().zip(rows) { - if preface_row.document_id != row.document_id { - return Err(CodeLexicalArtifactErrorV1::Corrupt( - "lexical artifact scoring preface document does not match its row" - .to_owned(), - )); - } - let entry = dictionary.entry(row_file_reference(&row.row)?)?; - let RowDictionaryEntryV1::File { - file_occurrence_id, - logical_path, - .. - } = entry.as_ref() - else { - return Err(CodeLexicalArtifactErrorV1::Corrupt( - "lexical artifact row file reference resolved to a non-file entry" - .to_owned(), - )); - }; - let occurrence = - FileOccurrenceId::new(file_occurrence_id.as_str()).map_err(|error| { - CodeLexicalArtifactErrorV1::Corrupt(format!( - "carried lexical row file occurrence is invalid: {error}" - )) - })?; - if retired.contains(&occurrence) { - retired_in_block = true; - retired_documents.insert(row.document_id); - retired_paths.insert(logical_path.clone()); - let decoded = decode_artifact_row( - generation, - &row.chunk_id, - &row.row, - &row.text, - &dictionary, - )?; - retired_ngrams.extend( - document_ngram_keys( - &super::super::normalized_search_text(&decoded), - decoded.sanitized_text.as_str(), - decoded.normalized_text.as_str(), - control, - )? - .into_iter() - .map(|(kind, ngram)| (kind, ngram, row.document_id)), - ); - for (field, length) in preface_row.field_lengths { - let length = i64::try_from(length).map_err(|_| { - CodeLexicalArtifactErrorV1::Contract( - "carried lexical field length exceeds i64".to_owned(), - ) - })?; - let code = field_code(field); - let delta = field_deltas.entry(code).or_insert(0); - *delta = delta.checked_sub(length).ok_or_else(|| { - CodeLexicalArtifactErrorV1::Contract( - "carried lexical field-stat delta overflowed".to_owned(), - ) - })?; - } - } else { - kept.push(CarriedKeptRowV1 { - document_id: row.document_id, - chunk_id: row.chunk_id, - row: row.row, - text: row.text, - field_lengths: preface_row.field_lengths, - trimmed_normalized_len: preface_row.trimmed_normalized_len, - }); - } - } - if !retired_in_block { - continue; - } - transaction - .execute( - "DELETE FROM row_blocks WHERE first_document = ?1", - [first_document], - ) - .map_err(sqlite_error)?; - if kept.is_empty() { - continue; - } - let encoded = encode_row_blocks( - &kept - .iter() - .map(|row| BlockRowV1 { - document_id: i64::from(row.document_id), - chunk_id: row.chunk_id.as_str(), - parent_chunk_id: None, - row: row.row.as_slice(), - text: row.text.as_str(), - field_lengths: &row.field_lengths, - trimmed_normalized_len: row.trimmed_normalized_len, - }) - .collect::>(), - )?; - let mut insert = transaction - .prepare("INSERT INTO row_blocks(first_document, payload) VALUES (?1, ?2)") - .map_err(sqlite_error)?; - for (first_document, payload) in encoded { - insert - .execute(params![first_document, payload]) - .map_err(sqlite_error)?; - } - } - // The dictionary borrow ends before the next statements. - drop(dictionary); - if !retired_ngrams.is_empty() { - let mut insert = transaction - .prepare(&format!( - "INSERT OR IGNORE INTO {CARRIED_RETIRED_NGRAM_TABLE}(kind, ngram, document_id) VALUES (?1, ?2, ?3)" - )) - .map_err(sqlite_error)?; - for (kind, ngram, document) in retired_ngrams { - insert - .execute(params![kind, ngram, i64::from(document)]) - .map_err(sqlite_error)?; - } - } - scrub_term_postings(transaction, &retired_documents, control)?; - scrub_exact_postings(transaction, &retired_documents, control)?; - scrub_fingerprints(transaction, &retired_paths, control)?; - delete_retired_clones(transaction, &retired_paths)?; - apply_field_deltas(transaction, &field_deltas)?; - let mut delete_chunks = transaction - .prepare("DELETE FROM row_chunks WHERE document_id = ?1") - .map_err(sqlite_error)?; - for document in &retired_documents { - delete_chunks - .execute([i64::from(*document)]) - .map_err(sqlite_error)?; - } - } - let max_document: Option = transaction - .query_row("SELECT MAX(document_id) FROM row_chunks", [], |row| { - row.get(0) - }) - .map_err(sqlite_error)?; - let shift = match max_document { - None => 0, - Some(max_document) => u64::try_from(max_document) - .map_err(|_| { - CodeLexicalArtifactErrorV1::Corrupt( - "carried lexical document id is negative".to_owned(), - ) - })? - .checked_add(1) - .ok_or_else(|| { - CodeLexicalArtifactErrorV1::Contract( - "carried lexical document shift overflowed".to_owned(), - ) - })?, - }; - Ok(shift) -} - -fn clear_source_receipts(transaction: &Transaction<'_>) -> Result<(), CodeLexicalArtifactErrorV1> { - // A sealed parent keeps `source_pages` and drops the per-page cursor - // table at the statistics step. Delete only the tables that are present; - // the cursor table is recreated empty before append. - let mut statements = vec![ - "DELETE FROM source_pages".to_owned(), - "UPDATE content_epoch SET epoch = 0 WHERE singleton = 1".to_owned(), - ]; - if table_exists(transaction, "source_page_cursors")? { - statements.insert(1, "DELETE FROM source_page_cursors".to_owned()); - } - transaction - .execute_batch(&statements.join(";\n")) - .map_err(sqlite_error) -} - -fn ensure_staging_tables(transaction: &Transaction<'_>) -> Result<(), CodeLexicalArtifactErrorV1> { - transaction - .execute_batch( - "CREATE TABLE IF NOT EXISTS finalization_state ( - singleton INTEGER PRIMARY KEY CHECK(singleton = 1), - state BLOB NOT NULL - ); - CREATE TABLE IF NOT EXISTS source_page_cursors ( - page_ordinal INTEGER PRIMARY KEY, - page_digest TEXT NOT NULL, - cumulative_digest TEXT NOT NULL, - payload_bytes INTEGER NOT NULL, - next_cursor BLOB NOT NULL - ); - CREATE TABLE IF NOT EXISTS row_chunk_pages ( - document_id INTEGER PRIMARY KEY, - chunk_id BLOB NOT NULL - ); - CREATE TABLE IF NOT EXISTS term_posting_runs ( - page_ordinal INTEGER NOT NULL, - term TEXT NOT NULL, - field INTEGER NOT NULL, - postings BLOB NOT NULL, - PRIMARY KEY(page_ordinal, term, field) - ) WITHOUT ROWID; - CREATE TABLE IF NOT EXISTS exact_posting_runs ( - page_ordinal INTEGER NOT NULL, - term_id INTEGER NOT NULL, - field INTEGER NOT NULL, - documents BLOB NOT NULL, - PRIMARY KEY(page_ordinal, term_id, field) - ) WITHOUT ROWID; - CREATE TABLE IF NOT EXISTS row_dictionary_pages ( - page_ordinal INTEGER NOT NULL, - entry_id INTEGER NOT NULL, - entry BLOB NOT NULL, - PRIMARY KEY(page_ordinal, entry_id) - ) WITHOUT ROWID; - CREATE TABLE IF NOT EXISTS field_stats_staging ( - field INTEGER PRIMARY KEY, - total_length INTEGER NOT NULL - ) WITHOUT ROWID; - CREATE TABLE IF NOT EXISTS clone_fingerprint_postings_pages ( - language TEXT NOT NULL, - class INTEGER NOT NULL, - normalization_revision INTEGER NOT NULL, - fingerprint INTEGER NOT NULL, - occurrence_ordinal INTEGER NOT NULL, - token_position INTEGER NOT NULL - );", - ) - .map_err(sqlite_error) -} - -fn write_carried_plan( - transaction: &Transaction<'_>, - metadata: &CodeLexicalProjectionMetadataV1, - shift: u64, - rebuild: &BTreeSet, -) -> Result<(), CodeLexicalArtifactErrorV1> { - let metadata_bytes = content_metadata_bytes(metadata)?; - let digest = metadata_digest(metadata)?; - let shift = i64::try_from(shift).map_err(|_| { - CodeLexicalArtifactErrorV1::Contract( - "carried lexical document shift exceeds i64".to_owned(), - ) - })?; - transaction - .execute( - "UPDATE artifact_state SET metadata = ?1, metadata_digest = ?2, receipt = ?3 WHERE singleton = 1", - params![ - metadata_bytes, - digest.as_str(), - vec![0u8; RECEIPT_RESERVATION_BYTES] - ], - ) - .map_err(sqlite_error)?; - transaction - .execute_batch(&format!( - "DROP TABLE IF EXISTS {CARRIED_SHIFT_TABLE}; - DROP TABLE IF EXISTS {CARRIED_REBUILD_TABLE}; - CREATE TABLE {CARRIED_SHIFT_TABLE}(singleton INTEGER PRIMARY KEY CHECK(singleton = 1), shift INTEGER NOT NULL); - CREATE TABLE {CARRIED_REBUILD_TABLE}(occurrence TEXT PRIMARY KEY) WITHOUT ROWID;" - )) - .map_err(sqlite_error)?; - transaction - .execute( - &format!("INSERT INTO {CARRIED_SHIFT_TABLE}(singleton, shift) VALUES (1, ?1)"), - [shift], - ) - .map_err(sqlite_error)?; - let mut insert = transaction - .prepare(&format!( - "INSERT INTO {CARRIED_REBUILD_TABLE}(occurrence) VALUES (?1)" - )) - .map_err(sqlite_error)?; - for occurrence in rebuild { - insert - .execute([occurrence.as_str()]) - .map_err(sqlite_error)?; - } - Ok(()) -} - -fn scrub_term_postings( - transaction: &Transaction<'_>, - retired: &HashSet, - control: &dyn CodeIndexExecutionControlV1, -) -> Result<(), CodeLexicalArtifactErrorV1> { - if retired.is_empty() { - return Ok(()); - } - let subtoken = field_code(LexicalFieldV1::Subtoken); - let mut statement = transaction - .prepare("SELECT term, lists FROM term_postings") - .map_err(sqlite_error)?; - let rows = statement - .query_map([], |row| { - Ok((row.get::<_, String>(0)?, row.get::<_, Vec>(1)?)) - }) - .map_err(sqlite_error)? - .collect::, _>>() - .map_err(sqlite_error)?; - drop(statement); - let mut update = transaction - .prepare("UPDATE term_postings SET in_fuzzy = ?2, lists = ?3 WHERE term = ?1") - .map_err(sqlite_error)?; - let mut delete = transaction - .prepare("DELETE FROM term_postings WHERE term = ?1") - .map_err(sqlite_error)?; - for (ordinal, (term, lists)) in rows.into_iter().enumerate() { - if ordinal.is_multiple_of(1024) { - checkpoint(control)?; - } - let inflated = term_lists_bytes(&lists)?; - let decoded = decode_term_lists(&inflated)?; - let mut changed = false; - let mut kept = Vec::new(); - for (field, _frequency, postings) in decoded { - match filter_posting_list(postings, true, retired)? { - None => kept.push((field, postings.to_vec())), - Some(filtered) if filtered.is_empty() => changed = true, - Some(filtered) => { - changed = true; - kept.push((field, filtered)); - } - } - } - if !changed { - continue; - } - if kept.is_empty() { - delete.execute([&term]).map_err(sqlite_error)?; - continue; - } - let encoded = kept - .iter() - .map(|(field, postings)| { - let count = u64::try_from(posting_count(postings, true)?).map_err(|_| { - CodeLexicalArtifactErrorV1::Contract( - "carried lexical posting count exceeds u64".to_owned(), - ) - })?; - Ok((*field, count, postings.clone())) - }) - .collect::, CodeLexicalArtifactErrorV1>>()?; - let in_fuzzy = encoded.iter().any(|(field, _, _)| *field != subtoken); - update - .execute(params![term, in_fuzzy, encode_term_lists(&encoded)?]) - .map_err(sqlite_error)?; - } - Ok(()) -} - -fn scrub_exact_postings( - transaction: &Transaction<'_>, - retired: &HashSet, - control: &dyn CodeIndexExecutionControlV1, -) -> Result<(), CodeLexicalArtifactErrorV1> { - if retired.is_empty() { - return Ok(()); - } - let mut statement = transaction - .prepare("SELECT term_id, field, documents FROM exact_postings") - .map_err(sqlite_error)?; - let rows = statement - .query_map([], |row| { - Ok(( - row.get::<_, i64>(0)?, - row.get::<_, i64>(1)?, - row.get::<_, Vec>(2)?, - )) - }) - .map_err(sqlite_error)? - .collect::, _>>() - .map_err(sqlite_error)?; - drop(statement); - let mut update = transaction - .prepare("UPDATE exact_postings SET documents = ?3 WHERE term_id = ?1 AND field = ?2") - .map_err(sqlite_error)?; - let mut delete = transaction - .prepare("DELETE FROM exact_postings WHERE term_id = ?1 AND field = ?2") - .map_err(sqlite_error)?; - for (ordinal, (term_id, field, documents)) in rows.into_iter().enumerate() { - if ordinal.is_multiple_of(1024) { - checkpoint(control)?; - } - let Some(filtered) = filter_posting_list(&documents, false, retired)? else { - continue; - }; - if filtered.is_empty() { - delete - .execute(params![term_id, field]) - .map_err(sqlite_error)?; - } else { - update - .execute(params![term_id, field, filtered]) - .map_err(sqlite_error)?; - } - } - Ok(()) -} - -fn filter_posting_list( - postings: &[u8], - frequencies: bool, - retired: &HashSet, -) -> Result>, CodeLexicalArtifactErrorV1> { - let mut encoder = PostingListEncoderV1::new(frequencies); - let mut changed = false; - for posting in PostingListDecoderV1::new(postings, frequencies) { - let (document, frequency) = posting?; - if retired.contains(&document) { - changed = true; - continue; - } - encoder.push(document, frequency).map_err(|_| { - CodeLexicalArtifactErrorV1::Corrupt( - "carried lexical posting list could not be rewritten".to_owned(), - ) - })?; - } - if !changed { - return Ok(None); - } - if encoder.len() == 0 { - return Ok(Some(Vec::new())); - } - Ok(Some(encoder.finish()?)) -} - -fn posting_count(postings: &[u8], frequencies: bool) -> Result { - let mut count = 0usize; - for posting in PostingListDecoderV1::new(postings, frequencies) { - let _ = posting?; - count = count.checked_add(1).ok_or_else(|| { - CodeLexicalArtifactErrorV1::Contract( - "carried lexical posting count overflowed".to_owned(), - ) - })?; - } - Ok(count) -} - -fn scrub_fingerprints( - transaction: &Transaction<'_>, - retired_paths: &BTreeSet, - control: &dyn CodeIndexExecutionControlV1, -) -> Result<(), CodeLexicalArtifactErrorV1> { - if retired_paths.is_empty() || !table_exists(transaction, "clone_fingerprint_postings")? { - return Ok(()); - } - let mut ordinals = transaction - .prepare("SELECT ordinal FROM clone_occurrences WHERE path = ?1") - .map_err(sqlite_error)?; - let mut retired_ordinals = HashSet::new(); - for path in retired_paths { - let rows = ordinals - .query_map([path.as_str()], |row| row.get::<_, i64>(0)) - .map_err(sqlite_error)?; - for ordinal in rows { - let ordinal = ordinal.map_err(sqlite_error)?; - let ordinal = u32::try_from(ordinal).map_err(|_| { - CodeLexicalArtifactErrorV1::Corrupt( - "carried clone occurrence ordinal exceeds u32".to_owned(), - ) - })?; - retired_ordinals.insert(ordinal); - } - } - drop(ordinals); - if retired_ordinals.is_empty() { - return Ok(()); - } - let mut statement = transaction - .prepare( - "SELECT language, class, normalization_revision, fingerprint, postings FROM clone_fingerprint_postings", - ) - .map_err(sqlite_error)?; - let rows = statement - .query_map([], |row| { - Ok(( - row.get::<_, String>(0)?, - row.get::<_, i64>(1)?, - row.get::<_, i64>(2)?, - row.get::<_, i64>(3)?, - row.get::<_, Vec>(4)?, - )) - }) - .map_err(sqlite_error)? - .collect::, _>>() - .map_err(sqlite_error)?; - drop(statement); - let mut update = transaction - .prepare( - "UPDATE clone_fingerprint_postings SET posting_count = ?5, postings = ?6 WHERE language = ?1 AND class = ?2 AND normalization_revision = ?3 AND fingerprint = ?4", - ) - .map_err(sqlite_error)?; - let mut delete = transaction - .prepare( - "DELETE FROM clone_fingerprint_postings WHERE language = ?1 AND class = ?2 AND normalization_revision = ?3 AND fingerprint = ?4", - ) - .map_err(sqlite_error)?; - for (ordinal, (language, class, revision, fingerprint, postings)) in - rows.into_iter().enumerate() - { - if ordinal.is_multiple_of(1024) { - checkpoint(control)?; - } - let decoded = decode_fingerprint_postings(&postings)?; - if !decoded - .iter() - .any(|(occurrence, _)| retired_ordinals.contains(occurrence)) - { - continue; - } - let kept = decoded - .into_iter() - .filter(|(occurrence, _)| !retired_ordinals.contains(occurrence)) - .collect::>(); - if kept.is_empty() { - delete - .execute(params![language, class, revision, fingerprint]) - .map_err(sqlite_error)?; - continue; - } - let count = i64::try_from(kept.len()).map_err(|_| { - CodeLexicalArtifactErrorV1::Contract( - "carried clone fingerprint posting count exceeds i64".to_owned(), - ) - })?; - update - .execute(params![ - language, - class, - revision, - fingerprint, - count, - encode_fingerprint_postings(&kept)? - ]) - .map_err(sqlite_error)?; - } - Ok(()) -} - -fn delete_retired_clones( - transaction: &Transaction<'_>, - retired_paths: &BTreeSet, -) -> Result<(), CodeLexicalArtifactErrorV1> { - if retired_paths.is_empty() { - return Ok(()); - } - let mut delete_postings = transaction - .prepare( - "DELETE FROM clone_exact_postings WHERE occurrence_ordinal IN (SELECT ordinal FROM clone_occurrences WHERE path = ?1)", - ) - .map_err(sqlite_error)?; - let mut delete_occurrences = transaction - .prepare("DELETE FROM clone_occurrences WHERE path = ?1") - .map_err(sqlite_error)?; - for path in retired_paths { - delete_postings - .execute([path.as_str()]) - .map_err(sqlite_error)?; - delete_occurrences - .execute([path.as_str()]) - .map_err(sqlite_error)?; - } - transaction - .execute( - "DELETE FROM clone_body_payloads WHERE ordinal NOT IN (SELECT payload_ordinal FROM clone_occurrences)", - [], - ) - .map_err(sqlite_error)?; - Ok(()) -} - -fn apply_field_deltas( - transaction: &Transaction<'_>, - deltas: &BTreeMap, -) -> Result<(), CodeLexicalArtifactErrorV1> { - let mut select = transaction - .prepare("SELECT total_length FROM field_stats WHERE field = ?1") - .map_err(sqlite_error)?; - let mut update = transaction - .prepare("UPDATE field_stats SET total_length = ?2 WHERE field = ?1") - .map_err(sqlite_error)?; - for (field, delta) in deltas { - let current: i64 = select - .query_row([field], |row| row.get(0)) - .optional() - .map_err(sqlite_error)? - .ok_or_else(|| { - CodeLexicalArtifactErrorV1::Corrupt( - "carried lexical field statistic is missing a retired field".to_owned(), - ) - })?; - let next = current.checked_add(*delta).ok_or_else(|| { - CodeLexicalArtifactErrorV1::Contract( - "carried lexical field statistic overflowed".to_owned(), - ) - })?; - if next < 0 { - return Err(CodeLexicalArtifactErrorV1::Corrupt( - "carried lexical field statistic became negative".to_owned(), - )); - } - update.execute(params![field, next]).map_err(sqlite_error)?; - } - Ok(()) -} - -fn merge_term_runs( - transaction: &Transaction<'_>, - control: &dyn CodeIndexExecutionControlV1, -) -> Result<(), CodeLexicalArtifactErrorV1> { - if !table_exists(transaction, "term_posting_runs")? { - return Ok(()); - } - let subtoken = field_code(LexicalFieldV1::Subtoken); - let mut select = transaction - .prepare( - "SELECT term, field, postings FROM term_posting_runs ORDER BY term, field, page_ordinal", - ) - .map_err(sqlite_error)?; - let runs = select - .query_map([], |row| { - Ok(( - row.get::<_, String>(0)?, - row.get::<_, i64>(1)?, - row.get::<_, Vec>(2)?, - )) - }) - .map_err(sqlite_error)? - .collect::, _>>() - .map_err(sqlite_error)?; - drop(select); - let mut existing = transaction - .prepare("SELECT lists FROM term_postings WHERE term = ?1") - .map_err(sqlite_error)?; - let mut upsert = transaction - .prepare( - "INSERT INTO term_postings(term, in_fuzzy, lists) VALUES (?1, ?2, ?3) - ON CONFLICT(term) DO UPDATE SET in_fuzzy = excluded.in_fuzzy, lists = excluded.lists", - ) - .map_err(sqlite_error)?; - let mut index = 0usize; - while index < runs.len() { - checkpoint(control)?; - let term = runs[index].0.clone(); - let mut fields: BTreeMap = BTreeMap::new(); - if let Some(lists) = existing - .query_row([&term], |row| row.get::<_, Vec>(0)) - .optional() - .map_err(sqlite_error)? - { - for (field, _frequency, postings) in decode_term_lists(&term_lists_bytes(&lists)?)? { - let mut encoder = PostingListEncoderV1::new(true); - append_staged_run(&mut encoder, postings, true)?; - fields.insert(field, encoder); - } - } - while index < runs.len() && runs[index].0 == term { - let field = runs[index].1; - let postings = &runs[index].2; - let encoder = fields - .entry(field) - .or_insert_with(|| PostingListEncoderV1::new(true)); - append_staged_run(encoder, postings, true)?; - index += 1; - } - let lists = fields - .into_iter() - .map(|(field, encoder)| { - let frequency = encoder.len(); - Ok((field, frequency, encoder.finish()?)) - }) - .collect::, CodeLexicalArtifactErrorV1>>()?; - let in_fuzzy = lists.iter().any(|(field, _, _)| *field != subtoken); - upsert - .execute(params![term, in_fuzzy, encode_term_lists(&lists)?]) - .map_err(sqlite_error)?; - } - Ok(()) -} - -fn merge_exact_runs( - transaction: &Transaction<'_>, - control: &dyn CodeIndexExecutionControlV1, -) -> Result<(), CodeLexicalArtifactErrorV1> { - if !table_exists(transaction, "exact_posting_runs")? { - return Ok(()); - } - let mut select = transaction - .prepare( - "SELECT term_id, field, documents FROM exact_posting_runs ORDER BY term_id, field, page_ordinal", - ) - .map_err(sqlite_error)?; - let runs = select - .query_map([], |row| { - Ok(( - row.get::<_, i64>(0)?, - row.get::<_, i64>(1)?, - row.get::<_, Vec>(2)?, - )) - }) - .map_err(sqlite_error)? - .collect::, _>>() - .map_err(sqlite_error)?; - drop(select); - let mut existing = transaction - .prepare("SELECT documents FROM exact_postings WHERE term_id = ?1 AND field = ?2") - .map_err(sqlite_error)?; - let mut upsert = transaction - .prepare( - "INSERT INTO exact_postings(term_id, field, documents) VALUES (?1, ?2, ?3) - ON CONFLICT(term_id, field) DO UPDATE SET documents = excluded.documents", - ) - .map_err(sqlite_error)?; - let mut index = 0usize; - while index < runs.len() { - checkpoint(control)?; - let term_id = runs[index].0; - let field = runs[index].1; - let mut encoder = PostingListEncoderV1::new(false); - if let Some(documents) = existing - .query_row(params![term_id, field], |row| row.get::<_, Vec>(0)) - .optional() - .map_err(sqlite_error)? - { - append_staged_run(&mut encoder, &documents, false)?; - } - while index < runs.len() && runs[index].0 == term_id && runs[index].1 == field { - append_staged_run(&mut encoder, &runs[index].2, false)?; - index += 1; - } - upsert - .execute(params![term_id, field, encoder.finish()?]) - .map_err(sqlite_error)?; - } - Ok(()) -} - -fn merge_row_chunks(transaction: &Transaction<'_>) -> Result<(), CodeLexicalArtifactErrorV1> { - if !table_exists(transaction, "row_chunk_pages")? { - return Ok(()); - } - transaction - .execute_batch( - "INSERT INTO row_chunks(chunk_id, document_id) - SELECT chunk_id, document_id FROM row_chunk_pages ORDER BY chunk_id;", - ) - .map_err(sqlite_error) -} - -fn merge_row_dictionary(transaction: &Transaction<'_>) -> Result<(), CodeLexicalArtifactErrorV1> { - if !table_exists(transaction, "row_dictionary_pages")? { - return Ok(()); - } - let collided: bool = transaction - .query_row( - "SELECT EXISTS( - SELECT 1 FROM row_dictionary_pages AS page - JOIN row_dictionary AS sealed ON sealed.entry_id = page.entry_id - WHERE sealed.entry != page.entry - )", - [], - |row| row.get(0), - ) - .map_err(sqlite_error)?; - if collided { - return Err(CodeLexicalArtifactErrorV1::Corrupt( - "carried lexical row dictionary identifier collided".to_owned(), - )); - } - transaction - .execute_batch( - "INSERT INTO row_dictionary(entry_id, entry) - SELECT entry_id, MIN(entry) FROM row_dictionary_pages - WHERE entry_id NOT IN (SELECT entry_id FROM row_dictionary) - GROUP BY entry_id;", - ) - .map_err(sqlite_error) -} - -fn merge_field_stats(transaction: &Transaction<'_>) -> Result<(), CodeLexicalArtifactErrorV1> { - if !table_exists(transaction, "field_stats_staging")? { - return Ok(()); - } - transaction - .execute_batch( - "UPDATE field_stats - SET total_length = total_length + ( - SELECT total_length FROM field_stats_staging - WHERE field_stats_staging.field = field_stats.field - ) - WHERE field IN (SELECT field FROM field_stats_staging); - INSERT INTO field_stats(field, total_length) - SELECT field, total_length FROM field_stats_staging - WHERE field NOT IN (SELECT field FROM field_stats);", - ) - .map_err(sqlite_error) -} - -fn merge_clone_fingerprints( - transaction: &Transaction<'_>, - control: &dyn CodeIndexExecutionControlV1, -) -> Result<(), CodeLexicalArtifactErrorV1> { - if !table_exists(transaction, "clone_fingerprint_postings_pages")? { - return Ok(()); - } - let mut select = transaction - .prepare( - "SELECT language, class, normalization_revision, fingerprint, occurrence_ordinal, token_position - FROM clone_fingerprint_postings_pages - ORDER BY language, class, normalization_revision, fingerprint, occurrence_ordinal, token_position", - ) - .map_err(sqlite_error)?; - let rows = select - .query_map([], |row| { - Ok(( - row.get::<_, String>(0)?, - row.get::<_, i64>(1)?, - row.get::<_, i64>(2)?, - row.get::<_, i64>(3)?, - row.get::<_, i64>(4)?, - row.get::<_, i64>(5)?, - )) - }) - .map_err(sqlite_error)? - .collect::, _>>() - .map_err(sqlite_error)?; - drop(select); - if rows.is_empty() { - return Ok(()); - } - let mut existing = transaction - .prepare( - "SELECT postings FROM clone_fingerprint_postings WHERE language = ?1 AND class = ?2 AND normalization_revision = ?3 AND fingerprint = ?4", - ) - .map_err(sqlite_error)?; - let mut upsert = transaction - .prepare( - "INSERT INTO clone_fingerprint_postings(language, class, normalization_revision, fingerprint, posting_count, postings) - VALUES (?1, ?2, ?3, ?4, ?5, ?6) - ON CONFLICT(language, class, normalization_revision, fingerprint) - DO UPDATE SET posting_count = excluded.posting_count, postings = excluded.postings", - ) - .map_err(sqlite_error)?; - let mut index = 0usize; - while index < rows.len() { - checkpoint(control)?; - let language = rows[index].0.clone(); - let class = rows[index].1; - let revision = rows[index].2; - let fingerprint = rows[index].3; - let mut postings = Vec::new(); - if let Some(stored) = existing - .query_row(params![language, class, revision, fingerprint], |row| { - row.get::<_, Vec>(0) - }) - .optional() - .map_err(sqlite_error)? - { - postings = decode_fingerprint_postings(&stored)?; - } - while index < rows.len() - && rows[index].0 == language - && rows[index].1 == class - && rows[index].2 == revision - && rows[index].3 == fingerprint - { - let occurrence = u32::try_from(rows[index].4).map_err(|_| { - CodeLexicalArtifactErrorV1::Contract( - "carried clone fingerprint occurrence exceeds u32".to_owned(), - ) - })?; - let position = u32::try_from(rows[index].5).map_err(|_| { - CodeLexicalArtifactErrorV1::Contract( - "carried clone fingerprint position exceeds u32".to_owned(), - ) - })?; - postings.push((occurrence, position)); - index += 1; - } - postings.sort_by_key(|posting| *posting); - postings.dedup(); - let count = i64::try_from(postings.len()).map_err(|_| { - CodeLexicalArtifactErrorV1::Contract( - "carried clone fingerprint posting count exceeds i64".to_owned(), - ) - })?; - upsert - .execute(params![ - language, - class, - revision, - fingerprint, - count, - encode_fingerprint_postings(&postings)? - ]) - .map_err(sqlite_error)?; - } - Ok(()) -} - -struct NgramDocumentDeltaV1 { - remove: BTreeSet, - add: BTreeSet, -} - -/// Whether patching the changed documents' n-gram keys is a smaller walk -/// than rebuilding every sealed list. -pub(super) fn carried_ngram_patch_fits( - transaction: &Transaction<'_>, -) -> Result { - if !table_exists(transaction, CARRIED_RETIRED_NGRAM_TABLE)? { - return Err(CodeLexicalArtifactErrorV1::Corrupt( - "carried lexical artifact is missing its retired n-gram roster".to_owned(), - )); - } - let total = count_sql(transaction, "SELECT COUNT(*) FROM row_chunks", [])?; - if total == 0 { - return Ok(false); - } - let shift = i64::try_from(read_carried_document_shift(transaction)?).map_err(|_| { - CodeLexicalArtifactErrorV1::Contract( - "carried lexical document shift exceeds i64".to_owned(), - ) - })?; - let added = count_sql( - transaction, - "SELECT COUNT(*) FROM row_chunks WHERE document_id >= ?1", - [shift], - )?; - let retired = count_sql( - transaction, - &format!("SELECT COUNT(DISTINCT document_id) FROM {CARRIED_RETIRED_NGRAM_TABLE}"), - [], - )?; - Ok(retired - .saturating_add(added) - .saturating_mul(CARRIED_NGRAM_PATCH_DOCUMENT_FACTOR) - < total) -} - -/// Remove retired documents from the n-gram lists they contributed and add -/// the documents appended after the carry shift. Lists no changed document -/// touches stay as the parent sealed them. -pub(super) fn patch_carried_ngram_postings( - transaction: &Transaction<'_>, - generation: &CodeGenerationId, - control: &dyn CodeIndexExecutionControlV1, -) -> Result<(), CodeLexicalArtifactErrorV1> { - let deltas = carried_ngram_deltas(transaction, generation, control)?; - let mut select = transaction - .prepare("SELECT documents FROM ngram_postings WHERE kind = ?1 AND ngram = ?2") - .map_err(sqlite_error)?; - let mut update = transaction - .prepare( - "UPDATE ngram_postings SET document_frequency = ?3, documents = ?4 WHERE kind = ?1 AND ngram = ?2", - ) - .map_err(sqlite_error)?; - let mut insert = transaction - .prepare( - "INSERT INTO ngram_postings(kind, ngram, document_frequency, documents) VALUES (?1, ?2, ?3, ?4)", - ) - .map_err(sqlite_error)?; - let mut delete = transaction - .prepare("DELETE FROM ngram_postings WHERE kind = ?1 AND ngram = ?2") - .map_err(sqlite_error)?; - for (ordinal, ((kind, ngram), delta)) in deltas.iter().enumerate() { - if ordinal.is_multiple_of(1024) { - checkpoint(control)?; - } - let stored: Option> = select - .query_row(params![kind, ngram], |row| row.get(0)) - .optional() - .map_err(sqlite_error)?; - if stored.is_none() && !delta.remove.is_empty() { - return Err(CodeLexicalArtifactErrorV1::Corrupt( - "carried lexical n-gram list is missing a retired document".to_owned(), - )); - } - let mut documents = match &stored { - Some(stored) => decode_document_set(stored)?, - None => RoaringBitmap::new(), - }; - for document in &delta.remove { - if !documents.remove(*document) { - return Err(CodeLexicalArtifactErrorV1::Corrupt( - "carried lexical n-gram list does not contain its retired document".to_owned(), - )); - } - } - for document in &delta.add { - if !documents.insert(*document) { - return Err(CodeLexicalArtifactErrorV1::Corrupt( - "carried lexical n-gram list already contains its appended document".to_owned(), - )); - } - } - if documents.is_empty() { - delete.execute(params![kind, ngram]).map_err(sqlite_error)?; - continue; - } - let frequency = i64::try_from(documents.len()).map_err(|_| { - CodeLexicalArtifactErrorV1::Contract( - "carried lexical n-gram document frequency exceeds i64".to_owned(), - ) - })?; - let sealed = seal_ngram_documents(&documents)?; - if stored.is_some() { - update - .execute(params![kind, ngram, frequency, sealed]) - .map_err(sqlite_error)?; - } else { - insert - .execute(params![kind, ngram, frequency, sealed]) - .map_err(sqlite_error)?; - } - } - Ok(()) -} - -pub(super) fn discard_carried_ngram_plan( - transaction: &Transaction<'_>, -) -> Result<(), CodeLexicalArtifactErrorV1> { - transaction - .execute_batch(&format!( - "DROP TABLE IF EXISTS {CARRIED_RETIRED_NGRAM_TABLE};" - )) - .map_err(sqlite_error) -} - -fn carried_ngram_deltas( - transaction: &Transaction<'_>, - generation: &CodeGenerationId, - control: &dyn CodeIndexExecutionControlV1, -) -> Result, CodeLexicalArtifactErrorV1> { - let mut deltas: BTreeMap<(i64, i64), NgramDocumentDeltaV1> = BTreeMap::new(); - let mut retired = transaction - .prepare(&format!( - "SELECT kind, ngram, document_id FROM {CARRIED_RETIRED_NGRAM_TABLE} ORDER BY kind, ngram" - )) - .map_err(sqlite_error)?; - let mut rows = retired.query([]).map_err(sqlite_error)?; - let mut visited = 0usize; - while let Some(row) = rows.next().map_err(sqlite_error)? { - if visited.is_multiple_of(1024) { - checkpoint(control)?; - } - visited += 1; - let kind: i64 = row.get(0).map_err(sqlite_error)?; - let ngram: i64 = row.get(1).map_err(sqlite_error)?; - let document = - u32::try_from(row.get::<_, i64>(2).map_err(sqlite_error)?).map_err(|_| { - CodeLexicalArtifactErrorV1::Corrupt( - "carried lexical retired n-gram document exceeds u32".to_owned(), - ) - })?; - deltas - .entry((kind, ngram)) - .or_insert_with(|| NgramDocumentDeltaV1 { - remove: BTreeSet::new(), - add: BTreeSet::new(), - }) - .remove - .insert(document); - } - drop(rows); - drop(retired); - let shift = i64::try_from(read_carried_document_shift(transaction)?).map_err(|_| { - CodeLexicalArtifactErrorV1::Contract( - "carried lexical document shift exceeds i64".to_owned(), - ) - })?; - let mut blocks = transaction - .prepare( - "SELECT first_document, payload FROM row_blocks WHERE first_document >= ?1 ORDER BY first_document", - ) - .map_err(sqlite_error)?; - let stored = blocks - .query_map(params![shift], |row| { - Ok((row.get::<_, i64>(0)?, row.get::<_, Vec>(1)?)) - }) - .map_err(sqlite_error)? - .collect::, _>>() - .map_err(sqlite_error)?; - drop(blocks); - let dictionary = ConnectionRowDictionaryV1::new(transaction); - for (first_document, payload) in stored { - checkpoint(control)?; - for row in decode_row_block(first_document, &payload)? { - if i64::from(row.document_id) < shift { - continue; - } - let decoded = - decode_artifact_row(generation, &row.chunk_id, &row.row, &row.text, &dictionary)?; - for (kind, ngram) in document_ngram_keys( - &super::super::normalized_search_text(&decoded), - decoded.sanitized_text.as_str(), - decoded.normalized_text.as_str(), - control, - )? { - deltas - .entry((kind, ngram)) - .or_insert_with(|| NgramDocumentDeltaV1 { - remove: BTreeSet::new(), - add: BTreeSet::new(), - }) - .add - .insert(row.document_id); - } - } - } - Ok(deltas) -} - -fn seal_ngram_documents(documents: &RoaringBitmap) -> Result, CodeLexicalArtifactErrorV1> { - let mut encoder = PostingListEncoderV1::new(false); - for document in documents { - encoder.push(document, 1)?; - } - encoder.finish_document_set() -} - -fn count_sql( - transaction: &Transaction<'_>, - sql: &str, - params: impl Params, -) -> Result { - transaction - .query_row(sql, params, |row| row.get(0)) - .map_err(sqlite_error) -} - -fn table_exists(connection: &Connection, table: &str) -> Result { - connection - .query_row( - "SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE type = 'table' AND name = ?1)", - [table], - |row| row.get(0), - ) - .map_err(sqlite_error) -} diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/format.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/format.rs index da2064baab..db110f0a81 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/format.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/format.rs @@ -14,9 +14,9 @@ use tracedecay_domain::{ }; use super::super::{CodeLexicalProjectionMetadataV1, LexicalFieldV1, ProjectedChunkV1}; -use super::CodeLexicalArtifactErrorV1; use super::clone_census::CodeLexicalCloneIndexCensusV1; use super::schema::{CODE_LEXICAL_ARTIFACT_FORMAT_REVISION_V1, digest_domain_for_revision}; +use super::{CodeLexicalArtifactErrorV1, sqlite_error}; pub(super) use super::schema::{SERVING_INDEX_STEP_COUNT_V11, STATISTICS_STEP_COUNT_V11}; @@ -429,6 +429,20 @@ const ARTIFACT_TABLE_LAYOUT: [(&str, bool, &[ColumnShapeV1]); 12] = [ pub(super) fn verify_artifact_table_layout( connection: &Connection, ) -> Result<(), CodeLexicalArtifactErrorV1> { + // The removed parent-carry path sealed incomplete source receipts under + // this same format revision. Its marker survives sealing, so rebuild it. + let carried: bool = connection + .query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE type = 'table' AND name = 'carried_document_shift')", + [], + |row| row.get(0), + ) + .map_err(sqlite_error)?; + if carried { + return Err(CodeLexicalArtifactErrorV1::Incompatible( + "parent-carried lexical artifact requires a full rebuild".to_owned(), + )); + } for (table, expected_without_rowid, expected_columns) in ARTIFACT_TABLE_LAYOUT { let without_rowid: Option = connection .query_row( diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/prepared.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/prepared.rs index 09209dcc32..f98e301b51 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/prepared.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/prepared.rs @@ -172,7 +172,6 @@ pub(super) fn prepare_page( page: &VerifiedSealedLexicalPageV1, previous_cursor: Option>, preparation_scratch_bytes: usize, - document_id_shift: u64, control: &dyn CodeIndexExecutionControlV1, ) -> Result { checkpoint(control)?; @@ -210,7 +209,6 @@ pub(super) fn prepare_page( } let document = first_document .checked_add(u64::try_from(offset).map_err(contract_number)?) - .and_then(|document| document.checked_add(document_id_shift)) .ok_or_else(|| { CodeLexicalArtifactErrorV1::Contract( "lexical artifact prepared document id overflowed".to_owned(), diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/row_codec.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/row_codec.rs index 689ec390cc..1d4f095f64 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/row_codec.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/row_codec.rs @@ -1054,20 +1054,6 @@ impl RowBlockV1 { } } -/// Scoring preface of one stored block, without inflating its rows. -pub(super) fn scoring_preface_rows( - first_document: i64, - stored: &[u8], -) -> Result, CodeLexicalArtifactErrorV1> { - let (preface, _) = split_scoring_preface(stored)?; - decode_preface_rows(first_document, preface) -} - -/// The file-dictionary reference at the start of one stored row. -pub(super) fn row_file_reference(row: &[u8]) -> Result { - RowCursorV1 { bytes: row }.take_reference() -} - /// Every row of one stored block. pub(super) fn decode_row_block( first_document: i64, diff --git a/crates/tracedecay/tests/daemon_suite/indexing_lifecycle_test.rs b/crates/tracedecay/tests/daemon_suite/indexing_lifecycle_test.rs index 94eb82adf4..fde2a04c63 100644 --- a/crates/tracedecay/tests/daemon_suite/indexing_lifecycle_test.rs +++ b/crates/tracedecay/tests/daemon_suite/indexing_lifecycle_test.rs @@ -145,27 +145,6 @@ fn write_cancellation_batch(project: &Path, scratch: &Path) { .expect("atomically install cancellation batch"); } -/// A multi-file corpus large enough that replaying every file through -/// `bulk_commit` stays visible to a status poll. The file count is not an -/// expected result; the restart assertion compares against whatever bound the -/// ready index actually published. -fn write_restart_resume_corpus(project: &Path) { - let root = project.join("src/resume_corpus"); - fs::create_dir_all(&root).expect("resume corpus directory"); - for file_index in 0..96_u32 { - let mut source = String::new(); - for symbol_index in 0..8_u32 { - writeln!( - source, - "pub fn resume_anchor_{file_index:04}_{symbol_index:03}(input: u32) -> u32 {{ input + {symbol_index} }}" - ) - .expect("format resume source"); - } - fs::write(root.join(format!("file_{file_index:04}.rs")), source) - .expect("write resume source"); - } -} - async fn wait_for_refreshing_old_generation( socket: &Path, handshake: &DaemonHandshake, @@ -202,43 +181,6 @@ async fn wait_for_refreshing_old_generation( .unwrap_or_else(|_| panic!("refresh never exposed its truthful in-flight receipt: {last}")) } -/// Waits until the refresh admitted after `old_generation` is running. -/// -/// The scheduler reports it in one of two in-flight shapes: still advertising -/// the committed generation (`refreshing`), or, once the successor's text -/// build starts, advertising that successor with its graph pending -/// (`indexing` with the successor's build progress). The first shape lasts -/// only until the successor build publishes its first phase. -async fn wait_for_refresh_in_flight( - socket: &Path, - handshake: &DaemonHandshake, - old_generation: &str, -) -> Value { - let mut last = Value::Null; - tokio::time::timeout(RECEIPT_TIMEOUT, async { - loop { - last = status(socket, handshake).await; - let worktree = &last["code_index_freshness"]["worktree"]; - let in_flight = last["code_index_freshness"]["status"] == "warming" - && worktree["coverage"] == "partial_refresh_in_progress" - && worktree["rebuild_in_flight"] == true; - let refreshing_committed = worktree["latest_generation_id"] == old_generation - && worktree["staleness_state"] == "refreshing"; - let building_successor = worktree["staleness_state"] == "indexing" - && worktree["progress"]["generation_id"] - .as_str() - .is_some_and(|generation| generation != old_generation) - && worktree["progress"]["phase"] != "ready"; - if in_flight && (refreshing_committed || building_successor) { - return last.clone(); - } - tokio::time::sleep(std::time::Duration::from_millis(25)).await; - } - }) - .await - .unwrap_or_else(|_| panic!("refresh never exposed its truthful in-flight receipt: {last}")) -} - fn read_active_generation(home: &Path, project: &Path) -> CodeIndexPublishedGenerationV1 { let layout = tracedecay_runtime_core::storage::resolve_layout(project, &home.join(".tracedecay")) @@ -988,147 +930,3 @@ async fn mounted_incremental_lifecycle_preserves_only_complete_compatible_genera "restarted daemon did not stop cleanly: {exit}" ); } - -#[tokio::test] -async fn restart_after_sigterm_rebuilds_only_changed_files() { - let (environment, project) = IsolatedHome::new(); - let project = canonical_existing_identity(&project).expect("canonical fixture project"); - initialize_repository(&project); - write_restart_resume_corpus(&project); - let revision = commit_all(&project, "resume corpus"); - let socket = daemon_socket_path(environment.home()); - let log_path = environment.scratch().join("restart-resume-daemon.log"); - let mut daemon = spawn_tracedecay_daemon_logged(environment.home(), &log_path, |_| {}); - let project_id = initialize_tracedecay(environment.home(), &project); - let identity = exact_identity(&project, project_id); - tracedecay_project::product_runtime::register_fixture_product_runtime(); - let handshake = tracedecay::daemon::handshake_for_current_client( - environment.profile(), - Some(project.clone()), - None, - false, - false, - ) - .expect("production daemon handshake"); - - let indexed = wait_for_terminal_generation( - &socket, - &handshake, - &project, - &identity, - "refs/heads/main", - Some(&revision), - None, - "resume_anchor_0001_000", - Some("src/resume_corpus/file_0001.rs"), - ) - .await; - let retained_files = - indexed.status["code_index_freshness"]["worktree"]["progress"]["total_files"] - .as_u64() - .expect("a ready index publishes the file bound it built"); - assert!( - retained_files > 1, - "restart replay is only observable when the retained corpus has more than one file: {}", - indexed.status - ); - - let edited = project.join("src/resume_corpus/file_0000.rs"); - let mut source = fs::read_to_string(&edited).expect("read corpus file"); - source.push_str("pub fn restart_resume_probe() -> u32 { 7 }\n"); - fs::write(&edited, source).expect("edit one corpus file"); - deliver_save( - environment.profile(), - &project, - &["src/resume_corpus/file_0000.rs"], - ) - .await; - let _refreshing = wait_for_refresh_in_flight(&socket, &handshake, &indexed.generation_id).await; - - let signal_result = unsafe { libc::kill(daemon.id() as libc::pid_t, libc::SIGTERM) }; - assert_eq!(signal_result, 0, "send graceful cancellation to daemon"); - let exit = daemon - .wait_for_exit(RECEIPT_TIMEOUT) - .expect("wait for cancelled daemon") - .expect("daemon must exit after SIGTERM"); - assert!( - exit.success(), - "daemon cancellation was not graceful: {exit}" - ); - - daemon = spawn_tracedecay_daemon_logged(environment.home(), &log_path, |_| {}); - let mut last = Value::Null; - let mut replayed_retained_corpus = false; - let mut successor_text_files = std::collections::BTreeSet::new(); - tokio::time::timeout(RECEIPT_TIMEOUT, async { - loop { - last = status(&socket, &handshake).await; - let progress = &last["code_index_freshness"]["worktree"]["progress"]; - let same_corpus = progress["total_files"].as_u64() == Some(retained_files); - let unfinished = progress["completed_files"] - .as_u64() - .is_some_and(|completed| completed < retained_files); - let successor = progress["generation_id"] - .as_str() - .is_some_and(|generation| generation != indexed.generation_id); - if progress["phase"] == "bulk_commit" && successor && same_corpus && unfinished { - replayed_retained_corpus = true; - } - // A source scan reports a zero bound before the text build knows - // its files. - if successor - && let Some(total) = progress["total_files"].as_u64().filter(|total| *total > 0) - { - successor_text_files.insert(total); - } - let worktree = &last["code_index_freshness"]["worktree"]; - if last["code_index_freshness"]["status"] == "current" - && worktree["latest_generation_id"] - .as_str() - .is_some_and(|generation| generation != indexed.generation_id) - { - return; - } - tokio::time::sleep(std::time::Duration::from_millis(5)).await; - } - }) - .await - .unwrap_or_else(|_| { - panic!( - "restart never became current on a new generation: {last}; daemon_log={}", - daemon_log_for_failure() - ) - }); - assert!( - !replayed_retained_corpus, - "restart bulk-committed the retained corpus instead of the changed file: {last}" - ); - assert_eq!( - successor_text_files, - std::collections::BTreeSet::from([1]), - "every progress report of the restarted text build must name only the edited file \ - (the retained corpus has {retained_files}): {last}" - ); - - let changed = search(&socket, &handshake, "restart_resume_probe").await; - assert_eq!( - changed["code_generation"].as_str(), - last["code_index_freshness"]["worktree"]["latest_generation_id"].as_str(), - "edited symbol was not served from the restarted generation: {changed}" - ); - assert!( - result_paths(&changed).contains(&"src/resume_corpus/file_0000.rs"), - "edited symbol was not found in the changed file: {changed}" - ); - let unchanged = search(&socket, &handshake, "resume_anchor_0001_000").await; - assert!( - result_paths(&unchanged).contains(&"src/resume_corpus/file_0001.rs"), - "unchanged file stopped being searchable after restart: {unchanged}" - ); - let phrase = search(&socket, &handshake, "fn resume_anchor_0001_000").await; - assert!( - result_paths(&phrase).contains(&"src/resume_corpus/file_0001.rs"), - "unchanged file dropped out of phrase search after restart: {phrase}" - ); - stop_daemon_gracefully(&mut daemon); -}