Event Streaming & Resilient Consumer Playbook

Building distributed, event-driven microservices often brings architectural dilemmas: When is a dedicated message broker necessary, and when does it introduce accidental complexity? How do you stream events to downstream services reliably without data loss? And how do you handle Kubernetes consumer replica sets without duplicate processing or race conditions?

This playbook provides an architectural decision framework and bounded implementation sketches for Alvyn, Server-Sent Events (SSE), and Transactional Outbox relays. The sketches are not a production-ready HA consumer framework; their limits and required operational safeguards are explicit below.

1. Architectural Blueprint: Direct Streaming vs. Message Brokers

A common reflex in event-driven architecture is to place an external message broker (like Apache Kafka, RabbitMQ, or NATS) between services as soon as events need to be shared. While brokers solve critical problems at scale, they also introduce operational overhead, client library dependencies, and potential consistency hurdles.

In an event-sourced architecture powered by Alvyn, PostgreSQL acts as an immutable, ordered, append-only event log. Understanding the trade-offs helps you choose the right communication model:

Communication Models: Fan-Out vs. Competing Consumers

When to Use Direct HTTP Streaming (Alvyn + SSE)

• CQRS Read-Model Projections: Replicating domain events to external databases (e.g., Elasticsearch, Redis, MongoDB).
• Downstream Microservices & 3rd-Party APIs: Allowing external partners or internal microservices to consume events using standard web standards without configuring VPNs or broker client libraries.
• Low-to-Medium Fleet Complexity: When consumers can process events using in-process concurrency or active-passive replica sets.
• Operational Simplicity: When you want to minimize the number of moving parts in your infrastructure stack.

When to Introduce a Dedicated Message Broker

• High-Volume Competing Consumers: When hundreds of independent worker pods must pull work items from a shared queue in round-robin fashion.
• Big Data & Analytics Pipelines: Ingesting 200,000+ events per second into stream-processing engines like Apache Flink or Apache Spark (ideal for Apache Kafka).
• Complex Transactional Task Queuing: Fine-grained per-message TTLs, priority queues, and dead-letter exchanges (ideal for RabbitMQ).
• Ultra-Low Latency RPC & Edge Computing: Sub-millisecond synchronous request-reply and edge mesh synchronization via leaf nodes (ideal for NATS JetStream).

2. Core Primitives & Guarantees (Demystified)

Before implementing the endpoints, distinguish Alvyn's ordered read cursor from application-owned durable checkpoints, retries, and side effects.

Concept 1: The Monotonic Bookmark (globalPosition)

Every event written to Alvyn receives a database-generated BIGSERIAL sequence number called globalPosition.

globalPosition is a unique sequence position, not transaction commit order or wall-clock order; gaps are normal. Alvyn's subscription watermark prevents advancing past unresolved lower positions. A checkpoint of 10522 acknowledges matching events through that position only if the consumer has successfully processed the entire delivered prefix. A completed parallel task alone is not a safe checkpoint.

Concept 2: The W3C SSE Standard (Last-Event-ID)

The W3C Server-Sent Events standard includes built-in resumption mechanics. When the server pushes an event, it includes an id: line:

id: 10522
event: OrderPlaced
data: {"orderId":"ORD-99","total":89.99}

When a network drop occurs, compliant HTTP clients automatically attach the Last-Event-ID header upon reconnecting:

GET /events HTTP/1.1
Host: api.example.com
Accept: text/event-stream
Last-Event-ID: 10522

Concept 3: Alvyn's lowerBound Parameter

When invoking eventStore.subscribe(), you supply lowerBound:

const stream = eventStore.subscribe({
  lowerBound: { id: "10522", type: "exclusive" },
});

This tells Alvyn: "Resume immediately after position 10522 (global_position > 10522)."

Concept 4: The Zero-Gap Single Query Loop

Alvyn does not maintain separate systems for "historical catch-up" and "live listening". Instead, it executes a continuous cursor query against PostgreSQL:

SELECT * FROM events
WHERE global_position > $1 AND global_position <= $2
ORDER BY global_position ASC
LIMIT 500;

1. Catch-Up Phase: While the consumer's cursor ($1) is behind the database head, Alvyn fetches full 500-event batches via primary key index scans.
2. Live Transition: When a query returns fewer than 500 events, the consumer is caught up. Alvyn pauses on an internal waker tied to PostgreSQL LISTEN/NOTIFY (with a periodic polling fallback).
3. Instant Wake-Up: The instant a new transaction commits, the waker triggers the exact same query from the updated cursor.

Concept 5: The Commit-Safe Watermark

In PostgreSQL, sequence numbers are allocated during insertion, but transactions commit in non-deterministic order:

Tx A (globalPosition 101) -------------> commits at t=2
Tx B (globalPosition 100) --------------------> commits at t=3 (delayed)

If a subscriber tailing at t=2 read position 101, advancing its cursor to 101, it would permanently miss event 100 when Tx B commits at t=3.

Alvyn eliminates this via computeSafeWatermark. The query upper bound ($2) only advances up to the safe watermark, holding back positions until all preceding concurrent transactions have either committed or aborted.

3. Step-by-Step Producer Implementation (HTTP SSE with Alvyn)

This minimal Express / Node.js endpoint illustrates bookmarks, disconnect cleanup, and heartbeats. Before deployment, add authentication and per-subject authorization, validate decimal cursor input before sending headers, bound slow-client buffering with backpressure/timeouts, and test proxy/disconnect behavior. It does not implement these production safeguards:

import type { Request, Response } from "express";
import { eventStore } from "./event-store";

export async function sseEventsHandler(req: Request, res: Response) {
  // 1. Set required SSE headers
  res.writeHead(200, {
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache, no-transform",
    Connection: "keep-alive",
    "X-Accel-Buffering": "no", // Disable buffering in Nginx / AWS ALB
  });
  res.flushHeaders();

  // 2. Extract Last-Event-ID from standard header or query param fallback
  const lastEventId = (req.headers["last-event-id"] ||
    req.query.lastEventId) as string | undefined;

  // 3. AbortController to cleanly release PostgreSQL connections on disconnect
  const abortController = new AbortController();
  res.on("close", () => {
    abortController.abort();
  });

  // 4. Heartbeat interval to prevent intermediate load balancers from dropping idle connections
  const heartbeatTimer = setInterval(() => {
    if (!res.writableEnded) {
      res.write(": keepalive\n\n");
    }
  }, 15000);

  try {
    // 5. Start Alvyn subscription from the client's bookmark
    const eventStream = eventStore.subscribe({
      lowerBound: lastEventId
        ? { id: lastEventId, type: "exclusive" }
        : undefined,
      signal: abortController.signal,
    });

    // 6. Stream events formatted according to W3C SSE standard
    for await (const event of eventStream) {
      res.write(`id: ${event.globalPosition.toString()}\n`);
      res.write(`event: ${event.type}\n`);
      res.write(
        `data: ${JSON.stringify({ ...event, globalPosition: event.globalPosition.toString() })}\n\n`,
      );
    }
  } catch (error: any) {
    if (!abortController.signal.aborted) {
      console.error("SSE stream error:", error);
      res.end();
    }
  } finally {
    clearInterval(heartbeatTimer);
    abortController.abort();
    if (!res.writableEnded) res.end();
  }
}

4. Step-by-Step Consumer Implementation (Resilient Client)

The following single-process demonstration uses a local file and parses only the single-line LF-delimited format emitted above. It is not an HA checkpoint store or a general SSE parser. Local checkpoint loss causes replay; downstream effects must be idempotent. Use a durable shared database checkpoint for failover and a standards-compliant parser with bounded frame size in production:

import fs from "node:fs/promises";
import path from "node:path";

const CHECKPOINT_PATH = path.resolve("./consumer.checkpoint");

// Load the last saved globalPosition
async function loadCheckpoint(): Promise<string | undefined> {
  try {
    return (await fs.readFile(CHECKPOINT_PATH, "utf-8")).trim();
  } catch (error) {
    if ((error as NodeJS.ErrnoException).code === "ENOENT") return undefined;
    throw error; // Do not silently reset progress on I/O errors
  }
}

// Persist checkpoint after successful handling
async function saveCheckpoint(position: string): Promise<void> {
  await fs.writeFile(`${CHECKPOINT_PATH}.tmp`, position, "utf-8");
  await fs.rename(`${CHECKPOINT_PATH}.tmp`, CHECKPOINT_PATH);
}

export async function startResilientConsumer(sseUrl: string) {
  let backoffMs = 1000;

  while (true) {
    const lastEventId = await loadCheckpoint();
    const controller = new AbortController();
    console.log(
      `Connecting to ${sseUrl} (Last-Event-ID: ${lastEventId ?? "BEGINNING"})...`,
    );

    try {
      const response = await fetch(sseUrl, {
        signal: controller.signal,
        headers: {
          Accept: "text/event-stream",
          ...(lastEventId ? { "Last-Event-ID": lastEventId } : {}),
        },
      });

      if (!response.ok || !response.body) {
        throw new Error(
          `HTTP error ${response.status}: ${response.statusText}`,
        );
      }

      // Reset backoff upon successful connection
      backoffMs = 1000;

      const reader = response.body.getReader();
      const decoder = new TextDecoder();
      let buffer = "";

      while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });
        if (buffer.length > 1_048_576)
          throw new Error("SSE buffer limit exceeded");
        const chunks = buffer.split("\n\n");
        buffer = chunks.pop() ?? "";

        for (const rawChunk of chunks) {
          if (rawChunk.startsWith(":")) continue; // Ignore keepalive heartbeats

          const lines = rawChunk.split("\n");
          let eventId = "";
          let eventType = "";
          let eventDataRaw = "";

          for (const line of lines) {
            if (line.startsWith("id: ")) eventId = line.slice(4).trim();
            else if (line.startsWith("event: "))
              eventType = line.slice(7).trim();
            else if (line.startsWith("data: "))
              eventDataRaw = line.slice(6).trim();
          }

          if (eventId && eventDataRaw) {
            const event = JSON.parse(eventDataRaw);

            // Execute domain business logic (must be idempotent)
            await handleDomainEvent(eventType, event);

            // Acknowledge checkpoint
            await saveCheckpoint(eventId);
          }
        }
      }
      throw new Error("SSE stream ended"); // Reconnect with backoff on clean EOF too
    } catch (error) {
      controller.abort();
      const jitterMs = Math.floor(Math.random() * 300); // 0-300ms random offset to prevent thundering herds
      const sleepMs = backoffMs + jitterMs;
      console.error(
        `SSE connection dropped. Retrying in ${sleepMs}ms...`,
        error,
      );
      await new Promise((resolve) => setTimeout(resolve, sleepMs));
      backoffMs = Math.min(backoffMs * 2, 30000); // Exponential backoff capped at 30s
    } finally {
      controller.abort();
    }
  }
}

async function handleDomainEvent(type: string, data: any) {
  // Domain business logic here (e.g. update local read model)
}

5. Multi-Instance Deployments & High-Availability Patterns

In production cloud environments (such as Kubernetes, AWS ECS, or Nomad), services are rarely deployed as single, isolated instances. Multiple replicas are standard for several reasons:

1. High Availability & Zero-Downtime: Running 2–3 replicas across availability zones ensures that rolling deployments, node maintenance, or spontaneous pod restarts do not disrupt service availability.
2. Hybrid / Multi-Purpose Services: A microservice often serves incoming REST or GraphQL traffic from web/mobile clients behind an ingress load balancer, while simultaneously running a background event consumer to update local caches, index search data, or trigger asynchronous workflows.
3. High I/O & Compute Throughput: Processing tasks (such as PDF generation, third-party API synchronization, or complex aggregations) require concurrent execution across CPU cores.

Because HTTP Server-Sent Events operates on standard HTTP streams, the server delivers events to every connected client (Fan-Out / Broadcast). To structure multi-instance deployments cleanly, choose the architectural pattern that aligns with your operational goals:

Architectural Decision Matrix

Pattern A: Active-Passive Leader Election (HA & Hybrid Services)

Use Case: You run 2–3 replicas of a service for High Availability or as a Hybrid Service (serving HTTP/REST traffic behind a load balancer while running a background event listener). You want the service to survive pod restarts without executing background event tasks multiple times in parallel.

An advisory lock alone is not a fenced lease. If the lock connection fails while the SSE connection or an external request remains alive, an old owner can keep working while a new owner acquires the lock. The local file consumer above cannot provide shared HA progress and must not be wrapped in a lock loop and called production-safe.

Bounded Recipe and Limits

1. For a PostgreSQL read model, prefer runProjection() and perform all writes through its transaction client. Checkpoint row locking and data updates then share a transaction without a custom SSE leader.
2. For external effects, prefer processOutbox() with durable broker acknowledgements, retries, and destination-side deduplication. Delivery remains at-least-once.
3. If an SSE leader is necessary, implement a separate owner lifecycle: acquire ownership on a dedicated connection, load a shared durable checkpoint, and pass cancellation through the fetch, parser, worker queues, and handlers. On connection loss or shutdown, stop accepting work and await/cancel in-flight tasks before releasing ownership. Never return a still-locked session to a pool.
4. Persist only the successfully handled contiguous prefix. Where possible, commit local effects and the checkpoint in one transaction. Require destination-enforced monotonic fencing tokens to reject stale owners, or idempotency keys where fencing is unavailable; cancellation alone cannot undo an already issued external request.
5. Test lock-session loss while SSE stays connected, process pauses, partial external success, checkpoint failures, and rolling restarts. Define retry, timeout, and replay budgets. Failover latency depends on failure detection; neither instant failover nor zero duplicates is promised.

This page deliberately does not supply an HA lock wrapper: Alvyn's SSE subscription does not implement a distributed ownership/fencing protocol for your external system.

Pattern B: In-Process Hash-Partitioned Worker Pool (High I/O Concurrency)

Use Case: A dedicated consumer pod needs to process a high volume of events where individual task execution is I/O-heavy (e.g., rendering PDFs, making external REST calls, or writing to downstream stores).

In most architectures, network transmission of JSON events over SSE is extremely fast (tens of thousands of events per second on a single connection); the bottleneck is downstream I/O. Instead of opening duplicate connections, maintain one SSE connection per consumer pod and dispatch events in-memory across an internal pool of worker queues by hashing the streamId (hash(streamId) % numberOfWorkers):

• Per-Entity Input Order: Events for Order-123 are routed to the same queue and executed in delivery order, not wall-clock order.
• I/O Concurrency: Different partitions can overlap asynchronous I/O. Promise queues do not provide multi-core CPU execution or eliminate races on shared resources.

import { createHash } from "node:crypto";

export class HashPartitionedWorkerPool {
  private queues: Array<Promise<void>> = [];

  constructor(private concurrency: number = 16) {
    if (!Number.isSafeInteger(concurrency) || concurrency < 1) {
      throw new Error("concurrency must be a positive safe integer");
    }
    this.queues = Array.from({ length: concurrency }, () => Promise.resolve());
  }

  private getPartition(key: string): number {
    const hash = createHash("md5").update(key).digest().readUInt32BE(0);
    return hash % this.concurrency;
  }

  public enqueue(streamId: string, task: () => Promise<void>): Promise<void> {
    const partition = this.getPartition(streamId);

    // Chain the task onto the specific queue for this streamId
    const chain = this.queues[partition].then(task);

    this.queues[partition] = chain;
    return chain;
  }
}

The returned promise rejects on failure; the affected partition stays failed so later work cannot pass the failed event. Observe every returned promise immediately. Bound each dispatch batch and await all outcomes before advancing the checkpoint:

// batch is a bounded, ordered list of consecutive delivered events
const outcomes = await Promise.allSettled(
  batch.map((event) => workers.enqueue(event.streamId, () => handle(event))),
);
const failed = outcomes.find((outcome) => outcome.status === "rejected");
if (failed?.status === "rejected") throw failed.reason;
if (batch.length > 0) {
  await saveCheckpoint(batch[batch.length - 1].globalPosition.toString());
}

Do not dispatch another batch after failure: stop intake, settle in-flight work, recreate the pool, and replay from the last durable checkpoint. Successful external effects in a failed batch may repeat, so handlers must be idempotent. The queue class alone does not bound memory; the caller must bound batch size and avoid accumulating unawaited batches.

Pattern C: Database-Level Sharding via Subject / Type Filters (Domain & Regional Partitioning)

Use Case: Workloads are naturally segregated by business domain or geographic region, and you want each pod in your fleet to handle a dedicated slice of the event catalog.

Partition the stream at the database query level using Alvyn's subject prefix or eventTypes filters:

// Dedicated pod processing only European orders
const stream = eventStore.subscribe({
  subject: "Order-EU-",
  recursive: true,
  lowerBound: { id: lastProcessedEuId },
});

Each logical shard needs a durable checkpoint and explicit ownership. Filters must be disjoint to avoid overlap; replicas with the same filter still receive the same events. Retries can duplicate effects, and changing shard filters requires a planned checkpoint/replay migration.

Pattern D: Competing Consumers via Message Broker (Alvyn Transactional Outbox)

Use Case: You require automatic, dynamic partition rebalancing across an elastic fleet of dozens or hundreds of worker pods without maintaining manual shard filters or leader locks.

Bridge Alvyn to Apache Kafka, RabbitMQ, or NATS JetStream using Alvyn's Transactional Outbox, as detailed in the next section.

6. The Hybrid Bridge: Relaying to Message Brokers via Alvyn Outbox

When your architecture requires true competing consumers across dozens of dynamically scaled pods, or integrates with an existing enterprise Kafka / RabbitMQ / NATS cluster, use Alvyn's Transactional Outbox.

This pattern eliminates the Dual-Write problem by atomically storing outbox entries within the same database transaction as the domain event:

1. Atomic Transaction: Domain events and outbox messages are committed together in one transaction.
2. Non-Blocking Polling: Background workers call eventStore.processOutbox(), which fetches pending messages using FOR UPDATE SKIP LOCKED.
3. Broker Publishing with ACK: Events are dispatched to Kafka, RabbitMQ, or NATS. Entries are marked processed only when the handler promise resolves.

Outbox Publisher Implementation

import { EventStore } from "alvyn";
import { kafkaProducer } from "./kafka-client"; // or RabbitMQ / NATS

export async function startOutboxRelay(
  eventStore: EventStore,
  signal: AbortSignal,
) {
  while (!signal.aborted) {
    const processed = await eventStore.processOutbox(async (entries) => {
      const messages = entries.map((entry) => ({
        key: entry.topic,
        value: JSON.stringify(entry.payload),
        headers: {
          globalPosition: entry.eventGlobalPosition.toString(),
        },
      }));

      // Send batch to Kafka topic with at-least-once delivery guarantee
      await kafkaProducer.send({
        topic: "domain-events",
        messages,
      });
    }, 100);

    if (processed === 0) {
      await new Promise((resolve) => setTimeout(resolve, 250));
    }
  }
}

// Delete processed outbox rows older than 7 days
const deletedCount = await eventStore.cleanupOutbox(7 * 24 * 60 * 60 * 1000);

7. Observability, Monitoring & Disaster Recovery

The relay example assumes a KafkaJS-compatible producer. Retry failed processOutbox() batches with bounded backoff in your supervisor. A broker acknowledgement followed by a failed PostgreSQL commit causes redelivery; the row lock is not an exactly-once guarantee for external effects.

Monitoring Consumer Lag (SQL)

To detect if a consumer is falling behind during high-traffic spikes, query the lag directly in PostgreSQL:

SELECT
  MAX(global_position) AS head_position,
  $1::BIGINT AS consumer_checkpoint,
  (MAX(global_position) - $1::BIGINT) AS consumer_lag
FROM event_store.events;

• This is a position-distance estimate, not an event count: sequences have gaps, filters exclude events, and the visible head can exceed the safe watermark. Use your configured schema in place of event_store.
• Alert on sustained lag growth and processing latency against a measured service objective; investigate slow handlers and watermark stalls before increasing concurrency.

Full Event Replays (Disaster Recovery & New Services)

When deploying a brand new downstream microservice (or recovering from a corrupted database), set the consumer checkpoint to 0 or omit Last-Event-ID.

Alvyn will stream every historical event from the beginning of time in 500-event chunks, then automatically transition into the live stream once caught up.

Idempotency Checklist

Because network retries guarantee at-least-once delivery, make your downstream handlers idempotent:

1. Deduplication Key: Store the processed globalPosition or event id in a unique PostgreSQL table (processed_events).
2. Transactional State Updates: Commit your local domain state change and the consumer checkpoint in the same database transaction.
3. Monotonic Version Checks: When updating read models, ensure updates are only applied if event.streamVersion > current_entity_version.