greentic_runner_host/
runtime.rs

1use std::collections::HashMap;
2use std::num::NonZeroUsize;
3use std::path::Path;
4use std::sync::Arc;
5use std::time::Instant;
6
7use anyhow::{Context, Result, bail};
8use arc_swap::ArcSwap;
9use axum::http::StatusCode;
10use lru::LruCache;
11use parking_lot::Mutex;
12use reqwest::Client;
13use serde_json::Value;
14use tokio::task::JoinHandle;
15
16use crate::config::HostConfig;
17use crate::engine::host::{SessionHost, StateHost};
18use crate::engine::runtime::StateMachineRuntime;
19use crate::pack::PackRuntime;
20use crate::runner::engine::FlowEngine;
21use crate::runner::mocks::MockLayer;
22use crate::secrets::{DynSecretsManager, read_secret_blocking};
23use crate::storage::session::DynSessionStore;
24use crate::storage::state::DynStateStore;
25use crate::wasi::RunnerWasiPolicy;
26use greentic_types::SecretRequirement;
27
28const TELEGRAM_CACHE_CAPACITY: usize = 1024;
29const WEBHOOK_CACHE_CAPACITY: usize = 256;
30
31/// Atomically swapped view of live tenant runtimes.
32pub struct ActivePacks {
33    inner: ArcSwap<HashMap<String, Arc<TenantRuntime>>>,
34}
35
36impl ActivePacks {
37    pub fn new() -> Self {
38        Self {
39            inner: ArcSwap::from_pointee(HashMap::new()),
40        }
41    }
42
43    pub fn load(&self, tenant: &str) -> Option<Arc<TenantRuntime>> {
44        self.inner.load().get(tenant).cloned()
45    }
46
47    pub fn snapshot(&self) -> Arc<HashMap<String, Arc<TenantRuntime>>> {
48        self.inner.load_full()
49    }
50
51    pub fn replace(&self, next: HashMap<String, Arc<TenantRuntime>>) {
52        self.inner.store(Arc::new(next));
53    }
54
55    pub fn len(&self) -> usize {
56        self.inner.load().len()
57    }
58
59    pub fn is_empty(&self) -> bool {
60        self.len() == 0
61    }
62}
63
64impl Default for ActivePacks {
65    fn default() -> Self {
66        Self::new()
67    }
68}
69
70/// Runtime bundle for a tenant pack.
71pub struct TenantRuntime {
72    tenant: String,
73    config: Arc<HostConfig>,
74    packs: Vec<Arc<PackRuntime>>,
75    digests: Vec<Option<String>>,
76    engine: Arc<FlowEngine>,
77    state_machine: Arc<StateMachineRuntime>,
78    http_client: Client,
79    telegram_cache: Mutex<LruCache<i64, StatusCode>>,
80    webhook_cache: Mutex<LruCache<String, Value>>,
81    messaging_rate: Mutex<RateLimiter>,
82    mocks: Option<Arc<MockLayer>>,
83    timer_handles: Mutex<Vec<JoinHandle<()>>>,
84    secrets: DynSecretsManager,
85}
86
87impl TenantRuntime {
88    #[allow(clippy::too_many_arguments)]
89    pub async fn load(
90        pack_path: &Path,
91        config: Arc<HostConfig>,
92        mocks: Option<Arc<MockLayer>>,
93        archive_source: Option<&Path>,
94        digest: Option<String>,
95        wasi_policy: Arc<RunnerWasiPolicy>,
96        session_host: Arc<dyn SessionHost>,
97        session_store: DynSessionStore,
98        state_store: DynStateStore,
99        state_host: Arc<dyn StateHost>,
100        secrets_manager: DynSecretsManager,
101    ) -> Result<Arc<Self>> {
102        let oauth_config = config.oauth_broker_config();
103        let pack = Arc::new(
104            PackRuntime::load(
105                pack_path,
106                Arc::clone(&config),
107                mocks.clone(),
108                archive_source,
109                Some(Arc::clone(&session_store)),
110                Some(Arc::clone(&state_store)),
111                Arc::clone(&wasi_policy),
112                Arc::clone(&secrets_manager),
113                oauth_config.clone(),
114                true,
115            )
116            .await
117            .with_context(|| {
118                format!(
119                    "failed to load pack {} for tenant {}",
120                    pack_path.display(),
121                    config.tenant
122                )
123            })?,
124        );
125        Self::from_packs(
126            config,
127            vec![(pack, digest)],
128            mocks,
129            session_host,
130            session_store,
131            state_store,
132            state_host,
133            secrets_manager,
134        )
135        .await
136    }
137
138    #[allow(clippy::too_many_arguments)]
139    pub async fn from_packs(
140        config: Arc<HostConfig>,
141        packs: Vec<(Arc<PackRuntime>, Option<String>)>,
142        mocks: Option<Arc<MockLayer>>,
143        session_host: Arc<dyn SessionHost>,
144        session_store: DynSessionStore,
145        _state_store: DynStateStore,
146        state_host: Arc<dyn StateHost>,
147        secrets_manager: DynSecretsManager,
148    ) -> Result<Arc<Self>> {
149        let telegram_capacity = NonZeroUsize::new(TELEGRAM_CACHE_CAPACITY)
150            .expect("telegram cache capacity must be > 0");
151        let webhook_capacity =
152            NonZeroUsize::new(WEBHOOK_CACHE_CAPACITY).expect("webhook cache capacity must be > 0");
153        let pack_runtimes = packs
154            .iter()
155            .map(|(pack, _)| Arc::clone(pack))
156            .collect::<Vec<_>>();
157        let digests = packs
158            .iter()
159            .map(|(_, digest)| digest.clone())
160            .collect::<Vec<_>>();
161        let engine = Arc::new(
162            FlowEngine::new(pack_runtimes.clone(), Arc::clone(&config))
163                .await
164                .context("failed to prime flow engine")?,
165        );
166        let state_machine = Arc::new(
167            StateMachineRuntime::from_flow_engine(
168                Arc::clone(&config),
169                Arc::clone(&engine),
170                session_host,
171                session_store,
172                state_host,
173                Arc::clone(&secrets_manager),
174                mocks.clone(),
175            )
176            .context("failed to initialise state machine runtime")?,
177        );
178        let http_client = Client::builder().build()?;
179        let rate_limits = config.rate_limits.clone();
180        Ok(Arc::new(Self {
181            tenant: config.tenant.clone(),
182            config,
183            packs: pack_runtimes,
184            digests,
185            engine,
186            state_machine,
187            http_client,
188            telegram_cache: Mutex::new(LruCache::new(telegram_capacity)),
189            webhook_cache: Mutex::new(LruCache::new(webhook_capacity)),
190            messaging_rate: Mutex::new(RateLimiter::new(
191                rate_limits.messaging_send_qps,
192                rate_limits.messaging_burst,
193            )),
194            mocks,
195            timer_handles: Mutex::new(Vec::new()),
196            secrets: secrets_manager,
197        }))
198    }
199
200    pub fn tenant(&self) -> &str {
201        &self.tenant
202    }
203
204    pub fn config(&self) -> &Arc<HostConfig> {
205        &self.config
206    }
207
208    pub fn main_pack(&self) -> &Arc<PackRuntime> {
209        self.packs
210            .first()
211            .expect("tenant runtime must contain at least one pack")
212    }
213
214    pub fn pack(&self) -> Arc<PackRuntime> {
215        Arc::clone(self.main_pack())
216    }
217
218    pub fn overlays(&self) -> Vec<Arc<PackRuntime>> {
219        self.packs.iter().skip(1).cloned().collect()
220    }
221
222    pub fn engine(&self) -> &Arc<FlowEngine> {
223        &self.engine
224    }
225
226    pub fn state_machine(&self) -> &Arc<StateMachineRuntime> {
227        &self.state_machine
228    }
229
230    pub fn http_client(&self) -> &Client {
231        &self.http_client
232    }
233
234    pub fn digest(&self) -> Option<&str> {
235        self.digests.first().and_then(|d| d.as_deref())
236    }
237
238    pub fn overlay_digests(&self) -> Vec<Option<String>> {
239        self.digests.iter().skip(1).cloned().collect()
240    }
241
242    pub fn required_secrets(&self) -> Vec<SecretRequirement> {
243        self.packs
244            .iter()
245            .flat_map(|pack| pack.required_secrets().iter().cloned())
246            .collect()
247    }
248
249    pub fn missing_secrets(&self) -> Vec<SecretRequirement> {
250        self.packs
251            .iter()
252            .flat_map(|pack| pack.missing_secrets(&self.config.tenant_ctx()))
253            .collect()
254    }
255
256    pub fn telegram_cache(&self) -> &Mutex<LruCache<i64, StatusCode>> {
257        &self.telegram_cache
258    }
259
260    pub fn webhook_cache(&self) -> &Mutex<LruCache<String, Value>> {
261        &self.webhook_cache
262    }
263
264    pub fn messaging_rate(&self) -> &Mutex<RateLimiter> {
265        &self.messaging_rate
266    }
267
268    pub fn mocks(&self) -> Option<&Arc<MockLayer>> {
269        self.mocks.as_ref()
270    }
271
272    pub fn register_timers(&self, handles: Vec<JoinHandle<()>>) {
273        self.timer_handles.lock().extend(handles);
274    }
275
276    pub fn get_secret(&self, key: &str) -> Result<String> {
277        if !self.config.secrets_policy.is_allowed(key) {
278            bail!("secret {key} is not permitted by bindings policy");
279        }
280        let bytes = read_secret_blocking(&self.secrets, key)
281            .context("failed to read secret from manager")?;
282        let value = String::from_utf8(bytes).context("secret value is not valid UTF-8")?;
283        Ok(value)
284    }
285}
286
287impl Drop for TenantRuntime {
288    fn drop(&mut self) {
289        for handle in self.timer_handles.lock().drain(..) {
290            handle.abort();
291        }
292    }
293}
294
295pub struct RateLimiter {
296    allowance: f64,
297    rate: f64,
298    burst: f64,
299    last_check: Instant,
300}
301
302impl RateLimiter {
303    pub fn new(qps: u32, burst: u32) -> Self {
304        let rate = qps.max(1) as f64;
305        let burst = burst.max(1) as f64;
306        Self {
307            allowance: burst,
308            rate,
309            burst,
310            last_check: Instant::now(),
311        }
312    }
313
314    pub fn try_acquire(&mut self) -> bool {
315        let now = Instant::now();
316        let elapsed = now.duration_since(self.last_check).as_secs_f64();
317        self.last_check = now;
318        self.allowance += elapsed * self.rate;
319        if self.allowance > self.burst {
320            self.allowance = self.burst;
321        }
322        if self.allowance < 1.0 {
323            false
324        } else {
325            self.allowance -= 1.0;
326            true
327        }
328    }
329}
greentic_runner_host/runtime.rs

greentic_runner_host/
runtime.rs
