From 29bc30b4d1c150e4cd0751f73c84dd5e3f13969e Mon Sep 17 00:00:00 2001 From: Jeremy Date: Wed, 14 Oct 2026 21:15:02 +0000 Subject: [PATCH 1/4] test(catalog): settle checkpoint observer before query admission --- rust/crates/pumas-core/src/api/catalog.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/rust/crates/pumas-core/src/api/catalog.rs b/rust/crates/pumas-core/src/api/catalog.rs index b63cace7..41d65a4c 100644 --- a/rust/crates/pumas-core/src/api/catalog.rs +++ b/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(); From b43bf6881e7a04f7f1b8d86f5e566bcc22799973 Mon Sep 17 00:00:00 2001 From: Jeremy Date: Wed, 14 Oct 2026 22:31:41 +0000 Subject: [PATCH 2/4] test(managed-python): observe provider exit while awaiting native readiness --- .../src/version_manager/managed_python.rs | 85 ++++++++++++++++--- 1 file changed, 72 insertions(+), 13 deletions(-) diff --git a/rust/crates/pumas-app-manager/src/version_manager/managed_python.rs b/rust/crates/pumas-app-manager/src/version_manager/managed_python.rs index 50296517..7a74dc85 100644 --- a/rust/crates/pumas-app-manager/src/version_manager/managed_python.rs +++ b/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>>, + ) -> Result { + 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::() { - 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::() { + 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::>>()); + 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")] From 299c95afcaa3f486208f999182404541b4c7b9de Mon Sep 17 00:00:00 2001 From: Jeremy Date: Wed, 14 Oct 2026 22:35:27 +0000 Subject: [PATCH 3/4] build(attribution): refresh managed provider source input hashes --- docs/release-attribution/0.8.0-rc.1-s3/inventory.json | 2 +- docs/release-attribution/0.8.0-rc.1/inventory.json | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/release-attribution/0.8.0-rc.1-s3/inventory.json b/docs/release-attribution/0.8.0-rc.1-s3/inventory.json index 90ae02e6..b161580c 100644 --- a/docs/release-attribution/0.8.0-rc.1-s3/inventory.json +++ b/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", diff --git a/docs/release-attribution/0.8.0-rc.1/inventory.json b/docs/release-attribution/0.8.0-rc.1/inventory.json index ee79399b..5ee956df 100644 --- a/docs/release-attribution/0.8.0-rc.1/inventory.json +++ b/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", From ee1948ef6e44e7b4b854d1155d74d347e00708f7 Mon Sep 17 00:00:00 2001 From: Jeremy Date: Wed, 14 Oct 2026 23:10:27 +0000 Subject: [PATCH 4/4] test(custody): await physical-store release before positive reopen --- .../src/version_manager/mod.rs | 103 +++++++++++++++--- 1 file changed, 85 insertions(+), 18 deletions(-) diff --git a/rust/crates/pumas-app-manager/src/version_manager/mod.rs b/rust/crates/pumas-app-manager/src/version_manager/mod.rs index 0f98f0ed..86c1e771 100644 --- a/rust/crates/pumas-app-manager/src/version_manager/mod.rs +++ b/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 { + // 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(),