festipod/backend/services/keyed-serializer.js

// One-at-a-time execution PER KEY, with a time limit — the ordering primitive behind every decision
// the backend takes about one pod object (grant or revoke its public read, index or de-index it in the
// catalog, add or remove an attendee).
//
// WHY. The platform runs the handlers of every webhook as soon as it arrives, without awaiting the
// previous one. Two notifications about the same object therefore run CONCURRENTLY: each reads, then
// acts — and one's action can land between the other's read and action.
//
// WHAT IT GUARANTEES — and what it does not:
//   - works for one key never overlap; each starts once the previous one for that key has settled
//     (finished, failed, or timed out), in the order they ENTERED `serialize` — which is the order the
//     handlers got there (after their own verification reads), NOT the order the user acted or the
//     pod emitted the activities;
//   - so it makes a decision order-proof only when the work derives that decision from state it reads
//     INSIDE: every pod change is notified after it happened, so whichever work runs last started
//     after the last change and acts on a state at least that recent (public-read `reconcile`, catalog
//     `sync`);
//   - a decision taken from a notification's own content (a Join versus a Leave) gains no ordering
//     from this; it must compare that content with recorded state under the lock
//     (`attendance-ledger.js`).
//
// BOUNDED. A work that has not settled after `timeoutMs` (default `DEFAULT_TIMEOUT_MS`) is abandoned:
// its caller's promise rejects with a `SerializedWorkTimeout`, its `signal` is aborted, and the key is
// released so the next work — the next notification — re-decides from a fresh read. A pod request that
// never answers can then block one object for at most that long, never until restart. The abandoned
// work cannot be stopped, so each work checks `signal.aborted` immediately before every write and
// gives up once it is set. A write ALREADY SENT when the time ran out can still land after the next
// work has read the state. So when an abandoned work finally settles, `onLateSettle(value)` (if given)
// is called with what the work resolved (undefined if it rejected): the CALLER decides from it whether
// a write actually landed late and, only then, queues one fresh decision for the key — a re-decision
// that must not itself schedule another (no chain while the pod stays slow). A work that never
// settles never wrote late.
//
// Keys are independent (different objects run concurrently); an idle key holds no memory.

// 30 s: a serialized decision makes a handful of pod round-trips — read the object, read up to two
// ACLs, write up to two (public-read); read, then one store write (catalog); a wait of at most 1.5 s
// for a fresh collection, the participant's participations collection made ready (a profile and an
// ACL read; at its creation, the create-and-attach, a profile re-read, one ACL write and one group
// grant), a read, two writes and their two read-backs (attendance) —
// each answering in tens of milliseconds on the local stack, a few seconds at worst on a loaded one. 30 s is over ten times the slowest
// legitimate run, and short enough that a stalled pod holds one object's decisions for half a minute
// rather than until the backend restarts.
const DEFAULT_TIMEOUT_MS = 30000;

class SerializedWorkTimeout extends Error {
  constructor(key, timeoutMs) {
    super(`serialized work for ${key} timed out after ${timeoutMs} ms: released, the next notification re-decides`);
    this.name = 'SerializedWorkTimeout';
    this.key = key;
  }
}

/**
 * createKeyedSerializer({ timeoutMs }) → serialize(key, work, { onLateSettle })
 *
 * `serialize(key, work)` runs `work(signal)` once every work previously queued for `key` has settled,
 * and resolves or rejects as `work` does — or rejects with a `SerializedWorkTimeout` (and aborts
 * `signal`) when `work` has not settled within `timeoutMs`; `onLateSettle(value)` is then called if
 * and when that abandoned work settles, with its resolved value (undefined when it rejected). `serialize.pending()` is the number of keys with queued work.
 */
function createKeyedSerializer({ timeoutMs = DEFAULT_TIMEOUT_MS } = {}) {
  const tails = new Map();

  function bounded(key, work, onLateSettle) {
    const controller = new AbortController();
    let timer;
    const expiry = new Promise((_, reject) => {
      timer = setTimeout(() => {
        controller.abort();
        reject(new SerializedWorkTimeout(key, timeoutMs));
      }, timeoutMs);
      if (typeof timer.unref === 'function') timer.unref();
    });
    const run = Promise.resolve().then(() => work(controller.signal));
    if (onLateSettle) {
      const late = value => {
        if (!controller.signal.aborted) return;
        try {
          onLateSettle(value);
        } catch {
          // a re-decision that cannot even be queued changes nothing; the next notification decides
        }
      };
      run.then(late, () => late(undefined));
    }
    // `race` handles `run` too, so a late rejection of an abandoned work is never unhandled.
    return Promise.race([run, expiry]).finally(() => clearTimeout(timer));
  }

  function serialize(key, work, { onLateSettle } = {}) {
    const previous = tails.get(key) || Promise.resolve();
    const result = previous.then(() => bounded(key, work, onLateSettle));
    // The chain continues whatever `work` does; the tail never rejects.
    const tail = result.then(
      () => undefined,
      () => undefined
    );
    tails.set(key, tail);
    tail.then(() => {
      if (tails.get(key) === tail) tails.delete(key);
    });
    return result;
  }

  serialize.pending = () => tails.size;
  return serialize;
}

module.exports = { createKeyedSerializer, SerializedWorkTimeout, DEFAULT_TIMEOUT_MS };