Distributed Streaming

How Fungi runs across multiple nodes.

For how to deploy, see Deployment / Kubernetes.

Topology

                ┌─────────────────┐
                │   JobManager    │
                │  (in server)    │
                └────────┬────────┘
                         │ gRPC
          ┌──────────────┼──────────────┐
          ↓              ↓              ↓
    ┌──────────┐   ┌──────────┐   ┌──────────┐
    │ TaskMgr  │   │ TaskMgr  │   │ TaskMgr  │
    │ Slot 1   │   │ Slot 1   │   │ Slot 1   │
    │ Slot 2   │   │ Slot 2   │   │ Slot 2   │
    └──────────┘   └──────────┘   └──────────┘
          │              │              │
          └──────────────┼──────────────┘
                         ↓
                  Object Storage
                  (checkpoints)

Components

Control Plane (gRPC)

Data Plane (gRPC streaming)

Checkpointing

Barriers flow inline with data:

Source-1 ──[data][barrier][data]──→ Operator
Source-2 ──[data][data][barrier]──→ Operator
                                     ↓
                          Wait for ALL barriers
                          Snapshot state
                          Forward barrier

States: Triggered → WaitingForAck → Committing → Completed

Exactly-Once

Two-phase commit for sinks:

1. Prepare — buffer data
2. Commit — make visible after checkpoint complete
3. Abort — discard if checkpoint fails

Recovery:

1. JobManager detects failure (missed heartbeats)
2. Reload state from last checkpoint
3. Resume from checkpoint position

API

#![allow(unused)]
fn main() {
use fungi_scheduler::{JobGraph, Scheduler, JobVertex};
use fungi_cluster::checkpoint::CheckpointCoordinator;

let mut graph = JobGraph::new("my-job");
graph.add_vertex(JobVertex::new("source", "kafka_source", 2));
graph.add_vertex(JobVertex::new("sink", "kafka_sink", 1));
graph.add_edge("source", "sink", PartitionType::Forward);

let mut scheduler = Scheduler::new();
scheduler.register_task_manager("tm-1", 4, 4096, 4);
scheduler.submit_job(graph)?;
let deployments = scheduler.schedule()?;

let coordinator = CheckpointCoordinator::new(storage, 60_000);
let checkpoint_id = coordinator.trigger_checkpoint("job-1", 3).await;
}

Troubleshooting