diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs index ac239ce49c..e5ad481701 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs @@ -38,6 +38,45 @@ use super::{ publication_authority_is_terminal, retained_noop_requires_follow_up_wake, }; +/// Runtime that polls code-index workers, separate from the daemon's serving +/// runtime. +/// +/// The first reconcile of a cold checkout is the strict-readiness critical +/// path. Sharing the daemon runtime left that worker unpolled for minutes +/// while project-open tasks occupied every serving thread: the pass log then +/// showed a 226s queue delay with the gates themselves taking 15µs, and the +/// dogfood deadline expired still inside text projection, graph never seated. +/// A process-wide runtime keeps indexing scheduled as soon as the wake is +/// posted. The future is already boxed, so the stack only has to poll it. +/// +/// Tests do not use this. The suite mounts many registries in one process; +/// a few shared threads then hold a waiter behind every other mount until +/// the per-test deadline. Each test already has a runtime that is not the +/// daemon's serving runtime. +#[cfg(not(test))] +fn code_index_worker_runtime() -> Result<&'static tokio::runtime::Runtime, CodeIndexSchedulerErrorV1> +{ + static RUNTIME: OnceLock> = OnceLock::new(); + match RUNTIME.get_or_init(|| { + let workers = std::thread::available_parallelism() + .map_or(2, usize::from) + .clamp(2, 4); + tokio::runtime::Builder::new_multi_thread() + .thread_name("td-code-index") + .worker_threads(workers) + .max_blocking_threads(workers.saturating_add(8)) + .thread_stack_size(8 * 1024 * 1024) + .enable_all() + .build() + .map_err(|error| error.to_string()) + }) { + Ok(runtime) => Ok(runtime), + Err(error) => Err(CodeIndexSchedulerErrorV1::Identity(format!( + "code-index worker runtime failed to start: {error}" + ))), + } +} + impl CodeIndexSchedulerRegistryV1 { #[cfg(test)] pub fn open_worktree( @@ -2332,10 +2371,16 @@ impl CodeIndexSchedulerRegistryV1 { let _ = result; } }); + #[cfg(test)] let task = tokio::spawn(hotpath::future!( worker_loop, label = "daemon.code_index.scheduler_worker" )); + #[cfg(not(test))] + let task = code_index_worker_runtime()?.spawn(hotpath::future!( + worker_loop, + label = "daemon.code_index.scheduler_worker" + )); entry.insert(MountedCodeIndexWorktreeV1 { project_id, repository_id, 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 394cc15419..ddec371375 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 @@ -2645,10 +2645,17 @@ impl LatestCodeTextGenerationV1 { control: &dyn CodeIndexExecutionControlV1, ) -> Result, RetrievalPortError> { let generation_id = &self.metadata.manifest().generation_id; + // The successor's working set stays at + // `CLONE_SUCCESSOR_MEMORY_BUDGET_BYTES_V1`. The charge is at least + // the reader budget so publication can transfer it onto the reader + // component. A fresh reader admission at that boundary is refused + // when graph replay is already on the RSS watermark. + let publication_charge = CLONE_SUCCESSOR_MEMORY_BUDGET_BYTES_V1 + .max(CODE_LEXICAL_ARTIFACT_QUERY_CACHE_BUDGET_BYTES_V1); let reservation = self.text_artifact_store.reserve_resident_memory( generation_id, "code-text-clone-successor", - CLONE_SUCCESSOR_MEMORY_BUDGET_BYTES_V1, + publication_charge, )?; let artifacts_root = code_text_artifacts_root(self.text_artifact_store.store_root()); ensure_private_text_artifacts_root(&artifacts_root)?; @@ -3292,15 +3299,11 @@ impl LatestCodeTextGenerationV1 { &sealed_identity, control, )?; - // The builder and source are gone, so its transient reservation no - // longer owns bytes. Release it before sampling the reader admission; - // the reader guard then carries the still-live unmodeled baseline. - drop(build_reservation); - let reader_reservation = store.reserve_resident_memory( - &self.metadata.manifest().generation_id, - "code-text-artifact-reader", - CODE_LEXICAL_ARTIFACT_QUERY_CACHE_BUDGET_BYTES_V1, - )?; + // The reader is a smaller charge of the bytes this build already + // holds. Dropping the build reservation and reserving the reader + // again is a gap: an overlapping graph replay can sit on the process + // RSS watermark, and the new admission is then refused forever. + let reader_reservation = reader_charge_from_held_reservation(build_reservation)?; let final_path = code_text_artifact_path(store.store_root(), &descriptor) .map_err(text_artifact_unavailable)?; let reader = CodeLexicalArtifactReaderV1::open_content_addressed( @@ -3405,7 +3408,11 @@ impl LatestCodeTextGenerationV1 { .finish(source_receipt, control) .map_err(map_text_artifact_error)?; drop(build.builder.take()); - drop(build.build_reservation.take()); + let held_reservation = build.build_reservation.take().ok_or_else(|| { + RetrievalPortError::Contract( + "clone-successor resident-memory charge disappeared before publication".to_owned(), + ) + })?; let descriptor = self.text_artifact_store.publish_with_prior( &build.staging_path, &self.metadata.manifest().generation_id, @@ -3413,11 +3420,7 @@ impl LatestCodeTextGenerationV1 { Some(&build.prior_descriptor), control, )?; - let reader_reservation = self.text_artifact_store.reserve_resident_memory( - &self.metadata.manifest().generation_id, - "code-text-artifact-reader", - CODE_LEXICAL_ARTIFACT_QUERY_CACHE_BUDGET_BYTES_V1, - )?; + let reader_reservation = reader_charge_from_held_reservation(held_reservation)?; let final_path = code_text_artifact_path(self.text_artifact_store.store_root(), &descriptor) .map_err(text_artifact_unavailable)?; @@ -3553,7 +3556,37 @@ impl LatestCodeTextGenerationV1 { build.builder = Some(builder); Ok(()) } +} + +/// Name a held publication charge as the reader and shrink it to the reader +/// budget. The bytes never leave the ledger, so the handoff is not a new +/// admission and measured RSS cannot refuse a charge that was already held. +fn reader_charge_from_held_reservation( + mut held: ResidentMemoryReservationV1, +) -> Result { + let reader_budget = + u64::try_from(CODE_LEXICAL_ARTIFACT_QUERY_CACHE_BUDGET_BYTES_V1).map_err(|_| { + RetrievalPortError::Contract("text-artifact reader budget exceeds u64".to_owned()) + })?; + if held.reserved_bytes() < reader_budget { + return Err(RetrievalPortError::Contract(format!( + "text-artifact publication charge {} is below the reader budget {reader_budget}", + held.reserved_bytes() + ))); + } + let component = ResidentMemoryComponentIdV1::new("code-text-artifact-reader") + .map_err(|error| RetrievalPortError::Contract(error.to_string()))?; + held.transfer_component(component, reader_budget) + .map_err(|error| { + RetrievalPortError::Contract(format!( + "text-artifact reader charge could not take over its held reservation: reserved {} measured {}", + error.reserved_bytes, error.measured_bytes + )) + })?; + Ok(held) +} +impl LatestCodeTextGenerationV1 { fn install_artifact_owners( &self, reader: CodeLexicalArtifactReaderV1, diff --git a/crates/tracedecay-runtime-core/src/resident_memory.rs b/crates/tracedecay-runtime-core/src/resident_memory.rs index f1669a488a..2a6e61292d 100644 --- a/crates/tracedecay-runtime-core/src/resident_memory.rs +++ b/crates/tracedecay-runtime-core/src/resident_memory.rs @@ -1134,6 +1134,57 @@ impl ProcessResidentMemoryV1 { Ok(()) } + /// Move one reservation's contribution onto `to_component` and keep only + /// `measured_bytes`, under the same lock. + /// + /// [`Self::reserve`] re-checks measured RSS. Dropping a charge and + /// reserving again is a gap: an overlapping consumer can sit on the + /// watermark and the new admission is refused even though these bytes + /// were already held. The ledger move does not ask for a new admission. + fn transfer_component( + &self, + from: &ResidentMemoryKeyV1, + to_component: ResidentMemoryComponentIdV1, + reserved_bytes: u64, + measured_bytes: u64, + ) -> Result<(), ResidentMemoryAdjustmentFailureV1> { + if measured_bytes > reserved_bytes { + return Err(ResidentMemoryAdjustmentFailureV1 { + reserved_bytes, + measured_bytes, + }); + } + let mut state = self.lock_state(); + let remove_source = { + let Some(charge) = state.charges.get_mut(from) else { + return Err(ResidentMemoryAdjustmentFailureV1 { + reserved_bytes, + measured_bytes, + }); + }; + if *charge < reserved_bytes { + return Err(ResidentMemoryAdjustmentFailureV1 { + reserved_bytes: *charge, + measured_bytes, + }); + } + *charge -= reserved_bytes; + *charge == 0 + }; + if remove_source { + state.charges.remove(from); + } + let released_bytes = reserved_bytes - measured_bytes; + state.used_bytes -= released_bytes; + hotpath::gauge!("runtime_core.resident.used_bytes").set(state.used_bytes as f64); + if measured_bytes > 0 { + let mut to = from.clone(); + to.component = to_component; + *state.charges.entry(to).or_default() += measured_bytes; + } + Ok(()) + } + fn release(&self, key: &ResidentMemoryKeyV1, reserved_bytes: u64) { if reserved_bytes == 0 { return; @@ -1230,6 +1281,29 @@ impl ResidentMemoryReservationV1 { self.reserved_bytes = measured_bytes; Ok(()) } + + /// Keep this charge and name it `component`, shrinking to `measured_bytes` + /// when the held amount is larger. + /// + /// The bytes stay in the ledger for the whole move. Callers that instead + /// drop this reservation and [`ProcessResidentMemoryV1::reserve`] the + /// destination open a gap where measured RSS can refuse a charge that + /// was already admitted. + pub fn transfer_component( + &mut self, + component: ResidentMemoryComponentIdV1, + measured_bytes: u64, + ) -> Result<(), ResidentMemoryAdjustmentFailureV1> { + self.authority.transfer_component( + &self.key, + component, + self.reserved_bytes, + measured_bytes, + )?; + self.key.component = component; + self.reserved_bytes = measured_bytes; + Ok(()) + } } impl Drop for ResidentMemoryReservationV1 { diff --git a/crates/tracedecay-runtime-core/src/resident_memory/tests.rs b/crates/tracedecay-runtime-core/src/resident_memory/tests.rs index 53bc42ca60..97a078593a 100644 --- a/crates/tracedecay-runtime-core/src/resident_memory/tests.rs +++ b/crates/tracedecay-runtime-core/src/resident_memory/tests.rs @@ -350,6 +350,57 @@ fn reservation_tracks_exact_identity_and_releases_on_drop() { assert_eq!(authority.snapshot().used_bytes, 0); } +#[test] +fn transfer_keeps_retained_bytes_when_a_new_admission_cannot() { + let authority = Arc::new(ProcessResidentMemoryV1::new(bytes(100))); + let build = key( + "project-a", + "worktree-a", + "generation-a", + "code-text-artifact-build", + ); + let reader = key( + "project-a", + "worktree-a", + "generation-a", + "code-text-artifact-reader", + ); + let mut held = authority + .reserve(build.clone(), bytes(80)) + .expect("build reservation"); + let _neighbor = authority + .reserve( + key("project-a", "worktree-a", "generation-a", "graph"), + bytes(20), + ) + .expect("neighbor fills the ceiling"); + let denied = authority + .reserve(reader.clone(), bytes(30)) + .expect_err("a fresh reader admission does not fit beside the held build charge"); + assert!(matches!( + denied, + ResidentMemoryAdmissionFailureV1::ReservationCeiling { .. } + )); + + held.transfer_component(reader.component, 30) + .expect("the held charge moves without a new admission"); + assert_eq!(held.key(), &reader); + assert_eq!(held.reserved_bytes(), 30); + let snapshot = authority.snapshot(); + assert_eq!(snapshot.used_bytes, 50); + assert_eq!(snapshot.charge_for(&build), 0); + assert_eq!(snapshot.charge_for(&reader), 30); + + let grown = held.transfer_component(reader.component, 40); + assert!(grown.is_err(), "a transfer cannot grow the held charge"); + assert_eq!(held.reserved_bytes(), 30); + assert_eq!(authority.snapshot().charge_for(&reader), 30); + + drop(held); + assert_eq!(authority.snapshot().used_bytes, 20); + assert_eq!(authority.snapshot().charge_for(&reader), 0); +} + #[test] fn rejection_reports_final_used_requested_and_limit_bytes() { let authority = Arc::new(ProcessResidentMemoryV1::new(bytes(100))); diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs index b5a72dfff6..6c33688210 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs @@ -1026,7 +1026,7 @@ async fn cursor_cas_lost_on_a_partially_covered_window_replays_the_tail() { }) ) .unwrap(); - let len = u64::try_from(std::fs::metadata(&path).unwrap().len()).unwrap(); + let len = std::fs::metadata(&path).unwrap().len(); let spy = SeamSpyAdmission::default(); spy.script_peer_covers_batch_prefix();