Stabilize custody and native provider readiness fixtures

docs/release-attribution/0.8.0-rc.1-s3/inventory.json

@@ -9239,7 +9239,7 @@
     "rust/Cargo.lock": "91fb9823de2f0346211b4f68ca400f03a628890e55dca90e0803e2dd7ad47e42",
     "rust/Cargo.toml": "4b30d577dea5f46bb7eb459a2b66a22b534f9671d561207a0a23749d29872fec",
     "rust/crates/pumas-app-manager/Cargo.toml": "f344b89815164c0acc2f6758ad8313a2b8cccba1004354046d542ca5a3d22a73",
-    "rust/crates/pumas-app-manager/src/version_manager/managed_python.rs": "e2e79a8c63500fa060e079edc21484bdc231e1941c6a5ebf65584c3efc95be6f",
+    "rust/crates/pumas-app-manager/src/version_manager/managed_python.rs": "9ce7243a0989bdf49d45d024f4ecbe69fc0964e0cb3feeffaf3e2cb4708f0c0e",
     "rust/crates/pumas-core/Cargo.toml": "d4403c6e15b9d2c3540473253835ed51bb2f0f82339141054636602450ad9c55",
     "rust/crates/pumas-rpc/Cargo.toml": "92b67b71767f6f495895c6d35a7c72107cfccc1cb60dedc400dd7f301a3a97f8",
     "scripts/release/artifact-plan.json": "66348a507f00ffdc74741fc7d70eab02f4a0eed4d3655354784cea2971662064",

docs/release-attribution/0.8.0-rc.1/inventory.json

@@ -8279,7 +8279,7 @@
     "rust/Cargo.lock": "91fb9823de2f0346211b4f68ca400f03a628890e55dca90e0803e2dd7ad47e42",
     "rust/Cargo.toml": "4b30d577dea5f46bb7eb459a2b66a22b534f9671d561207a0a23749d29872fec",
     "rust/crates/pumas-app-manager/Cargo.toml": "f344b89815164c0acc2f6758ad8313a2b8cccba1004354046d542ca5a3d22a73",
-    "rust/crates/pumas-app-manager/src/version_manager/managed_python.rs": "e2e79a8c63500fa060e079edc21484bdc231e1941c6a5ebf65584c3efc95be6f",
+    "rust/crates/pumas-app-manager/src/version_manager/managed_python.rs": "9ce7243a0989bdf49d45d024f4ecbe69fc0964e0cb3feeffaf3e2cb4708f0c0e",
     "rust/crates/pumas-core/Cargo.toml": "d4403c6e15b9d2c3540473253835ed51bb2f0f82339141054636602450ad9c55",
     "rust/crates/pumas-rpc/Cargo.toml": "92b67b71767f6f495895c6d35a7c72107cfccc1cb60dedc400dd7f301a3a97f8",
     "scripts/release/artifact-plan.json": "66348a507f00ffdc74741fc7d70eab02f4a0eed4d3655354784cea2971662064",

rust/crates/pumas-app-manager/src/version_manager/managed_python.rs

@@ -1530,20 +1530,75 @@ mod tests {
         assert!(!rejected_marker.exists());
     }
 
-    #[cfg(any(target_os = "windows", target_os = "macos"))]
-    async fn wait_for_descendant_marker(path: &Path) -> u32 {
-        tokio::time::timeout(Duration::from_secs(10), async {
+    // Allow native shell startup plus 5s for exact-handle acquisition while
+    // staying below the cancellation command's 30s provider deadline.
+    const DESCENDANT_STARTUP_BUDGET: Duration = Duration::from_secs(20);
+
+    async fn wait_for_descendant_marker(
+        path: &Path,
+        running: &mut tokio::task::JoinHandle<ProviderResult<Vec<u8>>>,
+    ) -> Result<u32, String> {
+        tokio::time::timeout(DESCENDANT_STARTUP_BUDGET, async {
             loop {
-                if let Ok(marker) = std::fs::read_to_string(path) {
-                    if let Ok(pid) = marker.trim().parse::<u32>() {
-                        return pid;
+                tokio::select! {
+                    biased;
+                    result = &mut *running => {
+                        return Err(format!(
+                            "provider command completed before descendant readiness: {result:?}"
+                        ));
+                    }
+                    _ = tokio::time::sleep(Duration::from_millis(10)) => {
+                        if let Ok(marker) = std::fs::read_to_string(path) {
+                            if let Ok(pid) = marker.trim().parse::<u32>() {
+                                return Ok(pid);
+                            }
+                        }
                     }
                 }
-                tokio::time::sleep(Duration::from_millis(10)).await;
             }
         })
         .await
-        .expect("provider descendant never started")
+        .map_err(|_| {
+            format!("provider descendant never started within {DESCENDANT_STARTUP_BUDGET:?}")
+        })?
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn descendant_readiness_reports_early_provider_failure() {
+        let root = tempfile::tempdir().unwrap();
+        let mut running = tokio::spawn(async {
+            Err(ManagedPythonFailure::new(
+                ManagedPythonFailureKind::Inconclusive,
+                "controlled provider startup failure",
+            ))
+        });
+        let failure = wait_for_descendant_marker(&root.path().join("missing.pid"), &mut running)
+            .await
+            .unwrap_err();
+        assert!(failure.contains("controlled provider startup failure"));
+        assert!(failure.contains("completed before descendant readiness"));
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn descendant_readiness_allows_slow_start_without_completing_provider() {
+        let root = tempfile::tempdir().unwrap();
+        let marker = root.path().join("delayed.pid");
+        let delayed_marker = marker.clone();
+        let writer = tokio::spawn(async move {
+            tokio::time::sleep(Duration::from_secs(12)).await;
+            std::fs::write(delayed_marker, "321\n").unwrap();
+        });
+        let mut running = tokio::spawn(std::future::pending::<ProviderResult<Vec<u8>>>());
+        assert_eq!(
+            wait_for_descendant_marker(&marker, &mut running)
+                .await
+                .unwrap(),
+            321
+        );
+        assert!(!running.is_finished());
+        writer.await.unwrap();
+        running.abort();
+        assert!(running.await.unwrap_err().is_cancelled());
     }
 
     #[cfg(target_os = "windows")]
@@ -1554,7 +1609,7 @@ mod tests {
                 "-NoProfile",
                 "-NonInteractive",
                 "-Command",
-                "$child = Start-Process powershell.exe -ArgumentList '-NoProfile','-NonInteractive','-Command','Start-Sleep -Seconds 30' -PassThru; Set-Content -LiteralPath $env:PUMAS_TEST_MARKER -Value $child.Id; Start-Sleep -Seconds 30",
+                "$ErrorActionPreference = 'Stop'; $child = Start-Process powershell.exe -ArgumentList '-NoProfile','-NonInteractive','-Command','Start-Sleep -Seconds 30' -PassThru; Set-Content -LiteralPath $env:PUMAS_TEST_MARKER -Value $child.Id; Start-Sleep -Seconds 30",
             ])
             .env("PUMAS_TEST_MARKER", marker);
         command
@@ -1628,12 +1683,14 @@ mod tests {
         let provider = Arc::new(ManagedPythonProvider::new(root.path(), cleanup.clone()).unwrap());
         let marker = root.path().join("cancelled-descendant.pid");
         let command = native_sleeping_tree_command(&marker);
-        let running = tokio::spawn(async move {
+        let mut running = tokio::spawn(async move {
             provider
                 .run_bounded(command, Duration::from_secs(30), 4096)
                 .await
         });
-        let descendant = wait_for_descendant_marker(&marker).await;
+        let descendant = wait_for_descendant_marker(&marker, &mut running)
+            .await
+            .expect("provider descendant readiness failed");
         #[cfg(target_os = "windows")]
         let descendant_handle = open_live_descendant_handle(descendant).await;
         #[cfg(target_os = "macos")]
@@ -1667,11 +1724,13 @@ mod tests {
         const DEADLINE: Duration = Duration::from_secs(60);
         let running_provider = provider.clone();
         let command = native_sleeping_tree_command(&marker);
-        let running =
+        let mut running =
             tokio::spawn(
                 async move { running_provider.run_bounded(command, DEADLINE, 4096).await },
             );
-        let descendant = wait_for_descendant_marker(&marker).await;
+        let descendant = wait_for_descendant_marker(&marker, &mut running)
+            .await
+            .expect("provider descendant readiness failed");
         #[cfg(target_os = "windows")]
         let descendant_handle = open_live_descendant_handle(descendant).await;
         #[cfg(target_os = "macos")]

rust/crates/pumas-app-manager/src/version_manager/mod.rs

@@ -1635,6 +1635,88 @@ mod tests {
     // Keep its bound separate from cancellation and shutdown assertions.
     const FIXTURE_READINESS_TIMEOUT: Duration = Duration::from_secs(10);
 
+    #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
+    async fn reopen_after_fork_release(
+        root: &Path,
+    ) -> pumas_library::Result<pumas_library::PumasApi> {
+        // Positive post-shutdown fixtures only: a parallel fork can retain the
+        // old physical descriptor until exec even after our last handle drops.
+        let physical_root = std::fs::canonicalize(root)?;
+        let busy = format!(
+            "Pumas library instance is already running for physical store {}. Drop existing owner handles before constructing another owner.",
+            physical_root.display()
+        );
+        let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
+        loop {
+            match pumas_library::PumasApi::builder(root)
+                .with_hf_client(false)
+                .with_process_manager(false)
+                .build()
+                .await
+            {
+                Err(pumas_library::PumasError::InvalidParams { message })
+                    if message == busy && tokio::time::Instant::now() < deadline =>
+                {
+                    tokio::time::sleep(Duration::from_millis(5)).await;
+                }
+                result => return result,
+            }
+        }
+    }
+
+    #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
+    #[tokio::test]
+    async fn native_positive_reopen_waits_for_retained_physical_owner() {
+        let root = TempDir::new().unwrap();
+        let api = pumas_library::PumasApi::builder(root.path())
+            .with_hf_client(false)
+            .with_process_manager(false)
+            .build()
+            .await
+            .unwrap();
+        api.shutdown_instance().await.unwrap();
+        // Retain the closed owner's physical share until the explicit release.
+        let (release, released) = tokio::sync::oneshot::channel();
+        let owner = tokio::spawn(async move {
+            released.await.unwrap();
+            drop(api);
+        });
+        let reopening = reopen_after_fork_release(root.path());
+        tokio::pin!(reopening);
+        assert!(
+            tokio::time::timeout(Duration::from_millis(25), &mut reopening)
+                .await
+                .is_err()
+        );
+        release.send(()).unwrap();
+        owner.await.unwrap();
+        let reopened = reopening.await.unwrap();
+        reopened.shutdown_instance().await.unwrap();
+    }
+
+    #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
+    #[tokio::test(start_paused = true)]
+    async fn native_positive_reopen_returns_other_builder_errors_without_retry() {
+        let root = TempDir::new().unwrap();
+        let file = root.path().join("not-a-directory");
+        std::fs::write(&file, b"retain").unwrap();
+        let expected = pumas_library::PumasApi::builder(&file)
+            .with_hf_client(false)
+            .with_process_manager(false)
+            .build()
+            .await
+            .err()
+            .expect("file must be refused");
+        let started = tokio::time::Instant::now();
+        let failure = reopen_after_fork_release(&file)
+            .await
+            .err()
+            .expect("file must remain refused");
+        assert_eq!(failure.to_string(), expected.to_string());
+        assert_eq!(tokio::time::Instant::now(), started);
+        assert_eq!(std::fs::read(file).unwrap(), b"retain");
+    }
+
     async fn create_test_manager() -> (VersionManager, TempDir) {
         let temp_dir = TempDir::new().unwrap();
 
@@ -2369,12 +2451,7 @@ mod tests {
             std::fs::write(&identity_path, serde_json::to_vec(&identity).unwrap()).unwrap();
             drop(manager);
             drop(api);
-            let reopened_api = pumas_library::PumasApi::builder(root.path())
-                .with_hf_client(false)
-                .with_process_manager(false)
-                .build()
-                .await
-                .unwrap();
+            let reopened_api = reopen_after_fork_release(root.path()).await.unwrap();
             let reopened = VersionManager::new_with_acquisition(
                 root.path(),
                 AppId::LlamaCpp,
@@ -2602,12 +2679,7 @@ mod tests {
         // A fresh service and manager must admit a new attempt for the same tag.
         let (retry_server, retry_observed, retry_release, retry_base_url) =
             native_archive_fixture(root.path()).await;
-        let reopened_api = pumas_library::PumasApi::builder(root.path())
-            .with_hf_client(false)
-            .with_process_manager(false)
-            .build()
-            .await
-            .unwrap();
+        let reopened_api = reopen_after_fork_release(root.path()).await.unwrap();
         let mut reopened = VersionManager::new_with_acquisition(
             root.path(),
             AppId::LlamaCpp,
@@ -2871,12 +2943,7 @@ mod tests {
         let source = tokio::net::TcpListener::bind(base_url.strip_prefix("http://").unwrap())
             .await
             .unwrap();
-        let reopened_api = pumas_library::PumasApi::builder(root.path())
-            .with_hf_client(false)
-            .with_process_manager(false)
-            .build()
-            .await
-            .unwrap();
+        let reopened_api = reopen_after_fork_release(root.path()).await.unwrap();
         let new_direct = || {
             VersionInstaller::new(
                 root.path().to_path_buf(),

rust/crates/pumas-core/src/api/catalog.rs

@@ -348,6 +348,11 @@ mod custody_tests {
         );
         let state = api.catalog().unwrap().clone();
         let checkpoint = api.catalog_checkpoint().await.unwrap();
+        // Checkpoint delivery can precede its nested-task observer cleanup.
+        // Start from zero so the later nonzero count identifies the queued query.
+        while state.tasks.owned_task_count_for_catalog_test() != 0 {
+            tokio::task::yield_now().await;
+        }
         let index = state.index.clone();
         let (entered_tx, entered_rx) = mpsc::channel();
         let (release_tx, release_rx) = mpsc::channel();