StreamTask

Consume an unbounded stream — flush per window, run until cancelled or exhausted, resume from a cursor.

A StreamTask consumes an impl Stream continuously: it pulls one item at a time, processes each into an output, and flushes per tumbling window — so memory stays bounded and downstream sees progress before the source ends. It runs until the stream is exhausted, until a window asks to stop, or until the run is cancelled; each flushed window commits a cursor so a crashed or cancelled run resumes where it left off. It is one of the Task family of processing models, alongside RouterTask, PollTask, TimerTask, BatchTask, and SteppedTask, and it reads typed dependencies from Resources like the rest. New to Cano? Read Workflows and Resources first; for the cursor-persistence half, Recovery.

use cano::prelude::*;
use futures_util::{Stream, stream};
use std::pin::Pin;

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
enum Step { Consume, Done }

struct ConsumeEvents;

#[task::stream(state = Step)]
impl ConsumeEvents {
    fn window(&self) -> StreamWindow {
        StreamWindow::Count(3) // flush every 3 processed items
    }

    async fn open(&self, _res: &Resources, cursor: Option<u64>)
        -> Result<Pin<Box<dyn Stream<Item = u64> + Send>>, CanoError>
    {
        // On a fresh run `cursor` is `None`; on resume it's the last committed offset.
        let start = cursor.map(|c| c + 1).unwrap_or(0);
        Ok(Box::pin(stream::iter(start..10)) as Pin<Box<dyn Stream<Item = u64> + Send>>)
    }

    async fn process_item(&self, _res: &Resources, item: u64)
        -> Result<(u64, u64), CanoError>
    {
        Ok((item * 2, item)) // (output, cursor reached by consuming this item)
    }

    async fn flush_window(&self, _res: &Resources, outputs: Vec<u64>)
        -> Result<WindowSignal<Step>, CanoError>
    {
        // Commit side effects / offsets for this window here.
        println!("flushed window of {} outputs", outputs.len());
        Ok(WindowSignal::Continue)
    }

    async fn on_close(&self, _res: &Resources, reason: CloseReason)
        -> Result<TaskResult<Step>, CanoError>
    {
        println!("stream ended ({reason:?})");
        Ok(TaskResult::Single(Step::Done))
    }
}

let workflow = Workflow::bare()
    .register_stream(Step::Consume, ConsumeEvents)
    .add_exit_state(Step::Done);

#BatchTask vs StreamTask

The two look similar — both fan a sub-operation over many items — but they solve opposite problems. A BatchTask loads a bounded Vec, processes all of it, and aggregates once at the end: O(N) memory, one emission, and it requires the data to end. A StreamTask is for unbounded / continuous sources: it emits incrementally per window, keeps memory bounded, runs until cancelled/exhausted, and resumes from a cursor.

#How the Stream Loop Works

A StreamTask has three associated types — type Item (one element pulled from the source), type Output (the per-item result accumulated into a window), and type Cursor (the resumable position; Serialize + DeserializeOwned + Send + Sync + 'static) — and four required methods:

Optional methods carry defaults: window() (defaults to StreamWindow::Count(1) — flush per item), on_item_error() (defaults to StreamErrorPolicy::FailFast), config() (defaults to TaskConfig::minimal() — no outer retry, because an outer retry would re-invoke open() and re-consume the stream), and name() (defaults to the type name).

#Quick Start with #[task::stream]

Attach #[task::stream(state = MyState)] to an inherent impl block. The macro infers Item from process_item's owned item parameter and Output / Cursor from the Ok tuple of its return type, injects default window / on_item_error / config / name if absent, synthesises the impl StreamTask<MyState> for ConsumeEvents header, and emits a companion impl Task<MyState> for ConsumeEvents whose run drives the in-memory loop — useful if you register it with plain register and don't want persistence.

use cano::prelude::*;
use futures_util::{Stream, stream};
use std::pin::Pin;

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
enum Step { Consume, Done }

#[derive(Debug, Clone)]
struct Event { offset: u64, payload: String }

#[derive(Debug)]
struct Processed { offset: u64, bytes: usize }

struct ConsumeEvents;

#[task::stream(state = Step)]
impl ConsumeEvents {
    fn window(&self) -> StreamWindow {
        StreamWindow::Count(3)
    }

    async fn open(&self, _res: &Resources, cursor: Option<u64>)
        -> Result<Pin<Box<dyn Stream<Item = Event> + Send>>, CanoError>
    {
        let start = cursor.map(|c| c + 1).unwrap_or(0);
        let events: Vec<Event> = (start..10)
            .map(|offset| Event { offset, payload: format!("payload-{offset}") })
            .collect();
        Ok(Box::pin(stream::iter(events)) as Pin<Box<dyn Stream<Item = Event> + Send>>)
    }

    async fn process_item(&self, _res: &Resources, item: Event)
        -> Result<(Processed, u64), CanoError>
    {
        Ok((
            Processed { offset: item.offset, bytes: item.payload.len() },
            item.offset, // the cursor reached by consuming this item
        ))
    }

    async fn flush_window(&self, _res: &Resources, outputs: Vec<Processed>)
        -> Result<WindowSignal<Step>, CanoError>
    {
        let bytes: usize = outputs.iter().map(|p| p.bytes).sum();
        println!("flushed {} events ({bytes} bytes)", outputs.len());
        Ok(WindowSignal::Continue)
    }

    async fn on_close(&self, _res: &Resources, reason: CloseReason)
        -> Result<TaskResult<Step>, CanoError>
    {
        println!("stream ended ({reason:?})");
        Ok(TaskResult::Single(Step::Done))
    }
}

#Windowing: StreamWindow

The window() method returns a tumbling-window trigger that controls how often flush_window fires and how much the driver buffers. Larger windows amortise the flush + checkpoint cost.

StreamBatch is an alias for StreamWindow — it is semantically identical but emphasises the batching semantics of Count(n) windows.

#Per-Item Error Policy

on_item_error() returns a StreamErrorPolicy deciding what the windowed loop does when process_item returns an Err:

#Cursor Persistence & Resume

Register a stream with Workflow::register_stream(state, task) — the durable, cancellable engine path. When a CheckpointStore plus a workflow id are attached (with_checkpoint_store + with_workflow_id), the engine persists the cursor returned by the last item of each flushed window as a CheckpointRow whose kind is RowKind::StepCursor (the cursor is serde_json-encoded). On Workflow::resume_from(workflow_id, token), the latest persisted cursor is rehydrated and passed to open(Some(cursor)) — so a crashed or cancelled run re-opens the source from the last committed position instead of starting over. Only fully flushed windows commit a cursor; a window that errors mid-flush commits nothing.

use cano::prelude::*;
use cano::RedbCheckpointStore;            // requires the `recovery` feature
use std::sync::Arc;

let checkpoint_store = RedbCheckpointStore::new("/var/lib/myapp/checkpoints.redb")?;
let workflow = Workflow::new(resources)
    .with_checkpoint_store(Arc::new(checkpoint_store))
    .with_workflow_id("event-consumer")
    .register_stream(Step::Consume, ConsumeEvents)   // cursor persisted per flushed window
    .add_exit_state(Step::Done);

// A crashed or cancelled run resumes from the last committed window:
// let result = workflow.resume_from("event-consumer", token).await?;
// open() is called with Some(cursor) — the position of the last fully-flushed window.

#The Cancellation Contract

Because an unbounded stream never ends on its own, cancellation is how you stop it cleanly. When the run's CancellationToken fires, the engine-driven driver performs a cooperative drain:

1. it flushes the in-flight (partial) window via flush_window;
2. it commits that window's cursor;
3. it calls on_close(CloseReason::Cancelled) for cleanup — its returned state is ignored;
4. the run ends as CanoError::Cancelled (category "cancelled").

The window boundary is not the only cancellable point: open() races the same effective token as the loop. A source that hangs while connecting — a broker that never hands back a consumer — is cancellable and timeoutable too, not just the gaps between windows. No cursor exists yet at that point, so there is nothing to flush or commit; the pending open future is simply dropped.

Workflow::with_total_timeout is honoured for Stream states, and it is honoured through that same drain. Instead of wrapping the state in a deadline that would drop the future mid-window, the engine folds the budget into the token the loop already watches — so a deadline trip still flushes the in-flight window and still commits its cursor, exactly like a real cancel. Only the error at the end differs:

#The Idempotency Contract

#register vs register_stream

A StreamTask can be registered two ways, and the difference is durability:

Plain register runs the macro-generated companion Task, which drives an in-memory windowed loop with no cursor persistence and no cancellation — it cannot be cancelled once running, and ends only via source exhaustion, a WindowSignal::Stop, or an error. Reach for register_stream for anything real; it is the path that observes the CancellationToken and persists cursors. register_stream also wraps on_close in panic safety, so a panic in cleanup becomes a CanoError instead of unwinding past the FSM — register does not do this.

#Explicit Trait-Impl Form

Prefer writing the trait header yourself — e.g. to name the associated types, or for a generic impl? Put a bare #[task::stream] on an impl StreamTask<...> for ... block and declare the three type lines (Item / Output / Cursor) yourself. The companion impl Task is still emitted.

use cano::prelude::*;
use futures_util::{Stream, stream};
use std::pin::Pin;

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
enum Step { Consume, Done }

struct Collector;

#[task::stream]
impl StreamTask<Step> for Collector {
    type Item = u32;
    type Output = u32;
    type Cursor = u64;

    fn window(&self) -> StreamWindow {
        StreamWindow::Count(2)
    }

    async fn open(&self, _res: &Resources, _cursor: Option<u64>)
        -> Result<Pin<Box<dyn Stream<Item = u32> + Send>>, CanoError>
    {
        Ok(Box::pin(stream::iter(vec![10u32, 20, 30, 40, 50]))
            as Pin<Box<dyn Stream<Item = u32> + Send>>)
    }

    async fn process_item(&self, _res: &Resources, item: u32)
        -> Result<(u32, u64), CanoError>
    {
        Ok((item * 2, item as u64))
    }

    async fn flush_window(&self, _res: &Resources, _outputs: Vec<u32>)
        -> Result<WindowSignal<Step>, CanoError>
    {
        Ok(WindowSignal::Continue)
    }

    async fn on_close(&self, _res: &Resources, _reason: CloseReason)
        -> Result<TaskResult<Step>, CanoError>
    {
        Ok(TaskResult::Single(Step::Done))
    }
}

#When to Use StreamTask

Reach for a StreamTask when:

• your source is unbounded or continuous — a Kafka topic, an SSE feed, a tailed file, a WebSocket — and never produces a final Vec to aggregate;
• you want incremental, per-window emission with bounded memory rather than one end-of-batch aggregate;
• you need the consumer to resume from a committed offset after a crash or a cancellation, not start over.

If your data is a bounded collection you want to map over and aggregate once, a BatchTask is simpler. If you have a long iterative job over a finite range that you want to crash-resume mid-loop, a SteppedTask fits better.