Protocols

Ironflow supports two streaming protocols for real-time event subscriptions:

Protocol	Endpoint	Use Case
WebSocket	ws://localhost:9123/ws	Browser clients, bidirectional communication
ConnectRPC	http://localhost:9123/ironflow.v1.PubSubService/*	Server-to-server, gRPC-compatible clients

Both protocols support pattern subscriptions, replay, CEL filtering, and consumer groups. Manual acknowledgment works for consumer groups over the WebSocket ack frame, through the AckEvent RPC, and on the SubscribeBidirectional stream. See Acknowledging a consumer-group event, Bidirectional streaming and Acknowledgment. It is not implemented for plain subscriptions.

---

WebSocket Protocol

Connect directly to ws://localhost:9123/ws for browser clients or custom integrations.

Message Types

Type	Direction	Description
subscribe	Client → Server	Request subscription
unsubscribe	Client → Server	Cancel subscription
ack	Client → Server	Event acknowledgment
subscription_result	Server → Client	Subscription confirmation
subscribed	Server → Client	Legacy subscription confirmation. Declared in the wire vocabulary, but the server never sends it. Read subscription_result.
event	Server → Client	Event delivery
subscription_error	Server → Client	Subscription-specific error
error	Server → Client	Connection-level error, not tied to a subscription — see Error Response

Subscribe Request

{
  "type": "subscribe",
  "subscription": {
    "pattern": "system.run.>",
    "options": {
      "replay": 10,
      "includeMetadata": true,
      "filter": "data.status == \"completed\"",
      "consumerGroup": "my-group",
      "ackMode": "manual",
      "topicOnly": false
    }
  }
}

topicOnly (default false) delivers each frame without its data payload — use it when the subscriber only needs to know that an event fired. The event is still marshaled and sent per client, so the saving is the payload itself, not the frame. data is present but null on suppressed frames.

Metadata is unaffected: it stays governed by includeMetadata.

startAfterSequence (optional, no default) resumes delivery after a stream sequence the client previously received on an event’s meta.sequence (#1848). It is a position, not a count, so a reconnecting client neither repeats nor skips work the way replay does. Sequence 0 is a legitimate “from the beginning of the stream”, so the field is omitted rather than zero when unset. Two rejections, both invalid_argument:

• combined with replay — they are different starting points, so the server refuses rather than merging them
• combined with consumerGroup — a group is durable and server-positioned, and a client cursor would move the position for every other member

Setting it routes the subscription to a dedicated JetStream consumer instead of the subject’s shared one. Delivery stays at-least-once: a resumed stream can repeat the frame that was in flight when the connection dropped.

A replay skips a projection’s updates from a failed rebuild until a rebuild succeeds, so it can deliver nothing for that projection. The subscription still opens, and live updates arrive once a rebuild runs.

A subscription resuming from a cursor and a consumer group are not filtered this way. They deliver what that client missed, in order, including state saves of a rebuild that later failed, which is what a connected subscriber received live. To tell whether a state is current, read GetProjection or watch system.projection_health.{name}.

A subscription with replay also gets a dedicated JetStream consumer, anchored right after the stream sequence the replay stopped at, so no event published during the replay is lost or sent twice (#3065). Each replaying subscription therefore holds one server-side consumer until it ends, and the server decodes every message once for each replaying subscriber, so replay costs more than a plain subscription. Unlike a cursor subscriber, a replaying one is not ended when it falls behind: the server waits up to 5 seconds for room in its buffer and then drops events at once until room returns. If the server cannot attach the live consumer after the replay, it retries twice, then falls back to a live path without the guarantee and logs a warning. If nothing attaches at all, it ends the subscription like a cursor subscriber that fell behind. The guarantee covers the server bridge. Over WebSocket, the 256-frame send buffer of the connection still drops frames for a client that reads slower than the server sends, including a large replay.

A subscriber with a cursor that falls behind is never sent an event past one it missed. When the server’s buffer for it is full, the server waits for room. If none comes for 5 seconds, it ends the subscription, sends the events already queued, and then closes the connection with code 1013 (Try Again Later). A client that reads nothing at all never reaches that close frame: the server’s write timeout drops the connection and the client sees 1006. In both cases, reconnect and resubscribe with the meta.sequence of the last event you received; the SDKs do this for you. The close ends every subscription on the connection, so a subscription without a cursor on the same connection can miss events while it reconnects. A subscription without a cursor or a replay keeps the old behavior: a full buffer drops events at once.

Subscription Result

{
  "type": "subscription_result",
  "results": [
    {
      "pattern": "system.run.>",
      "status": "ok",
      "subscriptionId": "sub_abc123"
    }
  ]
}

Event Delivery

{
  "type": "event",
  "subscriptionId": "sub_abc123",
  "eventId": "evt_xyz789",
  "topic": "system.run.xyz.created",
  "data": {
    "id": "xyz",
    "version": 1,
    "functionId": "process-order",
    "status": "running"
  },
  "meta": {
    "timestamp": "2024-01-15T10:30:00Z",
    "sequence": 12345
  }
}

When includeMetadata: true is set in the subscribe options, the meta field is included in each event delivery. It also carries sequenceExact, the same sequence as a decimal string, for clients whose JSON parser cannot hold every uint64. timestamp is the time the event was published, and a redelivery repeats it, unlike ConnectRPC, which stamps the frame when it builds it. For user-emitted events, meta also contains a custom field with any metadata attached at emit time:

{
  "type": "event",
  "subscriptionId": "sub_abc123",
  "eventId": "evt_xyz789",
  "topic": "events:order.placed",
  "data": { "orderId": "123" },
  "meta": {
    "timestamp": "2024-01-15T10:30:00Z",
    "sequence": 12346,
    "custom": {
      "source": "checkout",
      "traceId": "abc-123"
    }
  }
}

Run frames and ordering

A system.run.{runId}.{event} frame carries the run’s state. Its data has id, functionId, eventId, executionMode, status, attempt, maxAttempts, createdAt, updatedAt, timestamp, and version. startedAt, completedAt, output and error appear when set.

version is the run row’s version as written by the transition the frame announces. It rises with every write to the run. Frames for one run can arrive out of order when more than one node serves it, for example when a pull worker polls one node and completes on another. A watcher that shows the status of the last frame it got can then show running after completed. Keep the highest version seen per run and drop a frame with a lower one:

{ "topic": "system.run.xyz.completed", "data": { "id": "xyz", "status": "completed", "version": 5 } }
{ "topic": "system.run.xyz.updated",   "data": { "id": "xyz", "status": "running",   "version": 4 } }

The second frame is stale. Rules a consumer should follow:

• A frame whose version ties the highest seen is kept. One write can publish two frames, .failed and .updated.
• A zero, absent, or invalid version is unknown. Keep it without changing the highest version, unless the run is remembered as deleted. Writers use zero only when no committed snapshot is available; they never attach an earlier state’s version to a fallback payload.
• Step frames (system.run.{runId}.step.{stepId}.{event}) carry the step row’s own version, ranked independently by step identity. The payload has id, runId, name, type, status, createdAt, updatedAt, timestamp, and version; optional fields are startedAt, endedAt, output, and error. New steps start at version 1. Existing version-zero rows remain unknown until their next write.
• A deleted frame has no version. Remember the deleted run ID and reject later run and step frames, including unknown versions. Duplicate deletions pass.
• Each SDK filter keeps a bounded cache shared by run and step identities (10,000 by default, configurable with maxRuns, MaxRuns, or max_runs). Eviction forgets ordering and deletion history, so late frames for forgotten identities can pass. Reconnects and cache loss require a fresh snapshot when authoritative state matters.

The SDKs ship this filter: createRunVersionFilter (Node and browser), RunVersionFilter (Go and Python). ironflow subscribe and the browser agents.subscribe apply it. Desktop Activity does not: it lists failures, one entry per .failed frame, and the earlier failure of a run that was resumed and failed again is still a distinct failure.

Unsubscribe Request

{
  "type": "unsubscribe",
  "subscriptionId": "sub_abc123"
}

Acknowledgment (Manual Ack Mode)

The ack frame acknowledges an event that a MANUAL consumer group delivered. It applies to subscriptions that set consumerGroup in the subscribe options. See the subscribe request example. It does nothing for plain (non-group) subscriptions. A plain subscription always acknowledges automatically.

{ "type": "ack", "ackType": "ack|nak|term", "eventId": "…", "subscriptionId": "…", "redeliverDelay": 500 }

ackType	Description
ack	Successfully processed
nak	Failed, request redelivery
term	Failed permanently, do not redeliver

subscriptionId is optional. It names the consumer-group subscription that received the event. Without it, the server tries every group that the connection joined. An unknown subscriptionId does nothing. redeliverDelay (milliseconds) applies to nak only. The frame has no reply. The server logs a failed ack and does not report it to the client.

Error Response

{
  "type": "subscription_error",
  "subscriptionId": "sub_abc123",
  "code": "INVALID_PATTERN",
  "message": "invalid namespace prefix",
  "retrying": false
}

A failure the server cannot attribute to one subscription arrives on a plain error frame instead — { "type": "error", "code": ..., "message": ... }, with no subscriptionId. Handle both, and route an error frame to a global handler: an unparseable message, an unknown message type, and a malformed subscribe/unsubscribe frame all arrive this way, and none of them names the request that caused it.

---

ConnectRPC Protocol

ConnectRPC provides gRPC-compatible streaming over HTTP. Use the PubSubService for subscriptions.

Service Definition

service PubSubService {
  // Emit an event (alias for Trigger with emit semantics)
  rpc Emit(EmitRequest) returns (EmitResponse);

  // Server-streaming subscription (auto-ack)
  rpc Subscribe(SubscribeRequest) returns (stream SubscriptionEvent);

  // Join one consumer group and carry its acks on the same stream
  rpc SubscribeBidirectional(stream SubscribeBidirectionalRequest) returns (stream SubscribeBidirectionalResponse);

  // Consumer group management
  rpc CreateConsumerGroup(CreateConsumerGroupRequest) returns (ConsumerGroup);
  rpc GetConsumerGroup(GetConsumerGroupRequest) returns (ConsumerGroup);
  rpc ListConsumerGroups(ListConsumerGroupsRequest) returns (ListConsumerGroupsResponse);
  rpc UpdateConsumerGroup(UpdateConsumerGroupRequest) returns (ConsumerGroup);
  rpc DeleteConsumerGroup(DeleteConsumerGroupRequest) returns (google.protobuf.Empty);

  // Consumer group streaming
  rpc JoinConsumerGroup(JoinConsumerGroupRequest) returns (stream SubscriptionEvent);

  // Ack, nak or term one event of a MANUAL consumer group
  rpc AckEvent(AckEventRequest) returns (AckEventResponse);

  // Developer pub/sub (does NOT trigger workflow functions)
  rpc Publish(PublishRequest) returns (PublishResponse);
  rpc ListTopics(ListTopicsRequest) returns (ListTopicsResponse);
  rpc GetTopicStats(GetTopicStatsRequest) returns (GetTopicStatsResponse);
}

Subscribe Request (ConnectRPC)

Endpoint: POST /ironflow.v1.PubSubService/Subscribe

Request:

{
  "pattern": "system.run.>",
  "options": {
    "replay": 10,
    "include_metadata": true,
    "filter": "data.status == \"completed\"",
    "consumer_group": "my-group"
  }
}

Resuming after a disconnect — start_after_sequence is the ConnectRPC spelling of the WebSocket startAfterSequence option above. It is a uint64, so Connect’s JSON encoding carries it as a string:

{
  "pattern": "system.run.>",
  "options": {
    "include_metadata": true,
    "start_after_sequence": "12345"
  }
}

It takes the same two invalid_argument rejections — combined with replay, or combined with consumer_group — so it cannot appear in the request above. It is optional in the proto so that sequence 0 (“from the beginning of the stream”) stays distinguishable from unset.

A resuming subscriber that reads too slowly is held back, not skipped. If the server’s buffer for it stays full for 5 seconds, the server sends what it holds and ends the stream with unavailable. Only that stream ends. Call Subscribe again with the sequence of the last event you received; the SDKs do this for you.

SubscribeOptions also declares namespace, ack_mode, and backpressure. The Subscribe handler does not read any of the three: the namespace comes from the pattern’s own prefix, and manual ack is unimplemented. ack_mode and backpressure are honored on consumer groups, where they are set on the group itself through CreateConsumerGroup / UpdateConsumerGroup. There is no topic_only on this transport — that option is WebSocket-only.

Response (streaming):

{
  "subscription_id": "sub_abc123",
  "event_id": "evt_xyz789",
  "topic": "system.run.xyz.created",
  "data": {
    "id": "xyz",
    "version": 1,
    "status": "running"
  },
  "metadata": {
    "timestamp": "2024-01-15T10:30:00.000Z"
  },
  "sequence": "12345"
}

EventMetadata also declares source and namespace, and the server sets neither — they arrive empty on every frame. Do not branch on them. timestamp is stamped when the frame is built, not when the event was published.

delivery_attempt is 1 on Subscribe: that RPC never redelivers. Consumer-group events count up on every redelivery.

Read data_value, not only data

data is a google.protobuf.Struct, which can only carry a JSON object. An event whose payload is a scalar, an array or null arrives in data_value (field 8) instead, and data is absent. A client that reads only data drops those events silently. Take data_value when present and fall back to data.

When include_metadata: true is set and the event has user-defined metadata (attached at emit time), it is included in the metadata object under the custom key:

{
  "subscription_id": "sub_abc123",
  "event_id": "evt_xyz789",
  "topic": "events:order.placed",
  "data": { "orderId": "123" },
  "metadata": {
    "timestamp": "2024-01-15T10:30:00.000Z",
    "custom": {
      "traceId": "abc-123",
      "source": "checkout"
    }
  },
  "sequence": "12346"
}

Join Consumer Group

Endpoint: POST /ironflow.v1.PubSubService/JoinConsumerGroup

Request:

{
  "namespace": "default",
  "group_name": "order-processors",
  "consumer_id": "consumer-1",
  "client_id": "my-app-instance"
}

Response: the same SubscriptionEvent frames as Subscribe, with two differences. subscription_id is a token the server mints for this member and sends only to it. The consumer_id in the request is a label, recorded as the member’s client ID when client_id is empty; it never appears in subscription_id. metadata is always sent, with timestamp and any custom values: include_metadata has no effect on this RPC. delivery_attempt is 1 on the first delivery and 2 on the first redelivery.

Acknowledging a consumer-group event

Endpoint: POST /ironflow.v1.PubSubService/AckEvent

AckEvent acknowledges an event that a MANUAL consumer group delivered. Only consumer groups hold events for acknowledgment. A subscription without a consumer group always acknowledges automatically.

Request:

Field	Type	Description
namespace	string	Namespace (default: "default"). The engine ignores it.
group_name	string	Consumer group name.
consumer_id	string	The subscription_id of the event. Only the member that received the event holds it; any other value is NotFound.
event_id	string	Event ID to acknowledge.
ack_type	AckType	ACK_TYPE_ACK, ACK_TYPE_NAK or ACK_TYPE_TERM. ACK_TYPE_UNSPECIFIED is rejected.
redeliver_delay_ms	int32	Delay before redelivery, in milliseconds. NAK only.

subscription_id on each event is the value to send as consumer_id.

Response: AckEventResponse, an empty message.

Errors:

Cause	Code
group_name, consumer_id or event_id is empty	InvalidArgument
redeliver_delay_ms is negative	InvalidArgument
ack_type is unspecified	InvalidArgument
The group does not exist in the caller’s environment	NotFound
The member is not joined to the group	NotFound
The event is not in flight (for example, already acknowledged)	Success

An acknowledgment for an event that is not in flight succeeds. A successful reply does not prove that the event was in flight. An unknown event_id, or a consumer that does not hold the event, also succeeds. A retry is safe.

Redelivery timing: For a MANUAL group, the group setting redeliver_delay_ms is the time that JetStream waits for an acknowledgment. When this time ends, JetStream redelivers the event on its own. A NAK carries its own redeliver_delay_ms. That value sets the delay before redelivery for that NAK.

A redelivered event carries the next delivery_attempt: 2 after the first NAK or timeout, 3 after the second.

Bidirectional streaming (consumer-group join and ack)

Endpoint: POST /ironflow.v1.PubSubService/SubscribeBidirectional

SubscribeBidirectional joins one consumer group and carries its acks on the same stream. The stream is the member. No ack token crosses the wire. The group must exist. Join and ack need events:subscribe, the same as JoinConsumerGroup. The stream needs HTTP/2, so a browser using fetch cannot use it. A browser keeps the WebSocket ack frame.

Each message holds one payload field.

Client to server:

Payload	Description
join	A JoinConsumerGroupRequest. It MUST be the first message, and it is allowed once.
ack	A SubscriptionAck: event_id, ack_type, redeliver_delay_ms (NAK only) and ack_id.

Server to client:

Payload	Description
joined	Sent once, before any event: subscription_id, group_name, pattern and ack_mode.
event	A SubscriptionEvent. Its subscription_id equals the one in joined.
ack_result	The answer to one ack: ack_id, event_id, code and message.
heartbeat	Sent after 30 seconds without a frame. The SDKs reconnect after 90 seconds with no frame of any kind.

The client picks a new ack_id for each ack. The server echoes it in ack_result. Event IDs repeat after a NAK, so the event ID cannot match a result. Results can arrive in any order relative to events.

Ack result codes:

Code	Meaning
ACK_RESULT_CODE_OK	The server applied the ack, or the event was not in flight. A retry is safe.
ACK_RESULT_CODE_INVALID_ARGUMENT	event_id is empty, ack_type is unspecified, or redeliver_delay_ms is negative. The stream stays open.
ACK_RESULT_CODE_INTERNAL	The NATS ack failed. The message is internal error (ref <id>); the server log has the cause under the same ref. A retry is safe.
ACK_RESULT_CODE_NOT_FOUND	The group or this member is gone, for example deleted. Do not retry: the stream ends, and the server redelivers the event to a member that joins.

Stream end:

Cause	Result
The first message is not join, a second join, an empty payload, or an empty group_name	InvalidArgument
No first message within 10 seconds	DeadlineExceeded
The group does not exist in the caller’s environment	NotFound
The client half-closes	The member leaves the group. The stream ends OK.
The group is deleted	The stream ends OK. A new join gets NotFound.

On an AUTO group, an ack gets OK and changes nothing, because the server acked the event at delivery. When a member leaves, its held events go back to the group after the group’s ack wait, as on the other ack transports.

Acks on the stream always reach the member’s node.

The stream needs a server that includes this change (released after v0.47.0). Against an older server the join fails with a SERVER_TOO_OLD error that names the first server version with the stream (0.48.0). The Go and Node SDKs do not retry it. Use the WebSocket transport there.

Go:

sub, err := client.JoinConsumerGroup(ctx, "order-processors",
    ironflow.WithJoinTransport("grpc"))
if err != nil {
    log.Fatal(err)
}
defer sub.Unsubscribe()

for event := range sub.Events() {
    if err := process(event); err != nil {
        _ = sub.Nak(event.ID, 10*time.Second)
        continue
    }
    if err := sub.Ack(event.ID); errors.Is(err, ironflow.ErrStaleAck) {
        continue // the stream reconnected; the server redelivers the event
    }
}

Ack, Nak and Term wait for the server result, up to 30 seconds. A refused ack returns *ironflow.AckError with Code "invalid_argument", "not_found" or "internal". A NAK delay outside 0 to 2³¹−1 ms returns "invalid_argument" without a send. WebSocket stays the default transport.

Node:

import { joinConsumerGroupStream } from "@ironflow/node/consumer-group-stream";

const sub = await joinConsumerGroupStream(
  { serverUrl: "http://localhost:9123" },
  "order-processors",
  {
    onEvent: async (event) => {
      await process(event);
      await sub.ack(event.eventId!);
    },
  },
);

The SDK holds events until the call returns, so sub is set when onEvent runs. onEvent is awaited, so events are handled in order. A handler that throws goes to onError, and the SDK does not ack the event. Import AckError, StaleAckError and SubscriptionClosedError from the same subpath to tell ack failures apart. Options and the full contract are in the Node SDK reference.

Python: has no wrapper yet. Join with join_consumer_group and ack with ack_event.

Using with cURL

# Subscribe to events (streaming response)
curl -X POST http://localhost:9123/ironflow.v1.PubSubService/Subscribe \
  -H "Content-Type: application/json" \
  -d '{
    "pattern": "system.run.>",
    "options": {
      "replay": 10,
      "include_metadata": true
    }
  }'

# Join a consumer group
curl -X POST http://localhost:9123/ironflow.v1.PubSubService/JoinConsumerGroup \
  -H "Content-Type: application/json" \
  -d '{
    "group_name": "order-processors"
  }'

---

Error Codes

These string codes are the WebSocket vocabulary — they arrive in the code field of a subscription_error, of an errored subscription_result entry, or of a connection-level error frame. ConnectRPC does not use the domain codes below: it returns standard Connect codes (invalid_argument, internal, not_found, unavailable, unimplemented) with the detail in the error message. An internal message is internal error (ref <id>); the server log has the cause under the same ref. The generic codes invalid_argument, internal and unavailable are the same lowercase Connect codes on both transports.

Code	Description	Recoverable
INVALID_PATTERN	Pattern syntax error or invalid namespace prefix, or a malformed subscribe/unsubscribe frame	No
invalid_argument	Option combination the server refuses: startAfterSequence with replay, or startAfterSequence with consumerGroup	No
LIMIT_EXCEEDED	More than 100 subscriptions on one connection, plain and consumer-group together	No
internal	Server-side failure, or pub/sub not configured	Maybe
unavailable	A consumer-group subscribe is refused because the PUBSUB stream has a fault. The message names the stream and the fault, such as stream PUBSUB: no leader. See Recover a stuck JetStream stream	Yes, once the fault clears
PARSE_ERROR	The frame is not valid JSON. Always on an error frame	No
UNKNOWN_TYPE	The frame’s type is not one the server handles. Always on an error frame	No

A replay above 1000 events is refused too, but not as LIMIT_EXCEEDED: it arrives with the message replay limit exceeded, as internal on WebSocket and as invalid_argument on ConnectRPC.

Two more codes are declared in the wire vocabulary but are not sent by the server today. Clients that already branch on them can keep doing so; nothing will produce them:

Code	Status
INVALID_NAMESPACE	Reserved. Namespace-prefix errors arrive as INVALID_PATTERN instead.
NATS_DISCONNECT	Reserved. The server does not surface NATS connection loss on this channel.

Note

RESUBSCRIBE_FAILED is not a server error code — it is generated client-side by the SDK when a subscription fails to restore after reconnection.

---

Choosing a Protocol

Consideration	WebSocket	ConnectRPC
Browser support	Native	Requires library
Bidirectional	ack frame acks consumer-group events; optional subscriptionId	SubscribeBidirectional joins a group and acks on one stream (Go, Node). AckEvent also works.
gRPC ecosystem	No	Yes
Proxy compatibility	May need configuration	Standard HTTP/2
SDK support	Browser, Node, Go SDKs	Browser, Go, Python SDKs; Node for consumer-group streams

Recommendation:

• Use @ironflow/browser for browser clients — ConnectRPC is the default; call detectTransport() and pass its result to configure() when WebSocket fallback is needed
• Use ConnectRPC for server-to-server communication (via Go SDK or direct HTTP)