API index and search · Build metadata

Source snapshot

packages/remote/src/mutation-queue.ts

1// The client-side mutation queue for daemon/serverless deployments. The optimistic engine2// requires mids to arrive at the daemon CONTIGUOUSLY (`mid == lmid + 1`); a websocket gives3// that ordering for free, but the serverless hop (client → API server over HTTP → daemon)4// does not — two parallel fetches can reorder. The queue restores the invariant the cheap5// way: envelopes are sent as in-order batches, and the next batch does not leave until the6// prior one is confirmed (the API server's response means the daemon durably processed it).7//8// Delivery is at-least-once: a failed flush retries the SAME batch with backoff, which is9// safe because the daemon absorbs `mid ≤ lmid` as an idempotent replay. A per-envelope10// REJECTION (the API server's policy saying no) is not retried — the daemon has already11// advanced lmid past it, the authoritative snap-back rides the lmid release, and the reason12// surfaces here through `onRejected` (the wire itself carries no rejection signal).1314import type { MutationEnvelope } from "@rindle/client";1516import type { MutationEnvelopeSender } from "./subscribe.ts";1718/** One envelope's outcome from the API server's mutate endpoint. */19export interface PushOutcome {20  accepted: boolean;21  reason?: string;22}2324export interface QueuedMutationSenderOptions {25  /** Deliver one in-order batch; resolve once the daemon durably processed it (i.e. the26   *  API server awaited its daemon call before responding). Throw to have the SAME batch27   *  retried. Return per-envelope outcomes, or void when everything was accepted. */28  send: (envelopes: MutationEnvelope[]) => Promise<PushOutcome[] | void>;29  /** A policy rejection for one envelope (never retried; the prediction snaps back via the30   *  lmid release — this callback is where the reason reaches the app). */31  onRejected?: (envelope: MutationEnvelope, reason: string) => void;32  /** A failed flush attempt, before its retry. The ONLY surface for a transport/authority33   *  failure: `send` throwing is otherwise absorbed by the retry loop, so a queue built without34   *  this hook fails perfectly silently while head-of-line blocking every later mutation. */35  onError?: (err: unknown, attempt: number) => void;36  /** Backoff before retry `attempt` (1-based). Default: 200ms · 2^(attempt-1), capped at 5s. */37  retryDelayMs?: (attempt: number) => number;38  /** Max envelopes per flush. Default 32. */39  maxBatch?: number;40}4142interface Pending {43  envelope: MutationEnvelope;44  /** Resolves when the envelope's batch is confirmed (accepted OR rejected — both are45   *  durably processed states). */46  resolve: () => void;47}4849const defaultDelay = (attempt: number): number => Math.min(200 * 2 ** (attempt - 1), 5_000);5051const sleep = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms));5253/** Run an app-supplied observer without letting it break the queue. A throw escaping `flush`54 *  would abandon the loop with envelopes still queued and nothing scheduled to drain them —55 *  the notification hooks must not be able to wedge the write path they report on. */56const notify = (what: string, fn: () => void): void => {57  try {58    fn();59  } catch (err) {60    console.error(`[rindle] mutation queue ${what} handler threw:`, err);61  }62};6364/** A `MutationEnvelopeSender` (the `pushMutation` option of `RemoteOptimisticSource`) that65 *  queues envelopes and delivers them as confirmed, in-order batches. */66export function createQueuedMutationSender(opts: QueuedMutationSenderOptions): MutationEnvelopeSender {67  const queue: Pending[] = [];68  const maxBatch = Math.max(1, opts.maxBatch ?? 32);69  const delay = opts.retryDelayMs ?? defaultDelay;70  let flushing = false;7172  const flush = async (): Promise<void> => {73    if (flushing) return;74    flushing = true;75    try {76      while (queue.length > 0) {77        const batch = queue.slice(0, maxBatch);78        let outcomes: PushOutcome[] | void;79        for (let attempt = 1; ; attempt++) {80          try {81            outcomes = await opts.send(batch.map((p) => p.envelope));82            break;83          } catch (err) {84            notify("onError", () => opts.onError?.(err, attempt));85            await sleep(delay(attempt));86          }87        }88        queue.splice(0, batch.length);89        batch.forEach((pending, i) => {90          const outcome = Array.isArray(outcomes) ? outcomes[i] : undefined;91          if (outcome && !outcome.accepted) {92            const reason = outcome.reason ?? "mutation rejected";93            notify("onRejected", () => opts.onRejected?.(pending.envelope, reason));94          }95          pending.resolve();96        });97      }98    } finally {99      flushing = false;100    }101    // Envelopes enqueued during the final await: a new flush owns them.102    if (queue.length > 0) void flush();103  };104105  return (envelope) =>106    new Promise<void>((resolve) => {107      queue.push({ envelope, resolve });108      void flush();109    });110}111