perf(code-index): build code-graph rows from sealed segments

crates/tracedecay-code-index-runtime/src/code_graph_seat.rs

@@ -20,7 +20,6 @@ use std::path::PathBuf;
 use std::pin::Pin;
 use std::sync::{Arc, atomic::AtomicBool};
 
-use tracedecay_code_index::production::CodeIndexPublishedGenerationV1;
 use tracedecay_domain::errors::Result;
 use tracedecay_domain::{CodeGenerationId, ProjectId, RefId, RepositoryId, WorktreeId};
 use tracedecay_graph_db::{GraphDbError, SealedGraphStateDigest, VerifiedGraphSnapshot};
@@ -41,9 +40,10 @@ pub struct CodeGraphReplayBindingV1 {
 pub trait CodeGraphSeatLeaseV1: Send {
     fn authority(&self) -> Arc<CanonicalCodeGraphStoreLeaseV1>;
 
+    /// Publishes the retained sealed generation's graph head, building its
+    /// rows from the sealed segments when the head has not landed yet.
     fn publish_verified_snapshot(
         &self,
-        generation: &CodeIndexPublishedGenerationV1,
         request_cancelled: Arc<AtomicBool>,
     ) -> std::result::Result<VerifiedGraphSnapshot, GraphDbError>;
 
@@ -76,6 +76,5 @@ pub trait CodeGraphSeatRuntimePortV1: Send + Sync {
         generation_id: CodeGenerationId,
         project_database: Arc<Database>,
         replay_binding: CodeGraphReplayBindingV1,
-        decoded_generation: Option<Arc<CodeIndexPublishedGenerationV1>>,
     ) -> CodeGraphSeatLeaseFutureV1<'_>;
 }

crates/tracedecay-code-index-runtime/src/code_index_scheduler/graph_activation.rs

@@ -392,7 +392,6 @@ impl CodeGraphActivationAuthorityV1 {
                         generation_id,
                         Arc::clone(project_database),
                         replay_binding,
-                        None,
                     ),
                     label = "code_graph.activation.recover_head.retain_runtime"
                 )
@@ -428,6 +427,64 @@ impl CodeGraphActivationAuthorityV1 {
         }
     }
 
+    /// Publishes a sealed generation's graph head straight from its segments
+    /// on disk, without decoding the generation.
+    ///
+    /// Graph prepare runs this before the serving decode so the corpus-sized
+    /// graph build and the decoded generation are never resident together;
+    /// the activation that follows recovers the head this published instead
+    /// of building it. `Ok(false)` abstains for a refused policy or a
+    /// non-persistent authority.
+    #[hotpath::measure(future = true, label = "code_graph.activation.publish_sealed")]
+    pub async fn publish_sealed_graph(
+        &self,
+        project_id: &ProjectId,
+        repository_id: &RepositoryId,
+        worktree_id: &WorktreeId,
+        latest: &LatestCodeTextGenerationV1,
+        replay_binding: CodeGraphReplayBindingV1,
+        cancellation: Arc<AtomicBool>,
+    ) -> Result<bool, CodeIndexSchedulerErrorV1> {
+        if self.policy() == CodeGraphActivationPolicyV1::RefusedByConfiguration {
+            return Ok(false);
+        }
+        match self {
+            Self::Persistent {
+                runtime,
+                project_database,
+                ..
+            } => {
+                let retained = hotpath::future!(
+                    runtime.retain_code_graph_runtime(
+                        project_id.clone(),
+                        repository_id.clone(),
+                        worktree_id.clone(),
+                        latest.metadata().snapshot().reference.clone(),
+                        latest.metadata().manifest().generation_id.clone(),
+                        Arc::clone(project_database),
+                        replay_binding,
+                    ),
+                    label = "code_graph.activation.publish_sealed.retain_runtime"
+                )
+                .await
+                .map_err(|error| CodeIndexSchedulerErrorV1::GraphActivation(error.to_string()))?;
+                tokio::task::spawn_blocking(move || {
+                    retained.publish_verified_snapshot(cancellation).map(drop)
+                })
+                .await
+                .map_err(|error| {
+                    CodeIndexSchedulerErrorV1::GraphActivation(format!(
+                        "sealed graph publication task failed: {error}"
+                    ))
+                })?
+                .map_err(CodeGraphProjectionError::from)?;
+                Ok(true)
+            }
+            #[cfg(any(test, feature = "test-helpers"))]
+            Self::Memory { .. } => Ok(false),
+        }
+    }
+
     #[hotpath::measure(future = true, label = "code_graph.activation.total")]
     pub async fn activate(
         &self,
@@ -464,7 +521,6 @@ impl CodeGraphActivationAuthorityV1 {
                         generation_id,
                         Arc::clone(project_database),
                         replay_binding,
-                        Some(latest.generation_handle()),
                     ),
                     label = "code_graph.activation.retain_runtime"
                 )
@@ -660,7 +716,7 @@ impl LatestCompleteCodeIndexV1 {
         let snapshot = hotpath::measure_block!(
             "code_graph.activation.publish_verified_snapshot",
             retained
-                .publish_verified_snapshot(&self.generation, Arc::clone(&cancellation))
+                .publish_verified_snapshot(Arc::clone(&cancellation))
                 .map_err(CodeGraphProjectionError::from)
                 .inspect_err(|error| {
                     // The publication stopped at the measured-RSS watermark.

crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs

@@ -477,6 +477,7 @@ pub(super) struct GenerationDecodeBudgetV1 {
 }
 
 const GENERATION_DECODE_RESIDENT_COMPONENT_V1: &str = "code-index-generation-decode-v1";
+const SEALED_GRAPH_BUILD_RESIDENT_COMPONENT_V1: &str = "code-graph-sealed-build-v1";
 
 /// The resident cost of materializing the active generation.
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -2298,6 +2299,28 @@ impl DaemonCodeIndexPublicationStoreV1 {
     /// generation is measured resident like every other owner.
     fn admit_active_decode(
         &self,
+    ) -> Result<Option<ResidentMemoryReservationV1>, CodeIndexPublicationStoreErrorV1> {
+        self.admit_active_generation_work(GENERATION_DECODE_RESIDENT_COMPONENT_V1, "decoding")
+    }
+
+    /// Charge building the active generation's code graph from its sealed
+    /// segments, the same way and the same bytes as decoding it: the build
+    /// holds the generation's cross-file resolution inputs and then its
+    /// compact graph store, both bounded by the generation it projects. The
+    /// caller holds the reservation for the build.
+    pub(super) fn admit_sealed_graph_build(
+        &self,
+    ) -> Result<Option<ResidentMemoryReservationV1>, CodeIndexPublicationStoreErrorV1> {
+        self.admit_active_generation_work(
+            SEALED_GRAPH_BUILD_RESIDENT_COMPONENT_V1,
+            "building the code graph of",
+        )
+    }
+
+    fn admit_active_generation_work(
+        &self,
+        component: &'static str,
+        work: &str,
     ) -> Result<Option<ResidentMemoryReservationV1>, CodeIndexPublicationStoreErrorV1> {
         let Some(admission) = self
             .decode_admission
@@ -2339,7 +2362,7 @@ impl DaemonCodeIndexPublicationStoreV1 {
                 Ok(())
             } else {
                 Err(format!(
-                    "decoding generation {generation_id} needs {} resident bytes; {available} are \
+                    "{work} generation {generation_id} needs {} resident bytes; {available} are \
                      available below the {watermark}-byte admission watermark",
                     requested.get()
                 ))
@@ -2366,8 +2389,7 @@ impl DaemonCodeIndexPublicationStoreV1 {
             );
             admissible().map_err(CodeIndexPublicationStoreErrorV1::ResidentMemoryRefused)?;
         }
-        let component = ResidentMemoryComponentIdV1::new(GENERATION_DECODE_RESIDENT_COMPONENT_V1)
-            .map_err(Self::unavailable)?;
+        let component = ResidentMemoryComponentIdV1::new(component).map_err(Self::unavailable)?;
         admission
             .resident_memory
             .reserve(

crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs

@@ -1683,6 +1683,89 @@ impl CodeIndexSchedulerRegistryV1 {
                 }
                 let mut result = match source_result {
                     Ok(mut outcome) if prepare_graph => {
+                        // Publish the graph head from the sealed segments
+                        // before the serving decode below. The graph build and
+                        // the decoded generation are this step's two
+                        // corpus-sized working sets and must not be resident
+                        // together; the activation after the decode recovers
+                        // the head published here.
+                        let mut graph_publish_refusal = None;
+                        if let Some(text) = graph_text.as_ref() {
+                            let generation_id = text.metadata().manifest().generation_id.clone();
+                            let binding_scheduler = Arc::clone(&worker_scheduler);
+                            let shutting_down = Arc::clone(&worker_shutting_down);
+                            let binding_passes = Arc::clone(&worker_reconcile_in_progress);
+                            // The build is admitted like the decode it
+                            // replaces: charged before it runs, parked when it
+                            // does not fit, and holding its reservation until
+                            // the head is published.
+                            let admitted_binding = tokio::task::spawn_blocking(move || {
+                                let (_step, scheduler) = Self::lock_scheduler_for_graph_step(
+                                    &binding_scheduler,
+                                    &shutting_down,
+                                    &binding_passes,
+                                )?;
+                                let binding =
+                                    scheduler.code_graph_replay_binding(&generation_id)?;
+                                let admission = scheduler
+                                    .active_generation_decoder()
+                                    .map(|decoder| decoder.admit_sealed_graph_build())
+                                    .transpose();
+                                Ok::<_, CodeIndexSchedulerErrorV1>((binding, admission))
+                            })
+                            .await;
+                            match admitted_binding {
+                                Ok(Ok((
+                                    _,
+                                    Err(CodeIndexPublicationStoreErrorV1::ResidentMemoryRefused(
+                                        detail,
+                                    )),
+                                ))) => graph_publish_refusal = Some(detail),
+                                Ok(Ok((_, Err(error)))) => tracing::warn!(
+                                    event = "code_index_graph_publish_admission_failed",
+                                    error = %error,
+                                    "sealed graph build admission failed; activation publishes \
+                                     the graph after the serving decode"
+                                ),
+                                Ok(Ok((replay_binding, Ok(reservation)))) => {
+                                    let published = worker_graph_activation
+                                        .publish_sealed_graph(
+                                            &worker_project_id,
+                                            &worker_repository_id,
+                                            &worker_worktree_id,
+                                            text,
+                                            replay_binding,
+                                            Arc::clone(&worker_shutting_down),
+                                        )
+                                        .await;
+                                    drop(reservation);
+                                    match published {
+                                        Ok(_) => {}
+                                        Err(error) if error.is_resident_memory_graph_refusal() => {
+                                            graph_publish_refusal = Some(error.to_string());
+                                        }
+                                        Err(error) => tracing::warn!(
+                                            event = "code_index_graph_publish_before_decode_failed",
+                                            error = %error,
+                                            "sealed graph publication failed before the serving \
+                                             decode; activation retries it after the decode"
+                                        ),
+                                    }
+                                }
+                                Ok(Err(error)) => tracing::warn!(
+                                    event = "code_index_graph_publish_binding_unavailable",
+                                    error = %error,
+                                    "sealed replay binding is unavailable; activation publishes \
+                                     the graph after the serving decode"
+                                ),
+                                Err(error) => tracing::warn!(
+                                    event = "code_index_graph_publish_binding_task_failed",
+                                    error = %error,
+                                    "sealed replay binding task failed; activation publishes the \
+                                     graph after the serving decode"
+                                ),
+                            }
+                        }
                         let graph_scheduler = Arc::clone(&worker_scheduler);
                         let graph_text = graph_text.clone();
                         let shutting_down = Arc::clone(&worker_shutting_down);
@@ -1691,6 +1774,11 @@ impl CodeIndexSchedulerRegistryV1 {
                         let prepare_wake = Arc::clone(&worker_wake);
                         match hotpath::future!(
                             tokio::task::spawn_blocking(move || {
+                                // A graph build the memory watermark stopped
+                                // parks exactly like a decode that does not fit.
+                                if let Some(detail) = graph_publish_refusal {
+                                    return Ok((None, None, false, Some(detail)));
+                                }
                                 let decoder = Self::lock_scheduler_for_graph_step(
                                     &graph_scheduler,
                                     &shutting_down,

crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs

@@ -1414,16 +1414,57 @@ fn occurrence_graph_store_is_available_before_catalog_warm() {
         tracedecay_graph_db::GraphNamespace::new("code-graph").expect("graph namespace"),
     )
     .expect("projection identity");
-    let manifest =
-        crate::code_index::graph_projection::build_published_code_graph_manifest_checked(
+    // Build the graph the way publication does: from the sealed segments on
+    // disk, one window of files at a time.
+    let binding = scheduler
+        .code_graph_replay_binding(&generation_id)
+        .expect("sealed replay binding");
+    let digest = tracedecay_domain::sha256_hex_suffix(binding.sealed_state_digest.as_str())
+        .expect("sha256 sealed digest");
+    let sealed_manifest = std::fs::read(
+        binding
+            .generations_root
+            .join(format!("generation-{digest}.json")),
+    )
+    .expect("sealed manifest");
+    let segments_root =
+        tracedecay_code_index_retention::code_index_generations::code_generation_segments_root(
+            binding.generations_root.parent().expect("store root"),
+        );
+    let source =
+        crate::code_index::production::SealedGenerationFileWindowsV1::open(&sealed_manifest)
+            .expect("sealed manifest opens");
+    let scratch = TempDir::new().expect("graph row scratch");
+    let manifest = crate::code_index::graph_projection::build_sealed_code_graph_rows(
+        projection.clone(),
+        &source,
+        &mut |request, buffer| {
+            let crate::code_index::production::SealedGenerationSegmentReadV1::Whole {
+                digest, ..
+            } = request
+            else {
+                panic!("the graph build reads whole file segments");
+            };
+            *buffer = std::fs::read(segments_root.join(format!(
+                "segment-{}.json",
+                tracedecay_domain::sha256_hex_suffix(digest.as_str()).expect("segment digest")
+            )))
+            .expect("sealed segment");
+            Ok(())
+        },
+        &projector_revision,
+        tracedecay_graph_db::GraphGenerationRowSpill::create(
+            scratch.path().join("rows"),
             projection,
-            latest.generation(),
-            &projector_revision,
-            &|| Ok(()),
         )
-        .expect("code graph manifest");
+        .expect("row spill"),
+        &|| Ok(()),
+    )
+    .expect("code graph rows")
+    .materialize(&|| Ok(()))
+    .expect("code graph manifest");
     let snapshot = tracedecay_graph_db::VerifiedGraphSnapshot::memory(
-        manifest.as_ref().clone(),
+        manifest,
         Arc::new(tracedecay_graph_db::NeverCancelled),
     )
     .expect("verified graph snapshot");