test: verify scoped authority storage lifetime and physical initialization

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

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<open_observation::Counters>,
 }
 
 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, &registration.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(),

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);
+    }
+}

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

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, &registration).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;
+}

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(),

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
+        crate::server::plugin_host::with_authority_database(&f.api, || {
-            .set_subject_permissions(
+            subject_authorization::plugin_execution_identity(&f.api, &f.source).unwrap();
-                &f.subject,
+            let (_manager, independent) = authority_storage::independent_store(&f.api);
-                &SubjectPermissionsUpdate {
+            independent
-                    operation_id: "expire-service-source".into(),
+                .set_subject_permissions(
-                    expected_change_id: Some(source.change_id),
+                    &f.subject,
-                    execution: source.execution,
+                    &SubjectPermissionsUpdate {
-                    delegation: source.delegation,
+                        operation_id: "expire-service-source".into(),
-                },
+                        expected_change_id: Some(source.change_id),
-            )
+                        execution: source.execution,
-            .unwrap();
+                        delegation: source.delegation,
+                    },
+                )
+                .unwrap();
+            assert!(subject_authorization::plugin_execution_identity(&f.api, &f.source).is_err());
+            Ok(())
+        })
+        .unwrap();
         assert_late_operations_rejected(
             &owners,
             &lease,