submilli_engine/runtime/
effects.rs

1//! Record native outcomes before exposing them to guest code.
2//!
3//! Argument conversion (including guest callbacks) runs before `execute`; result
4//! conversion runs afterwards. The history port owns ordering and persistence.
5//! An absent history keeps ordinary execution direct. No Wasm handles are encoded.
6use std::io::Write;
7
8use futures::future::BoxFuture;
9use serde::{Deserialize, Serialize, de::DeserializeOwned};
10use sha2::{Digest, Sha256};
11use wasmtime::Caller;
12
13use super::StoreData;
14use super::durable::EffectId;
15use super::fuel;
16use super::host::{self, fatal_host_error};
17
18/// Independent of the filesystem journal's much smaller metadata limit.
19pub const MAX_OUTCOME_BYTES: usize = 64 * 1024 * 1024;
20pub const CODEC_VERSION: u32 = 4;
21
22#[derive(Debug, thiserror::Error)]
23#[error("durable execution: {0}")]
24pub struct HistoryError(pub String);
25
26/// A call's identity is allocated by the history implementation, not the interpreter.
27pub struct EffectCall {
28    pub id: EffectId,
29    pub sequence: u64,
30    pub recorded: Option<RecordedOutcome>,
31}
32
33pub type RecordedOutcome = Result<OutcomeBytes, NativeError>;
34
35/// Encoded bytes use MessagePack's binary representation, never an integer array.
36#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
37#[serde(transparent)]
38pub struct OutcomeBytes(#[serde(with = "binary")] pub Vec<u8>);
39
40/// Implementations must serialize calls and commit before `complete` returns.
41/// A missing outcome is an instruction to execute, regardless of idempotency.
42#[async_trait::async_trait]
43pub trait EffectHistory: Send + Sync {
44    async fn begin(
45        &self,
46        operation: &str,
47        fingerprint: [u8; 32],
48    ) -> Result<EffectCall, HistoryError>;
49    async fn complete(
50        &self,
51        call: &EffectCall,
52        outcome: RecordedOutcome,
53    ) -> Result<(), HistoryError>;
54    /// Additional persistence work, reported once and settled by the host boundary.
55    /// Storage must not reject a completed native outcome because fuel ran out.
56    fn take_storage_fuel(&self) -> u64 {
57        0
58    }
59    /// Finish owned persistence I/O before releasing execution resources.
60    async fn drain(&self) -> Result<(), HistoryError> {
61        Ok(())
62    }
63    /// Verify that no effect is pending and all recorded calls have been consumed.
64    async fn ensure_history_consumed(&self) -> Result<(), HistoryError>;
65}
66
67/// Execute one native action, or return its committed native outcome.
68///
69/// `arguments` are hashed in memory only. The action owns them, avoiding copies
70/// of large requests. It must not call guest code or manufacture guest exceptions:
71/// return native values/errors and convert them after this boundary instead.
72pub async fn execute<'store, A, T, F>(
73    caller: &mut Caller<'store, StoreData>,
74    operation: &'static str,
75    arguments: A,
76    action: F,
77) -> wasmtime::Result<T>
78where
79    A: Serialize + Send + 'static,
80    T: Serialize + DeserializeOwned + Send + 'static,
81    F: for<'call> FnOnce(
82            &'call mut Caller<'store, StoreData>,
83            EffectId,
84            A,
85        ) -> BoxFuture<'call, wasmtime::Result<T>>
86        + Send,
87{
88    let Some(history) = caller.data().effects.clone() else {
89        let result = action(caller, EffectId::stub()?, arguments).await;
90        return result.map_err(|error| host::throw_host_error(caller, error));
91    };
92    let fingerprint = fingerprint(caller, &arguments)?;
93    let call = history.begin(operation, fingerprint).await;
94    fuel::settle_host_fuel(&mut *caller, history.take_storage_fuel())?;
95    let call = call.map_err(fatal_host_error)?;
96    if let Some(recorded) = &call.recorded {
97        return match recorded {
98            Ok(bytes) => {
99                fuel::charge(&mut *caller, fuel::PARSE, bytes.0.len() as u64)?;
100                decode(&bytes.0)
101            }
102            Err(error) => Err(host::throw_host_error(caller, error.clone().into_error())),
103        };
104    }
105
106    let result = action(caller, call.id.clone(), arguments).await;
107    if let Err(error) = &result
108        && (host::ends_the_run(error) || error.is::<wasmtime::ThrownException>())
109    {
110        return result;
111    }
112    let recorded_result = match &result {
113        Ok(value) => {
114            let encoded = encode(value)?;
115            // Do not commit a representation the replay decoder refuses structurally.
116            validate_outcome(&encoded)?;
117            fuel::settle(&mut *caller, fuel::COPY, encoded.len() as u64)?;
118            Ok(OutcomeBytes(encoded))
119        }
120        Err(error) => Err(NativeError::capture(error)),
121    };
122    let completed = history.complete(&call, recorded_result).await;
123    fuel::settle_host_fuel(&mut *caller, history.take_storage_fuel())?;
124    completed.map_err(fatal_host_error)?;
125    result.map_err(|error| host::throw_host_error(caller, error))
126}
127
128pub(crate) mod binary {
129    use serde::{Deserializer, Serializer};
130    pub fn serialize<S: Serializer>(bytes: &[u8], serializer: S) -> Result<S::Ok, S::Error> {
131        serializer.serialize_bytes(bytes)
132    }
133    pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Vec<u8>, D::Error> {
134        struct Bytes;
135        impl<'de> serde::de::Visitor<'de> for Bytes {
136            type Value = Vec<u8>;
137            fn expecting(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
138                f.write_str("binary outcome")
139            }
140            fn visit_bytes<E: serde::de::Error>(self, bytes: &[u8]) -> Result<Vec<u8>, E> {
141                Ok(bytes.to_vec())
142            }
143        }
144        deserializer.deserialize_bytes(Bytes)
145    }
146}
147
148pub(crate) fn fingerprint<A: Serialize>(
149    caller: &mut Caller<'_, StoreData>,
150    arguments: &A,
151) -> wasmtime::Result<[u8; 32]> {
152    let mut writer = FingerprintWriter {
153        caller,
154        hash: Sha256::new(),
155        failure: None,
156    };
157    let result =
158        arguments.serialize(&mut rmp_serde::Serializer::new(&mut writer).with_struct_map());
159    if let Some(error) = writer.failure {
160        return Err(error);
161    }
162    result.map_err(fatal_host_error)?;
163    Ok(writer.hash.finalize().into())
164}
165
166struct FingerprintWriter<'a, 'store> {
167    caller: &'a mut Caller<'store, StoreData>,
168    hash: Sha256,
169    failure: Option<wasmtime::Error>,
170}
171impl Write for FingerprintWriter<'_, '_> {
172    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
173        if let Err(error) = fuel::charge(&mut *self.caller, fuel::HASH, bytes.len() as u64) {
174            self.failure = Some(error);
175            return Err(std::io::Error::other("effect fingerprint exceeded fuel"));
176        }
177        self.hash.update(bytes);
178        Ok(bytes.len())
179    }
180    fn flush(&mut self) -> std::io::Result<()> {
181        Ok(())
182    }
183}
184
185fn encode<T: Serialize>(value: &T) -> wasmtime::Result<Vec<u8>> {
186    let mut writer = OutcomeWriter { bytes: Vec::new() };
187    value
188        .serialize(&mut rmp_serde::Serializer::new(&mut writer).with_struct_map())
189        .map_err(fatal_host_error)?;
190    Ok(writer.bytes)
191}
192struct OutcomeWriter {
193    bytes: Vec<u8>,
194}
195impl Write for OutcomeWriter {
196    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
197        let length = self
198            .bytes
199            .len()
200            .checked_add(bytes.len())
201            .filter(|n| *n <= MAX_OUTCOME_BYTES)
202            .ok_or_else(|| std::io::Error::other("native outcome exceeds history limit"))?;
203        if length > self.bytes.capacity() {
204            let capacity = length
205                .max(self.bytes.capacity().saturating_mul(2))
206                .min(MAX_OUTCOME_BYTES);
207            self.bytes
208                .try_reserve_exact(capacity - self.bytes.len())
209                .map_err(std::io::Error::other)?;
210        }
211        self.bytes.extend_from_slice(bytes);
212        Ok(bytes.len())
213    }
214    fn flush(&mut self) -> std::io::Result<()> {
215        Ok(())
216    }
217}
218fn validate_outcome(bytes: &[u8]) -> wasmtime::Result<()> {
219    if bytes.len() > MAX_OUTCOME_BYTES {
220        return Err(fatal_host_error("native outcome exceeds history limit"));
221    }
222    // Validate lengths, nesting and complete consumption before deserialization.
223    super::effects_memory::decoded_heap_bound(bytes, 1, false)?;
224    Ok(())
225}
226
227fn decode<T: DeserializeOwned>(bytes: &[u8]) -> wasmtime::Result<T> {
228    validate_outcome(bytes)?;
229    let mut decoder = rmp_serde::Deserializer::new(std::io::Cursor::new(bytes));
230    decoder.set_max_depth(128);
231    let value = T::deserialize(&mut decoder).map_err(fatal_host_error)?;
232    if decoder.position() != bytes.len() as u64 {
233        return Err(fatal_host_error("trailing native outcome bytes"));
234    }
235    Ok(value)
236}
237
238/// Guest-visible error data, captured before the host wrapper raises an exception.
239#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
240pub enum NativeError {
241    Error(String),
242    Range(String),
243    QuotaExceeded(String),
244    Timeout(String),
245    Type(String),
246    Syntax(String),
247    Uri(String),
248    PermissionDenied(host::PermissionDenied),
249}
250impl NativeError {
251    pub(crate) fn capture(error: &wasmtime::Error) -> Self {
252        let message = error.to_string();
253        if let Some(denial) = error.downcast_ref::<host::PermissionDenied>() {
254            return Self::PermissionDenied(denial.clone());
255        }
256        if error.is::<host::RangeError>() {
257            return Self::Range(message);
258        }
259        if error.is::<host::QuotaExceededError>() {
260            return Self::QuotaExceeded(message);
261        }
262        if error.is::<host::TimeoutError>() {
263            return Self::Timeout(message);
264        }
265        if error.is::<host::TypeError>() {
266            return Self::Type(message);
267        }
268        if error.is::<host::SyntaxError>() {
269            return Self::Syntax(message);
270        }
271        if error.is::<host::UriError>() {
272            return Self::Uri(message);
273        }
274        Self::Error(message)
275    }
276    pub(crate) fn into_error(self) -> wasmtime::Error {
277        match self {
278            Self::Error(message) => wasmtime::Error::msg(message),
279            Self::Range(message) => wasmtime::Error::new(host::RangeError(message)),
280            Self::QuotaExceeded(message) => wasmtime::Error::new(host::QuotaExceededError(message)),
281            Self::Timeout(message) => wasmtime::Error::new(host::TimeoutError(message)),
282            Self::Type(message) => wasmtime::Error::new(host::TypeError(message)),
283            Self::Syntax(message) => wasmtime::Error::new(host::SyntaxError(message)),
284            Self::Uri(message) => wasmtime::Error::new(host::UriError(message)),
285            Self::PermissionDenied(denial) => wasmtime::Error::new(denial),
286        }
287    }
288}
289
290/// Validate a stored MessagePack envelope before Serde sees allocation hints.
291/// The caller must bound the blob size before validating the encoded structure.
292/// `element_bytes` is the largest sequence element in the persistence schema.
293pub fn stored_heap_bound(
294    bytes: &[u8],
295    element_bytes: usize,
296    dynamic_maps: bool,
297) -> Result<u64, HistoryError> {
298    super::effects_memory::decoded_heap_bound(bytes, element_bytes, dynamic_maps)
299        .map_err(|error| HistoryError(error.to_string()))
300}
301
302#[cfg(test)]
303mod tests {
304    use super::*;
305    #[test]
306    fn native_floats_keep_nan_bits_and_signed_zero() {
307        for bits in [
308            0x8000_0000_0000_0000,
309            0x7ff8_0000_0000_1234,
310            0x7ff0_0000_0000_0000,
311        ] {
312            let encoded = encode(&f64::from_bits(bits)).unwrap();
313            let restored: f64 = rmp_serde::from_slice(&encoded).unwrap();
314            assert_eq!(restored.to_bits(), bits);
315        }
316    }
317    #[test]
318    fn binary_is_not_expanded_into_integer_elements() {
319        let value = OutcomeBytes(vec![255; 4096]);
320        let encoded = encode(&value).unwrap();
321        assert!(encoded.len() < 4100);
322        let restored: OutcomeBytes = rmp_serde::from_slice(&encoded).unwrap();
323        assert_eq!(restored, value);
324        drop(encoded);
325    }
326    #[test]
327    fn missing_and_empty_native_strings_remain_distinct() {
328        for value in [None, Some(String::new()), Some("text".to_owned())] {
329            let encoded = encode(&value).unwrap();
330            let restored: Option<String> = rmp_serde::from_slice(&encoded).unwrap();
331            assert_eq!(restored, value);
332        }
333    }
334}
submilli_engine/runtime/effects.rs

submilli_engine/runtime/
effects.rs
