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 e5f360fbe6..78c802ef5a 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 @@ -989,6 +989,35 @@ fn text_artifact_unavailable(error: impl std::fmt::Display) -> RetrievalPortErro RetrievalPortError::AuthorityUnavailable(error.to_string()) } +/// 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) +} + /// Durable text-artifact store bound to one worktree's generation store root. /// /// Publishes finalized staging artifacts under `code-text-artifacts-v1/` and @@ -2659,10 +2688,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)?; @@ -3331,15 +3367,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( @@ -3449,7 +3481,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, @@ -3457,11 +3493,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)?; diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/serving.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/serving.rs index 21f8baa73e..95b3c64221 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/serving.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/serving.rs @@ -2721,11 +2721,15 @@ fn text_artifact_ceilings_reserve_through_process_resident_memory() { .expect("advance artifact build under an adequate authority") {} let snapshot = adequate.snapshot(); + let reader_budget = u64::try_from(CODE_LEXICAL_ARTIFACT_QUERY_CACHE_BUDGET_BYTES_V1) + .expect("reader budget fits u64"); assert!( snapshot.charges.iter().any(|charge| { - charge.key.component.as_str() == "code-text-artifact-reader" && charge.bytes > 0 + charge.key.component.as_str() == "code-text-artifact-reader" + && charge.bytes == reader_budget }), - "serving artifact owners must hold the measured reader charge: {snapshot:?}" + "serving artifact owners hold exactly the reader budget, not the larger publication \ + charge they take over: {snapshot:?}" ); assert!( !snapshot diff --git a/crates/tracedecay-runtime-core/src/resident_memory.rs b/crates/tracedecay-runtime-core/src/resident_memory.rs index a40c4d5b50..2554a6f231 100644 --- a/crates/tracedecay-runtime-core/src/resident_memory.rs +++ b/crates/tracedecay-runtime-core/src/resident_memory.rs @@ -1217,6 +1217,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; @@ -1313,6 +1364,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 292ad8f2cd..a8548a32fb 100644 --- a/crates/tracedecay-runtime-core/src/resident_memory/tests.rs +++ b/crates/tracedecay-runtime-core/src/resident_memory/tests.rs @@ -477,6 +477,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)));