エッジトリガーのepollは部分読みで止まる。イベント配信も、同じ理由で壊れる

# Edge-triggered vs level-triggered epoll: what happens after a partial read?
# Linux only. Run: python3 epoll_demo.py
import os, select

def run(flags, label):
    r, w = os.pipe()
    os.set_blocking(r, False)
    ep = select.epoll()
    ep.register(r, select.EPOLLIN | flags)
    os.write(w, b"x" * 2048)                      # (2) the writer writes 2 kB
    first = ep.poll(timeout=1)                    # (3) the fd is reported ready
    os.read(r, 1024)                              # (4) the reader consumes only 1 kB
    second = ep.poll(timeout=0.5)                 # (5) wait again
    print(f"{label:15} first={bool(first)} second={bool(second)} (1024 bytes still unread)")
    if not second:
        # The documented fix for edge-triggered use: read until EAGAIN.
        try:
            while os.read(r, 1024):
                pass
        except BlockingIOError:
            print(f"{'':15} drained until EAGAIN")
    ep.close(); os.close(r); os.close(w)

run(0, "level-triggered")
run(select.EPOLLET, "edge-triggered")

// Which consumer design converges when the channel drops, duplicates and reorders messages?
// Deterministic: a seeded PRNG and a virtual clock (1 tick = 10 ms). Run: node sim.mjs
import { createInvalidator } from "./invalidator.mjs";

const KEYS = ["a", "b", "c", "d", "e"];
const rng = (a) => () => { // mulberry32: small, seedable, good enough for a simulation
  a = (a + 0x6d2b79f5) | 0;
  let t = Math.imul(a ^ (a >>> 15), 1 | a);
  t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t;
  return ((t ^ (t >>> 14)) >>> 0) / 2 ** 32;
};

async function trial(seed, { pDrop, pDup, maxDelay }) {
  const rand = rng(seed), pick = (n) => Math.floor(rand() * n);
  let tick = 0, q = [];
  const at = (d, fn) => { const h = { t: tick + d, fn, dead: false }; q.push(h); return h; };
  const timers = { setTimeout: (fn, ms) => at(Math.max(1, Math.round(ms / 10)), fn), clearTimeout: (h) => (h.dead = true) };

  const server = Object.fromEntries(KEYS.map((k) => [k, 0]));       // key -> version (the value is the version)
  const mk = () => ({ val: Object.fromEntries(KEYS.map((k) => [k, 0])), fetches: 0 });
  const A = mk(), B = mk(), C = mk(), D = mk(), E = mk();            // five consumers, same updates, own channel each

  const refetch = (c) => (k) => new Promise((done) => {
    c.fetches++;
    const answer = server[k];                                       // the server answers with the value NOW
    at(1 + pick(maxDelay), () => { if (answer > c.val[k]) c.val[k] = answer; done(); }); // late answers may arrive out of order
  });
  const invC = createInvalidator({ refetch: refetch(C), timers });
  const invD = createInvalidator({ refetch: refetch(D), timers });
  const fetchE = refetch(E);

  const handlers = {
    A: (m) => m.type === "event" && (A.val[m.k] = m.v),                                    // apply payload in arrival order
    B: (m) => m.type === "event" && m.v > B.val[m.k] && (B.val[m.k] = m.v),                // apply payload if newer
    C: (m) => m.type === "event" && invC.invalidate(m.k),                                  // hint only
    D: (m) => (m.type === "event" ? invD.invalidate(m.k) : KEYS.forEach((k) => m.vers[k] > D.val[k] && invD.invalidate(k))),
    E: (m) => (m.type === "event" ? m.v > E.val[m.k] && (E.val[m.k] = m.v) : KEYS.forEach((k) => m.vers[k] > E.val[k] && fetchE(k))),
  };
  const send = (m) => { for (const h of Object.values(handlers)) {       // an independent lossy channel per consumer
    if (rand() < pDrop) continue;
    for (let i = 0; i < (rand() < pDup ? 2 : 1); i++) at(pick(maxDelay), () => h(m));
  } };

  const END = 200, STOP = 600;
  for (tick = 0; tick <= STOP; tick++) {
    if (tick < END && rand() < 0.5) { const k = KEYS[pick(KEYS.length)]; send({ type: "event", k, v: ++server[k] }); }
    if (tick % 20 === 0 && tick > END) send({ type: "heartbeat", vers: { ...server } }); // the "level": cheap, repeated, idempotent
    for (let moved = true; moved;) {
      moved = false;
      for (const h of q.filter((x) => x.t <= tick && !x.dead)) { h.dead = true; h.fn(); moved = true; }
      q = q.filter((x) => !x.dead);
      await new Promise(setImmediate);                               // let promise continuations run
    }
  }
  const ok = (c) => KEYS.every((k) => c.val[k] === server[k]);
  return { A, B, C, D, E, ok };
}

for (const cfg of [{ pDrop: 0, pDup: 0.2, maxDelay: 8 }, { pDrop: 0.2, pDup: 0.2, maxDelay: 8 }]) {
  const N = 300, tally = Object.fromEntries("ABCDE".split("").map((n) => [n, { ok: 0, fetches: 0 }]));
  for (let s = 1; s <= N; s++) {
    const r = await trial(s, cfg);
    for (const n of "ABCDE") { tally[n].ok += r.ok(r[n]) ? 1 : 0; tally[n].fetches += r[n].fetches; }
  }
  console.log(`\ndrop=${cfg.pDrop} dup=${cfg.pDup} reorder<=${cfg.maxDelay} ticks, ${N} trials`);
  const names = { A: "A delta, apply in arrival order", B: "B delta, apply if newer", C: "C hint only (invalidate+refetch)", D: "D hint + periodic level check", E: "E delta + periodic level check" };
  for (const n of "ABCDE") console.log(`${names[n].padEnd(36)} converged ${String(tally[n].ok).padStart(3)}/${N}  fetches/trial ${(tally[n].fetches / N).toFixed(1)}`);
}

// A level-triggered consumer: events only say "this key may be stale".
// The consumer owns the question "what is the current value?" and always answers it by refetching.
export function createInvalidator({
  refetch,                       // async (key) => void; must read the CURRENT value and store it
  windowMs = 50,                 // coalescing window: fixed, so latency is bounded
  maxBackoffMs = 5000,
  timers = { setTimeout, clearTimeout },
}) {
  const dirty = new Set();       // keys that may be stale and are not being fetched right now
  const fetching = new Set();    // keys with a refetch in flight
  const failures = new Map();    // key -> consecutive failures
  let timer = null;

  function schedule(ms = windowMs) {
    if (timer === null) timer = timers.setTimeout(flush, ms);
  }

  function flush() {
    timer = null;
    for (const key of [...dirty]) {
      if (fetching.has(key)) continue;           // stays dirty; re-armed when that fetch settles
      dirty.delete(key);
      fetching.add(key);
      Promise.resolve(refetch(key)).then(
        () => { failures.delete(key); settle(key, windowMs); },
        () => {
          const n = (failures.get(key) ?? 0) + 1;
          failures.set(key, n);
          dirty.add(key);                        // still stale
          // full jitter backoff so that many clients do not retry in lockstep
          settle(key, Math.random() * Math.min(maxBackoffMs, windowMs * 2 ** n));
        },
      );
    }
  }

  function settle(key, nextMs) {
    fetching.delete(key);
    if (dirty.has(key)) schedule(nextMs);        // an event arrived during the fetch, or the fetch failed
  }

  return {
    invalidate(key) { dirty.add(key); schedule(); },                          // any order, any number of times
    invalidateAll(keys) { for (const k of keys) dirty.add(k); schedule(); },  // after (re)connect
    pending: () => dirty.size + fetching.size,
  };
}

import assert from "node:assert/strict";
import { test } from "node:test";
import { createInvalidator } from "./invalidator.mjs";

const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

test("100 events on 2 keys become 2 fetches", async () => {
  const fetched = [];
  const inv = createInvalidator({ refetch: async (k) => void fetched.push(k), windowMs: 20 });
  for (let i = 0; i < 100; i++) inv.invalidate(i % 2 ? "list" : "item:1");
  await sleep(80);
  assert.deepEqual(fetched.sort(), ["item:1", "list"]);
});

test("an event that arrives during a fetch causes a second fetch", async () => {
  let version = 1, shown = 0, calls = 0;
  const inv = createInvalidator({
    windowMs: 10,
    refetch: async () => {
      calls++;
      const v = version;            // the server answers with the value at request time...
      await sleep(40);              // ...and the response is slow
      shown = v;
    },
  });
  inv.invalidate("k");              // version 1 is announced
  await sleep(20);                  // fetch #1 is now in flight, holding version 1
  version = 2; inv.invalidate("k"); // version 2 is announced while fetch #1 is running
  await sleep(200);
  assert.equal(calls, 2);
  assert.equal(shown, 2);           // without the re-arm, the UI would stay at 1
});

test("a failed fetch is retried with backoff instead of being forgotten", async () => {
  let calls = 0;
  const inv = createInvalidator({
    windowMs: 5,
    maxBackoffMs: 20,
    refetch: async () => { if (++calls < 3) throw new Error("boom"); },
  });
  inv.invalidate("k");
  await sleep(300);
  assert.equal(calls, 3);            // two failures, then success, then it stops
  assert.equal(inv.pending(), 0);
});