Channels

Channels move large or binary payloads between iii workers without putting the data in a JSON function payload. Use a channel for a file that is larger than the JSON message limit, such as a report.csv of 100 MiB. For small JSON, use a regular worker.call(...).

For the underlying model (how channels are addressed, multiplexed, and torn down), see Channels architecture.

Payload size

The limit applies to the complete serialized UTF-8 JSON envelope, measured in bytes. The Rust, Node, Python, and Go SDKs support messages up to 64 MiB. The browser SDK has a 16 MiB outbound limit and does not fragment messages because browser WebSocket APIs do not expose frame control. Browser workers also have a 16 MiB receive limit. For Rust, Node, Python, and Go, messages up to 16 MiB are sent as one text frame. Larger messages, up to and including 64 MiB, are sent as one fragmented WebSocket text message using multiple frames. The engine sends every message larger than 1 MiB to a worker as one fragmented text message of 1 MiB frames, so no frame it sends exceeds a worker’s frame size limit. Applications send and receive these messages through the usual function APIs. The receiving worker must run SDK version 0.24.5 or later to accept messages larger than 16 MiB. An updated engine checks the serialized envelope against the target worker’s receive limit and returns payload_too_large to the caller when the target cannot accept it. The engine sets that limit when the worker registers:

A browser worker gets 16 MiB.
A worker that declares the large-messages.v1 protocol feature gets 64 MiB, whatever version it reports. The Rust, Node, Python, and Go SDKs declare it from 0.25.0; the browser SDK does not.
Any other worker gets 64 MiB when the version it reports is an SDK version of 0.24.5 or later, written as semver (0.24.6, 0.25.0-rc.5) or PEP 440 (0.25.0rc5, 0.24.5.post1). It gets 16 MiB when the version is missing, cannot be parsed, or is older, including pre-releases of 0.24.5. Other features, such as observe.v1, do not count.

SDK 0.24.5 through 0.25.0-rc.5 do not declare large-messages.v1, so the engine relies on the version they report. A Rust worker on one of them that sets WorkerMetadata.version to its own version instead of the SDK version therefore stays at 16 MiB. Upgrade it to SDK 0.25.0 or later. Above 64 MiB, the Rust, Node, Python, and Go SDKs reject the message locally with payload_too_large; the error message links to this Payload size section. Engines older than the updated size check forward messages as a single frame without checking the target’s limit. An older worker may then disconnect if the message is too large for it. Use an engine and SDKs from the same release, or newer. The engine reads up to 64 MiB per message from a worker, whether the message arrives as one frame or as fragments, so it also accepts the single large frames of SDKs older than 0.24.5. When a worker sends a larger message, the engine drops it and the connection stays open. A call gets payload_too_large as its result, and when a function returns a result that is too large, its caller gets payload_too_large instead. Other calls on the connection are not affected. A message the engine cannot answer, such as a call that expects no result, is dropped and logged. Observer activations follow the same limits, and the engine keeps at most 64 MiB of one observed event (every field its observers capture, input and output counted separately). An event over either limit is not delivered: the observer counts a TOO_LARGE drop and the engine logs a warning. See Observers / Observed call size. Use a channel for payloads above 64 MiB. Channels stream payload bytes separately from the JSON function envelope. See Using channels.

Using channels

A channel is created by one worker and has two local stream ends, writer and reader, and two serializable refs, writerRef and readerRef, that another function receives in its payload. The subsections below cover the local ends.

Create a channel

createChannel() returns the channel. Write the payload to writer and close it when finished. The bytes flow through the engine to whichever worker holds the matching reader.

Node / TypeScript
Python
Rust
Go

import { registerWorker } from 'iii-sdk'
import { createChannel } from 'iii-sdk/helpers'

const worker = registerWorker('ws://localhost:49134')
const channel = await createChannel(worker)

channel.writer.stream.write(Buffer.from('file contents'))
channel.writer.stream.end()

import asyncio

from iii import register_worker
from iii.helpers import create_channel_async


async def main() -> None:
    worker = register_worker("ws://localhost:49134")

    channel = await create_channel_async(worker)
    await channel.writer.write(b"file contents")
    await channel.writer.close_async()


asyncio.run(main())

use iii_sdk::{register_worker, Error, InitOptions};

#[tokio::main]
async fn main() -> Result<(), Error> {
    let worker = register_worker("ws://localhost:49134", InitOptions::default());

    let channel = iii_sdk::helpers::create_channel(&worker, None).await?;
    channel.writer.write(b"file contents").await?;
    channel.writer.close().await?;

    worker.shutdown();
    Ok(())
}

package main

import (
	"context"
	"log"

	iii "github.com/iii-hq/iii/sdk/packages/go/iii"
)

func main() {
	ctx := context.Background()
	client := iii.RegisterWorker("ws://localhost:49134")

	channel, err := client.CreateChannel(ctx, nil)
	if err != nil {
		log.Fatal(err)
	}
	if err := channel.Writer.Write(ctx, []byte("file contents")); err != nil {
		log.Fatal(err)
	}
	if err := channel.Writer.Close(); err != nil {
		log.Fatal(err)
	}
}

Read from a channel

Read the local reader until the other end closes. The bytes arrive in the order they were written by whichever worker holds the matching writer. If the writer’s connection ends without a close (it aborted, crashed or lost its connection), the read fails with an error instead of ending: treat it as an incomplete transfer. See Backpressure and lifecycle for how a writer aborts.

Node / TypeScript
Python
Rust
Go

import { registerWorker } from 'iii-sdk'
import { createChannel } from 'iii-sdk/helpers'

const worker = registerWorker('ws://localhost:49134')
const channel = await createChannel(worker)

// Another worker holds channel.writerRef and writes to it.
let bytes = 0
for await (const chunk of channel.reader.stream) {
  bytes += chunk.length
}
console.log(`read ${bytes} bytes`)

import asyncio

from iii import register_worker
from iii.helpers import create_channel_async


async def main() -> None:
    worker = register_worker("ws://localhost:49134")
    channel = await create_channel_async(worker)

    # Another worker holds channel.writer_ref and writes to it.
    total = 0
    async for chunk in channel.reader:
        total += len(chunk)
    print(f"read {total} bytes")


asyncio.run(main())

use iii_sdk::{register_worker, Error, InitOptions};

#[tokio::main]
async fn main() -> Result<(), Error> {
    let worker = register_worker("ws://localhost:49134", InitOptions::default());
    let channel = iii_sdk::helpers::create_channel(&worker, None).await?;

    // Another worker holds channel.writer_ref and writes to it.
    let bytes = channel.reader.read_all().await?;
    println!("read {} bytes", bytes.len());

    worker.shutdown();
    Ok(())
}

package main

import (
	"context"
	"log"

	iii "github.com/iii-hq/iii/sdk/packages/go/iii"
)

func main() {
	ctx := context.Background()
	client := iii.RegisterWorker("ws://localhost:49134")

	channel, err := client.CreateChannel(ctx, nil)
	if err != nil {
		log.Fatal(err)
	}

	// Another worker holds channel.WriterRef and writes to it.
	bytes, err := channel.Reader.ReadAll(ctx)
	if err != nil {
		log.Fatal(err)
	}
	log.Printf("read %d bytes", len(bytes))
}

Using channels across functions

A channel only becomes useful once two code paths own its ends. Typically one function holds the local writer / reader and another function receives the matching ref in its payload. The two subsections below cover that handoff: first how to send a ref alongside a normal call, then how the receiving function opens the ref to read or write.

Pass a channel ref to another function

Pass the readerRef (or writerRef) as part of a normal function invocation. The receiving function uses the ref to read from (or write to) the channel.

Node / TypeScript
Python
Rust
Go

import { registerWorker } from 'iii-sdk'
import { createChannel } from 'iii-sdk/helpers'

const worker = registerWorker('ws://localhost:49134')
const channel = await createChannel(worker)

channel.writer.stream.write(Buffer.from('file contents'))
channel.writer.stream.end()

const result = await worker.call({
  function_id: 'files::process',
  payload: { filename: 'report.csv', reader: channel.readerRef },
})

import asyncio

from iii import register_worker
from iii.helpers import create_channel_async


async def main() -> None:
    worker = register_worker("ws://localhost:49134")
    channel = await create_channel_async(worker)

    await channel.writer.write(b"file contents")
    await channel.writer.close_async()

    result = await worker.call_async(
        {
            "function_id": "files::process",
            "payload": {
                "filename": "report.csv",
                "reader": channel.reader_ref.model_dump(),
            },
        }
    )


asyncio.run(main())

use iii_sdk::{register_worker, CallRequest, Error, InitOptions};
use serde_json::json;

#[tokio::main]
async fn main() -> Result<(), Error> {
    let worker = register_worker("ws://localhost:49134", InitOptions::default());

    let channel = iii_sdk::helpers::create_channel(&worker, None).await?;
    channel.writer.write(b"file contents").await?;
    channel.writer.close().await?;

    let result = worker
        .call(CallRequest::new(
            "files::process",
            json!({
                "filename": "report.csv",
                "reader": channel.reader_ref,
            }),
        ))
        .await?;

    println!("files::process returned {result}");

    worker.shutdown();
    Ok(())
}

package main

import (
	"context"
	"encoding/json"
	"log"

	iii "github.com/iii-hq/iii/sdk/packages/go/iii"
)

func main() {
	ctx := context.Background()
	client := iii.RegisterWorker("ws://localhost:49134")

	channel, err := client.CreateChannel(ctx, nil)
	if err != nil {
		log.Fatal(err)
	}
	if err := channel.Writer.Write(ctx, []byte("file contents")); err != nil {
		log.Fatal(err)
	}
	if err := channel.Writer.Close(); err != nil {
		log.Fatal(err)
	}

	payload, err := json.Marshal(map[string]any{
		"filename": "report.csv",
		"reader":   channel.ReaderRef,
	})
	if err != nil {
		log.Fatal(err)
	}

	result, err := client.Call(ctx, iii.CallRequest{
		FunctionID: "files::process",
		Payload:    payload,
	})
	if err != nil {
		log.Fatal(err)
	}
	log.Printf("files::process returned %s", result)
}

The channel belongs to the worker that created it. Keep that worker connected until the receiving function has attached to the ref. If the creator disconnects first, unattached ends are dropped. See Channel ownership.

Read from a channel ref

The receiving function gets the ref in its payload and reads it until the other end closes.

Node / TypeScript
Python
Rust
Go

import { registerWorker } from 'iii-sdk'
import type { ChannelReader } from 'iii-sdk/channel'

const worker = registerWorker('ws://localhost:49134')

worker.registerFunction('files::process', async (input: { reader: ChannelReader }) => {
  let bytes = 0
  for await (const chunk of input.reader.stream) {
    bytes += chunk.length
  }
  return { bytes }
})

from iii import register_worker

worker = register_worker("ws://localhost:49134")


async def process_file(payload: dict) -> dict:
    total = 0
    async for chunk in payload["reader"]:
        total += len(chunk)
    return {"bytes": total}


worker.register_function("files::process", process_file)

use iii_sdk::channel::ChannelReader;
use iii_sdk::helpers::{extract_channel_refs, ChannelDirection};
use iii_sdk::{register_worker, Error, InitOptions, RegisterFunction};
use serde_json::{json, Value};

#[tokio::main]
async fn main() -> Result<(), Error> {
    let worker = register_worker("ws://localhost:49134", InitOptions::default());
    let address = worker.address().to_string();

    worker.register_function(
        "files::process",
        RegisterFunction::new_async(move |input: Value| {
            let address = address.clone();
            async move {
                let refs = extract_channel_refs(&input);
                let (_, reader_ref) = refs
                    .iter()
                    .find(|(key, r)| {
                        key == "reader" && matches!(r.direction, ChannelDirection::Read)
                    })
                    .ok_or_else(|| Error::Handler("missing reader channel ref".to_string()))?;

                let reader = ChannelReader::new(&address, reader_ref);
                let bytes = reader.read_all().await?.len();

                Ok(json!({ "bytes": bytes }))
            }
        }),
    );

    worker.shutdown();
    Ok(())
}

package main

import (
	"context"
	"encoding/json"
	"errors"
	"log"

	iii "github.com/iii-hq/iii/sdk/packages/go/iii"
)

func main() {
	client := iii.RegisterWorker("ws://localhost:49134")

	_, err := client.RegisterFunction("files::process",
		func(ctx context.Context, data json.RawMessage) (any, error) {
			refs, err := iii.ExtractChannelRefs(data)
			if err != nil {
				return nil, err
			}
			readerRef, ok := refs["reader"]
			if !ok || readerRef.Direction != iii.ChannelRead {
				return nil, errors.New("missing reader channel ref")
			}

			reader := client.OpenReader(readerRef)
			defer reader.Close()

			bytes, err := reader.ReadAll(ctx)
			if err != nil {
				return nil, err
			}
			return map[string]int{"bytes": len(bytes)}, nil
		})
	if err != nil {
		log.Fatal(err)
	}
}

Write to a channel ref

The receiving function gets the ref in its payload, writes the payload to it, and closes it.

Node / TypeScript
Python
Rust
Go

import { registerWorker } from 'iii-sdk'
import type { ChannelWriter } from 'iii-sdk/channel'

const worker = registerWorker('ws://localhost:49134')

worker.registerFunction('files::generate', async (input: { writer: ChannelWriter }) => {
  input.writer.stream.write(Buffer.from('file contents'))
  input.writer.stream.end()
  return { ok: true }
})

from iii import register_worker

worker = register_worker("ws://localhost:49134")


async def generate_file(payload: dict) -> dict:
    writer = payload["writer"]
    await writer.write(b"file contents")
    await writer.close_async()
    return {"ok": True}


worker.register_function("files::generate", generate_file)

use iii_sdk::channel::ChannelWriter;
use iii_sdk::helpers::{extract_channel_refs, ChannelDirection};
use iii_sdk::{register_worker, Error, InitOptions, RegisterFunction};
use serde_json::{json, Value};

#[tokio::main]
async fn main() -> Result<(), Error> {
    let worker = register_worker("ws://localhost:49134", InitOptions::default());
    let address = worker.address().to_string();

    worker.register_function(
        "files::generate",
        RegisterFunction::new_async(move |input: Value| {
            let address = address.clone();
            async move {
                let refs = extract_channel_refs(&input);
                let (_, writer_ref) = refs
                    .iter()
                    .find(|(key, r)| {
                        key == "writer" && matches!(r.direction, ChannelDirection::Write)
                    })
                    .ok_or_else(|| Error::Handler("missing writer channel ref".to_string()))?;

                let writer = ChannelWriter::new(&address, writer_ref);
                writer.write(b"file contents").await?;
                writer.close().await?;

                Ok(json!({ "ok": true }))
            }
        }),
    );

    worker.shutdown();
    Ok(())
}

package main

import (
	"context"
	"encoding/json"
	"errors"
	"log"

	iii "github.com/iii-hq/iii/sdk/packages/go/iii"
)

func main() {
	client := iii.RegisterWorker("ws://localhost:49134")

	_, err := client.RegisterFunction("files::generate",
		func(ctx context.Context, data json.RawMessage) (any, error) {
			refs, err := iii.ExtractChannelRefs(data)
			if err != nil {
				return nil, err
			}
			writerRef, ok := refs["writer"]
			if !ok || writerRef.Direction != iii.ChannelWrite {
				return nil, errors.New("missing writer channel ref")
			}

			writer := client.OpenWriter(writerRef)
			if err := writer.Write(ctx, []byte("file contents")); err != nil {
				return nil, err
			}
			if err := writer.Close(); err != nil {
				return nil, err
			}
			return map[string]bool{"ok": true}, nil
		})
	if err != nil {
		log.Fatal(err)
	}
}

For the per-language channel API surface, see the SDK reference: Node, Python, Rust, and Go.

Getting Started

Using iii

Understanding iii

Creating Workers

Upgrading

Payload size

Using channels

Create a channel

Read from a channel

Using channels across functions

Pass a channel ref to another function

Read from a channel ref

Write to a channel ref

​Payload size

​Using channels

​Create a channel

​Read from a channel

​Using channels across functions

​Pass a channel ref to another function

​Read from a channel ref

​Write to a channel ref

Payload size

Using channels

Create a channel

Read from a channel

Using channels across functions

Pass a channel ref to another function

Read from a channel ref

Write to a channel ref