From a4318a6da9aaddb5ff633c18afdce6281199e088 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sun, 20 Sep 2026 03:02:36 +0000 Subject: [PATCH] fix(code-index): keep the reader charge on both publication tails Both text-artifact publication tails released the charge they already held and asked the process authority for the reader again. `reserve` re-samples measured RSS and refuses every admission above the 8 MiB pressure floor while the over-budget latch is set, and the reader budget is 256 MiB. Releasing a ledger charge does not lower measured RSS, so the release cannot clear that latch: an overlapping graph replay sitting on the watermark turns a finalized artifact into a generation that can never seat its owners. `ResidentMemoryReservationV1::transfer_component` renames a held charge and shrinks it under one lock, which asks for no admission. Both tails hand their charge to the reader that way. The clone successor now holds at least the reader budget so its tail has something to hand over; its working set is unchanged. Co-Authored-By: Claude Fable 5.1 --- .../src/code_index_scheduler/serving.rs | 64 ++++++++++++---- .../src/code_index_scheduler/tests/serving.rs | 8 +- .../src/resident_memory.rs | 74 +++++++++++++++++++ .../src/resident_memory/tests.rs | 51 +++++++++++++ 4 files changed, 179 insertions(+), 18 deletions(-) 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)));