sz_rust_workflow/integration/
plugin_unload_watcher.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2024-2026 SZ-Rust Team
3//
4use std::sync::Arc;
5
6use crate::error::WorkflowResult;
7use crate::repository::InstanceRepository;
8
9/// 插件卸载联动观察器，对齐 spec 5.4.1 规则 5。
10///
11/// 插件卸载后，扫描在途实例的待执行 plugin 节点，
12/// 引用该插件能力的节点标记为不可用。
13pub struct PluginUnloadWatcher {
14    instance_repo: Arc<dyn InstanceRepository>,
15}
16
17impl PluginUnloadWatcher {
18    pub fn new(instance_repo: Arc<dyn InstanceRepository>) -> Self {
19        Self { instance_repo }
20    }
21
22    /// 插件卸载时的回调。
23    ///
24    /// 扫描所有 running 实例，在实例上下文中标记受影响节点为不可用。
25    pub async fn on_plugin_unload(&self, plugin_name: &str) -> WorkflowResult<()> {
26        let instances = self.instance_repo.list_running().await?;
27        for mut inst in instances {
28            let mut changed = false;
29            if let serde_json::Value::Object(ref mut obj) = inst.context {
30                if let Some(unavailable) = obj
31                    .get_mut("_unavailable_plugins")
32                    .and_then(|v| v.as_array_mut())
33                {
34                    unavailable.push(serde_json::Value::String(plugin_name.to_string()));
35                    changed = true;
36                } else {
37                    obj.insert(
38                        "_unavailable_plugins".into(),
39                        serde_json::json!([plugin_name]),
40                    );
41                    changed = true;
42                }
43            }
44            if changed {
45                let expected = inst.version_lock;
46                let _ = self
47                    .instance_repo
48                    .update_with_version(&inst, expected)
49                    .await?;
50            }
51        }
52        Ok(())
53    }
54
55    /// 检查某插件是否对某实例不可用。
56    pub fn is_plugin_unavailable(
57        instance: &crate::instance::FlowInstance,
58        plugin_name: &str,
59    ) -> bool {
60        instance
61            .context
62            .get("_unavailable_plugins")
63            .and_then(|v| v.as_array())
64            .map(|arr| arr.iter().any(|v| v.as_str() == Some(plugin_name)))
65            .unwrap_or(false)
66    }
67}
68
69#[cfg(test)]
70mod tests {
71    use super::*;
72    use crate::repository::InMemoryInstanceRepository;
73
74    #[tokio::test]
75    async fn on_plugin_unload_marks_instances() {
76        let repo = Arc::new(InMemoryInstanceRepository::default());
77        let inst = crate::instance::FlowInstance::new(
78            "i1",
79            "test",
80            semver::Version::new(1, 0, 0),
81            "u1",
82            serde_json::json!({}),
83            "start",
84        );
85        repo.create(&inst).await.unwrap();
86
87        let watcher = PluginUnloadWatcher::new(repo.clone());
88        watcher.on_plugin_unload("crm").await.unwrap();
89
90        let got = repo.get("i1").await.unwrap().unwrap();
91        assert!(PluginUnloadWatcher::is_plugin_unavailable(&got, "crm"));
92        assert!(!PluginUnloadWatcher::is_plugin_unavailable(&got, "erp"));
93    }
94
95    #[tokio::test]
96    async fn no_running_instances() {
97        let repo = Arc::new(InMemoryInstanceRepository::default());
98        let watcher = PluginUnloadWatcher::new(repo);
99        watcher.on_plugin_unload("crm").await.unwrap();
100    }
101}
sz_rust_workflow/integration/plugin_unload_watcher.rs

sz_rust_workflow/integration/
plugin_unload_watcher.rs
