feat(extensions): 原子切换并恢复包配置组

frontend/src-tauri/src/extension_store.rs

@@ -236,14 +236,17 @@ impl ExtensionStore {
                 ordinary(&sidecar)?;
             }
         }
-        let db = Connection::open(database)?;
+        let mut db = Connection::open(database)?;
         db.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;")?;
         let version: i64 = db.query_row("PRAGMA user_version", [], |r| r.get(0))?;
-        if version > 2 {
+        if version > 3 {
             return Err(HostError::new("EXTENSION_SCHEMA_INCOMPATIBLE"));
         }
-        if version == 1 {
-            let backup = root.join(format!("extensions.schema1.{}.sqlite3", Uuid::new_v4()));
+        if (1..3).contains(&version) {
+            let backup = root.join(format!(
+                "extensions.schema{version}.{}.sqlite3",
+                Uuid::new_v4()
+            ));
             db.execute("VACUUM INTO ?1", [backup.to_string_lossy().as_ref()])?;
             OpenOptions::new().write(true).open(backup)?.sync_all()?;
         }
@@ -251,13 +254,88 @@ impl ExtensionStore {
             CREATE TABLE IF NOT EXISTS versions (package_key TEXT PRIMARY KEY,source TEXT NOT NULL,namespace TEXT NOT NULL,package_id TEXT NOT NULL,version TEXT NOT NULL,fingerprint TEXT NOT NULL,release TEXT NOT NULL,manifest TEXT NOT NULL,inventory TEXT NOT NULL,archive_hash TEXT NOT NULL,size INTEGER NOT NULL,signer BLOB NOT NULL,state TEXT NOT NULL);
             CREATE TABLE IF NOT EXISTS stage_operations (id TEXT PRIMARY KEY,fingerprint TEXT NOT NULL,receipt TEXT NOT NULL);
             CREATE TABLE IF NOT EXISTS prepared_packages (package_key TEXT PRIMARY KEY REFERENCES versions(package_key),directory TEXT NOT NULL UNIQUE,tree_sha256 TEXT NOT NULL);
-            PRAGMA user_version=2; COMMIT;")?;
+            CREATE TABLE IF NOT EXISTS extension_active(slot TEXT PRIMARY KEY,target TEXT NOT NULL,revision TEXT NOT NULL,pending_operation TEXT);
+            CREATE TABLE IF NOT EXISTS extension_transactions(id TEXT PRIMARY KEY,fingerprint TEXT NOT NULL,before_state TEXT NOT NULL,after_state TEXT NOT NULL,state TEXT NOT NULL);
+            PRAGMA user_version=3; COMMIT;")?;
+        crate::extension_transaction::recover(&mut db)?;
         Ok(Self {
             root,
             db,
             _lock: lock,
         })
     }
+    /// Atomically selects a prepared group after installer policy checks. This
+    /// method does not stop processes, validate configuration schemas or issue permits.
+    pub fn switch_prepared(
+        &mut self,
+        operation: &str,
+        vault_id: &str,
+        changes: &[crate::extension_transaction::Change],
+    ) -> Result<crate::extension_transaction::Receipt> {
+        use cap_fs_ext::DirExt;
+        let vault = Uuid::parse_str(vault_id)
+            .map_err(|_| HostError::new("VAULT_INVALID"))?
+            .to_string();
+        if vault != vault_id || changes.is_empty() || changes.len() > 200 {
+            return Err(HostError::new("EXTENSION_TRANSACTION_INVALID"));
+        }
+        let root = cap_std::fs::Dir::open_ambient_dir(&self.root, cap_std::ambient_authority())?;
+        let prepared = root.open_dir_nofollow("prepared")?;
+        for change in changes {
+            let (source, json, key, directory, tree): (String,String,Vec<u8>,String,String) = self.db.query_row(
+                "SELECT v.source,v.release,v.signer,p.directory,p.tree_sha256 FROM versions v JOIN prepared_packages p ON p.package_key=v.package_key WHERE v.package_key=?1",
+                [&change.target.package_key], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?,r.get(4)?)))?;
+            let release: Release = serde_json::from_str(&json)
+                .map_err(|_| HostError::new("EXTENSION_STORE_CORRUPT"))?;
+            let public: [u8; 32] = key
+                .try_into()
+                .map_err(|_| HostError::new("EXTENSION_STORE_CORRUPT"))?;
+            let (inventory, _) = release.verify_package(
+                &public,
+                &release.key_id,
+                &release.namespace,
+                false,
+                false,
+                &self.archive(&change.target.package_key)?,
+            )?;
+            let slot = hash(
+                &serde_json::to_vec(&(&vault, &source, &release.namespace, &release.package_id))
+                    .unwrap(),
+            );
+            if change.target.slot != slot
+                || change.target.directory != directory
+                || change.target.tree_sha256 != tree
+                || Uuid::parse_str(&directory)
+                    .map(|v| v.to_string())
+                    .ok()
+                    .as_ref()
+                    != Some(&directory)
+            {
+                return Err(HostError::new("EXTENSION_INSTALL_CONFLICT"));
+            }
+            if crate::extension_unpack::verify_tree(
+                &prepared.open_dir_nofollow(&directory)?,
+                &inventory,
+            )? != tree
+            {
+                return Err(HostError::new("EXTENSION_STORE_CORRUPT"));
+            }
+        }
+        crate::extension_transaction::switch(&mut self.db, operation, changes)
+    }
+    pub fn finish_installation(
+        &mut self,
+        operation: &str,
+        healthy: bool,
+    ) -> Result<crate::extension_transaction::Receipt> {
+        crate::extension_transaction::finish(&mut self.db, operation, healthy)
+    }
+    pub fn active_installation(
+        &self,
+        slot: &str,
+    ) -> Result<Option<crate::extension_transaction::Active>> {
+        crate::extension_transaction::active(&self.db, slot)
+    }
     /// Prepare a verified staged package. The caller supplies current signer/revocation
     /// policy; persisted preparation does not bypass that policy on replay.
     pub fn prepare(
@@ -544,6 +622,93 @@ mod tests {
         }
     }
     #[test]
+    fn prepared_switch_rechecks_content_vault_binding_and_recovers_on_open() {
+        use crate::extension_transaction::{Change, Target};
+        let temp = tempfile::tempdir().unwrap();
+        let (release, archive, key) = fixture();
+        let mut store = ExtensionStore::open(temp.path()).unwrap();
+        let staged = store
+            .stage(request(
+                &Uuid::new_v4().to_string(),
+                &release,
+                &archive,
+                &key,
+            ))
+            .unwrap();
+        let prepared = store
+            .prepare(
+                &staged.package_key,
+                Signer {
+                    public_key: &key,
+                    key_id: "test-key",
+                    namespace: "examples",
+                    revoked: false,
+                },
+                false,
+            )
+            .unwrap();
+        let vault = Uuid::new_v4().to_string();
+        let slot = hash(
+            &serde_json::to_vec(&(
+                &vault,
+                "https://catalog.example/",
+                &release.namespace,
+                &release.package_id,
+            ))
+            .unwrap(),
+        );
+        let changes = vec![Change {
+            target: Target {
+                slot: slot.clone(),
+                package_key: staged.package_key,
+                directory: prepared.directory.clone(),
+                tree_sha256: prepared.tree_sha256,
+                configuration: serde_json::json!({"review":true}),
+            },
+            expected_revision: None,
+        }];
+        assert!(store
+            .switch_prepared(
+                &Uuid::new_v4().to_string(),
+                &Uuid::new_v4().to_string(),
+                &changes
+            )
+            .is_err());
+        let operation = Uuid::new_v4().to_string();
+        store.switch_prepared(&operation, &vault, &changes).unwrap();
+        assert!(store
+            .active_installation(&slot)
+            .unwrap()
+            .unwrap()
+            .pending_operation
+            .is_some());
+        drop(store);
+        let mut store = ExtensionStore::open(temp.path()).unwrap();
+        assert!(store.active_installation(&slot).unwrap().is_none());
+        assert_eq!(
+            store.finish_installation(&operation, true).unwrap().state,
+            "rolled_back"
+        );
+        let operation = Uuid::new_v4().to_string();
+        store.switch_prepared(&operation, &vault, &changes).unwrap();
+        store.finish_installation(&operation, true).unwrap();
+        drop(store);
+        let mut store = ExtensionStore::open(temp.path()).unwrap();
+        assert_eq!(
+            store.active_installation(&slot).unwrap().unwrap().target,
+            changes[0].target
+        );
+        fs::write(
+            temp.path()
+                .join("prepared")
+                .join(&prepared.directory)
+                .join("persona.json"),
+            b"bad",
+        )
+        .unwrap();
+        assert!(store.switch_prepared(&operation, &vault, &changes).is_err());
+    }
+    #[test]
     fn schema_one_upgrade_preserves_versions_and_creates_readable_backup() {
         let temp = tempfile::tempdir().unwrap();
         let (release, archive, key) = fixture();

frontend/src-tauri/src/extension_transaction.rs

@@ -0,0 +1,298 @@
+//! Atomic package/configuration pointers. A pointer is never a runtime permission.
+use crate::workspace::{hash, HostError, Result};
+use rusqlite::{params, Connection, OptionalExtension};
+use serde::{Deserialize, Serialize};
+use std::collections::BTreeSet;
+
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[serde(deny_unknown_fields)]
+pub struct Target {
+    pub slot: String,
+    pub package_key: String,
+    pub directory: String,
+    pub tree_sha256: String,
+    pub configuration: serde_json::Value,
+}
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+pub struct Change {
+    pub target: Target,
+    pub expected_revision: Option<String>,
+}
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+pub struct Active {
+    pub target: Target,
+    pub revision: String,
+    pub pending_operation: Option<String>,
+}
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
+pub struct Receipt {
+    pub operation_id: String,
+    pub state: String,
+}
+pub fn schema(db: &Connection) -> Result<()> {
+    db.execute_batch("CREATE TABLE IF NOT EXISTS extension_active(slot TEXT PRIMARY KEY,target TEXT NOT NULL,revision TEXT NOT NULL,pending_operation TEXT);
+        CREATE TABLE IF NOT EXISTS extension_transactions(id TEXT PRIMARY KEY,fingerprint TEXT NOT NULL,before_state TEXT NOT NULL,after_state TEXT NOT NULL,state TEXT NOT NULL);")?;
+    Ok(())
+}
+pub fn active(db: &Connection, slot: &str) -> Result<Option<Active>> {
+    let row: Option<(String, String, Option<String>)> = db
+        .query_row(
+            "SELECT target,revision,pending_operation FROM extension_active WHERE slot=?1",
+            [slot],
+            |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
+        )
+        .optional()?;
+    row.map(|(target, revision, pending_operation)| {
+        Ok(Active {
+            target: serde_json::from_str(&target)
+                .map_err(|_| HostError::new("EXTENSION_STORE_CORRUPT"))?,
+            revision,
+            pending_operation,
+        })
+    })
+    .transpose()
+}
+fn encoded<T: Serialize>(value: &T) -> Result<String> {
+    serde_json::to_string(value).map_err(|_| HostError::new("EXTENSION_TRANSACTION_INVALID"))
+}
+pub fn switch(db: &mut Connection, operation: &str, changes: &[Change]) -> Result<Receipt> {
+    switch_inner(db, operation, changes, |_| Ok(()))
+}
+fn switch_inner(
+    db: &mut Connection,
+    operation: &str,
+    changes: &[Change],
+    mut checkpoint: impl FnMut(&str) -> Result<()>,
+) -> Result<Receipt> {
+    if uuid::Uuid::parse_str(operation).is_err() || changes.is_empty() || changes.len() > 200 {
+        return Err(HostError::new("EXTENSION_TRANSACTION_INVALID"));
+    }
+    let serialized = encoded(&changes)?;
+    if serialized.len() > 4 * 1024 * 1024 {
+        return Err(HostError::new("EXTENSION_TRANSACTION_INVALID"));
+    }
+    let fingerprint = hash(serialized.as_bytes());
+    let transaction = db.transaction()?;
+    let prior: Option<(String, String)> = transaction
+        .query_row(
+            "SELECT fingerprint,state FROM extension_transactions WHERE id=?1",
+            [operation],
+            |r| Ok((r.get(0)?, r.get(1)?)),
+        )
+        .optional()?;
+    if let Some((previous, state)) = prior {
+        if previous != fingerprint {
+            return Err(HostError::new("OPERATION_REUSED"));
+        }
+        return Ok(Receipt {
+            operation_id: operation.into(),
+            state,
+        });
+    }
+    let mut slots = BTreeSet::new();
+    let mut before = Vec::new();
+    for change in changes {
+        let target = &change.target;
+        for digest in [&target.slot, &target.package_key, &target.tree_sha256] {
+            if digest.len() != 64
+                || !digest
+                    .bytes()
+                    .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
+            {
+                return Err(HostError::new("EXTENSION_TRANSACTION_INVALID"));
+            }
+        }
+        if uuid::Uuid::parse_str(&target.directory)
+            .map(|v| v.to_string())
+            .ok()
+            .as_ref()
+            != Some(&target.directory)
+            || !target.configuration.is_object()
+            || !slots.insert(&target.slot)
+        {
+            return Err(HostError::new("EXTENSION_TRANSACTION_INVALID"));
+        }
+        let previous = active(&transaction, &target.slot)?;
+        if previous
+            .as_ref()
+            .is_some_and(|p| p.pending_operation.is_some())
+        {
+            return Err(HostError::new("EXTENSION_TRANSACTION_BUSY"));
+        }
+        if previous.as_ref().map(|p| &p.revision) != change.expected_revision.as_ref() {
+            return Err(HostError::new("EXTENSION_INSTALL_CONFLICT"));
+        }
+        before.push(previous);
+    }
+    transaction.execute(
+        "INSERT INTO extension_transactions VALUES (?1,?2,?3,?4,'checking')",
+        params![operation, fingerprint, encoded(&before)?, serialized],
+    )?;
+    checkpoint("journal_recorded")?;
+    for change in changes {
+        let target = encoded(&change.target)?;
+        transaction.execute("INSERT INTO extension_active VALUES (?1,?2,?3,?4) ON CONFLICT(slot) DO UPDATE SET target=excluded.target,revision=excluded.revision,pending_operation=excluded.pending_operation",
+            params![change.target.slot,target,hash(target.as_bytes()),operation])?;
+        checkpoint("pointer_recorded")?;
+    }
+    transaction.commit()?;
+    checkpoint("switch_committed")?;
+    Ok(Receipt {
+        operation_id: operation.into(),
+        state: "checking".into(),
+    })
+}
+
+/// `healthy` must come from the Host's matching package/config health probe.
+/// Recovery calls this with false; it never reissues any execution permits.
+pub fn finish(db: &mut Connection, operation: &str, healthy: bool) -> Result<Receipt> {
+    let tx = db.transaction()?;
+    let (before, after, state): (String, String, String) = tx.query_row(
+        "SELECT before_state,after_state,state FROM extension_transactions WHERE id=?1",
+        [operation],
+        |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
+    )?;
+    if state != "checking" {
+        return Ok(Receipt {
+            operation_id: operation.into(),
+            state,
+        });
+    }
+    let changes: Vec<Change> =
+        serde_json::from_str(&after).map_err(|_| HostError::new("EXTENSION_STORE_CORRUPT"))?;
+    let previous: Vec<Option<Active>> =
+        serde_json::from_str(&before).map_err(|_| HostError::new("EXTENSION_STORE_CORRUPT"))?;
+    if previous.len() != changes.len() {
+        return Err(HostError::new("EXTENSION_STORE_CORRUPT"));
+    }
+    for (change, old) in changes.iter().zip(previous) {
+        let current = active(&tx, &change.target.slot)?
+            .ok_or_else(|| HostError::new("EXTENSION_STORE_CORRUPT"))?;
+        if current.pending_operation.as_deref() != Some(operation)
+            || current.target != change.target
+        {
+            return Err(HostError::new("EXTENSION_STORE_CORRUPT"));
+        }
+        if healthy {
+            tx.execute(
+                "UPDATE extension_active SET pending_operation=NULL WHERE slot=?1",
+                [&change.target.slot],
+            )?;
+        } else if let Some(old) = old {
+            tx.execute("UPDATE extension_active SET target=?2,revision=?3,pending_operation=NULL WHERE slot=?1",
+                params![change.target.slot,encoded(&old.target)?,old.revision])?;
+        } else {
+            tx.execute(
+                "DELETE FROM extension_active WHERE slot=?1",
+                [&change.target.slot],
+            )?;
+        }
+    }
+    let state = if healthy { "complete" } else { "rolled_back" };
+    tx.execute(
+        "UPDATE extension_transactions SET state=?2 WHERE id=?1",
+        params![operation, state],
+    )?;
+    tx.commit()?;
+    Ok(Receipt {
+        operation_id: operation.into(),
+        state: state.into(),
+    })
+}
+pub fn recover(db: &mut Connection) -> Result<usize> {
+    let ids: Vec<String> = db
+        .prepare("SELECT id FROM extension_transactions WHERE state='checking' ORDER BY id")?
+        .query_map([], |r| r.get(0))?
+        .collect::<std::result::Result<_, _>>()?;
+    for id in &ids {
+        finish(db, id, false)?;
+    }
+    Ok(ids.len())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    fn open(path: &std::path::Path) -> Connection {
+        let db = Connection::open(path).unwrap();
+        db.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;")
+            .unwrap();
+        schema(&db).unwrap();
+        db
+    }
+    fn change(slot: char, version: u8, previous: Option<String>) -> Change {
+        Change {
+            target: Target {
+                slot: slot.to_string().repeat(64),
+                package_key: format!("{version:x}").repeat(64),
+                directory: uuid::Uuid::new_v4().to_string(),
+                tree_sha256: "a".repeat(64),
+                configuration: serde_json::json!({"version": version}),
+            },
+            expected_revision: previous,
+        }
+    }
+    #[test]
+    fn group_switch_crashes_recover_matching_packages_and_configuration() {
+        for boundary in ["journal_recorded", "pointer_recorded", "switch_committed"] {
+            for _ in 0..20 {
+                let temp = tempfile::tempdir().unwrap();
+                let path = temp.path().join("state.sqlite3");
+                let mut db = open(&path);
+                let old = vec![change('a', 1, None), change('b', 1, None)];
+                let id = uuid::Uuid::new_v4().to_string();
+                switch(&mut db, &id, &old).unwrap();
+                finish(&mut db, &id, true).unwrap();
+                let next: Vec<_> = old
+                    .iter()
+                    .map(|c| {
+                        change(
+                            c.target.slot.chars().next().unwrap(),
+                            2,
+                            Some(active(&db, &c.target.slot).unwrap().unwrap().revision),
+                        )
+                    })
+                    .collect();
+                let update = uuid::Uuid::new_v4().to_string();
+                assert!(
+                    switch_inner(&mut db, &update, &next, |at| if at == boundary {
+                        Err(HostError::new("INJECTED"))
+                    } else {
+                        Ok(())
+                    })
+                    .is_err()
+                );
+                drop(db);
+                let mut db = open(&path);
+                recover(&mut db).unwrap();
+                for original in &old {
+                    let current = active(&db, &original.target.slot).unwrap().unwrap();
+                    assert_eq!(current.target, original.target);
+                    assert!(current.pending_operation.is_none());
+                }
+                assert_eq!(recover(&mut db).unwrap(), 0);
+            }
+        }
+    }
+    #[test]
+    fn cas_busy_replay_and_failed_first_install() {
+        let mut db = Connection::open_in_memory().unwrap();
+        schema(&db).unwrap();
+        let c = vec![change('a', 1, None)];
+        let id = uuid::Uuid::new_v4().to_string();
+        switch(&mut db, &id, &c).unwrap();
+        assert!(switch(&mut db, &uuid::Uuid::new_v4().to_string(), &c).is_err());
+        assert_eq!(switch(&mut db, &id, &c).unwrap().state, "checking");
+        finish(&mut db, &id, false).unwrap();
+        assert!(active(&db, &c[0].target.slot).unwrap().is_none());
+        assert_eq!(switch(&mut db, &id, &c).unwrap().state, "rolled_back");
+        assert_eq!(finish(&mut db, &id, true).unwrap().state, "rolled_back");
+        let id = uuid::Uuid::new_v4().to_string();
+        switch(&mut db, &id, &c).unwrap();
+        finish(&mut db, &id, true).unwrap();
+        assert!(switch(&mut db, &uuid::Uuid::new_v4().to_string(), &c).is_err());
+        let mut different = c.clone();
+        different[0].target.configuration = serde_json::json!({"changed":true});
+        assert!(switch(&mut db, &id, &different).is_err());
+    }
+}

frontend/src-tauri/src/lib.rs

@@ -42,3 +42,6 @@ pub mod extension_unpack;
 
 #[cfg(feature = "desktop")]
 pub mod extension_permit;
+
+#[cfg(feature = "desktop")]
+pub mod extension_transaction;