From 5fb49c3fa755fe1e89bc10e5bf6d6e79d6e7898a Mon Sep 17 00:00:00 2001 From: "zackary.l.jackson" Date: Tue, 20 Oct 2026 19:43:41 +0000 Subject: [PATCH 1/9] fix(global-db): skip retention-released Codex alias owners Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../src/observation_adapter.rs | 5 +- .../src/observation_batch_tests.rs | 103 +++++++++++++++++- 2 files changed, 104 insertions(+), 4 deletions(-) diff --git a/crates/tracedecay-global-db/src/observation_adapter.rs b/crates/tracedecay-global-db/src/observation_adapter.rs index 5bc6062563..bcba3ef873 100644 --- a/crates/tracedecay-global-db/src/observation_adapter.rs +++ b/crates/tracedecay-global-db/src/observation_adapter.rs @@ -1285,10 +1285,13 @@ async fn read_retrieval_aliases_from_snapshot( let mut rows = snapshot .query( "SELECT alias.owner_json, alias.alias_kind, alias.locator_digest, - alias.anchor_id, binding.observation_id + alias.anchor_id, owner.observation_id FROM retrieval_anchor_aliases AS alias LEFT JOIN observation_retrieval_anchors AS binding ON binding.anchor_id = alias.anchor_id + LEFT JOIN observations AS owner + ON owner.observation_id = binding.observation_id + AND json_extract(owner.observation_json, '$.__retention_released') IS NULL JOIN json_each(?1) AS requested ON alias.owner_json = json_extract(requested.value, '$.scope_json') diff --git a/crates/tracedecay-global-db/src/observation_batch_tests.rs b/crates/tracedecay-global-db/src/observation_batch_tests.rs index 5898e9ff89..3a6aae3899 100644 --- a/crates/tracedecay-global-db/src/observation_batch_tests.rs +++ b/crates/tracedecay-global-db/src/observation_batch_tests.rs @@ -9,7 +9,7 @@ use std::sync::atomic::{AtomicU64, Ordering}; use tempfile::TempDir; use tracedecay_domain::{ CanonicalMessageRoleV1, CanonicalObservationEnvelopeV1, CanonicalObservationEvidenceV1, - CanonicalObservationFactV1, CanonicalObservationRelationsV1, DurableObservationV1, + CanonicalObservationFactV1, CanonicalObservationRelationsV1, DurableObservationV1, FactOwnerV1, ObservationCollisionOutcomeV1, ObservationId, ObservationIdentityMaterialV1, ObservationOrderingDomainV1, ObservationScopeV1, ObservationSourceCursorV1, ObservationSourceGenerationV1, ObservationSourceIdentityV1, ObservationSourceRangeV1, @@ -18,8 +18,10 @@ use tracedecay_domain::{ SanitizationReceiptV1, SanitizerDispositionV1, SensitivityV1, SessionId, UtcMicros, }; use tracedecay_store::{ - AnchoredObservationWrite, FOREGROUND_BATCH_MAX_OPERATIONS, ObservationBatchPersistOutcome, - ObservationPersistOutcome, ObservationStore, ObservationStoreError, ObservationWrite, + AnchorDispositionReasonClassV1, AnchorDispositionStateV1, AnchoredObservationWrite, + FOREGROUND_BATCH_MAX_OPERATIONS, ObservationBatchPersistOutcome, ObservationPersistOutcome, + ObservationStore, ObservationStoreError, ObservationWrite, RetrievalAnchorDispositionRecordV1, + RetrievalAnchorDispositionStore, }; use tracing::field::{Field, Visit}; use tracing::span::{Attributes, Id, Record}; @@ -837,6 +839,101 @@ async fn non_codex_alias_collision_stays_typed_over_a_corrupt_alias_owner() { ); } +#[asupersync::test] +async fn codex_sibling_over_a_retention_released_owner_is_a_typed_alias_collision() { + const DAY_MICROS: i64 = 24 * 60 * 60 * 1_000_000; + + let (_tmp, runtime, store) = open_profile_store().await; + let thread = THREAD_ROLLOUT.write(); + assert!(matches!( + store.persist_observation(thread.clone()).await.unwrap(), + ObservationPersistOutcome::Committed(_) + )); + + let database = runtime + .registered_database(HostAdmissionScope::Profile) + .unwrap(); + let now = tracedecay_contracts::clock::now_micros(); + let anchor = thread.retrieval_anchor(); + let record = RetrievalAnchorDispositionRecordV1::new( + "observation-batch:retention-released-owner", + anchor.anchor_id().clone(), + FactOwnerV1::from(anchor.owner().clone()), + AnchorDispositionStateV1::Deleted, + None, + AnchorDispositionReasonClassV1::Retention, + UtcMicros(now.0 - 40 * DAY_MICROS), + ) + .unwrap(); + database + .runtime_database() + .append_disposition(record) + .await + .unwrap(); + let report = database + .run_observation_retention( + None, + &crate::observation::retention::ObservationRetentionConfig::default(), + crate::observation::retention::RetentionMode::Apply, + now, + ) + .await + .unwrap(); + assert_eq!( + ( + report.anchors_released.acted, + report.observations_released.acted + ), + (1, 1) + ); + + let unrelated = sequential_writes( + &SessionId::new("session.observation-batch.unrelated").unwrap(), + 1, + ) + .into_iter() + .next() + .unwrap(); + let revert = REVERT_ROLLOUT.write(); + let batch = store + .persist_observations(vec![unrelated.clone(), revert.clone()]) + .await; + + let error = batch.unwrap_err(); + assert!( + matches!( + &error, + ObservationStoreError::RetrievalAnchorAliasCollision { + existing_anchor_id, + .. + } if **existing_anchor_id == *thread.retrieval_anchor_id() + ), + "expected the released owner's alias collision, got {error:?}" + ); + assert!( + store + .get_observation(unrelated.observation().observation_id()) + .await + .unwrap() + .is_none() + ); + assert!( + store + .get_observation(revert.observation().observation_id()) + .await + .unwrap() + .is_none() + ); + assert!(matches!( + store.persist_observation(revert.clone()).await, + Err(ObservationStoreError::RetrievalAnchorAliasCollision { .. }) + )); + assert!(matches!( + store.persist_observation(unrelated.clone()).await, + Ok(ObservationPersistOutcome::Committed(_)) + )); +} + #[asupersync::test] async fn persist_observations_dispatches_one_runtime_command_independent_of_batch_size() { let mut expected_snapshot_probes = None; From 5498869055eb103163cfc9b59f24cfc7bf8ac8e6 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Tue, 20 Oct 2026 18:35:17 -0700 Subject: [PATCH 2/9] test(sessions): verify durable completion after rekey --- .../registry_tests.rs | 119 +++++++++++------- 1 file changed, 73 insertions(+), 46 deletions(-) diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry_tests.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry_tests.rs index ad4495ee87..737f3d4cb8 100644 --- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry_tests.rs +++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry_tests.rs @@ -4,17 +4,23 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::time::Duration; use futures_util::FutureExt; +use tracedecay_domain::SessionId; use tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness; use tracedecay_runtime_core::native_runtime; +use tracedecay_session_temporal_store::SessionTemporalStore; use tracedecay_sessions::serving::{ SessionProjectionServingState, SessionProjectionServingStatusPort, SessionProjectionStaleReason, }; +use tracedecay_store::{ + SessionRefreshBeginOrJoinRequestV1, SessionRefreshDispositionV1, SessionRefreshFrontierV1, + SessionRefreshReceiptRequestV1, SessionRefreshStore, SessionRefreshTerminalStateV1, +}; use super::super::history::{ SessionHistoricalCapacityRelease, SessionHistoricalIngestOutcome, SessionHistoricalIngestPass, SessionHistoricalIngestor, SharedSessionHistoricalIngestor, }; -use super::super::wake::SessionTemporalRefreshWakeState; +use super::super::wake::{SessionTemporalRefreshWake, SessionTemporalRefreshWakeState}; use super::SessionTemporalRefreshSchedulerRegistry; use crate::StoreOwnerKey; @@ -137,6 +143,46 @@ async fn mount_held_project( history } +/// A replacement worker may consume a handed-off request before rekey returns. +/// Join its real pass and require the durable completion, rather than racing +/// the consumer for a transient queue length or dirty flag. These empty sessions +/// have no observation history for discovery to reconstruct a lost request from. +async fn assert_rekey_completed_requests( + harness: &RegisteredGlobalDbHarness, + wake: &SessionTemporalRefreshWake, + sessions: &[&str], +) { + assert!( + wake.wake_and_wait_until_idle(Duration::from_secs(5)).await, + "the retained route must join the replacement worker's pass" + ); + let store = SessionTemporalStore::new(harness.registered.as_ref()); + for name in sessions { + let session_id = SessionId::new(*name).expect("session id"); + let joined = store + .begin_or_join_session_refresh(SessionRefreshBeginOrJoinRequestV1::new( + session_id.clone(), + SessionRefreshFrontierV1::new(0, 0).expect("empty frontier"), + )) + .await + .expect("join the handed-off refresh"); + assert_eq!( + joined.disposition(), + SessionRefreshDispositionV1::Joined, + "{name}: the replacement must have begun the handed-off request" + ); + let receipt = store + .session_refresh_receipt(SessionRefreshReceiptRequestV1::new( + joined.operation_id().clone(), + session_id, + )) + .await + .expect("read handed-off refresh receipt") + .expect("the replacement must complete the handed-off request"); + assert_eq!(receipt.state(), SessionRefreshTerminalStateV1::Complete); + } +} + /// Mounts a project whose worker holds its history pass and a profile entry. async fn mount_drain_fixture( harness: &RegisteredGlobalDbHarness, @@ -410,20 +456,11 @@ async fn a_retried_rekey_recovers_the_dropped_attempts_route_and_requests() { rekey.await; assert!(registry.project_state(&old_owner).await.is_none()); - let new_state = registry + registry .project_state(&new_owner) .await .expect("new owner entry"); - assert_eq!( - new_state.queued_request_count(), - 1, - "the retry must land the retired owner's queued request on the new owner" - ); - assert!( - old_wake.wake(), - "the old route must reach the new owner once the rekey completes" - ); - assert!(new_state.take_dirty()); + assert_rekey_completed_requests(&harness, &old_wake, &["session.dropped-rekey"]).await; registry.shutdown().await; } @@ -496,19 +533,11 @@ async fn a_dropped_rekey_keeps_its_route_and_requests_after_the_worker_finished( registry .rekey_project(&old_owner, new_owner.clone(), harness.registered.clone()) .await; - let resumed = registry + registry .project_state(&new_owner) .await .expect("new owner entry"); - assert_eq!( - resumed.queued_request_count(), - 1, - "a rekey dropped after its join must retain the requests it staged" - ); - assert!( - old_wake.wake(), - "the old route must reach the new owner once the retry completes" - ); + assert_rekey_completed_requests(&harness, &old_wake, &["session.finished-rekey"]).await; registry.shutdown().await; } @@ -601,6 +630,8 @@ async fn a_rekey_after_a_readmission_lands_every_route_and_request() { !a_wake.wake(), "the parked handoff stays owed to its own destination" ); + let readmitted_history = + mount_held_project(®istry, &harness, &a_owner, HeldHistory::default()).await; let readmitted = registry .project_state(&a_owner) .await @@ -610,24 +641,29 @@ async fn a_rekey_after_a_readmission_lands_every_route_and_request() { tracedecay_store::SessionRefreshFrontierV1::new(0, 0).unwrap(), )); - registry - .rekey_project(&a_owner, d_owner.clone(), harness.registered.clone()) - .await; + let mut rekey = std::pin::pin!(registry.rekey_project( + &a_owner, + d_owner.clone(), + harness.registered.clone(), + )); + assert!( + rekey.as_mut().now_or_never().is_none(), + "the readmitted worker must still hold the queued request before its handoff" + ); + readmitted_history.release.notify_one(); + rekey.await; assert!(registry.project_state(&a_owner).await.is_none()); - let landed = registry + registry .project_state(&d_owner) .await .expect("final owner entry"); - assert_eq!( - landed.queued_request_count(), - 2, - "the parked request and the readmitted entry's request must both reach the final owner" - ); - assert!( - a_wake.wake(), - "the parked route must reach the new owner once the rekey completes" - ); + assert_rekey_completed_requests( + &harness, + &a_wake, + &["session.readmitted-a", "session.readmitted-live"], + ) + .await; assert!( c_wake.wake(), "the live route the readmission installed must reach the new owner too" @@ -744,20 +780,11 @@ async fn a_mapped_rekey_lands_the_workers_queued_requests() { rekey.await; assert!(registry.project_state(&old_owner).await.is_none()); - let new_state = registry + registry .project_state(&new_owner) .await .expect("new owner entry"); - assert_eq!( - new_state.queued_request_count(), - 1, - "the new owner must inherit the mapped worker's queued request" - ); - assert!( - old_wake.wake(), - "the old route must reach the new owner once the rekey completes" - ); - assert!(new_state.take_dirty()); + assert_rekey_completed_requests(&harness, &old_wake, &["session.mapped-rekey"]).await; registry.shutdown().await; } From 15e808cfe4108b11941f8a977da4ac78c9b7c237 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Tue, 20 Oct 2026 23:31:45 -0700 Subject: [PATCH 3/9] test(hooks): verify capture outcomes for every callback The spool admits staged envelopes at least once, so a commit-lock timeout can append another frame after the original was already acknowledged. Sequence counts therefore cannot identify distinct captured callbacks. Verify each concurrent Cursor session has exactly one completed capture outcome with accepted_for_replay status. This still fails for an unbound callback and detects missing or duplicated outcomes per session. --- .../multi_connection_test/ownership.rs | 45 ++++++++++++++----- 1 file changed, 33 insertions(+), 12 deletions(-) diff --git a/crates/tracedecay/tests/storage_suite/multi_connection_test/ownership.rs b/crates/tracedecay/tests/storage_suite/multi_connection_test/ownership.rs index 08de819f7a..1ac4f4fbea 100644 --- a/crates/tracedecay/tests/storage_suite/multi_connection_test/ownership.rs +++ b/crates/tracedecay/tests/storage_suite/multi_connection_test/ownership.rs @@ -207,19 +207,40 @@ fn twelve_mcp_cli_and_hook_clients_share_one_daemon_profile_store_owner() { let project_id = tracedecay_runtime_core::storage::default_profile_project_id(&project_path); let data_root = tracedecay_runtime_core::storage::profile_sharded_data_root(&profile_root, &project_id); - let host = tracedecay_domain::NativeHostIdentityV1::CursorDesktop; - let (_, capture_report) = tracedecay_hooks::HookSpoolV1::open( - data_root.join("hook-v2-spool").join(host.hook_key()), - tracedecay_hooks::HookSpoolConfigV1::stock(host), - tracedecay_contracts::now_micros(), - ) - .expect("open native capture receipt"); - // Sequence survives daemon replay/acknowledgement, so this proves each - // callback was captured even if the owner already drained its envelope. + // A commit-lock timeout can stage an envelope whose first spool frame + // the daemon already acknowledged. Importing it appends a duplicate + // frame, so the sequence counts attempts rather than distinct callbacks. + // Each callback's completed receipt instead proves that its own capture + // landed, including one safely staged for replay, rather than returning + // the same transport response for an unbound hook. + let analytics = std::fs::read_to_string(data_root.join("hook_analytics.jsonl")) + .expect("read native capture outcomes"); + let mut captured_sessions = analytics + .lines() + .map(|line| serde_json::from_str::(line).expect("hook analytics row")) + .filter(|row| { + row["event"] == "hook_completed" + && row["agent"] == "cursor" + && row["hook_name"] == "afterFileEdit" + }) + .map(|row| { + assert_eq!( + row["disposition"]["status"], "accepted_for_replay", + "native callback must be durably captured: {row}" + ); + row["session_id"] + .as_str() + .expect("capture names the callback session") + .to_owned() + }) + .collect::>(); + captured_sessions.sort(); assert_eq!( - capture_report.next_sequence, - 1 + CONCURRENT_CLIENTS_PER_PATH as u64, - "every native callback must be captured, not merely accepted as unbound" + captured_sessions, + (0..CONCURRENT_CLIENTS_PER_PATH) + .map(|ordinal| format!("session-{ordinal}")) + .collect::>(), + "every native callback must have exactly one successful capture outcome" ); let daemon_stderr = std::fs::read_to_string(&daemon_stderr_path).expect("read daemon stderr"); assert!( From 85bd4c7b34a65e9f67379f931a28aaf4a3d1c068 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Tue, 20 Oct 2026 23:31:47 -0700 Subject: [PATCH 4/9] fix(memory): reserve text builds below the admission ceiling Text admission sized from shared cgroup headroom, then reserved an extra watermark band against that same room. A 1.5 GiB build was refused with more than 1.5 GiB available because the reserve requested another 0.6 GiB. Add ceiling-aware reservation to the canonical process authority and use it for text artifacts. The ledger and measured process still obey the watermark, while the cgroup is charged only for the actual allocation. Remove the synthetic margin charge and its immediate shrink. Exercise actual concurrent reservations, a cgroup-constrained admission, and text builds with exactly their required cgroup headroom. Native regressions run in CI; local rustfmt and diff checks pass. --- .../src/code_index_scheduler.rs | 3 +- .../src/code_index_scheduler/serving.rs | 98 ++++++------------- .../tests/code_index_scheduler/serving.rs | 89 ++++++++++------- .../src/resident_memory.rs | 54 +++++++--- .../src/resident_memory/tests.rs | 39 ++++++++ 5 files changed, 168 insertions(+), 115 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..26a77bcc2d 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 @@ -947,29 +947,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 +1147,9 @@ 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. + /// Reserve one artifact memory ceiling below the process-wide admission + /// watermark. Only the component's bytes are charged to the ledger and + /// checked against shared cgroup headroom. fn reserve_resident_memory( &self, generation_id: &CodeGenerationId, @@ -1184,20 +1160,14 @@ impl DaemonCodeTextArtifactStoreV1 { .map(|(reservation, _)| reservation) } - /// Measure headroom and size the build: `(observed, watermark headroom, - /// admitted)` bytes. + /// Measure headroom and size the build below the admission watermark. fn text_artifact_admission( &self, preferred: NonZeroU64, minimum: NonZeroU64, - ) -> Result<(u64, u64), RetrievalPortError> { + ) -> Result { 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(), @@ -1207,7 +1177,7 @@ impl DaemonCodeTextArtifactStoreV1 { self.resident_memory .wait_for_headroom(minimum.get(), admission_watermark); })?; - Ok((watermark_headroom, admitted_bytes)) + Ok(admitted_bytes) } fn reserve_resident_memory_up_to( @@ -1235,56 +1205,50 @@ 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 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); } - admission => admission, - }?; + // 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) + } + 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 + let reservation = self .resident_memory - .reserve( + .reserve_below( ResidentMemoryKeyV1 { project_id: self.project_id.clone(), worktree_id: self.worktree_id.clone(), generation_id: generation_id.clone(), component, }, - accounted, + admitted, + self.resident_memory.admission_watermark_bytes(), ) .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 admitted = usize::try_from(admitted.get()).map_err(|error| { RetrievalPortError::Contract(format!( "text-artifact admitted reservation exceeds the platform limit: {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..6e9b36212f 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 @@ -2087,41 +2087,53 @@ 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; + use tracedecay_domain::WorktreeId; + use tracedecay_runtime_core::resident_memory::{ + ResidentMemoryAdmissionFailureV1, ResidentMemoryComponentIdV1, ResidentMemoryKeyV1, + }; - 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" - ); - used_bytes += requested.get(); - } + let limit = NonZeroU64::new(1_000).expect("nonzero process ceiling"); + let authority = Arc::new(ProcessResidentMemoryV1::with_pressure( + limit, + Arc::new(ResidentMemoryPressureV1::new(limit)), + )); + let key = ResidentMemoryKeyV1 { + project_id: ProjectId::new("project.overlapping-text-builds").expect("project"), + worktree_id: WorktreeId::new("worktree.overlapping-text-builds").expect("worktree"), + generation_id: CodeGenerationId::new("generation.overlapping-text-builds") + .expect("generation"), + component: ResidentMemoryComponentIdV1::new("code-text-artifact-build").expect("component"), + }; + let requested = NonZeroU64::new(300).expect("nonzero build request"); + let builds = [(); 3].map(|()| { + authority + .reserve_below( + key.clone(), + requested, + authority.admission_watermark_bytes(), + ) + .expect("each build fits beneath the shared 900-byte watermark") + }); assert_eq!( - used_bytes, 900, + authority.snapshot().used_bytes, + 900, "the retained ledger owns the three build ceilings, not three margins" ); - for overflow in [ - super::super::text_artifact_resident_memory_charge( - NonZeroU64::new(u64::MAX).expect("maximum nonzero request"), - 1, + assert!(matches!( + authority.reserve_below( + key, + NonZeroU64::new(100).expect("remaining hard-limit room"), + authority.admission_watermark_bytes(), ), - 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:?}" - ); - } + Err(ResidentMemoryAdmissionFailureV1::ReservationCeiling { + used_bytes: 900, + requested_bytes: 100, + limit_bytes: 900, + }) + )); + drop(builds); + assert_eq!(authority.snapshot().used_bytes, 0); } #[test] @@ -2199,7 +2211,14 @@ fn text_artifact_ceilings_reserve_through_process_resident_memory() { RESIDENT_MEMORY_PRESSURE_HIGH_WATERMARK_PERMILLE_V1, ); let fitting_live = watermark - build_floor; - for (live_bytes, admitted) in [(fitting_live, true), (fitting_live + 1, false)] { + for (live_bytes, cgroup_headroom, admitted) in [ + (fitting_live, None, true), + (fitting_live + 1, None, false), + // Shared cgroup room must cover the actual build, not a second copy + // of the band above the process admission watermark. + (0, Some(build_floor), true), + (0, Some(build_floor - 1), false), + ] { let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( measured_limit, Arc::new(move || { @@ -2207,7 +2226,7 @@ fn text_artifact_ceilings_reserve_through_process_resident_memory() { resident_bytes: live_bytes, unreclaimable_bytes: live_bytes, swapped_bytes: 0, - cgroup_headroom_bytes: None, + cgroup_headroom_bytes: cgroup_headroom, }) }), )); @@ -2224,11 +2243,13 @@ fn text_artifact_ceilings_reserve_through_process_resident_memory() { let latest = scheduler .latest_complete() .expect("measured latest generation"); - let advanced = latest.advance_text_serving(1); + // Open the build and its reservation without consuming source pages, + // so every case tests admission of the same unpublished artifact. + let advanced = latest.advance_text_serving(0); if admitted { assert!( advanced.is_ok(), - "{live_bytes} B live leaves the build floor below the watermark: {advanced:?}" + "{live_bytes} B live with cgroup room {cgroup_headroom:?} admits the build floor: {advanced:?}" ); } else { assert!( @@ -2236,7 +2257,7 @@ fn text_artifact_ceilings_reserve_through_process_resident_memory() { advanced, Err(tracedecay_query::retrieval::RetrievalPortError::ResidentMemoryRefused(_)) ), - "{live_bytes} B live leaves less than the build floor: {advanced:?}" + "{live_bytes} B live with cgroup room {cgroup_headroom:?} leaves less than the build floor: {advanced:?}" ); assert_eq!( measured.snapshot().used_bytes, diff --git a/crates/tracedecay-runtime-core/src/resident_memory.rs b/crates/tracedecay-runtime-core/src/resident_memory.rs index 34a9c445be..606ac2e31b 100644 --- a/crates/tracedecay-runtime-core/src/resident_memory.rs +++ b/crates/tracedecay-runtime-core/src/resident_memory.rs @@ -1521,14 +1521,39 @@ impl ProcessResidentMemoryV1 { key: ResidentMemoryKeyV1, requested_bytes: NonZeroU64, ) -> Result { - if let Some(failure) = self.observed_over_budget_refusal(requested_bytes) { + self.reserve_below(key, requested_bytes, u64::MAX) + } + + /// Reserve below a caller's admission ceiling using the same occupancy + /// as [`Self::headroom_below`]. The ceiling constrains both the ledger and + /// the process measurement; shared cgroup capacity constrains only the + /// requested allocation. Reserving the band above the ceiling as fake + /// component bytes would charge that band against cgroup room as well. + #[tracing::instrument( + name = "runtime_core.resident.reserve_below", + level = "trace", + skip_all + )] + pub fn reserve_below( + self: &Arc, + key: ResidentMemoryKeyV1, + requested_bytes: NonZeroU64, + ceiling: u64, + ) -> Result { + if let Some(failure) = self.observed_over_budget_refusal_below(requested_bytes, ceiling) { return Err(self.refused(failure)); } - if let Some(reservation) = self.try_reserve(&key, requested_bytes) { + let ledger_ceiling = ceiling.min(self.limit_bytes.get()); + if let Some(reservation) = self.try_reserve(&key, requested_bytes, ledger_ceiling) { return Ok(reservation); } - Err(self.refused(self.admission_failure(requested_bytes))) + let failure = ResidentMemoryAdmissionFailureV1::ReservationCeiling { + used_bytes: self.lock_state().used_bytes, + requested_bytes: requested_bytes.get(), + limit_bytes: ledger_ceiling, + }; + Err(self.refused(failure)) } /// Reserves one process-shared component without fabricating a project, @@ -1637,10 +1662,11 @@ impl ProcessResidentMemoryV1 { self: &Arc, key: &ResidentMemoryKeyV1, requested_bytes: NonZeroU64, + ceiling: u64, ) -> Option { let mut state = self.lock_state(); let next_used = state.used_bytes.checked_add(requested_bytes.get())?; - if next_used > self.limit_bytes.get() { + if next_used > ceiling { return None; } state.used_bytes = next_used; @@ -1662,30 +1688,34 @@ impl ProcessResidentMemoryV1 { fn observed_over_budget_refusal( &self, requested_bytes: NonZeroU64, + ) -> Option { + self.observed_over_budget_refusal_below(requested_bytes, u64::MAX) + } + + fn observed_over_budget_refusal_below( + &self, + requested_bytes: NonZeroU64, + ceiling: u64, ) -> Option { let pressure = self.pressure.state(); - let observed_bytes = pressure.observed_bytes()?; + let measured_ceiling = ceiling.min(self.pressure.limit_bytes()); + let observed_bytes = self.pressure.observed_bytes_for_limit(measured_ceiling)?; let high_watermark_bytes = self.pressure.high_watermark_bytes(); if requested_bytes.get() <= RESIDENT_MEMORY_PRESSURE_ADMISSION_FLOOR_BYTES_V1 || (!pressure.is_over_budget() - && requested_bytes.get() - <= self.pressure.limit_bytes().saturating_sub(observed_bytes)) + && requested_bytes.get() <= measured_ceiling.saturating_sub(observed_bytes)) { return None; } Some(ResidentMemoryAdmissionFailureV1::ObservedOverBudget { observed_bytes, - limit_bytes: self.pressure.limit_bytes(), + limit_bytes: measured_ceiling, high_watermark_bytes, requested_bytes: requested_bytes.get(), floor_bytes: RESIDENT_MEMORY_PRESSURE_ADMISSION_FLOOR_BYTES_V1, }) } - fn admission_failure(&self, requested_bytes: NonZeroU64) -> ResidentMemoryAdmissionFailureV1 { - self.admission_failure_from_used(self.lock_state().used_bytes, requested_bytes) - } - fn admission_failure_from_used( &self, used_bytes: u64, diff --git a/crates/tracedecay-runtime-core/src/resident_memory/tests.rs b/crates/tracedecay-runtime-core/src/resident_memory/tests.rs index 557293b4ec..93e1948c3e 100644 --- a/crates/tracedecay-runtime-core/src/resident_memory/tests.rs +++ b/crates/tracedecay-runtime-core/src/resident_memory/tests.rs @@ -1335,6 +1335,45 @@ fn cgroup_headroom_respects_a_larger_but_fuller_ancestor() { ); } +#[test] +fn reservations_below_a_watermark_use_the_headroom_measurement() { + 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 refused = authority + .reserve_below( + owner.clone(), + bytes(headroom.available_bytes + 1), + watermark, + ) + .expect_err("growth beyond shared cgroup headroom must be refused"); + assert!(refused.is_observed_over_budget()); + assert_eq!(authority.snapshot().used_bytes, 0); + + let held = authority + .reserve_below(owner, bytes(headroom.available_bytes), 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; From 4eca19408621db3e40063108e1c0e82150658298 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Tue, 20 Oct 2026 23:31:48 -0700 Subject: [PATCH 5/9] test(cli): retain daemon evidence for parked index failures The first-generation and dirty-branch journeys reported only a generic convergence park while their daemon stderr was discarded. Keep a separate log for each existing journey and include it on failure so CI names the refused operation and its actual cause. Keep the existing success assertions, readiness deadlines and polling. --- .../cli_non_interactive_test.rs | 23 +++++++++++++++---- 1 file changed, 18 insertions(+), 5 deletions(-) diff --git a/crates/tracedecay-cli/tests/core_cli_suite/cli_non_interactive_test.rs b/crates/tracedecay-cli/tests/core_cli_suite/cli_non_interactive_test.rs index 25be1f0fc4..3aa1fcd802 100644 --- a/crates/tracedecay-cli/tests/core_cli_suite/cli_non_interactive_test.rs +++ b/crates/tracedecay-cli/tests/core_cli_suite/cli_non_interactive_test.rs @@ -1200,14 +1200,21 @@ fn init_wait_holds_until_first_generation_is_ready() { git(project.path(), &["init", "-b", "main"]); commit_all(project.path(), "fixture repository"); + let daemon_log = home.path().join("init-wait-daemon.log"); + let _daemon = + crate::common::spawn_tracedecay_daemon_logged(home.path(), &daemon_log, |command| { + command.env("RUST_LOG", "info"); + }); let mut command = tracedecay_command(home.path(), project.path()); command.args(["init", "--wait"]); let output = run_with_timeout(command, cli_timeout()); assert!( output.status.success(), - "init --wait should succeed once the first generation is ready\nstdout:\n{}\nstderr:\n{}", + "init --wait should succeed once the first generation is ready\nstdout:\n{}\nstderr:\n{}\ndaemon:\n{}", String::from_utf8_lossy(&output.stdout), - String::from_utf8_lossy(&output.stderr) + String::from_utf8_lossy(&output.stderr), + std::fs::read_to_string(&daemon_log) + .unwrap_or_else(|error| format!("daemon log unavailable: {error}")) ); let stderr = String::from_utf8_lossy(&output.stderr); assert!( @@ -3328,7 +3335,11 @@ fn branch_search_serves_a_committed_generation_behind_dirty_worktree_state() { .unwrap(); init_project_fixture(home.path(), &project_root); - let _daemon = crate::common::spawn_tracedecay_daemon(home.path()); + let daemon_log = home.path().join("dirty-branch-daemon.log"); + let _daemon = + crate::common::spawn_tracedecay_daemon_logged(home.path(), &daemon_log, |command| { + command.env("RUST_LOG", "info"); + }); let ready_deadline = Instant::now() + Duration::from_secs(30); loop { let mut ready = tracedecay_command_without_daemon(home.path(), &project_root); @@ -3352,9 +3363,11 @@ fn branch_search_serves_a_committed_generation_behind_dirty_worktree_state() { } assert!( Instant::now() < ready_deadline, - "dirty worktree generation did not become queryable\nstdout:\n{}\nstderr:\n{}", + "dirty worktree generation did not become queryable\nstdout:\n{}\nstderr:\n{}\ndaemon:\n{}", String::from_utf8_lossy(&ready.stdout), - String::from_utf8_lossy(&ready.stderr) + String::from_utf8_lossy(&ready.stderr), + std::fs::read_to_string(&daemon_log) + .unwrap_or_else(|error| format!("daemon log unavailable: {error}")) ); std::thread::sleep(Duration::from_millis(100)); } From 25b8aed923f6bdc5935a4216eda328c4fc6e0b2b Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Tue, 20 Oct 2026 23:31:50 -0700 Subject: [PATCH 6/9] test(dashboard): expose explorer timeout state in CI failures --- .../tracedecay/tests/dashboard_api_test/explorer.rs | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/crates/tracedecay/tests/dashboard_api_test/explorer.rs b/crates/tracedecay/tests/dashboard_api_test/explorer.rs index 28d218f51a..a6b363ce0f 100644 --- a/crates/tracedecay/tests/dashboard_api_test/explorer.rs +++ b/crates/tracedecay/tests/dashboard_api_test/explorer.rs @@ -1,21 +1,27 @@ -use std::time::Duration; +use std::time::{Duration, Instant}; use crate::dashboard_api_support::*; use serde_json::{Value, json}; fn wait_for_query(agent: &ureq::Agent, base_url: &str, run_id: &str) -> Value { + let started = Instant::now(); + let mut last_response = None; for _ in 0..100 { let (status, body) = get_json(agent, &format!("{base_url}/api/explorer/queries/{run_id}")); assert_eq!(status, 200, "query status should resolve: {body}"); if matches!( body["payload"]["state"].as_str(), - Some("completed" | "partial" | "cancelled" | "error") + Some("completed" | "partial" | "cancelled" | "timed_out" | "error") ) { return body; } + last_response = Some(body); std::thread::sleep(Duration::from_millis(10)); } - panic!("explorer query {run_id} did not reach a terminal state"); + panic!( + "explorer query {run_id} did not reach a terminal state after {:?}; last response: {last_response:?}", + started.elapsed() + ); } #[test] From 1348d654da83d5b9b85fae508a5a914267cd1d93 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Wed, 21 Oct 2026 00:21:13 -0700 Subject: [PATCH 7/9] fix(memory): size text reservations inside ledger admission Reuse the reviewed reserve_up_to implementation from 863e868d3894ac2c0d1f51248e8b35b176040ef0 so text builders can reserve a supported smaller budget when headroom changes. Keep sizing, charging, minimum-floor validation, and the bounded shedding retry on one admission path, then park only for the minimum working set. Remove the unused reserve_below facade. Preserve cgroup headroom and real artifact floor regressions, and carry the existing sized-admission and shared-ledger behavior tests. --- .../src/code_index_scheduler/serving.rs | 90 ++++++-------- .../tests/code_index_scheduler/serving.rs | 19 ++- .../src/resident_memory.rs | 108 +++++++++++------ .../src/resident_memory/tests.rs | 112 ++++++++++++++++-- 4 files changed, 218 insertions(+), 111 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 26a77bcc2d..e6a5be1c4a 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, @@ -1147,9 +1146,8 @@ impl DaemonCodeTextArtifactStoreV1 { } } - /// Reserve one artifact memory ceiling below the process-wide admission - /// watermark. Only the component's bytes are charged to the ledger and - /// checked against shared cgroup headroom. + /// 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, @@ -1160,26 +1158,6 @@ impl DaemonCodeTextArtifactStoreV1 { .map(|(reservation, _)| reservation) } - /// Measure headroom and size the build below the admission watermark. - fn text_artifact_admission( - &self, - preferred: NonZeroU64, - minimum: NonZeroU64, - ) -> Result { - let admission_watermark = self.resident_memory.admission_watermark_bytes(); - let headroom = self.resident_memory.headroom_below(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(admitted_bytes) - } - fn reserve_resident_memory_up_to( &self, generation_id: &CodeGenerationId, @@ -1205,7 +1183,33 @@ impl DaemonCodeTextArtifactStoreV1 { "text-artifact minimum resident-memory reservation must be nonzero".to_owned(), ) })?; - let admitted_bytes = match 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 @@ -1213,8 +1217,6 @@ impl DaemonCodeTextArtifactStoreV1 { 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", @@ -1223,33 +1225,17 @@ impl DaemonCodeTextArtifactStoreV1 { refused = %detail, "text-artifact build shed retained state before re-measuring headroom" ); - self.text_artifact_admission(preferred, minimum) + reserve() + } + result => result, + } + .inspect_err(|error| { + if matches!(error, RetrievalPortError::ResidentMemoryRefused(_)) { + self.resident_memory + .wait_for_headroom(minimum.get(), admission_watermark); } - 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 reservation = self - .resident_memory - .reserve_below( - ResidentMemoryKeyV1 { - project_id: self.project_id.clone(), - worktree_id: self.worktree_id.clone(), - generation_id: generation_id.clone(), - component, - }, - admitted, - self.resident_memory.admission_watermark_bytes(), - ) - .map_err(|error| { - RetrievalPortError::ResidentMemoryRefused(format!( - "text-artifact resident-memory admission was refused: {error}" - )) - })?; - 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}" )) 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 6e9b36212f..64b6aab9f1 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 @@ -32,7 +32,7 @@ use tracedecay_domain::{ PrivacyDomainId, ProjectId, ProviderEvaluationStateV1, PublicRetrieverStatus, QueryNormalizationRevision, RelationEdgeKindV1, RetrievalBudget, RetrievalRequest, RetrievalScope, RetrievalSnapshot, RetrieverKind, RetrieverOutcome, SanitizerRevision, - ScoreDomainId, SingleRootScopeV1, TemporalModeV1, UtcMicros, VectorWatermark, + ScoreDomainId, SingleRootScopeV1, TemporalModeV1, UtcMicros, VectorWatermark, WorktreeId, encode_lowercase_hex, sha256_hex_suffix, }; use tracedecay_query::retrieval::{ @@ -49,7 +49,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_PRESSURE_HIGH_WATERMARK_PERMILLE_V1, ResidentMemoryAdmissionFailureV1, + ResidentMemoryComponentIdV1, ResidentMemoryKeyV1, ResidentMemoryPressureV1, resident_memory_watermark_bytes_v1, }; @@ -2087,15 +2088,13 @@ fn reader_reservation_refusal_precedes_missing_artifact_access() { #[test] fn overlapping_text_builds_share_one_admission_watermark_headroom() { - use tracedecay_domain::WorktreeId; - use tracedecay_runtime_core::resident_memory::{ - ResidentMemoryAdmissionFailureV1, ResidentMemoryComponentIdV1, ResidentMemoryKeyV1, - }; - let limit = NonZeroU64::new(1_000).expect("nonzero process ceiling"); let authority = Arc::new(ProcessResidentMemoryV1::with_pressure( limit, - Arc::new(ResidentMemoryPressureV1::new(limit)), + Arc::new(ResidentMemoryPressureV1::with_sampler( + limit, + Arc::new(|| None), + )), )); let key = ResidentMemoryKeyV1 { project_id: ProjectId::new("project.overlapping-text-builds").expect("project"), @@ -2107,7 +2106,7 @@ fn overlapping_text_builds_share_one_admission_watermark_headroom() { let requested = NonZeroU64::new(300).expect("nonzero build request"); let builds = [(); 3].map(|()| { authority - .reserve_below( + .reserve_up_to( key.clone(), requested, authority.admission_watermark_bytes(), @@ -2121,7 +2120,7 @@ fn overlapping_text_builds_share_one_admission_watermark_headroom() { "the retained ledger owns the three build ceilings, not three margins" ); assert!(matches!( - authority.reserve_below( + authority.reserve_up_to( key, NonZeroU64::new(100).expect("remaining hard-limit room"), authority.admission_watermark_bytes(), diff --git a/crates/tracedecay-runtime-core/src/resident_memory.rs b/crates/tracedecay-runtime-core/src/resident_memory.rs index 606ac2e31b..a16e42ca7d 100644 --- a/crates/tracedecay-runtime-core/src/resident_memory.rs +++ b/crates/tracedecay-runtime-core/src/resident_memory.rs @@ -1521,39 +1521,70 @@ impl ProcessResidentMemoryV1 { key: ResidentMemoryKeyV1, requested_bytes: NonZeroU64, ) -> Result { - self.reserve_below(key, requested_bytes, u64::MAX) + if let Some(failure) = self.observed_over_budget_refusal(requested_bytes) { + return Err(self.refused(failure)); + } + if let Some(reservation) = self.try_reserve(&key, requested_bytes) { + return Ok(reservation); + } + + Err(self.refused(self.admission_failure(requested_bytes))) } - /// Reserve below a caller's admission ceiling using the same occupancy - /// as [`Self::headroom_below`]. The ceiling constrains both the ledger and - /// the process measurement; shared cgroup capacity constrains only the - /// requested allocation. Reserving the band above the ceiling as fake - /// component bytes would charge that band against cgroup room as well. + /// 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_below", + name = "runtime_core.resident.reserve_up_to", level = "trace", skip_all )] - pub fn reserve_below( + pub fn reserve_up_to( self: &Arc, key: ResidentMemoryKeyV1, - requested_bytes: NonZeroU64, + preferred_bytes: NonZeroU64, ceiling: u64, ) -> Result { - if let Some(failure) = self.observed_over_budget_refusal_below(requested_bytes, ceiling) { - return Err(self.refused(failure)); - } - let ledger_ceiling = ceiling.min(self.limit_bytes.get()); - if let Some(reservation) = self.try_reserve(&key, requested_bytes, ledger_ceiling) { - return Ok(reservation); + 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); } - - let failure = ResidentMemoryAdmissionFailureV1::ReservationCeiling { - used_bytes: self.lock_state().used_bytes, - requested_bytes: requested_bytes.get(), - limit_bytes: ledger_ceiling, - }; - Err(self.refused(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, @@ -1621,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 { @@ -1662,11 +1701,10 @@ impl ProcessResidentMemoryV1 { self: &Arc, key: &ResidentMemoryKeyV1, requested_bytes: NonZeroU64, - ceiling: u64, ) -> Option { let mut state = self.lock_state(); let next_used = state.used_bytes.checked_add(requested_bytes.get())?; - if next_used > ceiling { + if next_used > self.limit_bytes.get() { return None; } state.used_bytes = next_used; @@ -1688,34 +1726,30 @@ impl ProcessResidentMemoryV1 { fn observed_over_budget_refusal( &self, requested_bytes: NonZeroU64, - ) -> Option { - self.observed_over_budget_refusal_below(requested_bytes, u64::MAX) - } - - fn observed_over_budget_refusal_below( - &self, - requested_bytes: NonZeroU64, - ceiling: u64, ) -> Option { let pressure = self.pressure.state(); - let measured_ceiling = ceiling.min(self.pressure.limit_bytes()); - let observed_bytes = self.pressure.observed_bytes_for_limit(measured_ceiling)?; + let observed_bytes = pressure.observed_bytes()?; let high_watermark_bytes = self.pressure.high_watermark_bytes(); if requested_bytes.get() <= RESIDENT_MEMORY_PRESSURE_ADMISSION_FLOOR_BYTES_V1 || (!pressure.is_over_budget() - && requested_bytes.get() <= measured_ceiling.saturating_sub(observed_bytes)) + && requested_bytes.get() + <= self.pressure.limit_bytes().saturating_sub(observed_bytes)) { return None; } Some(ResidentMemoryAdmissionFailureV1::ObservedOverBudget { observed_bytes, - limit_bytes: measured_ceiling, + limit_bytes: self.pressure.limit_bytes(), high_watermark_bytes, requested_bytes: requested_bytes.get(), floor_bytes: RESIDENT_MEMORY_PRESSURE_ADMISSION_FLOOR_BYTES_V1, }) } + fn admission_failure(&self, requested_bytes: NonZeroU64) -> ResidentMemoryAdmissionFailureV1 { + self.admission_failure_from_used(self.lock_state().used_bytes, requested_bytes) + } + fn admission_failure_from_used( &self, used_bytes: u64, diff --git a/crates/tracedecay-runtime-core/src/resident_memory/tests.rs b/crates/tracedecay-runtime-core/src/resident_memory/tests.rs index 93e1948c3e..10fe13f033 100644 --- a/crates/tracedecay-runtime-core/src/resident_memory/tests.rs +++ b/crates/tracedecay-runtime-core/src/resident_memory/tests.rs @@ -1336,7 +1336,7 @@ fn cgroup_headroom_respects_a_larger_but_fuller_ancestor() { } #[test] -fn reservations_below_a_watermark_use_the_headroom_measurement() { +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( @@ -1355,18 +1355,8 @@ fn reservations_below_a_watermark_use_the_headroom_measurement() { 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 refused = authority - .reserve_below( - owner.clone(), - bytes(headroom.available_bytes + 1), - watermark, - ) - .expect_err("growth beyond shared cgroup headroom must be refused"); - assert!(refused.is_observed_over_budget()); - assert_eq!(authority.snapshot().used_bytes, 0); - let held = authority - .reserve_below(owner, bytes(headroom.available_bytes), watermark) + .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); @@ -1658,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); +} From d24077509d57f2eae43d8f974334142dbac66266 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Wed, 21 Oct 2026 10:36:12 -0700 Subject: [PATCH 8/9] fix(memory): isolate fixtures and preserve admission progress --- .../src/agents/claude/tests.rs | 44 ++- .../src/code_index_scheduler/memory_tests.rs | 38 +- .../src/code_index_scheduler/reconcile.rs | 19 +- .../registry/resident_memory.rs | 16 +- .../src/code_index_scheduler/serving.rs | 106 ++++- .../code_index_scheduler/tests/residency.rs | 12 +- .../tests/code_index_scheduler/reconcile.rs | 16 +- .../tests/code_index_scheduler/serving.rs | 369 +++++++++++++++++- 8 files changed, 551 insertions(+), 69 deletions(-) diff --git a/crates/tracedecay-agent-hosts/src/agents/claude/tests.rs b/crates/tracedecay-agent-hosts/src/agents/claude/tests.rs index 707bc9da32..db426a6db3 100644 --- a/crates/tracedecay-agent-hosts/src/agents/claude/tests.rs +++ b/crates/tracedecay-agent-hosts/src/agents/claude/tests.rs @@ -1,4 +1,7 @@ -use super::super::{load_json_file_strict, safe_write_json_file, safe_write_text_file}; +use super::super::{ + load_json_file_strict, safe_write_json_file, safe_write_text_file, + unexpected_bundle_entrypoints, +}; use super::*; use serde_json::json; #[cfg(unix)] @@ -148,9 +151,46 @@ fn native_cache_content_drift_and_binary_relocation_require_refresh() { .unwrap() ); copy_rendered_bundle_to_native_cache(home.path(), new_bin); + let source = plugin_deploy_dir(home.path()); + let cache = claude_current_cached_plugin_root(home.path()); + let rendered = rendered_plugin_files(new_bin, &ServeArgs::default()).unwrap(); + let relatives = rendered + .iter() + .map(|(relative, _)| (*relative).to_owned()) + .collect::>(); + for (relative, expected) in rendered { + assert_eq!( + std::fs::read_to_string(source.join(relative)).unwrap(), + expected, + "relocated marketplace bytes differ at {relative}" + ); + assert_eq!( + std::fs::read_to_string(cache.join(relative)).unwrap(), + expected, + "refreshed native cache bytes differ at {relative}" + ); + } + for root in [&source, &cache] { + let unexpected = unexpected_bundle_entrypoints( + root, + &relatives, + &[".claude-plugin", "agents", "commands", "hooks", "skills"], + ) + .unwrap(); + assert!( + unexpected.is_empty(), + "refresh left auto-discovered entrypoints outside the catalog at {}: {unexpected:?}", + root.display() + ); + } + assert!( + claude_plugin_registration_is_active(home.path()).unwrap(), + "refresh must preserve the exact native plugin registration" + ); assert!( claude_plugin_is_natively_active(home.path(), Some((new_bin, &ServeArgs::default()))) - .unwrap() + .unwrap(), + "matching refreshed bytes must retain the exact discovery inventory" ); } diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/memory_tests.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/memory_tests.rs index 0ebe34ada8..62139a246d 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/memory_tests.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/memory_tests.rs @@ -73,6 +73,17 @@ fn expected_worker_reservation_on(runtime: &CodeIndexWorkerRuntimeV1, remaining_ ) } +/// Ledger scenarios own their measured input independently of sibling tests. +pub(super) fn ledger_memory(limit: NonZeroU64) -> Arc { + Arc::new(ProcessResidentMemoryV1::with_pressure( + limit, + Arc::new(ResidentMemoryPressureV1::with_sampler( + limit, + Arc::new(|| None), + )), + )) +} + #[test] fn captured_source_bytes_are_charged_until_the_snapshot_drops() { let project = fixture(); @@ -84,9 +95,7 @@ fn captured_source_bytes_are_charged_until_the_snapshot_drops() { Arc::new(SharedCodeIndexBytePoolV1::default()), ) .expect("open scheduler"); - let authority = Arc::new(ProcessResidentMemoryV1::new( - NonZeroU64::new(1024 * 1024).expect("source memory limit"), - )); + let authority = ledger_memory(NonZeroU64::new(1024 * 1024).expect("source memory limit")); scheduler.bind_resident_memory(Arc::clone(&authority)); let captured = scheduler @@ -119,9 +128,7 @@ fn reconcile_leaves_no_captured_source_charged_after_the_build() { Arc::new(SharedCodeIndexBytePoolV1::default()), ) .expect("open scheduler"); - let authority = Arc::new(ProcessResidentMemoryV1::new( - DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, - )); + let authority = ledger_memory(DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1); scheduler.bind_resident_memory(Arc::clone(&authority)); let captured = scheduler .capture_authoritative_snapshot(None) @@ -171,7 +178,7 @@ fn source_capture_refuses_before_build_when_retention_cannot_be_charged() { Arc::new(SharedCodeIndexBytePoolV1::default()), ) .expect("open scheduler"); - let authority = Arc::new(ProcessResidentMemoryV1::new(NonZeroU64::MIN)); + let authority = ledger_memory(NonZeroU64::MIN); scheduler.bind_resident_memory(Arc::clone(&authority)); assert!(matches!( @@ -291,9 +298,7 @@ fn worker_memory_reservation_is_charged_and_released_by_raii() { 2, DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1.get(), ); - let authority = Arc::new(ProcessResidentMemoryV1::new( - DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, - )); + let authority = ledger_memory(DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1); scheduler.bind_resident_memory(Arc::clone(&authority)); let reservation = scheduler .reserve_worker_memory() @@ -325,14 +330,7 @@ fn default_authority_worker_reserve_leaves_typed_snapshot_headroom() { let runtime = bind_automatic_worker_runtime(&scheduler, 128, 128 * 1024 * 1024 * 1024); // This case isolates the ledger's worker/snapshot split. Live process // pressure can narrow the slab and is exercised with explicit samples below. - let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( - DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, - Arc::new(|| None), - )); - let authority = Arc::new(ProcessResidentMemoryV1::with_pressure( - DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, - pressure, - )); + let authority = ledger_memory(DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1); scheduler.bind_resident_memory(Arc::clone(&authority)); let _worker = scheduler @@ -376,12 +374,12 @@ fn worker_memory_reservation_refusal_is_typed() { 2, DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1.get(), ); - let authority = Arc::new(ProcessResidentMemoryV1::new( + let authority = ledger_memory( NonZeroU64::new( tracedecay_code_index::parallelism::INDEX_WORKER_RESIDENT_BUDGET_BYTES_V1 - 1, ) .expect("positive test limit"), - )); + ); scheduler.bind_resident_memory(authority); assert!(matches!( scheduler.reserve_worker_memory(), diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/reconcile.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/reconcile.rs index 14e34cf58d..d8bbfe3f5e 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/reconcile.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/reconcile.rs @@ -37,6 +37,8 @@ use tracedecay_domain::{ }; use tracedecay_graph_db::GraphConflictContextV1; use tracedecay_privacy::CODE_SOURCE_SANITIZER_VERSION_V1; +#[cfg(any(test, feature = "test-helpers"))] +use tracedecay_runtime_core::resident_memory::ResidentMemoryPressureV1; use tracedecay_runtime_core::resident_memory::{ ProcessResidentMemoryV1, RESIDENT_MEMORY_PRESSURE_ADMISSION_FLOOR_BYTES_V1, RESIDENT_OWNER_IDLE_WINDOW_V1, ResidentMemoryAdmissionFailureV1, ResidentMemoryComponentIdV1, @@ -1250,7 +1252,9 @@ impl HistoricalCodeIndexGenerationOwnerV1 { } impl CodeIndexWorktreeSchedulerV1 { - /// A standalone owner under the path policy a fresh profile resolves. + /// A standalone fixture owner under the path policy a fresh profile resolves. + /// Concurrent cases do not contribute measured pressure to this owner; + /// pressure scenarios bind their own explicit readings after opening it. #[cfg(any(test, feature = "test-helpers"))] pub fn open( project_id: ProjectId, @@ -1258,14 +1262,23 @@ impl CodeIndexWorktreeSchedulerV1 { store_root: PathBuf, byte_pool: Arc, ) -> Result { - Self::open_with_policy( + let mut scheduler = Self::open_with_policy( project_id, project_root, store_root, byte_pool, CodeIndexHintPolicyV1::default(), crate::config::registry_default_index_path_policy(), - ) + )?; + let limit = detected_process_resident_memory_limit_v1(); + scheduler.bind_resident_memory(Arc::new(ProcessResidentMemoryV1::with_pressure( + limit, + Arc::new(ResidentMemoryPressureV1::with_sampler( + limit, + Arc::new(|| None), + )), + ))); + Ok(scheduler) } pub fn open_with_policy( diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/resident_memory.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/resident_memory.rs index b971b0dc1a..486d42d8fb 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/resident_memory.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/resident_memory.rs @@ -2,11 +2,12 @@ use std::sync::Arc; -#[cfg(any(test, feature = "test-helpers"))] -use tracedecay_runtime_core::resident_memory::DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1; pub use tracedecay_runtime_core::resident_memory::ProcessResidentMemoryV1; #[cfg(any(test, feature = "test-helpers"))] -use tracedecay_runtime_core::resident_memory::RESIDENT_OWNER_IDLE_WINDOW_V1; +use tracedecay_runtime_core::resident_memory::{ + DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, RESIDENT_OWNER_IDLE_WINDOW_V1, + ResidentMemoryPressureV1, +}; use tracedecay_runtime_core::resident_memory::{ResidentOwnersV1, process_resident_owners_v1}; use super::CodeIndexSchedulerRegistryV1; @@ -15,10 +16,17 @@ impl CodeIndexSchedulerRegistryV1 { #[cfg(any(test, feature = "test-helpers"))] #[cfg_attr(not(test), allow(dead_code))] pub fn new(max_worktrees: usize) -> Self { + // A fixture owns its ledger and pressure readings. Concurrent tests' + // RSS is not an input; pressure journeys pass an explicit authority. + let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( + DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, + Arc::new(|| None), + )); Self::with_resident_memory_and_progress_producer_incarnation( max_worktrees, - Arc::new(ProcessResidentMemoryV1::new( + Arc::new(ProcessResidentMemoryV1::with_pressure( DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, + pressure, )), 1, ) 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 e6a5be1c4a..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,7 +2796,13 @@ impl LatestCodeTextGenerationV1 { maximum_work: usize, control: &dyn CodeIndexExecutionControlV1, ) -> Result { - let _workers = self + 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 .enter(&self.text_artifact_store.resident_memory) @@ -2774,8 +2815,23 @@ 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) => { + 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) + } + }; if result.as_ref().is_err_and(|error| { matches!( error, @@ -3297,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) => { @@ -3588,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/src/code_index_scheduler/tests/residency.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs index 4cd725b7cb..875d53625d 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs @@ -1476,14 +1476,6 @@ fn ledger_leaving_one_mib( .expect("the held memory fits the ledger") } -fn sixteen_gib_authority() -> Arc { - let limit = NonZeroU64::new(16 * 1024 * 1024 * 1024).unwrap(); - Arc::new(ProcessResidentMemoryV1::with_pressure( - limit, - Arc::new(ResidentMemoryPressureV1::new(limit)), - )) -} - /// A 16 GiB authority reading a fixed process size, so a held ledger refuses /// a refresh however much memory the tests beside it hold. fn fixed_reading_authority() -> Arc { @@ -1510,7 +1502,7 @@ async fn a_refresh_refused_for_memory_publishes_when_an_idle_owner_releases_it() let fixture = GitFixture::new(&[("src/main.rs", "fn main() {}\n")]); let store = TempDir::new().expect("store root"); let owners = Arc::new(ResidentOwnersV1::new(IDLE_WINDOW)); - let resident_memory = sixteen_gib_authority(); + let resident_memory = fixed_reading_authority(); let (registry, _) = mounted_core_query_worktree_in( CodeIndexSchedulerRegistryV1::with_resident_memory(1, Arc::clone(&resident_memory)) .with_resident_owners(Arc::clone(&owners)), @@ -1576,7 +1568,7 @@ async fn a_refresh_refused_by_the_ledger_publishes_when_the_blocking_reservation let fixture = GitFixture::new(&[("src/main.rs", "fn main() {}\n")]); let store = TempDir::new().expect("store root"); let owners = Arc::new(ResidentOwnersV1::new(IDLE_WINDOW)); - let resident_memory = sixteen_gib_authority(); + let resident_memory = fixed_reading_authority(); let (registry, _) = mounted_core_query_worktree_in( CodeIndexSchedulerRegistryV1::with_resident_memory(1, Arc::clone(&resident_memory)) .with_resident_owners(Arc::clone(&owners)), diff --git a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/reconcile.rs b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/reconcile.rs index 30f6f177a3..182c11a2ba 100644 --- a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/reconcile.rs +++ b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/reconcile.rs @@ -30,9 +30,11 @@ use tracedecay_domain::{ use tracedecay_runtime_core::resident_memory::{ DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, ProcessResidentMemoryV1, ResidentHoldingV1, ResidentOwnerBytesV1, ResidentOwnerKindV1, ResidentOwnerReleaseV1, ResidentOwnerSampleV1, - ResidentOwnerScopeV1, ResidentOwnerV1, ResidentOwnersV1, sampled_process_resident_bytes_v1, + ResidentOwnerScopeV1, ResidentOwnerV1, ResidentOwnersV1, + detected_process_resident_memory_limit_v1, sampled_process_resident_bytes_v1, }; +use super::super::memory_tests::ledger_memory; #[cfg(unix)] use super::stage_raw_git_path; use super::{ @@ -1437,12 +1439,17 @@ fn retained_decode_incremental_seal_and_text_publish_stay_below_the_high_waterma let fixture = GitFixture::new(&sources); let store = TempDir::new().expect("store root"); + // This scale journey measures real process pressure through admission + // as well as its RSS assertions, so retain the production pressure cell. + let memory_limit = detected_process_resident_memory_limit_v1(); + { let mut seed = scheduler( &fixture, store.path().to_path_buf(), Arc::new(SharedCodeIndexBytePoolV1::default()), ); + seed.bind_resident_memory(Arc::new(ProcessResidentMemoryV1::new(memory_limit))); published(seed.reconcile_now().expect("seed retained generation")); } @@ -1455,6 +1462,7 @@ fn retained_decode_incremental_seal_and_text_publish_stay_below_the_high_waterma store.path().to_path_buf(), Arc::new(SharedCodeIndexBytePoolV1::default()), ); + scheduler.bind_resident_memory(Arc::new(ProcessResidentMemoryV1::new(memory_limit))); assert_eq!(scheduler.publication.sealed_decode_count(), 0); assert!(matches!( scheduler @@ -11410,7 +11418,7 @@ fn graph_off_changed_source_worker_memory_denial_retries_without_decode() { ) > tight_limit.get(), "the fixture limit must deny the installed worker plan" ); - let tight = Arc::new(ProcessResidentMemoryV1::new(tight_limit)); + let tight = ledger_memory(tight_limit); scheduler.bind_resident_memory(Arc::clone(&tight)); let denied = scheduler.reconcile_retained_text_generation_with(&metadata_a, true); assert!( @@ -11435,9 +11443,7 @@ fn graph_off_changed_source_worker_memory_denial_retries_without_decode() { "a denied worker reservation must not leak a charge" ); - let adequate = Arc::new(ProcessResidentMemoryV1::new( - DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, - )); + let adequate = ledger_memory(DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1); scheduler.bind_resident_memory(Arc::clone(&adequate)); let outcome = scheduler .reconcile_retained_text_generation_with(&metadata_a, 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 64b6aab9f1..00751d63f6 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,11 @@ use std::{ fmt::Write as _, num::NonZeroU64, path::{Path, PathBuf}, - sync::{Arc, mpsc}, + sync::{ + Arc, Mutex, RwLock, + atomic::{AtomicBool, Ordering}, + mpsc, + }, thread, time::{Duration, Instant}, }; @@ -20,6 +24,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, @@ -56,6 +61,7 @@ use tracedecay_runtime_core::resident_memory::{ use tracedecay_session_temporal_store::SessionTemporalAccess; +use super::super::memory_tests::ledger_memory; use super::{ ALPHA_LIB_V1, CALLER_PAGE, CALLER_STAR, GitFixture, OwnerSignals, ReadyRetrievalControlV1, SERVING_SEAT_FAILURE_CEILING, active_text_artifact_path, application_context, @@ -2068,9 +2074,9 @@ fn reader_reservation_refusal_precedes_missing_artifact_access() { store.path().to_path_buf(), Arc::new(SharedCodeIndexBytePoolV1::default()), ); - scheduler.bind_resident_memory(Arc::new(ProcessResidentMemoryV1::new( + scheduler.bind_resident_memory(ledger_memory( std::num::NonZeroU64::new(1024 * 1024).expect("tight memory limit"), - ))); + )); let latest = scheduler.latest_complete().expect("restored generation"); assert!( matches!( @@ -2135,6 +2141,263 @@ fn overlapping_text_builds_share_one_admission_watermark_headroom() { assert_eq!(authority.snapshot().used_bytes, 0); } +#[test] +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()), + ); + scheduler.bind_resident_memory(Arc::clone(&memory)); + published(scheduler.reconcile_now().expect("seal source generation")); + } + let mut scheduler = scheduler( + &fixture, + store.path().to_path_buf(), + Arc::new(SharedCodeIndexBytePoolV1::default()), + ); + 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 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(_)) + )); + 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()); + 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] +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 concurrent_text_admission_refusal_cannot_block_a_ready_owner() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn concurrent_text_owner() {}\n")]); + let store = TempDir::new().expect("store root"); + let limit = NonZeroU64::new(4 * 1024 * 1024 * 1024).expect("process limit"); + let (sampling, sampled) = mpsc::channel(); + let (resume, resumed) = mpsc::channel(); + let resumed = Mutex::new(resumed); + let paused = AtomicBool::new(false); + let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( + limit, + Arc::new(move || { + let refused = thread::current().name() == Some("refused-text-advance"); + if refused && !paused.swap(true, Ordering::AcqRel) { + sampling.send(()).expect("announce refused sample"); + resumed + .lock() + .expect("sample resume receiver") + .recv_timeout(SERVING_SEAT_FAILURE_CEILING) + .expect("resume refused sample"); + } + Some(ProcessResidentSampleV1 { + resident_bytes: 0, + unreclaimable_bytes: 0, + swapped_bytes: 0, + cgroup_headroom_bytes: refused.then_some(0), + }) + }), + )); + let memory = Arc::new(ProcessResidentMemoryV1::with_pressure(limit, pressure)); + let mut scheduler = scheduler( + &fixture, + store.path().to_path_buf(), + Arc::new(SharedCodeIndexBytePoolV1::default()), + ); + scheduler.bind_resident_memory(memory); + published(scheduler.reconcile_now().expect("seal source generation")); + let text = scheduler.latest_complete().expect("text generation"); + thread::scope(|scope| { + let refused_text = text.clone(); + let refused = thread::Builder::new() + .name("refused-text-advance".to_owned()) + .spawn_scoped(scope, move || refused_text.advance_text_serving(1)) + .expect("spawn refused advance"); + sampled + .recv_timeout(SERVING_SEAT_FAILURE_CEILING) + .expect("first advance reaches memory admission"); + let ready_text = text.clone(); + let (starting, started) = mpsc::channel(); + let ready = scope.spawn(move || { + starting.send(()).expect("announce concurrent advance"); + ready_text.production_query_owners() + }); + started + .recv_timeout(SERVING_SEAT_FAILURE_CEILING) + .expect("second advance starts while the first is admitted"); + resume.send(()).expect("release refused sample"); + assert!(matches!( + refused.join().expect("refused advance joins"), + Err(RetrievalPortError::ResidentMemoryRefused(_)) + )); + ready + .join() + .expect("ready advance joins") + .expect("concurrent owner serves after the refusal"); + }); + let ready = build_progress_snapshot(&scheduler); + assert!(text.query_owners_are_ready()); + assert_eq!(ready.phase, CodeIndexBuildPhaseV1::Ready); + assert_eq!(ready.blocked_reason, None); +} + #[test] fn text_build_budget_shrinks_to_available_headroom_without_dropping_below_its_floor() { const GIB: u64 = 1024 * 1024 * 1024; @@ -2179,9 +2442,7 @@ fn text_artifact_ceilings_reserve_through_process_resident_memory() { Arc::new(SharedCodeIndexBytePoolV1::default()), ); published(scheduler.reconcile_now().expect("publish")); - let tight = Arc::new(ProcessResidentMemoryV1::new( - std::num::NonZeroU64::new(1024 * 1024).expect("tight limit"), - )); + let tight = ledger_memory(std::num::NonZeroU64::new(1024 * 1024).expect("tight limit")); scheduler.bind_resident_memory(Arc::clone(&tight)); let latest = scheduler.latest_complete().expect("latest generation"); let denied = latest.advance_text_serving(usize::MAX); @@ -2268,9 +2529,7 @@ fn text_artifact_ceilings_reserve_through_process_resident_memory() { // An adequate authority admits the build and holds the reader charge // for as long as the artifact owners serve. - let adequate = Arc::new(ProcessResidentMemoryV1::new( - DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, - )); + let adequate = ledger_memory(DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1); { // A fresh scheduler restores the durable generation; the unchanged // fixture needs no republish. @@ -2778,6 +3037,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 3a570a073b0c860722a2336b3eed9c43b73d61b0 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Wed, 21 Oct 2026 17:23:26 -0700 Subject: [PATCH 9/9] test(sessions): retain rekey completion checks after merge --- .../registry_tests.rs | 74 +++++++++++++------ 1 file changed, 51 insertions(+), 23 deletions(-) diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry_tests.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry_tests.rs index ac150a4d45..d52c9c9eb7 100644 --- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry_tests.rs +++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry_tests.rs @@ -4,17 +4,23 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::time::Duration; use futures_util::FutureExt; +use tracedecay_domain::SessionId; use tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness; use tracedecay_runtime_core::native_runtime; +use tracedecay_session_temporal_store::SessionTemporalStore; use tracedecay_sessions::serving::{ SessionProjectionServingState, SessionProjectionServingStatusPort, SessionProjectionStaleReason, }; +use tracedecay_store::{ + SessionRefreshBeginOrJoinRequestV1, SessionRefreshDispositionV1, SessionRefreshFrontierV1, + SessionRefreshReceiptRequestV1, SessionRefreshStore, SessionRefreshTerminalStateV1, +}; use super::super::history::{ SessionHistoricalCapacityRelease, SessionHistoricalIngestOutcome, SessionHistoricalIngestPass, SessionHistoricalIngestor, SharedSessionHistoricalIngestor, }; -use super::super::wake::SessionTemporalRefreshWakeState; +use super::super::wake::{SessionTemporalRefreshWake, SessionTemporalRefreshWakeState}; use super::SessionTemporalRefreshSchedulerRegistry; use crate::StoreOwnerKey; @@ -137,6 +143,46 @@ async fn mount_held_project( history } +/// A replacement worker may consume a handed-off request before rekey returns. +/// Join its real pass and require the durable completion, rather than racing +/// the consumer for a transient queue length or dirty flag. These empty sessions +/// have no observation history for discovery to reconstruct a lost request from. +async fn assert_rekey_completed_requests( + harness: &RegisteredGlobalDbHarness, + wake: &SessionTemporalRefreshWake, + sessions: &[&str], +) { + assert!( + wake.wake_and_wait_until_idle(Duration::from_secs(5)).await, + "the retained route must join the replacement worker's pass" + ); + let store = SessionTemporalStore::new(harness.registered.as_ref()); + for name in sessions { + let session_id = SessionId::new(*name).expect("session id"); + let joined = store + .begin_or_join_session_refresh(SessionRefreshBeginOrJoinRequestV1::new( + session_id.clone(), + SessionRefreshFrontierV1::new(0, 0).expect("empty frontier"), + )) + .await + .expect("join the handed-off refresh"); + assert_eq!( + joined.disposition(), + SessionRefreshDispositionV1::Joined, + "{name}: the replacement must have begun the handed-off request" + ); + let receipt = store + .session_refresh_receipt(SessionRefreshReceiptRequestV1::new( + joined.operation_id().clone(), + session_id, + )) + .await + .expect("read handed-off refresh receipt") + .expect("the replacement must complete the handed-off request"); + assert_eq!(receipt.state(), SessionRefreshTerminalStateV1::Complete); + } +} + /// Mounts a project whose worker holds its history pass and a profile entry. async fn mount_drain_fixture( harness: &RegisteredGlobalDbHarness, @@ -410,20 +456,11 @@ async fn a_retried_rekey_recovers_the_dropped_attempts_route_and_requests() { rekey.await; assert!(registry.project_state(&old_owner).await.is_none()); - let new_state = registry + registry .project_state(&new_owner) .await .expect("new owner entry"); - assert_eq!( - new_state.queued_request_count(), - 1, - "the retry must land the retired owner's queued request on the new owner" - ); - assert!( - old_wake.wake(), - "the old route must reach the new owner once the rekey completes" - ); - assert!(new_state.take_dirty()); + assert_rekey_completed_requests(&harness, &old_wake, &["session.dropped-rekey"]).await; registry.shutdown().await; } @@ -787,20 +824,11 @@ async fn a_mapped_rekey_lands_the_workers_queued_requests() { rekey.await; assert!(registry.project_state(&old_owner).await.is_none()); - let new_state = registry + registry .project_state(&new_owner) .await .expect("new owner entry"); - assert_eq!( - new_state.queued_request_count(), - 1, - "the new owner must inherit the mapped worker's queued request" - ); - assert!( - old_wake.wake(), - "the old route must reach the new owner once the rekey completes" - ); - assert!(new_state.take_dirty()); + assert_rekey_completed_requests(&harness, &old_wake, &["session.mapped-rekey"]).await; registry.shutdown().await; }