termivar_scanner/plugin/
execution.rs

1use futures::FutureExt;
2use std::{panic::AssertUnwindSafe, time::Instant};
3
4use super::{
5    context::{PluginContext, PluginExecutionRequest, PluginExecutionResult},
6    recorder::sanitize_error_safely,
7    registry::PluginRegistry,
8    PluginError,
9};
10
11impl PluginRegistry {
12    /// Executes one invocation and returns observation evidence only.
13    pub async fn execute(
14        &self,
15        plugin_id: &str,
16        request: PluginExecutionRequest,
17    ) -> Result<PluginExecutionResult, PluginError> {
18        let (plugin, stats, _invocation_lease) = {
19            let entry = self.entries.get(plugin_id).ok_or(PluginError::NotFound)?;
20            if !entry.config.enabled {
21                return Err(PluginError::Disabled);
22            }
23            let stats = entry.stats.clone();
24            let lease = stats.acquire_invocation()?;
25            (entry.plugin.clone(), stats, lease)
26        };
27
28        let context = PluginContext::from_request(plugin_id.to_owned(), request)?;
29        context.ensure_active()?;
30        stats.record_execution();
31        let started = Instant::now();
32        let plugin_future =
33            match std::panic::catch_unwind(AssertUnwindSafe(|| plugin.execute(&context))) {
34                Ok(future) => future,
35                Err(_) => {
36                    context.discard();
37                    stats.record_error();
38                    return Err(PluginError::Panicked);
39                },
40            };
41        let mut execution = Some(Box::pin(AssertUnwindSafe(plugin_future).catch_unwind()));
42        let wall = tokio::time::sleep_until(context.deadline);
43        tokio::pin!(wall);
44
45        let completion = match execution.as_mut() {
46            Some(execution_future) => tokio::select! {
47                biased;
48                () = context.cancellation.cancelled() => Err(PluginError::Cancelled),
49                () = &mut wall => Err(PluginError::WallTimeExceeded),
50                result = execution_future.as_mut() => match result {
51                    Ok(result) => match result {
52                        Ok(()) => Ok(()),
53                        Err(error) => Err(sanitize_error_safely(context.redaction.as_ref(), error)
54                            .unwrap_or(PluginError::HostCallbackPanicked)),
55                    },
56                    Err(_) => Err(PluginError::Panicked),
57                },
58            },
59            None => Err(PluginError::HostStateUnavailable),
60        };
61        let drop_result = std::panic::catch_unwind(AssertUnwindSafe(|| drop(execution.take())));
62        if drop_result.is_err() {
63            context.discard();
64            stats.record_error();
65            return Err(PluginError::Panicked);
66        }
67
68        if let Err(error) = completion {
69            context.discard();
70            stats.record_error();
71            return Err(error);
72        }
73
74        match context.finish() {
75            Ok((observations, usage)) => {
76                stats.record_success();
77                Ok(PluginExecutionResult {
78                    plugin_id: plugin_id.to_owned(),
79                    observations,
80                    usage,
81                    elapsed_ms: elapsed_ms(started),
82                })
83            },
84            Err(error) => {
85                context.discard();
86                stats.record_error();
87                Err(error)
88            },
89        }
90    }
91}
92
93fn elapsed_ms(started: Instant) -> u64 {
94    u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX)
95}
termivar_scanner/plugin/execution.rs

termivar_scanner/plugin/
execution.rs
