diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler.rs index 22ea5e6d6e..788aa597d1 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler.rs @@ -252,6 +252,7 @@ use serving::{ use serving::{ CodeIndexCommittedProgressSampleV1, CodeTextProjectionSlotV1, TEXT_ARTIFACT_PAGE_CHUNKS_V1, clone_successor_source_batch_limits_from_charges, map_sealed_page_source_error, - sha256_private_file_and_size, text_artifact_builder_budget, - text_artifact_resident_memory_charges, text_artifact_source_batch_limits, + sha256_private_file_and_size, text_artifact_admitted_build_budget, + text_artifact_builder_budget, text_artifact_resident_memory_charges, + text_artifact_source_batch_limits, }; 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 5d7ae15289..394cc15419 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 @@ -1035,6 +1035,34 @@ pub(super) fn text_artifact_resident_memory_charges( Ok((accounted, retained)) } +pub(super) fn text_artifact_admitted_build_budget( + preferred_bytes: u64, + minimum_bytes: u64, + limit_bytes: u64, + used_bytes: u64, + observed_bytes: u64, + watermark_headroom: u64, +) -> Result { + if minimum_bytes == 0 || preferred_bytes < minimum_bytes { + return Err(RetrievalPortError::Contract( + "text-artifact build budget bounds are invalid".to_owned(), + )); + } + let unmodeled_live_bytes = observed_bytes.saturating_sub(used_bytes); + let available_for_growth = limit_bytes + .saturating_sub(used_bytes) + .saturating_sub(unmodeled_live_bytes) + .saturating_sub(watermark_headroom); + let admitted_bytes = preferred_bytes.min(available_for_growth); + if admitted_bytes < minimum_bytes { + return Err(RetrievalPortError::AuthorityUnavailable(format!( + "text-artifact build needs at least {minimum_bytes} bytes; \ + {available_for_growth} bytes are available below the resident-memory watermark" + ))); + } + Ok(admitted_bytes) +} + impl DaemonCodeTextArtifactStoreV1 { pub(super) fn bind( store_root: &Path, @@ -1067,9 +1095,20 @@ impl DaemonCodeTextArtifactStoreV1 { component: &'static str, bytes: usize, ) -> Result { + self.reserve_resident_memory_up_to(generation_id, component, bytes, bytes) + .map(|(reservation, _)| reservation) + } + + fn reserve_resident_memory_up_to( + &self, + generation_id: &CodeGenerationId, + component: &'static str, + preferred_bytes: usize, + minimum_bytes: usize, + ) -> Result<(ResidentMemoryReservationV1, usize), RetrievalPortError> { let component = ResidentMemoryComponentIdV1::new(component) .map_err(|error| RetrievalPortError::Contract(error.to_string()))?; - let requested = u64::try_from(bytes) + let preferred = u64::try_from(preferred_bytes) .ok() .and_then(std::num::NonZeroU64::new) .ok_or_else(|| { @@ -1077,6 +1116,14 @@ impl DaemonCodeTextArtifactStoreV1 { "text-artifact resident-memory reservation must be nonzero".to_owned(), ) })?; + let minimum = u64::try_from(minimum_bytes) + .ok() + .and_then(std::num::NonZeroU64::new) + .ok_or_else(|| { + RetrievalPortError::Contract( + "text-artifact minimum resident-memory reservation must be nonzero".to_owned(), + ) + })?; let snapshot = self.resident_memory.snapshot(); let observed_bytes = sampled_process_resident_bytes_v1().map_or(0, |observed| { self.resident_memory @@ -1092,8 +1139,21 @@ impl DaemonCodeTextArtifactStoreV1 { .high_watermark_bytes() .min(snapshot.limit_bytes); let watermark_headroom = snapshot.limit_bytes.saturating_sub(admission_watermark); + let admitted_bytes = text_artifact_admitted_build_budget( + preferred.get(), + minimum.get(), + snapshot.limit_bytes, + snapshot.used_bytes, + observed_bytes, + watermark_headroom, + )?; + let admitted = NonZeroU64::new(admitted_bytes).ok_or_else(|| { + RetrievalPortError::Contract( + "text-artifact admitted resident-memory reservation must be nonzero".to_owned(), + ) + })?; let (accounted, retained) = text_artifact_resident_memory_charges( - requested, + admitted, unmodeled_live_bytes, watermark_headroom, )?; @@ -1102,7 +1162,9 @@ impl DaemonCodeTextArtifactStoreV1 { hotpath::gauge!("query.artifact.admission.unmodeled_live_bytes") .set(unmodeled_live_bytes as f64); hotpath::gauge!("query.artifact.admission.requested_growth_bytes") - .set(requested.get() as f64); + .set(preferred.get() as f64); + hotpath::gauge!("query.artifact.admission.admitted_growth_bytes") + .set(admitted.get() as f64); hotpath::gauge!("query.artifact.admission.accounted_bytes").set(accounted.get() as f64); hotpath::gauge!("query.artifact.admission.retained_bytes").set(retained.get() as f64); let mut reservation = self @@ -1116,13 +1178,22 @@ impl DaemonCodeTextArtifactStoreV1 { }, accounted, ) - .map_err(|_| RetrievalPortError::BudgetExceeded)?; + .map_err(|error| { + RetrievalPortError::AuthorityUnavailable(format!( + "text-artifact resident-memory admission was refused: {error}" + )) + })?; reservation.shrink_to(retained.get()).map_err(|error| { RetrievalPortError::Contract(format!( "text-artifact resident-memory headroom release failed: {error}" )) })?; - Ok(reservation) + let admitted = usize::try_from(admitted.get()).map_err(|error| { + RetrievalPortError::Contract(format!( + "text-artifact admitted reservation exceeds the platform limit: {error}" + )) + })?; + Ok((reservation, admitted)) } fn acquire_store_write_lock(&self) -> Result { @@ -2723,14 +2794,9 @@ impl LatestCodeTextGenerationV1 { control: &dyn CodeIndexExecutionControlV1, ) -> Result { let store = &self.text_artifact_store; - let build_memory_budget = code_lexical_artifact_build_memory_budget_for( + let preferred_build_memory_budget = code_lexical_artifact_build_memory_budget_for( store.resident_memory.snapshot().limit_bytes, ); - let (source_batch_pages, source_batch_bytes, _) = - text_artifact_source_batch_limits(build_memory_budget); - hotpath::gauge!("query.artifact.build_memory_budget_bytes").set(build_memory_budget); - hotpath::gauge!("query.artifact.source_batch_pages_max").set(source_batch_pages); - hotpath::gauge!("query.artifact.source_batch_bytes_max").set(source_batch_bytes); let generation_id = self.metadata.manifest().generation_id.clone(); if let Some(descriptor) = store.published_descriptor(&generation_id)? && let Some(outcome) = @@ -2739,12 +2805,24 @@ impl LatestCodeTextGenerationV1 { return Ok(outcome); } // The builder's advertised memory ceiling is reserved through the - // process resident-memory authority before the build allocates. - let build_reservation = store.reserve_resident_memory( + // process resident-memory authority before the build allocates. The + // host-scaled figure is a preferred ceiling, not a minimum: under a + // large stale serving graph, admit any supported budget down to the + // builder's established 1.5 GiB floor so the replacement can finish + // and release that graph. + let (build_reservation, build_memory_budget) = store.reserve_resident_memory_up_to( &generation_id, "code-text-artifact-build", - build_memory_budget, + preferred_build_memory_budget, + CODE_LEXICAL_ARTIFACT_BUILD_MEMORY_BUDGET_BYTES_V1, )?; + let (source_batch_pages, source_batch_bytes, _) = + text_artifact_source_batch_limits(build_memory_budget); + hotpath::gauge!("query.artifact.preferred_build_memory_budget_bytes") + .set(preferred_build_memory_budget); + hotpath::gauge!("query.artifact.build_memory_budget_bytes").set(build_memory_budget); + hotpath::gauge!("query.artifact.source_batch_pages_max").set(source_batch_pages); + hotpath::gauge!("query.artifact.source_batch_bytes_max").set(source_batch_bytes); let sealed_identity = store.sealed_identity(&generation_id)?; let sealed_hex = sha256_hex_suffix(sealed_identity.digest.as_str()).ok_or_else(|| { RetrievalPortError::Contract( 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 82bc63a013..dd5516dc71 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 @@ -2334,9 +2334,11 @@ fn reader_reservation_refusal_precedes_missing_artifact_access() { std::num::NonZeroU64::new(1024 * 1024).expect("tight memory limit"), ))); let latest = scheduler.latest_complete().expect("restored generation"); - assert_eq!( - latest.advance_text_serving(1), - Err(tracedecay_query::retrieval::RetrievalPortError::BudgetExceeded), + assert!( + matches!( + latest.advance_text_serving(1), + Err(tracedecay_query::retrieval::RetrievalPortError::AuthorityUnavailable(_)) + ), "the reservation gate must win before the missing path is inspected" ); assert!( @@ -2390,6 +2392,48 @@ fn overlapping_text_builds_share_one_admission_watermark_headroom() { } } +#[test] +fn text_build_budget_shrinks_to_available_headroom_without_dropping_below_its_floor() { + const GIB: u64 = 1024 * 1024 * 1024; + const MIB: u64 = 1024 * 1024; + let limit = 26 * GIB; + let preferred = limit / 8; + let minimum = 1536 * MIB; + let watermark_headroom = limit - (limit * 900 / 1000); + let observed = 21 * GIB; + let available = limit - observed - watermark_headroom; + + assert_eq!( + super::super::text_artifact_admitted_build_budget( + preferred, + minimum, + limit, + 0, + observed, + watermark_headroom, + ), + Ok(available), + "a replacement build must use the supported smaller budget instead of deadlocking behind the stale graph" + ); + assert_eq!( + super::super::text_artifact_admitted_build_budget( + preferred, + minimum, + limit, + 0, + 22 * GIB, + watermark_headroom, + ), + Err( + tracedecay_query::retrieval::RetrievalPortError::AuthorityUnavailable(format!( + "text-artifact build needs at least {minimum} bytes; {} bytes are available below the resident-memory watermark", + limit - 22 * GIB - watermark_headroom + )) + ), + "less than the builder's supported floor must remain a typed capacity refusal" + ); +} + /// The artifact build and reader ceilings must reserve through the process /// resident-memory authority: an authority too small for the advertised /// build ceiling refuses the build as a typed unavailability, and a serving @@ -2416,9 +2460,9 @@ fn text_artifact_ceilings_reserve_through_process_resident_memory() { assert!( matches!( denied, - Err(tracedecay_query::retrieval::RetrievalPortError::BudgetExceeded) + Err(tracedecay_query::retrieval::RetrievalPortError::AuthorityUnavailable(_)) ), - "an unreservable build ceiling must refuse as a typed budget state: {denied:?}" + "an unreservable build ceiling must refuse as typed availability: {denied:?}" ); assert_eq!( tight.snapshot().used_bytes, @@ -2447,10 +2491,12 @@ fn text_artifact_ceilings_reserve_through_process_resident_memory() { let latest = scheduler .latest_complete() .expect("measured latest generation"); - assert_eq!( - latest.advance_text_serving(1), - Err(tracedecay_query::retrieval::RetrievalPortError::BudgetExceeded), - "fresh RSS plus the requested build ceiling exceeds the process authority" + assert!( + matches!( + latest.advance_text_serving(1), + Err(tracedecay_query::retrieval::RetrievalPortError::AuthorityUnavailable(_)) + ), + "fresh RSS plus the minimum build ceiling exceeds the process authority" ); assert_eq!( measured.snapshot().used_bytes,