rustfs_targets/runtime/
sidecar.rs

1// Copyright 2024 RustFS Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use crate::TargetDomain;
16use crate::runtime::sidecar_protocol::SidecarHandshake;
17use serde::{Deserialize, Serialize};
18use std::time::Duration;
19use thiserror::Error;
20
21const DEFAULT_FAILURE_THRESHOLD: usize = 3;
22
23#[derive(Debug, Clone, PartialEq, Eq)]
24pub struct SidecarRuntimePolicy {
25    pub allow_external_sidecars: bool,
26    pub require_sandbox: bool,
27    pub require_provenance: bool,
28    pub max_queue_depth: usize,
29    pub operation_timeout: Duration,
30    pub failure_threshold: usize,
31    pub redact_error_details: bool,
32}
33
34impl Default for SidecarRuntimePolicy {
35    fn default() -> Self {
36        Self {
37            allow_external_sidecars: false,
38            require_sandbox: true,
39            require_provenance: true,
40            max_queue_depth: 0,
41            operation_timeout: Duration::from_secs(5),
42            failure_threshold: DEFAULT_FAILURE_THRESHOLD,
43            redact_error_details: true,
44        }
45    }
46}
47
48impl SidecarRuntimePolicy {
49    pub fn verified_external(max_queue_depth: usize, operation_timeout: Duration, failure_threshold: usize) -> Self {
50        Self {
51            allow_external_sidecars: true,
52            require_sandbox: true,
53            require_provenance: true,
54            max_queue_depth,
55            operation_timeout,
56            failure_threshold: failure_threshold.max(1),
57            redact_error_details: true,
58        }
59    }
60
61    pub fn failure_threshold(&self) -> usize {
62        self.failure_threshold.max(1)
63    }
64
65    pub fn validate_activation(&self, safety_checks: &SidecarRuntimeSafetyChecks) -> Result<(), SidecarRuntimePolicyError> {
66        validate_runtime_policy(self, safety_checks)
67    }
68}
69
70#[derive(Debug, Clone, Copy, PartialEq, Eq)]
71pub struct SidecarRuntimeSafetyChecks {
72    pub sandboxed: bool,
73    pub provenance_verified: bool,
74    pub queue_depth: usize,
75}
76
77impl SidecarRuntimeSafetyChecks {
78    pub fn verified(queue_depth: usize) -> Self {
79        Self {
80            sandboxed: true,
81            provenance_verified: true,
82            queue_depth,
83        }
84    }
85}
86
87#[derive(Debug, Error, PartialEq, Eq)]
88pub enum SidecarRuntimePolicyError {
89    #[error("external sidecar runtime is disabled by policy")]
90    ExternalSidecarDisabled,
91
92    #[error("sidecar runtime requires sandbox isolation")]
93    SandboxRequired,
94
95    #[error("sidecar runtime requires verified provenance")]
96    ProvenanceRequired,
97
98    #[error("sidecar runtime queue depth {queue_depth} exceeds policy bound {max_queue_depth}")]
99    QueueDepthExceeded { queue_depth: usize, max_queue_depth: usize },
100}
101
102#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
103#[serde(rename_all = "snake_case")]
104pub struct SidecarPluginRuntime {
105    pub endpoint: String,
106    pub handshake: SidecarHandshake,
107    pub healthy: bool,
108    pub failure_count: usize,
109    pub degraded_to_builtin: bool,
110    pub last_error: Option<String>,
111}
112
113impl SidecarPluginRuntime {
114    pub fn new(endpoint: impl Into<String>, handshake: SidecarHandshake) -> Self {
115        Self {
116            endpoint: endpoint.into(),
117            handshake,
118            healthy: false,
119            failure_count: 0,
120            degraded_to_builtin: false,
121            last_error: None,
122        }
123    }
124
125    pub fn enable(&mut self, expected_plugin_id: &str, required_domain: TargetDomain) -> Result<(), String> {
126        self.handshake.validate(expected_plugin_id)?;
127        if !self.handshake.supported_domains.contains(&required_domain) {
128            return Err(format!(
129                "sidecar plugin {} does not support required domain {:?}",
130                self.handshake.plugin_id, required_domain
131            ));
132        }
133
134        self.healthy = true;
135        self.degraded_to_builtin = false;
136        self.last_error = None;
137        self.failure_count = 0;
138        Ok(())
139    }
140
141    pub fn enable_with_policy(
142        &mut self,
143        expected_plugin_id: &str,
144        required_domain: TargetDomain,
145        policy: &SidecarRuntimePolicy,
146        safety_checks: &SidecarRuntimeSafetyChecks,
147    ) -> Result<(), String> {
148        self.handshake.validate(expected_plugin_id)?;
149        if !self.handshake.supported_domains.contains(&required_domain) {
150            return Err(format!(
151                "sidecar plugin {} does not support required domain {:?}",
152                self.handshake.plugin_id, required_domain
153            ));
154        }
155        policy.validate_activation(safety_checks).map_err(|err| err.to_string())?;
156
157        self.healthy = true;
158        self.degraded_to_builtin = false;
159        self.last_error = None;
160        self.failure_count = 0;
161        Ok(())
162    }
163
164    pub fn mark_unhealthy(&mut self) {
165        self.healthy = false;
166    }
167
168    pub fn record_failure(&mut self, error: impl Into<String>) {
169        self.failure_count = self.failure_count.saturating_add(1);
170        self.healthy = false;
171        self.last_error = Some(error.into());
172        if self.failure_count >= DEFAULT_FAILURE_THRESHOLD {
173            self.degraded_to_builtin = true;
174        }
175    }
176
177    pub fn record_failure_with_policy(&mut self, policy: &SidecarRuntimePolicy, error: impl Into<String>) {
178        self.failure_count = self.failure_count.saturating_add(1);
179        self.healthy = false;
180        self.last_error = Some(if policy.redact_error_details {
181            "sidecar operation failed".to_string()
182        } else {
183            error.into()
184        });
185        if self.failure_count >= policy.failure_threshold() {
186            self.degraded_to_builtin = true;
187        }
188    }
189
190    /// Simulates a send bounded by the policy's operation timeout. On timeout the
191    /// failure is recorded through the policy (so error details are redacted when
192    /// `redact_error_details` is set and the configurable failure threshold —
193    /// not a hardcoded constant — drives circuit breaking). A successful send
194    /// clears the failure count so transient blips never accumulate toward the
195    /// breaker.
196    pub fn send_with_timeout(&mut self, policy: &SidecarRuntimePolicy, simulated_latency: Duration) -> Result<(), String> {
197        if simulated_latency > policy.operation_timeout {
198            self.record_failure_with_policy(
199                policy,
200                format!(
201                    "sidecar send timeout after {:?} (budget {:?})",
202                    simulated_latency, policy.operation_timeout
203                ),
204            );
205            return Err(self
206                .last_error
207                .clone()
208                .unwrap_or_else(|| "sidecar operation failed".to_string()));
209        }
210        self.healthy = true;
211        self.last_error = None;
212        // Success resets the circuit-breaker accounting: a healthy send must not
213        // leave stale failures that could trip the breaker on the next blip.
214        self.failure_count = 0;
215        self.degraded_to_builtin = false;
216        Ok(())
217    }
218
219    pub fn shutdown(&mut self) {
220        self.healthy = false;
221    }
222}
223
224fn validate_runtime_policy(
225    policy: &SidecarRuntimePolicy,
226    safety_checks: &SidecarRuntimeSafetyChecks,
227) -> Result<(), SidecarRuntimePolicyError> {
228    if !policy.allow_external_sidecars {
229        return Err(SidecarRuntimePolicyError::ExternalSidecarDisabled);
230    }
231    if policy.require_sandbox && !safety_checks.sandboxed {
232        return Err(SidecarRuntimePolicyError::SandboxRequired);
233    }
234    if policy.require_provenance && !safety_checks.provenance_verified {
235        return Err(SidecarRuntimePolicyError::ProvenanceRequired);
236    }
237    if safety_checks.queue_depth > policy.max_queue_depth {
238        return Err(SidecarRuntimePolicyError::QueueDepthExceeded {
239            queue_depth: safety_checks.queue_depth,
240            max_queue_depth: policy.max_queue_depth,
241        });
242    }
243    Ok(())
244}
245
246#[cfg(test)]
247mod tests {
248    use super::{SidecarPluginRuntime, SidecarRuntimePolicy, SidecarRuntimeSafetyChecks};
249    use crate::TargetDomain;
250    use crate::runtime::sidecar_protocol::{SIDECAR_RUNTIME_PROTOCOL_VERSION, SidecarHandshake, SidecarPluginCapability};
251    use std::time::Duration;
252
253    fn notify_sidecar_handshake() -> SidecarHandshake {
254        SidecarHandshake {
255            protocol_version: SIDECAR_RUNTIME_PROTOCOL_VERSION.to_string(),
256            plugin_id: "external:webhook".to_string(),
257            plugin_version: "1.2.3".to_string(),
258            supported_domains: vec![TargetDomain::Notify],
259            capabilities: vec![
260                SidecarPluginCapability::HealthCheck,
261                SidecarPluginCapability::SendEvent,
262                SidecarPluginCapability::Shutdown,
263            ],
264        }
265    }
266
267    #[test]
268    fn sidecar_runtime_enable_marks_runtime_healthy() {
269        let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
270
271        runtime
272            .enable("external:webhook", TargetDomain::Notify)
273            .expect("sidecar runtime should enable");
274
275        assert!(runtime.healthy);
276    }
277
278    #[test]
279    fn sidecar_runtime_policy_rejects_external_activation_by_default() {
280        let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
281
282        let result = runtime.enable_with_policy(
283            "external:webhook",
284            TargetDomain::Notify,
285            &SidecarRuntimePolicy::default(),
286            &SidecarRuntimeSafetyChecks::verified(0),
287        );
288
289        assert_eq!(
290            result.as_ref().map_err(String::as_str),
291            Err("external sidecar runtime is disabled by policy")
292        );
293        assert!(!runtime.healthy);
294    }
295
296    #[test]
297    fn sidecar_runtime_policy_requires_sandbox_and_provenance() {
298        let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
299        let policy = SidecarRuntimePolicy::verified_external(16, Duration::from_secs(5), 3);
300
301        let missing_sandbox = runtime.enable_with_policy(
302            "external:webhook",
303            TargetDomain::Notify,
304            &policy,
305            &SidecarRuntimeSafetyChecks {
306                sandboxed: false,
307                provenance_verified: true,
308                queue_depth: 0,
309            },
310        );
311
312        assert_eq!(
313            missing_sandbox.as_ref().map_err(String::as_str),
314            Err("sidecar runtime requires sandbox isolation")
315        );
316
317        let missing_provenance = runtime.enable_with_policy(
318            "external:webhook",
319            TargetDomain::Notify,
320            &policy,
321            &SidecarRuntimeSafetyChecks {
322                sandboxed: true,
323                provenance_verified: false,
324                queue_depth: 0,
325            },
326        );
327
328        assert_eq!(
329            missing_provenance.as_ref().map_err(String::as_str),
330            Err("sidecar runtime requires verified provenance")
331        );
332        assert!(!runtime.healthy);
333    }
334
335    #[test]
336    fn sidecar_runtime_policy_enforces_queue_bound() {
337        let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
338        let policy = SidecarRuntimePolicy::verified_external(2, Duration::from_secs(5), 3);
339
340        let result = runtime.enable_with_policy(
341            "external:webhook",
342            TargetDomain::Notify,
343            &policy,
344            &SidecarRuntimeSafetyChecks::verified(3),
345        );
346
347        assert_eq!(
348            result.as_ref().map_err(String::as_str),
349            Err("sidecar runtime queue depth 3 exceeds policy bound 2")
350        );
351        assert!(!runtime.healthy);
352    }
353
354    #[test]
355    fn sidecar_runtime_policy_allows_verified_external_activation() {
356        let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
357        let policy = SidecarRuntimePolicy::verified_external(16, Duration::from_secs(5), 3);
358
359        runtime
360            .enable_with_policy(
361                "external:webhook",
362                TargetDomain::Notify,
363                &policy,
364                &SidecarRuntimeSafetyChecks::verified(1),
365            )
366            .expect("verified sidecar runtime should enable");
367
368        assert!(runtime.healthy);
369        assert_eq!(policy.failure_threshold(), 3);
370    }
371
372    #[test]
373    fn sidecar_runtime_policy_redacts_failure_details() {
374        let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
375        let policy = SidecarRuntimePolicy::verified_external(16, Duration::from_secs(5), 2);
376
377        runtime.record_failure_with_policy(&policy, "secret token leaked in transport error");
378        runtime.record_failure_with_policy(&policy, "another secret error");
379
380        assert_eq!(runtime.last_error.as_deref(), Some("sidecar operation failed"));
381        assert!(runtime.degraded_to_builtin);
382    }
383
384    #[test]
385    fn sidecar_runtime_enable_rejects_domain_mismatch() {
386        let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
387
388        let result = runtime.enable("external:webhook", TargetDomain::Audit);
389
390        assert!(result.is_err());
391        assert!(!runtime.healthy);
392    }
393
394    #[test]
395    fn sidecar_runtime_shutdown_marks_runtime_unhealthy() {
396        let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
397        runtime
398            .enable("external:webhook", TargetDomain::Notify)
399            .expect("sidecar runtime should enable");
400
401        runtime.shutdown();
402
403        assert!(!runtime.healthy);
404    }
405
406    #[test]
407    fn sidecar_runtime_degrades_to_builtin_after_failure_threshold() {
408        let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
409
410        runtime.record_failure("send failed");
411        runtime.record_failure("send failed again");
412        runtime.record_failure("send failed third time");
413
414        assert!(runtime.degraded_to_builtin);
415        assert!(!runtime.healthy);
416        assert_eq!(runtime.failure_count, 3);
417    }
418
419    #[test]
420    fn sidecar_runtime_send_timeout_redacts_error_and_uses_policy_threshold() {
421        let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
422        // redact_error_details defaults to true; threshold 2 is honored, not the
423        // hardcoded DEFAULT_FAILURE_THRESHOLD.
424        let policy = SidecarRuntimePolicy::verified_external(16, Duration::from_millis(50), 2);
425
426        let result = runtime.send_with_timeout(&policy, Duration::from_millis(75));
427
428        assert!(result.is_err());
429        // The raw budget/latency detail must not leak; it is redacted.
430        assert_eq!(runtime.last_error.as_deref(), Some("sidecar operation failed"));
431        assert_eq!(runtime.failure_count, 1);
432        assert!(!runtime.degraded_to_builtin);
433
434        // Second timeout hits the configured threshold and degrades.
435        let _ = runtime.send_with_timeout(&policy, Duration::from_millis(75));
436        assert_eq!(runtime.failure_count, 2);
437        assert!(runtime.degraded_to_builtin);
438    }
439
440    #[test]
441    fn sidecar_runtime_successful_send_clears_failure_count() {
442        let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
443        let policy = SidecarRuntimePolicy::verified_external(16, Duration::from_millis(50), 3);
444
445        // Accumulate a failure, then a successful send must reset the breaker
446        // accounting so a later single failure does not immediately degrade.
447        let _ = runtime.send_with_timeout(&policy, Duration::from_millis(75));
448        assert_eq!(runtime.failure_count, 1);
449
450        runtime
451            .send_with_timeout(&policy, Duration::from_millis(10))
452            .expect("a within-budget send should succeed");
453        assert_eq!(runtime.failure_count, 0);
454        assert!(!runtime.degraded_to_builtin);
455        assert!(runtime.healthy);
456        assert!(runtime.last_error.is_none());
457    }
458}
rustfs_targets/runtime/sidecar.rs

rustfs_targets/runtime/
sidecar.rs
