Struct RetriesHook 

pub struct RetriesHook<T, O: Ordering = NoOrder, B: Boundedness = Unbounded, S = OnProcess> { /* private fields */ }

A hook handle controlling an assume_retries operator over a stream of T elements with ordering O (mirroring the type of the stream whose retries are assumed). S is the handle’s scope.

A decision for a retries hook says how many times a buffered element is released — the point where the simulator injects the duplicates the AtLeastOnce type says downstream must tolerate. Because every element admits arbitrarily many retries, the decision space is infinite and this non-determinism can never be explored autonomously: an assume_retries under simulation must be bound to a hook. See hydro_lang::sim::hooks for the decisions offered.

Implementations

impl<T, O: Ordering, Scope: ScriptableHookScope> RetriesHook<T, O, Unbounded, Scope>

pub fn pause(&self)

Available on crate feature sim only.

Declares that buffering at this operator is intended: while paused, the hook is exempt from the missing-decision error, never causes its tick to run, and — if its tick runs anyway because other hooks feed it — contributes its “nothing new” behavior each time. Scripting any decision implicitly resumes the hook.

A pause takes its place in the script like everything else: requested while a decision is still pending, the hold begins once that decision has been consumed.

pub fn resume(&self)

Available on crate feature sim only.

Ends a Self::pause (and clears Self::auto_pause mode).

pub fn auto_pause(&self)

Available on crate feature sim only.

Sets a standing mode where this hook only ever acts when scripted: it holds immediately, and every scripted decision leaves a fresh hold in place behind it.

This deliberately opts out of the forgotten-hook protection: if the test forgets a step, the operator silently holds its data instead of failing. The one auto_pause() line at the top of a test is the reviewer-visible marker that this hook’s timing is entirely script-driven, missed steps and all.

pub async fn pause_while<Fut: Future>(&self, body: Fut) -> Fut::Output

Available on crate feature sim only.

Pauses the hook exactly for the duration of body (resuming even on panic), so a bracketed buffering phase cannot leak a paused hook.

pub fn pause_until( &self, predicate: impl Fn(&OrderingStatus) -> bool + Unpin, ) -> PauseUntilFuture<OrderingStatus, impl Fn(&OrderingStatus) -> bool + Unpin>

Available on crate feature sim only.

Pauses the hook and returns a future that resolves once the hook’s pending-input status satisfies predicate — a synchronization point for scripts where the right decision is not knowable upfront. The status is read on demand at every scheduling point. After the future resolves, the hook is unpaused; the ordinary missing-decision error applies from there on.

pub fn pause_until_count( &self, n: usize, ) -> PauseUntilFuture<OrderingStatus, impl Fn(&OrderingStatus) -> bool + Unpin>

Available on crate feature sim only.

Pauses a top-level retries hook until at least n elements are buffered; see Self::pause_until.

impl<T, Scope: ScriptableHookScope> RetriesHook<T, NoOrder, Unbounded, Scope>

where T: Serialize + DeserializeOwned,

pub fn release(&self, value: T, copies: usize) -> DecisionFuture

Available on crate feature sim only.

Scripts a top-level assume_retries action to release copies copies of the buffered element equal to value, consuming it. This is where the simulator injects the duplicates the AtLeastOnce input type says downstream must tolerate; copies is the total number of deliveries, so release(v, 1) observes v exactly once, with no duplicate.

Any buffered element may be named — the copies’ placement is unobservable on a NoOrder output, so the cardinality is the entire decision.

Panics

Panics immediately if copies is zero: releasing an element zero times would be a loss, which AtLeastOnce does not permit.

impl<T, Scope: ScriptableHookScope> RetriesHook<T, TotalOrder, Unbounded, Scope>

where T: Serialize + DeserializeOwned,

pub fn release(&self, values: impl IntoIterator<Item = T>) -> DecisionFuture

Available on crate feature sim only.

Scripts a top-level assume_retries action to release these values, which must match a prefix of the buffered queue in arrival order, where a value may be repeated back-to-back to simulate retries. This is where the simulator injects the duplicates the AtLeastOnce input type says downstream must tolerate. The decision fires only once it can be honored in full; matched elements before the last value are consumed (a later value can only start once the earlier one’s copies ended).

This is the non-final form: the trailing value’s element stays buffered with its copies left open, so later decisions may release more copies of it — scripting release never over-commits to an element being done. The element remains pending input until a Self::release_final consumes it (or the buffering is declared with the pause family). Elements left un-released by a partial decision likewise stay pending, and downstream tick-input hooks batch independently of how releases are grouped into decisions.

Panics

Panics immediately if values is empty: a decision must release something (use the pause family to hold data on purpose).

pub fn release_final( &self, values: impl IntoIterator<Item = T>, ) -> DecisionFuture

Available on crate feature sim only.

Like Self::release, but final: the trailing value’s element is consumed along with the rest of the matched prefix, declaring that no more copies of it will be observed. With no values, finalizes a front element whose copies were all released by earlier release decisions, delivering nothing more.

pub fn release_next(&self, copies: usize) -> DecisionFuture

Available on crate feature sim only.

Scripts a top-level assume_retries action to release copies copies of the front of the buffered queue, whatever it is, consuming it (waiting for an element to arrive if the queue is empty). Unlike Self::release_final, this does not assert the released value; prefer the value-naming forms, which fail loudly when the script falls out of sync with the program.

Panics

Panics immediately if copies is zero: releasing an element zero times would be a loss, which AtLeastOnce does not permit.

impl<T, O: Ordering, Scope: ScriptableHookScope> RetriesHook<T, O, Bounded, Scope>

where T: Serialize + DeserializeOwned,

pub fn release(&self, values: impl IntoIterator<Item = T>) -> DecisionFuture

Available on crate feature sim only.

Scripts an in-tick assume_retries observation to consume its complete input and release exactly these values. Every value received by the operator during that tick must be released at least once, and only received values may appear.

On a TotalOrder stream the scripted sequence must follow the input order, where a value may repeat only back-to-back (a repeat cannot come after a different value). On a NoOrder stream only the multiset of scripted values matters.

impl<T, O: Ordering, B: Boundedness, C> RetriesHook<T, O, B, OnCluster<C>>

pub fn on(&self, member_id: u32) -> RetriesHook<T, O, B, OnMember<C>>

Selects one cluster member’s instance of the hooked operator, returning an OnMember-scoped handle with the full scripting API.

Each member’s instance is scripted independently: handle.on(0).release(2) scripts member 0’s next batch and says nothing about the other members, each of which needs its own decision (or pause) when it holds buffered input. A member ID outside the cluster’s sizing is reported when the scripting call is awaited.

Trait Implementations

impl<T, O: Ordering, B: Boundedness, S> Clone for RetriesHook<T, O, B, S>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<T, O: Ordering, B: Boundedness, S> Copy for RetriesHook<T, O, B, S>

impl<T, O: Ordering, B: Boundedness, S> Debug for RetriesHook<T, O, B, S>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<T, O: Ordering, B: Boundedness, S: BindableHookScope> SimHook for RetriesHook<T, O, B, S>

where T: Clone + Serialize + DeserializeOwned + PartialEq,

fn create(next_id: &mut dyn FnMut() -> usize) -> Self

Creates every handle in this value, allocating fresh IDs via next_id.