From f3d4aba54e54bc0f70675cbdeb756383f08187ab Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 12 Oct 2026 07:55:07 +0900 Subject: [PATCH] fix: isolate finite DB and Host execution ownership in fixtures --- .../src/assignment_delivery_tests.rs | 16 +- .../src/assignment_tracking_tests.rs | 18 +- crates/workspace-server/src/auth.rs | 2 +- crates/workspace-server/src/authority.rs | 8 +- crates/workspace-server/src/config_source.rs | 12 +- .../src/frozen_runtime_binding_tests.rs | 2 +- crates/workspace-server/src/hosts.rs | 2 +- .../src/hosts/tests/workspace_auth_tests.rs | 3 +- .../src/legacy_evidence_column_tests.rs | 4 +- crates/workspace-server/src/memory_staging.rs | 2 +- .../src/migration_entrypoint_tests.rs | 6 +- .../workspace-server/src/repository_access.rs | 2 +- crates/workspace-server/src/retention.rs | 2 +- .../src/schema_equivalence_tests.rs | 4 +- crates/workspace-server/src/server.rs | 400 ++++++++++++------ .../src/server/subjektiv_job.rs | 4 +- .../src/server/tests/auth_logging_tests.rs | 6 +- .../src/server/tests/drive_tests.rs | 6 +- .../server/tests/drive_web_roundtrip_tests.rs | 4 +- .../src/server/tests/subject_spawn_tests.rs | 2 +- .../tests/ticket_worker_launch_tests.rs | 2 +- .../src/server/tests/worker_drive_tests.rs | 2 +- .../server/tests/worker_operations_tests.rs | 9 +- .../server/tests/workspace_admission_tests.rs | 2 +- .../src/server/worker_operations.rs | 21 +- .../src/server_assignment_attention_tests.rs | 2 +- .../src/server_assignment_delivery_tests.rs | 4 +- .../src/server_db_execution_tests.rs | 93 +++- .../src/server_ticket_case_signed_tests.rs | 2 +- .../src/server_ticket_tags_tests.rs | 2 +- .../src/server_workspace_config.rs | 6 +- .../src/server_workspace_config_tests.rs | 4 +- .../src/server_workspace_config_wip_tests.rs | 2 +- crates/workspace-server/src/store.rs | 192 +++++---- .../src/ticket_tag_upgrade_tests.rs | 4 +- .../src/workdir_create_operations.rs | 2 +- .../workspace-server/src/workdir_removal.rs | 2 +- .../workspace-server/src/worker_projection.rs | 2 +- .../src/worker_source/runtime_request.rs | 2 +- .../workspace-server/src/workspace_catalog.rs | 28 +- .../src/workspace_deletion.rs | 4 +- .../src/workspace_signing_identity.rs | 6 +- .../src/workspace_subscription.rs | 9 +- 43 files changed, 579 insertions(+), 328 deletions(-) diff --git a/crates/workspace-server/src/assignment_delivery_tests.rs b/crates/workspace-server/src/assignment_delivery_tests.rs index 7a93d3c12..59b9d5a78 100644 --- a/crates/workspace-server/src/assignment_delivery_tests.rs +++ b/crates/workspace-server/src/assignment_delivery_tests.rs @@ -119,7 +119,7 @@ async fn assignment_delivery_response_loss_returns_immutable_payload_after_progr .update_ticket_assignment_tracking("w", "tracking", &update, &source) .unwrap(); drop(store); - let store = SqliteWorkspaceStore::open(directory.path().join("workspace.db")).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(directory.path().join("workspace.db")).unwrap(); assert_eq!( store .list_pending_ticket_assignment_deliveries("w", 100) @@ -180,7 +180,7 @@ async fn assignment_delivery_simultaneous_reservations_converge_to_one_receipt() let path = directory.path().join("workspace.db"); let input = pending.input_event_id.clone(); std::thread::spawn(move || { - let store = SqliteWorkspaceStore::open(path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(path).unwrap(); barrier.wait(); store .reserve_ticket_assignment_delivery("w", "a", &input, &tracking_store_source()) @@ -540,7 +540,7 @@ async fn assignment_delivery_rendered_content_is_immutable_across_restart_and_co "original rendered prompt" ); drop(store); - let store = SqliteWorkspaceStore::open(directory.path().join("workspace.db")).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(directory.path().join("workspace.db")).unwrap(); assert_eq!( store .prepare_ticket_assignment_delivery_content( @@ -606,7 +606,7 @@ fn assignment_delivery_schema92_upgrade_uses_reconciled_migration_evidence_not_l let operations = stored_fixture_rows(&conn, "ticket_assignment_operations"); let tracking = stored_fixture_rows(&conn, "typed_ticket_assignment_tracking_history"); drop(conn); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); let pending = store .list_pending_ticket_assignment_deliveries("w", 100) .unwrap(); @@ -656,7 +656,7 @@ async fn assignment_startup_simultaneous_reservations_admit_runtime_effects_once let barrier = barrier.clone(); let path = directory.path().join("workspace.db"); std::thread::spawn(move || { - let store = SqliteWorkspaceStore::open(path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(path).unwrap(); barrier.wait(); delivery_store_start_reserve(&store, "start-a", "a", &tracking_store_source()) .unwrap() @@ -688,8 +688,8 @@ fn assignment_delivery_fresh_and_upgraded_operational_schemas_are_identical() { let conn = tracking_schema91_fixture(&path); migrate_ticket_tracking_v91_to_v92(&conn).unwrap(); drop(conn); - let upgraded = SqliteWorkspaceStore::open(&path).unwrap(); - let fresh = SqliteWorkspaceStore::in_memory().unwrap(); + let upgraded = SqliteWorkspaceStore::open_for_test(&path).unwrap(); + let fresh = SqliteWorkspaceStore::in_memory_for_test().unwrap(); let objects = |store: &SqliteWorkspaceStore| { store.with_conn(|conn| { let mut statement=conn.prepare("SELECT type,name,sql FROM sqlite_schema WHERE name LIKE 'ticket_assignment_delivery_%' OR name LIKE 'ticket_assignment_operation_fence%' ORDER BY name")?; @@ -713,7 +713,7 @@ async fn assignment_delivery_missing_immutable_request_guard_refuses_startup() { .unwrap(); drop(store); assert!( - matches!(SqliteWorkspaceStore::open(directory.path().join("workspace.db")),Err(Error::Store(message)) if message.contains("ticket_assignment_delivery_request_no_update")) + matches!(SqliteWorkspaceStore::open_for_test(directory.path().join("workspace.db")),Err(Error::Store(message)) if message.contains("ticket_assignment_delivery_request_no_update")) ); } diff --git a/crates/workspace-server/src/assignment_tracking_tests.rs b/crates/workspace-server/src/assignment_tracking_tests.rs index 7c85c4812..8bafcfb44 100644 --- a/crates/workspace-server/src/assignment_tracking_tests.rs +++ b/crates/workspace-server/src/assignment_tracking_tests.rs @@ -78,7 +78,7 @@ async fn tracking_store_every_replay_fences_current_singleton_and_authenticated_ ); // Model the authority change after authenticated source capture, before // BEGIN IMMEDIATE on a second connection. Receipt presence cannot bypass it. - let second = SqliteWorkspaceStore::open(directory.path().join("workspace.db")).unwrap(); + let second = SqliteWorkspaceStore::open_for_test(directory.path().join("workspace.db")).unwrap(); second.with_conn_mut(|conn| { let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?; match change { @@ -233,7 +233,7 @@ async fn tracking_store_same_principal_reissue_and_exact_retries_keep_original_r .is_err() ); drop(store); - let reopened = SqliteWorkspaceStore::open(directory.path().join("workspace.db")).unwrap(); + let reopened = SqliteWorkspaceStore::open_for_test(directory.path().join("workspace.db")).unwrap(); assert_eq!( reopened .get_ticket_assignment_tracking("w", "tracking", "a") @@ -441,7 +441,7 @@ fn tracking_store_real_schema91_to92_retains_queue_cases_principals_audit_receip .collect::>(), vec![92, 93] ); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); store .with_conn(|conn| { assert_eq!(current_schema_version(conn)?, 93); @@ -522,7 +522,7 @@ fn tracking_store_real_schema91_to92_retains_queue_cases_principals_audit_receip .list_ticket_assignment_tracking_history("w", "pending", "a-pending") .unwrap(); drop(store); - let reopened = SqliteWorkspaceStore::open(&path).unwrap(); + let reopened = SqliteWorkspaceStore::open_for_test(&path).unwrap(); assert_eq!( reopened .list_ticket_assignment_tracking_history("w", "pending", "a-pending") @@ -547,7 +547,7 @@ fn tracking_store_schema91_closed_unknown_assignment_uses_assign_audit_for_first let releases_before = stored_fixture_rows(&conn, "ticket_assignment_work_releases"); drop(conn); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); let backend = ticket::SqliteTicketBackend::open_verified(&path, "w").unwrap(); let reopen = || { let observed = backend.show("historical-done".into()).unwrap(); @@ -642,7 +642,7 @@ fn tracking_store_schema91_closed_unknown_assignment_uses_assign_audit_for_first ); // A second observer holds the same Assign baseline. Only the first // explicit writer may consume it; BEGIN IMMEDIATE serializes the CAS. - let second = SqliteWorkspaceStore::open(&path).unwrap(); + let second = SqliteWorkspaceStore::open_for_test(&path).unwrap(); assert_eq!( second .get_ticket_assignment_tracking("w", "historical-done", "a-historical-done") @@ -737,7 +737,7 @@ fn tracking_store_schema91_closed_unknown_assignment_uses_assign_audit_for_first }).unwrap(); drop(second); drop(store); - let reopened_store = SqliteWorkspaceStore::open(&path).unwrap(); + let reopened_store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); assert_eq!( reopened_store .get_ticket_assignment_tracking("w", "historical-done", "a-historical-done") @@ -793,7 +793,7 @@ fn tracking_store_schema91_missing_assign_audit_remains_unknown_without_cas_base ) .unwrap(); drop(conn); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); let unknown = store .get_ticket_assignment_tracking("w", "historical-done", "a-historical-done") .unwrap() @@ -1065,7 +1065,7 @@ async fn orchestrator_assignment_all_runtime_retries_fence_singleton_and_binding &source, ) .unwrap(); - let second = SqliteWorkspaceStore::open(directory.path().join("workspace.db")).unwrap(); + let second = SqliteWorkspaceStore::open_for_test(directory.path().join("workspace.db")).unwrap(); second.with_conn_mut(|conn| {let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;match change { "replace" => {tx.execute("UPDATE worker_singleton_owners SET worker_id='replacement',generation=2 WHERE workspace_id='w'",[])?;}, "revoke" => {tx.execute("UPDATE workspace_runtime_bindings SET revoked_at='revoked' WHERE workspace_id='w'",[])?;}, diff --git a/crates/workspace-server/src/auth.rs b/crates/workspace-server/src/auth.rs index bfe4ab4c3..b5badef0d 100644 --- a/crates/workspace-server/src/auth.rs +++ b/crates/workspace-server/src/auth.rs @@ -203,7 +203,7 @@ mod tests { }) .unwrap(); let store = Arc::new( - SqliteWorkspaceStore::in_memory() + SqliteWorkspaceStore::in_memory_for_test() .unwrap() .with_db_executor(executor), ); diff --git a/crates/workspace-server/src/authority.rs b/crates/workspace-server/src/authority.rs index abda3e7b2..950a5665a 100644 --- a/crates/workspace-server/src/authority.rs +++ b/crates/workspace-server/src/authority.rs @@ -3241,7 +3241,7 @@ mod tests { async fn unfinished_dependencies_remain_diagnostic_without_disabling_manual_start() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("workspace.db"); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "w".into(), @@ -3316,7 +3316,7 @@ mod tests { async fn manual_worker_eligibility_accepts_explicit_states_without_reviving_released_work() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("workspace.db"); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "w".into(), @@ -3449,7 +3449,7 @@ mod tests { write_ticket(dir.path(), "00000000001J5", "Second ticket", "queued"); write_ticket(dir.path(), "00000000001J6", "Third ticket", "planning"); let db_path = dir.path().join("workspace.db"); - let store = SqliteWorkspaceStore::open(&db_path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&db_path).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-test".to_string(), @@ -3881,7 +3881,7 @@ VALUES ('workspace-test', 'ticket', 4); async fn objective_mutations_write_sqlite_records_and_audit_events() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().join("workspace.db"); - let store = SqliteWorkspaceStore::open(&db_path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&db_path).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-test".to_string(), diff --git a/crates/workspace-server/src/config_source.rs b/crates/workspace-server/src/config_source.rs index 58158f464..7ac608622 100644 --- a/crates/workspace-server/src/config_source.rs +++ b/crates/workspace-server/src/config_source.rs @@ -795,7 +795,7 @@ mod tests { } async fn open_store() -> SqliteWorkspaceStore { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); store.upsert_workspace(&workspace()).await.unwrap(); store } @@ -1454,7 +1454,7 @@ mod tests { #[tokio::test] async fn invalid_candidate_is_never_persisted() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); store.upsert_workspace(&workspace()).await.unwrap(); let current = store.load_workspace_config("w-config").unwrap().unwrap(); let main = current.snapshot.get(&path(MAIN_CONFIG_ENTRYPOINT)).unwrap(); @@ -1478,7 +1478,7 @@ mod tests { #[tokio::test] async fn valid_candidate_commits_snapshot_content_and_provenance_atomically() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); store.upsert_workspace(&workspace()).await.unwrap(); let current = store.load_workspace_config("w-config").unwrap().unwrap(); let committed = store @@ -1496,7 +1496,7 @@ mod tests { #[tokio::test] async fn stale_cas_cannot_overwrite_newer_tree() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); store.upsert_workspace(&workspace()).await.unwrap(); let current = store.load_workspace_config("w-config").unwrap().unwrap(); let request = commit_request(¤t, vec![update_main(¤t, "{ answer = 42; }")]); @@ -1514,7 +1514,7 @@ mod tests { #[tokio::test] async fn committed_content_remains_retrievable_after_later_commit() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); store.upsert_workspace(&workspace()).await.unwrap(); let current = store.load_workspace_config("w-config").unwrap().unwrap(); let first = store @@ -1772,7 +1772,7 @@ mod tests { #[test] fn migration_creates_config_authority_tables() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); store .with_conn(|conn| { for table in [ diff --git a/crates/workspace-server/src/frozen_runtime_binding_tests.rs b/crates/workspace-server/src/frozen_runtime_binding_tests.rs index a23c29b04..9c8bef71f 100644 --- a/crates/workspace-server/src/frozen_runtime_binding_tests.rs +++ b/crates/workspace-server/src/frozen_runtime_binding_tests.rs @@ -818,7 +818,7 @@ mod current_authority { use super::*; fn binding_store() -> (SqliteWorkspaceStore, WorkspaceRuntimeBinding) { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); let runtime_key = worker_runtime::auth::RuntimeIdentityMaterial::generate("runtime-key").unwrap(); store.with_conn(|conn| { diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 765a557a7..0b41a9a47 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -6394,7 +6394,7 @@ mod tests { }) .unwrap(); let store = Arc::new( - SqliteWorkspaceStore::in_memory() + SqliteWorkspaceStore::in_memory_for_test() .unwrap() .with_db_executor(executor.clone()), ); diff --git a/crates/workspace-server/src/hosts/tests/workspace_auth_tests.rs b/crates/workspace-server/src/hosts/tests/workspace_auth_tests.rs index f68964f35..29c0cb52c 100644 --- a/crates/workspace-server/src/hosts/tests/workspace_auth_tests.rs +++ b/crates/workspace-server/src/hosts/tests/workspace_auth_tests.rs @@ -32,7 +32,8 @@ struct Fixture { impl Fixture { async fn new() -> Self { let temp = tempfile::tempdir().unwrap(); - let store = Arc::new(SqliteWorkspaceStore::open(temp.path().join("server.db")).unwrap()); + let store = + Arc::new(SqliteWorkspaceStore::open_for_test(temp.path().join("server.db")).unwrap()); store .upsert_account(&AccountRecord { account_id: "owner".into(), diff --git a/crates/workspace-server/src/legacy_evidence_column_tests.rs b/crates/workspace-server/src/legacy_evidence_column_tests.rs index a59f2a0ca..33994b292 100644 --- a/crates/workspace-server/src/legacy_evidence_column_tests.rs +++ b/crates/workspace-server/src/legacy_evidence_column_tests.rs @@ -45,7 +45,7 @@ fn evidence_column_cutover_preserves_all_values_and_current_schema_names() { let artifacts = evidence_rows(&conn, "artifacts"); let receipts = evidence_rows(&conn, "repository_secret_legacy_receipts"); drop(conn); - let upgraded = SqliteWorkspaceStore::open(&path).unwrap(); + let upgraded = SqliteWorkspaceStore::open_for_test(&path).unwrap(); upgraded .with_conn(|conn| { assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION); @@ -73,7 +73,7 @@ fn evidence_column_cutover_preserves_all_values_and_current_schema_names() { Ok(()) }) .unwrap(); - for store in [upgraded, SqliteWorkspaceStore::in_memory().unwrap()] { + for store in [upgraded, SqliteWorkspaceStore::in_memory_for_test().unwrap()] { store.with_conn(|conn| { let mut stmt = conn.prepare("SELECT name,sql FROM sqlite_schema WHERE sql IS NOT NULL AND name NOT LIKE 'sqlite_%'")?; let objects = stmt.query_map([],|row|Ok((row.get::<_,String>(0)?,row.get::<_,String>(1)?)))?.collect::>>()?; diff --git a/crates/workspace-server/src/memory_staging.rs b/crates/workspace-server/src/memory_staging.rs index ff8caadf7..631d1b33e 100644 --- a/crates/workspace-server/src/memory_staging.rs +++ b/crates/workspace-server/src/memory_staging.rs @@ -217,7 +217,7 @@ mod tests { async fn authority() -> (TempDir, SqliteWorkspaceAuthority) { let temp = TempDir::new().unwrap(); let db_path = temp.path().join("workspace.db"); - let store = SqliteWorkspaceStore::open(&db_path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&db_path).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-test".to_string(), diff --git a/crates/workspace-server/src/migration_entrypoint_tests.rs b/crates/workspace-server/src/migration_entrypoint_tests.rs index d0eb2375d..41329a730 100644 --- a/crates/workspace-server/src/migration_entrypoint_tests.rs +++ b/crates/workspace-server/src/migration_entrypoint_tests.rs @@ -32,7 +32,7 @@ fn opening_schema_84_with_uncertain_removal_preserves_old_server_recovery() { drop(conn); for attempt in 0..3 { let error = match attempt { - 0 => SqliteWorkspaceStore::open(&path).err().unwrap(), + 0 => SqliteWorkspaceStore::open_for_test(&path).err().unwrap(), 1 => SqliteWorkspaceStore::migration_plan(&path).unwrap_err(), _ => SqliteWorkspaceStore::migrate_database(&path).unwrap_err(), }; @@ -62,7 +62,7 @@ fn opening_schema_84_with_uncertain_removal_preserves_old_server_recovery() { ) .unwrap(); drop(conn); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); store.with_conn(|conn| { assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION); assert_eq!(conn.query_row("SELECT input_fingerprint FROM worker_removal_operations WHERE operation_id='removal'",[],|r|r.get::<_,String>(0))?,"legacy-receipt"); @@ -109,7 +109,7 @@ fn migration_dry_run_and_apply_share_the_original_repository_secret_authority() assert!(!table_exists(&conn, "server_secret_objects").unwrap()); drop(conn); assert_eq!(SqliteWorkspaceStore::migrate_database(&path).unwrap(), plan); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); store.with_conn(|conn| { assert_eq!(current_schema_version(conn)?,LATEST_SCHEMA_VERSION); let (nonce, mut sealed): (Vec,Vec) = conn.query_row( diff --git a/crates/workspace-server/src/repository_access.rs b/crates/workspace-server/src/repository_access.rs index 304909848..c8a13c3f4 100644 --- a/crates/workspace-server/src/repository_access.rs +++ b/crates/workspace-server/src/repository_access.rs @@ -2290,7 +2290,7 @@ mod tests { ) { let dir = tempfile::tempdir().unwrap(); let database = dir.path().join("server.db"); - let store = Arc::new(SqliteWorkspaceStore::open(&database).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&database).unwrap()); let timestamp = now(); for workspace_id in ["workspace-a", "workspace-b"] { futures::executor::block_on(store.upsert_workspace(&WorkspaceRecord { diff --git a/crates/workspace-server/src/retention.rs b/crates/workspace-server/src/retention.rs index aa6ca692d..4aa310885 100644 --- a/crates/workspace-server/src/retention.rs +++ b/crates/workspace-server/src/retention.rs @@ -1116,7 +1116,7 @@ mod tests { WorkerId::from_legacy_u64(1) } fn setup() -> SqliteWorkspaceStore { - let s = SqliteWorkspaceStore::in_memory().unwrap(); + let s = SqliteWorkspaceStore::in_memory_for_test().unwrap(); s.with_conn(|c| { c.execute( "INSERT INTO accounts(account_id,kind,handle,display_name,created_at,updated_at) \ diff --git a/crates/workspace-server/src/schema_equivalence_tests.rs b/crates/workspace-server/src/schema_equivalence_tests.rs index 37f5e218a..4063a675b 100644 --- a/crates/workspace-server/src/schema_equivalence_tests.rs +++ b/crates/workspace-server/src/schema_equivalence_tests.rs @@ -38,8 +38,8 @@ fn fresh_and_upgraded_state_contract_schemas_enforce_the_same_constraints() { let temp = tempfile::tempdir().unwrap(); let path = temp.path().join("upgraded.db"); prepare_retained_schema(&path, 85); - let upgraded = SqliteWorkspaceStore::open(&path).unwrap(); - let fresh = SqliteWorkspaceStore::in_memory().unwrap(); + let upgraded = SqliteWorkspaceStore::open_for_test(&path).unwrap(); + let fresh = SqliteWorkspaceStore::in_memory_for_test().unwrap(); let mut mismatches = Vec::new(); for table in [ "workspace_tags", diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 3dc7aa8fe..938cc3357 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -1589,6 +1589,7 @@ fn orchestrator_attention_state( #[derive(Clone)] pub struct WorkspaceApi { + host_executors: HostExecutors, pub(crate) config: ServerConfig, pub(crate) store: Arc, config_store: Arc, @@ -1687,6 +1688,7 @@ impl ServerApiContractService { struct WorkerRemovalService { workspace_id: String, store: Arc, + host_executors: HostExecutors, runtime: Weak, workdir_sessions: Arc>, workdir_session_locks: Arc>>>>, @@ -1700,6 +1702,7 @@ impl WorkerRemovalService { Self { workspace_id: api.config.workspace_id.clone(), store: api.store.clone(), + host_executors: api.host_executors.clone(), runtime: Arc::downgrade(&api.runtime), workdir_sessions: api.workdir_sessions.clone(), workdir_session_locks: api.workdir_session_locks.clone(), @@ -1870,12 +1873,15 @@ impl WorkerRemovalService { let runtime_id = target_runtime_id.to_owned(); let worker_id = target_worker_id.to_owned(); let reason = reason.to_owned(); - run_host_workflow(async move { - service - .execute_async_on_host(source, &runtime_id, &worker_id, &reason) - .await - .map_err(Error::Store) - }) + run_host_workflow( + self.host_executors.host().map_err(|e| e.to_string())?, + async move { + service + .execute_async_on_host(source, &runtime_id, &worker_id, &reason) + .await + .map_err(Error::Store) + }, + ) .await .map_err(|error| error.to_string()) } @@ -1976,12 +1982,15 @@ impl WorkerRemovalService { let target = target.clone(); let reason = reason.to_owned(); let parent = parent_workspace_deletion_operation_id.map(str::to_owned); - run_host_workflow(async move { - service - .execute_target_removal_on_host(&runtime, &target, &reason, parent.as_deref()) - .await - .map_err(Error::Store) - }) + run_host_workflow( + self.host_executors.host().map_err(|e| e.to_string())?, + async move { + service + .execute_target_removal_on_host(&runtime, &target, &reason, parent.as_deref()) + .await + .map_err(Error::Store) + }, + ) .await .map_err(|error| error.to_string()) } @@ -2235,6 +2244,7 @@ use workspace_admission::WorkspaceAdmission; #[derive(Clone)] pub struct WorkspaceServerApi { + host_executors: HostExecutors, template: Arc, store: Arc, catalog: WorkspaceCatalogService, @@ -2265,6 +2275,7 @@ impl WorkspaceServerApi { workspace_signing_material_root(&template.database_path), )); Self { + host_executors: HostExecutors::shared(), template: Arc::new(template), catalog: WorkspaceCatalogService::new(store.clone(), signing_materials.clone()), signing_materials, @@ -2280,6 +2291,13 @@ impl WorkspaceServerApi { } } + #[cfg(test)] + fn new_for_test(template: ServerConfig, store: Arc) -> Self { + let mut api = Self::new(template, store); + api.host_executors = HostExecutors::default(); + api + } + // Own the complete synchronous store stage; callers retain their async // lifecycle guards across it and never split a transaction around await. async fn execute_store_stage( @@ -2312,7 +2330,12 @@ impl WorkspaceServerApi { let config = self .template .for_catalog_workspace(&workspace, repositories)?; - let api = WorkspaceApi::new(config, self.store.clone()).await?; + let api = WorkspaceApi::new_with_host_executors( + config, + self.store.clone(), + self.host_executors.clone(), + ) + .await?; apis.insert(workspace_id.to_string(), api.clone()); Ok(Some(api)) } @@ -2387,7 +2410,7 @@ impl WorkspaceServerApi { let workers = self .execute_store_stage(move |store| store.list_worker_registry(&id, i64::MAX as usize)) .await?; - host_io_executor()?.run(move || -> Result<_> { + api.host_executors.host()?.run(move || -> Result<_> { for registry_worker in workers { let worker_key = registry_worker.display_name; match api.runtime.worker(®istry_worker.worker) { @@ -2493,7 +2516,9 @@ impl WorkspaceServerApi { let removal_api = api.clone(); let workdir_id = workdir.workdir_id.clone(); let parent_id = operation_id.to_owned(); - match host_io_executor()? + match api + .host_executors + .host()? .run(move || { execute_workdir_removal_for_workspace_deletion( &removal_api, @@ -3487,6 +3512,19 @@ impl WorkspaceApi { } pub async fn new(config: ServerConfig, store: Arc) -> Result { + Self::new_with_host_executors(config, store, HostExecutors::shared()).await + } + + #[cfg(test)] + async fn new_for_test(config: ServerConfig, store: Arc) -> Result { + Self::new_with_host_executors(config, store, HostExecutors::default()).await + } + + async fn new_with_host_executors( + config: ServerConfig, + store: Arc, + host_executors: HostExecutors, + ) -> Result { let resource_broker = BackendResourceBroker::default(); if store.get_workspace(&config.workspace_id).await?.is_none() { return Err(crate::Error::Config(format!( @@ -3563,6 +3601,7 @@ impl WorkspaceApi { store, Arc::new(execution_backend), resource_broker, + host_executors, Some(worker_remove_dispatcher), ) .await?; @@ -3623,6 +3662,7 @@ impl WorkspaceApi { store, execution_backend, BackendResourceBroker::default(), + HostExecutors::default(), None, ) .await @@ -3633,6 +3673,7 @@ impl WorkspaceApi { store: Arc, execution_backend: Arc, resource_broker: BackendResourceBroker, + host_executors: HostExecutors, worker_remove_dispatcher: Option< Arc, >, @@ -3783,6 +3824,7 @@ impl WorkspaceApi { let drive_host = Arc::new(crate::workspace_drive_host::WorkspaceDriveHost::new(&feature_storage).await?); let mut api = Self { + host_executors, drive_host, config_store, repository_secrets, @@ -3877,11 +3919,13 @@ impl WorkspaceApi { schedule_external_workdir_expiry(&api, &grant)?; } let workdir_recovery_api = api.clone(); - host_io_executor()? + api.host_executors + .host()? .run(move || recover_workdir_removals(&workdir_recovery_api)) .await?; recover_runtime_removals(&api).await; - host_io_executor()? + api.host_executors + .host()? .run(move || { let recovery_now = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true); let deliveries = api @@ -4323,7 +4367,7 @@ impl WorkspaceApi { let attempt_id = attempt.attempt_id.clone(); // Admit before claiming: overload must leave durable Pending/Failed // ownership recoverable, never strand a Running claim without a job. - let admission = backend_job_cleanup_executor().and_then(|executor| { + let admission = self.host_executors.cleanup().and_then(|executor| { executor.dispatch(move || { let cleanup = || -> Result<()> { let (attempt, claimed) = api.store.claim_backend_job_worker_cleanup( @@ -4679,7 +4723,7 @@ impl WorkspaceApi { let attempt_id = attempt_id.to_string(); handle.spawn(async move { tokio::time::sleep(delay).await; - let admission = host_io_executor().and_then(|executor| executor.dispatch(move || { + let admission = api.host_executors.host().and_then(|executor| executor.dispatch(move || { let Ok(Some(attempt)) = api.store .get_backend_job_attempt(&api.config.workspace_id, &job_id, &attempt_id) @@ -11686,7 +11730,9 @@ impl server_api::ServerApi for ServerApiContractService { let api = self.workspace_api()?.clone(); let headers = current_worker_contract_headers(&context)?; let stage_api = api.clone(); - let worker = host_io_executor() + let worker = api + .host_executors + .host() .map_err(ApiError::from) .map_err(ApiError::into_repository_api_error)? .run(move || -> ApiResult<_> { @@ -11720,7 +11766,9 @@ impl server_api::ServerApi for ServerApiContractService { let api = self.workspace_api()?.clone(); let headers = current_worker_contract_headers(&context)?; let stage_api = api.clone(); - let worker = host_io_executor() + let worker = api + .host_executors + .host() .map_err(ApiError::from) .map_err(ApiError::into_repository_api_error)? .run(move || -> ApiResult<_> { @@ -13774,7 +13822,7 @@ async fn start_manual_ticket_worker_assignment( let operation_id = operation_id.to_owned(); let selected = selected.to_vec(); let delegation = delegation.cloned(); - run_host_workflow(async move { + run_host_workflow(api.host_executors.host()?, async move { start_manual_ticket_worker_assignment_on_host( &api, &record, @@ -14263,7 +14311,9 @@ async fn scoped_set_ticket_assignment_with_context( validate_workspace_scope(&api, &path.0.0)?; let source_api = api.clone(); let context = context.clone(); - let (author, source) = host_io_executor()? + let (author, source) = api + .host_executors + .host()? .run(move || authenticated_assignment_source(&source_api, &headers, &context)) .await?; apply_set_ticket_assignment(State(api), path, request, author, Some(source)).await @@ -14415,7 +14465,9 @@ async fn scoped_clear_ticket_assignment_with_context( validate_workspace_scope(&api, &path.0.0)?; let source_api = api.clone(); let context = context.clone(); - let (author, source) = host_io_executor()? + let (author, source) = api + .host_executors + .host()? .run(move || authenticated_assignment_source(&source_api, &headers, &context)) .await?; apply_clear_ticket_assignment(State(api), path, query, author, Some(source)).await @@ -15263,7 +15315,7 @@ fn schedule_ticket_item_check( } }; let api = api.clone(); - let admission = host_io_executor().and_then(|executor| { + let admission = api.host_executors.host().and_then(|executor| { executor.dispatch(move || { if let Err(error) = api.dispatch_backend_job_reservation(reservation) { tracing::warn!( @@ -15319,7 +15371,7 @@ async fn execute_ticket_rest_operation_with_context( let job_api = api.clone(); let workspace_id = workspace_id.to_owned(); let context = context.cloned(); - host_io_executor()?.run(move || { + api.host_executors.host()?.run(move || { let api = &job_api; let workspace_id = workspace_id.as_str(); let context = context.as_ref(); @@ -16483,7 +16535,7 @@ async fn scoped_list_merge_requests( AxumPath(workspace_id): AxumPath, Query(query): Query, ) -> ApiResult> { - host_io_executor()?.run(move || { + api.host_executors.host()?.run(move || { let workspace_id = parse_workspace_id(&workspace_id)?; require_workspace_access(&workspace_id, &api)?; @@ -16700,7 +16752,8 @@ async fn scoped_show_merge_request( AxumPath((workspace_id, merge_request_id)): AxumPath<(String, String)>, Query(query): Query, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { let workspace_id = parse_workspace_id(&workspace_id)?; require_workspace_access(&workspace_id, &api)?; @@ -16795,7 +16848,8 @@ async fn scoped_merge_request_readiness( State(api): State, AxumPath((workspace_id, merge_request_id)): AxumPath<(String, String)>, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { let workspace_id = parse_workspace_id(&workspace_id)?; let store = merge_request_store(&api, &workspace_id)?; @@ -16869,7 +16923,8 @@ async fn scoped_open_merge_request( AxumPath((workspace_id, ticket_id)): AxumPath<(String, String)>, Json(input): Json, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { let workspace_id = parse_workspace_id(&workspace_id)?; let ticket_id = resolve_workspace_ticket_reference(&api, &workspace_id, &ticket_id)?; @@ -16948,7 +17003,8 @@ async fn scoped_merge_request_thread( AxumPath((workspace_id, merge_request_id)): AxumPath<(String, String)>, Query(query): Query, ) -> ApiResult>> { - host_io_executor()? + api.host_executors + .host()? .run(move || { let workspace_id = parse_workspace_id(&workspace_id)?; let events = merge_request_store(&api, &workspace_id)?.thread_page_by_id( @@ -17019,7 +17075,8 @@ async fn scoped_register_reviewer_child_session( AxumPath(workspace_id): AxumPath, Json(input): Json, ) -> ApiResult { - host_io_executor()? + api.host_executors + .host()? .run(move || { let workspace_id = parse_workspace_id(&workspace_id)?; require_workspace_access(&workspace_id, &api)?; @@ -17045,7 +17102,8 @@ async fn scoped_register_merge_request_review_capability( AxumPath((workspace_id, merge_request_id)): AxumPath<(String, String)>, Json(input): Json, ) -> ApiResult { - host_io_executor()? + api.host_executors + .host()? .run(move || { let workspace_id = parse_workspace_id(&workspace_id)?; require_workspace_access(&workspace_id, &api)?; @@ -17160,7 +17218,8 @@ async fn scoped_submit_merge_request_review( AxumPath((workspace_id, merge_request_id)): AxumPath<(String, String)>, Json(input): Json, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { let workspace_id = parse_workspace_id(&workspace_id)?; let store = merge_request_store(&api, &workspace_id)?; @@ -17235,7 +17294,8 @@ async fn scoped_revoke_merge_request_review( AxumPath((workspace_id, merge_request_id)): AxumPath<(String, String)>, Json(input): Json, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { let workspace_id = parse_workspace_id(&workspace_id)?; require_workspace_access(&workspace_id, &api)?; @@ -17287,7 +17347,8 @@ async fn scoped_complete_merge_request( AxumPath((workspace_id, merge_request_id)): AxumPath<(String, String)>, Json(input): Json, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { let workspace_id = parse_workspace_id(&workspace_id)?; require_workspace_access(&workspace_id, &api)?; @@ -18396,7 +18457,8 @@ async fn open_current_worker_workdir_session_locked( // The accepted Runtime authorization must retain the actual session lock, // not just the async waiter that submitted it. let stage_guard = session_guard.clone(); - host_io_executor()? + api.host_executors + .host()? .run(move || { let _session_guard = stage_guard; stage_api @@ -18634,7 +18696,8 @@ async fn sync_runtime_worker_workdir_attachments_locked( let stage_api = api.clone(); let stage_worker = worker.clone(); let stage_guard = session_guard.clone(); - host_io_executor()? + api.host_executors + .host()? .run(move || { let _session_guard = stage_guard; replace_runtime_worker_workdir_attachments(&stage_api, &stage_worker, attachments) @@ -18695,7 +18758,8 @@ async fn scoped_submit_backend_job_result( headers: HeaderMap, Json(request): Json, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { validate_workspace_scope(&api, &path.workspace_id)?; let worker = current_worker_identity(&api, &path.workspace_id, &headers)?; @@ -18771,7 +18835,9 @@ async fn import_current_worker_workdir_observation( let stage_api = api.clone(); let stage_worker = worker.clone(); let stage_guard = session_guard.cloned(); - let observed_worker = host_io_executor()? + let observed_worker = api + .host_executors + .host()? .run(move || { let _session_guard = stage_guard; stage_api @@ -18802,7 +18868,9 @@ async fn scoped_attach_current_worker_workdir( Json(request): Json, ) -> ApiResult> { let stage_api = api.clone(); - let worker = host_io_executor()? + let worker = api + .host_executors + .host()? .run(move || -> ApiResult<_> { validate_workspace_scope(&stage_api, &path.workspace_id)?; Ok(current_worker_identity( @@ -19123,7 +19191,9 @@ async fn scoped_detach_current_worker_workdir( let alias = workdir::WorkdirAttachmentAlias::new(alias) .map_err(|error| Error::InvalidInput(error.to_string()))?; let stage_api = api.clone(); - let worker = host_io_executor()? + let worker = api + .host_executors + .host()? .run(move || current_worker_identity(&stage_api, &workspace_id, &headers)) .await?; detach_current_worker_workdir( @@ -19383,7 +19453,7 @@ async fn scoped_execute_current_worker_workdir_operation( validate_workspace_scope(&api, &path.workspace_id)?; context.stage = "worker_identity"; let stage_api = api.clone(); - let worker = host_io_executor()?.run(move || { + let worker = api.host_executors.host()?.run(move || { current_worker_identity(&stage_api, &path.workspace_id, &headers) }).await?; context.runtime_id_hash = Some(crate::worker_source::diagnostic_id_hash(&worker.runtime_id)); @@ -19608,7 +19678,7 @@ async fn run_orchestrator_turn_end_hook(api: WorkspaceApi) { let mut worker_states = HashMap::new(); while let Some(update) = subscription.recv().await { let stage_api = api.clone(); - let result = host_io_executor().map(|executor| async move { + let result = api.host_executors.host().map(|executor| async move { executor .run(move || -> Result<_> { let keep_running = @@ -19714,7 +19784,7 @@ fn enqueue_orchestrator_assignment_attention( ) -> Option> { let api = api.clone(); let (done, completion) = tokio::sync::oneshot::channel(); - match host_io_executor().and_then(|executor| { + match api.host_executors.host().and_then(|executor| { executor.dispatch(move || { if let Some(orchestrator) = find_workspace_orchestrator(&api) { dispatch_orchestrator_assignment_attention(&api, &orchestrator.worker); @@ -20180,33 +20250,39 @@ const SUBJEKTIV_MEMORY_CONSOLIDATION_PROFILE: &str = "subjektiv-memory-consolida // Teardown waits must not occupy the ordinary Host lane: a Job may need a // user stop/control operation on that lane before its Run can finish. Cleanup // therefore has its own finite owner, still independent of the DB callback lane. -fn backend_job_cleanup_executor() -> Result { - static CLEANUP: std::sync::OnceLock> = - std::sync::OnceLock::new(); - CLEANUP - .get_or_init(|| { - crate::db_executor::DbExecutor::new(crate::db_executor::DbExecutorLimits { - workers: std::num::NonZeroUsize::new(2).unwrap(), - queued: std::num::NonZeroUsize::new(16).unwrap(), - }) - .ok() - }) - .clone() - .ok_or(Error::DbExecutorUnavailable) +#[derive(Clone, Default)] +struct HostExecutors { + host: Arc>>, + cleanup: Arc>>, } -fn host_io_executor() -> Result { - static HOST: std::sync::OnceLock> = - std::sync::OnceLock::new(); - HOST.get_or_init(|| { - crate::db_executor::DbExecutor::new(crate::db_executor::DbExecutorLimits { - workers: std::num::NonZeroUsize::new(2).unwrap(), - queued: std::num::NonZeroUsize::new(16).unwrap(), - }) - .ok() - }) - .clone() - .ok_or(Error::DbExecutorUnavailable) +impl HostExecutors { + fn shared() -> Self { + static SHARED: std::sync::OnceLock = std::sync::OnceLock::new(); + SHARED.get_or_init(Self::default).clone() + } + + fn lane( + owner: &std::sync::OnceLock>, + ) -> Result { + owner + .get_or_init(|| { + crate::db_executor::DbExecutor::new(crate::db_executor::DbExecutorLimits { + workers: std::num::NonZeroUsize::new(2).unwrap(), + queued: std::num::NonZeroUsize::new(16).unwrap(), + }) + .ok() + }) + .clone() + .ok_or(Error::DbExecutorUnavailable) + } + + fn host(&self) -> Result { + Self::lane(&self.host) + } + fn cleanup(&self) -> Result { + Self::lane(&self.cleanup) + } } // Some existing lifecycle workflows interleave owned async resource locks with @@ -20216,6 +20292,7 @@ fn host_io_executor() -> Result { // No SQL transaction contains an await/Runtime call, and this lane is separate // from the DB worker needed by Runtime callbacks. Do not nest this helper. async fn run_host_workflow( + executor: crate::db_executor::DbExecutor, future: impl std::future::Future> + Send + 'static, ) -> std::result::Result where @@ -20224,10 +20301,7 @@ where { let runtime = tokio::runtime::Handle::try_current().map_err(|_| E::from(Error::DbExecutorUnavailable))?; - host_io_executor() - .map_err(E::from)? - .run(move || runtime.block_on(future)) - .await + executor.run(move || runtime.block_on(future)).await } fn open_subjektiv_store(api: &WorkspaceApi) -> ApiResult { @@ -22452,7 +22526,8 @@ async fn start_subjektiv_staging_consolidation( } let subject = subject_id.to_owned(); - host_io_executor()? + api.host_executors + .host()? .run(move || subjektiv_job::dispatch(&api, &subject, candidate_count, total_bytes)) .await } @@ -22462,7 +22537,8 @@ async fn scoped_memory_consolidation( AxumPath(path): AxumPath, Json(operation): Json, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { validate_workspace_scope(&api, &path.workspace_id)?; Ok(Json(start_memory_staging_consolidation(api, operation)?)) @@ -23499,7 +23575,8 @@ async fn scoped_get_workspace_worker( State(api): State, AxumPath(path): AxumPath, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { validate_workspace_scope(&api, &path.workspace_id)?; let worker_ref = api @@ -23656,7 +23733,8 @@ async fn list_known_workers( AxumPath(path): AxumPath, headers: HeaderMap, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { validate_workspace_scope(&api, &path.workspace_id)?; let source = authenticate_worker_mutation_source(&api, &path.workspace_id, &headers)?; @@ -23719,7 +23797,7 @@ async fn spawn_known_worker_authenticated( context: Option<&server_api::ServerRequestContext>, ) -> ApiResult> { let context = context.cloned(); - run_host_workflow(async move { + run_host_workflow(api.host_executors.host()?, async move { spawn_known_worker_authenticated_on_host( State(api), AxumPath(path), @@ -23958,7 +24036,8 @@ async fn scoped_list_worker_observation_sessions( AxumPath(path): AxumPath, headers: HeaderMap, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { validate_workspace_scope(&api, &path.workspace_id)?; let source = authenticate_worker_mutation_source(&api, &path.workspace_id, &headers)?; @@ -24084,7 +24163,7 @@ async fn scoped_start_workspace_orchestrator( State(api): State, AxumPath(path): AxumPath, ) -> ApiResult> { - host_io_executor()?.run(move || { + api.host_executors.host()?.run(move || { #[cfg(test)] tests::seed_test_delivery_binding(&api); validate_workspace_scope(&api, &path.workspace_id)?; @@ -24345,7 +24424,8 @@ async fn scoped_list_runtime_working_directories( ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; require_active_workspace_runtime_binding(&api, &path.runtime_id).await?; - host_io_executor()? + api.host_executors + .host()? .run(move || { let (items, diagnostics) = runtime_working_directory_summaries(&api, &path.runtime_id)?; let items = items.into_iter().map(Into::into).collect(); @@ -24379,7 +24459,8 @@ async fn scoped_runtime_working_directory_detail( ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; require_active_workspace_runtime_binding(&api, &path.runtime_id).await?; - host_io_executor()? + api.host_executors + .host()? .run(move || { working_directory_detail_for_runtime(api, &path.runtime_id, &path.working_directory_id) }) @@ -24469,7 +24550,8 @@ async fn cleanup_runtime_working_directory( ) -> ApiResult { validate_workspace_scope(&api, &path.workspace_id)?; require_active_workspace_runtime_binding(&api, &path.runtime_id).await?; - host_io_executor()? + api.host_executors + .host()? .run(move || { let registered_runtime = registered_workdir_runtime_id(&api, &path.working_directory_id)?; @@ -25703,7 +25785,8 @@ async fn scoped_working_directory_detail( State(api): State, AxumPath(path): AxumPath, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { validate_workspace_scope(&api, &path.workspace_id)?; let record = api @@ -25752,7 +25835,8 @@ async fn cleanup_working_directory( request: WorkingDirectoryRemovalRequest, ) -> ApiResult { validate_workspace_scope(&api, &path.workspace_id)?; - host_io_executor()? + api.host_executors + .host()? .run(move || { execute_workdir_removal( &api, @@ -26067,7 +26151,9 @@ async fn create_workspace_working_directory( if reserved.state == "succeeded" { let stage_api = api.clone(); - return host_io_executor()? + return api + .host_executors + .host()? .run(move || { working_directory_detail_for_runtime( stage_api, @@ -26113,7 +26199,9 @@ async fn create_workspace_working_directory( let stage_api = api.clone(); let stage_runtime_id = reserved.resolved_runtime_id.clone(); - let runtime = match host_io_executor()? + let runtime = match api + .host_executors + .host()? .run(move || { Ok::<_, Error>( stage_api @@ -26199,7 +26287,9 @@ async fn create_workspace_working_directory( working_directory_request.backend_workdir_id = Some(reserved.working_directory_id.clone()); let stage_api = api.clone(); let stage_reserved = reserved.clone(); - let existing = match host_io_executor()? + let existing = match api + .host_executors + .host()? .run(move || { Ok::<_, Error>(stage_api.runtime.working_directory( &stage_reserved.resolved_runtime_id, @@ -26228,7 +26318,9 @@ async fn create_workspace_working_directory( } else { let stage_api = api.clone(); let stage_runtime_id = reserved.resolved_runtime_id.clone(); - match host_io_executor()? + match api + .host_executors + .host()? .run(move || { Ok::<_, Error>( stage_api @@ -27174,7 +27266,9 @@ async fn execute_runtime_cleanup_with_context( context.authorize_workspace(api).await?; let plan_api = api.clone(); let plan_runtime = runtime_id.to_owned(); - let plan = host_io_executor()? + let plan = api + .host_executors + .host()? .run(move || build_runtime_cleanup_plan(&plan_api, &plan_runtime)) .await?; // Rebuild the actual candidate set immediately before executing. The digest @@ -27297,7 +27391,9 @@ async fn execute_runtime_cleanup_with_context( let removal_api = api.clone(); let removal_candidate = candidate.clone(); let requester = format!("runtime-cleanup:{runtime_id}"); - let removal = host_io_executor()? + let removal = api + .host_executors + .host()? .run(move || { execute_workdir_removal( &removal_api, @@ -27354,7 +27450,9 @@ async fn execute_runtime_cleanup_with_context( let plan_api = api.clone(); let plan_runtime = runtime_id.to_owned(); - let plan_after = host_io_executor()? + let plan_after = api + .host_executors + .host()? .run(move || build_runtime_cleanup_plan(&plan_api, &plan_runtime)) .await?; let executed_at = now_registry_timestamp(); @@ -27664,7 +27762,8 @@ async fn scoped_runtime_cleanup_plan( State(api): State, AxumPath(path): AxumPath, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { validate_workspace_scope(&api, &path.workspace_id)?; let plan = build_runtime_cleanup_plan(&api, path.runtime_id.as_str())?; @@ -27715,7 +27814,8 @@ async fn scoped_create_attachment_upload_grant( AxumPath(path): AxumPath, Query(query): Query, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { validate_workspace_scope(&api, &path.workspace_id)?; resolve_workspace_worker_reference( @@ -27857,7 +27957,8 @@ async fn scoped_cancel_attachment_upload( Extension(actor): Extension, AxumPath(path): AxumPath, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { validate_workspace_scope(&api, &path.workspace_id)?; let completed = { @@ -28885,7 +28986,8 @@ async fn repository_log( async fn list_hosts( State(api): State, ) -> ApiResult>> { - host_io_executor()? + api.host_executors + .host()? .run(move || { let limit = api.config.max_records.min(200); let runtime_hosts = api.runtime.list_hosts(limit); @@ -28903,7 +29005,8 @@ async fn list_hosts( async fn list_runtimes( State(api): State, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { let limit = api.config.max_records.min(200); let runtimes = api.runtime.list_runtimes(limit); @@ -29100,7 +29203,10 @@ async fn execute_runtime_removal( let api = api.clone(); // Retain registration/commit recovery as one owned lifecycle workflow; // SQL transaction boundaries and commit-before-Runtime order are unchanged. - run_host_workflow(async move { execute_runtime_removal_on_host(&api, operation).await }).await + run_host_workflow(api.host_executors.host()?, async move { + execute_runtime_removal_on_host(&api, operation).await + }) + .await } async fn execute_runtime_removal_on_host( @@ -29582,7 +29688,7 @@ async fn create_workspace_worker_authenticated( context: Option<&server_api::ServerRequestContext>, ) -> ApiResult> { let context = context.cloned(); - run_host_workflow(async move { + run_host_workflow(api.host_executors.host()?, async move { create_workspace_worker_authenticated_on_host( api, headers, @@ -30411,7 +30517,8 @@ async fn get_runtime_worker( State(api): State, AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { let worker_ref = resolve_workspace_worker_reference(&api, &runtime_id, &worker_id)?; let worker = api @@ -30508,7 +30615,8 @@ async fn list_runtime_workers( AxumPath(runtime_id): AxumPath, Query(query): Query, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { let limit = api.config.max_records.min(200); let (runtime_workers, source) = match query.status { @@ -30941,7 +31049,7 @@ async fn create_runtime_worker_authenticated( context: Option<&server_api::ServerRequestContext>, ) -> ApiResult> { let context = context.cloned(); - run_host_workflow(async move { + run_host_workflow(api.host_executors.host()?, async move { create_runtime_worker_authenticated_on_host( State(api), AxumPath(runtime_id), @@ -31159,7 +31267,8 @@ async fn sync_runtime_config_bundle( AxumPath(runtime_id): AxumPath, Json(request): Json, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { let bundle = serde_json::from_value::(request.bundle) .map_err(|error| invalid_contract_value("Runtime ConfigBundle", error))?; @@ -31181,7 +31290,8 @@ async fn check_runtime_config_bundle( AxumPath((runtime_id, bundle_id)): AxumPath<(String, String)>, Query(query): Query, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { let result = api .runtime @@ -31230,7 +31340,8 @@ async fn delete_runtime_worker_uploaded_file( State(api): State, AxumPath((runtime_id, worker_id, artifact_id)): AxumPath<(String, String, String)>, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { let worker = resolve_workspace_worker_reference(&api, &runtime_id, &worker_id)?; api.runtime @@ -31246,7 +31357,8 @@ async fn runtime_worker_completions( AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>, Json(request): Json, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { let request = worker_completions_request_from_api(request)?; let worker = resolve_workspace_worker_reference(&api, &runtime_id, &worker_id)?; @@ -31825,7 +31937,8 @@ async fn list_host_workers( State(api): State, AxumPath(host_id): AxumPath, ) -> ApiResult> { - host_io_executor()? + api.host_executors + .host()? .run(move || { let limit = api.config.max_records.min(200); let runtime_workers = api @@ -38312,7 +38425,7 @@ mod tests { cached_status: "connected".to_owned(), timeout: std::time::Duration::from_secs(1), }); - SqliteWorkspaceStore::open(&api.config.database_path) + SqliteWorkspaceStore::open_for_test(&api.config.database_path) .unwrap() .upsert_workspace_runtime_binding( WorkspaceRuntimeBinding { @@ -39819,9 +39932,9 @@ mod tests { async fn workspace_api_does_not_register_a_missing_workspace() { let dir = tempfile::tempdir().unwrap(); let config = test_server_config(dir.path()); - let store = Arc::new(SqliteWorkspaceStore::open(&config.database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap()); - let error = match WorkspaceApi::new(config, store.clone()).await { + let error = match WorkspaceApi::new_for_test(config, store.clone()).await { Ok(_) => panic!("missing Workspace registration must fail closed"), Err(error) => error, }; @@ -39838,11 +39951,12 @@ mod tests { let workspace = tempfile::tempdir().unwrap(); init_clean_git_workspace(workspace.path()); let config = test_server_config(workspace.path()); - let store = Arc::new(SqliteWorkspaceStore::open(config.database_path.clone()).unwrap()); + let store = + Arc::new(SqliteWorkspaceStore::open_for_test(config.database_path.clone()).unwrap()); seed_test_registered_workspace(store.as_ref(), &config) .await .unwrap(); - let api = WorkspaceApi::new(config, store).await.unwrap(); + let api = WorkspaceApi::new_for_test(config, store).await.unwrap(); let workspace_id = api.config.workspace_id.clone(); let result = scoped_start_workspace_orchestrator( @@ -39897,7 +40011,7 @@ mod tests { let workspace = tempfile::tempdir().unwrap(); init_clean_git_workspace(workspace.path()); let config = test_server_config(workspace.path()); - let store = SqliteWorkspaceStore::open(config.database_path.clone()).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(config.database_path.clone()).unwrap(); let api = WorkspaceApi::new_with_execution_backend( config, Arc::new(store), @@ -44441,8 +44555,8 @@ mod tests { let restored_config = test_server_config(&root) .with_embedded_runtime_store_root(root.join("restored-embedded-runtime")); let restored_store = - SqliteWorkspaceStore::open(restored_config.database_path.clone()).unwrap(); - let restored = WorkspaceApi::new(restored_config, Arc::new(restored_store)) + SqliteWorkspaceStore::open_for_test(restored_config.database_path.clone()).unwrap(); + let restored = WorkspaceApi::new_for_test(restored_config, Arc::new(restored_store)) .await .unwrap(); assert!( @@ -46768,7 +46882,8 @@ mod tests { notification_target: None, limits: crate::backend_job::BackendJobLimits::default(), }; - let store = Arc::new(SqliteWorkspaceStore::open(config.database_path.clone()).unwrap()); + let store = + Arc::new(SqliteWorkspaceStore::open_for_test(config.database_path.clone()).unwrap()); seed_test_registered_workspace(store.as_ref(), &config) .await .unwrap(); @@ -46906,7 +47021,8 @@ mod tests { drop(store); drop(execution); - let restarted_store = SqliteWorkspaceStore::open(config.database_path.clone()).unwrap(); + let restarted_store = + SqliteWorkspaceStore::open_for_test(config.database_path.clone()).unwrap(); let restarted_execution = Arc::new(DeterministicExecutionBackend::default()); restarted_execution.accept_restores(); let restarted = WorkspaceApi::new_with_execution_backend( @@ -47096,7 +47212,7 @@ mod tests { async fn backend_job_delivery_recovery_replays_same_tracked_notification_id() { let temp = tempfile::tempdir().unwrap(); let config = test_server_config(temp.path()); - let store = SqliteWorkspaceStore::open(config.database_path.clone()).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(config.database_path.clone()).unwrap(); let execution = Arc::new(DeterministicExecutionBackend::default()); let api = WorkspaceApi::new_with_execution_backend( config.clone(), @@ -47671,7 +47787,7 @@ mod tests { if let Some(parent) = config.database_path.parent() { fs::create_dir_all(parent).unwrap(); } - SqliteWorkspaceStore::open(&config.database_path).unwrap() + SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap() } #[tokio::test] @@ -47732,7 +47848,7 @@ mod tests { .unwrap(); let mut template = test_server_config(dir.path()); template.static_assets_dir = Some(static_dir); - let store = Arc::new(SqliteWorkspaceStore::open(&template.database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&template.database_path).unwrap()); let app = build_workspace_server_router(template, store) .await .unwrap(); @@ -47815,7 +47931,7 @@ mod tests { async fn repository_free_workspace_can_be_created_selected_and_used_then_register_repository() { let temp = tempfile::tempdir().unwrap(); let config = test_server_config(temp.path()); - let store = Arc::new(SqliteWorkspaceStore::open(&config.database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap()); let owner = seed_test_api_token(store.as_ref(), "empty-owner"); let foreign = seed_test_api_token(store.as_ref(), "empty-foreign"); let app = build_workspace_server_router(config.clone(), store.clone()) @@ -48067,7 +48183,7 @@ mod tests { .. } = &config.auth; let expected_origin = expected_origin.clone(); - let store = Arc::new(SqliteWorkspaceStore::open(&config.database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap()); store .upsert_account(&AccountRecord { account_id: "account-auth".to_owned(), @@ -49344,7 +49460,7 @@ mod tests { .unwrap(); let mut template = test_server_config(dir.path()); template.static_assets_dir = Some(static_dir); - let store = Arc::new(SqliteWorkspaceStore::open(&template.database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&template.database_path).unwrap()); let token = seed_test_api_token(store.as_ref(), "two-workspaces"); let catalog = WorkspaceCatalogService::new( store.clone(), @@ -49485,7 +49601,7 @@ mod tests { let repository = dir.path().join("repository"); std::fs::create_dir_all(repository.join(".git")).unwrap(); let template = test_server_config(dir.path()); - let store = Arc::new(SqliteWorkspaceStore::open(&template.database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&template.database_path).unwrap()); let app = build_workspace_server_router(template, store) .await .unwrap(); @@ -49554,7 +49670,7 @@ mod tests { async fn memory_consolidation_backlog_ignores_legacy_filesystem_staging() { let dir = tempfile::tempdir().unwrap(); let config = test_server_config(dir.path()); - let store = SqliteWorkspaceStore::open(&config.database_path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap(); let api = WorkspaceApi::new_with_execution_backend( config, Arc::new(store), @@ -49587,7 +49703,7 @@ mod tests { async fn memory_consolidation_reuses_existing_idle_worker() { let dir = tempfile::tempdir().unwrap(); let config = test_server_config(dir.path()); - let store = SqliteWorkspaceStore::open(&config.database_path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap(); let api = WorkspaceApi::new_with_execution_backend( config, Arc::new(store), @@ -52982,7 +53098,7 @@ mod tests { workspace_root: impl Into, ) -> (WorkspaceApi, Arc) { let config = test_server_config(workspace_root); - let store = SqliteWorkspaceStore::open(config.database_path.clone()).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(config.database_path.clone()).unwrap(); let execution = Arc::new(DeterministicExecutionBackend::default()); let api = WorkspaceApi::new_with_execution_backend(config, Arc::new(store), execution.clone()) @@ -53006,7 +53122,7 @@ mod tests { config.repositories[0].source = source; config.repositories[0].path = None; config.repositories[0].default_selector = Some("develop".to_string()); - let store = SqliteWorkspaceStore::open(config.database_path.clone()).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(config.database_path.clone()).unwrap(); WorkspaceApi::new_with_execution_backend( config, Arc::new(store), @@ -53035,7 +53151,7 @@ mod tests { docs.path = Some(docs_root); docs.default_selector = Some("develop".to_string()); config.repositories.push(docs); - let store = SqliteWorkspaceStore::open(config.database_path.clone()).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(config.database_path.clone()).unwrap(); WorkspaceApi::new_with_execution_backend( config, Arc::new(store), @@ -53470,6 +53586,7 @@ mod tests { store, Arc::new(DeterministicExecutionBackend::default()), BackendResourceBroker::default(), + HostExecutors::default(), Some(dispatcher.clone()), ) .await @@ -53715,7 +53832,7 @@ mod tests { cached_status: "connected".to_string(), timeout: std::time::Duration::from_secs(1), }); - let store = SqliteWorkspaceStore::open(config.database_path.clone()).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(config.database_path.clone()).unwrap(); let owner_account_id = format!("account-{TEST_WORKSPACE_ID}"); store .upsert_account(&crate::store::AccountRecord { @@ -54015,7 +54132,8 @@ mod tests { let mut revoked = trust; revoked.revoked_at = Some("2026-08-11T00:01:00Z".to_string()); - let authority = SqliteWorkspaceStore::open(api.config.database_path.clone()).unwrap(); + let authority = + SqliteWorkspaceStore::open_for_test(api.config.database_path.clone()).unwrap(); authority .upsert_workspace_runtime_binding(revoked, true) .unwrap(); @@ -54945,7 +55063,7 @@ mod tests { .unwrap() .digest ); - let restarted = SqliteWorkspaceStore::open(&api.config.database_path).unwrap(); + let restarted = SqliteWorkspaceStore::open_for_test(&api.config.database_path).unwrap(); assert_eq!( changed.digest, restarted @@ -55491,7 +55609,7 @@ mod tests { .unwrap(); assert_eq!(capability_changed.items, reattached.items); assert_ne!(capability_changed.digest, reattached.digest); - let restarted = SqliteWorkspaceStore::open(&api.config.database_path).unwrap(); + let restarted = SqliteWorkspaceStore::open_for_test(&api.config.database_path).unwrap(); let (restarted_page, restarted_digest) = restarted .list_worker_workdir_links_page_with_digest( TEST_WORKSPACE_ID, @@ -55510,7 +55628,7 @@ mod tests { .unwrap() .run_to_completion(128, std::time::Duration::from_millis(1), None) .unwrap(); - let restored = SqliteWorkspaceStore::from_connection(snapshot).unwrap(); + let restored = SqliteWorkspaceStore::from_connection_for_test(snapshot).unwrap(); assert_eq!( restored .list_worker_workdir_links_page_with_digest( @@ -55555,7 +55673,7 @@ mod tests { ) .unwrap(); assert_ne!(read_write_digest, capability_changed.digest); - let reader = SqliteWorkspaceStore::open(&api.config.database_path).unwrap(); + let reader = SqliteWorkspaceStore::open_for_test(&api.config.database_path).unwrap(); let barrier = Arc::new(std::sync::Barrier::new(2)); std::thread::scope(|scope| { let worker = &fixture.worker; @@ -57415,7 +57533,7 @@ mod tests { let identity = worker_runtime::auth::RuntimeIdentityMaterial::generate("runtime-test").unwrap(); configure_runtime_request_auth(&mut api, &identity, "runtime-test"); - let store = SqliteWorkspaceStore::open(&api.config.database_path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&api.config.database_path).unwrap(); let memory_settings = store .get_workspace_memory_settings(TEST_WORKSPACE_ID) .unwrap(); @@ -59615,9 +59733,9 @@ mod tests { std::fs::write(static_dir.join("index.html"), "
Yoi Workspace
").unwrap(); std::fs::write(static_dir.join("assets/app.js"), "console.log('yoi');").unwrap(); - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); let mut config = test_server_config(dir.path()); - let sqlite_store = SqliteWorkspaceStore::open(&config.database_path).unwrap(); + let sqlite_store = SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap(); sqlite_store .upsert_workspace(&WorkspaceRecord { workspace_id: TEST_WORKSPACE_ID.to_string(), @@ -59652,7 +59770,7 @@ mod tests { "API Ticket", ticket::TicketState::Open, ); - let sqlite_store = SqliteWorkspaceStore::open(&config.database_path).unwrap(); + let sqlite_store = SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap(); sqlite_store .upsert_workspace(&WorkspaceRecord { workspace_id: TEST_WORKSPACE_ID.to_string(), @@ -62176,7 +62294,7 @@ mod tests { async fn objective_mutation_endpoints_round_trip_through_workspace_authority() { let dir = tempfile::tempdir().unwrap(); let config = test_server_config(dir.path()); - let store = Arc::new(SqliteWorkspaceStore::open(&config.database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap()); let owner_account_id = format!("account-{TEST_WORKSPACE_ID}"); store .upsert_account(&crate::store::AccountRecord { diff --git a/crates/workspace-server/src/server/subjektiv_job.rs b/crates/workspace-server/src/server/subjektiv_job.rs index 6aaa1e83c..33ef01ed4 100644 --- a/crates/workspace-server/src/server/subjektiv_job.rs +++ b/crates/workspace-server/src/server/subjektiv_job.rs @@ -68,7 +68,9 @@ pub(super) async fn retire_legacy_worker(api: &WorkspaceApi, subject_id: &str) - let job_api = api.clone(); let subject_id = subject_id.to_owned(); let key = format!("{SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX}{subject_id}"); - let (worker, complete) = host_io_executor()? + let (worker, complete) = api + .host_executors + .host()? .run(move || { let api = &job_api; let key = format!("{SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX}{subject_id}"); diff --git a/crates/workspace-server/src/server/tests/auth_logging_tests.rs b/crates/workspace-server/src/server/tests/auth_logging_tests.rs index b00d2e1b4..d5bf3334a 100644 --- a/crates/workspace-server/src/server/tests/auth_logging_tests.rs +++ b/crates/workspace-server/src/server/tests/auth_logging_tests.rs @@ -54,7 +54,7 @@ async fn rejected_runtime_proofs_log_reason_and_verification_without_secrets_on_ configure_runtime_request_auth(&mut api, &identity, "runtime-test"); let revoked = RuntimeIdentityMaterial::generate("revoked-runtime-secret").unwrap(); configure_runtime_request_auth(&mut api, &revoked, "revoked-runtime-secret"); - SqliteWorkspaceStore::open(&api.config.database_path) + SqliteWorkspaceStore::open_for_test(&api.config.database_path) .unwrap() .revoke_workspace_runtime_binding( TEST_WORKSPACE_ID, @@ -70,7 +70,7 @@ async fn rejected_runtime_proofs_log_reason_and_verification_without_secrets_on_ let body = "body-secret"; let apps = [ build_router(api.clone()), - workspace_server_router(WorkspaceServerApi::new( + workspace_server_router(WorkspaceServerApi::new_for_test( api.config.clone(), api.store.clone(), )), @@ -347,7 +347,7 @@ async fn repository_runtime_proof_401_keeps_generic_envelope_on_both_routers() { let path = format!("/api/w/{TEST_WORKSPACE_ID}/repositories"); for app in [ build_router(api.clone()), - workspace_server_router(WorkspaceServerApi::new( + workspace_server_router(WorkspaceServerApi::new_for_test( api.config.clone(), api.store.clone(), )), diff --git a/crates/workspace-server/src/server/tests/drive_tests.rs b/crates/workspace-server/src/server/tests/drive_tests.rs index 1d05ff7db..3b7ec65ca 100644 --- a/crates/workspace-server/src/server/tests/drive_tests.rs +++ b/crates/workspace-server/src/server/tests/drive_tests.rs @@ -40,7 +40,7 @@ async fn cached_drive_handle_rejects_mutation_after_deletion_reservation() { }) .await .unwrap(); - let api = WorkspaceApi::new(config.clone(), store.clone()) + let api = WorkspaceApi::new_for_test(config.clone(), store.clone()) .await .unwrap(); let cached = api.drive().await.unwrap(); @@ -104,7 +104,7 @@ async fn deleting_workspace_reopens_fenced_drive_and_purges_blobs_before_feature // Use a separate embedded-runtime namespace for the first API: this test // reconstructs Drive authority in-process rather than spawning/restarting // product processes (runtime observers may retain their namespace). - let first_api = WorkspaceApi::new(config.clone(), store.clone()) + let first_api = WorkspaceApi::new_for_test(config.clone(), store.clone()) .await .unwrap(); // Merely constructing a Workspace API must not create blob storage. @@ -160,7 +160,7 @@ async fn deleting_workspace_reopens_fenced_drive_and_purges_blobs_before_feature .unwrap() .unwrap(); drop(first_api); - let recovered = WorkspaceServerApi::new(config.clone(), store.clone()); + let recovered = WorkspaceServerApi::new_for_test(config.clone(), store.clone()); let api = recovered .api_for_workspace(&config.workspace_id) .await diff --git a/crates/workspace-server/src/server/tests/drive_web_roundtrip_tests.rs b/crates/workspace-server/src/server/tests/drive_web_roundtrip_tests.rs index 02ab40e35..71efdb137 100644 --- a/crates/workspace-server/src/server/tests/drive_web_roundtrip_tests.rs +++ b/crates/workspace-server/src/server/tests/drive_web_roundtrip_tests.rs @@ -39,10 +39,10 @@ impl RealDriveFixture { let temp = tempfile::tempdir().unwrap(); let mut config = test_server_config(temp.path()); config.repositories.clear(); - let store = Arc::new(SqliteWorkspaceStore::open(&config.database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap()); let owner = seed_test_api_token(store.as_ref(), "drive-web-owner"); let second = seed_test_api_token(store.as_ref(), "drive-web-second"); - let server_api = WorkspaceServerApi::new(config, store.clone()); + let server_api = WorkspaceServerApi::new_for_test(config, store.clone()); let server = LoopbackServer::start(workspace_server_router(server_api.clone())).await; // Exercise T-717's real repositoryless Workspace bootstrap, not a // pre-existing test Workspace with its repository merely hidden in UI. diff --git a/crates/workspace-server/src/server/tests/subject_spawn_tests.rs b/crates/workspace-server/src/server/tests/subject_spawn_tests.rs index b54c4bdad..37519b22c 100644 --- a/crates/workspace-server/src/server/tests/subject_spawn_tests.rs +++ b/crates/workspace-server/src/server/tests/subject_spawn_tests.rs @@ -27,7 +27,7 @@ async fn subject_session_callback_completes_while_create_and_deletion_are_waitin ) .unwrap(); let worker_id = reserved.worker_id.to_string(); - let server = WorkspaceServerApi::new(api.config.clone(), api.store.clone()); + let server = WorkspaceServerApi::new_for_test(api.config.clone(), api.store.clone()); // Reuse the fixture API, but exercise the real outer dispatcher and both auth layers. server .apis diff --git a/crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs b/crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs index 64ca15dc3..733df3809 100644 --- a/crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs +++ b/crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs @@ -144,7 +144,7 @@ impl LaunchFixture { async fn new(remote: bool) -> Self { let temp = tempfile::tempdir().unwrap(); let config = test_server_config(temp.path()); - let store = SqliteWorkspaceStore::open(config.database_path.clone()).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(config.database_path.clone()).unwrap(); let backend = Arc::new(LaunchExecutionBackend::default()); let api = WorkspaceApi::new_with_execution_backend(config, Arc::new(store), backend.clone()) diff --git a/crates/workspace-server/src/server/tests/worker_drive_tests.rs b/crates/workspace-server/src/server/tests/worker_drive_tests.rs index 2b24e45a5..481c309ad 100644 --- a/crates/workspace-server/src/server/tests/worker_drive_tests.rs +++ b/crates/workspace-server/src/server/tests/worker_drive_tests.rs @@ -151,7 +151,7 @@ impl Fixture { let temp = tempfile::tempdir().unwrap(); let mut config = test_server_config(temp.path()); config.repositories.clear(); - let store = Arc::new(SqliteWorkspaceStore::open(&config.database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap()); let mut api = WorkspaceApi::new_with_execution_backend( config, store, diff --git a/crates/workspace-server/src/server/tests/worker_operations_tests.rs b/crates/workspace-server/src/server/tests/worker_operations_tests.rs index 2229bdb59..67ae6fdce 100644 --- a/crates/workspace-server/src/server/tests/worker_operations_tests.rs +++ b/crates/workspace-server/src/server/tests/worker_operations_tests.rs @@ -1329,8 +1329,9 @@ async fn restore_guard_internal_stale_conflict_retires_intent_for_later_explicit ); assert_eq!(api.runtime.worker(&identity).unwrap().state, "stopped"); // Retirement, unlike in-memory refresh, must survive caller journal restart. - api.config_store = - Arc::new(SqliteWorkspaceStore::open(api.config.database_path.clone()).unwrap()); + api.config_store = Arc::new( + SqliteWorkspaceStore::open_for_test(api.config.database_path.clone()).unwrap(), + ); let fresh = api.runtime.worker(&identity).unwrap(); let next = api .config_store @@ -1490,7 +1491,7 @@ async fn restore_guard_internal_intent_survives_client_store_restart_without_rep ); // Reopen the durable caller journal, discarding all client-side memory of request identity. api.config_store = - Arc::new(SqliteWorkspaceStore::open(api.config.database_path.clone()).unwrap()); + Arc::new(SqliteWorkspaceStore::open_for_test(api.config.database_path.clone()).unwrap()); let pinned = api .config_store .pin_internal_worker_restore_intent( @@ -1599,7 +1600,7 @@ async fn restore_guard_cross_runtime_ssh_completion_retries_pinned_snapshot_at_w crate::repository_source::repository_source_fingerprint(&source); config.repositories[0].source = source; config.repositories[0].path = None; - let store = SqliteWorkspaceStore::open(config.database_path.clone()).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(config.database_path.clone()).unwrap(); let execution = Arc::new(DeterministicExecutionBackend::default()); execution.accept_restores(); let api = WorkspaceApi::new_with_execution_backend(config, Arc::new(store), execution.clone()) diff --git a/crates/workspace-server/src/server/tests/workspace_admission_tests.rs b/crates/workspace-server/src/server/tests/workspace_admission_tests.rs index c71ea73df..142ae143e 100644 --- a/crates/workspace-server/src/server/tests/workspace_admission_tests.rs +++ b/crates/workspace-server/src/server/tests/workspace_admission_tests.rs @@ -56,7 +56,7 @@ impl Fixture { api.runtime.register_or_replace(runtime); let now = Arc::new(AtomicI64::new(START)); let clock = now.clone(); - let mut server = WorkspaceServerApi::new(api.config.clone(), api.store.clone()); + let mut server = WorkspaceServerApi::new_for_test(api.config.clone(), api.store.clone()); server.proof_clock = Arc::new(move || clock.load(Ordering::SeqCst)); server .apis diff --git a/crates/workspace-server/src/server/worker_operations.rs b/crates/workspace-server/src/server/worker_operations.rs index ed38158bf..447ae7e29 100644 --- a/crates/workspace-server/src/server/worker_operations.rs +++ b/crates/workspace-server/src/server/worker_operations.rs @@ -356,7 +356,9 @@ impl WorkspaceWorker { ) -> ApiResult { let guard = self.authorize_operation(context, "restore").await?; let worker = self.clone(); - host_io_executor()? + self.api + .host_executors + .host()? .run(move || { let _guard = guard; worker.restore_internal(request) @@ -381,7 +383,10 @@ impl WorkspaceWorker { ) -> ApiResult { let guard = self.authorize_operation(context, "stop").await?; let worker = self.clone(); - let (result, guard) = host_io_executor()? + let (result, guard) = self + .api + .host_executors + .host()? .run(move || -> ApiResult<_> { let result = worker .api @@ -399,7 +404,9 @@ impl WorkspaceWorker { let session_guard = session_lock.lock_owned().await; close_current_worker_session_locked(&self.api, &self.identity).await?; let worker = self.clone(); - host_io_executor()? + self.api + .host_executors + .host()? .run(move || -> ApiResult<_> { let (_control, _session) = (guard, session_guard); if let Some(record) = worker @@ -425,7 +432,9 @@ impl WorkspaceWorker { ) -> ApiResult { let guard = self.authorize_operation(context, "cancel").await?; let worker = self.clone(); - host_io_executor()? + self.api + .host_executors + .host()? .run(move || -> ApiResult<_> { let _guard = guard; Ok(worker @@ -457,7 +466,9 @@ impl WorkspaceWorker { request.kind = WorkerInputKind::UserIfIdle; } let worker = self.clone(); - host_io_executor()? + self.api + .host_executors + .host()? .run(move || -> ApiResult<_> { let (_control, _session) = (guard, session_guard); Ok(worker diff --git a/crates/workspace-server/src/server_assignment_attention_tests.rs b/crates/workspace-server/src/server_assignment_attention_tests.rs index d72a8e735..c70499891 100644 --- a/crates/workspace-server/src/server_assignment_attention_tests.rs +++ b/crates/workspace-server/src/server_assignment_attention_tests.rs @@ -593,7 +593,7 @@ async fn real_attention_worker_start_restore(lose_response: bool) { let dir = tempfile::tempdir().unwrap(); init_clean_git_workspace(dir.path()); let config = test_server_config(dir.path()); - let store = Arc::new(SqliteWorkspaceStore::open(&config.database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap()); seed_test_registered_workspace(store.as_ref(), &config) .await .unwrap(); diff --git a/crates/workspace-server/src/server_assignment_delivery_tests.rs b/crates/workspace-server/src/server_assignment_delivery_tests.rs index 489674c19..3f44c5c89 100644 --- a/crates/workspace-server/src/server_assignment_delivery_tests.rs +++ b/crates/workspace-server/src/server_assignment_delivery_tests.rs @@ -700,7 +700,7 @@ async fn delivery_real_fixture( DeliveryScriptedClient, ) { let config = test_server_config(root); - let store = Arc::new(SqliteWorkspaceStore::open(&config.database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap()); seed_test_registered_workspace(store.as_ref(), &config) .await .unwrap(); @@ -924,7 +924,7 @@ async fn assignment_real_worker_response_loss_restore_and_concurrent_scans_commi tokio::task::yield_now().await; let reopened = WorkspaceApi::new_with_execution_backend( config.clone(), - Arc::new(SqliteWorkspaceStore::open(&config.database_path).unwrap()), + Arc::new(SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap()), execution.clone(), ) .await diff --git a/crates/workspace-server/src/server_db_execution_tests.rs b/crates/workspace-server/src/server_db_execution_tests.rs index 22052925a..1db761f13 100644 --- a/crates/workspace-server/src/server_db_execution_tests.rs +++ b/crates/workspace-server/src/server_db_execution_tests.rs @@ -63,7 +63,7 @@ impl Fixture { backend_base_url: Some("http://localhost:8787".into()), }; let store = Arc::new( - SqliteWorkspaceStore::open(&config.database_path) + SqliteWorkspaceStore::open_for_test(&config.database_path) .unwrap() .with_db_executor(executor.clone()), ); @@ -90,10 +90,10 @@ impl Fixture { revoked_at: None, }) .unwrap(); - let api = WorkspaceApi::new(config.clone(), store.clone()) + let api = WorkspaceApi::new_for_test(config.clone(), store.clone()) .await .unwrap(); - let server = WorkspaceServerApi::new(config, store); + let server = WorkspaceServerApi::new_for_test(config, store); // Cache only the real inner router; avoid starting the unrelated hook. server .apis @@ -452,7 +452,9 @@ async fn ticket_create_waiting_for_sqlite_does_not_block_timer_or_control() { assert!(body["id"].is_string(), "{body}"); // Ticket dispatch can call Runtime callbacks, so its operation owner is // the separate bounded host executor, not the store executor. - observed(host_io_executor().unwrap().drain()).await.unwrap(); + observed(fixture.api.host_executors.host().unwrap().drain()) + .await + .unwrap(); fixture.executor.drain().await.unwrap(); let count: i64 = reader.query_row("SELECT count(*) FROM typed_tickets WHERE workspace_id = ?1 AND title = 'locked Ticket'", [WORKSPACE], |row| row.get(0)).unwrap(); assert_eq!(count, 1, "one accepted HTTP mutation must commit once"); @@ -604,3 +606,86 @@ async fn cancelled_queued_http_mutation_keeps_capacity_until_commit_and_drain() "drain must release actual-job capacity" ); } + +#[tokio::test(start_paused = true)] +async fn saturated_cancelled_host_jobs_keep_their_budget_but_not_other_fixture_or_cleanup_admission() + { + use std::sync::{ + Condvar, + atomic::{AtomicUsize, Ordering}, + }; + // A drop guard releases workers even when an assertion panics. + struct Release(Arc<(Mutex, Condvar)>); + impl Drop for Release { + fn drop(&mut self) { + *self.0.0.lock().unwrap() = true; + self.0.1.notify_all(); + } + } + let owner = HostExecutors::default(); + let other = HostExecutors::default(); + let host = owner.host().unwrap(); + let gate = Arc::new((Mutex::new(false), Condvar::new())); + let release = Release(gate.clone()); + let completed = Arc::new(AtomicUsize::new(0)); + let mut waiters = Vec::new(); + for _ in 0..2 { + let host = host.clone(); + let gate = gate.clone(); + let completed = completed.clone(); + let (started, accepted) = tokio::sync::oneshot::channel(); + waiters.push(tokio::spawn(async move { + host.run(move || { + started.send(()).unwrap(); + let mut released = gate.0.lock().unwrap(); + while !*released { + released = gate.1.wait(released).unwrap(); + } + completed.fetch_add(1, Ordering::SeqCst); + Ok::<_, Error>(()) + }) + .await + })); + accepted.await.unwrap(); + } + for _ in 0..16 { + let completed = completed.clone(); + host.dispatch(move || { + completed.fetch_add(1, Ordering::SeqCst); + }) + .unwrap(); + } + for waiter in waiters { + waiter.abort(); + assert!(waiter.await.unwrap_err().is_cancelled()); + } + assert!(matches!( + host.run(|| Ok::<_, Error>(())).await, + Err(Error::DbCapacityExceeded) + )); + // Paused Tokio time can advance without completing or releasing actual Host + // work. Neither another fixture nor cleanup needs this saturated Host lane. + tokio::time::advance(std::time::Duration::from_secs(31)).await; + assert_eq!(completed.load(Ordering::SeqCst), 0); + other + .host() + .unwrap() + .run(|| Ok::<_, Error>(())) + .await + .unwrap(); + owner + .cleanup() + .unwrap() + .run(|| Ok::<_, Error>(())) + .await + .unwrap(); + assert!(matches!( + host.run(|| Ok::<_, Error>(())).await, + Err(Error::DbCapacityExceeded) + )); + drop(release); + host.drain().await.unwrap(); + assert_eq!(completed.load(Ordering::SeqCst), 18); + host.run(|| Ok::<_, Error>(())).await.unwrap(); + host.drain().await.unwrap(); +} diff --git a/crates/workspace-server/src/server_ticket_case_signed_tests.rs b/crates/workspace-server/src/server_ticket_case_signed_tests.rs index 78ba317b2..8f7c38b52 100644 --- a/crates/workspace-server/src/server_ticket_case_signed_tests.rs +++ b/crates/workspace-server/src/server_ticket_case_signed_tests.rs @@ -724,7 +724,7 @@ async fn catalog_case_dispatch_authenticates_and_routes_the_canonical_workspace_ reason: ticket::TicketClosureReason::Completed, }, ); - let catalog = WorkspaceServerApi::new(api.config.clone(), api.store.clone()); + let catalog = WorkspaceServerApi::new_for_test(api.config.clone(), api.store.clone()); // Retain only this in-process fixture: do not create a real Runtime or a // background Orchestrator hook while testing catalog authentication/routing. catalog diff --git a/crates/workspace-server/src/server_ticket_tags_tests.rs b/crates/workspace-server/src/server_ticket_tags_tests.rs index b636a8356..1608982bd 100644 --- a/crates/workspace-server/src/server_ticket_tags_tests.rs +++ b/crates/workspace-server/src/server_ticket_tags_tests.rs @@ -529,7 +529,7 @@ async fn runtime_tag_mutations_reject_invalid_proofs_and_changed_binding_authori *tampered.body_mut() = Body::from(format!("{} ", input)); runtime_tag_response(app.clone(), tampered, StatusCode::UNAUTHORIZED).await; } - let store = SqliteWorkspaceStore::open(&api.config.database_path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&api.config.database_path).unwrap(); let original = api .store .get_workspace_runtime_binding(TEST_WORKSPACE_ID, "runtime-tags") diff --git a/crates/workspace-server/src/server_workspace_config.rs b/crates/workspace-server/src/server_workspace_config.rs index dbdb24e6c..f1372b1e4 100644 --- a/crates/workspace-server/src/server_workspace_config.rs +++ b/crates/workspace-server/src/server_workspace_config.rs @@ -634,7 +634,7 @@ pub(super) async fn commit( // a DB worker or an open SQL transaction. let (complete, publication) = tokio::sync::oneshot::channel(); let runtime = persistence_api.runtime.clone(); - let admission = host_io_executor().and_then(|executor| { + let admission = persistence_api.host_executors.host().and_then(|executor| { executor.dispatch(move || { let _guard = guard; let _diagnostics = runtime @@ -681,7 +681,9 @@ pub(super) async fn create_grant( let guard = lock(api, &worker).await?; let runtime = api.runtime.clone(); let lookup_worker = worker.clone(); - let (summary, guard) = host_io_executor() + let (summary, guard) = api + .host_executors + .host() .map_err(pre_error)? .run(move || { Ok::<_, Error>((|| { diff --git a/crates/workspace-server/src/server_workspace_config_tests.rs b/crates/workspace-server/src/server_workspace_config_tests.rs index 72fb9e910..1c932be1a 100644 --- a/crates/workspace-server/src/server_workspace_config_tests.rs +++ b/crates/workspace-server/src/server_workspace_config_tests.rs @@ -318,7 +318,7 @@ mod workspace_config_integration { }; for scoped in [false, true] { let app = if scoped { - workspace_server_router(WorkspaceServerApi::new( + workspace_server_router(WorkspaceServerApi::new_for_test( api.config.clone(), api.store.clone(), )) @@ -1385,7 +1385,7 @@ mod workspace_config_integration { let api = &fixture.api; let granted = grant(api, &fixture.worker, Access::ReadWrite).await; let attached = attach(api, &fixture.worker).await; - let restarted = SqliteWorkspaceStore::open(&api.config.database_path).unwrap(); + let restarted = SqliteWorkspaceStore::open_for_test(&api.config.database_path).unwrap(); assert_eq!( restarted .get_workspace_config_grant(TEST_WORKSPACE_ID, &granted.grant_id) diff --git a/crates/workspace-server/src/server_workspace_config_wip_tests.rs b/crates/workspace-server/src/server_workspace_config_wip_tests.rs index b5a229f55..1e989edd8 100644 --- a/crates/workspace-server/src/server_workspace_config_wip_tests.rs +++ b/crates/workspace-server/src/server_workspace_config_wip_tests.rs @@ -800,7 +800,7 @@ async fn workspace_config_actual_remote_and_embedded_clients_use_authenticated_w let base = format!("http://{addr}"); let mut config = test_server_config(root.path()); config.backend_base_url = Some("server-test".into()); - let store = Arc::new(SqliteWorkspaceStore::open(&config.database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap()); let captured = Arc::new(Mutex::new(None)); let abort = Arc::new(Mutex::new(None)); let factory = ConfigCaptureFactory { diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index 1517c10eb..2d954fa27 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -2278,13 +2278,40 @@ impl SqliteWorkspaceStore { } pub fn from_connection(conn: Connection) -> Result { + Self::from_connection_with_executor(conn, crate::db_executor::DbExecutor::shared()) + } + + fn from_connection_with_executor( + conn: Connection, + db_executor: crate::db_executor::DbExecutor, + ) -> Result { prepare_connection(&conn)?; Ok(Self { conn: Arc::new(Mutex::new(conn)), - db_executor: crate::db_executor::DbExecutor::shared(), + db_executor, }) } + // Ordinary fixtures own the same finite budget as production, not another + // fixture's admission/SQL/drain lifetime or a permanent global worker. + #[cfg(test)] + pub(crate) fn open_for_test(path: impl AsRef) -> Result { + Self::from_connection_for_test(Connection::open(path)?) + } + + #[cfg(test)] + pub(crate) fn in_memory_for_test() -> Result { + Self::from_connection_for_test(Connection::open_in_memory()?) + } + + #[cfg(test)] + pub(crate) fn from_connection_for_test(conn: Connection) -> Result { + Self::from_connection_with_executor( + conn, + crate::db_executor::DbExecutor::new(crate::db_executor::DbExecutorLimits::default())?, + ) + } + /// Override execution limits for a hermetic fixture; subsequent clones share it. pub fn with_db_executor(mut self, executor: crate::db_executor::DbExecutor) -> Self { self.db_executor = executor; @@ -14861,7 +14888,7 @@ mod tests { fn drive_grants_migrate_and_preserve_audit_through_revoke_worker_delete_and_restart() { let t = tempfile::tempdir().unwrap(); let path = t.path().join("server.db"); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); let legacy = Connection::open_in_memory().unwrap(); configure_sqlite(&legacy).unwrap(); legacy.execute_batch("CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY); @@ -14951,7 +14978,7 @@ mod tests { Some(other) ); drop(store); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); assert_eq!( store .get_workspace_drive_grant("drive-ws", &first.grant_id) @@ -14986,7 +15013,7 @@ mod tests { } #[test] fn drive_grants_baseline_constraints_and_worker_lifetime_fail_closed() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); let proposed = drive_grant_fixture(&store); let first = store.create_workspace_drive_grant(&proposed).unwrap(); store.with_conn(|c| { @@ -15022,7 +15049,7 @@ mod tests { fn schema_v74_migrates_historical_worker_projection_rows_without_counter_state() { let temp = tempfile::tempdir().unwrap(); let path = temp.path().join("workspace.db"); - drop(SqliteWorkspaceStore::open(&path).unwrap()); + drop(SqliteWorkspaceStore::open_for_test(&path).unwrap()); let conn = Connection::open(&path).unwrap(); configure_sqlite(&conn).unwrap(); conn.execute_batch( @@ -15347,7 +15374,7 @@ mod tests { #[tokio::test] async fn external_workdir_grant_persists_typed_source_and_fences_generations() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); store .upsert_account(&AccountRecord { account_id: "owner".to_string(), @@ -15719,7 +15746,7 @@ mod tests { fn runtime_removal_test_store() -> (tempfile::TempDir, SqliteWorkspaceStore) { let temp = tempfile::tempdir().unwrap(); - let store = SqliteWorkspaceStore::open(temp.path().join("server.db")).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(temp.path().join("server.db")).unwrap(); store .with_conn(|conn| { conn.execute_batch( @@ -16103,7 +16130,7 @@ mod tests { let path = temp.path().join("server.db"); prepare_retained_schema(&path, 59); - let migrated = SqliteWorkspaceStore::open(&path).unwrap(); + let migrated = SqliteWorkspaceStore::open_for_test(&path).unwrap(); migrated .with_conn(|conn| { let table_count: i64 = conn.query_row( @@ -16139,7 +16166,7 @@ mod tests { let path = temp.path().join("server.db"); prepare_retained_schema(&path, 60); - drop(SqliteWorkspaceStore::open(&path).unwrap()); + drop(SqliteWorkspaceStore::open_for_test(&path).unwrap()); let conn = Connection::open(&path).unwrap(); assert_eq!( current_schema_version(&conn).unwrap(), @@ -16169,7 +16196,7 @@ mod tests { #[test] fn schema_v59_runtime_removal_migration_rolls_back_partial_ddl_and_marker() { let temp = tempfile::tempdir().unwrap(); - let store = SqliteWorkspaceStore::open(temp.path().join("server.db")).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(temp.path().join("server.db")).unwrap(); store .with_conn(|conn| { let trigger_names = { @@ -16433,7 +16460,7 @@ mod tests { fn fresh_database_uses_only_the_canonical_workspace_baseline() { let temp = tempfile::tempdir().unwrap(); let path = temp.path().join("server.db"); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); store .with_conn(|conn| { let migrations = conn @@ -16581,7 +16608,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx .collect::>(), (83..=LATEST_SCHEMA_VERSION).collect::>() ); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); let assignment = store .get_active_ticket_worker_assignment("workspace-a", "ticket") .unwrap() @@ -17052,7 +17079,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx let applied = SqliteWorkspaceStore::migrate_database(&path).unwrap(); assert_eq!(applied, plan); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); store .with_conn(|conn| { assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION); @@ -17601,7 +17628,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx "migrated databases must reject incomplete singleton leases" ); drop(conn); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); let legacy = RuntimeWorkerRef::new("legacy-runtime", "legacy-orchestrator"); let lease = store .require_current_worker_singleton_owner("workspace-a", &legacy) @@ -17789,7 +17816,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx .unwrap(); drop(conn); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); store .with_conn(|conn| { assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION); @@ -17890,7 +17917,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx .unwrap(); drop(conn); - let error = SqliteWorkspaceStore::open(&path) + let error = SqliteWorkspaceStore::open_for_test(&path) .err() .expect("migration must fail") .to_string(); @@ -17915,7 +17942,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx fn runtime_binding_identity_and_trust_uniqueness_are_workspace_scoped() { let temp = tempfile::tempdir().unwrap(); let path = temp.path().join("server.db"); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); store .with_conn(|conn| { conn.execute_batch( @@ -18017,7 +18044,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx ); drop(store); - let reopened = SqliteWorkspaceStore::open(&path).unwrap(); + let reopened = SqliteWorkspaceStore::open_for_test(&path).unwrap(); assert!( reopened .list_workspace_runtime_bindings("workspace-a", false) @@ -18067,7 +18094,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx fn workspace_runtime_connection_persists_without_authentication_evidence() { let temp = tempfile::tempdir().unwrap(); let path = temp.path().join("server.db"); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); store .with_conn(|conn| { conn.execute_batch( @@ -18112,7 +18139,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx .unwrap(); assert!(store.workspace_runtime_binding_matches(&persisted).unwrap()); drop(store); - let reopened = SqliteWorkspaceStore::open(&path).unwrap(); + let reopened = SqliteWorkspaceStore::open_for_test(&path).unwrap(); assert_eq!( reopened .get_workspace_runtime_binding("workspace-a", "runtime-a") @@ -18366,7 +18393,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx #[test] fn runtime_binding_key_mutations_are_identity_guarded_idempotent_and_audited() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); store .with_conn(|conn| { conn.execute_batch( @@ -18504,7 +18531,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx #[test] fn embedded_runtime_binding_can_explicitly_rotate_restart_identity() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); store .with_conn(|conn| { conn.execute_batch( @@ -18604,7 +18631,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx conn.execute_batch("DROP TABLE typed_ticket_artifacts") .unwrap(); - let result = SqliteWorkspaceStore::from_connection(conn); + let result = SqliteWorkspaceStore::from_connection_for_test(conn); let error = match result { Ok(_) => panic!("schema drift unexpectedly passed startup verification"), Err(error) => error, @@ -18616,7 +18643,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx async fn workspace_schema_rejects_cross_workspace_ticket_repository_reference_at_write_time() { let dir = tempfile::tempdir().unwrap(); let database_path = dir.path().join("workspace.sqlite"); - let store = SqliteWorkspaceStore::open(&database_path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&database_path).unwrap(); for workspace_id in ["workspace-a", "workspace-b"] { store .upsert_workspace(&WorkspaceRecord { @@ -18670,14 +18697,14 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx ); drop(backend); - SqliteWorkspaceStore::open(&database_path).unwrap(); + SqliteWorkspaceStore::open_for_test(&database_path).unwrap(); } #[tokio::test] async fn migrates_sqlite_and_preserves_workspace_record() { let dir = tempfile::tempdir().unwrap(); let db = dir.path().join("control-plane.sqlite"); - let store = SqliteWorkspaceStore::open(&db).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&db).unwrap(); assert_eq!(store.schema_version().await.unwrap(), LATEST_SCHEMA_VERSION); assert!( @@ -18695,7 +18722,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx }; store.upsert_workspace(&record).await.unwrap(); - let reopened = SqliteWorkspaceStore::open(&db).unwrap(); + let reopened = SqliteWorkspaceStore::open_for_test(&db).unwrap(); assert_eq!( reopened.schema_version().await.unwrap(), LATEST_SCHEMA_VERSION @@ -18710,7 +18737,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx async fn workspace_display_name_update_is_timestamp_guarded_and_preserves_identity() { let dir = tempfile::tempdir().unwrap(); let database_path = dir.path().join("server.db"); - let store = SqliteWorkspaceStore::open(&database_path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&database_path).unwrap(); let record = WorkspaceRecord { workspace_id: "workspace-a".to_string(), owner_account_id: "owner-account".to_string(), @@ -18749,7 +18776,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx updated ); drop(store); - let reopened = SqliteWorkspaceStore::open(database_path).unwrap(); + let reopened = SqliteWorkspaceStore::open_for_test(database_path).unwrap(); assert_eq!( reopened .get_workspace(&record.workspace_id) @@ -18763,7 +18790,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx #[tokio::test] async fn objective_creation_allocates_and_resolves_workspace_resource_key() { let dir = tempfile::tempdir().unwrap(); - let store = SqliteWorkspaceStore::open(dir.path().join("server.db")).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(dir.path().join("server.db")).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-a".into(), @@ -18809,7 +18836,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx #[tokio::test] async fn worker_create_reservation_allocates_uuid_before_runtime_and_replays_exact_input() { let dir = tempfile::tempdir().unwrap(); - let store = SqliteWorkspaceStore::open(dir.path().join("server.db")).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(dir.path().join("server.db")).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-a".to_string(), @@ -19113,7 +19140,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx async fn worker_singleton_restart_retry_replays_exact_workdir_intent_and_rejects_changes() { let dir = tempfile::tempdir().unwrap(); let database = dir.path().join("server.db"); - let store = SqliteWorkspaceStore::open(&database).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&database).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-a".to_string(), @@ -19173,7 +19200,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx .unwrap(); drop(store); - let reopened = SqliteWorkspaceStore::open(&database).unwrap(); + let reopened = SqliteWorkspaceStore::open_for_test(&database).unwrap(); let retried = reopened .reserve_worker_create( "workspace-a", @@ -19217,7 +19244,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx async fn worker_singleton_ownership_is_workspace_scoped_durable_and_fences_old_generations() { let dir = tempfile::tempdir().unwrap(); let database = dir.path().join("server.db"); - let store = SqliteWorkspaceStore::open(&database).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&database).unwrap(); for workspace_id in ["workspace-a", "workspace-b"] { store .upsert_workspace(&WorkspaceRecord { @@ -19330,7 +19357,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx ); drop(store); - let reopened = SqliteWorkspaceStore::open(database).unwrap(); + let reopened = SqliteWorkspaceStore::open_for_test(database).unwrap(); assert_eq!( reopened .current_worker_singleton_owner("workspace-a", key) @@ -19343,7 +19370,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx async fn concurrent_worker_singleton_reservations_have_one_winner() { let dir = tempfile::tempdir().unwrap(); let database = dir.path().join("server.db"); - let store = SqliteWorkspaceStore::open(&database).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&database).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-a".to_string(), @@ -19363,7 +19390,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx let database = database.clone(); let barrier = barrier.clone(); std::thread::spawn(move || { - let store = SqliteWorkspaceStore::open(database).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(database).unwrap(); let memory = store.get_workspace_memory_settings("workspace-a").unwrap(); barrier.wait(); store.reserve_worker_create( @@ -19383,7 +19410,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx .collect::>(); assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1); assert_eq!(results.iter().filter(|result| result.is_err()).count(), 1); - let store = SqliteWorkspaceStore::open(database).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(database).unwrap(); let owner = store .current_worker_singleton_owner("workspace-a", "shared-key") .unwrap() @@ -19399,7 +19426,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx #[tokio::test] async fn workspace_flow_sources_pin_content_and_builtins_stay_resources() { let dir = tempfile::tempdir().unwrap(); - let store = SqliteWorkspaceStore::open(dir.path().join("server.db")).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(dir.path().join("server.db")).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-a".to_string(), @@ -19492,7 +19519,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx "returning to earlier content must reuse the immutable content" ); drop(store); - let reopened = SqliteWorkspaceStore::open(dir.path().join("server.db")).unwrap(); + let reopened = SqliteWorkspaceStore::open_for_test(dir.path().join("server.db")).unwrap(); assert_eq!( reopened .get_flow_source_content( @@ -19748,7 +19775,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx } ); drop(conn); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); let backend = ticket::SqliteTicketBackend::open_verified(&path, "case-workspace").unwrap(); store .with_conn(|conn| { @@ -19915,7 +19942,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx ); drop(store); drop(backend); - let reopened = SqliteWorkspaceStore::open(&path).unwrap(); + let reopened = SqliteWorkspaceStore::open_for_test(&path).unwrap(); reopened .with_conn(|conn| { assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION); @@ -19982,7 +20009,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx before ); drop(conn); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); let backend = ticket::SqliteTicketBackend::open_verified(&path, "case-workspace").unwrap(); assert_eq!( backend @@ -20116,8 +20143,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("equivalence.db"); prepare_retained_schema(&path, 90); - let upgraded = SqliteWorkspaceStore::open(&path).unwrap(); - let fresh = SqliteWorkspaceStore::in_memory().unwrap(); + let upgraded = SqliteWorkspaceStore::open_for_test(&path).unwrap(); + let fresh = SqliteWorkspaceStore::in_memory_for_test().unwrap(); for object in [ "workspace_tags", "typed_ticket_tags", @@ -20170,7 +20197,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx #[test] fn schema_v91_case_constraint_verification_rejects_legacy_view_predicates() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); store.with_conn(|conn| { conn.execute_batch("DROP VIEW ticket_active_worker_assignments; CREATE VIEW ticket_active_worker_assignments AS @@ -20331,7 +20358,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx // boundary under test. Targets need not be materialized by a Runtime. async fn manual_worker_start_store() -> (tempfile::TempDir, SqliteWorkspaceStore) { let dir = tempfile::tempdir().unwrap(); - let store = SqliteWorkspaceStore::open(dir.path().join("workspace.db")).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(dir.path().join("workspace.db")).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "w".into(), @@ -20419,7 +20446,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx let threads = (0..2) .map(|index| { let store = - SqliteWorkspaceStore::open(dir.path().join("workspace.db")).unwrap(); + SqliteWorkspaceStore::open_for_test(dir.path().join("workspace.db")) + .unwrap(); let barrier = barrier.clone(); std::thread::spawn(move || { barrier.wait(); @@ -20784,7 +20812,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx .retain_ticket_claim_recovery("w", "claim", &fingerprint, &changed) .is_err() ); - let reopened = SqliteWorkspaceStore::open(dir.path().join("workspace.db")).unwrap(); + let reopened = + SqliteWorkspaceStore::open_for_test(dir.path().join("workspace.db")).unwrap(); assert_eq!( reopened.get_ticket_claim_recovery("w", "claim").unwrap(), Some(recovery) @@ -20842,7 +20871,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx store .rollback_ticket_assignment_operation("w", "claim") .unwrap(); - let reopened = SqliteWorkspaceStore::open(dir.path().join("workspace.db")).unwrap(); + let reopened = + SqliteWorkspaceStore::open_for_test(dir.path().join("workspace.db")).unwrap(); let receipt = reopened .get_ticket_assignment_operation("w", "claim") .unwrap() @@ -21062,7 +21092,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx use ticket::TicketBackend; let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("workspace.db"); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "w".into(), @@ -21190,7 +21220,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx async fn ticket_worker_assignment_replaces_current_and_preserves_audit_history() { let dir = tempfile::tempdir().unwrap(); let db = dir.path().join("server.db"); - let store = SqliteWorkspaceStore::open(&db).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&db).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-a".to_string(), @@ -21379,7 +21409,7 @@ INSERT INTO worker_registry ( ) .unwrap(); drop(store); - let store = SqliteWorkspaceStore::open(&db).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&db).unwrap(); let pending = store .get_ticket_assignment_operation("workspace-a", "reserved-operation") .unwrap() @@ -21503,7 +21533,7 @@ INSERT INTO worker_registry ( async fn role_assignment_routes_and_manual_start_are_state_fenced_atomically() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().join("workspace.db"); - let store = SqliteWorkspaceStore::open(&db_path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&db_path).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-role".to_string(), @@ -21792,7 +21822,7 @@ INSERT INTO worker_registry ( fn startup_rejects_missing_workspace_resource_constraint_trigger() { let temp = tempfile::tempdir().unwrap(); let path = temp.path().join("server.db"); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); store .with_conn(|conn| { conn.execute_batch("DROP TRIGGER ticket_worker_assignments_validate_insert")?; @@ -21801,7 +21831,7 @@ INSERT INTO worker_registry ( .unwrap(); drop(store); - let error = match SqliteWorkspaceStore::open(&path) { + let error = match SqliteWorkspaceStore::open_for_test(&path) { Ok(_) => panic!("missing assignment constraint trigger must fail closed"), Err(error) => error, }; @@ -21817,7 +21847,7 @@ INSERT INTO worker_registry ( fn startup_accepts_retained_assignment_history_after_parent_cleanup() { let temp = tempfile::tempdir().unwrap(); let path = temp.path().join("server.db"); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); store .with_conn(|conn| { conn.execute_batch( @@ -21881,7 +21911,7 @@ INSERT INTO worker_registry ( .unwrap(); drop(store); - let reopened = SqliteWorkspaceStore::open(&path).unwrap(); + let reopened = SqliteWorkspaceStore::open_for_test(&path).unwrap(); reopened .with_conn(|conn| { let retained: i64 = conn.query_row( @@ -21909,7 +21939,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn workspace_bootstrap_validates_key_and_rejects_duplicate_workspace() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); store .upsert_account(&AccountRecord { account_id: "owner-account".to_string(), @@ -22041,7 +22071,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn repository_records_round_trip() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); assert_eq!(store.schema_version().await.unwrap(), LATEST_SCHEMA_VERSION); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), @@ -22122,7 +22152,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn memory_authority_records_round_trip_and_close_staging() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); assert_eq!(store.schema_version().await.unwrap(), LATEST_SCHEMA_VERSION); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), @@ -22197,7 +22227,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn legacy_memory_reset_is_atomic_scoped_idempotent_and_exact() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); for (workspace_id, display_name) in [ ("local-dev", "Local Dev"), ("other-workspace", "Other Workspace"), @@ -22364,7 +22394,7 @@ INSERT INTO worker_registry ( async fn worker_workdir_registry_round_trips_and_preserves_pinned_retention() { let temp = tempfile::tempdir().unwrap(); let db = temp.path().join("workspace.db"); - let store = SqliteWorkspaceStore::open(&db).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&db).unwrap(); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: "owner-account".to_string(), @@ -22733,7 +22763,7 @@ INSERT INTO worker_registry ( ); } drop(store); - let store = SqliteWorkspaceStore::open(&db).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&db).unwrap(); assert_eq!( store .list_worker_workdir_links_page("local-dev", &worker.worker, 1, 0, None) @@ -22763,7 +22793,7 @@ INSERT INTO worker_registry ( let workdir_conflict = store.attach_worker_workdir(&workdir_conflict).unwrap(); drop(store); - let reopened = SqliteWorkspaceStore::open(&db).unwrap(); + let reopened = SqliteWorkspaceStore::open_for_test(&db).unwrap(); assert_eq!( reopened .list_worker_workdir_links("local-dev", &worker.worker) @@ -22781,7 +22811,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn worker_projection_preserves_equal_local_ids_from_distinct_runtimes() { let temp = tempfile::tempdir().unwrap(); - let store = SqliteWorkspaceStore::open(temp.path().join("workspace.db")).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(temp.path().join("workspace.db")).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "local-dev".to_string(), @@ -22907,7 +22937,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn worker_projection_reconciles_snapshots_and_preserves_unavailable_lifecycle() { let temp = tempfile::tempdir().unwrap(); - let store = SqliteWorkspaceStore::open(temp.path().join("workspace.db")).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(temp.path().join("workspace.db")).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "local-dev".to_string(), @@ -23082,7 +23112,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn late_runtime_event_does_not_recreate_retention_deleted_catalog_worker() { let temp = tempfile::tempdir().unwrap(); - let store = SqliteWorkspaceStore::open(temp.path().join("workspace.db")).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(temp.path().join("workspace.db")).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "local-dev".to_string(), @@ -23177,7 +23207,7 @@ INSERT INTO worker_registry ( async fn worker_control_grants_are_idempotent_scoped_and_support_internal_invalidation() { let dir = tempfile::tempdir().unwrap(); let database = dir.path().join("control-grants.db"); - let store = SqliteWorkspaceStore::open(&database).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&database).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-control".to_string(), @@ -23247,7 +23277,7 @@ INSERT INTO worker_registry ( ); drop(store); - let store = SqliteWorkspaceStore::open(&database).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&database).unwrap(); assert_eq!( store .list_active_worker_control_grants( @@ -23305,7 +23335,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn account_and_login_records_round_trip() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); assert_eq!(store.schema_version().await.unwrap(), LATEST_SCHEMA_VERSION); let now = "2026-07-22T00:00:00Z".to_string(); let account = AccountRecord { @@ -23664,9 +23694,9 @@ INSERT INTO worker_registry ( async fn backend_job_resource_reserve_converges_across_connections_without_changing_input() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("resource-jobs.sqlite"); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); seed_resource_job_workspace(&store).await; - let other = SqliteWorkspaceStore::open(&path).unwrap(); + let other = SqliteWorkspaceStore::open_for_test(&path).unwrap(); let requests = [ resource_job_request("first", "subject-a", "candidate-1"), resource_job_request("followup", "subject-a", "candidate-2"), @@ -23782,7 +23812,7 @@ INSERT INTO worker_registry ( async fn backend_job_resource_retry_and_unknown_are_fenced_by_durable_cleanup() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("cleanup-jobs.sqlite"); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); seed_resource_job_workspace(&store).await; let request = resource_job_request("failed", "subject-a", "candidate-1"); let first = store @@ -23888,7 +23918,7 @@ INSERT INTO worker_registry ( .unwrap(); finish_resource_job_cleanup(&store, &second); drop(store); - let restarted = SqliteWorkspaceStore::open(&path).unwrap(); + let restarted = SqliteWorkspaceStore::open_for_test(&path).unwrap(); let active = restarted .find_active_backend_job_for_resource( "resource-workspace", @@ -23948,7 +23978,7 @@ INSERT INTO worker_registry ( async fn backend_job_resource_retry_races_new_intent_atomically_and_remains_bounded() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("retry-race.sqlite"); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); seed_resource_job_workspace(&store).await; // Generic resources use the same common reservation API without domain grants. let mut request = resource_job_request("generic", "subject-a", "candidate-1"); @@ -23971,7 +24001,7 @@ INSERT INTO worker_registry ( let mut followup = request.clone(); followup.job_id = "generic-followup".into(); followup.input = serde_json::json!({"candidate": "new"}); - let other = SqliteWorkspaceStore::open(&path).unwrap(); + let other = SqliteWorkspaceStore::open_for_test(&path).unwrap(); let barrier = std::sync::Barrier::new(2); let (retry, reserve) = std::thread::scope(|scope| { let a = scope.spawn(|| { @@ -24057,7 +24087,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn backend_job_all_older_attempt_cleanups_fence_retry_and_resource_release() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); seed_resource_job_workspace(&store).await; let mut request = resource_job_request("legacy-overlap", "subject-a", "candidate-1"); request.limits.max_attempts = 3; @@ -24181,7 +24211,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn backend_job_empty_batch_surface_refresh_locks_subject_and_defers_candidates() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); seed_resource_job_workspace(&store).await; for surface_state in ["ungenerated", "stale", "failed"] { let subject = format!("subject-{surface_state}"); @@ -24301,7 +24331,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn backend_job_completed_resource_releases_only_after_worker_cleanup() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); seed_resource_job_workspace(&store).await; let request = resource_job_request("completed", "subject-a", "candidate-1"); let first = store @@ -24355,7 +24385,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn backend_jobs_fence_results_project_ownership_and_bound_retries() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); let workspace_id = "workspace-job"; store .upsert_account(&AccountRecord { diff --git a/crates/workspace-server/src/ticket_tag_upgrade_tests.rs b/crates/workspace-server/src/ticket_tag_upgrade_tests.rs index 0ea9c64e7..69190eb1f 100644 --- a/crates/workspace-server/src/ticket_tag_upgrade_tests.rs +++ b/crates/workspace-server/src/ticket_tag_upgrade_tests.rs @@ -263,7 +263,7 @@ fn assert_retained_classification_upgrade(workspace_version: i64, component: i64 let tags_before = retained.as_ref().map(|_| classification_rows(&conn)); let auth_before = fixture_auth_rows(&conn); drop(conn); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); store.with_conn(|conn| { assert_eq!(current_schema_version(conn)?,93); assert_eq!(conn.query_row("SELECT MAX(version) FROM ticket_schema_migrations",[],|row|row.get::<_,i64>(0))?,11); @@ -490,7 +490,7 @@ fn workspace90_cases9_add_classification_without_reinterpreting_reasoned_history let metadata_before = stored_fixture_rows(&conn, "typed_ticket_case_metadata"); let auth_before = fixture_auth_rows(&conn); drop(conn); - let store = SqliteWorkspaceStore::open(&path).unwrap(); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); store .with_conn(|conn| { assert_eq!(current_schema_version(conn)?,93); diff --git a/crates/workspace-server/src/workdir_create_operations.rs b/crates/workspace-server/src/workdir_create_operations.rs index a41fc8ca7..a98284b1e 100644 --- a/crates/workspace-server/src/workdir_create_operations.rs +++ b/crates/workspace-server/src/workdir_create_operations.rs @@ -750,7 +750,7 @@ mod tests { #[test] fn retry_keeps_resolved_config_and_credential_candidates_after_fallback_moves() { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); futures::executor::block_on(store.upsert_workspace(&WorkspaceRecord { workspace_id: "workspace".to_string(), owner_account_id: "owner-account".to_string(), diff --git a/crates/workspace-server/src/workdir_removal.rs b/crates/workspace-server/src/workdir_removal.rs index 9c1f2a327..13b481fb4 100644 --- a/crates/workspace-server/src/workdir_removal.rs +++ b/crates/workspace-server/src/workdir_removal.rs @@ -925,7 +925,7 @@ mod tests { } async fn seeded_store() -> (SqliteWorkspaceStore, WorkdirRegistryRecord) { - let store = SqliteWorkspaceStore::in_memory().unwrap(); + let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); store .upsert_account(&AccountRecord { account_id: "account-a".to_string(), diff --git a/crates/workspace-server/src/worker_projection.rs b/crates/workspace-server/src/worker_projection.rs index cb6a10dde..fc8140132 100644 --- a/crates/workspace-server/src/worker_projection.rs +++ b/crates/workspace-server/src/worker_projection.rs @@ -463,7 +463,7 @@ mod tests { ) { let temp = tempfile::tempdir().unwrap(); let store = Arc::new( - SqliteWorkspaceStore::open(temp.path().join("workspace.db")) + SqliteWorkspaceStore::open_for_test(temp.path().join("workspace.db")) .unwrap() .with_db_executor(crate::db_executor::DbExecutor::new(Default::default()).unwrap()), ); diff --git a/crates/workspace-server/src/worker_source/runtime_request.rs b/crates/workspace-server/src/worker_source/runtime_request.rs index d299b520d..b8c2e7226 100644 --- a/crates/workspace-server/src/worker_source/runtime_request.rs +++ b/crates/workspace-server/src/worker_source/runtime_request.rs @@ -392,7 +392,7 @@ mod tests { }) .unwrap(); let store = Arc::new( - SqliteWorkspaceStore::open(&db_path) + SqliteWorkspaceStore::open_for_test(&db_path) .unwrap() .with_db_executor(executor), ); diff --git a/crates/workspace-server/src/workspace_catalog.rs b/crates/workspace-server/src/workspace_catalog.rs index 5d9679d9f..78ce46f45 100644 --- a/crates/workspace-server/src/workspace_catalog.rs +++ b/crates/workspace-server/src/workspace_catalog.rs @@ -285,7 +285,7 @@ mod tests { #[tokio::test] async fn create_is_atomic_and_exact_retries_converge() { - let store = Arc::new(SqliteWorkspaceStore::in_memory().unwrap()); + let store = Arc::new(SqliteWorkspaceStore::in_memory_for_test().unwrap()); let service = WorkspaceCatalogService::new( store.clone(), Arc::new( @@ -339,7 +339,7 @@ mod tests { #[test] fn create_rejects_noncanonical_repository_keys_without_persisting_a_workspace() { - let store = Arc::new(SqliteWorkspaceStore::in_memory().unwrap()); + let store = Arc::new(SqliteWorkspaceStore::in_memory_for_test().unwrap()); let service = WorkspaceCatalogService::new( store.clone(), Arc::new( @@ -378,7 +378,7 @@ mod tests { for with_repository in [false, true] { let temp = tempfile::tempdir().unwrap(); let database_path = temp.path().join("server.db"); - let store = Arc::new(SqliteWorkspaceStore::open(&database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&database_path).unwrap()); let owner_account_id = owner_account(store.as_ref()); let materials = Arc::new(InMemoryWorkspaceSigningMaterialStore::default()); let service = WorkspaceCatalogService::new( @@ -418,7 +418,7 @@ mod tests { drop(service); drop(store); - let store = Arc::new(SqliteWorkspaceStore::open(&database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&database_path).unwrap()); let restarted = WorkspaceCatalogService::new(store.clone(), materials); let created = restarted.create(request, owner_account_id).unwrap(); let identity = store @@ -435,7 +435,7 @@ mod tests { for with_repository in [false, true] { let temp = tempfile::tempdir().unwrap(); let database_path = temp.path().join("server.db"); - let store = Arc::new(SqliteWorkspaceStore::open(&database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&database_path).unwrap()); let owner_account_id = owner_account(store.as_ref()); let materials = Arc::new(InMemoryWorkspaceSigningMaterialStore::default()); let service = WorkspaceCatalogService::new(store.clone(), materials.clone()); @@ -486,7 +486,7 @@ mod tests { drop(service); drop(store); - let store = Arc::new(SqliteWorkspaceStore::open(&database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&database_path).unwrap()); let restarted = WorkspaceCatalogService::new(store.clone(), materials); let created = restarted.create(request, owner_account_id).unwrap(); let identity = store @@ -500,7 +500,7 @@ mod tests { #[tokio::test] async fn idempotency_key_reuse_with_different_payload_is_rejected() { - let store = Arc::new(SqliteWorkspaceStore::in_memory().unwrap()); + let store = Arc::new(SqliteWorkspaceStore::in_memory_for_test().unwrap()); let owner_account_id = owner_account(store.as_ref()); let service = WorkspaceCatalogService::new( store, @@ -532,7 +532,7 @@ mod tests { #[test] fn repository_free_create_initializes_authority_and_concurrent_retries_converge() { - let store = Arc::new(SqliteWorkspaceStore::in_memory().unwrap()); + let store = Arc::new(SqliteWorkspaceStore::in_memory_for_test().unwrap()); let owner = owner_account(store.as_ref()); let service = WorkspaceCatalogService::new( store.clone(), @@ -611,7 +611,7 @@ mod tests { #[test] fn replay_reads_current_repository_at_initial_key_and_never_recreates_it() { - let store = Arc::new(SqliteWorkspaceStore::in_memory().unwrap()); + let store = Arc::new(SqliteWorkspaceStore::in_memory_for_test().unwrap()); let owner = owner_account(store.as_ref()); let service = WorkspaceCatalogService::new( store.clone(), @@ -673,13 +673,13 @@ mod tests { repository: None, }; let (owner, created) = { - let store = Arc::new(SqliteWorkspaceStore::open(&database).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&database).unwrap()); let owner = owner_account(store.as_ref()); let service = WorkspaceCatalogService::new(store, materials.clone()); let created = service.create(request.clone(), owner.clone()).unwrap(); (owner, created) }; - let store = Arc::new(SqliteWorkspaceStore::open(&database).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&database).unwrap()); let reopened = WorkspaceCatalogService::new(store.clone(), materials); let replay = reopened.create(request, owner).unwrap(); assert_eq!(replay.workspace, created.workspace); @@ -739,7 +739,7 @@ mod tests { #[test] fn create_rejects_non_user_account_owners() { - let store = Arc::new(SqliteWorkspaceStore::in_memory().unwrap()); + let store = Arc::new(SqliteWorkspaceStore::in_memory_for_test().unwrap()); store .upsert_account(&AccountRecord { account_id: "organization-owner".to_string(), @@ -777,7 +777,7 @@ mod tests { #[test] fn catalog_list_is_scoped_to_the_required_owner_account() { - let store = Arc::new(SqliteWorkspaceStore::in_memory().unwrap()); + let store = Arc::new(SqliteWorkspaceStore::in_memory_for_test().unwrap()); let owner_a = owner_account(store.as_ref()); let owner_b = "account-owner-b".to_string(); store @@ -845,7 +845,7 @@ mod tests { #[test] fn remote_repository_creation_persists_typed_source_without_auth_metadata() { - let store = Arc::new(SqliteWorkspaceStore::in_memory().unwrap()); + let store = Arc::new(SqliteWorkspaceStore::in_memory_for_test().unwrap()); let owner_account_id = owner_account(store.as_ref()); let service = WorkspaceCatalogService::new( store.clone(), diff --git a/crates/workspace-server/src/workspace_deletion.rs b/crates/workspace-server/src/workspace_deletion.rs index e241f8aee..b68c7e4af 100644 --- a/crates/workspace-server/src/workspace_deletion.rs +++ b/crates/workspace-server/src/workspace_deletion.rs @@ -922,7 +922,7 @@ mod tests { use tempfile::tempdir; fn setup() -> (SqliteWorkspaceStore, String, String) { - seed_workspaces(SqliteWorkspaceStore::in_memory().expect("store")) + seed_workspaces(SqliteWorkspaceStore::in_memory_for_test().expect("store")) } fn seed_workspaces(store: SqliteWorkspaceStore) -> (SqliteWorkspaceStore, String, String) { @@ -989,7 +989,7 @@ mod tests { let dir = tempdir().expect("tempdir"); let path = dir.path().join("server.db"); let (store, owner, workspace_id) = - seed_workspaces(SqliteWorkspaceStore::open(&path).expect("store")); + seed_workspaces(SqliteWorkspaceStore::open_for_test(&path).expect("store")); for workspace in ["workspace-a", "workspace-b"] { let backend = ticket::SqliteTicketBackend::open_verified(&path, workspace).unwrap(); let initial = backend diff --git a/crates/workspace-server/src/workspace_signing_identity.rs b/crates/workspace-server/src/workspace_signing_identity.rs index f724f2ca1..959f28805 100644 --- a/crates/workspace-server/src/workspace_signing_identity.rs +++ b/crates/workspace-server/src/workspace_signing_identity.rs @@ -780,7 +780,7 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let database_path = temp.path().join("server.db"); - let store = Arc::new(SqliteWorkspaceStore::open(&database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&database_path).unwrap()); store .upsert_account(&AccountRecord { account_id: "account-1".to_string(), @@ -833,7 +833,7 @@ mod tests { drop(failing_service); drop(store); - let store = Arc::new(SqliteWorkspaceStore::open(&database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&database_path).unwrap()); let service = WorkspaceSigningIdentityService::new(store.clone(), materials.clone()); let provisioned = service .provision_existing("workspace-1", "account-1") @@ -962,7 +962,7 @@ mod tests { ); drop(service); drop(store); - let store = Arc::new(SqliteWorkspaceStore::open(&database_path).unwrap()); + let store = Arc::new(SqliteWorkspaceStore::open_for_test(&database_path).unwrap()); let restarted = WorkspaceSigningIdentityService::new(store.clone(), materials.clone()); let recovered = restarted .provision_existing("workspace-2", "account-1") diff --git a/crates/workspace-server/src/workspace_subscription.rs b/crates/workspace-server/src/workspace_subscription.rs index eaa88d40e..a62faa34e 100644 --- a/crates/workspace-server/src/workspace_subscription.rs +++ b/crates/workspace-server/src/workspace_subscription.rs @@ -738,10 +738,11 @@ mod tests { AccountRecord, ControlPlaneStore, SqliteWorkspaceStore, WorkerRegistryRecord, WorkspaceRecord, }; - let store = - std::sync::Arc::new(SqliteWorkspaceStore::in_memory().unwrap().with_db_executor( - crate::db_executor::DbExecutor::new(Default::default()).unwrap(), - )); + let store = std::sync::Arc::new( + SqliteWorkspaceStore::in_memory_for_test() + .unwrap() + .with_db_executor(crate::db_executor::DbExecutor::new(Default::default()).unwrap()), + ); store .upsert_account(&AccountRecord { account_id: "owner".into(),