Slice Blocks

When building distributed applications, you often need to process incoming requests while observing the current state of your system. For example, a counter service needs to respond to "get" requests with the current count, or a key-value store needs to look up values for incoming queries.

The challenge is that in Hydro, live collections update asynchronously. A Stream of requests arrives over time, and a Singleton holding state changes as updates are processed. How do you combine these two asynchronous sources in a meaningful way?

The `sliced!` Macro

Hydro provides the sliced! macro to solve this problem. It allows you to take a slice of multiple live collections at a point in time, process them together, and emit results back into the asynchronous world.

use hydro_lang::prelude::*;

let get_response = sliced! {
    let request_batch = use::batch(get_requests, nondet!(/** batch boundaries are never observed */));
    let count_snapshot = use::snapshot(current_count, nondet!(/** each request reads the count at the time it is processed */));

    let count_ref = count_snapshot.by_ref();
    request_batch.map(q!(|_req| *count_ref))
};

A sliced! block starts with one or more use statements — hooks, with syntax inspired by React hooks — each specifying a live collection to slice and a style controlling how it is sliced. Each hook returns the sliced version of the collection, which is bounded for the duration of the slice. All hooks must appear before the body of the slice, and all collections consumed by hooks must live at the same location.

The body of the slice transforms the bounded collections and returns a live collection (or a tuple of them), which is automatically "unsliced" back into an unbounded collection that continues across slice boundaries (see Returning Values from a Slice).

Batches and Snapshots

The two fundamental hooks reveal a bounded version of a live collection: use::batch(collection, nondet!(...)) reveals a batch of new elements for stream-like collections, and use::snapshot(collection, nondet!(...)) reveals a snapshot of the current value for singleton-like collections.

Input collection	Hook	Revealed as
`Stream`	`use::batch`	Batch of elements that arrived since the last slice
`Singleton`	`use::snapshot`	Snapshot of the current value
`Optional`	`use::snapshot`	Snapshot of the current value (possibly absent)
`KeyedStream`	`use::batch`	Batch of new elements, grouped per key
`KeyedSingleton` (unbounded values)	`use::snapshot`	Snapshot of the current entries
`KeyedSingleton` (`BoundedValue`)	`use::batch`	Batch of newly arrived entries

In all cases, the revealed collection is frozen for the duration of the slice, so you can safely observe it in its entirety (including with reference handles).

note

The style-less use(collection, nondet!(...)) hook, which picked between batching and snapshotting automatically based on the collection type, is deprecated in favor of the explicit use::batch and use::snapshot hooks.

When you slice a Stream, the batch contains all elements that have arrived since the previous slice was processed. The boundaries of these batches are non-deterministic—they depend on network timing, processing speed, and other runtime factors.

In the animation below, elements arrive continuously on the input stream. When a slice is taken, all pending elements are collected into a batch for processing. The batch is then transformed (in this case, each integer is converted to a string), and the results are emitted back to the output stream.

let numbers = process.source_iter(q!(vec![1, 2, 3, 4, 5]));

let stringified = sliced! {
    let batch = use::batch(numbers, nondet!(/** batch boundaries don't affect final result */));
    batch.map(q!(|x| x.to_string()))
};
// Eventually emits: "1", "2", "3", "4", "5" (in batches)

The key insight is that while batch boundaries are non-deterministic, the eventual result is deterministic—all elements will eventually be processed and emitted.

When you slice a Singleton, you receive a snapshot of its current value at the moment the slice is taken. If the singleton is updated between slices, subsequent slices will observe the new value. The animation below shows how a singleton's value changes over time as updates are processed:

A sliced! block can combine any number of live collections, and all hooks are sliced together at the same logical point in time. For example, slicing a Stream alongside a Singleton reveals a batch of stream elements and a snapshot of the singleton's value at the same moment, allowing you to perform joins and lookups across several sources of state:

let requests = process.source_iter(q!(vec![1, 2, 3]));
let scale = process.singleton(q!(10));

let scaled = sliced! {
    let batch = use::batch(requests, nondet!(/** batch boundaries don't affect per-element results */));
    let scale_snapshot = use::snapshot(scale, nondet!(/** the scale is constant, so all snapshots are identical */));
    batch.cross_singleton(scale_snapshot).map(q!(|(x, s)| x * s))
};
// 10, 20, 30

Bounded-Value Keyed Singletons

A KeyedSingleton with the BoundedValue bound gets special treatment. Because each key's value is immutable once it appears, there is no need to re-observe existing entries: use::batch reveals a batch containing only the newly arrived entries, and each entry is revealed in exactly one slice. This makes BoundedValue keyed singletons behave like a stream of request/response entries:

let events: Stream<(&str, i32), _, Unbounded> = process
    .source_iter(q!(vec![("alice", 1), ("bob", 2), ("alice", 3)]))
    .into();
let first_events = events.into_keyed().first(); // KeyedSingleton<&str, i32, _, BoundedValue>

let processed = sliced! {
    let new_entries = use::batch(first_events, nondet!(/** each entry is handled independently, so batching is not observable */));
    new_entries.entries()
};
// ("alice", 1), ("bob", 2) in some order

Guarantees

Although the timing of slices is non-deterministic, hooks provide several guarantees that make it possible to reason about correctness:

Batches partition the input: every element of a stream (or entry of a BoundedValue keyed singleton) appears in exactly one batch, and batches preserve the order of the underlying stream. Concatenating all batches yields the original collection.
Snapshots are monotone: the snapshot revealed in a later slice includes at least all data that contributed to the snapshot in an earlier slice. State never appears to "go backwards" — but a snapshot may lag behind the latest writes, since updates propagate asynchronously.
A single point in time: all hooks in one sliced! block are sliced together, at the same logical point in time.

Because batch boundaries and snapshot timing remain non-deterministic, every use of an external collection requires a nondet! guard explaining why this non-determinism is acceptable. See Non-Determinism and nondet! for how to write these explanations.

Reading State with References

The most common use of a slice is to answer a batch of requests using a snapshot of some state. While you can pair each request with the snapshot using cross_singleton (as above), the more ergonomic idiom is to capture the snapshot with a reference handle (by_ref()) and read it directly inside a q!() closure:

let increments = process.source_iter(q!(vec![(), (), ()]));
let get_requests = process.source_iter(q!(vec!["alice", "bob"]));

let current_count = increments.count();

let get_response = sliced! {
    let request_batch = use::batch(get_requests, nondet!(/** batch boundaries are never observed */));
    let count_snapshot = use::snapshot(current_count, nondet!(/** each request reads the count at the time it is processed */));

    let count_ref = count_snapshot.by_ref();
    request_batch.map(q!(|requester| (requester, *count_ref)))
};

Because the snapshot revealed by the hook is bounded, by_ref() gives you a handle that resolves to a plain &usize at runtime. Reference handles can also mutate state within a slice via by_mut(); see References and Mutations.

Atomic Snapshots: `use::atomic`

By default, a snapshot may lag arbitrarily behind outputs your program has already released, which can violate guarantees like read-after-write consistency. The use::atomic(collection, nondet!(...)) hook strengthens the batching and snapshotting hooks for collections in an atomic context (created with .atomic()): the revealed batch or snapshot is guaranteed to be consistent with respect to the outputs released via end_atomic() on that same atomic context.

let increment_request_processing = increment_requests.atomic();
let current_count = increment_request_processing.clone().count();
let increment_ack = increment_request_processing.end_atomic();

let get_response = sliced! {
    let request_batch = use::batch(get_requests, nondet!(/** we never observe batch boundaries */));
    let count_snapshot = use::atomic(current_count, nondet!(/** atomicity guarantees consistency wrt increments */));
    let count_ref = count_snapshot.by_ref();
    request_batch.map(q!(|_| *count_ref))
};

If a client has received an acknowledgement released by end_atomic(), any later use::atomic snapshot will reflect the acknowledged operation. See Atomic Collections for the full story.

State Hooks: `use::state` and `use::state_null`

Sometimes you need to maintain state across slice iterations—for example, accumulating a running count or buffering elements until a condition is met. State hooks declare collections that are internal to the slice and persist across slice iterations. They are declared with let mut, and the value assigned to the binding at the end of the body is carried over to the next iteration of the slice.

Use use::state(|l| initial) when the state has a known initial value. The closure receives the slice's location and returns the state for the first iteration:

let running_count = sliced! {
    let batch = use::batch(input_stream, nondet!(/** batch boundaries don't affect the final count */));
    let mut counter = use::state(|l| l.singleton(q!(0)));

    // Increment the counter by the number of items in this batch
    let new_count = counter.clone().zip(batch.count())
        .map(q!(|(old, add)| old + add));
    counter = new_count.clone();
    new_count.into_stream()
};

Use use::state_null::<Type>() when the state should start out null: an empty Stream, an absent Optional, and so on. Because there is no initial value to infer the type from, you must annotate it explicitly. A common pattern is buffering elements until a condition is met, such as holding payloads until a leader is elected:

let payloads_with_leader = sliced! {
    let mut unsent_payloads = use::state_null::<Stream<_, _, _, TotalOrder>>();

    let payload_batch = use::batch(payloads, nondet!(/** ... */));
    let latest_leader = use::snapshot(leader_id, nondet!(/** ... */));

    // Combine buffered and new payloads
    let all_payloads = unsent_payloads.chain(payload_batch);

    // If no leader, buffer everything; otherwise clear the buffer
    unsent_payloads = all_payloads.clone().filter_if(latest_leader.clone().is_none());
    all_payloads.cross_singleton(latest_leader)
};

Unlike the other hooks, state hooks do not take a nondet! guard: the state itself is just a value carried between iterations. But because the state evolves according to non-deterministically sliced inputs, code using state hooks deserves the same careful review as the hooks that feed it.

Instead of reassigning the state binding, you can also mutate state in place with mutable references (by_mut), which is often clearer when several inputs read and write the same state.

State Hooks vs. Sliced Singletons

State hooks differ from singletons consumed with use::snapshot in an important way:

Sliced singletons observe external state that is derived deterministically (e.g. by fold) and updates independently of the slice.
State hooks are internal to the slice and hold values you compute between iterations.

Prefer deriving state with deterministic APIs and observing it via use::snapshot when possible; the type system provides stronger guarantees for such state. Reach for state hooks when the update logic fundamentally depends on the slice structure (buffering, batched accumulation, multi-input mutation).

Returning Values from a Slice

The body of a sliced! block returns bounded collections, which are automatically unsliced back into live collections that evolve across slices:

Returned from body	Unsliced result
`Stream` (bounded)	Unbounded `Stream` concatenating the elements from every slice
`Singleton`	Unbounded `Singleton` continually updated to the latest slice's value
`Optional`	Unbounded `Optional` continually updated to the latest slice's value
`KeyedStream` (bounded)	Unbounded `KeyedStream` concatenating each key's elements from every slice
Tuple of the above	Tuple of unsliced collections

A KeyedSingleton cannot be returned directly; convert it with .into_keyed_stream() and return the resulting KeyedStream instead.

To keep an output inside the atomic context associated with the slice (so that downstream consumers can establish consistency guarantees), wrap it with yield_atomic; see Atomic Collections.

Scheduling and Timers

Slices are lazy—a slice only re-runs when there is new input to process on at least one of its used collections. If no new data arrives, the slice will not execute again. This means that if you need a slice to run on a regular schedule (e.g., to periodically emit a heartbeat or poll for changes), you must explicitly provide a time-based input.

The simplest approach is to use sample_every, which produces a stream of periodic snapshots from a live collection. You can directly process this stream, or use this sampled stream inside a sliced! block to combine it with other inputs:

let sampled_state: Stream<...> = state.sample_every(
    q!(Duration::from_secs(1)),
    nondet!(/** sampling timing is non-deterministic */),
);

sampled_state.clone().for_each(q!(|sample| {
    println!("Sampled state: {:?}", sample);
}));

let periodic_output = sliced! {
    let sample_batch = use::batch(sampled_state, nondet!(/** batch boundaries are non-deterministic */));
    let other_state_snapshot = use::snapshot(other_state, nondet!(/** snapshot timing is non-deterministic */));

    sample_batch.cross_singleton(other_state_snapshot)
        .map(q!(|(s1, s2)| process(s1, s2)))
};

Alternatively, you can use source_interval to create a raw stream of timer ticks and use it directly in a slice. However, be careful: because slices can be triggered by any of their inputs, the slice may sometimes run when the interval stream has an empty batch (i.e., no new timer events since the last run). Your slice logic must handle this case, for example by using first() or count() to check whether any interval events are present before acting on them.

let ticks = process.source_interval(
    q!(Duration::from_secs(1)),
    nondet!(/** timer is non-deterministic */),
);

let periodic_output = sliced! {
    let tick_batch = use::batch(ticks, nondet!(/** batch boundaries are non-deterministic */));
    let state = use::snapshot(current_state, nondet!(/** snapshot timing is non-deterministic */));

    // Only emit when there is at least one tick in this batch
    state.filter_if(tick_batch.first().is_some()).into_stream()
};

Using Slices Well

Slices are a powerful tool for manipulating asynchronous data, but should only be used when necessary, since they introduce non-determinism: batch boundaries, snapshot timing, and the interleaving of slices are all decided at runtime. This is why every hook requires a nondet! guard.

Keep slices focused: Each sliced! block should have a clear purpose. If you're doing multiple unrelated operations, consider separate blocks.
Document non-determinism: The explanation in each nondet! call should explain why the non-determinism doesn't affect correctness — see Non-Determinism and nondet! for what makes a good explanation.
Test with simulation: Use exhaustive simulation testing to verify your code handles all possible batch boundaries and snapshot timings correctly.

The sliced! Macro​

Batches and Snapshots​

Bounded-Value Keyed Singletons​

Guarantees​

Reading State with References​

Atomic Snapshots: use::atomic​

State Hooks: use::state and use::state_null​

State Hooks vs. Sliced Singletons​

Returning Values from a Slice​

Scheduling and Timers​

Using Slices Well​