dwara_core/wasm/runner.rs
1//! Plugin runner — the integration layer between the proxy pipeline
2//! and the proxy-wasm host (DW-055).
3//!
4//! The [`PluginRunner`] holds compiled plugin modules keyed by name and
5//! provides per-request methods to run each phase. The proxy pipeline
6//! calls these methods at the appropriate points in the request path
7//! (wired end-to-end by `dataplane::plugin_dispatch`, DW-157). This
8//! module compiles unconditionally in the OSS build — there are no
9//! cargo features for the plugin runtime.
10
11use std::collections::HashMap;
12use std::path::PathBuf;
13use std::sync::Arc;
14
15use super::host::{PhaseResult, PluginInstance, PluginLimits, PluginModule, WasmEngine};
16
17/// Holds compiled plugin modules and runs them per-request.
18///
19/// Created at config publish time from the gateway's `plugins` list.
20/// Cheap to clone (Arc internals). Each route's `plugins` field names
21/// the plugins to run; the proxy pipeline calls the runner's phase
22/// methods with those names.
23#[derive(Clone)]
24pub struct PluginRunner {
25 engine: WasmEngine,
26 modules: Arc<HashMap<String, PluginModule>>,
27}
28
29/// The result of running a phase across all plugins on a route.
30#[derive(Debug)]
31pub enum PhaseOutcome {
32 /// All plugins returned Continue. The request proceeds normally.
33 /// The (possibly modified) headers/body are available from the
34 /// instances.
35 Continue,
36 /// A plugin short-circuited with a local response. The proxy should
37 /// return this response immediately.
38 LocalResponse(super::host::LocalResponse),
39 /// A plugin trapped (out of fuel, memory error, or panic). The
40 /// proxy should return a 500.
41 Trap(String),
42}
43
44/// Per-request plugin state — holds the instances for all plugins on
45/// the route. Created at the start of request processing and dropped
46/// at the end.
47pub struct PluginInstances {
48 instances: Vec<(String, PluginInstance)>,
49}
50
51impl PluginRunner {
52 /// Build a runner from the gateway's `plugins` list, compiling
53 /// each WASM plugin's module. A local plugin loads from its
54 /// `wasm:` path; a registry `source:` plugin (DW-165) has no
55 /// `wasm` field and loads from its resolved, verified artifact
56 /// path in `resolved` (see [`crate::wasm::source`]). Plugins that
57 /// fail to compile are skipped (with a log warning); the gateway
58 /// starts even if a plugin is broken.
59 pub fn new(
60 plugins: &[crate::config::PluginConfig],
61 resolved: &HashMap<String, PathBuf>,
62 ) -> Result<Self, String> {
63 let engine = WasmEngine::new()?;
64 let mut modules = HashMap::new();
65
66 for plugin in plugins {
67 // DW-119: a plugin is either `wasm:` or `native:`. The
68 // runner only compiles WASM plugins; native filters are
69 // handled by the unified plugin chain (plugins domain).
70 let wasm_path: String = if let Some(p) = &plugin.wasm {
71 p.clone()
72 } else if plugin.source.is_some() {
73 // DW-165: registry-sourced plugin — the verified local
74 // artifact path from the resolver.
75 match resolved.get(&plugin.name) {
76 Some(p) => p.display().to_string(),
77 // No resolved path = resolution failed and the
78 // lifecycle already marked the plugin Crashed;
79 // nothing to compile here.
80 None => continue,
81 }
82 } else {
83 continue;
84 };
85 let wasm_bytes = match std::fs::read(&wasm_path) {
86 Ok(bytes) => bytes,
87 Err(e) => {
88 tracing::warn!(
89 plugin = %plugin.name,
90 path = %wasm_path,
91 error = %e,
92 "DW-055: failed to read plugin wasm file; skipping"
93 );
94 continue;
95 }
96 };
97
98 let limits = plugin
99 .limits
100 .as_ref()
101 .map(|l| PluginLimits {
102 fuel: l.fuel.unwrap_or(1_000_000),
103 memory_mb: l.memory_mb.unwrap_or(32),
104 timeout_ms: l.timeout_ms.unwrap_or(100),
105 })
106 .unwrap_or_default();
107
108 let plugin_config = plugin
109 .config
110 .as_ref()
111 .map(|c| c.as_bytes().to_vec())
112 .unwrap_or_default();
113 let vm_config = Vec::new();
114
115 match engine.compile(&wasm_bytes, limits, plugin_config, vm_config) {
116 Ok(module) => {
117 modules.insert(plugin.name.clone(), module);
118 }
119 Err(e) => {
120 tracing::warn!(
121 plugin = %plugin.name,
122 error = %e,
123 "DW-055: failed to compile plugin wasm; skipping"
124 );
125 }
126 }
127 }
128
129 Ok(Self {
130 engine,
131 modules: Arc::new(modules),
132 })
133 }
134
135 /// Create per-request instances for all plugins named on a route.
136 /// Returns `None` when no named plugin has a compiled module (the
137 /// caller treats that as fail-closed unavailability when the route
138 /// does reference WASM plugins — DW-157). A plugin whose module
139 /// fails to instantiate is skipped here with a warn; the caller is
140 /// expected to verify coverage via [`PluginInstances::contains`].
141 pub fn instantiate(&self, plugin_names: &[String]) -> Option<PluginInstances> {
142 if plugin_names.is_empty() {
143 return None;
144 }
145 let mut instances = Vec::new();
146 for name in plugin_names {
147 if let Some(module) = self.modules.get(name) {
148 match module.instantiate(&self.engine) {
149 Ok(inst) => instances.push((name.clone(), inst)),
150 Err(e) => {
151 tracing::warn!(
152 plugin = %name,
153 error = %e,
154 "DW-055: failed to instantiate plugin"
155 );
156 }
157 }
158 }
159 }
160 if instances.is_empty() {
161 None
162 } else {
163 Some(PluginInstances { instances })
164 }
165 }
166
167 /// Whether any plugins are configured.
168 pub fn is_empty(&self) -> bool {
169 self.modules.is_empty()
170 }
171
172 /// Whether a compiled module exists for `name` (DW-157). The
173 /// lifecycle manager uses this at load time to mark plugins whose
174 /// .wasm failed to read or compile as crashed (fail-closed),
175 /// instead of leaving them reading as healthy with no module
176 /// behind them.
177 pub fn has(&self, name: &str) -> bool {
178 self.modules.contains_key(name)
179 }
180}
181
182impl PluginInstances {
183 /// The empty instance set (DW-157): the adapter for chains with no
184 /// WASM plugins (native-only routes) — every per-name dispatch
185 /// passes through unchanged.
186 pub fn empty() -> Self {
187 Self {
188 instances: Vec::new(),
189 }
190 }
191
192 /// Whether an instance exists for `name`. The dataplane's
193 /// fail-closed gate (DW-157) uses this to verify that instantiation
194 /// covered every WASM plugin a route references before the request
195 /// proceeds.
196 pub fn contains(&self, name: &str) -> bool {
197 self.instances.iter().any(|(n, _)| n == name)
198 }
199
200 /// The named plugin's instance (DW-157). The unified-chain adapter
201 /// dispatches per plugin NAME (the chain interleaves native filters
202 /// between WASM entries), so it needs mutable access to exactly one
203 /// instance at a time. `None` when no instance exists for `name`.
204 pub fn instance_mut(&mut self, name: &str) -> Option<&mut PluginInstance> {
205 self.instances
206 .iter_mut()
207 .find(|(n, _)| n == name)
208 .map(|(_, inst)| inst)
209 }
210
211 /// The named plugin's current request header map (DW-167): the
212 /// callout driver reads the resumed plugin's phase payload back
213 /// through the adapter after delivering a callout response.
214 pub fn instance_request_headers(&self, name: &str) -> Option<Vec<(String, String)>> {
215 self.instances
216 .iter()
217 .find(|(n, _)| n == name)
218 .map(|(_, inst)| inst.request_headers().to_vec())
219 }
220
221 /// The named plugin's current response header map (DW-167; see
222 /// [`PluginInstances::instance_request_headers`]).
223 pub fn instance_response_headers(&self, name: &str) -> Option<Vec<(String, String)>> {
224 self.instances
225 .iter()
226 .find(|(n, _)| n == name)
227 .map(|(_, inst)| inst.response_headers().to_vec())
228 }
229
230 /// The named plugin's current request body (DW-167; see
231 /// [`PluginInstances::instance_request_headers`]).
232 pub fn instance_request_body(&self, name: &str) -> Option<Vec<u8>> {
233 self.instances
234 .iter()
235 .find(|(n, _)| n == name)
236 .map(|(_, inst)| inst.request_body().to_vec())
237 }
238
239 /// The named plugin's current response body (DW-167; see
240 /// [`PluginInstances::instance_request_headers`]).
241 pub fn instance_response_body(&self, name: &str) -> Option<Vec<u8>> {
242 self.instances
243 .iter()
244 .find(|(n, _)| n == name)
245 .map(|(_, inst)| inst.response_body().to_vec())
246 }
247
248 /// Run `proxy_on_request_headers` on all instances. Returns the
249 /// outcome and the (possibly modified) headers.
250 pub fn on_request_headers(
251 &mut self,
252 headers: Vec<(String, String)>,
253 ) -> (PhaseOutcome, Vec<(String, String)>) {
254 let mut current_headers = headers;
255 for (_, instance) in &mut self.instances {
256 match instance.on_request_headers(current_headers.clone()) {
257 PhaseResult::Continue => {
258 current_headers = instance.request_headers().to_vec();
259 }
260 PhaseResult::LocalResponse(resp) => {
261 return (PhaseOutcome::LocalResponse(resp), current_headers);
262 }
263 PhaseResult::Trap(e) => {
264 return (PhaseOutcome::Trap(e), current_headers);
265 }
266 PhaseResult::Pause => {
267 return (
268 PhaseOutcome::Trap(
269 "proxy_http_call is not supported on this \
270 execution path (no callout driver)"
271 .to_string(),
272 ),
273 current_headers,
274 );
275 }
276 }
277 }
278 (PhaseOutcome::Continue, current_headers)
279 }
280
281 /// Run `proxy_on_request_body` on all instances. Returns the
282 /// outcome and the (possibly modified) body.
283 pub fn on_request_body(&mut self, body: Vec<u8>) -> (PhaseOutcome, Vec<u8>) {
284 let mut current_body = body;
285 for (_, instance) in &mut self.instances {
286 match instance.on_request_body(current_body.clone()) {
287 PhaseResult::Continue => {
288 current_body = instance.request_body().to_vec();
289 }
290 PhaseResult::LocalResponse(resp) => {
291 return (PhaseOutcome::LocalResponse(resp), current_body);
292 }
293 PhaseResult::Trap(e) => {
294 return (PhaseOutcome::Trap(e), current_body);
295 }
296 PhaseResult::Pause => {
297 return (
298 PhaseOutcome::Trap(
299 "proxy_http_call is not supported on this \
300 execution path (no callout driver)"
301 .to_string(),
302 ),
303 current_body,
304 );
305 }
306 }
307 }
308 (PhaseOutcome::Continue, current_body)
309 }
310
311 /// Run `proxy_on_response_headers` on all instances. Returns the
312 /// outcome and the (possibly modified) headers.
313 pub fn on_response_headers(
314 &mut self,
315 headers: Vec<(String, String)>,
316 ) -> (PhaseOutcome, Vec<(String, String)>) {
317 let mut current_headers = headers;
318 for (_, instance) in &mut self.instances {
319 match instance.on_response_headers(current_headers.clone()) {
320 PhaseResult::Continue => {
321 current_headers = instance.response_headers().to_vec();
322 }
323 PhaseResult::LocalResponse(resp) => {
324 return (PhaseOutcome::LocalResponse(resp), current_headers);
325 }
326 PhaseResult::Trap(e) => {
327 return (PhaseOutcome::Trap(e), current_headers);
328 }
329 PhaseResult::Pause => {
330 return (
331 PhaseOutcome::Trap(
332 "proxy_http_call is not supported on this \
333 execution path (no callout driver)"
334 .to_string(),
335 ),
336 current_headers,
337 );
338 }
339 }
340 }
341 (PhaseOutcome::Continue, current_headers)
342 }
343
344 /// Run `proxy_on_response_body` on all instances. Returns the
345 /// outcome and the (possibly modified) body.
346 pub fn on_response_body(&mut self, body: Vec<u8>) -> (PhaseOutcome, Vec<u8>) {
347 let mut current_body = body;
348 for (_, instance) in &mut self.instances {
349 match instance.on_response_body(current_body.clone()) {
350 PhaseResult::Continue => {
351 current_body = instance.response_body().to_vec();
352 }
353 PhaseResult::LocalResponse(resp) => {
354 return (PhaseOutcome::LocalResponse(resp), current_body);
355 }
356 PhaseResult::Trap(e) => {
357 return (PhaseOutcome::Trap(e), current_body);
358 }
359 PhaseResult::Pause => {
360 return (
361 PhaseOutcome::Trap(
362 "proxy_http_call is not supported on this \
363 execution path (no callout driver)"
364 .to_string(),
365 ),
366 current_body,
367 );
368 }
369 }
370 }
371 (PhaseOutcome::Continue, current_body)
372 }
373
374 /// Call `proxy_on_done` and `proxy_on_log` on all instances (the
375 /// cleanup path).
376 pub fn on_done(&mut self) {
377 for (_, instance) in &mut self.instances {
378 instance.on_done();
379 }
380 }
381}