hydro_lang/
sim_hooks.rs

1//! Handle types for **simulator hooks**: scripting the decisions of unsafe operators.
2//!
3//! Every unsafe operator (like [`Stream::batch`](crate::live_collections::Stream::batch) or
4//! [`Singleton::snapshot`](crate::live_collections::Singleton::snapshot)) takes a
5//! [`NonDet`](crate::nondet::NonDet) guard. A guard can optionally carry a **hook handle**,
6//! which lets a simulation test take manual control of the non-deterministic decision made
7//! by that operator (which elements form the next batch, which version of a piece of state
8//! a snapshot reveals, ...). See `hydro_lang::sim::hooks` for the test-side scripting API.
9//!
10//! Handles are created from the [`FlowBuilder`](crate::compile::builder::FlowBuilder) via
11//! [`FlowBuilder::sim_hook`](crate::compile::builder::FlowBuilder::sim_hook) *before* the
12//! program under test is constructed, and attached to the operator they control with the
13//! `nondet!(... hook = handle)` syntax. Handles are small and `Copy`: the same value is
14//! passed into the program during construction and used later inside the test body to
15//! script decisions.
16//!
17//! This module contains only the handle types themselves (plain data), so components can
18//! expose hookable signatures (e.g. `nondet_batch: NonDet<Option<BatchHook<u32>>>`, passed
19//! directly to the `batch` operator it controls) without pulling
20//! in any simulator machinery; binding a hook in a flow that is *deployed* rather than
21//! simulated is harmless metadata that non-simulator backends ignore.
22
23use std::hash::Hash;
24use std::marker::PhantomData;
25
26use serde::Serialize;
27use serde::de::DeserializeOwned;
28
29use crate::live_collections::boundedness::{Boundedness, Unbounded};
30use crate::live_collections::stream::{ExactlyOnce, Ordering, Retries, TotalOrder};
31
32/// A simulator hook handle (or a set of them) that can be created in one call to
33/// [`FlowBuilder::sim_hook`](crate::compile::builder::FlowBuilder::sim_hook).
34///
35/// Individual handle types implement this trait, and a struct of handles (a component's
36/// "testing interface") can implement it by creating every field. Fields are typed
37/// `Option<...>` so the struct doubles as a composite hook payload: its [`Default`]
38/// ("no hooks") is what a plain `nondet!(...)` guard carries, while `flow.sim_hook()`
39/// fills in every handle:
40///
41/// ```rust,ignore
42/// #[derive(Clone, Copy, Default)]
43/// pub struct CounterHooks {
44///     pub batch: Option<BatchHook<u32>>,
45///     pub snapshot: Option<SnapshotHook<u64>>,
46/// }
47///
48/// impl SimHook for CounterHooks {
49///     fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
50///         CounterHooks {
51///             batch: SimHook::create(next_id),
52///             snapshot: SimHook::create(next_id),
53///         }
54///     }
55/// }
56/// ```
57///
58/// Such structs nest (a field can itself be a struct of handles), and since handles are
59/// `Copy` a test can pass the struct around or destructure it freely.
60///
61/// Each handle's [`SimHook`] impl carries the trait bounds that the *scripted simulation
62/// codegen* for its operator kind requires (serde round-tripping for decisions, equality
63/// for value-naming decisions, `Hash + Eq + Clone` keys for keyed buffers). Since handles
64/// can only be created through this trait, binding a hook to an operator over unsupported
65/// types fails at the `flow.sim_hook()` call — an ordinary compile error in the test crate
66/// — instead of surfacing as a rustc failure inside the generated simulation dylib.
67pub trait SimHook {
68    /// Creates every handle in this value, allocating fresh IDs via `next_id`.
69    fn create(next_id: &mut dyn FnMut() -> usize) -> Self;
70}
71
72impl<H: SimHook> SimHook for Option<H> {
73    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
74        Some(H::create(next_id))
75    }
76}
77
78/// A hook handle controlling a `batch` operator over a stream of `T` elements with ordering
79/// `O` and retry guarantee `R` (mirroring the type of the stream being batched).
80///
81/// A decision for a batch hook says which buffered elements form the next batch released
82/// into the tick. See `hydro_lang::sim::hooks` for the decisions offered.
83pub struct BatchHook<T, O: Ordering = TotalOrder, R: Retries = ExactlyOnce> {
84    pub(crate) id: usize,
85    pub(crate) _phantom: PhantomData<fn(T, O, R)>,
86}
87
88impl<T, O: Ordering, R: Retries> Clone for BatchHook<T, O, R> {
89    fn clone(&self) -> Self {
90        *self
91    }
92}
93impl<T, O: Ordering, R: Retries> Copy for BatchHook<T, O, R> {}
94
95impl<T, O: Ordering, R: Retries> std::fmt::Debug for BatchHook<T, O, R> {
96    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
97        f.debug_struct("BatchHook").field("id", &self.id).finish()
98    }
99}
100
101impl<T, O: Ordering, R: Retries> SimHook for BatchHook<T, O, R>
102where
103    T: Serialize + DeserializeOwned + PartialEq,
104{
105    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
106        BatchHook {
107            id: next_id(),
108            _phantom: PhantomData,
109        }
110    }
111}
112
113/// A hook handle controlling a `snapshot` operator over a singleton of `T`.
114///
115/// A decision for a snapshot hook picks which buffered version of the state the next tick
116/// execution observes. See `hydro_lang::sim::hooks` for the decisions offered.
117pub struct SnapshotHook<T> {
118    pub(crate) id: usize,
119    pub(crate) _phantom: PhantomData<fn(T)>,
120}
121
122impl<T> Clone for SnapshotHook<T> {
123    fn clone(&self) -> Self {
124        *self
125    }
126}
127impl<T> Copy for SnapshotHook<T> {}
128
129impl<T> std::fmt::Debug for SnapshotHook<T> {
130    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
131        f.debug_struct("SnapshotHook")
132            .field("id", &self.id)
133            .finish()
134    }
135}
136
137impl<T> SimHook for SnapshotHook<T>
138where
139    T: Clone + PartialEq + Serialize + DeserializeOwned,
140{
141    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
142        SnapshotHook {
143            id: next_id(),
144            _phantom: PhantomData,
145        }
146    }
147}
148
149/// A hook handle controlling an `assume_ordering` operator over `T` elements.
150///
151/// A top-level decision selects the next buffered element to release. An `assume_ordering`
152/// inside a tick instead takes one exhaustive ordering of that tick's complete input. See
153/// `hydro_lang::sim::hooks` for the decisions offered.
154pub struct OrderingHook<T, B: Boundedness = Unbounded> {
155    pub(crate) id: usize,
156    pub(crate) _phantom: PhantomData<fn(T, B)>,
157}
158
159impl<T, B: Boundedness> Clone for OrderingHook<T, B> {
160    fn clone(&self) -> Self {
161        *self
162    }
163}
164impl<T, B: Boundedness> Copy for OrderingHook<T, B> {}
165
166impl<T, B: Boundedness> std::fmt::Debug for OrderingHook<T, B> {
167    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
168        f.debug_struct("OrderingHook")
169            .field("id", &self.id)
170            .finish()
171    }
172}
173
174impl<T, B: Boundedness> SimHook for OrderingHook<T, B>
175where
176    T: Serialize + DeserializeOwned + PartialEq,
177{
178    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
179        OrderingHook {
180            id: next_id(),
181            _phantom: PhantomData,
182        }
183    }
184}
185
186/// A hook handle controlling a `batch` operator over a keyed stream with keys `K`, values
187/// `V`, per-key value ordering `O`, and retry guarantee `R` (mirroring the type of the
188/// keyed stream being batched).
189///
190/// A decision for a keyed batch hook says which buffered `(key, value)` entries form the
191/// next batch released into the tick. See `hydro_lang::sim::hooks` for the decisions
192/// offered.
193pub struct KeyedBatchHook<K, V, O: Ordering = TotalOrder, R: Retries = ExactlyOnce> {
194    pub(crate) id: usize,
195    pub(crate) _phantom: PhantomData<fn(K, V, O, R)>,
196}
197
198impl<K, V, O: Ordering, R: Retries> Clone for KeyedBatchHook<K, V, O, R> {
199    fn clone(&self) -> Self {
200        *self
201    }
202}
203impl<K, V, O: Ordering, R: Retries> Copy for KeyedBatchHook<K, V, O, R> {}
204
205impl<K, V, O: Ordering, R: Retries> std::fmt::Debug for KeyedBatchHook<K, V, O, R> {
206    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
207        f.debug_struct("KeyedBatchHook")
208            .field("id", &self.id)
209            .finish()
210    }
211}
212
213impl<K, V, O: Ordering, R: Retries> SimHook for KeyedBatchHook<K, V, O, R>
214where
215    K: Hash + Eq + Clone + Serialize + DeserializeOwned,
216    V: Serialize + DeserializeOwned + PartialEq,
217{
218    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
219        KeyedBatchHook {
220            id: next_id(),
221            _phantom: PhantomData,
222        }
223    }
224}
225
226/// A hook handle controlling a `snapshot` (or `batch`) operator over a keyed singleton
227/// with keys `K` and values `V`.
228///
229/// A decision for a keyed snapshot hook picks which buffered version of each key's state
230/// the next tick execution observes. See `hydro_lang::sim::hooks` for the decisions
231/// offered.
232pub struct KeyedSnapshotHook<K, V> {
233    pub(crate) id: usize,
234    pub(crate) _phantom: PhantomData<fn(K, V)>,
235}
236
237impl<K, V> Clone for KeyedSnapshotHook<K, V> {
238    fn clone(&self) -> Self {
239        *self
240    }
241}
242impl<K, V> Copy for KeyedSnapshotHook<K, V> {}
243
244impl<K, V> std::fmt::Debug for KeyedSnapshotHook<K, V> {
245    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
246        f.debug_struct("KeyedSnapshotHook")
247            .field("id", &self.id)
248            .finish()
249    }
250}
251
252impl<K, V> SimHook for KeyedSnapshotHook<K, V>
253where
254    K: Hash + Eq + Clone + Serialize + DeserializeOwned,
255    V: Clone + PartialEq + Serialize + DeserializeOwned,
256{
257    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
258        KeyedSnapshotHook {
259            id: next_id(),
260            _phantom: PhantomData,
261        }
262    }
263}
264
265/// A hook handle controlling an `assume_ordering` operator over a keyed stream with keys
266/// `K` and values `V`.
267///
268/// A top-level decision selects the next buffered `(key, value)` entry to release. An
269/// `assume_ordering` inside a tick instead takes one exhaustive per-key ordering of that
270/// tick's complete input. See `hydro_lang::sim::hooks` for the decisions offered.
271pub struct KeyedOrderingHook<K, V, B: Boundedness = Unbounded> {
272    pub(crate) id: usize,
273    pub(crate) _phantom: PhantomData<fn(K, V, B)>,
274}
275
276impl<K, V, B: Boundedness> Clone for KeyedOrderingHook<K, V, B> {
277    fn clone(&self) -> Self {
278        *self
279    }
280}
281impl<K, V, B: Boundedness> Copy for KeyedOrderingHook<K, V, B> {}
282
283impl<K, V, B: Boundedness> std::fmt::Debug for KeyedOrderingHook<K, V, B> {
284    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
285        f.debug_struct("KeyedOrderingHook")
286            .field("id", &self.id)
287            .finish()
288    }
289}
290
291impl<K, V, B: Boundedness> SimHook for KeyedOrderingHook<K, V, B>
292where
293    K: Hash + Eq + Clone + Serialize + DeserializeOwned,
294    V: Serialize + DeserializeOwned + PartialEq,
295{
296    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
297        KeyedOrderingHook {
298            id: next_id(),
299            _phantom: PhantomData,
300        }
301    }
302}
303
304/// A hook handle controlling an `entries_partially_ordered` operator over a keyed stream
305/// with keys `K` and values `V`.
306///
307/// The operator preserves the order of values within each key while interleaving across
308/// keys non-deterministically. A top-level decision releases the front entry of one key's
309/// buffer; inside a tick, a single decision supplies the complete interleaving. See
310/// `hydro_lang::sim::hooks` for the decisions offered.
311pub struct PartialOrderingHook<K, V, B: Boundedness = Unbounded> {
312    pub(crate) id: usize,
313    pub(crate) _phantom: PhantomData<fn(K, V, B)>,
314}
315
316impl<K, V, B: Boundedness> Clone for PartialOrderingHook<K, V, B> {
317    fn clone(&self) -> Self {
318        *self
319    }
320}
321impl<K, V, B: Boundedness> Copy for PartialOrderingHook<K, V, B> {}
322
323impl<K, V, B: Boundedness> std::fmt::Debug for PartialOrderingHook<K, V, B> {
324    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
325        f.debug_struct("PartialOrderingHook")
326            .field("id", &self.id)
327            .finish()
328    }
329}
330
331impl<K, V, B: Boundedness> SimHook for PartialOrderingHook<K, V, B>
332where
333    K: Hash + Eq + Clone + Serialize + DeserializeOwned,
334    V: Serialize + DeserializeOwned + PartialEq,
335{
336    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
337        PartialOrderingHook {
338            id: next_id(),
339            _phantom: PhantomData,
340        }
341    }
342}
343
344/// A hook handle controlling a `merge_ordered` operator over streams of `T` elements.
345///
346/// The operator preserves the order of each input while interleaving the two inputs
347/// non-deterministically. A top-level decision releases the front element of one input's
348/// buffer; inside a tick, a single decision supplies the complete interleaving. See
349/// `hydro_lang::sim::hooks` for the decisions offered.
350pub struct MergeOrderedHook<T, B: Boundedness = Unbounded> {
351    pub(crate) id: usize,
352    pub(crate) _phantom: PhantomData<fn(T, B)>,
353}
354
355impl<T, B: Boundedness> Clone for MergeOrderedHook<T, B> {
356    fn clone(&self) -> Self {
357        *self
358    }
359}
360impl<T, B: Boundedness> Copy for MergeOrderedHook<T, B> {}
361
362impl<T, B: Boundedness> std::fmt::Debug for MergeOrderedHook<T, B> {
363    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
364        f.debug_struct("MergeOrderedHook")
365            .field("id", &self.id)
366            .finish()
367    }
368}
369
370impl<T, B: Boundedness> SimHook for MergeOrderedHook<T, B>
371where
372    T: Serialize + DeserializeOwned + PartialEq,
373{
374    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
375        MergeOrderedHook {
376            id: next_id(),
377            _phantom: PhantomData,
378        }
379    }
380}
381
382/// A hook handle controlling a `merge_ordered` operator over keyed streams with keys `K`
383/// and values `V`.
384///
385/// The operator preserves each input's order within every key while interleaving the two
386/// inputs non-deterministically (cross-key order is unconstrained). A top-level decision
387/// releases the front entry of one key's buffer in one input; inside a tick, a single
388/// decision supplies the complete interleaving. See `hydro_lang::sim::hooks` for the
389/// decisions offered.
390pub struct KeyedMergeOrderedHook<K, V, B: Boundedness = Unbounded> {
391    pub(crate) id: usize,
392    pub(crate) _phantom: PhantomData<fn(K, V, B)>,
393}
394
395impl<K, V, B: Boundedness> Clone for KeyedMergeOrderedHook<K, V, B> {
396    fn clone(&self) -> Self {
397        *self
398    }
399}
400impl<K, V, B: Boundedness> Copy for KeyedMergeOrderedHook<K, V, B> {}
401
402impl<K, V, B: Boundedness> std::fmt::Debug for KeyedMergeOrderedHook<K, V, B> {
403    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
404        f.debug_struct("KeyedMergeOrderedHook")
405            .field("id", &self.id)
406            .finish()
407    }
408}
409
410impl<K, V, B: Boundedness> SimHook for KeyedMergeOrderedHook<K, V, B>
411where
412    K: Hash + Eq + Clone + Serialize + DeserializeOwned,
413    V: Serialize + DeserializeOwned + PartialEq,
414{
415    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
416        KeyedMergeOrderedHook {
417            id: next_id(),
418            _phantom: PhantomData,
419        }
420    }
421}
hydro_lang/sim_hooks.rs

hydro_lang/
sim_hooks.rs
