How the iii-stream Worker Integrates with Redis for Real-Time Streaming Patterns

The iii-stream worker uses a pluggable Redis adapter that persists stream items as Redis hashes while distributing real-time events through Redis Pub/Sub channels, enabling durable storage and low-latency fan-out to WebSocket clients.

The iii-stream worker is the built-in streaming primitive of the iii engine (iii-hq/iii). By configuring the worker to use the Redis adapter, you gain ACID-compliant persistence alongside a pub/sub backbone that powers live data streaming to connected clients.

Selecting the Redis Adapter

The stream worker loads adapters dynamically based on the adapter.name field in StreamModuleConfig. When set to "redis", the engine invokes the factory registered in engine/src/workers/stream/adapters/redis_adapter.rs:

crate::register_adapter!(<StreamAdapterRegistration> name: "redis", make_adapter);

The make_adapter function constructs a RedisAdapter instance by reading the optional redis_url from configuration (defaulting to redis://localhost:6379):

fn make_adapter(_engine: Arc<Engine>, config: Option<Value>) -> StreamAdapterFuture {
    Box::pin(async move {
        let redis_url = config
            .as_ref()
            .and_then(|c| c.get("redis_url"))
            .and_then(|v| v.as_str())
            .unwrap_or("redis://localhost:6379")
            .to_string();
        Ok(Arc::new(RedisAdapter::new(redis_url).await?) as Arc<dyn StreamAdapter>)
    })
}

Connection Architecture

The RedisAdapter maintains separate Redis clients for publishing and subscribing. In engine/src/workers/stream/adapters/redis_adapter.rs, the new constructor establishes:

• A connection manager for publishing operations (wrapped in a Mutex for thread-safe access)
• A dedicated async client for Pub/Sub subscriptions

let client = Client::open(redis_url.as_str())?;
let manager = timeout(DEFAULT_REDIS_CONNECTION_TIMEOUT, client.get_connection_manager())
    .await?
    .map_err(|e| anyhow::anyhow!("Failed to connect to Redis at {}: {}", redis_url, e))?;
let publisher = Arc::new(Mutex::new(manager));
let subscriber = Arc::new(client);

This dual-client design isolates command throughput from event listening, preventing subscription latency from blocking storage writes.

Persistent Storage with Redis Hashes

All stream operations map to Redis hash commands keyed by stream:<stream_name>:<group_id>. The adapter implements atomic operations using Lua scripts to ensure ACID semantics without client-side locks.

Atomic Set and Delete

The set method uses a Lua script to retrieve the old value before writing the new JSON payload:

let script = redis::Script::new(
    r#"
        local old_value = redis.call('HGET', KEYS[1], ARGV[1])
        redis.call('HSET', KEYS[1], ARGV[1], ARGV[2])
        return old_value
    "#,
);

Similarly, delete atomically returns the removed value before deleting the hash field, ensuring clients receive the previous state even under concurrent load.

Partial Updates with JSON Operations

The update method applies JSON Patch operations in-place using a pre-defined Lua script (JSON_UPDATE_SCRIPT). This script parses a list of UpdateOp structs, applies them to the stored JSON, and returns both the previous and current values without requiring a full read-modify-write cycle from the client.

Group Queries

For listing streams and groups, the adapter uses non-blocking HSCAN and SCAN commands rather than KEYS. All Redis commands lock the publisher mutex only for the duration of network I/O, maximizing concurrency across multiple stream workers.

Real-Time Event Distribution

Beyond persistence, the Redis adapter powers the real-time streaming pattern through Pub/Sub.

Publishing Events

Every mutation (set, delete, update) calls emit_event, which serializes a StreamWrapperMessage and publishes it to the stream::events channel:

let event_json = serde_json::to_string(&message)?;
conn.publish::<_, _, ()>(&STREAM_TOPIC, &event_json).await?;

Fan-Out to WebSocket Clients

The adapter runs a background watch_events task that subscribes to the same channel, deserializes messages once, and broadcasts them to all active WebSocket connections tracked in a RwLock<HashMap<String, Arc<dyn StreamConnection>>>:

let mut pubsub = self.subscriber.get_async_pubsub().await?;
pubsub.subscribe(&STREAM_TOPIC).await?;
while let Some(msg) = msg.next().await {
    let msg: StreamWrapperMessage = serde_json::from_str(&payload)?;
    for connection in connections.values() {
        connection.handle_stream_message(&msg).await?;
    }
}

The StreamWorker spawns this watcher at boot time in engine/src/workers/stream/stream.rs:

let watch_handle = tokio::spawn(async move { adapter.watch_events().await });

When a client connects via WebSocket to the worker's HTTP server, the adapter registers the connection in the shared map, ensuring subsequent Redis events reach that socket immediately.

Configuration and Usage Examples

Enabling Redis in Engine Configuration

Configure the worker in engine/config.yaml to use the Redis adapter:

workers:
  - id: iii-stream
    config:
      port: 3112
      host: "0.0.0.0"
      adapter:
        name: "redis"
        config:
          redis_url: "redis://redis.example.com:6379"

Setting adapter.name to "redis" triggers the factory that instantiates RedisAdapter with the provided URL.

Writing Stream Items via SDK

Using the TypeScript SDK to persist and update data:

import { iii } from "iii-sdk";

await iii.callFunction({
  function_id: "stream::set",
  payload: {
    stream_name: "chat",
    group_id: "room-42",
    item_id: "msg-123",
    data: { author: "alice", text: "Hello world!" }
  }
});

await iii.callFunction({
  function_id: "stream::update",
  payload: {
    stream_name: "chat",
    group_id: "room-42",
    item_id: "msg-123",
    ops: [{ op: "append", path: "reactions", value: "thumbs_up" }]
  }
});

Subscribing to Live Updates

Connect via WebSocket to receive real-time events broadcast by the Redis watcher:

const socket = new WebSocket("ws://localhost:3112/");

socket.addEventListener("message", (ev) => {
  const msg = JSON.parse(ev.data);
  if (msg.type === "stream") {
    console.log("Stream event:", msg);
  }
});

When any mutation occurs, the Redis adapter publishes the change to the stream::events channel, the worker's watch_events task forwards it, and connected clients receive the payload immediately.

Summary

• The iii-stream worker integrates with Redis through a configurable adapter system located in engine/src/workers/stream/adapters/redis_adapter.rs.
• Persistent storage uses Redis hashes keyed by stream:<stream_name>:<group_id>, with Lua scripts ensuring atomic set, delete, and JSON update operations.
• Real-time distribution relies on Redis Pub/Sub via the stream::events channel, with the adapter acting as both publisher (on writes) and subscriber (for WebSocket fan-out).
• The adapter maintains dual Redis clients—one connection manager for commands and one async client for Pub/Sub—to optimize throughput and latency.
• Configuration requires setting adapter.name to "redis" and optionally specifying a redis_url in the worker configuration.

Frequently Asked Questions

What Redis data structures does the iii-stream worker use?

The worker uses Redis hashes to store stream items, where each hash key follows the pattern stream:<stream_name>:<group_id> and fields represent individual item IDs. For real-time notifications, it uses Redis Pub/Sub with a single channel named stream::events to broadcast changes to all connected WebSocket clients.

How does the iii-stream worker handle concurrent writes to the same stream item?

The RedisAdapter uses Lua scripts for all write operations (set, delete, update). These scripts execute atomically on the Redis server, ensuring that read-modify-write cycles are race-condition free without requiring client-side locks or transactions.

Can I use a Redis Cluster with the iii-stream worker?

The current implementation in redis_adapter.rs uses redis::Client::open with a single connection URL. For Redis Cluster support, you would need to modify the make_adapter factory to initialize a redis::cluster::ClusterClient instead of the standard client, though this is not implemented in the current iii-hq/iii codebase.

What is the default Redis URL if not specified in configuration?

If the redis_url key is omitted from the adapter configuration, the factory defaults to redis://localhost:6379. This is hardcoded in the make_adapter function within engine/src/workers/stream/adapters/redis_adapter.rs.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}