fix: move long native polls off the worker for #4066

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

@@ -957,7 +957,9 @@ impl CodeIndexSchedulerRegistryV1 {
                     // banked permit resolves the wait at once, so the worker was
                     // never idle.
                     if !notified.as_mut().enable() {
-                        worker_residency.release_decode_when_parked(&worker_resident_owners);
+                        worker_residency
+                            .release_decode_when_parked_off_worker(&worker_resident_owners)
+                            .await;
                         super::CodeIndexWorkerPhaseV1::enter(
                             &worker_phase_signal,
                             super::CodeIndexWorkerPhaseV1::Parked,

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

@@ -781,6 +781,27 @@ impl CodeIndexSchedulerRegistryV1 {
             .release_decode_when_parked(&self.resident_owners);
     }
 
+    /// Make this worktree's next park release spend `cpu` first. The receiver
+    /// resolves once that CPU is spent.
+    #[cfg(test)]
+    pub async fn inject_park_release_cpu_for_test(
+        &self,
+        project_root: &Path,
+        cpu: std::time::Duration,
+    ) -> tokio::sync::oneshot::Receiver<()> {
+        let project_root = canonical_existing_identity(project_root).expect("canonical root");
+        let (spun, spent) = tokio::sync::oneshot::channel();
+        let mounted = self.mounted.lock().await;
+        *mounted
+            .get(&project_root)
+            .expect("mounted worktree")
+            .residency
+            .injected_park_release_cpu
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some((cpu, spun));
+        spent
+    }
+
     /// Drop only the decoded seat, as a deferred serving decode does once a
     /// published graph head serves from the text owner.
     #[cfg(test)]

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

@@ -25,6 +25,7 @@ use std::sync::{Arc, Mutex, PoisonError, RwLock, Weak};
 use std::time::Instant;
 
 use crate::code_index_task_support::spawn_detached;
+use tracedecay_runtime_core::native_runtime;
 
 use tracedecay_code_index::graph_projection::{
     CodeGraphCatalogReleaseV1, CodeGraphEngineReleaseV1,
@@ -69,6 +70,10 @@ pub(super) struct WorktreeResidencyV1 {
     park_release_after_pass_armed: AtomicBool,
     park_release_after_reads_drain_retrying: AtomicBool,
     pub(super) park_headroom: ParkHeadroomV1,
+    /// CPU the next park release spends first, and who hears it finish.
+    #[cfg(test)]
+    pub(super) injected_park_release_cpu:
+        Mutex<Option<(std::time::Duration, tokio::sync::oneshot::Sender<()>)>>,
 }
 
 /// The owner-headroom epochs a worktree's own park releases advanced.
@@ -131,6 +136,8 @@ impl WorktreeResidencyV1 {
             park_release_after_pass_armed: AtomicBool::new(false),
             park_release_after_reads_drain_retrying: AtomicBool::new(false),
             park_headroom: ParkHeadroomV1::default(),
+            #[cfg(test)]
+            injected_park_release_cpu: Mutex::new(None),
         }
     }
 
@@ -183,6 +190,25 @@ impl WorktreeResidencyV1 {
         }
     }
 
+    /// [`Self::release_decode_when_parked`] on the blocking pool. Its callers
+    /// are native tasks, and the release walks, frees, and unmaps a decode or
+    /// graph engine and rewrites its generation markers. Inside a native
+    /// task's poll that holds the worker, and past `LONG_POLL_CPU` a debug
+    /// build fails the task (#4066).
+    pub(super) async fn release_decode_when_parked_off_worker(
+        self: &Arc<Self>,
+        owners: &Arc<ResidentOwnersV1>,
+    ) {
+        let (residency, owners) = (Arc::clone(self), Arc::clone(owners));
+        if let Err(error) = native_runtime::spawn_blocking(move || {
+            residency.release_decode_when_parked(&owners);
+        })
+        .await
+        {
+            tracing::error!(%error, "the parked decode release did not run");
+        }
+    }
+
     /// Drop the seated decode, catalog, and engine once the worker parks.
     ///
     /// Exact and lexical serve from the sealed text artifact. Callers reseat
@@ -200,6 +226,8 @@ impl WorktreeResidencyV1 {
     /// re-decode. The decode is announced as this worktree's own park
     /// release: a retry re-decodes it (issue #3588).
     pub(super) fn release_decode_when_parked(self: &Arc<Self>, owners: &Arc<ResidentOwnersV1>) {
+        #[cfg(test)]
+        self.spin_injected_park_release_cpu();
         if self.busy() {
             self.arm_park_release_after_pass_settles(owners);
             return;
@@ -356,7 +384,9 @@ impl WorktreeResidencyV1 {
             let (residency, owners) = (Weak::clone(&residency), Weak::clone(&weak_owners));
             spawn_detached("code_index_park_release_after_reads", async move {
                 if let (Some(residency), Some(owners)) = (residency.upgrade(), owners.upgrade()) {
-                    residency.release_decode_when_parked(&owners);
+                    residency
+                        .release_decode_when_parked_off_worker(&owners)
+                        .await;
                 }
             });
         }));
@@ -386,7 +416,9 @@ impl WorktreeResidencyV1 {
                 residency
                     .park_release_after_pass_armed
                     .store(false, Ordering::Release);
-                residency.release_decode_when_parked(&owners);
+                residency
+                    .release_decode_when_parked_off_worker(&owners)
+                    .await;
             }
         });
     }
@@ -418,7 +450,9 @@ impl WorktreeResidencyV1 {
         let settled = text.catalog_warm_notify();
         spawn_detached("code_index_park_release_after_catalog_warm", async move {
             settled.notified().await;
-            retry();
+            if let Err(error) = native_runtime::spawn_blocking(move || retry()).await {
+                tracing::error!(%error, "the parked graph release retry did not run");
+            }
         });
     }
 
@@ -449,6 +483,22 @@ impl WorktreeResidencyV1 {
         self.reconcile_in_progress.running()
     }
 
+    #[cfg(test)]
+    fn spin_injected_park_release_cpu(&self) {
+        let injected = self
+            .injected_park_release_cpu
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .take();
+        if let Some((cpu, spun)) = injected {
+            let started = Instant::now();
+            while started.elapsed() < cpu {
+                std::hint::spin_loop();
+            }
+            let _ = spun.send(());
+        }
+    }
+
     /// Register every one of this worktree's owners with `owners`. The returned
     /// guard keeps the owners alive and registered for the mount's lifetime.
     pub(super) fn register(

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

@@ -2128,3 +2128,64 @@ async fn a_memory_activation_longer_than_one_native_poll_still_seats() {
     wait_for_live_complete_generation(&registry, fixture.path()).await;
     registry.shutdown().await;
 }
+
+async fn commit_and_index(
+    registry: &CodeIndexSchedulerRegistryV1,
+    fixture: &GitFixture,
+    source: &str,
+) {
+    let previous = registry
+        .latest_generation_id(fixture.path())
+        .await
+        .expect("a seated generation");
+    fixture.edit("src/lib.rs", source);
+    git(fixture.path(), &["commit", "-qam", source]);
+    assert!(matches!(
+        registry
+            .notify_path(fixture.path(), fixture.path().join("src/lib.rs"))
+            .await,
+        super::super::CodeIndexDemandAdmissionV1::Queued
+    ));
+    wait_for_generation_change(registry, fixture.path(), &previous).await;
+}
+
+/// A park release that outlasts the CPU one native poll may spend leaves the
+/// worker indexing. The release frees and unmaps decodes and graph engines
+/// synchronously: run inside the worker's poll, the debug long-poll guard
+/// fails the worker and no later edit is indexed (#4066).
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn a_park_release_longer_than_one_native_poll_keeps_the_worker_indexing() {
+    let fixture = GitFixture::new(&[("src/lib.rs", "pub fn before_park() -> u32 { 1 }\n")]);
+    let store = TempDir::new().expect("store root");
+    let (registry, _) = mounted_text_query_worktree_at(
+        CodeIndexSchedulerRegistryV1::new(1),
+        fixture.path(),
+        store.path().to_path_buf(),
+    )
+    .await;
+    let spent = registry
+        .inject_park_release_cpu_for_test(
+            fixture.path(),
+            tracedecay_runtime_core::native_runtime::LONG_POLL_CPU * 2,
+        )
+        .await;
+
+    commit_and_index(
+        &registry,
+        &fixture,
+        "pub fn parks_after_this() -> u32 { 2 }\n",
+    )
+    .await;
+    tokio::time::timeout(SERVING_SEAT_FAILURE_CEILING, spent)
+        .await
+        .expect("the worker parks after indexing the edit")
+        .expect("the injected release ran");
+    commit_and_index(
+        &registry,
+        &fixture,
+        "pub fn indexed_after_the_long_release() -> u32 { 3 }\n",
+    )
+    .await;
+
+    registry.shutdown().await;
+}

crates/tracedecay-store-runtime/src/store_locator_resolver.rs

@@ -550,8 +550,19 @@ impl StoreRuntimeResolver for LocalStoreRuntimeResolverV1 {
         key: &'a StoreRuntimeKey,
     ) -> StoreRuntimeRegistryFuture<'a, Result<ResolvedStoreLocator, StoreRuntimeRegistryFailure>>
     {
+        // Resolution canonicalizes paths and reads and parses
+        // `/proc/self/mountinfo`. Its callers are native tasks, where that
+        // synchronous work holds the worker; past `LONG_POLL_CPU` a debug
+        // build fails the task (#4066).
+        let resolver = self.clone();
+        let key = key.clone();
         Box::pin(async move {
-            match self.resolve_graph_key(key) {
+            let resolution = tracedecay_runtime_core::native_runtime::spawn_blocking(move || {
+                resolver.resolve_graph_key(&key)
+            })
+            .await
+            .map_err(|message| StoreRuntimeRegistryFailure::ResolverFailed { message })?;
+            match resolution {
                 LocalStoreLocatorResolutionV1::Resolved(locator) => {
                     Ok(locator.into_registry_locator())
                 }

crates/tracedecay-store-runtime/src/store_locator_resolver/graph.rs

@@ -15,6 +15,8 @@ impl LocalStoreRuntimeResolverV1 {
     /// `StoreRuntimeRegistry::retain_code_graph_store`; resolving a code shard
     /// here fails closed instead of recreating per-worktree graph sharding.
     pub fn resolve_graph_key(&self, key: &StoreRuntimeKey) -> LocalStoreLocatorResolutionV1 {
+        #[cfg(test)]
+        tests::spin_injected_resolution_cpu(key);
         if matches!(key.shard_id().scope, StoreShardScopeV1::Code { .. }) {
             return LocalStoreLocatorResolutionV1::Unavailable(LocalStoreLocatorUnavailableV1 {
                 shard_id: key.shard_id().clone(),
@@ -211,4 +213,64 @@ mod tests {
         let database = mount_and_resolve(&registry, registration).unwrap();
         assert!(database.snapshot().is_ok());
     }
+
+    static INJECTED_RESOLUTION_CPU: std::sync::Mutex<std::collections::BTreeMap<String, Duration>> =
+        std::sync::Mutex::new(std::collections::BTreeMap::new());
+
+    pub(super) fn spin_injected_resolution_cpu(key: &StoreRuntimeKey) {
+        let cpu = INJECTED_RESOLUTION_CPU
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .remove(&format!("{:?}", key.shard_id()));
+        if let Some(cpu) = cpu {
+            let spun = std::time::Instant::now();
+            while spun.elapsed() < cpu {
+                std::hint::spin_loop();
+            }
+        }
+    }
+
+    /// Graph store attachment awaits this resolution from native tasks: the
+    /// code-index worker and the retained graph-open hook tasks. A debug
+    /// build fails a native task whose poll runs past `LONG_POLL_CPU`, so the
+    /// resolution must not run inside the poll (#4066).
+    #[cfg(all(debug_assertions, unix))]
+    #[test]
+    fn a_graph_resolution_longer_than_one_native_poll_does_not_fail_its_task() {
+        use tracedecay_runtime_core::native_runtime::{self, NativeTask, NativeTaskOutcome};
+        use tracedecay_runtime_core::shard_runtime::registry::StoreRuntimeResolver;
+
+        let temporary = TempDir::new().unwrap();
+        let brain = BrainId::try_from("brain-long-resolve".to_owned()).unwrap();
+        let profile = UserProfileId::try_from("profile-long-resolve".to_owned()).unwrap();
+        let resolver = LocalStoreRuntimeResolverV1::new(LocalProfileStoreAuthorityV1::new(
+            brain.clone(),
+            profile.clone(),
+            temporary.path().canonicalize().unwrap(),
+        ));
+        let key = StoreRuntimeKey::new(
+            StoreShardIdV1::project(
+                brain,
+                profile,
+                ProjectId::try_from("project-long-resolve".to_owned()).unwrap(),
+            ),
+            StoreIncarnationV1::new(1).unwrap(),
+        );
+        INJECTED_RESOLUTION_CPU
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .insert(
+                format!("{:?}", key.shard_id()),
+                native_runtime::LONG_POLL_CPU * 2,
+            );
+        // The unenrolled project does not resolve; only where it ran is under test.
+        let resolution = NativeTask::spawn(async move {
+            let _ = resolver.resolve_graph(&key).await;
+        })
+        .expect("native resolution task");
+        assert_eq!(
+            native_runtime::block_on(resolution.wait()).expect("process runtime"),
+            NativeTaskOutcome::Completed(()),
+        );
+    }
 }