Reliable Consumers

Naven Events uses leases and at-least-once delivery. A consumer must assume that it can receive the same Event again after a crash, network timeout, or lost acknowledgement.

Consumer loop

const navenApiUrl = process.env.NAVEN_API_URL!;
const navenApiKey = process.env.NAVEN_API_KEY!;
const queueId = process.env.NAVEN_QUEUE_ID!;

type Envelope<T> = {
  code: number;
  message: string;
  data: T;
};

async function navenRequest<T>(
  path: string,
  init: RequestInit,
): Promise<T> {
  const response = await fetch(`${navenApiUrl}${path}`, {
    ...init,
    headers: {
      Authorization: `Bearer ${navenApiKey}`,
      "Content-Type": "application/json",
      ...init.headers,
    },
  });

  const body = await response.json() as Envelope<T>;
  if (!response.ok || body.code !== 0) {
    throw new Error(`${response.status}: ${body.message}`);
  }
  return body.data;
}

async function consumeForever() {
  for (;;) {
    const claim = await navenRequest<{
      event: {
        id: string;
        scheduleId: string | null;
        payload: Record<string, unknown>;
        scheduledAt: string;
        attemptCount: number;
      };
      leaseToken: string;
      leaseExpiresAt: string;
    } | null>(`/v1/event-queues/${queueId}/claim`, {
      method: "POST",
      body: JSON.stringify({
        leaseSeconds: 900,
        waitSeconds: 20,
      }),
    });

    if (!claim) continue;
    await processClaim(claim);
  }
}

Automatic heartbeat

Heartbeat more frequently than the lease duration. One third of the lease is a reasonable starting interval:

async function processClaim(claim: {
  event: {
    id: string;
    payload: Record<string, unknown>;
    scheduledAt: string;
  };
  leaseToken: string;
}) {
  const heartbeat = setInterval(() => {
    void navenRequest(`/v1/events/${claim.event.id}/heartbeat`, {
      method: "POST",
      body: JSON.stringify({
        leaseToken: claim.leaseToken,
        leaseSeconds: 900,
      }),
    });
  }, 300_000);

  try {
    await executeIdempotently(claim.event);

    await navenRequest(`/v1/events/${claim.event.id}/ack`, {
      method: "POST",
      body: JSON.stringify({ leaseToken: claim.leaseToken }),
    });
  } catch (error) {
    await navenRequest(`/v1/events/${claim.event.id}/nack`, {
      method: "POST",
      body: JSON.stringify({
        leaseToken: claim.leaseToken,
        error: error instanceof Error ? error.message : "Execution failed",
        retryDelaySeconds: 30,
      }),
    });
  } finally {
    clearInterval(heartbeat);
  }
}

If a heartbeat or acknowledgement returns 409, the lease expired or was reassigned. Stop applying new side effects and reconcile any operation that may already have succeeded.