Skip to content
LeaniLeani

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.
  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,
} from "@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> {
  transaction(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;


  try {
    await consumers.inspect(options.processor, options.consumer, {
      signal: options.signal,
    });
  } catch (error) {
    if (!(error instanceof LeaniError) || error.status !== 404) throw error;
    await consumers.create(options.processor, {
      id: options.consumer,
      role: "required",
      start: { position: "earliest_retained" },
      leaseTtlSeconds: options.leaseTtlSeconds ?? 60,
      credential: options.credential,
      signal: options.signal,
    });
  }


  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(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