Explore if we can have a single binary + a cluster/microservice mode, similar to VictoriaMetrics

Right now, Controller, Arroyo, and QueryEngineRust are built and deployed separately. This is great for scale out, but difficult to develop and deploy by a beginner user. Similar to VictoriaMetrics, we can think of both a single-node single-binary application, and a cluster version that has different components that can be scaled independently.

Originally by @linear[bot] on 2026-01-20:

PRO-367

@milindsrivastava1997

Originally by @milindsrivastava1997 on 2026-01-20:

Unified Binary Architecture: Single-Node and Cluster Deployment

Goal

Create a unified binary (asap) that can run in two deployment modes:

Single-Node Mode: All components (query engine + streaming engine) run in a single process with in-memory communication
Cluster Mode: Components run as independent, horizontally-scalable microservices communicating over the network (Kafka/gRPC)

Additionally, the architecture should allow swapping the streaming engine (currently Arroyo) for a custom mini-batch engine in the future without modifying core query engine logic.

Analogy: VictoriaMetrics Architecture

VictoriaMetrics implements this exact pattern and serves as our reference architecture.

VictoriaMetrics Structure

Mode	Components	Communication
Single-Node	One `victoria-metrics` binary	Direct function calls, shared memory
Cluster	Separate `vminsert`, `vmselect`, `vmstorage` binaries	Network RPC (gRPC/HTTP)

Key Design Principles from VictoriaMetrics

1. Core Logic is Deployment-Agnostic

The storage engine (lib/storage) has no knowledge of whether it's running in single-node or cluster mode. It exposes the same API regardless:

// Same interface used in both modes
storage.SetMinScrapeIntervalForDeduplication(duration)
storage.SetBigMergeWorkersCount(count)

2. Network Layer is a Thin Wrapper

In cluster mode, the network layer simply wraps the same core operations with RPC serialization. The core logic remains unchanged.

3. Unified Configuration

Both modes use identical configuration patterns (flags + environment variables). The storage layer receives the same configuration values regardless of how they were provided or which mode is running.

4. ~90-95% Code Reuse

The vast majority of code lives in shared libraries (lib/). Only the thin orchestration and networking layers differ between modes.

Current State: QueryEngineRust + Arroyo

┌─────────────────────┐         ┌─────────────────────┐
│   QueryEngineRust   │         │       Arroyo        │
│   (query serving)   │◄─Kafka──│  (stream processing)│
└─────────────────────┘         └─────────────────────┘
        │                                │
        └────── Separate binaries ───────┘
              Separate deployments
              Network overhead even
              when co-located

Current Pain Points:

Two separate binaries to build, deploy, and monitor
Kafka overhead even when running on the same machine
No easy way to run a lightweight local/dev setup
Tight coupling to Arroyo makes future engine changes difficult

Target Architecture

┌──────────────────────────────────────────────────────────────────┐
│                        asap binary                            │
├──────────────────────────────────────────────────────────────────┤
│                                                                   │
│  ┌─────────────────┐              ┌────────────────────────────┐ │
│  │  Query Layer    │              │  Streaming Engine          │ │
│  │  (HttpServer)   │              │  (trait: StreamingEngine)  │ │
│  └────────┬────────┘              └─────────────┬──────────────┘ │
│           │                                      │                │
│           │        ┌──────────────────┐          │                │
│           └───────►│   Core Engine    │◄─────────┘                │
│                    │   + Store        │                           │
│                    └──────────────────┘                           │
│                    (deployment-agnostic)                          │
│                                                                   │
└──────────────────────────────────────────────────────────────────┘

StreamingEngine implementations:
  ├── ArroyoEngine      (current)
  ├── MiniBatchEngine   (future)
  └── MockEngine        (testing)

Deployment modes:
  ├── single-node: in-memory channels, one process
  └── cluster: Kafka/gRPC, independent scaling

Benefits

Operational Benefits

Benefit	Description
Simplified Deployment	One binary to build, ship, and run for simple use cases
Reduced Latency	Single-node mode eliminates Kafka serialization overhead
Easier Development	Run everything locally with `asap --mode single-node`
Flexible Scaling	Same binary, different flags for cluster mode
Unified Monitoring	Single process = single set of metrics in simple deployments

Architectural Benefits

Benefit	Description
Engine Swappability	Replace Arroyo with custom mini-batch engine without touching query layer
Clear Boundaries	Traits enforce separation between core logic and I/O adapters
Testability	Mock implementations for unit/integration testing
Code Reuse	Core logic shared across all deployment modes

Developer Experience

Benefit	Description
Quick Start	`asap` just works out of the box for evaluation
Gradual Complexity	Start single-node, migrate to cluster when needed
Consistent Behavior	Same core logic guarantees identical results across modes

Core Abstractions

StreamingEngine Trait

The key abstraction that enables engine swappability:

#[async_trait]
pub trait StreamingEngine: Send + Sync {
    /// Start processing, sending results to the provided sink
    async fn run(&self, sink: Box<dyn PrecomputeSink>) -> Result<()>;

    /// Graceful shutdown
    async fn shutdown(&self) -> Result<()>;

    /// Health status for orchestration
    fn health(&self) -> EngineHealth;
}

PrecomputeSink Trait

Abstracts the destination for precomputed results:

#[async_trait]
pub trait PrecomputeSink: Send + Sync {
    async fn send(&self, output: PrecomputedOutput) -> Result<()>;
}

// Implementations:
// - ChannelSink: in-memory mpsc channel (single-node)
// - KafkaSink: Kafka producer (cluster mode)
// - DirectSink: direct function call to store (maximum performance)

IngestPort Trait

Abstracts how the query engine receives precomputed data:

#[async_trait]
pub trait IngestPort: Send + Sync {
    async fn recv(&mut self) -> Option<PrecomputedOutput>;
}

// Implementations:
// - ChannelIngest: receives from mpsc channel (single-node)
// - KafkaIngest: existing KafkaConsumer (cluster mode)

Configuration Strategy

Following VictoriaMetrics' pattern: unified configuration with CLI flags, environment variables, and config files.

Unified Config File

# config.yaml
mode: single-node  # or "cluster"

engine:
  type: arroyo     # or "minibatch" in future

  arroyo:
    parallelism: 4
    checkpoint_interval: 30s
    state_dir: /data/checkpoints

  minibatch:       # future
    window_size: 10s
    batch_size: 10000

query_engine:
  http_port: 8088
  prometheus_scrape_interval: 15s
  forward_unsupported_queries: true
  prometheus_server: http://localhost:9090

kafka:
  brokers: ["localhost:9092"]
  input_topic: metrics-raw
  output_topic: metrics-precomputed

# Cluster-mode specific
cluster:
  query_engine:
    replicas: 3
  streaming_workers:
    replicas: 5

CLI Usage

# Single-node (all-in-one, great for dev/eval)
asap --config config.yaml

# Single-node with CLI overrides
asap --config config.yaml --query-engine.http-port 9090

# Cluster mode - run query engine only
asap --config config.yaml --mode cluster --component query-engine

# Cluster mode - run streaming workers (scale these independently)
asap --config config.yaml --mode cluster --component streaming-worker

# Environment variable overrides (12-factor app friendly)
ASAP_MODE=cluster ASAP_QUERY_ENGINE_HTTP_PORT=8080 asap

Directory Structure

asap-internal/
├── lib/
│   ├── core/                     # Shared traits and types
│   │   ├── Cargo.toml
│   │   └── src/
│   │       ├── lib.rs
│   │       ├── streaming.rs      # StreamingEngine trait
│   │       ├── sink.rs           # PrecomputeSink trait
│   │       ├── ingest.rs         # IngestPort trait
│   │       └── types.rs          # PrecomputedOutput, shared types
│   │
│   └── config/                   # Unified configuration
│       ├── Cargo.toml
│       └── src/
│           ├── lib.rs
│           └── schema.rs         # Config structs (clap + serde)
│
├── engines/
│   ├── arroyo-adapter/           # Wraps Arroyo as StreamingEngine
│   │   ├── Cargo.toml
│   │   └── src/lib.rs
│   │
│   └── minibatch/                # Future custom engine
│       ├── Cargo.toml
│       └── src/lib.rs
│
├── QueryEngineRust/              # Existing (minimal changes)
│   ├── Cargo.toml
│   └── src/
│       ├── lib.rs                # Already exists, expose more
│       ├── main.rs               # Keep for standalone mode
│       └── ...
│
├── arroyo/                       # Existing (unchanged)
│   └── ...
│
└── asap/                     # New unified binary
    ├── Cargo.toml
    └── src/
        ├── main.rs               # CLI entry point
        ├── single_node.rs        # Single-node orchestration
        └── cluster.rs            # Cluster component runners

Migration Path

Phase 1: Create Core Abstractions

Goal: Define traits without changing existing code behavior.

Tasks:

Create lib/core crate with StreamingEngine, PrecomputeSink, IngestPort traits
Move PrecomputedOutput and shared types to lib/core
Update QueryEngineRust to depend on lib/core for types

Outcome: Clean interfaces defined, existing code still works unchanged.

Phase 2: Create Arroyo Adapter

Goal: Wrap Arroyo behind the StreamingEngine trait.

Tasks:

Create engines/arroyo-adapter crate
Implement StreamingEngine trait wrapping Arroyo's pipeline execution
Implement PrecomputeSink that writes to Arroyo's existing Kafka sink
Add ability for Arroyo to write to a custom sink (channel) instead of Kafka

Outcome: Arroyo can be used through the StreamingEngine abstraction.

Phase 3: Implement Sink/Ingest Adapters

Goal: Create the communication adapters for both modes.

Tasks:

Implement ChannelSink (mpsc sender)
Implement ChannelIngest (mpsc receiver)
Refactor KafkaConsumer to implement IngestPort trait
Verify existing Kafka sink works with PrecomputeSink trait

Outcome: Both in-memory and Kafka communication paths available.

Phase 4: Create Unified Binary (Single-Node)

Goal: Single binary that runs everything in one process.

Tasks:

Create asap crate
Create lib/config with unified configuration schema
Implement single-node mode wiring (channel-based communication)
Add CLI with clap (flags + env var support)
Test end-to-end: raw metrics in → precomputed queries out

Outcome: asap --mode single-node works.

Phase 5: Add Cluster Mode

Goal: Same binary can run individual components for independent scaling.

Tasks:

Add --component flag to CLI
Implement run_query_engine_only() (reads from Kafka)
Implement run_streaming_worker_only() (writes to Kafka)
Add health check endpoints for orchestration
Document Kubernetes/Docker Compose deployment patterns

Outcome: asap --mode cluster --component query-engine works.

Phase 6: (Future) Custom Mini-Batch Engine

Goal: Replace Arroyo with a custom, simpler streaming engine.

Tasks:

Implement MiniBatchEngine with StreamingEngine trait
Add engine.type: minibatch configuration support
Benchmark against Arroyo implementation
Gradual rollout with feature flag

Outcome: Can switch engines via configuration without code changes to query layer.

Success Criteria

Criterion	Measurement
Single binary deployment	`asap` runs full stack with one command
Zero Kafka overhead (single-node)	Benchmark shows latency improvement
Independent scaling (cluster)	Can scale streaming workers without touching query engine
Engine swappability	Can switch `engine.type` without query layer changes
Configuration parity	Same config file works for both modes
No regression	Existing functionality preserved through migration

Open Questions

State management in single-node mode: How should checkpointing work when Arroyo runs embedded? Use local filesystem?
Graceful shutdown coordination: In single-node mode, what's the shutdown sequence? Drain queries → checkpoint streaming → exit?
Resource isolation: Should single-node mode have configurable resource limits per component (e.g., streaming gets 4 cores, query gets 2)?
Hybrid mode: Is there value in a mode where query engine is embedded but streaming workers are distributed?

References

@milindsrivastava1997

Originally by @milindsrivastava1997 on 2026-01-21:

Arroyo Embedding Risks Assessment

This document summarizes the findings from investigating whether Arroyo can be embedded as a library within QueryEngineRust to create a unified single-node binary.

Executive Summary

Embedding Arroyo as a library faces three critical blockers related to its distributed architecture assumptions. Arroyo was designed as a distributed system where workers coordinate with a central controller - it has no standalone/embedded mode.

Critical Risks

1. Worker Requires Controller (CRITICAL)

Finding: The Arroyo worker cannot start without an active Controller.

Evidence (arroyo-worker/src/lib.rs:225-231):

let mut client = retry!(
    controller_client("worker", &config.worker.tls).await,
    2,
    Duration::from_millis(100),
    Duration::from_secs(2),
    |e| warn!("Failed to connect to controller: {e}, retrying...")
)?;

Impact:

Worker blocks on startup until controller responds
Continuous gRPC heartbeat every 5 seconds (run_control_loop)
Job assignments come from controller, not local configuration
Cannot run a worker in isolation

Required to fix: Remove/mock controller client, inject job config directly, implement local control loop.

2. Global Singletons (CRITICAL)

Finding: Multiple lazy_static, OnceLock, and ArcSwap singletons prevent multiple instances and cause initialization conflicts.

Configuration singleton (arroyo-rpc/src/config.rs:29):

static CONFIG: ArcSwapOption<Config> = ArcSwapOption::const_empty();

pub fn initialize_config(path: Option<&Path>, dir: Option<&Path>) {
    // ...
    panic!("Unable to initialize configuration; it's already initialized!");
}

Metrics singletons (arroyo-metrics/src/lib.rs:43-88):

lazy_static! {
    pub static ref MESSAGE_RECV_COUNTER: IntCounterVec = register_int_counter_vec!(...).unwrap();
    pub static ref MESSAGES_SENT: IntCounterVec = register_int_counter_vec!(...).unwrap();
    // 6+ more global counters
}

Other globals:

STORAGE_PROVIDER in arroyo-state (OnceCell)
DB_BACKUP_NOTIFIER in arroyo-rpc (OnceLock)
Various lazy_statics in arroyo-node, arroyo-controller

Impact:

Cannot run multiple Arroyo instances in one process
Config initialization panics if called twice
Prometheus metrics fail on duplicate registration
Parent binary cannot use same metric names

3. Configuration Namespace Pollution (CRITICAL)

Finding: Arroyo uses ARROYO__* environment variable prefix that can conflict with parent application.

Evidence (arroyo-rpc/src/config.rs:141-190):

// Priority (highest to lowest):
// 1. ARROYO__* environment variables  <-- Will override everything
// 2. Config file specified in code
// 3. *.toml or *.yaml in config directory
// ...

Impact:

Any ARROYO__* env var in the environment overrides programmatic config
Cannot isolate Arroyo config from parent application
Deployment environment may have conflicting variables

Example conflict:

# If parent application or environment has:
export ARROYO__WORKER__ID=999
export ARROYO__CHECKPOINT_URL=s3://other-bucket

# Arroyo will use these instead of intended values

Medium Risks

4. Hardcoded Connector Registry

Finding: Sinks are registered in a hardcoded list, not dynamically.

Evidence (arroyo-connectors/src/lib.rs:41-70):

pub fn connectors() -> HashMap<&'static str, Box<dyn ErasedConnector>> {
    let connectors: Vec<Box<dyn ErasedConnector>> = vec![
        Box::new(blackhole::BlackholeConnector {}),
        Box::new(kafka::KafkaConnector {}),
        // ... 20+ hardcoded connectors
    ];
    connectors.into_iter().map(|c| (c.name(), c)).collect()
}

Impact: Adding a custom in-memory channel sink requires modifying this file or forking the crate.

Workaround: Fork arroyo-connectors or modify to support dynamic registration.

5. Global Checkpoint Storage

Finding: Checkpoint storage provider is a global singleton.

Evidence (arroyo-state/src/lib.rs:167-186):

static STORAGE_PROVIDER: tokio::sync::OnceCell<Arc<StorageProvider>> =
    tokio::sync::OnceCell::const_new();

Impact:

Storage backend set once at startup, cannot change
Must configure checkpoint_url before any checkpoint operation

Workaround: Disable checkpointing for embedded mode, or accept the constraint.

Architecture Diagram

Current Arroyo Architecture (Distributed-First)
═══════════════════════════════════════════════

┌─────────────────────────────────────────────────────────────┐
│                         Controller                           │
│  - Job scheduling                                            │
│  - Checkpoint coordination                                   │
│  - Worker registration                                       │
└──────────────────────────┬──────────────────────────────────┘
                           │ gRPC (mandatory)
                           │ heartbeat every 5s
                           ▼
┌─────────────────────────────────────────────────────────────┐
│                          Worker                              │
│  - BLOCKS until controller responds                          │
│  - Receives job assignments via gRPC                         │
│  - Cannot run standalone                                     │
└─────────────────────────────────────────────────────────────┘

Global State (process-wide singletons)
══════════════════════════════════════
CONFIG ─────────────► panics if initialized twice
METRICS ────────────► prometheus registry, one per process
STORAGE_PROVIDER ───► checkpoint backend, set once

Implications for Unified Binary

Goal	Feasibility	Notes
Single binary with all components	Possible	Must run full Arroyo stack (controller + worker)
Remove Kafka in single-node mode	Hard	Requires forking worker to bypass controller
Multiple Arroyo instances per process	Impossible	Global singletons prevent this
Swap Arroyo for custom engine	Hard	Deep coupling to Arroyo's control plane
Clean library interface	Not available	Arroyo is a system, not a library

Revised Approach Options

Option A: Orchestrate Full Stack (Low Risk)

Run controller + worker + query engine in same binary, accept internal gRPC/Kafka.

async fn run_single_node() {
    tokio::spawn(arroyo_controller::run());  // Internal controller
    tokio::spawn(arroyo_worker::run());      // Connects via localhost
    tokio::spawn(query_engine::run());       // Consumes from Kafka
}

Effort: ~500-800 lines | Arroyo changes: None

Option B: Fork Worker for Embedded Mode (Medium Risk)

Create standalone worker that accepts jobs directly without controller.

Effort: ~800-1300 lines | Arroyo changes: ~300-500 lines modified

Option C: Use Arroyo Operators as Library (High Effort)

Extract operator implementations, build custom execution engine.

Effort: ~1500-2500 lines | Arroyo changes: Moderate

Option D: Keep Separate Processes (No Risk)

Bundle via Docker Compose / Kubernetes, no code changes.

Effort: ~100-200 lines (deployment config) | Arroyo changes: None

Recommendation

Start with Option A - it achieves the single-binary deployment goal with minimal risk. The internal gRPC/Kafka overhead within localhost is negligible for most use cases.

Pursue Option B only if profiling shows the internal communication is a bottleneck.

Option C aligns with the long-term goal of a custom mini-batch engine but should be deferred until that work begins.

References

arroyo-worker/src/lib.rs - Worker initialization and control loop
arroyo-rpc/src/config.rs - Configuration loading and globals
arroyo-metrics/src/lib.rs - Prometheus metric singletons
arroyo-state/src/lib.rs - Checkpoint storage globals
arroyo-connectors/src/lib.rs - Connector registry

@milindsrivastava1997

Originally by @milindsrivastava1997 on 2026-01-23:

Possible candidates for creating an embeddable stream processing library as a replacement for arroyo+ArroyoSketch:

Denormalized. Seems to be exactly what we want, but repo is abandoned
Extract core functionality from arroyo
timely-dataflow
Make one using datafusion. Datafusion contributors were discussing adding streaming support, but discussion is inactive from Jul 2024
Fluvio

@milindsrivastava1997

Originally by @milindsrivastava1997 on 2026-01-27:

Some datafusion related links:

@milindsrivastava1997

Originally by @milindsrivastava1997 on 2026-01-27:

Pinging Micah would be a good option

Explore if we can have a single binary + a cluster/microservice mode, similar to VictoriaMetrics #102

Description

Activity

milindsrivastava1997 commented on Mar 4, 2026

milindsrivastava1997 commented on Mar 4, 2026

Unified Binary Architecture: Single-Node and Cluster Deployment

Goal

Analogy: VictoriaMetrics Architecture

VictoriaMetrics Structure

Key Design Principles from VictoriaMetrics

Current State: QueryEngineRust + Arroyo

Target Architecture

Benefits

Operational Benefits

Architectural Benefits

Developer Experience

Core Abstractions

StreamingEngine Trait

PrecomputeSink Trait

IngestPort Trait

Configuration Strategy

Unified Config File

CLI Usage

Directory Structure

Migration Path

Phase 1: Create Core Abstractions

Phase 2: Create Arroyo Adapter

Phase 3: Implement Sink/Ingest Adapters

Phase 4: Create Unified Binary (Single-Node)

Phase 5: Add Cluster Mode

Phase 6: (Future) Custom Mini-Batch Engine

Success Criteria

Open Questions

References

milindsrivastava1997 commented on Mar 4, 2026

Arroyo Embedding Risks Assessment

Executive Summary

Critical Risks

1. Worker Requires Controller (CRITICAL)

2. Global Singletons (CRITICAL)

3. Configuration Namespace Pollution (CRITICAL)

Medium Risks

4. Hardcoded Connector Registry

5. Global Checkpoint Storage

Architecture Diagram

Implications for Unified Binary

Revised Approach Options

Option A: Orchestrate Full Stack (Low Risk)

Option B: Fork Worker for Embedded Mode (Medium Risk)

Option C: Use Arroyo Operators as Library (High Effort)

Option D: Keep Separate Processes (No Risk)

Recommendation

References

milindsrivastava1997 commented on Mar 4, 2026

milindsrivastava1997 commented on Mar 4, 2026

milindsrivastava1997 commented on Mar 4, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions