From a9b07ead3fee2b7926cf3366e5ea2c6e1daf01ae Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 19 Sep 2026 06:57:07 +0000 Subject: [PATCH 1/4] fix(code-index): seat graph while text is projecting PR dogfood on 3f71bccb2dba timed out in bulk_commit with graph pending. The first pass sat unscheduled for 226s on the daemon runtime, then the published text projection was joined before any graph attempt. Run workers on their own runtime and activate the graph during that projection. Shrink the build charge into the reader instead of dropping it, so an overlapping replay cannot refuse the reader admission. Co-authored-by: Zack Jackson --- .../code_index_scheduler/registry/mount.rs | 152 ++++++++++++------ .../registry/test_gates.rs | 24 +++ .../src/code_index_scheduler/serving.rs | 27 ++-- .../code_index_scheduler/tests/reconcile.rs | 51 ++++++ 4 files changed, 193 insertions(+), 61 deletions(-) diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs index ac239ce49c..a5df91a6c1 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs @@ -38,6 +38,39 @@ use super::{ publication_authority_is_terminal, retained_noop_requires_follow_up_wake, }; +/// Runtime that polls code-index workers, separate from the daemon's serving +/// runtime. +/// +/// The first reconcile of a cold checkout is the strict-readiness critical +/// path. Sharing the daemon runtime left that worker unpolled for minutes +/// while project-open tasks occupied every serving thread: the pass log then +/// showed a 226s queue delay with the gates themselves taking 15µs, and the +/// dogfood deadline expired still inside text projection, graph never seated. +/// A process-wide runtime keeps indexing scheduled as soon as the wake is +/// posted. The future is already boxed, so the stack only has to poll it. +fn code_index_worker_runtime() -> Result<&'static tokio::runtime::Runtime, CodeIndexSchedulerErrorV1> +{ + static RUNTIME: OnceLock> = OnceLock::new(); + match RUNTIME.get_or_init(|| { + let workers = std::thread::available_parallelism() + .map_or(2, usize::from) + .clamp(2, 4); + tokio::runtime::Builder::new_multi_thread() + .thread_name("td-code-index") + .worker_threads(workers) + .max_blocking_threads(workers.saturating_add(8)) + .thread_stack_size(8 * 1024 * 1024) + .enable_all() + .build() + .map_err(|error| error.to_string()) + }) { + Ok(runtime) => Ok(runtime), + Err(error) => Err(CodeIndexSchedulerErrorV1::Identity(format!( + "code-index worker runtime failed to start: {error}" + ))), + } +} + impl CodeIndexSchedulerRegistryV1 { #[cfg(test)] pub fn open_worktree( @@ -1049,13 +1082,17 @@ impl CodeIndexSchedulerRegistryV1 { Ok(Ok(CodeIndexReconcileOutcomeV1::Published(_))) ); let mut graph_text = retained_text.clone(); - // The replacement owner's bounded projection must finish - // before a fresh graph publication starts. Both consume the - // same sealed generation and are corpus-sized: overlapping - // them lets graph replay hold the source while text opens it, - // then leaves text unable to reacquire its reservation after - // graph publication reaches the process RSS watermark. + // Text projection and graph activation both read the sealed + // generation. They run together: awaiting the whole artifact + // (bulk commit, then an undivided ngram index) before the + // first graph attempt is how strict readiness timed out still + // in `bulk_commit` with `graph=pending`. The text build keeps + // its reservation for the whole projection and shrinks that + // same charge into the reader, so graph holding RSS cannot + // open a gap where the reader admission is refused. let mut published_text_projection_outcome = None; + let mut published_text_projection = None; + let mut text_projection_in_flight = false; if published_pass { *worker_text_generation .write() @@ -1097,48 +1134,31 @@ impl CodeIndexSchedulerRegistryV1 { } Ok(Ok(Ok(None)) | Err(_)) | Err(_) => None, }; - // Finish the replacement text owner before optional - // O(store) graph work below. Exact and lexical are the - // required fresh-index product; graph activation is an - // optional projection and must not consume the source or - // resident-memory headroom needed to build them. - // Yielding back to the loop instead would hand the next - // pass a checkout that has already moved, and on a shared - // repository that pass publishes again - which is exactly - // how a sealed generation stayed unseated forever. + // Start the replacement text owner, but do not join it + // before graph activation. Exact and lexical still have to + // finish before the serving swap (the source proof is + // renewed from the projection outcome below). Graph + // activation reads the sealed generation directly and + // shares that window instead of starting only after the + // artifact is ready. Yielding back to the loop instead + // would hand the next pass a checkout that has already + // moved, and on a shared repository that pass publishes + // again - which is exactly how a sealed generation stayed + // unseated forever. if graph_activation_enabled && !graph_activation_deferred && let Some(text) = graph_text.clone() { - let projection = tokio::spawn(Self::drive_text_projection( - text, - Arc::clone(&worker_shutting_down), - Arc::clone(&worker_convergence_park), - None, - #[cfg(test)] - worker_project_root.clone(), - )); - published_text_projection_outcome = Some(match projection.await { - Ok(outcome) => outcome, - Err(error) => { - if let Some(text) = graph_text.as_ref() { - text.mark_text_serving_failed(); - } - park_convergence( - &worker_convergence_park, - format!("code text projection task failed abnormally: {error}"), - CONVERGENCE_PARK_TASK_FAILURE_REMEDIATION_V1, - None, - false, - ); - tracing::warn!( - event = "code_index_text_projection_task_failed", - error = %error, - "published text projection task failed before graph seating" - ); - PublishedTextProjectionOutcomeV1::Unfinished - } - }); + published_text_projection = + Some(tokio::spawn(Self::drive_text_projection( + text, + Arc::clone(&worker_shutting_down), + Arc::clone(&worker_convergence_park), + None, + #[cfg(test)] + worker_project_root.clone(), + ))); + text_projection_in_flight = true; } else if graph_text .as_ref() .is_none_or(LatestCodeTextGenerationV1::text_projection_needs_work) @@ -1178,7 +1198,8 @@ impl CodeIndexSchedulerRegistryV1 { matches!(&source_result, Ok(Ok(_))), published_pass, if published_pass { - exact_and_lexical_ready_for_graph(graph_text.as_ref()) + text_projection_in_flight + || exact_and_lexical_ready_for_graph(graph_text.as_ref()) } else { graph_text.is_some() }, @@ -1436,12 +1457,14 @@ impl CodeIndexSchedulerRegistryV1 { // verified graph head, which reads the durable graph store // instead of replaying the sealed source. If recovery did not // satisfy this pass, wait for text before the corpus-sized - // decode and replay below. Otherwise a failed fresh text pass - // becomes a retained pass on its next wake and recreates the - // same source and resident-memory contention we avoid above. - // Same named predicate as the published seat gate above. + // decode and replay below. A publication's text projection is + // already in flight and keeps its build reservation, so its + // graph replay shares that window instead of waiting for the + // artifact. A later retained pass still waits, or a failed + // fresh text pass would recreate the source contention. if prepare_graph && !graph_already_serves + && !text_projection_in_flight && !exact_and_lexical_ready_for_graph(graph_text.as_ref()) { prepare_graph = false; @@ -1695,9 +1718,34 @@ impl CodeIndexSchedulerRegistryV1 { } } } + // Graph activation has had the projection window. Join the + // text owner before the source-proof renewal and the serving + // swap: those still need the projection's terminal outcome. + if let Some(projection) = published_text_projection.take() { + published_text_projection_outcome = Some(match projection.await { + Ok(outcome) => outcome, + Err(error) => { + if let Some(text) = graph_text.as_ref() { + text.mark_text_serving_failed(); + } + park_convergence( + &worker_convergence_park, + format!("code text projection task failed abnormally: {error}"), + CONVERGENCE_PARK_TASK_FAILURE_REMEDIATION_V1, + None, + false, + ); + tracing::warn!( + event = "code_index_text_projection_task_failed", + error = %error, + "published text projection task failed before graph seating" + ); + PublishedTextProjectionOutcomeV1::Unfinished + } + }); + } // Process the fresh text outcome at the existing source-proof - // and serving-swap boundary. Graph work above ran only when - // the outcome was ready. + // and serving-swap boundary. if let Some(outcome) = published_text_projection_outcome.take() { match outcome { PublishedTextProjectionOutcomeV1::Finished => { @@ -2332,7 +2380,7 @@ impl CodeIndexSchedulerRegistryV1 { let _ = result; } }); - let task = tokio::spawn(hotpath::future!( + let task = code_index_worker_runtime()?.spawn(hotpath::future!( worker_loop, label = "daemon.code_index.scheduler_worker" )); diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/test_gates.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/test_gates.rs index 102974e294..6fe904126b 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/test_gates.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/test_gates.rs @@ -60,6 +60,30 @@ impl CodeIndexSchedulerRegistryV1 { } } + /// Graph-serving state of the mounted text owner, the same projection + /// status reads. `None` means the worktree is unmounted or has no text + /// owner yet. + #[cfg(test)] + pub async fn code_graph_serving_readiness_for_test( + &self, + project_root: &Path, + ) -> Option { + let Ok(project_root) = project_root.canonicalize() else { + return None; + }; + let text = { + let mounted = self.mounted.lock().await; + let Some(worktree) = mounted.get(&project_root) else { + return None; + }; + Arc::clone(&worktree.text_generation) + }; + text.read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .as_ref() + .map(|text| text.code_graph_serving_readiness()) + } + /// Test-only observation of an exact mounted worktree's active owner pass. #[cfg(test)] pub async fn reconcile_in_progress_for_test(&self, project_root: &Path) -> bool { 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 394cc15419..cb4a732c33 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 @@ -3292,15 +3292,24 @@ impl LatestCodeTextGenerationV1 { &sealed_identity, control, )?; - // The builder and source are gone, so its transient reservation no - // longer owns bytes. Release it before sampling the reader admission; - // the reader guard then carries the still-live unmodeled baseline. - drop(build_reservation); - let reader_reservation = store.reserve_resident_memory( - &self.metadata.manifest().generation_id, - "code-text-artifact-reader", - CODE_LEXICAL_ARTIFACT_QUERY_CACHE_BUDGET_BYTES_V1, - )?; + // Keep the build charge and shrink it to the reader budget. Dropping + // it and reserving again is a gap: a graph replay that overlapped this + // projection can be sitting on the process RSS watermark, and the new + // admission is then refused forever. The reader is a smaller charge + // of the same generation, so it stays inside the bytes already held. + let reader_budget = u64::try_from(CODE_LEXICAL_ARTIFACT_QUERY_CACHE_BUDGET_BYTES_V1) + .map_err(|_| { + RetrievalPortError::Contract("text-artifact reader budget exceeds u64".to_owned()) + })?; + let mut reader_reservation = build_reservation; + if reader_reservation.reserved_bytes() > reader_budget { + reader_reservation.shrink_to(reader_budget).map_err(|error| { + RetrievalPortError::Contract(format!( + "text-artifact reader charge could not shrink inside its build reservation: reserved {} measured {}", + error.reserved_bytes, error.measured_bytes + )) + })?; + } let final_path = code_text_artifact_path(store.store_root(), &descriptor) .map_err(text_artifact_unavailable)?; let reader = CodeLexicalArtifactReaderV1::open_content_addressed( diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs index 188d869742..5b39b0f3ca 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs @@ -2628,6 +2628,57 @@ async fn graph_read_during_reconcile_records_a_busy_follow_up() { registry.shutdown().await; } +/// Strict readiness needs a ready graph and a finished text artifact. Awaiting +/// the artifact before the first graph attempt leaves the graph pending for +/// the whole build, which is the PR-dogfood timeout (still in bulk commit, +/// graph pending, at the deadline). Graph activation has to move while the +/// published text projection is held. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn graph_activation_runs_while_published_text_projection_is_held() { + let fixture = GitFixture::new(ALPHA_LIB_V1); + let store = TempDir::new().expect("store root"); + let registry = CodeIndexSchedulerRegistryV1::with_background_reconcile_permits(1, 1); + let canonical_root = fixture.path().canonicalize().expect("canonical fixture"); + let (projection_started, release_projection) = registry + .pause_next_published_text_projection(canonical_root) + .await; + registry + .mount_worktree( + test_project_id(), + fixture.path(), + store.path().to_path_buf(), + ) + .await + .expect("mount worktree"); + tokio::time::timeout(Duration::from_secs(30), projection_started) + .await + .expect("publication did not reach text projection") + .expect("projection gate stays armed"); + + let graph_ready = tokio::time::timeout(Duration::from_secs(30), async { + loop { + if matches!( + registry + .code_graph_serving_readiness_for_test(fixture.path()) + .await, + Some( + tracedecay_contracts::code_index_freshness::CodeGraphServingReadinessV1::Ready + ) + ) { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await; + release_projection + .send(()) + .expect("release publication projection"); + graph_ready + .expect("graph serving becomes ready while the published text projection is still held"); + registry.shutdown().await; +} + /// A publication can finish source capture long before its text artifact is /// ready. The serving swap must reverify after that projection, otherwise the /// exact active generation seats after its bounded proof expires and every From 946ad3bce053612dbbf29920f014c364c336929b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 19 Sep 2026 12:19:29 +0000 Subject: [PATCH 2/4] fix(code-index): keep the reader charge on both tails The published-projection overlap treated an in-flight text build as graph admission. Activation then started before exact and lexical owners were ready, including while a parked owner still held the source. Those convergence tests fail, so the seat order stays. Both publication tails hand their held charge to the reader component instead of dropping it and reserving again. Measured RSS cannot refuse a charge that never left the ledger. The clone successor reserves at least the reader budget so that handoff does not under-charge the reader. Workers stay on their own runtime so the first pass is not stuck behind project-open. Co-authored-by: Zack Jackson --- .../code_index_scheduler/registry/mount.rs | 117 ++++++++---------- .../registry/test_gates.rs | 24 ---- .../src/code_index_scheduler/serving.rs | 74 +++++++---- .../code_index_scheduler/tests/reconcile.rs | 51 -------- .../src/resident_memory.rs | 74 +++++++++++ .../src/resident_memory/tests.rs | 51 ++++++++ 6 files changed, 225 insertions(+), 166 deletions(-) diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs index a5df91a6c1..179c2103c3 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs @@ -1082,17 +1082,13 @@ impl CodeIndexSchedulerRegistryV1 { Ok(Ok(CodeIndexReconcileOutcomeV1::Published(_))) ); let mut graph_text = retained_text.clone(); - // Text projection and graph activation both read the sealed - // generation. They run together: awaiting the whole artifact - // (bulk commit, then an undivided ngram index) before the - // first graph attempt is how strict readiness timed out still - // in `bulk_commit` with `graph=pending`. The text build keeps - // its reservation for the whole projection and shrinks that - // same charge into the reader, so graph holding RSS cannot - // open a gap where the reader admission is refused. + // The replacement owner's bounded projection must finish + // before a fresh graph publication starts. Both consume the + // same sealed generation and are corpus-sized: overlapping + // them lets graph replay hold the source while text opens it, + // then leaves text unable to reacquire its reservation after + // graph publication reaches the process RSS watermark. let mut published_text_projection_outcome = None; - let mut published_text_projection = None; - let mut text_projection_in_flight = false; if published_pass { *worker_text_generation .write() @@ -1134,31 +1130,48 @@ impl CodeIndexSchedulerRegistryV1 { } Ok(Ok(Ok(None)) | Err(_)) | Err(_) => None, }; - // Start the replacement text owner, but do not join it - // before graph activation. Exact and lexical still have to - // finish before the serving swap (the source proof is - // renewed from the projection outcome below). Graph - // activation reads the sealed generation directly and - // shares that window instead of starting only after the - // artifact is ready. Yielding back to the loop instead - // would hand the next pass a checkout that has already - // moved, and on a shared repository that pass publishes - // again - which is exactly how a sealed generation stayed - // unseated forever. + // Finish the replacement text owner before optional + // O(store) graph work below. Exact and lexical are the + // required fresh-index product; graph activation is an + // optional projection and must not consume the source or + // resident-memory headroom needed to build them. + // Yielding back to the loop instead would hand the next + // pass a checkout that has already moved, and on a shared + // repository that pass publishes again - which is exactly + // how a sealed generation stayed unseated forever. if graph_activation_enabled && !graph_activation_deferred && let Some(text) = graph_text.clone() { - published_text_projection = - Some(tokio::spawn(Self::drive_text_projection( - text, - Arc::clone(&worker_shutting_down), - Arc::clone(&worker_convergence_park), - None, - #[cfg(test)] - worker_project_root.clone(), - ))); - text_projection_in_flight = true; + let projection = tokio::spawn(Self::drive_text_projection( + text, + Arc::clone(&worker_shutting_down), + Arc::clone(&worker_convergence_park), + None, + #[cfg(test)] + worker_project_root.clone(), + )); + published_text_projection_outcome = Some(match projection.await { + Ok(outcome) => outcome, + Err(error) => { + if let Some(text) = graph_text.as_ref() { + text.mark_text_serving_failed(); + } + park_convergence( + &worker_convergence_park, + format!("code text projection task failed abnormally: {error}"), + CONVERGENCE_PARK_TASK_FAILURE_REMEDIATION_V1, + None, + false, + ); + tracing::warn!( + event = "code_index_text_projection_task_failed", + error = %error, + "published text projection task failed before graph seating" + ); + PublishedTextProjectionOutcomeV1::Unfinished + } + }); } else if graph_text .as_ref() .is_none_or(LatestCodeTextGenerationV1::text_projection_needs_work) @@ -1198,8 +1211,7 @@ impl CodeIndexSchedulerRegistryV1 { matches!(&source_result, Ok(Ok(_))), published_pass, if published_pass { - text_projection_in_flight - || exact_and_lexical_ready_for_graph(graph_text.as_ref()) + exact_and_lexical_ready_for_graph(graph_text.as_ref()) } else { graph_text.is_some() }, @@ -1457,14 +1469,12 @@ impl CodeIndexSchedulerRegistryV1 { // verified graph head, which reads the durable graph store // instead of replaying the sealed source. If recovery did not // satisfy this pass, wait for text before the corpus-sized - // decode and replay below. A publication's text projection is - // already in flight and keeps its build reservation, so its - // graph replay shares that window instead of waiting for the - // artifact. A later retained pass still waits, or a failed - // fresh text pass would recreate the source contention. + // decode and replay below. Otherwise a failed fresh text pass + // becomes a retained pass on its next wake and recreates the + // same source and resident-memory contention we avoid above. + // Same named predicate as the published seat gate above. if prepare_graph && !graph_already_serves - && !text_projection_in_flight && !exact_and_lexical_ready_for_graph(graph_text.as_ref()) { prepare_graph = false; @@ -1718,34 +1728,9 @@ impl CodeIndexSchedulerRegistryV1 { } } } - // Graph activation has had the projection window. Join the - // text owner before the source-proof renewal and the serving - // swap: those still need the projection's terminal outcome. - if let Some(projection) = published_text_projection.take() { - published_text_projection_outcome = Some(match projection.await { - Ok(outcome) => outcome, - Err(error) => { - if let Some(text) = graph_text.as_ref() { - text.mark_text_serving_failed(); - } - park_convergence( - &worker_convergence_park, - format!("code text projection task failed abnormally: {error}"), - CONVERGENCE_PARK_TASK_FAILURE_REMEDIATION_V1, - None, - false, - ); - tracing::warn!( - event = "code_index_text_projection_task_failed", - error = %error, - "published text projection task failed before graph seating" - ); - PublishedTextProjectionOutcomeV1::Unfinished - } - }); - } // Process the fresh text outcome at the existing source-proof - // and serving-swap boundary. + // and serving-swap boundary. Graph work above ran only when + // the outcome was ready. if let Some(outcome) = published_text_projection_outcome.take() { match outcome { PublishedTextProjectionOutcomeV1::Finished => { diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/test_gates.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/test_gates.rs index 6fe904126b..102974e294 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/test_gates.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/test_gates.rs @@ -60,30 +60,6 @@ impl CodeIndexSchedulerRegistryV1 { } } - /// Graph-serving state of the mounted text owner, the same projection - /// status reads. `None` means the worktree is unmounted or has no text - /// owner yet. - #[cfg(test)] - pub async fn code_graph_serving_readiness_for_test( - &self, - project_root: &Path, - ) -> Option { - let Ok(project_root) = project_root.canonicalize() else { - return None; - }; - let text = { - let mounted = self.mounted.lock().await; - let Some(worktree) = mounted.get(&project_root) else { - return None; - }; - Arc::clone(&worktree.text_generation) - }; - text.read() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .as_ref() - .map(|text| text.code_graph_serving_readiness()) - } - /// Test-only observation of an exact mounted worktree's active owner pass. #[cfg(test)] pub async fn reconcile_in_progress_for_test(&self, project_root: &Path) -> bool { 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 cb4a732c33..ddec371375 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 @@ -2645,10 +2645,17 @@ impl LatestCodeTextGenerationV1 { control: &dyn CodeIndexExecutionControlV1, ) -> Result, RetrievalPortError> { let generation_id = &self.metadata.manifest().generation_id; + // The successor's working set stays at + // `CLONE_SUCCESSOR_MEMORY_BUDGET_BYTES_V1`. The charge is at least + // the reader budget so publication can transfer it onto the reader + // component. A fresh reader admission at that boundary is refused + // when graph replay is already on the RSS watermark. + let publication_charge = CLONE_SUCCESSOR_MEMORY_BUDGET_BYTES_V1 + .max(CODE_LEXICAL_ARTIFACT_QUERY_CACHE_BUDGET_BYTES_V1); let reservation = self.text_artifact_store.reserve_resident_memory( generation_id, "code-text-clone-successor", - CLONE_SUCCESSOR_MEMORY_BUDGET_BYTES_V1, + publication_charge, )?; let artifacts_root = code_text_artifacts_root(self.text_artifact_store.store_root()); ensure_private_text_artifacts_root(&artifacts_root)?; @@ -3292,24 +3299,11 @@ impl LatestCodeTextGenerationV1 { &sealed_identity, control, )?; - // Keep the build charge and shrink it to the reader budget. Dropping - // it and reserving again is a gap: a graph replay that overlapped this - // projection can be sitting on the process RSS watermark, and the new - // admission is then refused forever. The reader is a smaller charge - // of the same generation, so it stays inside the bytes already held. - let reader_budget = u64::try_from(CODE_LEXICAL_ARTIFACT_QUERY_CACHE_BUDGET_BYTES_V1) - .map_err(|_| { - RetrievalPortError::Contract("text-artifact reader budget exceeds u64".to_owned()) - })?; - let mut reader_reservation = build_reservation; - if reader_reservation.reserved_bytes() > reader_budget { - reader_reservation.shrink_to(reader_budget).map_err(|error| { - RetrievalPortError::Contract(format!( - "text-artifact reader charge could not shrink inside its build reservation: reserved {} measured {}", - error.reserved_bytes, error.measured_bytes - )) - })?; - } + // The reader is a smaller charge of the bytes this build already + // holds. Dropping the build reservation and reserving the reader + // again is a gap: an overlapping graph replay can sit on the process + // RSS watermark, and the new admission is then refused forever. + let reader_reservation = reader_charge_from_held_reservation(build_reservation)?; let final_path = code_text_artifact_path(store.store_root(), &descriptor) .map_err(text_artifact_unavailable)?; let reader = CodeLexicalArtifactReaderV1::open_content_addressed( @@ -3414,7 +3408,11 @@ impl LatestCodeTextGenerationV1 { .finish(source_receipt, control) .map_err(map_text_artifact_error)?; drop(build.builder.take()); - drop(build.build_reservation.take()); + let held_reservation = build.build_reservation.take().ok_or_else(|| { + RetrievalPortError::Contract( + "clone-successor resident-memory charge disappeared before publication".to_owned(), + ) + })?; let descriptor = self.text_artifact_store.publish_with_prior( &build.staging_path, &self.metadata.manifest().generation_id, @@ -3422,11 +3420,7 @@ impl LatestCodeTextGenerationV1 { Some(&build.prior_descriptor), control, )?; - let reader_reservation = self.text_artifact_store.reserve_resident_memory( - &self.metadata.manifest().generation_id, - "code-text-artifact-reader", - CODE_LEXICAL_ARTIFACT_QUERY_CACHE_BUDGET_BYTES_V1, - )?; + let reader_reservation = reader_charge_from_held_reservation(held_reservation)?; let final_path = code_text_artifact_path(self.text_artifact_store.store_root(), &descriptor) .map_err(text_artifact_unavailable)?; @@ -3562,7 +3556,37 @@ impl LatestCodeTextGenerationV1 { build.builder = Some(builder); Ok(()) } +} + +/// Name a held publication charge as the reader and shrink it to the reader +/// budget. The bytes never leave the ledger, so the handoff is not a new +/// admission and measured RSS cannot refuse a charge that was already held. +fn reader_charge_from_held_reservation( + mut held: ResidentMemoryReservationV1, +) -> Result { + let reader_budget = + u64::try_from(CODE_LEXICAL_ARTIFACT_QUERY_CACHE_BUDGET_BYTES_V1).map_err(|_| { + RetrievalPortError::Contract("text-artifact reader budget exceeds u64".to_owned()) + })?; + if held.reserved_bytes() < reader_budget { + return Err(RetrievalPortError::Contract(format!( + "text-artifact publication charge {} is below the reader budget {reader_budget}", + held.reserved_bytes() + ))); + } + let component = ResidentMemoryComponentIdV1::new("code-text-artifact-reader") + .map_err(|error| RetrievalPortError::Contract(error.to_string()))?; + held.transfer_component(component, reader_budget) + .map_err(|error| { + RetrievalPortError::Contract(format!( + "text-artifact reader charge could not take over its held reservation: reserved {} measured {}", + error.reserved_bytes, error.measured_bytes + )) + })?; + Ok(held) +} +impl LatestCodeTextGenerationV1 { fn install_artifact_owners( &self, reader: CodeLexicalArtifactReaderV1, diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs index 5b39b0f3ca..188d869742 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs @@ -2628,57 +2628,6 @@ async fn graph_read_during_reconcile_records_a_busy_follow_up() { registry.shutdown().await; } -/// Strict readiness needs a ready graph and a finished text artifact. Awaiting -/// the artifact before the first graph attempt leaves the graph pending for -/// the whole build, which is the PR-dogfood timeout (still in bulk commit, -/// graph pending, at the deadline). Graph activation has to move while the -/// published text projection is held. -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn graph_activation_runs_while_published_text_projection_is_held() { - let fixture = GitFixture::new(ALPHA_LIB_V1); - let store = TempDir::new().expect("store root"); - let registry = CodeIndexSchedulerRegistryV1::with_background_reconcile_permits(1, 1); - let canonical_root = fixture.path().canonicalize().expect("canonical fixture"); - let (projection_started, release_projection) = registry - .pause_next_published_text_projection(canonical_root) - .await; - registry - .mount_worktree( - test_project_id(), - fixture.path(), - store.path().to_path_buf(), - ) - .await - .expect("mount worktree"); - tokio::time::timeout(Duration::from_secs(30), projection_started) - .await - .expect("publication did not reach text projection") - .expect("projection gate stays armed"); - - let graph_ready = tokio::time::timeout(Duration::from_secs(30), async { - loop { - if matches!( - registry - .code_graph_serving_readiness_for_test(fixture.path()) - .await, - Some( - tracedecay_contracts::code_index_freshness::CodeGraphServingReadinessV1::Ready - ) - ) { - return; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - }) - .await; - release_projection - .send(()) - .expect("release publication projection"); - graph_ready - .expect("graph serving becomes ready while the published text projection is still held"); - registry.shutdown().await; -} - /// A publication can finish source capture long before its text artifact is /// ready. The serving swap must reverify after that projection, otherwise the /// exact active generation seats after its bounded proof expires and every diff --git a/crates/tracedecay-runtime-core/src/resident_memory.rs b/crates/tracedecay-runtime-core/src/resident_memory.rs index f1669a488a..2a6e61292d 100644 --- a/crates/tracedecay-runtime-core/src/resident_memory.rs +++ b/crates/tracedecay-runtime-core/src/resident_memory.rs @@ -1134,6 +1134,57 @@ impl ProcessResidentMemoryV1 { Ok(()) } + /// Move one reservation's contribution onto `to_component` and keep only + /// `measured_bytes`, under the same lock. + /// + /// [`Self::reserve`] re-checks measured RSS. Dropping a charge and + /// reserving again is a gap: an overlapping consumer can sit on the + /// watermark and the new admission is refused even though these bytes + /// were already held. The ledger move does not ask for a new admission. + fn transfer_component( + &self, + from: &ResidentMemoryKeyV1, + to_component: ResidentMemoryComponentIdV1, + reserved_bytes: u64, + measured_bytes: u64, + ) -> Result<(), ResidentMemoryAdjustmentFailureV1> { + if measured_bytes > reserved_bytes { + return Err(ResidentMemoryAdjustmentFailureV1 { + reserved_bytes, + measured_bytes, + }); + } + let mut state = self.lock_state(); + let remove_source = { + let Some(charge) = state.charges.get_mut(from) else { + return Err(ResidentMemoryAdjustmentFailureV1 { + reserved_bytes, + measured_bytes, + }); + }; + if *charge < reserved_bytes { + return Err(ResidentMemoryAdjustmentFailureV1 { + reserved_bytes: *charge, + measured_bytes, + }); + } + *charge -= reserved_bytes; + *charge == 0 + }; + if remove_source { + state.charges.remove(from); + } + let released_bytes = reserved_bytes - measured_bytes; + state.used_bytes -= released_bytes; + hotpath::gauge!("runtime_core.resident.used_bytes").set(state.used_bytes as f64); + if measured_bytes > 0 { + let mut to = from.clone(); + to.component = to_component; + *state.charges.entry(to).or_default() += measured_bytes; + } + Ok(()) + } + fn release(&self, key: &ResidentMemoryKeyV1, reserved_bytes: u64) { if reserved_bytes == 0 { return; @@ -1230,6 +1281,29 @@ impl ResidentMemoryReservationV1 { self.reserved_bytes = measured_bytes; Ok(()) } + + /// Keep this charge and name it `component`, shrinking to `measured_bytes` + /// when the held amount is larger. + /// + /// The bytes stay in the ledger for the whole move. Callers that instead + /// drop this reservation and [`ProcessResidentMemoryV1::reserve`] the + /// destination open a gap where measured RSS can refuse a charge that + /// was already admitted. + pub fn transfer_component( + &mut self, + component: ResidentMemoryComponentIdV1, + measured_bytes: u64, + ) -> Result<(), ResidentMemoryAdjustmentFailureV1> { + self.authority.transfer_component( + &self.key, + component, + self.reserved_bytes, + measured_bytes, + )?; + self.key.component = component; + self.reserved_bytes = measured_bytes; + Ok(()) + } } impl Drop for ResidentMemoryReservationV1 { diff --git a/crates/tracedecay-runtime-core/src/resident_memory/tests.rs b/crates/tracedecay-runtime-core/src/resident_memory/tests.rs index 53bc42ca60..97a078593a 100644 --- a/crates/tracedecay-runtime-core/src/resident_memory/tests.rs +++ b/crates/tracedecay-runtime-core/src/resident_memory/tests.rs @@ -350,6 +350,57 @@ fn reservation_tracks_exact_identity_and_releases_on_drop() { assert_eq!(authority.snapshot().used_bytes, 0); } +#[test] +fn transfer_keeps_retained_bytes_when_a_new_admission_cannot() { + let authority = Arc::new(ProcessResidentMemoryV1::new(bytes(100))); + let build = key( + "project-a", + "worktree-a", + "generation-a", + "code-text-artifact-build", + ); + let reader = key( + "project-a", + "worktree-a", + "generation-a", + "code-text-artifact-reader", + ); + let mut held = authority + .reserve(build.clone(), bytes(80)) + .expect("build reservation"); + let _neighbor = authority + .reserve( + key("project-a", "worktree-a", "generation-a", "graph"), + bytes(20), + ) + .expect("neighbor fills the ceiling"); + let denied = authority + .reserve(reader.clone(), bytes(30)) + .expect_err("a fresh reader admission does not fit beside the held build charge"); + assert!(matches!( + denied, + ResidentMemoryAdmissionFailureV1::ReservationCeiling { .. } + )); + + held.transfer_component(reader.component, 30) + .expect("the held charge moves without a new admission"); + assert_eq!(held.key(), &reader); + assert_eq!(held.reserved_bytes(), 30); + let snapshot = authority.snapshot(); + assert_eq!(snapshot.used_bytes, 50); + assert_eq!(snapshot.charge_for(&build), 0); + assert_eq!(snapshot.charge_for(&reader), 30); + + let grown = held.transfer_component(reader.component, 40); + assert!(grown.is_err(), "a transfer cannot grow the held charge"); + assert_eq!(held.reserved_bytes(), 30); + assert_eq!(authority.snapshot().charge_for(&reader), 30); + + drop(held); + assert_eq!(authority.snapshot().used_bytes, 20); + assert_eq!(authority.snapshot().charge_for(&reader), 0); +} + #[test] fn rejection_reports_final_used_requested_and_limit_bytes() { let authority = Arc::new(ProcessResidentMemoryV1::new(bytes(100))); From 7f062995054418ee62f5e9266b0bf86acac1c0f9 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 19 Sep 2026 14:20:10 +0000 Subject: [PATCH 3/4] fix(sessions): drop a no-op length conversion Metadata::len is already u64. try_from is a useless conversion and -D warnings fails the workspace clippy check. Co-authored-by: Zack Jackson --- .../runtime/observation/jsonl_observation_admission/tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs index b5a72dfff6..6c33688210 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs @@ -1026,7 +1026,7 @@ async fn cursor_cas_lost_on_a_partially_covered_window_replays_the_tail() { }) ) .unwrap(); - let len = u64::try_from(std::fs::metadata(&path).unwrap().len()).unwrap(); + let len = std::fs::metadata(&path).unwrap().len(); let spy = SeamSpyAdmission::default(); spy.script_peer_covers_batch_prefix(); From fac0a0a475405b78a0691673db4c3d3085a1de59 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 19 Sep 2026 14:20:10 +0000 Subject: [PATCH 4/4] fix(code-index): do not share the worker runtime in tests The suite mounts many registries in one process. A process-wide runtime of a few threads then holds a waiter behind every other mount. The search decode test hit the 360s deadline, then passed in a fresh process. Tests already have a runtime that is not the daemon serving runtime, so they spawn there. The daemon still uses the dedicated worker runtime. Co-authored-by: Zack Jackson --- .../src/code_index_scheduler/registry/mount.rs | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs index 179c2103c3..e5ad481701 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs @@ -48,6 +48,12 @@ use super::{ /// dogfood deadline expired still inside text projection, graph never seated. /// A process-wide runtime keeps indexing scheduled as soon as the wake is /// posted. The future is already boxed, so the stack only has to poll it. +/// +/// Tests do not use this. The suite mounts many registries in one process; +/// a few shared threads then hold a waiter behind every other mount until +/// the per-test deadline. Each test already has a runtime that is not the +/// daemon's serving runtime. +#[cfg(not(test))] fn code_index_worker_runtime() -> Result<&'static tokio::runtime::Runtime, CodeIndexSchedulerErrorV1> { static RUNTIME: OnceLock> = OnceLock::new(); @@ -2365,6 +2371,12 @@ impl CodeIndexSchedulerRegistryV1 { let _ = result; } }); + #[cfg(test)] + let task = tokio::spawn(hotpath::future!( + worker_loop, + label = "daemon.code_index.scheduler_worker" + )); + #[cfg(not(test))] let task = code_index_worker_runtime()?.spawn(hotpath::future!( worker_loop, label = "daemon.code_index.scheduler_worker"