Durable Objects Concurrency - Single Thread, Input & Output Gates, blockConcurrencyWhile & the Races That Remain

// Inventory: demonstrates the one race input gates do NOT stop - awaiting non-storage I/O.
import { DurableObject } from "cloudflare:workers";
import type { Env } from "./index";

// Stand-in for a payment / fraud API call. A timer, like fetch(), is not storage I/O,
// so the input gate opens and other requests may run while we wait.
const externalCall = (ms: number) => new Promise<void>((r) => setTimeout(r, ms));

export class Inventory extends DurableObject<Env> {
  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
    ctx.blockConcurrencyWhile(async () => {
      ctx.storage.sql.exec("CREATE TABLE IF NOT EXISTS stock(id INTEGER PRIMARY KEY CHECK(id = 1), qty INTEGER NOT NULL, version INTEGER NOT NULL)");
    });
  }

  reset(qty: number): void {
    this.ctx.storage.sql.exec("INSERT INTO stock VALUES (1, ?, 0) ON CONFLICT(id) DO UPDATE SET qty = excluded.qty, version = 0", qty);
  }

  qty(): number { return this.ctx.storage.sql.exec("SELECT qty FROM stock").one().qty as number; }

  // BUG: check, await external I/O, then act on the stale check.
  async reserveRacy(): Promise<string> {
    const qty = this.qty();
    if (qty <= 0) return "sold-out";
    await externalCall(50);                       // other reserveRacy calls interleave here
    this.ctx.storage.sql.exec("UPDATE stock SET qty = ? WHERE id = 1", qty - 1); // writes a stale value
    return "reserved";
  }

  // FIX 1: claim first (synchronous, atomic), then do the slow call, and compensate on failure.
  async reserveClaimFirst(): Promise<string> {
    const row = this.ctx.storage.sql
      .exec("UPDATE stock SET qty = qty - 1, version = version + 1 WHERE id = 1 AND qty > 0 RETURNING qty").toArray();
    if (row.length === 0) return "sold-out";
    try {
      await externalCall(50);
      return "reserved";
    } catch (e) {
      this.ctx.storage.sql.exec("UPDATE stock SET qty = qty + 1 WHERE id = 1"); // compensation
      throw e;
    }
  }

  // FIX 2: optimistic check-and-set - remember the version, re-check it after the await.
  async reserveCas(): Promise<string> {
    for (let attempt = 0; attempt < 5; attempt++) {
      const { qty, version } = this.ctx.storage.sql.exec("SELECT qty, version FROM stock").one() as { qty: number; version: number };
      if (qty <= 0) return "sold-out";
      await externalCall(50);
      const updated = this.ctx.storage.sql
        .exec("UPDATE stock SET qty = qty - 1, version = version + 1 WHERE id = 1 AND version = ? RETURNING qty", version).toArray();
      if (updated.length === 1) return "reserved";
      // someone else won the race; loop and re-read
    }
    return "conflict-retry-later";
  }
}

racy  results={"reserved":6} final qty=2
claim results={"reserved":3,"sold-out":3} final qty=0
cas   results={"reserved":3,"sold-out":3} final qty=0

# SIMULATION (not Cloudflare code) of a Durable Object's single thread, input gate and the external-I/O race.
# asyncio gives us one thread with interleaving at every await, exactly like JavaScript in the object.
import asyncio

class Inventory:
    def __init__(self, qty): self.qty, self.version = qty, 0
    async def storage_read(self):
        # Storage I/O: the input gate stays CLOSED, so no other request is delivered meanwhile.
        return self.qty, self.version
    async def external_call(self):
        await asyncio.sleep(0.01)      # fetch()/timer: the input gate OPENS, other requests run here

    async def racy(self):
        qty, _ = await self.storage_read()
        if qty <= 0: return "sold-out"
        await self.external_call()     # 6 buyers all passed the check above
        self.qty = qty - 1             # stale write: everyone writes 2
        return "reserved"

    async def claim_first(self):
        if self.qty <= 0: return "sold-out"   # check and decrement with no await between = atomic
        self.qty -= 1; self.version += 1
        await self.external_call()
        return "reserved"

    async def cas(self):
        for _ in range(5):
            qty, ver = await self.storage_read()
            if qty <= 0: return "sold-out"
            await self.external_call()
            if self.version == ver:            # nobody changed it while we waited
                self.qty -= 1; self.version += 1; return "reserved"
        return "conflict-retry-later"

async def run(mode):
    inv = Inventory(3)
    res = await asyncio.gather(*(getattr(inv, mode)() for _ in range(6)))
    tally = {k: res.count(k) for k in sorted(set(res))}
    print(f"{mode:<11} results={tally} final qty={inv.qty}")

for m in ["racy", "claim_first", "cas"]:
    asyncio.run(run(m))

racy        results={'reserved': 6} final qty=2
claim_first results={'reserved': 3, 'sold-out': 3} final qty=0
cas         results={'reserved': 3, 'sold-out': 3} final qty=0

// SIMULATION (not Cloudflare code): where interleaving happens, and what blockConcurrencyWhile costs.
const log: string[] = [];
const externalCall = (ms: number) => new Promise<void>((r) => setTimeout(r, ms)); // like fetch(): gate opens

async function handler(name: string) {
  log.push(`${name}: start (sync code runs to completion, nobody can interrupt)`);
  await externalCall(10);                          // other handlers may run during this await
  log.push(`${name}: resumed after external call`);
}

// Tiny stand-in for blockConcurrencyWhile: one event at a time, even across awaits.
let chain: Promise<unknown> = Promise.resolve();
function blockConcurrencyWhile<T>(fn: () => Promise<T>): Promise<T> {
  const run = chain.then(fn); chain = run.catch(() => undefined); return run;
}

async function main() {
  await Promise.all([handler("req1"), handler("req2")]);
  console.log("plain awaits:\n  " + log.join("\n  "));

  log.length = 0;
  await Promise.all(["req1", "req2"].map((n) => blockConcurrencyWhile(() => handler(n))));
  console.log("inside blockConcurrencyWhile:\n  " + log.join("\n  "));

  // Throughput ceiling when every request holds the object for its critical section.
  for (const ms of [1, 5, 50, 300]) {
    console.log(`critical section ${String(ms).padStart(3)} ms -> at most ${Math.floor(1000 / ms)} req/s for this object`);
  }
}
main();
export {};

plain awaits:
  req1: start (sync code runs to completion, nobody can interrupt)
  req2: start (sync code runs to completion, nobody can interrupt)
  req1: resumed after external call
  req2: resumed after external call
inside blockConcurrencyWhile:
  req1: start (sync code runs to completion, nobody can interrupt)
  req1: resumed after external call
  req2: start (sync code runs to completion, nobody can interrupt)
  req2: resumed after external call
critical section   1 ms -> at most 1000 req/s for this object
critical section   5 ms -> at most 200 req/s for this object
critical section  50 ms -> at most 20 req/s for this object
critical section 300 ms -> at most 3 req/s for this object