perf(index): substantially accelerate text catch-up

.gitignore

@@ -3,7 +3,6 @@
 .DS_Store
 .codegraph
 .tracedecay
-.mcp.json
 !plugin/.mcp.json
 .claude/settings.local.json
 .cursor/

.mcp.json

@@ -0,0 +1,8 @@
+{
+  "mcpServers": {
+    "hotpath": {
+      "type": "http",
+      "url": "http://127.0.0.1:6771/mcp"
+    }
+  }
+}

crates/tracedecay-code-index/src/extract.rs

@@ -199,7 +199,6 @@ impl TreeSitterExtractor {
         })
     }
 
-    #[hotpath::measure]
     pub(crate) fn extract_preparsed(
         &self,
         file: &ReceiptBoundCodeFileV1,
@@ -208,30 +207,32 @@ impl TreeSitterExtractor {
         parsed_len: usize,
         cancellation: &dyn ExtractionCancellation,
     ) -> Result<ExtractedCodeFileV1, ExtractionFailureV1> {
-        if cancellation.is_cancelled() {
-            return Err(ExtractionFailureV1::Cancelled);
-        }
-        let authority = file.authority().clone();
-        let file = file.validated_file();
-        validate_descriptor(file, descriptor)?;
-        let admitted_prefix = file
-            .sanitized_bytes
-            .get(..parsed_len)
-            .and_then(|bytes| std::str::from_utf8(bytes).ok());
-        if admitted_prefix.is_none() {
-            return Err(ExtractionFailureV1::ParseFailed {
-                detail: "retained parse prefix is not an admitted UTF-8 boundary".to_owned(),
-            });
-        }
-        finish_extraction(
-            authority,
-            file,
-            descriptor,
-            artifact,
-            parsed_len,
-            parsed_len < file.sanitized_bytes.len(),
-            cancellation,
-        )
+        crate::hotpath_observe::measure_hot_loop!("code_index.extract.incremental", {
+            if cancellation.is_cancelled() {
+                return Err(ExtractionFailureV1::Cancelled);
+            }
+            let authority = file.authority().clone();
+            let file = file.validated_file();
+            validate_descriptor(file, descriptor)?;
+            let admitted_prefix = file
+                .sanitized_bytes
+                .get(..parsed_len)
+                .and_then(|bytes| std::str::from_utf8(bytes).ok());
+            if admitted_prefix.is_none() {
+                return Err(ExtractionFailureV1::ParseFailed {
+                    detail: "retained parse prefix is not an admitted UTF-8 boundary".to_owned(),
+                });
+            }
+            finish_extraction(
+                authority,
+                file,
+                descriptor,
+                artifact,
+                parsed_len,
+                parsed_len < file.sanitized_bytes.len(),
+                cancellation,
+            )
+        })
     }
 
     /// Typed evidence for one file whose bounded retained parse exceeded its
@@ -441,7 +442,7 @@ fn rows_digest(
         .map(CanonicalUnresolvedRefRow::from)
         .collect::<Vec<_>>();
     let mut imports = artifact.imports.clone();
-    hotpath::measure_block!("code_index_rows_digest_sort", {
+    crate::hotpath_observe::measure_hot_loop!("code_index_rows_digest_sort", {
         sort_canonical_rows(&mut nodes);
         sort_canonical_rows(&mut edges);
         sort_canonical_rows(&mut unresolved);
@@ -462,7 +463,7 @@ fn rows_digest(
         unresolved_refs: Vec<CanonicalUnresolvedRefRow<'a>>,
     }
 
-    hotpath::measure_block!(
+    crate::hotpath_observe::measure_hot_loop!(
         "code_index_rows_digest_hash",
         canonical_sha256(&RowsPayload {
             separator: EXTRACTION_ROWS_SEPARATOR,
@@ -485,7 +486,7 @@ fn rows_digest(
 pub(crate) fn parser_import_rows_digest(
     imports: &[ExtractedImportEvidenceV1],
 ) -> Result<ManifestDigest, ExtractionFailureV1> {
-    hotpath::measure_block!("code_index_parser_import_rows_digest", {
+    crate::hotpath_observe::measure_hot_loop!("code_index_parser_import_rows_digest", {
         let mut imports = imports.to_vec();
         imports.sort();
         canonical_sha256(&(PARSER_IMPORT_ROWS_DIGEST_SEPARATOR, imports.as_slice())).map_err(
@@ -497,55 +498,56 @@ pub(crate) fn parser_import_rows_digest(
 }
 
 impl LanguageExtractor for TreeSitterExtractor {
-    #[hotpath::measure]
     fn extract(
         &self,
         file: &ReceiptBoundCodeFileV1,
         descriptor: &LanguageDescriptorV1,
         cancellation: &dyn ExtractionCancellation,
     ) -> Result<ExtractedCodeFileV1, ExtractionFailureV1> {
-        if cancellation.is_cancelled() {
-            return Err(ExtractionFailureV1::Cancelled);
-        }
-        let authority = file.authority().clone();
-        let file = file.validated_file();
-        validate_descriptor(file, descriptor)?;
-
-        let parser = self.resolve_parser(file, descriptor).ok_or({
-            ExtractionFailureV1::GrammarUnavailable {
-                language: descriptor.language.clone(),
+        crate::hotpath_observe::measure_hot_loop!("code_index.extract.full", {
+            if cancellation.is_cancelled() {
+                return Err(ExtractionFailureV1::Cancelled);
             }
-        })?;
-        if canonical_language_id(parser.language_name()) != descriptor.language.as_str() {
-            return Err(ExtractionFailureV1::IncompatibleDescriptor {
-                detail: format!(
-                    "descriptor {} resolved to a {} parser",
-                    descriptor.language,
-                    parser.language_name()
-                ),
-            });
-        }
-
-        let source = std::str::from_utf8(&file.sanitized_bytes).map_err(|error| {
-            ExtractionFailureV1::ParseFailed {
-                detail: format!("sanitized bytes are not valid UTF-8: {error}"),
+            let authority = file.authority().clone();
+            let file = file.validated_file();
+            validate_descriptor(file, descriptor)?;
+
+            let parser = self.resolve_parser(file, descriptor).ok_or({
+                ExtractionFailureV1::GrammarUnavailable {
+                    language: descriptor.language.clone(),
+                }
+            })?;
+            if canonical_language_id(parser.language_name()) != descriptor.language.as_str() {
+                return Err(ExtractionFailureV1::IncompatibleDescriptor {
+                    detail: format!(
+                        "descriptor {} resolved to a {} parser",
+                        descriptor.language,
+                        parser.language_name()
+                    ),
+                });
             }
-        })?;
-        let parsed_len =
-            crate::chunks::snap_down(source, source.len().min(MAX_EXTRACTION_SOURCE_BYTES));
-        let extraction_source = &source[..parsed_len];
-        let source_was_capped = parsed_len < source.len();
-
-        let artifact = parser.extract_artifact(&file.file.logical_path, extraction_source);
-        finish_extraction(
-            authority,
-            file,
-            descriptor,
-            artifact,
-            parsed_len,
-            source_was_capped,
-            cancellation,
-        )
+
+            let source = std::str::from_utf8(&file.sanitized_bytes).map_err(|error| {
+                ExtractionFailureV1::ParseFailed {
+                    detail: format!("sanitized bytes are not valid UTF-8: {error}"),
+                }
+            })?;
+            let parsed_len =
+                crate::chunks::snap_down(source, source.len().min(MAX_EXTRACTION_SOURCE_BYTES));
+            let extraction_source = &source[..parsed_len];
+            let source_was_capped = parsed_len < source.len();
+
+            let artifact = parser.extract_artifact(&file.file.logical_path, extraction_source);
+            finish_extraction(
+                authority,
+                file,
+                descriptor,
+                artifact,
+                parsed_len,
+                source_was_capped,
+                cancellation,
+            )
+        })
     }
 }
 

crates/tracedecay-code-index/src/generations.rs

@@ -270,7 +270,7 @@ impl<R: LanguageRegistry> GenerationPlanner<R> {
     /// Plan and seal one immutable generation while binding every inferred and
     /// explicitly declared rebuild cause into its identity and publication
     /// fence.
-    #[hotpath::measure]
+    #[hotpath::measure(label = "code_index.build.plan_full")]
     pub fn plan_generation_with_invalidation(
         &self,
         snapshot: &ValidatedCodeSnapshotV1,
@@ -364,7 +364,7 @@ impl<R: LanguageRegistry> GenerationPlanner<R> {
     /// quarantined corruption are not inferable from a sanitized snapshot,
     /// so callers must declare them explicitly. Declared reasons are merged
     /// with descriptor, sanitizer, chunker, and privacy incompatibilities.
-    #[hotpath::measure]
+    #[hotpath::measure(label = "code_index.build.plan_increment")]
     pub fn plan_increment_with_invalidation(
         &self,
         prior_manifest: &CodeGenerationManifestV1,
@@ -681,7 +681,7 @@ pub enum GenerationJoinErrorV1 {
 /// every eligible chunk names exactly one code generation and file
 /// occurrence). Cross-generation documents or chunks, undeclared chunks, and
 /// duplicates are typed rejections — never silently joined.
-#[hotpath::measure]
+#[hotpath::measure(label = "code_index.build.generation_join")]
 pub fn join_chunks_to_generation(
     generation: &CodeGenerationManifestV1,
     document: &CodeSearchDocumentV1,

crates/tracedecay-code-index/src/hotpath_observe.rs

@@ -17,7 +17,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 #[cfg(feature = "hotpath")]
 use std::time::Instant;
 
-#[cfg(feature = "hotpath")]
+#[cfg(any(feature = "hotpath", test))]
 const HOT_LOOP_SAMPLE_PERIOD: u64 = 32;
 
 #[cfg(feature = "hotpath")]
@@ -27,23 +27,46 @@ static WORKERS_ACTIVE: AtomicUsize = AtomicUsize::new(0);
 #[cfg(feature = "hotpath")]
 static WORKERS_POOL_COORDINATION: AtomicUsize = AtomicUsize::new(0);
 #[cfg(feature = "hotpath")]
-static GREP_FILE_SAMPLE: AtomicU64 = AtomicU64::new(0);
+static HOT_LOOP_SAMPLE: AtomicU64 = AtomicU64::new(0);
+
+#[cfg(feature = "hotpath")]
+#[must_use]
+#[inline(always)]
+fn is_hot_loop_sample(sequence: u64) -> bool {
+    sequence.is_multiple_of(HOT_LOOP_SAMPLE_PERIOD)
+}
 
 #[must_use]
 #[inline(always)]
 pub(crate) fn sample_hot_loop() -> bool {
     #[cfg(feature = "hotpath")]
     {
-        GREP_FILE_SAMPLE
-            .fetch_add(1, Ordering::Relaxed)
-            .is_multiple_of(HOT_LOOP_SAMPLE_PERIOD)
+        is_hot_loop_sample(HOT_LOOP_SAMPLE.fetch_add(1, Ordering::Relaxed))
     }
     #[cfg(not(feature = "hotpath"))]
     {
         false
     }
 }
 
+/// Measure one fixed-rate sample from a hot file loop.
+///
+/// The label must be a literal so callers cannot create path- or
+/// generation-shaped cardinality. The disabled path still evaluates only the
+/// work expression; its sampler is an inlined constant `false` with no atomic
+/// or clock access.
+macro_rules! measure_hot_loop {
+    ($label:literal, $work:expr) => {{
+        if $crate::hotpath_observe::sample_hot_loop() {
+            hotpath::measure_block!($label, $work)
+        } else {
+            $work
+        }
+    }};
+}
+
+pub(crate) use measure_hot_loop;
+
 #[cfg(any(feature = "hotpath", test))]
 #[inline(always)]
 fn decrement_if_positive(counter: &AtomicUsize) -> bool {
@@ -297,21 +320,26 @@ pub(crate) fn record_seal_bytes(bytes: u64) {
     }
 }
 
-pub(crate) struct TtfqStart(#[cfg(feature = "hotpath")] Instant);
+/// Start of one production-owner generation build. The matching observation
+/// ends only after the immutable generation has been published and is
+/// queryable through that owner; daemon scheduling/wake latency is measured
+/// separately by the reconcile cadence receipt.
+pub(crate) struct BuildToQueryableStart(#[cfg(feature = "hotpath")] Instant);
 
 #[inline(always)]
-pub(crate) fn start_ttfq() -> TtfqStart {
-    TtfqStart(
+pub(crate) fn start_build_to_queryable() -> BuildToQueryableStart {
+    BuildToQueryableStart(
         #[cfg(feature = "hotpath")]
         Instant::now(),
     )
 }
 
 #[inline(always)]
-pub(crate) fn record_ttfq(started: TtfqStart) {
+pub(crate) fn record_build_to_queryable(started: BuildToQueryableStart) {
     #[cfg(feature = "hotpath")]
     {
-        hotpath::gauge!("code_index_ttfq_micros").set(started.0.elapsed().as_micros() as f64);
+        hotpath::gauge!("code_index_build_to_queryable_micros")
+            .set(started.0.elapsed().as_micros() as f64);
     }
     #[cfg(not(feature = "hotpath"))]
     {
@@ -347,6 +375,22 @@ pub(crate) fn record_rebuild_state(state: &'static str) {
 mod tests {
     use super::*;
 
+    #[cfg(feature = "hotpath")]
+    #[test]
+    fn hotpath_file_probe_cadence_uses_one_fixed_slot_per_period() {
+        let sampled = (0..(HOT_LOOP_SAMPLE_PERIOD * 2))
+            .filter(|sequence| is_hot_loop_sample(*sequence))
+            .collect::<Vec<_>>();
+
+        assert_eq!(sampled, vec![0, HOT_LOOP_SAMPLE_PERIOD]);
+    }
+
+    #[cfg(not(feature = "hotpath"))]
+    #[test]
+    fn hotpath_file_probe_sampler_is_dormant_without_the_feature() {
+        assert!((0..(HOT_LOOP_SAMPLE_PERIOD * 2)).all(|_| !sample_hot_loop()));
+    }
+
     #[test]
     fn pending_queue_decrements_when_each_worker_starts() {
         let queue = PendingWorkQueue::new(3);