Manual Acknowledgment

Manual acknowledgment ensures at-least-once delivery by requiring explicit confirmation that an event was processed. This is essential for critical reliability patterns like payment processing or database synchronization.

Note

Manual ack needs a consumer group. Manual ack works for MANUAL consumer groups. The Go, Node and browser SDKs send ack, nak and term over the WebSocket, and the server applies them. The browser SDK needs transport: "websocket", because its default ConnectRPC transport cannot send them. Python calls the AckEvent RPC. Go and Node can also join and ack over one ConnectRPC stream; see Ack over one stream. See Protocols for the frame format. A plain subscription without a consumer group always acknowledges automatically.

Ack Modes

Mode	Behavior
auto	Events are acknowledged automatically upon delivery.
manual	You must explicitly call ack(), nak(), or term() for each event.

---

Acknowledgment Methods

Method	Description
ack(eventId)	Success: Confirms processing is complete.
nak(eventId, delay?)	Retry: Requests redelivery (optionally after a delay).
term(eventId)	Terminal Failure: Permanently rejects the event (no redelivery).

---

Example Usage

The examples join a consumer group. The group must already exist with ackMode set to manual. A browser join to a group that does not exist creates it in auto mode, and acks on an auto group do nothing. See Creating Consumer Groups.

TypeScript

import { ironflow, type AckableSubscription } from "@ironflow/browser";

// The default ConnectRPC transport cannot send acks. Use the WebSocket transport.
ironflow.configure({ serverUrl: "http://localhost:9123", transport: "websocket" });

// `joinConsumerGroup()` sets `consumerGroup` and `ackMode: "manual"` for you.
let sub: AckableSubscription;
sub = await ironflow.joinConsumerGroup("order-workers", "events:payment.>", {
  onEvent: async (event) => {
    try {
      await processPayment(event.data);
      // Use the 'sub' object to acknowledge
      await sub.ack(event.eventId);
    } catch (err) {
      // Retry in 5 seconds
      await sub.nak(event.eventId, 5000);
    }
  },
});

Go

// Join an existing MANUAL consumer group
sub, err := client.JoinConsumerGroup(ctx, "order-workers")
if err != nil {
    log.Fatal(err)
}
defer sub.Unsubscribe()

for event := range sub.Events() {
    if err := processPayment(event.Data); err != nil {
        sub.Nak(event.ID, 5*time.Second)
    } else {
        sub.Ack(event.ID)
    }
}

Python

import os

from ironflow import IronflowRPC
from ironflow.rpc.v1 import AckEventRequest, AckType, JoinConsumerGroupRequest

with IronflowRPC(api_key=os.environ["IRONFLOW_API_KEY"]) as rpc:
    for event in rpc.pubsub.join_consumer_group(JoinConsumerGroupRequest(group_name="order-workers")):
        kind = AckType.ACK if process(event) else AckType.NAK
        rpc.pubsub.ack_event(AckEventRequest(
            group_name="order-workers",
            consumer_id=event.subscription_id,
            event_id=event.event_id,
            ack_type=kind,
            redeliver_delay_ms=5000,
        ))

Manual acknowledgment applies to MANUAL consumer groups. A subscription without a consumer group always acknowledges automatically. Send event.subscription_id as consumer_id. See Protocols for the request fields and error codes.

Ack over one stream (Go and Node)

Go and Node can join a MANUAL group over PubSubService.SubscribeBidirectional. The stream is the group member, and it carries the events and the acks. Each ack, nak or term waits for the server result, so a failed ack returns an error instead of a silent loss. WebSocket stays the default transport.

Go

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

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

Node

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

const sub = await joinConsumerGroupStream(
  { serverUrl: "http://localhost:9123", apiKey: process.env.IRONFLOW_API_KEY },
  "order-workers",
  {
    onEvent: async (event) => {
      try {
        await processPayment(event.data);
      } catch {
        await sub.nak(event.eventId!, 5000);
        return;
      }
      await sub.ack(event.eventId!);
    },
  },
);

The Node SDK holds events until joinConsumerGroupStream returns, so sub is set when onEvent runs. The stream needs Ironflow server v0.48.0 or later. 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 SDK does not retry it. Use the WebSocket transport there.

If the stream drops, the SDK reconnects with backoff. The member ID is new after a reconnect, so an ack for an event of the old stream fails with ErrStaleAck (Go) or StaleAckError (Node). The server redelivers that event. The browser cannot stream both ways, so it keeps the WebSocket ack frame. See Protocols for the frames and result codes.

Best Practices

1. Timeout Handling: Ensure your processing logic completes before the server’s visibility timeout.
2. Idempotency: Always implement idempotency in your onEvent handlers, as network issues can cause duplicate deliveries before an ack is received.
3. Use TERM for Invalid Data: Don’t use nak() for validation errors; use term() to prevent an infinite retry loop of bad data.