From 7891e015964e71af6f13b79e040a7da2e87c0c71 Mon Sep 17 00:00:00 2001 From: "renovate[bot]" <29139614+renovate[bot]@users.noreply.github.com> Date: Tue, 20 Oct 2026 18:59:17 +0000 Subject: [PATCH 1/4] chore(deps): update dependency react-router to v8.5.0 --- pnpm-lock.yaml | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 2deae74482..58b9051aab 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -219,7 +219,7 @@ importers: version: 19.3.1(react@19.3.1) react-router: specifier: ^8.0.0 - version: 8.4.0(react-dom@19.3.1(react@19.3.1))(react@19.3.1) + version: 8.5.0(@types/react-dom@19.3.0(@types/react@19.3.0))(@types/react@19.3.0)(react-dom@19.3.1(react@19.3.1))(react@19.3.1) zod: specifier: ^4.0.0 version: 4.6.7 @@ -2469,13 +2469,19 @@ packages: '@types/react': optional: true - react-router@8.4.0: - resolution: {integrity: sha512-JtydAkBvU9UTEe5qNC+POhu+ZBNM7rlKY2IegwXc7Idj7xfRG16x5RZrw3mju+XlqBxfvb2ky9P3jUp9WyWC1g==} + react-router@8.5.0: + resolution: {integrity: sha512-9nEd/deifcTJIIcUaNIyQGkyXD0D6QVB9XD2O5GgHpP3xu3U5RXm87YQ8++Wn8JQWgMBM/5+PjWPM/7n7O7WHA==} engines: {node: '>=22.22.0'} peerDependencies: + '@types/react': '>=19.2.0' + '@types/react-dom': '>=19.2.0' react: '>=19.2.7' react-dom: '>=19.2.7' peerDependenciesMeta: + '@types/react': + optional: true + '@types/react-dom': + optional: true react-dom: optional: true @@ -4715,12 +4721,14 @@ snapshots: optionalDependencies: '@types/react': 19.3.0 - react-router@8.4.0(react-dom@19.3.1(react@19.3.1))(react@19.3.1): + react-router@8.5.0(@types/react-dom@19.3.0(@types/react@19.3.0))(@types/react@19.3.0)(react-dom@19.3.1(react@19.3.1))(react@19.3.1): dependencies: '@remix-run/route-pattern': 0.22.1 cookie-es: 3.1.1 react: 19.3.1 optionalDependencies: + '@types/react': 19.3.0 + '@types/react-dom': 19.3.0(@types/react@19.3.0) react-dom: 19.3.1(react@19.3.1) react-style-singleton@2.2.3(@types/react@19.3.0)(react@19.3.1): From f8db82ea71d35b41a23bb54167b681cf44be2801 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Wed, 21 Oct 2026 01:16:36 -0700 Subject: [PATCH 2/4] fix(memory): repair text admission and blocked progress --- .../src/code_index_scheduler.rs | 3 +- .../src/code_index_scheduler/serving.rs | 171 +++++++----------- .../tests/code_index_scheduler/serving.rs | 118 ++++++++---- .../src/resident_memory.rs | 66 ++++++- .../src/resident_memory/tests.rs | 127 +++++++++++++ .../tests/rmcp_route/host_index_handoff.rs | 2 +- 6 files changed, 344 insertions(+), 143 deletions(-) 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 baca65a33f..a325e59813 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler.rs @@ -300,8 +300,7 @@ pub use serving::{ use serving::{ CodeIndexCommittedProgressSampleV1, CodeTextProjectionSlotV1, map_sealed_page_source_error, sha256_private_file_and_size, text_artifact_admitted_build_budget, - text_artifact_builder_budget, text_artifact_resident_memory_charge, - text_artifact_source_batch_limits, + text_artifact_builder_budget, text_artifact_source_batch_limits, }; #[cfg(test)] use tracedecay_query::retrieval::ports::TEXT_ARTIFACT_PAGE_CHUNKS_V1; 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 de14bd949d..68501f16e5 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 @@ -5,7 +5,6 @@ mod family_report; use std::{ collections::{BTreeSet, VecDeque}, io::Read, - num::NonZeroU64, path::{Path, PathBuf}, sync::{ Arc, Condvar, Mutex, MutexGuard, OnceLock, PoisonError, RwLock, @@ -947,29 +946,6 @@ pub struct DaemonCodeTextArtifactStoreV1 { worktree_id: WorktreeId, } -/// The bytes one artifact admission reserves: its ceiling plus the band above -/// the admission watermark, so the reserve call enforces that watermark -/// against both the ledger and the measured resident set. The band is not -/// memory the artifact owns and is released right after the reserve; -/// retaining it in every overlapping build would charge the same -/// process-wide margin repeatedly. The process's live set is not charged: -/// the reserve already compares the request with the measured resident set, -/// which contains it. -pub(super) fn text_artifact_resident_memory_charge( - requested: NonZeroU64, - watermark_headroom: u64, -) -> Result { - requested - .get() - .checked_add(watermark_headroom) - .and_then(NonZeroU64::new) - .ok_or_else(|| { - RetrievalPortError::Contract( - "text-artifact resident-memory accounting overflowed".to_owned(), - ) - }) -} - pub(super) fn text_artifact_admitted_build_budget( preferred_bytes: u64, minimum_bytes: u64, @@ -1170,10 +1146,8 @@ impl DaemonCodeTextArtifactStoreV1 { } } - /// Reserve one artifact memory ceiling. The atomic reserve also includes - /// the process-wide high-watermark headroom, then releases that - /// check-only margin before returning while the component ceiling stays - /// charged. + /// Charge the artifact's ceiling below the process admission watermark. + /// Sizing and charging share one measured headroom reading. fn reserve_resident_memory( &self, generation_id: &CodeGenerationId, @@ -1184,32 +1158,6 @@ impl DaemonCodeTextArtifactStoreV1 { .map(|(reservation, _)| reservation) } - /// Measure headroom and size the build: `(observed, watermark headroom, - /// admitted)` bytes. - fn text_artifact_admission( - &self, - preferred: NonZeroU64, - minimum: NonZeroU64, - ) -> Result<(u64, u64), RetrievalPortError> { - let admission_watermark = self.resident_memory.admission_watermark_bytes(); - let headroom = self.resident_memory.headroom_below(admission_watermark); - let watermark_headroom = self - .resident_memory - .snapshot() - .limit_bytes - .saturating_sub(admission_watermark); - let admitted_bytes = text_artifact_admitted_build_budget( - preferred.get(), - minimum.get(), - headroom.available_bytes, - ) - .inspect_err(|_| { - self.resident_memory - .wait_for_headroom(minimum.get(), admission_watermark); - })?; - Ok((watermark_headroom, admitted_bytes)) - } - fn reserve_resident_memory_up_to( &self, generation_id: &CodeGenerationId, @@ -1235,57 +1183,59 @@ impl DaemonCodeTextArtifactStoreV1 { "text-artifact minimum resident-memory reservation must be nonzero".to_owned(), ) })?; - let (watermark_headroom, admitted_bytes) = - match self.text_artifact_admission(preferred, minimum) { - Err(RetrievalPortError::ResidentMemoryRefused(detail)) => { - let released = self - .resident_owners - .shed(minimum.get(), std::time::Instant::now()); - for release in &released { - log_resident_owner_release_v1(release); - } - // Freed pages sit in the pool workers' allocator heaps - // whether or not an owner was shed. - let trim = release_process_allocator_memory_v1(); - tracing::info!( - event = "code_text_artifact_build_shed_retained_state", - released_owners = released.len(), - trimmed_bytes = trim.released_bytes(), - refused = %detail, - "text-artifact build shed retained state before re-measuring headroom" - ); - self.text_artifact_admission(preferred, minimum) + let admission_watermark = self.resident_memory.admission_watermark_bytes(); + let reserve = || { + let reservation = self + .resident_memory + .reserve_up_to( + ResidentMemoryKeyV1 { + project_id: self.project_id.clone(), + worktree_id: self.worktree_id.clone(), + generation_id: generation_id.clone(), + component, + }, + preferred, + admission_watermark, + ) + .map_err(|error| { + RetrievalPortError::ResidentMemoryRefused(format!( + "text-artifact resident-memory admission was refused: {error}" + )) + })?; + let admitted = text_artifact_admitted_build_budget( + preferred.get(), + minimum.get(), + reservation.reserved_bytes(), + )?; + Ok((reservation, admitted)) + }; + let (reservation, admitted) = match reserve() { + Err(RetrievalPortError::ResidentMemoryRefused(detail)) => { + let released = self + .resident_owners + .shed(minimum.get(), std::time::Instant::now()); + for release in &released { + log_resident_owner_release_v1(release); } - admission => admission, - }?; - let admitted = NonZeroU64::new(admitted_bytes).ok_or_else(|| { - RetrievalPortError::Contract( - "text-artifact admitted resident-memory reservation must be nonzero".to_owned(), - ) - })?; - let accounted = text_artifact_resident_memory_charge(admitted, watermark_headroom)?; - let mut reservation = self - .resident_memory - .reserve( - ResidentMemoryKeyV1 { - project_id: self.project_id.clone(), - worktree_id: self.worktree_id.clone(), - generation_id: generation_id.clone(), - component, - }, - accounted, - ) - .map_err(|error| { - RetrievalPortError::ResidentMemoryRefused(format!( - "text-artifact resident-memory admission was refused: {error}" - )) - })?; - reservation.shrink_to(admitted.get()).map_err(|error| { - RetrievalPortError::Contract(format!( - "text-artifact resident-memory headroom release failed: {error}" - )) + let trim = release_process_allocator_memory_v1(); + tracing::info!( + event = "code_text_artifact_build_shed_retained_state", + released_owners = released.len(), + trimmed_bytes = trim.released_bytes(), + refused = %detail, + "text-artifact build shed retained state before re-measuring headroom" + ); + reserve() + } + result => result, + } + .inspect_err(|error| { + if matches!(error, RetrievalPortError::ResidentMemoryRefused(_)) { + self.resident_memory + .wait_for_headroom(minimum.get(), admission_watermark); + } })?; - let admitted = usize::try_from(admitted.get()).map_err(|error| { + let admitted = usize::try_from(admitted).map_err(|error| { RetrievalPortError::Contract(format!( "text-artifact admitted reservation exceeds the platform limit: {error}" )) @@ -2811,7 +2761,7 @@ impl LatestCodeTextGenerationV1 { maximum_work: usize, control: &dyn CodeIndexExecutionControlV1, ) -> Result { - let _workers = self + let workers = self .text_artifact_store .worker_runtime .enter(&self.text_artifact_store.resident_memory) @@ -2824,8 +2774,17 @@ impl LatestCodeTextGenerationV1 { ), ) => RetrievalPortError::ResidentMemoryRefused(error.to_string()), other => RetrievalPortError::AuthorityUnavailable(other.to_string()), - })?; - let result = self.advance_text_serving_inner(maximum_work, control); + }); + let result = match workers { + Ok(_workers) => self.advance_text_serving_inner(maximum_work, control), + Err(error) => Err(error), + }; + // A refusal before the builder opens has no batch/finalization error + // to classify. Preserve that refusal on the same generation's progress + // instead of leaving its completed source scan looking unblocked. + if matches!(&result, Err(RetrievalPortError::ResidentMemoryRefused(_))) { + self.publish_text_progress_blocked(CodeIndexBuildBlockedReasonV1::ResidentMemory); + } if result.as_ref().is_err_and(|error| { matches!( error, diff --git a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs index 05bf0e812e..fcb1c9d59a 100644 --- a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs +++ b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs @@ -20,6 +20,7 @@ use tracedecay_contracts::{ PageRequest, PhraseSearchRequest, QualifiedNameRequest, ResolvedScope, ResultProjection, RetrievalOrder, RetrievalPortContext, RetrievalPortOutcome, RetrievalRequestMeta, SourceMetadataRequest, callable_code_operation, + code_index_freshness::{CodeIndexBuildBlockedReasonV1, CodeIndexBuildPhaseV1}, retrieval::{ CodeFacetDimension, CodeFacetRequest, CodeHierarchyRequest, CodeImpactRequest, CodeImplementationsRequest, CodeNavigationRequest, CodeTimelineRequest, @@ -49,8 +50,8 @@ use tracedecay_query::retrieval::{ }; use tracedecay_runtime_core::resident_memory::{ DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, ProcessResidentMemoryV1, ProcessResidentSampleV1, - RESIDENT_MEMORY_PRESSURE_HIGH_WATERMARK_PERMILLE_V1, ResidentMemoryPressureV1, - resident_memory_watermark_bytes_v1, + RESIDENT_MEMORY_PRESSURE_HIGH_WATERMARK_PERMILLE_V1, ResidentMemoryComponentIdV1, + ResidentMemoryPressureV1, resident_memory_watermark_bytes_v1, }; use tracedecay_session_temporal_store::SessionTemporalAccess; @@ -2086,42 +2087,93 @@ fn reader_reservation_refusal_precedes_missing_artifact_access() { } #[test] -fn overlapping_text_builds_share_one_admission_watermark_headroom() { - let limit_bytes = 1_000_u64; - let watermark_headroom = 100_u64; - let requested = NonZeroU64::new(300).expect("nonzero build request"); - let mut used_bytes = 0_u64; - - for _ in 0..3 { - let accounted = - super::super::text_artifact_resident_memory_charge(requested, watermark_headroom) - .expect("bounded admission accounting"); - assert!( - used_bytes + accounted.get() <= limit_bytes, - "each overlapping build fits beneath the same 900-byte high watermark" +fn retained_text_admission_reports_its_block_and_recovers_after_headroom_release() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn reserved_until_release() {}\n")]); + let store = TempDir::new().expect("store root"); + let limit = NonZeroU64::new(4 * 1024 * 1024 * 1024).expect("test process limit"); + let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( + limit, + Arc::new(|| { + Some(ProcessResidentSampleV1 { + resident_bytes: 0, + unreclaimable_bytes: 0, + swapped_bytes: 0, + cgroup_headroom_bytes: None, + }) + }), + )); + let memory = Arc::new(ProcessResidentMemoryV1::with_pressure( + limit, + Arc::clone(&pressure), + )); + { + let mut scheduler = scheduler( + &fixture, + store.path().to_path_buf(), + Arc::new(SharedCodeIndexBytePoolV1::default()), ); - used_bytes += requested.get(); + scheduler.bind_resident_memory(Arc::clone(&memory)); + published(scheduler.reconcile_now().expect("seal source generation")); } - - assert_eq!( - used_bytes, 900, - "the retained ledger owns the three build ceilings, not three margins" + let mut scheduler = scheduler( + &fixture, + store.path().to_path_buf(), + Arc::new(SharedCodeIndexBytePoolV1::default()), ); - for overflow in [ - super::super::text_artifact_resident_memory_charge( - NonZeroU64::new(u64::MAX).expect("maximum nonzero request"), - 1, - ), - super::super::text_artifact_resident_memory_charge(requested, u64::MAX), - ] { - assert!( - matches!( - overflow, - Err(tracedecay_query::retrieval::RetrievalPortError::Contract(_)) - ), - "overflow must remain a typed contract refusal: {overflow:?}" + scheduler.bind_resident_memory(Arc::clone(&memory)); + let text = scheduler + .servable_retained_text_generation() + .expect("publication store") + .expect("retained sealed text generation"); + let scanned = build_progress_snapshot(&scheduler); + assert_eq!(scanned.phase, CodeIndexBuildPhaseV1::SourceScan); + assert_eq!(scanned.blocked_reason, None); + let build_floor = u64::try_from(CODE_LEXICAL_ARTIFACT_BUILD_MEMORY_BUDGET_BYTES_V1) + .expect("build floor fits u64"); + let held_bytes = memory.admission_watermark_bytes() - build_floor + 1; + let competing = memory + .reserve_process_shared( + ResidentMemoryComponentIdV1::new("held-text-headroom").expect("component"), + NonZeroU64::new(held_bytes).expect("competing reservation"), + ) + .expect("hold actual ledger headroom"); + let mut headroom = pressure.subscribe_headroom(); + + for _ in 0..2 { + let refused = text.advance_text_serving(1); + assert!(matches!( + refused, + Err(RetrievalPortError::ResidentMemoryRefused(_)) + )); + let blocked = build_progress_snapshot(&scheduler); + assert_eq!(blocked.generation_id, scanned.generation_id); + assert_eq!(blocked.phase, CodeIndexBuildPhaseV1::SourceScan); + assert_eq!(blocked.committed_pages, 0); + assert_eq!( + blocked.blocked_reason, + Some(CodeIndexBuildBlockedReasonV1::ResidentMemory) ); + assert!(text.text_projection_needs_work()); + assert_eq!(memory.snapshot().used_bytes, held_bytes); } + + drop(headroom.borrow_and_update()); + drop(competing); + assert!( + headroom + .has_changed() + .expect("headroom authority remains live") + ); + assert_eq!(memory.snapshot().used_bytes, 0); + text.production_query_owners() + .expect("same owner builds after headroom is released"); + let ready = build_progress_snapshot(&scheduler); + assert_eq!(ready.generation_id, scanned.generation_id); + assert_eq!(ready.phase, CodeIndexBuildPhaseV1::Ready); + assert!(ready.committed_pages > 0); + assert_eq!(ready.blocked_reason, None); + assert!(text.query_owners_are_ready()); + assert!(active_text_artifact_path(store.path()).is_file()); } #[test] diff --git a/crates/tracedecay-runtime-core/src/resident_memory.rs b/crates/tracedecay-runtime-core/src/resident_memory.rs index 34a9c445be..a16e42ca7d 100644 --- a/crates/tracedecay-runtime-core/src/resident_memory.rs +++ b/crates/tracedecay-runtime-core/src/resident_memory.rs @@ -1531,6 +1531,62 @@ impl ProcessResidentMemoryV1 { Err(self.refused(self.admission_failure(requested_bytes))) } + /// Size and charge new work from the same measured headroom reading. + /// The returned ceiling can be smaller than `preferred_bytes`; consumers + /// with a minimum working set must release a reservation below that floor. + /// Concurrent ledger charges are checked under the charge's own lock. + /// The consumer registers its working-set minimum with `wait_for_headroom` + /// when the returned budget cannot support its next operation. + #[tracing::instrument( + name = "runtime_core.resident.reserve_up_to", + level = "trace", + skip_all + )] + pub fn reserve_up_to( + self: &Arc, + key: ResidentMemoryKeyV1, + preferred_bytes: NonZeroU64, + ceiling: u64, + ) -> Result { + self.pressure.measure_admission_bytes(); + let ceiling = ceiling.min(self.limit_bytes.get()); + let mut state = self.lock_state(); + let headroom = self.headroom_from_state(&state, ceiling); + let admitted = preferred_bytes + .get() + .min(headroom.available_bytes) + .min(ceiling.saturating_sub(state.used_bytes)); + if admitted == 0 { + let failure = if self.pressure.state().is_over_budget() + || headroom.observed_bytes >= ceiling.min(self.pressure.limit_bytes()) + { + ResidentMemoryAdmissionFailureV1::ObservedOverBudget { + observed_bytes: headroom.observed_bytes, + limit_bytes: self.pressure.limit_bytes(), + high_watermark_bytes: self.pressure.high_watermark_bytes(), + requested_bytes: preferred_bytes.get(), + floor_bytes: RESIDENT_MEMORY_PRESSURE_ADMISSION_FLOOR_BYTES_V1, + } + } else { + ResidentMemoryAdmissionFailureV1::ReservationCeiling { + used_bytes: state.used_bytes, + requested_bytes: preferred_bytes.get(), + limit_bytes: ceiling, + } + }; + drop(state); + return Err(failure); + } + // `admitted` is at most ceiling - used_bytes, so both ledger sums fit. + state.used_bytes += admitted; + *state.charges.entry(key.clone()).or_default() += admitted; + Ok(ResidentMemoryReservationV1 { + authority: Arc::clone(self), + key, + reserved_bytes: admitted, + }) + } + /// Reserves one process-shared component without fabricating a project, /// worktree, or code-generation owner. These reservations use the same /// process ceiling and RAII release authority as project generations. @@ -1596,12 +1652,20 @@ impl ProcessResidentMemoryV1 { /// [`Self::reserve`] holds them. pub fn headroom_below(&self, ceiling: u64) -> ResidentMemoryHeadroomV1 { self.pressure.measure_admission_bytes(); + self.headroom_from_state(&self.lock_state(), ceiling) + } + + fn headroom_from_state( + &self, + state: &ResidentMemoryStateV1, + ceiling: u64, + ) -> ResidentMemoryHeadroomV1 { let measured_ceiling = ceiling.min(self.pressure.limit_bytes()); let observed_bytes = self .pressure .observed_bytes_for_limit(measured_ceiling) .unwrap_or(0); - let used_bytes = self.lock_state().used_bytes; + let used_bytes = state.used_bytes; let available_bytes = if self.pressure.state().is_over_budget() { 0 } else { diff --git a/crates/tracedecay-runtime-core/src/resident_memory/tests.rs b/crates/tracedecay-runtime-core/src/resident_memory/tests.rs index 557293b4ec..10fe13f033 100644 --- a/crates/tracedecay-runtime-core/src/resident_memory/tests.rs +++ b/crates/tracedecay-runtime-core/src/resident_memory/tests.rs @@ -1335,6 +1335,35 @@ fn cgroup_headroom_respects_a_larger_but_fuller_ancestor() { ); } +#[test] +fn sized_reservations_charge_only_the_measured_cgroup_headroom() { + const MIB: u64 = 1024 * 1024; + let limit = bytes(800 * MIB); + let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( + limit, + Arc::new(|| { + Some(ProcessResidentSampleV1 { + resident_bytes: 100 * MIB, + unreclaimable_bytes: 100 * MIB, + swapped_bytes: 0, + cgroup_headroom_bytes: Some(200 * MIB), + }) + }), + )); + let authority = Arc::new(ProcessResidentMemoryV1::with_pressure(limit, pressure)); + let watermark = authority.admission_watermark_bytes(); + let headroom = authority.headroom_below(watermark); + assert_eq!(headroom.available_bytes, 200 * MIB); + let owner = key("project-a", "worktree-a", "generation-a", "text-build"); + let held = authority + .reserve_up_to(owner, bytes(headroom.available_bytes + 1), watermark) + .expect("the actual allocation fits without charging the 80 MiB watermark band"); + assert_eq!(held.reserved_bytes(), 200 * MIB); + assert_eq!(authority.snapshot().used_bytes, 200 * MIB); + drop(held); + assert_eq!(authority.snapshot().used_bytes, 0); +} + #[test] fn narrowed_budgets_and_waiters_share_cgroup_capacity_without_charging_peers() { const MIB: u64 = 1024 * 1024; @@ -1619,3 +1648,101 @@ fn allocator_reclaim_preserves_cgroup_pressure_until_sample_recovers() { .is_ok() ); } + +#[test] +fn sized_admission_remeasures_and_charges_the_supported_smaller_budget() { + const MIB: u64 = 1024 * 1024; + let measured = Arc::new(AtomicU64::new(300 * MIB)); + let sampled = Arc::clone(&measured); + let reads = Arc::new(AtomicU64::new(0)); + let sampled_reads = Arc::clone(&reads); + let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( + bytes(1000 * MIB), + Arc::new(move || { + sampled_reads.fetch_add(1, Ordering::SeqCst); + let observed = sampled.load(Ordering::SeqCst); + Some(ProcessResidentSampleV1 { + resident_bytes: observed, + unreclaimable_bytes: observed, + swapped_bytes: 0, + cgroup_headroom_bytes: None, + }) + }), + )); + let authority = Arc::new(ProcessResidentMemoryV1::with_pressure( + bytes(1000 * MIB), + Arc::clone(&pressure), + )); + let watermark = authority.admission_watermark_bytes(); + assert_eq!(watermark, 900 * MIB); + assert_eq!( + authority.headroom_below(watermark).available_bytes, + 600 * MIB + ); + measured.store(700 * MIB, Ordering::SeqCst); + let before = reads.load(Ordering::SeqCst); + let reservation = authority + .reserve_up_to( + key("project-a", "worktree-a", "generation-a", "text-build"), + bytes(600 * MIB), + watermark, + ) + .expect("the smaller build fits despite growth since planning"); + assert_eq!(reads.load(Ordering::SeqCst), before + 1); + assert_eq!(reservation.reserved_bytes(), 200 * MIB); + assert_eq!(authority.snapshot().used_bytes, 200 * MIB); + drop(reservation); + assert_eq!(authority.snapshot().used_bytes, 0); + + measured.store(950 * MIB, Ordering::SeqCst); + let refused = authority + .reserve_up_to( + key("project-a", "worktree-a", "generation-b", "text-build"), + bytes(600 * MIB), + watermark, + ) + .expect_err("sizing must still refuse measured pressure"); + assert!(refused.is_observed_over_budget()); + assert_eq!(authority.snapshot().used_bytes, 0); +} + +#[test] +fn sized_admissions_share_the_watermark_without_charging_its_margin() { + let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( + bytes(1000), + Arc::new(|| None), + )); + let authority = Arc::new(ProcessResidentMemoryV1::with_pressure( + bytes(1000), + pressure, + )); + let watermark = authority.admission_watermark_bytes(); + let owner = key("project-a", "worktree-a", "generation-a", "text-build"); + let mut reservations = (0..3) + .map(|_| { + authority + .reserve_up_to(owner.clone(), bytes(300), watermark) + .unwrap() + }) + .collect::>(); + assert_eq!(authority.snapshot().used_bytes, 900); + let refused = authority + .reserve_up_to(owner.clone(), bytes(300), watermark) + .unwrap_err(); + assert_eq!( + refused, + ResidentMemoryAdmissionFailureV1::ReservationCeiling { + used_bytes: 900, + requested_bytes: 300, + limit_bytes: 900, + } + ); + drop(reservations.pop()); + let maximum = authority + .reserve_up_to(owner, bytes(u64::MAX), watermark) + .expect("bounded sizing cannot overflow while reserving available bytes"); + assert_eq!(maximum.reserved_bytes(), 300); + assert_eq!(authority.snapshot().used_bytes, 900); + drop((reservations, maximum)); + assert_eq!(authority.snapshot().used_bytes, 0); +} diff --git a/crates/tracedecay/src/daemon/tests/rmcp_route/host_index_handoff.rs b/crates/tracedecay/src/daemon/tests/rmcp_route/host_index_handoff.rs index 06cc07fb88..deb3f268e7 100644 --- a/crates/tracedecay/src/daemon/tests/rmcp_route/host_index_handoff.rs +++ b/crates/tracedecay/src/daemon/tests/rmcp_route/host_index_handoff.rs @@ -19,7 +19,7 @@ async fn ready_reading( .await .unwrap_or_else(|error| panic!("{phase} readiness read failed: {error:?}")); let CodeIndexReadinessWaitReadV1::Reached { reading } = reading else { - panic!("exact worktree did not become ready: {reading:?}"); + panic!("exact worktree did not become ready during {phase}: {reading:?}"); }; *reading } From 32fef3e13a0b4f98b736eddc814c971ac4d95de5 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Wed, 21 Oct 2026 01:33:45 -0700 Subject: [PATCH 3/4] fix(memory): keep delayed refusals behind current text progress --- .../src/code_index_scheduler/serving.rs | 105 ++++++++-- .../tests/code_index_scheduler/serving.rs | 197 +++++++++++++++++- 2 files changed, 279 insertions(+), 23 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 68501f16e5..719d44539f 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 @@ -229,6 +229,29 @@ impl CodeIndexBuildProgressSlotStateV1 { let _ = self.publish(&generation_id, self.owner_epoch, snapshot); } + /// A text refusal belongs to its pending owner. A caller without the + /// projection claim must also identify the progress it observed before + /// attempting admission, so a delayed refusal cannot overwrite new work. + pub(super) fn block_owned_text( + &mut self, + generation_id: &CodeGenerationId, + owner_epoch: u64, + reason: CodeIndexBuildBlockedReasonV1, + expected_progress_epoch: Option, + ) -> bool { + let Some(current) = self.snapshot() else { + return false; + }; + if current.phase == CodeIndexBuildPhaseV1::Ready + || expected_progress_epoch.is_some_and(|epoch| current.progress_epoch != epoch) + { + return false; + } + let mut snapshot = current.as_ref().clone(); + snapshot.blocked_reason = Some(reason); + self.publish(generation_id, owner_epoch, snapshot) + } + pub fn snapshot(&self) -> Option> { self.snapshot.as_ref().map(Arc::clone) } @@ -2673,9 +2696,13 @@ impl LatestCodeTextGenerationV1 { // its phase before it re-attempts the work that was refused, so // clearing here erased the reason between two identical refusals // and status reported `blocked_reason: null` throughout a stall. - // A committed boundary is the honest clear -- it builds a fresh - // snapshot with no blocked reason -- and only that runs after work - // actually landed. + // A committed boundary clears the refusal. Ready is also a + // successful boundary: it is published only after the verified + // query owners are installed, even if finalization completed + // without another pending boundary in between. + if phase == CodeIndexBuildPhaseV1::Ready { + snapshot.blocked_reason = None; + } let _ = slot.publish(generation_id, self.text_progress_owner_epoch, snapshot); } }; @@ -2692,18 +2719,26 @@ impl LatestCodeTextGenerationV1 { fn publish_text_artifact_block(&self, error: &CodeLexicalArtifactErrorV1) { match error { CodeLexicalArtifactErrorV1::Unreserved(_) => { - self.publish_text_progress_blocked(CodeIndexBuildBlockedReasonV1::ResidentMemory); + self.publish_text_progress_blocked( + CodeIndexBuildBlockedReasonV1::ResidentMemory, + None, + ); } CodeLexicalArtifactErrorV1::Io(_) | CodeLexicalArtifactErrorV1::Missing(_) => { self.publish_text_progress_blocked( CodeIndexBuildBlockedReasonV1::ArtifactStoreUnavailable, + None, ); } _ => {} } } - fn publish_text_progress_blocked(&self, reason: CodeIndexBuildBlockedReasonV1) { + fn publish_text_progress_blocked( + &self, + reason: CodeIndexBuildBlockedReasonV1, + expected_progress_epoch: Option, + ) { let generation_id = &self.metadata.manifest().generation_id; { let _span = tracing::trace_span!("query.artifact.progress.publish").entered(); @@ -2712,12 +2747,12 @@ impl LatestCodeTextGenerationV1 { .text_progress_slot .write() .unwrap_or_else(std::sync::PoisonError::into_inner); - let Some(current) = slot.snapshot() else { - return; - }; - let mut snapshot = current.as_ref().clone(); - snapshot.blocked_reason = Some(reason); - let _ = slot.publish(generation_id, self.text_progress_owner_epoch, snapshot); + let _ = slot.block_owned_text( + generation_id, + self.text_progress_owner_epoch, + reason, + expected_progress_epoch, + ); } }; } @@ -2761,6 +2796,12 @@ impl LatestCodeTextGenerationV1 { maximum_work: usize, control: &dyn CodeIndexExecutionControlV1, ) -> Result { + let progress_before_admission = self + .text_progress_slot + .read() + .unwrap_or_else(PoisonError::into_inner) + .snapshot() + .map(|snapshot| snapshot.progress_epoch); let workers = self .text_artifact_store .worker_runtime @@ -2777,14 +2818,20 @@ impl LatestCodeTextGenerationV1 { }); let result = match workers { Ok(_workers) => self.advance_text_serving_inner(maximum_work, control), - Err(error) => Err(error), + Err(error) => { + if matches!(&error, RetrievalPortError::ResidentMemoryRefused(_)) + && let Some(epoch) = progress_before_admission + { + // Worker admission owns no projection claim. Only the + // unchanged progress it observed can inherit this refusal. + self.publish_text_progress_blocked( + CodeIndexBuildBlockedReasonV1::ResidentMemory, + Some(epoch), + ); + } + Err(error) + } }; - // A refusal before the builder opens has no batch/finalization error - // to classify. Preserve that refusal on the same generation's progress - // instead of leaving its completed source scan looking unblocked. - if matches!(&result, Err(RetrievalPortError::ResidentMemoryRefused(_))) { - self.publish_text_progress_blocked(CodeIndexBuildBlockedReasonV1::ResidentMemory); - } if result.as_ref().is_err_and(|error| { matches!( error, @@ -3306,7 +3353,17 @@ impl LatestCodeTextGenerationV1 { let outcome = { let _span = tracing::trace_span!("query.artifact.head_open").entered(); self.open_published_head_or_begin_build(control) - }?; + } + .inspect_err(|error| { + if matches!(error, RetrievalPortError::ResidentMemoryRefused(_)) { + // Keep the claim until this attempt's refusal is visible; + // no later wake can publish progress ahead of this block. + self.publish_text_progress_blocked( + CodeIndexBuildBlockedReasonV1::ResidentMemory, + None, + ); + } + })?; match outcome { TextHeadOpenOutcomeV1::Served => return Ok(true), TextHeadOpenOutcomeV1::Build(initialized) => { @@ -3597,7 +3654,15 @@ impl LatestCodeTextGenerationV1 { &content_key, build_reservation, control, - )?; + ) + .inspect_err(|error| { + if matches!(error, RetrievalPortError::ResidentMemoryRefused(_)) { + self.publish_text_progress_blocked( + CodeIndexBuildBlockedReasonV1::ResidentMemory, + None, + ); + } + })?; Ok(true) } diff --git a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs index fcb1c9d59a..708739c618 100644 --- a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs +++ b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs @@ -3,7 +3,7 @@ use std::{ fmt::Write as _, num::NonZeroU64, path::{Path, PathBuf}, - sync::{Arc, mpsc}, + sync::{Arc, RwLock, mpsc}, thread, time::{Duration, Instant}, }; @@ -2139,8 +2139,12 @@ fn retained_text_admission_reports_its_block_and_recovers_after_headroom_release .expect("hold actual ledger headroom"); let mut headroom = pressure.subscribe_headroom(); - for _ in 0..2 { - let refused = text.advance_text_serving(1); + for claimed_path in [true, false] { + let refused = if claimed_path { + text.advance_artifact_text_serving(1, &text.text_execution_control()) + } else { + text.advance_text_serving(1) + }; assert!(matches!( refused, Err(RetrievalPortError::ResidentMemoryRefused(_)) @@ -2176,6 +2180,101 @@ fn retained_text_admission_reports_its_block_and_recovers_after_headroom_release assert!(active_text_artifact_path(store.path()).is_file()); } +#[test] +fn ready_publication_clears_a_finalization_refusal_without_another_pending_boundary() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn ready_after_refusal() {}\n")]); + let store = TempDir::new().expect("store root"); + let limit = NonZeroU64::new(4 * 1024 * 1024 * 1024).expect("test process limit"); + let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( + limit, + Arc::new(|| { + Some(ProcessResidentSampleV1 { + resident_bytes: 0, + unreclaimable_bytes: 0, + swapped_bytes: 0, + cgroup_headroom_bytes: None, + }) + }), + )); + let mut scheduler = scheduler( + &fixture, + store.path().to_path_buf(), + Arc::new(SharedCodeIndexBytePoolV1::default()), + ); + scheduler.bind_resident_memory(Arc::new(ProcessResidentMemoryV1::with_pressure( + limit, pressure, + ))); + published(scheduler.reconcile_now().expect("seal source generation")); + let _workers = scheduler + .ensure_worker_plan() + .expect("fixture worker runtime"); + let text = scheduler.latest_complete().expect("text generation"); + let control = text.text_execution_control(); + assert!( + !text + .advance_text_serving(0) + .expect("initialize pending text progress") + ); + let finalization_deadline = Instant::now() + SERVING_SEAT_FAILURE_CEILING; + while !matches!( + build_progress_snapshot(&scheduler).phase, + CodeIndexBuildPhaseV1::IndexBuild | CodeIndexBuildPhaseV1::Verification + ) { + assert!( + Instant::now() < finalization_deadline, + "text build did not reach finalization" + ); + assert!( + !text.advance_text_serving(1).expect("begin finalization"), + "the bounded build must expose a pending finalization boundary" + ); + } + { + let mut slot = text.text_projection_build.lock_slot(); + let super::super::CodeTextProjectionSlotV1::Building(build) = &mut *slot else { + panic!("resumable text builder"); + }; + while !matches!( + build + .builder + .advance_started_finalization(1_000, &control) + .expect("finish the durable artifact before publication"), + CodeLexicalArtifactFinalizationStepV1::Ready(_) + ) { + assert!( + Instant::now() < finalization_deadline, + "durable finalization did not finish" + ); + } + } + text.publish_text_progress_phase(CodeIndexBuildPhaseV1::Verification, 0, 0); + let before = build_progress_snapshot(&scheduler); + let progress = scheduler.build_progress_slot(); + assert!( + progress + .write() + .expect("progress authority") + .block_owned_text( + &text.metadata().manifest().generation_id, + text.text_progress_owner_epoch, + CodeIndexBuildBlockedReasonV1::ResidentMemory, + Some(before.progress_epoch), + ) + ); + + // A resumed finalization can go directly from its prior refusal to + // installing query owners. No intervening Pending snapshot clears it. + assert!( + text.advance_artifact_text_serving(1, &control) + .expect("publish the finalized artifact") + ); + let ready = build_progress_snapshot(&scheduler); + assert!(text.query_owners_are_ready()); + assert_eq!(ready.phase, CodeIndexBuildPhaseV1::Ready); + assert_eq!(ready.blocked_reason, None); + assert_eq!(ready.committed_pages, before.committed_pages); +} + #[test] fn text_build_budget_shrinks_to_available_headroom_without_dropping_below_its_floor() { const GIB: u64 = 1024 * 1024 * 1024; @@ -2810,6 +2909,98 @@ fn progress_owner_epoch_rejects_original_a_after_a_b_a_replacement() { after_stale_original.progress_epoch, replacement_a.progress_epoch, "matching generation identity must not let the original A owner replace the newer A epoch" ); + assert!(!slot.block_owned_text( + &generation_a, + original_a_owner, + CodeIndexBuildBlockedReasonV1::ResidentMemory, + Some(after_stale_original.progress_epoch), + )); + assert_eq!( + slot.snapshot().expect("replacement owner").blocked_reason, + None + ); +} + +#[test] +fn delayed_memory_refusal_cannot_replace_newer_pending_or_ready_progress() { + for phase in [ + CodeIndexBuildPhaseV1::BulkCommit, + CodeIndexBuildPhaseV1::Ready, + ] { + let generation = + CodeGenerationId::new("generation.progress-refusal").expect("generation identity"); + let mut state = super::super::CodeIndexBuildProgressSlotStateV1::default(); + let owner = state.replace_generation(generation.clone()); + assert!(state.publish( + &generation, + owner, + progress_snapshot_for_generation(&generation, 1), + )); + let slot = Arc::new(RwLock::new(state)); + let (observed_tx, observed_rx) = mpsc::sync_channel(1); + let (resume_tx, resume_rx) = mpsc::sync_channel(1); + let delayed_slot = Arc::clone(&slot); + let delayed_generation = generation.clone(); + let delayed = thread::spawn(move || { + let observed_epoch = delayed_slot + .read() + .expect("observe pending progress") + .snapshot() + .expect("pending snapshot") + .progress_epoch; + observed_tx.send(()).expect("admission began"); + resume_rx + .recv_timeout(SERVING_SEAT_FAILURE_CEILING) + .expect("admission refused after concurrent progress"); + delayed_slot + .write() + .expect("publish refusal") + .block_owned_text( + &delayed_generation, + owner, + CodeIndexBuildBlockedReasonV1::ResidentMemory, + Some(observed_epoch), + ) + }); + observed_rx + .recv_timeout(SERVING_SEAT_FAILURE_CEILING) + .expect("worker observed its admission epoch"); + let newer = { + let mut current = slot.write().expect("publish concurrent progress"); + let mut next = progress_snapshot_for_generation(&generation, 2); + next.phase = phase; + assert!(current.publish(&generation, owner, next)); + current.snapshot().expect("newer progress") + }; + resume_tx.send(()).expect("release delayed refusal"); + assert!(!delayed.join().expect("delayed worker")); + + let mut current = slot.write().expect("read surviving progress"); + let after = current.snapshot().expect("current progress"); + assert_eq!(after.progress_epoch, newer.progress_epoch); + assert_eq!(after.phase, phase); + assert_eq!(after.committed_pages, 2); + assert_eq!(after.blocked_reason, None); + let accepts_current_refusal = current.block_owned_text( + &generation, + owner, + CodeIndexBuildBlockedReasonV1::ResidentMemory, + Some(after.progress_epoch), + ); + assert_eq!( + accepts_current_refusal, + phase != CodeIndexBuildPhaseV1::Ready + ); + assert_eq!( + current + .snapshot() + .expect("current refusal verdict") + .blocked_reason, + (phase != CodeIndexBuildPhaseV1::Ready) + .then_some(CodeIndexBuildBlockedReasonV1::ResidentMemory), + "an unchanged pending owner must still report its genuine wait" + ); + } } #[test] From 5f9f80666335beeb5a22dc98f5ec00e7c2e2b0b9 Mon Sep 17 00:00:00 2001 From: Zack Jackson Date: Wed, 21 Oct 2026 04:09:59 -0700 Subject: [PATCH 4/4] test(memory): use native headroom change notification --- .../tests/code_index_scheduler/serving.rs | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs index 708739c618..be4f194e80 100644 --- a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs +++ b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs @@ -2163,11 +2163,7 @@ fn retained_text_admission_reports_its_block_and_recovers_after_headroom_release drop(headroom.borrow_and_update()); drop(competing); - assert!( - headroom - .has_changed() - .expect("headroom authority remains live") - ); + assert!(headroom.has_changed()); assert_eq!(memory.snapshot().used_bytes, 0); text.production_query_owners() .expect("same owner builds after headroom is released");