faucet_transform_wasm/
config.rs

1//! Config types for the WASM transform. No I/O or wasmtime here.
2
3use schemars::JsonSchema;
4use serde::{Deserialize, Serialize};
5
6/// Configuration for the `wasm` transform.
7#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
8pub struct WasmTransformConfig {
9    /// Filesystem path to the precompiled `.wasm` module (absolute, or relative
10    /// to the working directory). URLs are not supported in v1.
11    pub module: String,
12    /// Name of the exported transform function. Default `"transform"`.
13    #[serde(default = "default_function")]
14    pub function: String,
15    /// Linear-memory cap for the module, in mebibytes. A record that grows
16    /// memory past this limit fails that record (routed by `on_error`).
17    /// Default `16`. The practical lower bound for ~1 KB JSON records is ~8 MB.
18    #[serde(default = "default_memory_limit_mb")]
19    pub memory_limit_mb: u32,
20    /// wasmtime fuel budget per record — a deterministic CPU bound. A record
21    /// that exhausts its fuel fails (routed by `on_error`). Default
22    /// `10_000_000`. Fuel is the effective CPU limit in v1 (there are no
23    /// blocking host calls, so a wall-clock timeout would be redundant).
24    #[serde(default = "default_fuel_limit")]
25    pub fuel_limit: u64,
26    /// What to do when a record fails inside the module (trap, fuel/memory
27    /// exhaustion, ABI violation, or non-JSON output). Default `fail`.
28    #[serde(default)]
29    pub on_error: WasmOnError,
30    /// Re-stat the module file's mtime before each page; recompile and swap in
31    /// the new module atomically at the page boundary if it changed. In-flight
32    /// records within a page always use one module. Default `false`.
33    #[serde(default)]
34    pub reload_on_change: bool,
35}
36
37/// Policy for a per-record module failure.
38#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
39#[serde(rename_all = "snake_case")]
40pub enum WasmOnError {
41    /// Abort the whole run with a [`faucet_core::FaucetError::Transform`]
42    /// (the default; matches every other transform's fail-fast behaviour).
43    #[default]
44    Fail,
45    /// Drop the failing record and continue (a warning + metric is emitted).
46    Skip,
47    /// Emit the record unchanged and continue (a warning + metric is emitted).
48    Passthrough,
49}
50
51fn default_function() -> String {
52    "transform".to_owned()
53}
54
55fn default_memory_limit_mb() -> u32 {
56    16
57}
58
59fn default_fuel_limit() -> u64 {
60    10_000_000
61}
62
63impl WasmTransformConfig {
64    /// The low-cardinality metric label for this module — the file basename.
65    pub(crate) fn module_label(&self) -> String {
66        std::path::Path::new(&self.module)
67            .file_name()
68            .map(|s| s.to_string_lossy().into_owned())
69            .unwrap_or_else(|| self.module.clone())
70    }
71}
72
73#[cfg(test)]
74mod tests {
75    use super::*;
76    use serde_json::json;
77
78    #[test]
79    fn decodes_minimal_config_with_defaults() {
80        let cfg: WasmTransformConfig =
81            serde_json::from_value(json!({"module": "./t.wasm"})).unwrap();
82        assert_eq!(cfg.module, "./t.wasm");
83        assert_eq!(cfg.function, "transform");
84        assert_eq!(cfg.memory_limit_mb, 16);
85        assert_eq!(cfg.fuel_limit, 10_000_000);
86        assert_eq!(cfg.on_error, WasmOnError::Fail);
87        assert!(!cfg.reload_on_change);
88    }
89
90    #[test]
91    fn decodes_full_config() {
92        let cfg: WasmTransformConfig = serde_json::from_value(json!({
93            "module": "/abs/redact.wasm",
94            "function": "run",
95            "memory_limit_mb": 32,
96            "fuel_limit": 5000,
97            "on_error": "skip",
98            "reload_on_change": true
99        }))
100        .unwrap();
101        assert_eq!(cfg.function, "run");
102        assert_eq!(cfg.memory_limit_mb, 32);
103        assert_eq!(cfg.fuel_limit, 5000);
104        assert_eq!(cfg.on_error, WasmOnError::Skip);
105        assert!(cfg.reload_on_change);
106    }
107
108    #[test]
109    fn rejects_missing_module() {
110        let err = serde_json::from_value::<WasmTransformConfig>(json!({})).unwrap_err();
111        assert!(err.to_string().contains("module"), "{err}");
112    }
113
114    #[test]
115    fn rejects_unknown_on_error() {
116        let err = serde_json::from_value::<WasmTransformConfig>(
117            json!({"module": "t.wasm", "on_error": "explode"}),
118        )
119        .unwrap_err();
120        assert!(
121            err.to_string().to_lowercase().contains("on_error")
122                || err.to_string().contains("explode"),
123            "{err}"
124        );
125    }
126
127    #[test]
128    fn module_label_is_basename() {
129        let cfg: WasmTransformConfig =
130            serde_json::from_value(json!({"module": "/a/b/redact_email.wasm"})).unwrap();
131        assert_eq!(cfg.module_label(), "redact_email.wasm");
132        let bare: WasmTransformConfig =
133            serde_json::from_value(json!({"module": "x.wasm"})).unwrap();
134        assert_eq!(bare.module_label(), "x.wasm");
135    }
136
137    #[test]
138    fn schema_builds_with_module_property() {
139        let schema = schemars::schema_for!(WasmTransformConfig);
140        let json = serde_json::to_value(&schema).unwrap();
141        assert!(
142            json.get("properties")
143                .and_then(|p| p.get("module"))
144                .is_some()
145        );
146    }
147}
faucet_transform_wasm/config.rs

faucet_transform_wasm/
config.rs
