From 5627f47e2b5e40d149bf9882bd5fab7ac206f1ff Mon Sep 17 00:00:00 2001 From: Hare Date: Sat, 31 Oct 2026 04:58:20 +0900 Subject: [PATCH] test: verify scoped authority storage lifetime and physical initialization --- crates/feature-storage/Cargo.toml | 3 + crates/feature-storage/src/lib.rs | 35 ++++ .../feature-storage/src/open_observation.rs | 64 ++++++ crates/workspace-server/Cargo.toml | 1 + .../server_plugin_authority_storage_tests.rs | 198 ++++++++++++++++++ .../src/server_plugin_host_tests.rs | 55 +++++ .../server_plugin_service_runtime_tests.rs | 56 ++++- 7 files changed, 401 insertions(+), 11 deletions(-) create mode 100644 crates/feature-storage/src/open_observation.rs create mode 100644 crates/workspace-server/src/server_plugin_authority_storage_tests.rs diff --git a/crates/feature-storage/Cargo.toml b/crates/feature-storage/Cargo.toml index 2702517ea..19d92dbc8 100644 --- a/crates/feature-storage/Cargo.toml +++ b/crates/feature-storage/Cargo.toml @@ -5,6 +5,9 @@ edition.workspace = true license.workspace = true publish = false +[features] +test-support = [] + [dependencies] rusqlite.workspace = true thiserror.workspace = true diff --git a/crates/feature-storage/src/lib.rs b/crates/feature-storage/src/lib.rs index b5f34fe40..349c9d2c7 100644 --- a/crates/feature-storage/src/lib.rs +++ b/crates/feature-storage/src/lib.rs @@ -121,13 +121,28 @@ impl FeatureRegistration { } } +#[cfg(feature = "test-support")] +#[path = "open_observation.rs"] +mod open_observation; +#[cfg(feature = "test-support")] +pub use open_observation::OpenObservation; + #[derive(Clone, Debug)] pub struct RegisteredFeature { manager_id: Uuid, feature_id: String, + #[cfg(feature = "test-support")] + observation: Arc, } impl RegisteredFeature { + /// Instance-local physical-operation counters for fixture assertions. The + /// observer retains no database and cannot replace or suppress durable I/O. + #[cfg(feature = "test-support")] + pub fn open_observation(&self) -> OpenObservation { + self.observation.snapshot() + } + pub fn feature_id(&self) -> &str { &self.feature_id } @@ -219,6 +234,8 @@ impl FeatureStorage { Ok(RegisteredFeature { manager_id: self.inner.id, feature_id, + #[cfg(feature = "test-support")] + observation: Arc::default(), }) } @@ -279,14 +296,32 @@ impl FeatureStorage { } let feature_dir = self.feature_dir(scope_id); + #[cfg(feature = "test-support")] + let mut counted_sync = |path: &Path| { + registration.observation.sync_attempt(); + let result = sync(path); + if result.is_ok() { + registration.observation.sync_success(); + } + result + }; + #[cfg(feature = "test-support")] + durable_create_directory(&feature_dir, &mut counted_sync)?; + #[cfg(not(feature = "test-support"))] durable_create_directory(&feature_dir, &mut sync)?; let path = feature_dir.join(format!("{}.sqlite", registration.feature_id)); + #[cfg(feature = "test-support")] + registration.observation.open_attempt(); let connection = Connection::open_with_flags( path, OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_CREATE, )?; + #[cfg(feature = "test-support")] + registration.observation.open_success(); configure_connection(&connection)?; migrate(&connection, ®istration.feature_id, migrations)?; + #[cfg(feature = "test-support")] + registration.observation.initialized(); let connection = Arc::new(FeatureDatabaseInner { scope_id: scope_id.to_string(), feature_id: registration.feature_id.clone(), diff --git a/crates/feature-storage/src/open_observation.rs b/crates/feature-storage/src/open_observation.rs new file mode 100644 index 000000000..46254568b --- /dev/null +++ b/crates/feature-storage/src/open_observation.rs @@ -0,0 +1,64 @@ +use std::sync::atomic::{AtomicU64, Ordering}; + +/// Read-only observations for one registration (and its clones) across its scopes. +/// Counts real I/O boundaries, not API opens, and retains no physical database. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct OpenObservation { + pub physical_open_attempts: u64, + pub physical_opens: u64, + pub completed_initializations: u64, + pub directory_sync_attempts: u64, + pub directory_sync_successes: u64, +} + +impl OpenObservation { + pub fn since(self, before: Self) -> Self { + Self { + physical_open_attempts: self.physical_open_attempts - before.physical_open_attempts, + physical_opens: self.physical_opens - before.physical_opens, + completed_initializations: self.completed_initializations + - before.completed_initializations, + directory_sync_attempts: self.directory_sync_attempts - before.directory_sync_attempts, + directory_sync_successes: self.directory_sync_successes + - before.directory_sync_successes, + } + } +} + +#[derive(Debug, Default)] +pub(super) struct Counters { + physical_open_attempts: AtomicU64, + physical_opens: AtomicU64, + completed_initializations: AtomicU64, + directory_sync_attempts: AtomicU64, + directory_sync_successes: AtomicU64, +} + +impl Counters { + pub(super) fn snapshot(&self) -> OpenObservation { + OpenObservation { + physical_open_attempts: self.physical_open_attempts.load(Ordering::Relaxed), + physical_opens: self.physical_opens.load(Ordering::Relaxed), + completed_initializations: self.completed_initializations.load(Ordering::Relaxed), + directory_sync_attempts: self.directory_sync_attempts.load(Ordering::Relaxed), + directory_sync_successes: self.directory_sync_successes.load(Ordering::Relaxed), + } + } + pub(super) fn open_attempt(&self) { + self.physical_open_attempts.fetch_add(1, Ordering::Relaxed); + } + pub(super) fn open_success(&self) { + self.physical_opens.fetch_add(1, Ordering::Relaxed); + } + pub(super) fn initialized(&self) { + self.completed_initializations + .fetch_add(1, Ordering::Relaxed); + } + pub(super) fn sync_attempt(&self) { + self.directory_sync_attempts.fetch_add(1, Ordering::Relaxed); + } + pub(super) fn sync_success(&self) { + self.directory_sync_successes + .fetch_add(1, Ordering::Relaxed); + } +} diff --git a/crates/workspace-server/Cargo.toml b/crates/workspace-server/Cargo.toml index 955b7d83e..fbabe2908 100644 --- a/crates/workspace-server/Cargo.toml +++ b/crates/workspace-server/Cargo.toml @@ -66,6 +66,7 @@ webauthn-rs = { workspace = true, features = ["conditional-ui"] } webauthn-rs-proto = "0.5.5" [dev-dependencies] +feature-storage = { workspace = true, features = ["test-support"] } native-tls = "0.2" tokio-native-tls = "0.3" wip-protocol.workspace = true diff --git a/crates/workspace-server/src/server_plugin_authority_storage_tests.rs b/crates/workspace-server/src/server_plugin_authority_storage_tests.rs new file mode 100644 index 000000000..072533214 --- /dev/null +++ b/crates/workspace-server/src/server_plugin_authority_storage_tests.rs @@ -0,0 +1,198 @@ +//! Focused physical-lifetime/live-authority regressions through existing owners. +//! Storage counters observe real directory sync/open; they never replace that I/O. +use super::*; + +pub(super) fn independent_store( + api: &WorkspaceApi, +) -> (crate::FeatureStorage, crate::subjektiv::SubjektivStore) { + let manager = crate::FeatureStorage::for_server_database(&api.config.database_path).unwrap(); + let scope = manager.workspace(TEST_WORKSPACE_ID).unwrap(); + let registration = crate::subjektiv::SubjektivStore::register(&scope).unwrap(); + let store = crate::subjektiv::SubjektivStore::open(&scope, ®istration).unwrap(); + // Handles do not own manager lifetime: dropping it intentionally fences them. + (manager, store) +} + +fn permission(f: &Fixture) -> server_api::authorization::OperationAuthorizationRequest { + server_api::authorization::OperationAuthorizationRequest { + operation: "plugin.admit".into(), + resource: PermissionResource::PluginInstallation { + workspace_id: TEST_WORKSPACE_ID.into(), + installation_id: f.installed.installation.installation_id.clone(), + }, + } +} + +fn revoke(store: &crate::subjektiv::SubjektivStore, subject: &str) { + let previous = store.subject_permissions(subject).unwrap().unwrap(); + store + .set_subject_permissions( + subject, + &SubjectPermissionsUpdate { + operation_id: "external-revoke".into(), + expected_change_id: Some(previous.change_id), + execution: vec![], + delegation: vec![], + }, + ) + .unwrap(); +} + +#[tokio::test] +async fn sync_authority_checks_reduce_physical_initializations_without_skipping_live_reads() { + let f = Fixture::new(None).await; + // Call the unchanged synchronous owners both without and with physical-handle + // retention. These are the identity and dual-permission portions of Host checks. + for identity in [true, false] { + let check = || { + if identity { + subject_authorization::plugin_execution_identity(&f.api, &f.source) + .map(|_| ()) + .map_err(|e| e.error) + } else { + for operation in ["plugin.service.consume", "plugin.admit"] { + let mut request = permission(&f); + request.operation = operation.into(); + subject_authorization::authorize_scope(&f.api, &f.source, &request) + .map_err(|e| e.error)?; + } + Ok(()) + } + }; + let before = f.api.subjektiv_registration.open_observation(); + check().unwrap(); + let unretained = f + .api + .subjektiv_registration + .open_observation() + .since(before); + assert!(unretained.physical_opens > 1, "{unretained:?}"); + assert_check_physical_initializations(&f, before, unretained.physical_opens); + let before = f.api.subjektiv_registration.open_observation(); + plugin_host::with_authority_database(&f.api, check).unwrap(); + assert_check_physical_initializations(&f, before, 1); + let retained = f + .api + .subjektiv_registration + .open_observation() + .since(before); + println!( + "authority_physical_io identity={identity} before={unretained:?} after={retained:?}" + ); + } + f.shutdown().await; +} + +#[tokio::test] +async fn scoped_subject_database_rereads_policy_changed_on_independent_connection() { + let f = Fixture::new(None).await; + let (_manager, other) = independent_store(&f.api); + let request = permission(&f); + let before = f.api.subjektiv_registration.open_observation(); + plugin_host::with_authority_database(&f.api, || { + subject_authorization::authorize_scope(&f.api, &f.source, &request).unwrap(); + revoke(&other, &f.subject); + let error = + subject_authorization::authorize_scope(&f.api, &f.source, &request).unwrap_err(); + assert!(matches!(error.error, Error::WorkspacePermissionDenied(_))); + Ok(()) + }) + .unwrap(); + assert_check_physical_initializations(&f, before, 1); + drop(other); + f.shutdown().await; +} + +#[tokio::test] +async fn scoped_subject_database_does_not_survive_deletion_or_refresh_itself_after_restore() { + let f = Fixture::new(None).await; + let snapshot_dir = tempfile::tempdir().unwrap(); + let snapshot = snapshot_dir.path().join("snapshot"); + f.api.feature_storage.backup(&snapshot).unwrap(); + let request = permission(&f); + plugin_host::with_authority_database(&f.api, || { + subject_authorization::authorize_scope(&f.api, &f.source, &request).unwrap(); + // The guard retains no connection lock: deletion completes synchronously. + f.api.feature_storage.delete().unwrap(); + assert!(subject_authorization::authorize_scope(&f.api, &f.source, &request).is_err()); + assert!(f.api.feature_storage.restore(&snapshot).is_err()); + let fresh = + crate::FeatureStorage::for_server_database(&f.api.config.database_path).unwrap(); + let scope = fresh.workspace(TEST_WORKSPACE_ID).unwrap(); + scope.restore(&snapshot).unwrap(); + let registration = crate::subjektiv::SubjektivStore::register(&scope).unwrap(); + let mut restored = f.api.clone(); + restored.feature_storage = scope; + restored.subjektiv_registration = registration; + // A fresh Host binding can read the restored DB; the old binding stays fenced. + subject_authorization::authorize_scope(&restored, &f.source, &request).unwrap(); + assert!(subject_authorization::authorize_scope(&f.api, &f.source, &request).is_err()); + restored.feature_storage.shutdown().unwrap(); + Ok(()) + }) + .unwrap(); + f.shutdown().await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn cancelled_authority_waiter_keeps_physical_database_until_accepted_check_finishes() { + let f = Fixture::new(None).await; + let (_manager, other) = independent_store(&f.api); + let request = permission(&f); + let executor = f.api.config_store.db_executor(); + let api = f.api.clone(); + let worker = f.source.clone(); + let (entered, started) = tokio::sync::oneshot::channel(); + let (release, released) = std::sync::mpsc::sync_channel(0); + let denied = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let checked = denied.clone(); + let lane = executor.clone(); + let before = f.api.subjektiv_registration.open_observation(); + let waiter = tokio::spawn(async move { + lane.run(move || { + plugin_host::with_authority_database(&api, || { + subject_authorization::authorize_scope(&api, &worker, &request).unwrap(); + entered.send(()).unwrap(); + released.recv().unwrap(); + checked.store( + subject_authorization::authorize_scope(&api, &worker, &request).is_err(), + std::sync::atomic::Ordering::SeqCst, + ); + Ok(()) + }) + }) + .await + }); + tokio::time::timeout(std::time::Duration::from_secs(10), started) + .await + .unwrap() + .unwrap(); + waiter.abort(); + assert!(waiter.await.unwrap_err().is_cancelled()); + // This open reuses the accepted job's physical handle even after its waiter dies. + drop( + f.api + .feature_storage + .open(&f.api.subjektiv_registration) + .unwrap(), + ); + assert_check_physical_initializations(&f, before, 1); + revoke(&other, &f.subject); + release.send(()).unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(10), executor.drain()) + .await + .unwrap() + .unwrap(); + assert!(denied.load(std::sync::atomic::Ordering::SeqCst)); + assert_check_physical_initializations(&f, before, 1); + // Draining the accepted work released the guard, rather than creating a cache. + drop( + f.api + .feature_storage + .open(&f.api.subjektiv_registration) + .unwrap(), + ); + assert_check_physical_initializations(&f, before, 2); + drop(other); + f.shutdown().await; +} diff --git a/crates/workspace-server/src/server_plugin_host_tests.rs b/crates/workspace-server/src/server_plugin_host_tests.rs index 959489436..5a1e175f3 100644 --- a/crates/workspace-server/src/server_plugin_host_tests.rs +++ b/crates/workspace-server/src/server_plugin_host_tests.rs @@ -4,6 +4,8 @@ // proof verification, current Subject permissions and all Host checks are real. // Run with `cargo test -p yoi-workspace-server --lib plugin_host_tests`. // The shared TLS fixture requires openssl; tests never mutate process environment. +#[path = "server_plugin_authority_storage_tests.rs"] +mod authority_storage; #[path = "server_plugin_data_tests.rs"] mod data; @@ -1361,6 +1363,59 @@ async fn remote_owner_for_with_limit( ) } +fn assert_check_physical_initializations( + f: &Fixture, + before: crate::feature_storage::OpenObservation, + expected: u64, +) { + let observed = f + .api + .subjektiv_registration + .open_observation() + .since(before); + let directory = f + .api + .config + .database_path + .parent() + .unwrap() + .join("feature-storage") + .join(TEST_WORKSPACE_ID) + .join("features"); + let resolved = fs::canonicalize(&directory).unwrap(); + let ancestors: std::collections::HashSet<_> = + directory.ancestors().chain(resolved.ancestors()).collect(); + assert_eq!(observed.completed_initializations, expected, "{observed:?}"); + assert_eq!(observed.physical_open_attempts, expected, "{observed:?}"); + assert_eq!(observed.physical_opens, expected, "{observed:?}"); + assert_eq!( + observed.directory_sync_attempts, + expected * ancestors.len() as u64, + "{observed:?}" + ); + assert_eq!( + observed.directory_sync_successes, + observed.directory_sync_attempts + ); +} + +#[tokio::test] +async fn tool_check_reuses_physical_subject_database_only_within_each_sync_scope() { + let f = Fixture::new(None).await; + let owner = f.owner(None).await; + for _ in 0..2 { + let before = f.api.subjektiv_registration.open_observation(); + assert!(matches!( + owner.call(prepare(&f)).await.unwrap(), + PluginHostResponse::Authorized + )); + // Identity and permission checks are two separate synchronous scopes. + // The next call must open again rather than keeping a cross-call cache. + assert_check_physical_initializations(&f, before, 2); + } + f.shutdown().await; +} + fn prepare(f: &Fixture) -> PluginHostRequest { PluginHostRequest::Prepare { plan_id: f.plan.plan_id.clone(), diff --git a/crates/workspace-server/src/server_plugin_service_runtime_tests.rs b/crates/workspace-server/src/server_plugin_service_runtime_tests.rs index bf50952f0..dde3cc410 100644 --- a/crates/workspace-server/src/server_plugin_service_runtime_tests.rs +++ b/crates/workspace-server/src/server_plugin_service_runtime_tests.rs @@ -37,6 +37,33 @@ mod service_connection { f } + #[tokio::test] + async fn service_check_reuses_physical_subject_database_only_within_each_sync_scope() { + let peer = Peer::new(HOST, vec![]).await; + let f = fixture(&peer).await; + let owner = f.owner(None).await; + for _ in 0..2 { + let before = f.api.subjektiv_registration.open_observation(); + assert_eq!( + request( + owner.as_ref(), + PluginServiceRequest::Prepare { + plan_id: f.plan.plan_id.clone(), + installation_id: f.installed.installation.installation_id.clone(), + service_name: SERVICE.into(), + operation: PluginServiceLifecycleOperation::Admit, + } + ) + .await + .unwrap(), + PluginServiceResponse::Prepared + ); + assert_check_physical_initializations(&f, before, 2); + } + assert_eq!(peer.connection_count(), 0); + f.shutdown().await; + } + async fn request( owner: &dyn PluginHostOwner, request: PluginServiceRequest, @@ -1092,17 +1119,24 @@ mod service_connection { )); // Delegation validity is source-identity based, not wall-clock based. // Even identical replacement policy expires this retained child grant. - store - .set_subject_permissions( - &f.subject, - &SubjectPermissionsUpdate { - operation_id: "expire-service-source".into(), - expected_change_id: Some(source.change_id), - execution: source.execution, - delegation: source.delegation, - }, - ) - .unwrap(); + crate::server::plugin_host::with_authority_database(&f.api, || { + subject_authorization::plugin_execution_identity(&f.api, &f.source).unwrap(); + let (_manager, independent) = authority_storage::independent_store(&f.api); + independent + .set_subject_permissions( + &f.subject, + &SubjectPermissionsUpdate { + operation_id: "expire-service-source".into(), + expected_change_id: Some(source.change_id), + execution: source.execution, + delegation: source.delegation, + }, + ) + .unwrap(); + assert!(subject_authorization::plugin_execution_identity(&f.api, &f.source).is_err()); + Ok(()) + }) + .unwrap(); assert_late_operations_rejected( &owners, &lease,