fix: isolate finite DB and Host execution ownership in fixtures

This commit is contained in:
2026-10-12 07:55:07 +09:00
parent 61533fadb6
commit f3d4aba54e
43 changed files with 579 additions and 328 deletions
@@ -119,7 +119,7 @@ async fn assignment_delivery_response_loss_returns_immutable_payload_after_progr
.update_ticket_assignment_tracking("w", "tracking", &update, &source) .update_ticket_assignment_tracking("w", "tracking", &update, &source)
.unwrap(); .unwrap();
drop(store); 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!( assert_eq!(
store store
.list_pending_ticket_assignment_deliveries("w", 100) .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 path = directory.path().join("workspace.db");
let input = pending.input_event_id.clone(); let input = pending.input_event_id.clone();
std::thread::spawn(move || { std::thread::spawn(move || {
let store = SqliteWorkspaceStore::open(path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(path).unwrap();
barrier.wait(); barrier.wait();
store store
.reserve_ticket_assignment_delivery("w", "a", &input, &tracking_store_source()) .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" "original rendered prompt"
); );
drop(store); 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!( assert_eq!(
store store
.prepare_ticket_assignment_delivery_content( .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 operations = stored_fixture_rows(&conn, "ticket_assignment_operations");
let tracking = stored_fixture_rows(&conn, "typed_ticket_assignment_tracking_history"); let tracking = stored_fixture_rows(&conn, "typed_ticket_assignment_tracking_history");
drop(conn); drop(conn);
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
let pending = store let pending = store
.list_pending_ticket_assignment_deliveries("w", 100) .list_pending_ticket_assignment_deliveries("w", 100)
.unwrap(); .unwrap();
@@ -656,7 +656,7 @@ async fn assignment_startup_simultaneous_reservations_admit_runtime_effects_once
let barrier = barrier.clone(); let barrier = barrier.clone();
let path = directory.path().join("workspace.db"); let path = directory.path().join("workspace.db");
std::thread::spawn(move || { std::thread::spawn(move || {
let store = SqliteWorkspaceStore::open(path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(path).unwrap();
barrier.wait(); barrier.wait();
delivery_store_start_reserve(&store, "start-a", "a", &tracking_store_source()) delivery_store_start_reserve(&store, "start-a", "a", &tracking_store_source())
.unwrap() .unwrap()
@@ -688,8 +688,8 @@ fn assignment_delivery_fresh_and_upgraded_operational_schemas_are_identical() {
let conn = tracking_schema91_fixture(&path); let conn = tracking_schema91_fixture(&path);
migrate_ticket_tracking_v91_to_v92(&conn).unwrap(); migrate_ticket_tracking_v91_to_v92(&conn).unwrap();
drop(conn); drop(conn);
let upgraded = SqliteWorkspaceStore::open(&path).unwrap(); let upgraded = SqliteWorkspaceStore::open_for_test(&path).unwrap();
let fresh = SqliteWorkspaceStore::in_memory().unwrap(); let fresh = SqliteWorkspaceStore::in_memory_for_test().unwrap();
let objects = |store: &SqliteWorkspaceStore| { let objects = |store: &SqliteWorkspaceStore| {
store.with_conn(|conn| { 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")?; 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(); .unwrap();
drop(store); drop(store);
assert!( 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"))
); );
} }
@@ -78,7 +78,7 @@ async fn tracking_store_every_replay_fences_current_singleton_and_authenticated_
); );
// Model the authority change after authenticated source capture, before // Model the authority change after authenticated source capture, before
// BEGIN IMMEDIATE on a second connection. Receipt presence cannot bypass it. // 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| { second.with_conn_mut(|conn| {
let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?; let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
match change { match change {
@@ -233,7 +233,7 @@ async fn tracking_store_same_principal_reissue_and_exact_retries_keep_original_r
.is_err() .is_err()
); );
drop(store); 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!( assert_eq!(
reopened reopened
.get_ticket_assignment_tracking("w", "tracking", "a") .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<_>>(), .collect::<Vec<_>>(),
vec![92, 93] vec![92, 93]
); );
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
store store
.with_conn(|conn| { .with_conn(|conn| {
assert_eq!(current_schema_version(conn)?, 93); 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") .list_ticket_assignment_tracking_history("w", "pending", "a-pending")
.unwrap(); .unwrap();
drop(store); drop(store);
let reopened = SqliteWorkspaceStore::open(&path).unwrap(); let reopened = SqliteWorkspaceStore::open_for_test(&path).unwrap();
assert_eq!( assert_eq!(
reopened reopened
.list_ticket_assignment_tracking_history("w", "pending", "a-pending") .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"); let releases_before = stored_fixture_rows(&conn, "ticket_assignment_work_releases");
drop(conn); 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 backend = ticket::SqliteTicketBackend::open_verified(&path, "w").unwrap();
let reopen = || { let reopen = || {
let observed = backend.show("historical-done".into()).unwrap(); 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 // A second observer holds the same Assign baseline. Only the first
// explicit writer may consume it; BEGIN IMMEDIATE serializes the CAS. // 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!( assert_eq!(
second second
.get_ticket_assignment_tracking("w", "historical-done", "a-historical-done") .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(); }).unwrap();
drop(second); drop(second);
drop(store); drop(store);
let reopened_store = SqliteWorkspaceStore::open(&path).unwrap(); let reopened_store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
assert_eq!( assert_eq!(
reopened_store reopened_store
.get_ticket_assignment_tracking("w", "historical-done", "a-historical-done") .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(); .unwrap();
drop(conn); drop(conn);
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
let unknown = store let unknown = store
.get_ticket_assignment_tracking("w", "historical-done", "a-historical-done") .get_ticket_assignment_tracking("w", "historical-done", "a-historical-done")
.unwrap() .unwrap()
@@ -1065,7 +1065,7 @@ async fn orchestrator_assignment_all_runtime_retries_fence_singleton_and_binding
&source, &source,
) )
.unwrap(); .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 { 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'",[])?;}, "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'",[])?;}, "revoke" => {tx.execute("UPDATE workspace_runtime_bindings SET revoked_at='revoked' WHERE workspace_id='w'",[])?;},
+1 -1
View File
@@ -203,7 +203,7 @@ mod tests {
}) })
.unwrap(); .unwrap();
let store = Arc::new( let store = Arc::new(
SqliteWorkspaceStore::in_memory() SqliteWorkspaceStore::in_memory_for_test()
.unwrap() .unwrap()
.with_db_executor(executor), .with_db_executor(executor),
); );
+4 -4
View File
@@ -3241,7 +3241,7 @@ mod tests {
async fn unfinished_dependencies_remain_diagnostic_without_disabling_manual_start() { async fn unfinished_dependencies_remain_diagnostic_without_disabling_manual_start() {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("workspace.db"); let path = dir.path().join("workspace.db");
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
store store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "w".into(), workspace_id: "w".into(),
@@ -3316,7 +3316,7 @@ mod tests {
async fn manual_worker_eligibility_accepts_explicit_states_without_reviving_released_work() { async fn manual_worker_eligibility_accepts_explicit_states_without_reviving_released_work() {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("workspace.db"); let path = dir.path().join("workspace.db");
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
store store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "w".into(), workspace_id: "w".into(),
@@ -3449,7 +3449,7 @@ mod tests {
write_ticket(dir.path(), "00000000001J5", "Second ticket", "queued"); write_ticket(dir.path(), "00000000001J5", "Second ticket", "queued");
write_ticket(dir.path(), "00000000001J6", "Third ticket", "planning"); write_ticket(dir.path(), "00000000001J6", "Third ticket", "planning");
let db_path = dir.path().join("workspace.db"); 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 store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "workspace-test".to_string(), 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() { async fn objective_mutations_write_sqlite_records_and_audit_events() {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("workspace.db"); 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 store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "workspace-test".to_string(), workspace_id: "workspace-test".to_string(),
+6 -6
View File
@@ -795,7 +795,7 @@ mod tests {
} }
async fn open_store() -> SqliteWorkspaceStore { 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.upsert_workspace(&workspace()).await.unwrap();
store store
} }
@@ -1454,7 +1454,7 @@ mod tests {
#[tokio::test] #[tokio::test]
async fn invalid_candidate_is_never_persisted() { 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(); store.upsert_workspace(&workspace()).await.unwrap();
let current = store.load_workspace_config("w-config").unwrap().unwrap(); let current = store.load_workspace_config("w-config").unwrap().unwrap();
let main = current.snapshot.get(&path(MAIN_CONFIG_ENTRYPOINT)).unwrap(); let main = current.snapshot.get(&path(MAIN_CONFIG_ENTRYPOINT)).unwrap();
@@ -1478,7 +1478,7 @@ mod tests {
#[tokio::test] #[tokio::test]
async fn valid_candidate_commits_snapshot_content_and_provenance_atomically() { 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(); store.upsert_workspace(&workspace()).await.unwrap();
let current = store.load_workspace_config("w-config").unwrap().unwrap(); let current = store.load_workspace_config("w-config").unwrap().unwrap();
let committed = store let committed = store
@@ -1496,7 +1496,7 @@ mod tests {
#[tokio::test] #[tokio::test]
async fn stale_cas_cannot_overwrite_newer_tree() { 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(); store.upsert_workspace(&workspace()).await.unwrap();
let current = store.load_workspace_config("w-config").unwrap().unwrap(); let current = store.load_workspace_config("w-config").unwrap().unwrap();
let request = commit_request(&current, vec![update_main(&current, "{ answer = 42; }")]); let request = commit_request(&current, vec![update_main(&current, "{ answer = 42; }")]);
@@ -1514,7 +1514,7 @@ mod tests {
#[tokio::test] #[tokio::test]
async fn committed_content_remains_retrievable_after_later_commit() { 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(); store.upsert_workspace(&workspace()).await.unwrap();
let current = store.load_workspace_config("w-config").unwrap().unwrap(); let current = store.load_workspace_config("w-config").unwrap().unwrap();
let first = store let first = store
@@ -1772,7 +1772,7 @@ mod tests {
#[test] #[test]
fn migration_creates_config_authority_tables() { fn migration_creates_config_authority_tables() {
let store = SqliteWorkspaceStore::in_memory().unwrap(); let store = SqliteWorkspaceStore::in_memory_for_test().unwrap();
store store
.with_conn(|conn| { .with_conn(|conn| {
for table in [ for table in [
@@ -818,7 +818,7 @@ mod current_authority {
use super::*; use super::*;
fn binding_store() -> (SqliteWorkspaceStore, WorkspaceRuntimeBinding) { fn binding_store() -> (SqliteWorkspaceStore, WorkspaceRuntimeBinding) {
let store = SqliteWorkspaceStore::in_memory().unwrap(); let store = SqliteWorkspaceStore::in_memory_for_test().unwrap();
let runtime_key = let runtime_key =
worker_runtime::auth::RuntimeIdentityMaterial::generate("runtime-key").unwrap(); worker_runtime::auth::RuntimeIdentityMaterial::generate("runtime-key").unwrap();
store.with_conn(|conn| { store.with_conn(|conn| {
+1 -1
View File
@@ -6394,7 +6394,7 @@ mod tests {
}) })
.unwrap(); .unwrap();
let store = Arc::new( let store = Arc::new(
SqliteWorkspaceStore::in_memory() SqliteWorkspaceStore::in_memory_for_test()
.unwrap() .unwrap()
.with_db_executor(executor.clone()), .with_db_executor(executor.clone()),
); );
@@ -32,7 +32,8 @@ struct Fixture {
impl Fixture { impl Fixture {
async fn new() -> Self { async fn new() -> Self {
let temp = tempfile::tempdir().unwrap(); 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 store
.upsert_account(&AccountRecord { .upsert_account(&AccountRecord {
account_id: "owner".into(), account_id: "owner".into(),
@@ -45,7 +45,7 @@ fn evidence_column_cutover_preserves_all_values_and_current_schema_names() {
let artifacts = evidence_rows(&conn, "artifacts"); let artifacts = evidence_rows(&conn, "artifacts");
let receipts = evidence_rows(&conn, "repository_secret_legacy_receipts"); let receipts = evidence_rows(&conn, "repository_secret_legacy_receipts");
drop(conn); drop(conn);
let upgraded = SqliteWorkspaceStore::open(&path).unwrap(); let upgraded = SqliteWorkspaceStore::open_for_test(&path).unwrap();
upgraded upgraded
.with_conn(|conn| { .with_conn(|conn| {
assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION); 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(()) Ok(())
}) })
.unwrap(); .unwrap();
for store in [upgraded, SqliteWorkspaceStore::in_memory().unwrap()] { for store in [upgraded, SqliteWorkspaceStore::in_memory_for_test().unwrap()] {
store.with_conn(|conn| { 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 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::<rusqlite::Result<Vec<_>>>()?; let objects = stmt.query_map([],|row|Ok((row.get::<_,String>(0)?,row.get::<_,String>(1)?)))?.collect::<rusqlite::Result<Vec<_>>>()?;
@@ -217,7 +217,7 @@ mod tests {
async fn authority() -> (TempDir, SqliteWorkspaceAuthority) { async fn authority() -> (TempDir, SqliteWorkspaceAuthority) {
let temp = TempDir::new().unwrap(); let temp = TempDir::new().unwrap();
let db_path = temp.path().join("workspace.db"); 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 store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "workspace-test".to_string(), workspace_id: "workspace-test".to_string(),
@@ -32,7 +32,7 @@ fn opening_schema_84_with_uncertain_removal_preserves_old_server_recovery() {
drop(conn); drop(conn);
for attempt in 0..3 { for attempt in 0..3 {
let error = match attempt { 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(), 1 => SqliteWorkspaceStore::migration_plan(&path).unwrap_err(),
_ => SqliteWorkspaceStore::migrate_database(&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(); .unwrap();
drop(conn); drop(conn);
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
store.with_conn(|conn| { store.with_conn(|conn| {
assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION); 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"); 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()); assert!(!table_exists(&conn, "server_secret_objects").unwrap());
drop(conn); drop(conn);
assert_eq!(SqliteWorkspaceStore::migrate_database(&path).unwrap(), plan); 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| { store.with_conn(|conn| {
assert_eq!(current_schema_version(conn)?,LATEST_SCHEMA_VERSION); assert_eq!(current_schema_version(conn)?,LATEST_SCHEMA_VERSION);
let (nonce, mut sealed): (Vec<u8>,Vec<u8>) = conn.query_row( let (nonce, mut sealed): (Vec<u8>,Vec<u8>) = conn.query_row(
@@ -2290,7 +2290,7 @@ mod tests {
) { ) {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let database = dir.path().join("server.db"); 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(); let timestamp = now();
for workspace_id in ["workspace-a", "workspace-b"] { for workspace_id in ["workspace-a", "workspace-b"] {
futures::executor::block_on(store.upsert_workspace(&WorkspaceRecord { futures::executor::block_on(store.upsert_workspace(&WorkspaceRecord {
+1 -1
View File
@@ -1116,7 +1116,7 @@ mod tests {
WorkerId::from_legacy_u64(1) WorkerId::from_legacy_u64(1)
} }
fn setup() -> SqliteWorkspaceStore { fn setup() -> SqliteWorkspaceStore {
let s = SqliteWorkspaceStore::in_memory().unwrap(); let s = SqliteWorkspaceStore::in_memory_for_test().unwrap();
s.with_conn(|c| { s.with_conn(|c| {
c.execute( c.execute(
"INSERT INTO accounts(account_id,kind,handle,display_name,created_at,updated_at) \ "INSERT INTO accounts(account_id,kind,handle,display_name,created_at,updated_at) \
@@ -38,8 +38,8 @@ fn fresh_and_upgraded_state_contract_schemas_enforce_the_same_constraints() {
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("upgraded.db"); let path = temp.path().join("upgraded.db");
prepare_retained_schema(&path, 85); prepare_retained_schema(&path, 85);
let upgraded = SqliteWorkspaceStore::open(&path).unwrap(); let upgraded = SqliteWorkspaceStore::open_for_test(&path).unwrap();
let fresh = SqliteWorkspaceStore::in_memory().unwrap(); let fresh = SqliteWorkspaceStore::in_memory_for_test().unwrap();
let mut mismatches = Vec::new(); let mut mismatches = Vec::new();
for table in [ for table in [
"workspace_tags", "workspace_tags",
File diff suppressed because it is too large Load Diff
@@ -68,7 +68,9 @@ pub(super) async fn retire_legacy_worker(api: &WorkspaceApi, subject_id: &str) -
let job_api = api.clone(); let job_api = api.clone();
let subject_id = subject_id.to_owned(); let subject_id = subject_id.to_owned();
let key = format!("{SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX}{subject_id}"); let key = format!("{SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX}{subject_id}");
let (worker, complete) = host_io_executor()? let (worker, complete) = api
.host_executors
.host()?
.run(move || { .run(move || {
let api = &job_api; let api = &job_api;
let key = format!("{SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX}{subject_id}"); let key = format!("{SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX}{subject_id}");
@@ -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"); configure_runtime_request_auth(&mut api, &identity, "runtime-test");
let revoked = RuntimeIdentityMaterial::generate("revoked-runtime-secret").unwrap(); let revoked = RuntimeIdentityMaterial::generate("revoked-runtime-secret").unwrap();
configure_runtime_request_auth(&mut api, &revoked, "revoked-runtime-secret"); 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() .unwrap()
.revoke_workspace_runtime_binding( .revoke_workspace_runtime_binding(
TEST_WORKSPACE_ID, TEST_WORKSPACE_ID,
@@ -70,7 +70,7 @@ async fn rejected_runtime_proofs_log_reason_and_verification_without_secrets_on_
let body = "body-secret"; let body = "body-secret";
let apps = [ let apps = [
build_router(api.clone()), build_router(api.clone()),
workspace_server_router(WorkspaceServerApi::new( workspace_server_router(WorkspaceServerApi::new_for_test(
api.config.clone(), api.config.clone(),
api.store.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"); let path = format!("/api/w/{TEST_WORKSPACE_ID}/repositories");
for app in [ for app in [
build_router(api.clone()), build_router(api.clone()),
workspace_server_router(WorkspaceServerApi::new( workspace_server_router(WorkspaceServerApi::new_for_test(
api.config.clone(), api.config.clone(),
api.store.clone(), api.store.clone(),
)), )),
@@ -40,7 +40,7 @@ async fn cached_drive_handle_rejects_mutation_after_deletion_reservation() {
}) })
.await .await
.unwrap(); .unwrap();
let api = WorkspaceApi::new(config.clone(), store.clone()) let api = WorkspaceApi::new_for_test(config.clone(), store.clone())
.await .await
.unwrap(); .unwrap();
let cached = api.drive().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 // Use a separate embedded-runtime namespace for the first API: this test
// reconstructs Drive authority in-process rather than spawning/restarting // reconstructs Drive authority in-process rather than spawning/restarting
// product processes (runtime observers may retain their namespace). // 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 .await
.unwrap(); .unwrap();
// Merely constructing a Workspace API must not create blob storage. // 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()
.unwrap(); .unwrap();
drop(first_api); drop(first_api);
let recovered = WorkspaceServerApi::new(config.clone(), store.clone()); let recovered = WorkspaceServerApi::new_for_test(config.clone(), store.clone());
let api = recovered let api = recovered
.api_for_workspace(&config.workspace_id) .api_for_workspace(&config.workspace_id)
.await .await
@@ -39,10 +39,10 @@ impl RealDriveFixture {
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
let mut config = test_server_config(temp.path()); let mut config = test_server_config(temp.path());
config.repositories.clear(); 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 owner = seed_test_api_token(store.as_ref(), "drive-web-owner");
let second = seed_test_api_token(store.as_ref(), "drive-web-second"); 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; let server = LoopbackServer::start(workspace_server_router(server_api.clone())).await;
// Exercise T-717's real repositoryless Workspace bootstrap, not a // Exercise T-717's real repositoryless Workspace bootstrap, not a
// pre-existing test Workspace with its repository merely hidden in UI. // pre-existing test Workspace with its repository merely hidden in UI.
@@ -27,7 +27,7 @@ async fn subject_session_callback_completes_while_create_and_deletion_are_waitin
) )
.unwrap(); .unwrap();
let worker_id = reserved.worker_id.to_string(); 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. // Reuse the fixture API, but exercise the real outer dispatcher and both auth layers.
server server
.apis .apis
@@ -144,7 +144,7 @@ impl LaunchFixture {
async fn new(remote: bool) -> Self { async fn new(remote: bool) -> Self {
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
let config = test_server_config(temp.path()); 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 backend = Arc::new(LaunchExecutionBackend::default());
let api = let api =
WorkspaceApi::new_with_execution_backend(config, Arc::new(store), backend.clone()) WorkspaceApi::new_with_execution_backend(config, Arc::new(store), backend.clone())
@@ -151,7 +151,7 @@ impl Fixture {
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
let mut config = test_server_config(temp.path()); let mut config = test_server_config(temp.path());
config.repositories.clear(); 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( let mut api = WorkspaceApi::new_with_execution_backend(
config, config,
store, store,
@@ -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"); assert_eq!(api.runtime.worker(&identity).unwrap().state, "stopped");
// Retirement, unlike in-memory refresh, must survive caller journal restart. // Retirement, unlike in-memory refresh, must survive caller journal restart.
api.config_store = api.config_store = Arc::new(
Arc::new(SqliteWorkspaceStore::open(api.config.database_path.clone()).unwrap()); SqliteWorkspaceStore::open_for_test(api.config.database_path.clone()).unwrap(),
);
let fresh = api.runtime.worker(&identity).unwrap(); let fresh = api.runtime.worker(&identity).unwrap();
let next = api let next = api
.config_store .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. // Reopen the durable caller journal, discarding all client-side memory of request identity.
api.config_store = 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 let pinned = api
.config_store .config_store
.pin_internal_worker_restore_intent( .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); crate::repository_source::repository_source_fingerprint(&source);
config.repositories[0].source = source; config.repositories[0].source = source;
config.repositories[0].path = None; 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()); let execution = Arc::new(DeterministicExecutionBackend::default());
execution.accept_restores(); execution.accept_restores();
let api = WorkspaceApi::new_with_execution_backend(config, Arc::new(store), execution.clone()) let api = WorkspaceApi::new_with_execution_backend(config, Arc::new(store), execution.clone())
@@ -56,7 +56,7 @@ impl Fixture {
api.runtime.register_or_replace(runtime); api.runtime.register_or_replace(runtime);
let now = Arc::new(AtomicI64::new(START)); let now = Arc::new(AtomicI64::new(START));
let clock = now.clone(); 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.proof_clock = Arc::new(move || clock.load(Ordering::SeqCst));
server server
.apis .apis
@@ -356,7 +356,9 @@ impl WorkspaceWorker {
) -> ApiResult<InternalWorkerRestoreResult> { ) -> ApiResult<InternalWorkerRestoreResult> {
let guard = self.authorize_operation(context, "restore").await?; let guard = self.authorize_operation(context, "restore").await?;
let worker = self.clone(); let worker = self.clone();
host_io_executor()? self.api
.host_executors
.host()?
.run(move || { .run(move || {
let _guard = guard; let _guard = guard;
worker.restore_internal(request) worker.restore_internal(request)
@@ -381,7 +383,10 @@ impl WorkspaceWorker {
) -> ApiResult<WorkerLifecycleResult> { ) -> ApiResult<WorkerLifecycleResult> {
let guard = self.authorize_operation(context, "stop").await?; let guard = self.authorize_operation(context, "stop").await?;
let worker = self.clone(); let worker = self.clone();
let (result, guard) = host_io_executor()? let (result, guard) = self
.api
.host_executors
.host()?
.run(move || -> ApiResult<_> { .run(move || -> ApiResult<_> {
let result = worker let result = worker
.api .api
@@ -399,7 +404,9 @@ impl WorkspaceWorker {
let session_guard = session_lock.lock_owned().await; let session_guard = session_lock.lock_owned().await;
close_current_worker_session_locked(&self.api, &self.identity).await?; close_current_worker_session_locked(&self.api, &self.identity).await?;
let worker = self.clone(); let worker = self.clone();
host_io_executor()? self.api
.host_executors
.host()?
.run(move || -> ApiResult<_> { .run(move || -> ApiResult<_> {
let (_control, _session) = (guard, session_guard); let (_control, _session) = (guard, session_guard);
if let Some(record) = worker if let Some(record) = worker
@@ -425,7 +432,9 @@ impl WorkspaceWorker {
) -> ApiResult<WorkerLifecycleResult> { ) -> ApiResult<WorkerLifecycleResult> {
let guard = self.authorize_operation(context, "cancel").await?; let guard = self.authorize_operation(context, "cancel").await?;
let worker = self.clone(); let worker = self.clone();
host_io_executor()? self.api
.host_executors
.host()?
.run(move || -> ApiResult<_> { .run(move || -> ApiResult<_> {
let _guard = guard; let _guard = guard;
Ok(worker Ok(worker
@@ -457,7 +466,9 @@ impl WorkspaceWorker {
request.kind = WorkerInputKind::UserIfIdle; request.kind = WorkerInputKind::UserIfIdle;
} }
let worker = self.clone(); let worker = self.clone();
host_io_executor()? self.api
.host_executors
.host()?
.run(move || -> ApiResult<_> { .run(move || -> ApiResult<_> {
let (_control, _session) = (guard, session_guard); let (_control, _session) = (guard, session_guard);
Ok(worker Ok(worker
@@ -593,7 +593,7 @@ async fn real_attention_worker_start_restore(lose_response: bool) {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
init_clean_git_workspace(dir.path()); init_clean_git_workspace(dir.path());
let config = test_server_config(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) seed_test_registered_workspace(store.as_ref(), &config)
.await .await
.unwrap(); .unwrap();
@@ -700,7 +700,7 @@ async fn delivery_real_fixture(
DeliveryScriptedClient, DeliveryScriptedClient,
) { ) {
let config = test_server_config(root); 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) seed_test_registered_workspace(store.as_ref(), &config)
.await .await
.unwrap(); .unwrap();
@@ -924,7 +924,7 @@ async fn assignment_real_worker_response_loss_restore_and_concurrent_scans_commi
tokio::task::yield_now().await; tokio::task::yield_now().await;
let reopened = WorkspaceApi::new_with_execution_backend( let reopened = WorkspaceApi::new_with_execution_backend(
config.clone(), config.clone(),
Arc::new(SqliteWorkspaceStore::open(&config.database_path).unwrap()), Arc::new(SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap()),
execution.clone(), execution.clone(),
) )
.await .await
@@ -63,7 +63,7 @@ impl Fixture {
backend_base_url: Some("http://localhost:8787".into()), backend_base_url: Some("http://localhost:8787".into()),
}; };
let store = Arc::new( let store = Arc::new(
SqliteWorkspaceStore::open(&config.database_path) SqliteWorkspaceStore::open_for_test(&config.database_path)
.unwrap() .unwrap()
.with_db_executor(executor.clone()), .with_db_executor(executor.clone()),
); );
@@ -90,10 +90,10 @@ impl Fixture {
revoked_at: None, revoked_at: None,
}) })
.unwrap(); .unwrap();
let api = WorkspaceApi::new(config.clone(), store.clone()) let api = WorkspaceApi::new_for_test(config.clone(), store.clone())
.await .await
.unwrap(); .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. // Cache only the real inner router; avoid starting the unrelated hook.
server server
.apis .apis
@@ -452,7 +452,9 @@ async fn ticket_create_waiting_for_sqlite_does_not_block_timer_or_control() {
assert!(body["id"].is_string(), "{body}"); assert!(body["id"].is_string(), "{body}");
// Ticket dispatch can call Runtime callbacks, so its operation owner is // Ticket dispatch can call Runtime callbacks, so its operation owner is
// the separate bounded host executor, not the store executor. // 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(); 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(); 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"); 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" "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<bool>, 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();
}
@@ -724,7 +724,7 @@ async fn catalog_case_dispatch_authenticates_and_routes_the_canonical_workspace_
reason: ticket::TicketClosureReason::Completed, 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 // Retain only this in-process fixture: do not create a real Runtime or a
// background Orchestrator hook while testing catalog authentication/routing. // background Orchestrator hook while testing catalog authentication/routing.
catalog catalog
@@ -529,7 +529,7 @@ async fn runtime_tag_mutations_reject_invalid_proofs_and_changed_binding_authori
*tampered.body_mut() = Body::from(format!("{} ", input)); *tampered.body_mut() = Body::from(format!("{} ", input));
runtime_tag_response(app.clone(), tampered, StatusCode::UNAUTHORIZED).await; 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 let original = api
.store .store
.get_workspace_runtime_binding(TEST_WORKSPACE_ID, "runtime-tags") .get_workspace_runtime_binding(TEST_WORKSPACE_ID, "runtime-tags")
@@ -634,7 +634,7 @@ pub(super) async fn commit(
// a DB worker or an open SQL transaction. // a DB worker or an open SQL transaction.
let (complete, publication) = tokio::sync::oneshot::channel(); let (complete, publication) = tokio::sync::oneshot::channel();
let runtime = persistence_api.runtime.clone(); 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 || { executor.dispatch(move || {
let _guard = guard; let _guard = guard;
let _diagnostics = runtime let _diagnostics = runtime
@@ -681,7 +681,9 @@ pub(super) async fn create_grant(
let guard = lock(api, &worker).await?; let guard = lock(api, &worker).await?;
let runtime = api.runtime.clone(); let runtime = api.runtime.clone();
let lookup_worker = worker.clone(); let lookup_worker = worker.clone();
let (summary, guard) = host_io_executor() let (summary, guard) = api
.host_executors
.host()
.map_err(pre_error)? .map_err(pre_error)?
.run(move || { .run(move || {
Ok::<_, Error>((|| { Ok::<_, Error>((|| {
@@ -318,7 +318,7 @@ mod workspace_config_integration {
}; };
for scoped in [false, true] { for scoped in [false, true] {
let app = if scoped { let app = if scoped {
workspace_server_router(WorkspaceServerApi::new( workspace_server_router(WorkspaceServerApi::new_for_test(
api.config.clone(), api.config.clone(),
api.store.clone(), api.store.clone(),
)) ))
@@ -1385,7 +1385,7 @@ mod workspace_config_integration {
let api = &fixture.api; let api = &fixture.api;
let granted = grant(api, &fixture.worker, Access::ReadWrite).await; let granted = grant(api, &fixture.worker, Access::ReadWrite).await;
let attached = attach(api, &fixture.worker).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!( assert_eq!(
restarted restarted
.get_workspace_config_grant(TEST_WORKSPACE_ID, &granted.grant_id) .get_workspace_config_grant(TEST_WORKSPACE_ID, &granted.grant_id)
@@ -800,7 +800,7 @@ async fn workspace_config_actual_remote_and_embedded_clients_use_authenticated_w
let base = format!("http://{addr}"); let base = format!("http://{addr}");
let mut config = test_server_config(root.path()); let mut config = test_server_config(root.path());
config.backend_base_url = Some("server-test".into()); 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 captured = Arc::new(Mutex::new(None));
let abort = Arc::new(Mutex::new(None)); let abort = Arc::new(Mutex::new(None));
let factory = ConfigCaptureFactory { let factory = ConfigCaptureFactory {
+111 -81
View File
@@ -2278,13 +2278,40 @@ impl SqliteWorkspaceStore {
} }
pub fn from_connection(conn: Connection) -> Result<Self> { pub fn from_connection(conn: Connection) -> Result<Self> {
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<Self> {
prepare_connection(&conn)?; prepare_connection(&conn)?;
Ok(Self { Ok(Self {
conn: Arc::new(Mutex::new(conn)), 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<Path>) -> Result<Self> {
Self::from_connection_for_test(Connection::open(path)?)
}
#[cfg(test)]
pub(crate) fn in_memory_for_test() -> Result<Self> {
Self::from_connection_for_test(Connection::open_in_memory()?)
}
#[cfg(test)]
pub(crate) fn from_connection_for_test(conn: Connection) -> Result<Self> {
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. /// Override execution limits for a hermetic fixture; subsequent clones share it.
pub fn with_db_executor(mut self, executor: crate::db_executor::DbExecutor) -> Self { pub fn with_db_executor(mut self, executor: crate::db_executor::DbExecutor) -> Self {
self.db_executor = executor; self.db_executor = executor;
@@ -14861,7 +14888,7 @@ mod tests {
fn drive_grants_migrate_and_preserve_audit_through_revoke_worker_delete_and_restart() { fn drive_grants_migrate_and_preserve_audit_through_revoke_worker_delete_and_restart() {
let t = tempfile::tempdir().unwrap(); let t = tempfile::tempdir().unwrap();
let path = t.path().join("server.db"); 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(); let legacy = Connection::open_in_memory().unwrap();
configure_sqlite(&legacy).unwrap(); configure_sqlite(&legacy).unwrap();
legacy.execute_batch("CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY); legacy.execute_batch("CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY);
@@ -14951,7 +14978,7 @@ mod tests {
Some(other) Some(other)
); );
drop(store); drop(store);
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
assert_eq!( assert_eq!(
store store
.get_workspace_drive_grant("drive-ws", &first.grant_id) .get_workspace_drive_grant("drive-ws", &first.grant_id)
@@ -14986,7 +15013,7 @@ mod tests {
} }
#[test] #[test]
fn drive_grants_baseline_constraints_and_worker_lifetime_fail_closed() { 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 proposed = drive_grant_fixture(&store);
let first = store.create_workspace_drive_grant(&proposed).unwrap(); let first = store.create_workspace_drive_grant(&proposed).unwrap();
store.with_conn(|c| { store.with_conn(|c| {
@@ -15022,7 +15049,7 @@ mod tests {
fn schema_v74_migrates_historical_worker_projection_rows_without_counter_state() { fn schema_v74_migrates_historical_worker_projection_rows_without_counter_state() {
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("workspace.db"); 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(); let conn = Connection::open(&path).unwrap();
configure_sqlite(&conn).unwrap(); configure_sqlite(&conn).unwrap();
conn.execute_batch( conn.execute_batch(
@@ -15347,7 +15374,7 @@ mod tests {
#[tokio::test] #[tokio::test]
async fn external_workdir_grant_persists_typed_source_and_fences_generations() { 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 store
.upsert_account(&AccountRecord { .upsert_account(&AccountRecord {
account_id: "owner".to_string(), account_id: "owner".to_string(),
@@ -15719,7 +15746,7 @@ mod tests {
fn runtime_removal_test_store() -> (tempfile::TempDir, SqliteWorkspaceStore) { fn runtime_removal_test_store() -> (tempfile::TempDir, SqliteWorkspaceStore) {
let temp = tempfile::tempdir().unwrap(); 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 store
.with_conn(|conn| { .with_conn(|conn| {
conn.execute_batch( conn.execute_batch(
@@ -16103,7 +16130,7 @@ mod tests {
let path = temp.path().join("server.db"); let path = temp.path().join("server.db");
prepare_retained_schema(&path, 59); prepare_retained_schema(&path, 59);
let migrated = SqliteWorkspaceStore::open(&path).unwrap(); let migrated = SqliteWorkspaceStore::open_for_test(&path).unwrap();
migrated migrated
.with_conn(|conn| { .with_conn(|conn| {
let table_count: i64 = conn.query_row( let table_count: i64 = conn.query_row(
@@ -16139,7 +16166,7 @@ mod tests {
let path = temp.path().join("server.db"); let path = temp.path().join("server.db");
prepare_retained_schema(&path, 60); prepare_retained_schema(&path, 60);
drop(SqliteWorkspaceStore::open(&path).unwrap()); drop(SqliteWorkspaceStore::open_for_test(&path).unwrap());
let conn = Connection::open(&path).unwrap(); let conn = Connection::open(&path).unwrap();
assert_eq!( assert_eq!(
current_schema_version(&conn).unwrap(), current_schema_version(&conn).unwrap(),
@@ -16169,7 +16196,7 @@ mod tests {
#[test] #[test]
fn schema_v59_runtime_removal_migration_rolls_back_partial_ddl_and_marker() { fn schema_v59_runtime_removal_migration_rolls_back_partial_ddl_and_marker() {
let temp = tempfile::tempdir().unwrap(); 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 store
.with_conn(|conn| { .with_conn(|conn| {
let trigger_names = { let trigger_names = {
@@ -16433,7 +16460,7 @@ mod tests {
fn fresh_database_uses_only_the_canonical_workspace_baseline() { fn fresh_database_uses_only_the_canonical_workspace_baseline() {
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("server.db"); let path = temp.path().join("server.db");
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
store store
.with_conn(|conn| { .with_conn(|conn| {
let migrations = conn let migrations = conn
@@ -16581,7 +16608,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
.collect::<Vec<_>>(), .collect::<Vec<_>>(),
(83..=LATEST_SCHEMA_VERSION).collect::<Vec<_>>() (83..=LATEST_SCHEMA_VERSION).collect::<Vec<_>>()
); );
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
let assignment = store let assignment = store
.get_active_ticket_worker_assignment("workspace-a", "ticket") .get_active_ticket_worker_assignment("workspace-a", "ticket")
.unwrap() .unwrap()
@@ -17052,7 +17079,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
let applied = SqliteWorkspaceStore::migrate_database(&path).unwrap(); let applied = SqliteWorkspaceStore::migrate_database(&path).unwrap();
assert_eq!(applied, plan); assert_eq!(applied, plan);
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
store store
.with_conn(|conn| { .with_conn(|conn| {
assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION); 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" "migrated databases must reject incomplete singleton leases"
); );
drop(conn); 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 legacy = RuntimeWorkerRef::new("legacy-runtime", "legacy-orchestrator");
let lease = store let lease = store
.require_current_worker_singleton_owner("workspace-a", &legacy) .require_current_worker_singleton_owner("workspace-a", &legacy)
@@ -17789,7 +17816,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
.unwrap(); .unwrap();
drop(conn); drop(conn);
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
store store
.with_conn(|conn| { .with_conn(|conn| {
assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION); assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION);
@@ -17890,7 +17917,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
.unwrap(); .unwrap();
drop(conn); drop(conn);
let error = SqliteWorkspaceStore::open(&path) let error = SqliteWorkspaceStore::open_for_test(&path)
.err() .err()
.expect("migration must fail") .expect("migration must fail")
.to_string(); .to_string();
@@ -17915,7 +17942,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
fn runtime_binding_identity_and_trust_uniqueness_are_workspace_scoped() { fn runtime_binding_identity_and_trust_uniqueness_are_workspace_scoped() {
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("server.db"); let path = temp.path().join("server.db");
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
store store
.with_conn(|conn| { .with_conn(|conn| {
conn.execute_batch( conn.execute_batch(
@@ -18017,7 +18044,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
); );
drop(store); drop(store);
let reopened = SqliteWorkspaceStore::open(&path).unwrap(); let reopened = SqliteWorkspaceStore::open_for_test(&path).unwrap();
assert!( assert!(
reopened reopened
.list_workspace_runtime_bindings("workspace-a", false) .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() { fn workspace_runtime_connection_persists_without_authentication_evidence() {
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("server.db"); let path = temp.path().join("server.db");
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
store store
.with_conn(|conn| { .with_conn(|conn| {
conn.execute_batch( conn.execute_batch(
@@ -18112,7 +18139,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
.unwrap(); .unwrap();
assert!(store.workspace_runtime_binding_matches(&persisted).unwrap()); assert!(store.workspace_runtime_binding_matches(&persisted).unwrap());
drop(store); drop(store);
let reopened = SqliteWorkspaceStore::open(&path).unwrap(); let reopened = SqliteWorkspaceStore::open_for_test(&path).unwrap();
assert_eq!( assert_eq!(
reopened reopened
.get_workspace_runtime_binding("workspace-a", "runtime-a") .get_workspace_runtime_binding("workspace-a", "runtime-a")
@@ -18366,7 +18393,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
#[test] #[test]
fn runtime_binding_key_mutations_are_identity_guarded_idempotent_and_audited() { 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 store
.with_conn(|conn| { .with_conn(|conn| {
conn.execute_batch( conn.execute_batch(
@@ -18504,7 +18531,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
#[test] #[test]
fn embedded_runtime_binding_can_explicitly_rotate_restart_identity() { fn embedded_runtime_binding_can_explicitly_rotate_restart_identity() {
let store = SqliteWorkspaceStore::in_memory().unwrap(); let store = SqliteWorkspaceStore::in_memory_for_test().unwrap();
store store
.with_conn(|conn| { .with_conn(|conn| {
conn.execute_batch( conn.execute_batch(
@@ -18604,7 +18631,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
conn.execute_batch("DROP TABLE typed_ticket_artifacts") conn.execute_batch("DROP TABLE typed_ticket_artifacts")
.unwrap(); .unwrap();
let result = SqliteWorkspaceStore::from_connection(conn); let result = SqliteWorkspaceStore::from_connection_for_test(conn);
let error = match result { let error = match result {
Ok(_) => panic!("schema drift unexpectedly passed startup verification"), Ok(_) => panic!("schema drift unexpectedly passed startup verification"),
Err(error) => error, 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() { async fn workspace_schema_rejects_cross_workspace_ticket_repository_reference_at_write_time() {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let database_path = dir.path().join("workspace.sqlite"); 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"] { for workspace_id in ["workspace-a", "workspace-b"] {
store store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
@@ -18670,14 +18697,14 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
); );
drop(backend); drop(backend);
SqliteWorkspaceStore::open(&database_path).unwrap(); SqliteWorkspaceStore::open_for_test(&database_path).unwrap();
} }
#[tokio::test] #[tokio::test]
async fn migrates_sqlite_and_preserves_workspace_record() { async fn migrates_sqlite_and_preserves_workspace_record() {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let db = dir.path().join("control-plane.sqlite"); 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_eq!(store.schema_version().await.unwrap(), LATEST_SCHEMA_VERSION);
assert!( assert!(
@@ -18695,7 +18722,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
}; };
store.upsert_workspace(&record).await.unwrap(); store.upsert_workspace(&record).await.unwrap();
let reopened = SqliteWorkspaceStore::open(&db).unwrap(); let reopened = SqliteWorkspaceStore::open_for_test(&db).unwrap();
assert_eq!( assert_eq!(
reopened.schema_version().await.unwrap(), reopened.schema_version().await.unwrap(),
LATEST_SCHEMA_VERSION 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() { async fn workspace_display_name_update_is_timestamp_guarded_and_preserves_identity() {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let database_path = dir.path().join("server.db"); 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 { let record = WorkspaceRecord {
workspace_id: "workspace-a".to_string(), workspace_id: "workspace-a".to_string(),
owner_account_id: "owner-account".to_string(), owner_account_id: "owner-account".to_string(),
@@ -18749,7 +18776,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
updated updated
); );
drop(store); drop(store);
let reopened = SqliteWorkspaceStore::open(database_path).unwrap(); let reopened = SqliteWorkspaceStore::open_for_test(database_path).unwrap();
assert_eq!( assert_eq!(
reopened reopened
.get_workspace(&record.workspace_id) .get_workspace(&record.workspace_id)
@@ -18763,7 +18790,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
#[tokio::test] #[tokio::test]
async fn objective_creation_allocates_and_resolves_workspace_resource_key() { async fn objective_creation_allocates_and_resolves_workspace_resource_key() {
let dir = tempfile::tempdir().unwrap(); 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 store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "workspace-a".into(), workspace_id: "workspace-a".into(),
@@ -18809,7 +18836,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
#[tokio::test] #[tokio::test]
async fn worker_create_reservation_allocates_uuid_before_runtime_and_replays_exact_input() { async fn worker_create_reservation_allocates_uuid_before_runtime_and_replays_exact_input() {
let dir = tempfile::tempdir().unwrap(); 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 store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "workspace-a".to_string(), 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() { async fn worker_singleton_restart_retry_replays_exact_workdir_intent_and_rejects_changes() {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let database = dir.path().join("server.db"); let database = dir.path().join("server.db");
let store = SqliteWorkspaceStore::open(&database).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&database).unwrap();
store store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "workspace-a".to_string(), workspace_id: "workspace-a".to_string(),
@@ -19173,7 +19200,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
.unwrap(); .unwrap();
drop(store); drop(store);
let reopened = SqliteWorkspaceStore::open(&database).unwrap(); let reopened = SqliteWorkspaceStore::open_for_test(&database).unwrap();
let retried = reopened let retried = reopened
.reserve_worker_create( .reserve_worker_create(
"workspace-a", "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() { async fn worker_singleton_ownership_is_workspace_scoped_durable_and_fences_old_generations() {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let database = dir.path().join("server.db"); 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"] { for workspace_id in ["workspace-a", "workspace-b"] {
store store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
@@ -19330,7 +19357,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
); );
drop(store); drop(store);
let reopened = SqliteWorkspaceStore::open(database).unwrap(); let reopened = SqliteWorkspaceStore::open_for_test(database).unwrap();
assert_eq!( assert_eq!(
reopened reopened
.current_worker_singleton_owner("workspace-a", key) .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() { async fn concurrent_worker_singleton_reservations_have_one_winner() {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let database = dir.path().join("server.db"); let database = dir.path().join("server.db");
let store = SqliteWorkspaceStore::open(&database).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&database).unwrap();
store store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "workspace-a".to_string(), workspace_id: "workspace-a".to_string(),
@@ -19363,7 +19390,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
let database = database.clone(); let database = database.clone();
let barrier = barrier.clone(); let barrier = barrier.clone();
std::thread::spawn(move || { 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(); let memory = store.get_workspace_memory_settings("workspace-a").unwrap();
barrier.wait(); barrier.wait();
store.reserve_worker_create( store.reserve_worker_create(
@@ -19383,7 +19410,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
.collect::<Vec<_>>(); .collect::<Vec<_>>();
assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1); assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1);
assert_eq!(results.iter().filter(|result| result.is_err()).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 let owner = store
.current_worker_singleton_owner("workspace-a", "shared-key") .current_worker_singleton_owner("workspace-a", "shared-key")
.unwrap() .unwrap()
@@ -19399,7 +19426,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
#[tokio::test] #[tokio::test]
async fn workspace_flow_sources_pin_content_and_builtins_stay_resources() { async fn workspace_flow_sources_pin_content_and_builtins_stay_resources() {
let dir = tempfile::tempdir().unwrap(); 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 store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "workspace-a".to_string(), 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" "returning to earlier content must reuse the immutable content"
); );
drop(store); 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!( assert_eq!(
reopened reopened
.get_flow_source_content( .get_flow_source_content(
@@ -19748,7 +19775,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
} }
); );
drop(conn); 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(); let backend = ticket::SqliteTicketBackend::open_verified(&path, "case-workspace").unwrap();
store store
.with_conn(|conn| { .with_conn(|conn| {
@@ -19915,7 +19942,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
); );
drop(store); drop(store);
drop(backend); drop(backend);
let reopened = SqliteWorkspaceStore::open(&path).unwrap(); let reopened = SqliteWorkspaceStore::open_for_test(&path).unwrap();
reopened reopened
.with_conn(|conn| { .with_conn(|conn| {
assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION); assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION);
@@ -19982,7 +20009,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
before before
); );
drop(conn); 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(); let backend = ticket::SqliteTicketBackend::open_verified(&path, "case-workspace").unwrap();
assert_eq!( assert_eq!(
backend backend
@@ -20116,8 +20143,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("equivalence.db"); let path = dir.path().join("equivalence.db");
prepare_retained_schema(&path, 90); prepare_retained_schema(&path, 90);
let upgraded = SqliteWorkspaceStore::open(&path).unwrap(); let upgraded = SqliteWorkspaceStore::open_for_test(&path).unwrap();
let fresh = SqliteWorkspaceStore::in_memory().unwrap(); let fresh = SqliteWorkspaceStore::in_memory_for_test().unwrap();
for object in [ for object in [
"workspace_tags", "workspace_tags",
"typed_ticket_tags", "typed_ticket_tags",
@@ -20170,7 +20197,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
#[test] #[test]
fn schema_v91_case_constraint_verification_rejects_legacy_view_predicates() { 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| { store.with_conn(|conn| {
conn.execute_batch("DROP VIEW ticket_active_worker_assignments; conn.execute_batch("DROP VIEW ticket_active_worker_assignments;
CREATE VIEW ticket_active_worker_assignments AS 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. // boundary under test. Targets need not be materialized by a Runtime.
async fn manual_worker_start_store() -> (tempfile::TempDir, SqliteWorkspaceStore) { async fn manual_worker_start_store() -> (tempfile::TempDir, SqliteWorkspaceStore) {
let dir = tempfile::tempdir().unwrap(); 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 store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "w".into(), workspace_id: "w".into(),
@@ -20419,7 +20446,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
let threads = (0..2) let threads = (0..2)
.map(|index| { .map(|index| {
let store = let store =
SqliteWorkspaceStore::open(dir.path().join("workspace.db")).unwrap(); SqliteWorkspaceStore::open_for_test(dir.path().join("workspace.db"))
.unwrap();
let barrier = barrier.clone(); let barrier = barrier.clone();
std::thread::spawn(move || { std::thread::spawn(move || {
barrier.wait(); barrier.wait();
@@ -20784,7 +20812,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
.retain_ticket_claim_recovery("w", "claim", &fingerprint, &changed) .retain_ticket_claim_recovery("w", "claim", &fingerprint, &changed)
.is_err() .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!( assert_eq!(
reopened.get_ticket_claim_recovery("w", "claim").unwrap(), reopened.get_ticket_claim_recovery("w", "claim").unwrap(),
Some(recovery) Some(recovery)
@@ -20842,7 +20871,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
store store
.rollback_ticket_assignment_operation("w", "claim") .rollback_ticket_assignment_operation("w", "claim")
.unwrap(); .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 let receipt = reopened
.get_ticket_assignment_operation("w", "claim") .get_ticket_assignment_operation("w", "claim")
.unwrap() .unwrap()
@@ -21062,7 +21092,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
use ticket::TicketBackend; use ticket::TicketBackend;
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("workspace.db"); let path = dir.path().join("workspace.db");
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
store store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "w".into(), 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() { async fn ticket_worker_assignment_replaces_current_and_preserves_audit_history() {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let db = dir.path().join("server.db"); let db = dir.path().join("server.db");
let store = SqliteWorkspaceStore::open(&db).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&db).unwrap();
store store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "workspace-a".to_string(), workspace_id: "workspace-a".to_string(),
@@ -21379,7 +21409,7 @@ INSERT INTO worker_registry (
) )
.unwrap(); .unwrap();
drop(store); drop(store);
let store = SqliteWorkspaceStore::open(&db).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&db).unwrap();
let pending = store let pending = store
.get_ticket_assignment_operation("workspace-a", "reserved-operation") .get_ticket_assignment_operation("workspace-a", "reserved-operation")
.unwrap() .unwrap()
@@ -21503,7 +21533,7 @@ INSERT INTO worker_registry (
async fn role_assignment_routes_and_manual_start_are_state_fenced_atomically() { async fn role_assignment_routes_and_manual_start_are_state_fenced_atomically() {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("workspace.db"); 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 store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "workspace-role".to_string(), workspace_id: "workspace-role".to_string(),
@@ -21792,7 +21822,7 @@ INSERT INTO worker_registry (
fn startup_rejects_missing_workspace_resource_constraint_trigger() { fn startup_rejects_missing_workspace_resource_constraint_trigger() {
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("server.db"); let path = temp.path().join("server.db");
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
store store
.with_conn(|conn| { .with_conn(|conn| {
conn.execute_batch("DROP TRIGGER ticket_worker_assignments_validate_insert")?; conn.execute_batch("DROP TRIGGER ticket_worker_assignments_validate_insert")?;
@@ -21801,7 +21831,7 @@ INSERT INTO worker_registry (
.unwrap(); .unwrap();
drop(store); 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"), Ok(_) => panic!("missing assignment constraint trigger must fail closed"),
Err(error) => error, Err(error) => error,
}; };
@@ -21817,7 +21847,7 @@ INSERT INTO worker_registry (
fn startup_accepts_retained_assignment_history_after_parent_cleanup() { fn startup_accepts_retained_assignment_history_after_parent_cleanup() {
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("server.db"); let path = temp.path().join("server.db");
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
store store
.with_conn(|conn| { .with_conn(|conn| {
conn.execute_batch( conn.execute_batch(
@@ -21881,7 +21911,7 @@ INSERT INTO worker_registry (
.unwrap(); .unwrap();
drop(store); drop(store);
let reopened = SqliteWorkspaceStore::open(&path).unwrap(); let reopened = SqliteWorkspaceStore::open_for_test(&path).unwrap();
reopened reopened
.with_conn(|conn| { .with_conn(|conn| {
let retained: i64 = conn.query_row( let retained: i64 = conn.query_row(
@@ -21909,7 +21939,7 @@ INSERT INTO worker_registry (
#[tokio::test] #[tokio::test]
async fn workspace_bootstrap_validates_key_and_rejects_duplicate_workspace() { 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 store
.upsert_account(&AccountRecord { .upsert_account(&AccountRecord {
account_id: "owner-account".to_string(), account_id: "owner-account".to_string(),
@@ -22041,7 +22071,7 @@ INSERT INTO worker_registry (
#[tokio::test] #[tokio::test]
async fn repository_records_round_trip() { 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); assert_eq!(store.schema_version().await.unwrap(), LATEST_SCHEMA_VERSION);
let workspace = WorkspaceRecord { let workspace = WorkspaceRecord {
workspace_id: "local-dev".to_string(), workspace_id: "local-dev".to_string(),
@@ -22122,7 +22152,7 @@ INSERT INTO worker_registry (
#[tokio::test] #[tokio::test]
async fn memory_authority_records_round_trip_and_close_staging() { 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); assert_eq!(store.schema_version().await.unwrap(), LATEST_SCHEMA_VERSION);
let workspace = WorkspaceRecord { let workspace = WorkspaceRecord {
workspace_id: "local-dev".to_string(), workspace_id: "local-dev".to_string(),
@@ -22197,7 +22227,7 @@ INSERT INTO worker_registry (
#[tokio::test] #[tokio::test]
async fn legacy_memory_reset_is_atomic_scoped_idempotent_and_exact() { 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 [ for (workspace_id, display_name) in [
("local-dev", "Local Dev"), ("local-dev", "Local Dev"),
("other-workspace", "Other Workspace"), ("other-workspace", "Other Workspace"),
@@ -22364,7 +22394,7 @@ INSERT INTO worker_registry (
async fn worker_workdir_registry_round_trips_and_preserves_pinned_retention() { async fn worker_workdir_registry_round_trips_and_preserves_pinned_retention() {
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
let db = temp.path().join("workspace.db"); let db = temp.path().join("workspace.db");
let store = SqliteWorkspaceStore::open(&db).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&db).unwrap();
let workspace = WorkspaceRecord { let workspace = WorkspaceRecord {
workspace_id: "local-dev".to_string(), workspace_id: "local-dev".to_string(),
owner_account_id: "owner-account".to_string(), owner_account_id: "owner-account".to_string(),
@@ -22733,7 +22763,7 @@ INSERT INTO worker_registry (
); );
} }
drop(store); drop(store);
let store = SqliteWorkspaceStore::open(&db).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&db).unwrap();
assert_eq!( assert_eq!(
store store
.list_worker_workdir_links_page("local-dev", &worker.worker, 1, 0, None) .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(); let workdir_conflict = store.attach_worker_workdir(&workdir_conflict).unwrap();
drop(store); drop(store);
let reopened = SqliteWorkspaceStore::open(&db).unwrap(); let reopened = SqliteWorkspaceStore::open_for_test(&db).unwrap();
assert_eq!( assert_eq!(
reopened reopened
.list_worker_workdir_links("local-dev", &worker.worker) .list_worker_workdir_links("local-dev", &worker.worker)
@@ -22781,7 +22811,7 @@ INSERT INTO worker_registry (
#[tokio::test] #[tokio::test]
async fn worker_projection_preserves_equal_local_ids_from_distinct_runtimes() { async fn worker_projection_preserves_equal_local_ids_from_distinct_runtimes() {
let temp = tempfile::tempdir().unwrap(); 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 store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "local-dev".to_string(), workspace_id: "local-dev".to_string(),
@@ -22907,7 +22937,7 @@ INSERT INTO worker_registry (
#[tokio::test] #[tokio::test]
async fn worker_projection_reconciles_snapshots_and_preserves_unavailable_lifecycle() { async fn worker_projection_reconciles_snapshots_and_preserves_unavailable_lifecycle() {
let temp = tempfile::tempdir().unwrap(); 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 store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "local-dev".to_string(), workspace_id: "local-dev".to_string(),
@@ -23082,7 +23112,7 @@ INSERT INTO worker_registry (
#[tokio::test] #[tokio::test]
async fn late_runtime_event_does_not_recreate_retention_deleted_catalog_worker() { async fn late_runtime_event_does_not_recreate_retention_deleted_catalog_worker() {
let temp = tempfile::tempdir().unwrap(); 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 store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "local-dev".to_string(), 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() { async fn worker_control_grants_are_idempotent_scoped_and_support_internal_invalidation() {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let database = dir.path().join("control-grants.db"); let database = dir.path().join("control-grants.db");
let store = SqliteWorkspaceStore::open(&database).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&database).unwrap();
store store
.upsert_workspace(&WorkspaceRecord { .upsert_workspace(&WorkspaceRecord {
workspace_id: "workspace-control".to_string(), workspace_id: "workspace-control".to_string(),
@@ -23247,7 +23277,7 @@ INSERT INTO worker_registry (
); );
drop(store); drop(store);
let store = SqliteWorkspaceStore::open(&database).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&database).unwrap();
assert_eq!( assert_eq!(
store store
.list_active_worker_control_grants( .list_active_worker_control_grants(
@@ -23305,7 +23335,7 @@ INSERT INTO worker_registry (
#[tokio::test] #[tokio::test]
async fn account_and_login_records_round_trip() { 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); assert_eq!(store.schema_version().await.unwrap(), LATEST_SCHEMA_VERSION);
let now = "2026-07-22T00:00:00Z".to_string(); let now = "2026-07-22T00:00:00Z".to_string();
let account = AccountRecord { let account = AccountRecord {
@@ -23664,9 +23694,9 @@ INSERT INTO worker_registry (
async fn backend_job_resource_reserve_converges_across_connections_without_changing_input() { async fn backend_job_resource_reserve_converges_across_connections_without_changing_input() {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("resource-jobs.sqlite"); 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; seed_resource_job_workspace(&store).await;
let other = SqliteWorkspaceStore::open(&path).unwrap(); let other = SqliteWorkspaceStore::open_for_test(&path).unwrap();
let requests = [ let requests = [
resource_job_request("first", "subject-a", "candidate-1"), resource_job_request("first", "subject-a", "candidate-1"),
resource_job_request("followup", "subject-a", "candidate-2"), 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() { async fn backend_job_resource_retry_and_unknown_are_fenced_by_durable_cleanup() {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("cleanup-jobs.sqlite"); 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; seed_resource_job_workspace(&store).await;
let request = resource_job_request("failed", "subject-a", "candidate-1"); let request = resource_job_request("failed", "subject-a", "candidate-1");
let first = store let first = store
@@ -23888,7 +23918,7 @@ INSERT INTO worker_registry (
.unwrap(); .unwrap();
finish_resource_job_cleanup(&store, &second); finish_resource_job_cleanup(&store, &second);
drop(store); drop(store);
let restarted = SqliteWorkspaceStore::open(&path).unwrap(); let restarted = SqliteWorkspaceStore::open_for_test(&path).unwrap();
let active = restarted let active = restarted
.find_active_backend_job_for_resource( .find_active_backend_job_for_resource(
"resource-workspace", "resource-workspace",
@@ -23948,7 +23978,7 @@ INSERT INTO worker_registry (
async fn backend_job_resource_retry_races_new_intent_atomically_and_remains_bounded() { async fn backend_job_resource_retry_races_new_intent_atomically_and_remains_bounded() {
let dir = tempfile::tempdir().unwrap(); let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("retry-race.sqlite"); 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; seed_resource_job_workspace(&store).await;
// Generic resources use the same common reservation API without domain grants. // Generic resources use the same common reservation API without domain grants.
let mut request = resource_job_request("generic", "subject-a", "candidate-1"); let mut request = resource_job_request("generic", "subject-a", "candidate-1");
@@ -23971,7 +24001,7 @@ INSERT INTO worker_registry (
let mut followup = request.clone(); let mut followup = request.clone();
followup.job_id = "generic-followup".into(); followup.job_id = "generic-followup".into();
followup.input = serde_json::json!({"candidate": "new"}); 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 barrier = std::sync::Barrier::new(2);
let (retry, reserve) = std::thread::scope(|scope| { let (retry, reserve) = std::thread::scope(|scope| {
let a = scope.spawn(|| { let a = scope.spawn(|| {
@@ -24057,7 +24087,7 @@ INSERT INTO worker_registry (
#[tokio::test] #[tokio::test]
async fn backend_job_all_older_attempt_cleanups_fence_retry_and_resource_release() { 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; seed_resource_job_workspace(&store).await;
let mut request = resource_job_request("legacy-overlap", "subject-a", "candidate-1"); let mut request = resource_job_request("legacy-overlap", "subject-a", "candidate-1");
request.limits.max_attempts = 3; request.limits.max_attempts = 3;
@@ -24181,7 +24211,7 @@ INSERT INTO worker_registry (
#[tokio::test] #[tokio::test]
async fn backend_job_empty_batch_surface_refresh_locks_subject_and_defers_candidates() { 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; seed_resource_job_workspace(&store).await;
for surface_state in ["ungenerated", "stale", "failed"] { for surface_state in ["ungenerated", "stale", "failed"] {
let subject = format!("subject-{surface_state}"); let subject = format!("subject-{surface_state}");
@@ -24301,7 +24331,7 @@ INSERT INTO worker_registry (
#[tokio::test] #[tokio::test]
async fn backend_job_completed_resource_releases_only_after_worker_cleanup() { 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; seed_resource_job_workspace(&store).await;
let request = resource_job_request("completed", "subject-a", "candidate-1"); let request = resource_job_request("completed", "subject-a", "candidate-1");
let first = store let first = store
@@ -24355,7 +24385,7 @@ INSERT INTO worker_registry (
#[tokio::test] #[tokio::test]
async fn backend_jobs_fence_results_project_ownership_and_bound_retries() { 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"; let workspace_id = "workspace-job";
store store
.upsert_account(&AccountRecord { .upsert_account(&AccountRecord {
@@ -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 tags_before = retained.as_ref().map(|_| classification_rows(&conn));
let auth_before = fixture_auth_rows(&conn); let auth_before = fixture_auth_rows(&conn);
drop(conn); drop(conn);
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
store.with_conn(|conn| { store.with_conn(|conn| {
assert_eq!(current_schema_version(conn)?,93); 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); 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 metadata_before = stored_fixture_rows(&conn, "typed_ticket_case_metadata");
let auth_before = fixture_auth_rows(&conn); let auth_before = fixture_auth_rows(&conn);
drop(conn); drop(conn);
let store = SqliteWorkspaceStore::open(&path).unwrap(); let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
store store
.with_conn(|conn| { .with_conn(|conn| {
assert_eq!(current_schema_version(conn)?,93); assert_eq!(current_schema_version(conn)?,93);
@@ -750,7 +750,7 @@ mod tests {
#[test] #[test]
fn retry_keeps_resolved_config_and_credential_candidates_after_fallback_moves() { 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 { futures::executor::block_on(store.upsert_workspace(&WorkspaceRecord {
workspace_id: "workspace".to_string(), workspace_id: "workspace".to_string(),
owner_account_id: "owner-account".to_string(), owner_account_id: "owner-account".to_string(),
@@ -925,7 +925,7 @@ mod tests {
} }
async fn seeded_store() -> (SqliteWorkspaceStore, WorkdirRegistryRecord) { async fn seeded_store() -> (SqliteWorkspaceStore, WorkdirRegistryRecord) {
let store = SqliteWorkspaceStore::in_memory().unwrap(); let store = SqliteWorkspaceStore::in_memory_for_test().unwrap();
store store
.upsert_account(&AccountRecord { .upsert_account(&AccountRecord {
account_id: "account-a".to_string(), account_id: "account-a".to_string(),
@@ -463,7 +463,7 @@ mod tests {
) { ) {
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
let store = Arc::new( let store = Arc::new(
SqliteWorkspaceStore::open(temp.path().join("workspace.db")) SqliteWorkspaceStore::open_for_test(temp.path().join("workspace.db"))
.unwrap() .unwrap()
.with_db_executor(crate::db_executor::DbExecutor::new(Default::default()).unwrap()), .with_db_executor(crate::db_executor::DbExecutor::new(Default::default()).unwrap()),
); );
@@ -392,7 +392,7 @@ mod tests {
}) })
.unwrap(); .unwrap();
let store = Arc::new( let store = Arc::new(
SqliteWorkspaceStore::open(&db_path) SqliteWorkspaceStore::open_for_test(&db_path)
.unwrap() .unwrap()
.with_db_executor(executor), .with_db_executor(executor),
); );
@@ -285,7 +285,7 @@ mod tests {
#[tokio::test] #[tokio::test]
async fn create_is_atomic_and_exact_retries_converge() { 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( let service = WorkspaceCatalogService::new(
store.clone(), store.clone(),
Arc::new( Arc::new(
@@ -339,7 +339,7 @@ mod tests {
#[test] #[test]
fn create_rejects_noncanonical_repository_keys_without_persisting_a_workspace() { 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( let service = WorkspaceCatalogService::new(
store.clone(), store.clone(),
Arc::new( Arc::new(
@@ -378,7 +378,7 @@ mod tests {
for with_repository in [false, true] { for with_repository in [false, true] {
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
let database_path = temp.path().join("server.db"); 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 owner_account_id = owner_account(store.as_ref());
let materials = Arc::new(InMemoryWorkspaceSigningMaterialStore::default()); let materials = Arc::new(InMemoryWorkspaceSigningMaterialStore::default());
let service = WorkspaceCatalogService::new( let service = WorkspaceCatalogService::new(
@@ -418,7 +418,7 @@ mod tests {
drop(service); drop(service);
drop(store); 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 restarted = WorkspaceCatalogService::new(store.clone(), materials);
let created = restarted.create(request, owner_account_id).unwrap(); let created = restarted.create(request, owner_account_id).unwrap();
let identity = store let identity = store
@@ -435,7 +435,7 @@ mod tests {
for with_repository in [false, true] { for with_repository in [false, true] {
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
let database_path = temp.path().join("server.db"); 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 owner_account_id = owner_account(store.as_ref());
let materials = Arc::new(InMemoryWorkspaceSigningMaterialStore::default()); let materials = Arc::new(InMemoryWorkspaceSigningMaterialStore::default());
let service = WorkspaceCatalogService::new(store.clone(), materials.clone()); let service = WorkspaceCatalogService::new(store.clone(), materials.clone());
@@ -486,7 +486,7 @@ mod tests {
drop(service); drop(service);
drop(store); 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 restarted = WorkspaceCatalogService::new(store.clone(), materials);
let created = restarted.create(request, owner_account_id).unwrap(); let created = restarted.create(request, owner_account_id).unwrap();
let identity = store let identity = store
@@ -500,7 +500,7 @@ mod tests {
#[tokio::test] #[tokio::test]
async fn idempotency_key_reuse_with_different_payload_is_rejected() { 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 owner_account_id = owner_account(store.as_ref());
let service = WorkspaceCatalogService::new( let service = WorkspaceCatalogService::new(
store, store,
@@ -532,7 +532,7 @@ mod tests {
#[test] #[test]
fn repository_free_create_initializes_authority_and_concurrent_retries_converge() { 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 owner = owner_account(store.as_ref());
let service = WorkspaceCatalogService::new( let service = WorkspaceCatalogService::new(
store.clone(), store.clone(),
@@ -611,7 +611,7 @@ mod tests {
#[test] #[test]
fn replay_reads_current_repository_at_initial_key_and_never_recreates_it() { 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 owner = owner_account(store.as_ref());
let service = WorkspaceCatalogService::new( let service = WorkspaceCatalogService::new(
store.clone(), store.clone(),
@@ -673,13 +673,13 @@ mod tests {
repository: None, repository: None,
}; };
let (owner, created) = { 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 owner = owner_account(store.as_ref());
let service = WorkspaceCatalogService::new(store, materials.clone()); let service = WorkspaceCatalogService::new(store, materials.clone());
let created = service.create(request.clone(), owner.clone()).unwrap(); let created = service.create(request.clone(), owner.clone()).unwrap();
(owner, created) (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 reopened = WorkspaceCatalogService::new(store.clone(), materials);
let replay = reopened.create(request, owner).unwrap(); let replay = reopened.create(request, owner).unwrap();
assert_eq!(replay.workspace, created.workspace); assert_eq!(replay.workspace, created.workspace);
@@ -739,7 +739,7 @@ mod tests {
#[test] #[test]
fn create_rejects_non_user_account_owners() { 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 store
.upsert_account(&AccountRecord { .upsert_account(&AccountRecord {
account_id: "organization-owner".to_string(), account_id: "organization-owner".to_string(),
@@ -777,7 +777,7 @@ mod tests {
#[test] #[test]
fn catalog_list_is_scoped_to_the_required_owner_account() { 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_a = owner_account(store.as_ref());
let owner_b = "account-owner-b".to_string(); let owner_b = "account-owner-b".to_string();
store store
@@ -845,7 +845,7 @@ mod tests {
#[test] #[test]
fn remote_repository_creation_persists_typed_source_without_auth_metadata() { 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 owner_account_id = owner_account(store.as_ref());
let service = WorkspaceCatalogService::new( let service = WorkspaceCatalogService::new(
store.clone(), store.clone(),
@@ -922,7 +922,7 @@ mod tests {
use tempfile::tempdir; use tempfile::tempdir;
fn setup() -> (SqliteWorkspaceStore, String, String) { 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) { fn seed_workspaces(store: SqliteWorkspaceStore) -> (SqliteWorkspaceStore, String, String) {
@@ -989,7 +989,7 @@ mod tests {
let dir = tempdir().expect("tempdir"); let dir = tempdir().expect("tempdir");
let path = dir.path().join("server.db"); let path = dir.path().join("server.db");
let (store, owner, workspace_id) = 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"] { for workspace in ["workspace-a", "workspace-b"] {
let backend = ticket::SqliteTicketBackend::open_verified(&path, workspace).unwrap(); let backend = ticket::SqliteTicketBackend::open_verified(&path, workspace).unwrap();
let initial = backend let initial = backend
@@ -780,7 +780,7 @@ mod tests {
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
let database_path = temp.path().join("server.db"); 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 store
.upsert_account(&AccountRecord { .upsert_account(&AccountRecord {
account_id: "account-1".to_string(), account_id: "account-1".to_string(),
@@ -833,7 +833,7 @@ mod tests {
drop(failing_service); drop(failing_service);
drop(store); 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 service = WorkspaceSigningIdentityService::new(store.clone(), materials.clone());
let provisioned = service let provisioned = service
.provision_existing("workspace-1", "account-1") .provision_existing("workspace-1", "account-1")
@@ -962,7 +962,7 @@ mod tests {
); );
drop(service); drop(service);
drop(store); 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 restarted = WorkspaceSigningIdentityService::new(store.clone(), materials.clone());
let recovered = restarted let recovered = restarted
.provision_existing("workspace-2", "account-1") .provision_existing("workspace-2", "account-1")
@@ -738,10 +738,11 @@ mod tests {
AccountRecord, ControlPlaneStore, SqliteWorkspaceStore, WorkerRegistryRecord, AccountRecord, ControlPlaneStore, SqliteWorkspaceStore, WorkerRegistryRecord,
WorkspaceRecord, WorkspaceRecord,
}; };
let store = let store = std::sync::Arc::new(
std::sync::Arc::new(SqliteWorkspaceStore::in_memory().unwrap().with_db_executor( SqliteWorkspaceStore::in_memory_for_test()
crate::db_executor::DbExecutor::new(Default::default()).unwrap(), .unwrap()
)); .with_db_executor(crate::db_executor::DbExecutor::new(Default::default()).unwrap()),
);
store store
.upsert_account(&AccountRecord { .upsert_account(&AccountRecord {
account_id: "owner".into(), account_id: "owner".into(),