Contracts

1. AsyncStateStore

Durable per-key state. Reference impls: tflo-state-files (async feature; runs sync I/O on spawn_blocking) and tflo-state-s3 (direct async impl over the S3Client trait).

#[async_trait]
pub trait AsyncStateStore: Send + Sync {
    async fn save(&self, key: &[u8], snapshot: &StateSnapshot) -> Result<(), String>;
    async fn load(&self, key: &[u8]) -> Result<Option<StateSnapshot>, String>;
    async fn list_keys(&self) -> Result<Vec<Vec<u8>>, String>;
    async fn save_batch(&self, items: &[(Vec<u8>, StateSnapshot)]) -> Result<(), String>;
    async fn delete(&self, key: &[u8]) -> Result<(), String>;
}

save_batch has a sequential-loop default. Cost-sensitive backends (S3, GCS) must override to use multi-object batched APIs — per-key PUT is the most common production cost amplification. The Arc<dyn AsyncStateStore> blanket impl already exists in core.

2. Cursor

Durable stream position. Pairs with an AsyncCursorStore<C> impl; the pairing is what Checkpointer orchestrates.

pub trait Cursor: Clone + Send + Sync + Debug + 'static {
    fn to_bytes(&self) -> Vec<u8>;
    fn from_bytes(data: &[u8]) -> Result<Self, String>;
    fn display(&self) -> String;
}

Existing impls: KafkaOffset ((topic, partition, offset)) and MqttCursor (last packet id + bounded QoS-2 dedup window + retained-topic sequence map).

3. ShardRouter<K>

Pluggable key → shard ownership. LocalShard owns every key (single-process default). KafkaShardRouter is driven by consumer-group rebalance callbacks.

pub trait ShardRouter<K>: Send + Sync {
    fn owns(&self, key: &K) -> bool;
    fn assignment_epoch(&self) -> u64;
}

Lifecycle hooks (on_assign / on_revoke) are intentionally outside the trait — they're runtime-coupled and live in connector crates. assignment_epoch must be strictly monotonic and bump after ownership changes; that is what fences out events that arrive in the rebalance window for keys you no longer own.

4. Operator

Custom node plugin. tflo-ops and tflo-fintech are both pure Operator plugins — the canonical templates for your own catalog crate.

pub trait Operator: Send + Sync + 'static {
    fn eval(&mut self, inputs: &[Computed], ts: i64) -> NodeOutput;
    fn reset(&mut self) {}
    fn name(&self) -> &str { "operator" }
    fn save(&self) -> Option<Vec<u8>> { None }
    fn load(&mut self, _bytes: &[u8]) -> Result<(), OperatorLoadError> { /* ... */ Ok(()) }
    fn type_id_version(&self) -> u32 { 0 }   // Phase 1 — opt-in versioning
}

pub trait StatelessOperator: Operator {}    // marker for declared intent

Auxiliary contracts

Connector-side traits that live in their crates, not in tflo-core:

Crash-safe orchestration: Checkpointer

tflo_core::state::Checkpointer is the single place that sequences the snapshot → state-store-save → cursor-store-save writes in the correct order. If you're building a new connector that needs checkpointing, use it rather than re-implementing the ordering — getting the snapshot / cursor write order wrong is the most common cause of duplicate processing on restart.