greentic_runner_host/
runtime.rs

1use std::collections::HashMap;
2use std::future::Future;
3use std::path::Path;
4use std::sync::Arc;
5
6use anyhow::{Context, Result, anyhow, bail};
7use arc_swap::ArcSwap;
8use parking_lot::Mutex;
9use reqwest::Client;
10use tokio::runtime::{Handle, Runtime};
11use tokio::task::JoinHandle;
12
13use crate::config::HostConfig;
14use crate::engine::host::{SessionHost, StateHost};
15use crate::engine::runtime::StateMachineRuntime;
16use crate::oauth::{OAuthBrokerConfig, request_resource_token};
17use crate::operator_metrics::OperatorMetrics;
18use crate::operator_registry::OperatorRegistry;
19use crate::pack::{ComponentResolution, PackRuntime};
20use crate::runner::adapt_events_email::{
21    EmailExecutionPlan, EmailSendRequest, build_email_execution_plan, execute_email_request,
22};
23use crate::runner::contract_cache::{ContractCache, ContractCacheStats};
24use crate::runner::engine::FlowEngine;
25use crate::runner::mocks::MockLayer;
26use crate::secrets::{DynSecretsManager, canonicalize_secret_key, read_secret_blocking};
27use crate::storage::session::DynSessionStore;
28use crate::storage::state::DynStateStore;
29use crate::trace::PackTraceInfo;
30use crate::wasi::RunnerWasiPolicy;
31use greentic_types::SecretRequirement;
32
33const RUNTIME_SECRETS_PACK_ID: &str = "_runner";
34
35/// Atomically swapped view of live tenant runtimes.
36pub struct ActivePacks {
37    inner: ArcSwap<HashMap<String, Arc<TenantRuntime>>>,
38}
39
40impl ActivePacks {
41    pub fn new() -> Self {
42        Self {
43            inner: ArcSwap::from_pointee(HashMap::new()),
44        }
45    }
46
47    pub fn load(&self, tenant: &str) -> Option<Arc<TenantRuntime>> {
48        self.inner.load().get(tenant).cloned()
49    }
50
51    pub fn snapshot(&self) -> Arc<HashMap<String, Arc<TenantRuntime>>> {
52        self.inner.load_full()
53    }
54
55    pub fn replace(&self, next: HashMap<String, Arc<TenantRuntime>>) {
56        self.inner.store(Arc::new(next));
57    }
58
59    pub fn len(&self) -> usize {
60        self.inner.load().len()
61    }
62
63    pub fn is_empty(&self) -> bool {
64        self.len() == 0
65    }
66}
67
68impl Default for ActivePacks {
69    fn default() -> Self {
70        Self::new()
71    }
72}
73
74/// Runtime bundle for a tenant pack.
75pub struct TenantRuntime {
76    tenant: String,
77    config: Arc<HostConfig>,
78    packs: Vec<Arc<PackRuntime>>,
79    digests: Vec<Option<String>>,
80    engine: Arc<FlowEngine>,
81    state_machine: Arc<StateMachineRuntime>,
82    http_client: Client,
83    mocks: Option<Arc<MockLayer>>,
84    timer_handles: Mutex<Vec<JoinHandle<()>>>,
85    secrets: DynSecretsManager,
86    operator_registry: OperatorRegistry,
87    operator_metrics: Arc<OperatorMetrics>,
88    contract_cache: ContractCache,
89}
90
91#[derive(Clone)]
92pub struct ResolvedComponent {
93    pub digest: String,
94    pub component_ref: String,
95    pub pack: Arc<PackRuntime>,
96}
97
98/// Block on a future whether or not we're already inside a tokio runtime.
99pub fn block_on<F: Future<Output = R>, R>(future: F) -> R {
100    if let Ok(handle) = Handle::try_current() {
101        handle.block_on(future)
102    } else {
103        Runtime::new()
104            .expect("failed to create tokio runtime")
105            .block_on(future)
106    }
107}
108
109impl TenantRuntime {
110    #[allow(clippy::too_many_arguments)]
111    pub async fn load(
112        pack_path: &Path,
113        config: Arc<HostConfig>,
114        mocks: Option<Arc<MockLayer>>,
115        archive_source: Option<&Path>,
116        digest: Option<String>,
117        wasi_policy: Arc<RunnerWasiPolicy>,
118        session_host: Arc<dyn SessionHost>,
119        session_store: DynSessionStore,
120        state_store: DynStateStore,
121        state_host: Arc<dyn StateHost>,
122        secrets_manager: DynSecretsManager,
123    ) -> Result<Arc<Self>> {
124        let oauth_config = config.oauth_broker_config();
125        let pack = Arc::new(
126            PackRuntime::load(
127                pack_path,
128                Arc::clone(&config),
129                mocks.clone(),
130                archive_source,
131                Some(Arc::clone(&session_store)),
132                Some(Arc::clone(&state_store)),
133                Arc::clone(&wasi_policy),
134                Arc::clone(&secrets_manager),
135                oauth_config.clone(),
136                true,
137                ComponentResolution::default(),
138            )
139            .await
140            .with_context(|| {
141                format!(
142                    "failed to load pack {} for tenant {}",
143                    pack_path.display(),
144                    config.tenant
145                )
146            })?,
147        );
148        Self::from_packs(
149            config,
150            vec![(pack, digest)],
151            mocks,
152            session_host,
153            session_store,
154            state_store,
155            state_host,
156            secrets_manager,
157        )
158        .await
159    }
160
161    #[allow(clippy::too_many_arguments)]
162    pub async fn from_packs(
163        config: Arc<HostConfig>,
164        packs: Vec<(Arc<PackRuntime>, Option<String>)>,
165        mocks: Option<Arc<MockLayer>>,
166        session_host: Arc<dyn SessionHost>,
167        session_store: DynSessionStore,
168        _state_store: DynStateStore,
169        state_host: Arc<dyn StateHost>,
170        secrets_manager: DynSecretsManager,
171    ) -> Result<Arc<Self>> {
172        let operator_registry = OperatorRegistry::build(&packs)?;
173        let operator_metrics = Arc::new(OperatorMetrics::default());
174        let pack_runtimes = packs
175            .iter()
176            .map(|(pack, _)| Arc::clone(pack))
177            .collect::<Vec<_>>();
178        let digests = packs
179            .iter()
180            .map(|(_, digest)| digest.clone())
181            .collect::<Vec<_>>();
182        let mut pack_trace = HashMap::new();
183        for (pack, digest) in &packs {
184            let pack_id = pack.metadata().pack_id.clone();
185            let pack_ref = config
186                .pack_bindings
187                .iter()
188                .find(|binding| binding.pack_id == pack_id)
189                .map(|binding| binding.pack_ref.clone())
190                .unwrap_or_else(|| pack_id.clone());
191            pack_trace.insert(
192                pack_id,
193                PackTraceInfo {
194                    pack_ref,
195                    resolved_digest: digest.clone(),
196                },
197            );
198        }
199        let engine = Arc::new(
200            FlowEngine::new(pack_runtimes.clone(), Arc::clone(&config))
201                .await
202                .context("failed to prime flow engine")?,
203        );
204        let state_machine = Arc::new(
205            StateMachineRuntime::from_flow_engine(
206                Arc::clone(&config),
207                Arc::clone(&engine),
208                pack_trace,
209                session_host,
210                session_store,
211                state_host,
212                Arc::clone(&secrets_manager),
213                mocks.clone(),
214            )
215            .context("failed to initialise state machine runtime")?,
216        );
217        let http_client = Client::builder().build()?;
218        Ok(Arc::new(Self {
219            tenant: config.tenant.clone(),
220            config,
221            packs: pack_runtimes,
222            digests,
223            engine,
224            state_machine,
225            http_client,
226            mocks,
227            timer_handles: Mutex::new(Vec::new()),
228            secrets: secrets_manager,
229            operator_registry,
230            operator_metrics,
231            contract_cache: ContractCache::from_env(),
232        }))
233    }
234
235    pub fn tenant(&self) -> &str {
236        &self.tenant
237    }
238
239    pub fn config(&self) -> &Arc<HostConfig> {
240        &self.config
241    }
242
243    pub fn operator_registry(&self) -> &OperatorRegistry {
244        &self.operator_registry
245    }
246
247    pub fn operator_metrics(&self) -> &OperatorMetrics {
248        &self.operator_metrics
249    }
250
251    pub fn contract_cache(&self) -> &ContractCache {
252        &self.contract_cache
253    }
254
255    pub fn contract_cache_stats(&self) -> ContractCacheStats {
256        self.contract_cache.stats()
257    }
258
259    pub fn main_pack(&self) -> &Arc<PackRuntime> {
260        self.packs
261            .first()
262            .expect("tenant runtime must contain at least one pack")
263    }
264
265    pub fn pack(&self) -> Arc<PackRuntime> {
266        Arc::clone(self.main_pack())
267    }
268
269    pub fn overlays(&self) -> Vec<Arc<PackRuntime>> {
270        self.packs.iter().skip(1).cloned().collect()
271    }
272
273    pub fn engine(&self) -> &Arc<FlowEngine> {
274        &self.engine
275    }
276
277    pub fn state_machine(&self) -> &Arc<StateMachineRuntime> {
278        &self.state_machine
279    }
280
281    pub fn http_client(&self) -> &Client {
282        &self.http_client
283    }
284
285    pub fn oauth_config(&self) -> Option<OAuthBrokerConfig> {
286        self.config.oauth_broker_config()
287    }
288
289    pub fn digest(&self) -> Option<&str> {
290        self.digests.first().and_then(|d| d.as_deref())
291    }
292
293    pub fn overlay_digests(&self) -> Vec<Option<String>> {
294        self.digests.iter().skip(1).cloned().collect()
295    }
296
297    pub fn required_secrets(&self) -> Vec<SecretRequirement> {
298        self.packs
299            .iter()
300            .flat_map(|pack| pack.required_secrets().iter().cloned())
301            .collect()
302    }
303
304    pub fn missing_secrets(&self) -> Vec<SecretRequirement> {
305        self.packs
306            .iter()
307            .flat_map(|pack| pack.missing_secrets(&self.config.tenant_ctx()))
308            .collect()
309    }
310
311    pub fn mocks(&self) -> Option<&Arc<MockLayer>> {
312        self.mocks.as_ref()
313    }
314
315    pub fn register_timers(&self, handles: Vec<JoinHandle<()>>) {
316        self.timer_handles.lock().extend(handles);
317    }
318
319    pub fn get_secret(&self, key: &str) -> Result<String> {
320        if crate::provider_core_only::is_enabled() {
321            bail!(crate::provider_core_only::blocked_message("secrets"))
322        }
323        if !self.config.secrets_policy.is_allowed(key) {
324            bail!("secret {key} is not permitted by bindings policy");
325        }
326        let ctx = self.config.tenant_ctx();
327        let canonical_key = canonicalize_secret_key(key);
328        let bytes =
329            read_secret_blocking(&self.secrets, &ctx, RUNTIME_SECRETS_PACK_ID, &canonical_key)
330                .context("failed to read secret from manager")?;
331        let value = String::from_utf8(bytes).context("secret value is not valid UTF-8")?;
332        Ok(value)
333    }
334
335    pub fn build_events_email_execution_plan(
336        &self,
337        tenant: &greentic_types::TenantCtx,
338        request: &EmailSendRequest,
339    ) -> Result<EmailExecutionPlan> {
340        let oauth = self
341            .oauth_config()
342            .ok_or_else(|| anyhow!("oauth broker config is not configured for tenant runtime"))?;
343        build_email_execution_plan(&oauth, tenant, request)
344    }
345
346    pub async fn execute_events_email_request(
347        &self,
348        access_token: &str,
349        request: &EmailSendRequest,
350    ) -> Result<()> {
351        execute_email_request(self.http_client(), access_token, request).await
352    }
353
354    pub async fn execute_events_email_with_oauth(
355        &self,
356        tenant: &greentic_types::TenantCtx,
357        request: &EmailSendRequest,
358    ) -> Result<()> {
359        let plan = self.build_events_email_execution_plan(tenant, request)?;
360        let token = request_resource_token(self.http_client(), &plan.token_request).await?;
361        self.execute_events_email_request(&token.access_token, request)
362            .await
363    }
364
365    pub fn pack_for_component(&self, component_ref: &str) -> Option<Arc<PackRuntime>> {
366        self.packs
367            .iter()
368            .find(|pack| pack.contains_component(component_ref))
369            .cloned()
370    }
371
372    pub fn pack_for_component_with_digest(
373        &self,
374        component_ref: &str,
375    ) -> Option<(Arc<PackRuntime>, Option<String>)> {
376        self.packs
377            .iter()
378            .zip(self.digests.iter())
379            .find(|(pack, _)| pack.contains_component(component_ref))
380            .map(|(pack, digest)| (Arc::clone(pack), digest.clone()))
381    }
382
383    pub fn resolve_component(&self, component_ref: &str) -> Option<ResolvedComponent> {
384        self.pack_for_component_with_digest(component_ref)
385            .map(|(pack, digest)| ResolvedComponent {
386                digest: digest
387                    .or_else(|| self.digest().map(ToString::to_string))
388                    .unwrap_or_else(|| "unknown".to_string()),
389                component_ref: component_ref.to_string(),
390                pack,
391            })
392    }
393}
394
395impl Drop for TenantRuntime {
396    fn drop(&mut self) {
397        for handle in self.timer_handles.lock().drain(..) {
398            handle.abort();
399        }
400    }
401}
greentic_runner_host/runtime.rs

greentic_runner_host/
runtime.rs
