Consume changes durably

The cursor in a Leani change envelope is an opaque resume and acknowledgement token. Store it beside the application state derived from that change. The destination transaction is the durable boundary; the later Leani acknowledgement advances retention and delivery state.

1. Create a stable named consumer once, or inspect the existing consumer after a restart.
2. Poll retained changes under a lease.
3. Apply one change idempotently and store its cursor in the same destination transaction. Key both by the consumer’s stream as well as the sequence: a node whose store is replaced numbers its changes from the start again.
4. Commit the destination transaction.
5. Acknowledge that cursor to Leani. If this request is lost, accept the same change again and rely on idempotent application.

Durable SDK consumer

import {
  LeaniError,
  createLeaniClient,
  isResetRequired,
  type ChangeEnvelope,
  type DurableConsumer,
} from "@smart-byte/leani-sdk";

export interface DestinationTransaction<T> {
  /** Must be idempotent because a committed change can be redelivered. */
  apply(change: ChangeEnvelope<T>): Promise<void>;
  /** Store this beside application state in the same transaction. */
  storeLeaniCursor(cursor: string, sequence: string): Promise<void>;
}

export interface Destination<T> {
  /**
   * `stream` names the durable consumer in the node's store. Sequences are
   * unique only within it: a node whose store is replaced numbers its
   * changes anew, and its consumer is created again.
   */
  transaction(
    stream: string,
    work: (transaction: DestinationTransaction<T>) => Promise<void>,
  ): Promise<void>;
}

export interface ConsumerOptions {
  baseUrl: string;
  processor: string;
  consumer: string;
  credential: string;
  leaseTtlSeconds?: number;
  signal?: AbortSignal;
}

export async function consumeDurably<T>(
  destination: Destination<T>,
  options: ConsumerOptions,
): Promise<void> {
  const leani = createLeaniClient({ baseUrl: options.baseUrl });
  const consumers = leani.processors.consumers;

  let consumer: DurableConsumer;
  try {
    consumer = await consumers.inspect(options.processor, options.consumer, {
      signal: options.signal,
    });
  } catch (error) {
    if (!(error instanceof LeaniError) || error.status !== 404) throw error;
    // Only an until_acknowledged stream accepts a required consumer that
    // fences pruning; a window stream prunes by its own limits.
    consumer = await consumers.create(options.processor, {
      id: options.consumer,
      role: "best_effort",
      start: { position: "earliest_retained" },
      leaseTtlSeconds: options.leaseTtlSeconds ?? 60,
      credential: options.credential,
      signal: options.signal,
    });
  }
  // The consumer's creation identifies the store it lives in.
  const stream =
    `${consumer.processorInstance}/${consumer.id}@${consumer.createdAtUnixMs}`;

  while (!options.signal?.aborted) {
    const page = await consumers.changes<T>(
      options.processor,
      options.consumer,
      options.credential,
      { limit: 256, signal: options.signal },
    );

    if (page.data.length === 0) {
      await consumers.renew(
        options.processor,
        options.consumer,
        options.credential,
        { signal: options.signal },
      );
      await new Promise((resolve) => setTimeout(resolve, 500));
      continue;
    }

    for (const change of page.data) {
      if (isResetRequired(change)) {
        throw new Error("retained history no longer covers this consumer; rebuild from a snapshot");
      }

      // Keep the lease alive while a fetched page is being applied. A slow
      // destination must not rely on the next acknowledgement for renewal.
      await consumers.renew(
        options.processor,
        options.consumer,
        options.credential,
        { signal: options.signal },
      );

      await destination.transaction(stream, async (transaction) => {
        await transaction.apply(change);
        await transaction.storeLeaniCursor(change.cursor, change.sequence);
      });

      // Acknowledge only after application state and its cursor commit together.
      await consumers.acknowledge(
        options.processor,
        options.consumer,
        options.credential,
        change.cursor,
        { signal: options.signal },
      );
    }
  }
}

Commit application state and its Leani cursor atomically before acknowledging.Verified source

Never acknowledge before your commit

An early acknowledgement can make Leani eligible to discard history your application has not durably applied. A failed acknowledgement after commit is safe: the event may be delivered again, so application writes must be idempotent.

The consumer’s credential goes with every call, and Leani accepts an acknowledgement only for a change it delivered to that consumer. This loop uses the generic consumer routes, which serve processors with canonical delivery ordering. A processor with a split live stream (block_versioned_idempotent, such as blobs-money) fences renewal and acknowledgement by streaming session: read it with the SDK’s streamLive() or subscribe() from @smart-byte/leani-sdk/backfill, and acknowledge each batch’s boundary.