Durable Objects Storage - SQLite vs Legacy KV, Transactions, Write Coalescing & Point-in-Time Recovery

// Counter: one object per counter name. Sharded counters spread one hot name across N objects.
import { DurableObject } from "cloudflare:workers";
import type { Env } from "./index";

export class Counter extends DurableObject<Env> {
  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
    ctx.blockConcurrencyWhile(async () => {
      ctx.storage.sql.exec("CREATE TABLE IF NOT EXISTS kv(k TEXT PRIMARY KEY, v INTEGER NOT NULL)");
    });
  }

  // RPC method: callers do `await stub.increment(5)`; no Request/Response parsing.
  increment(by = 1): number {
    // UPSERT + RETURNING is one synchronous statement: atomic, no await, no interleaving.
    return this.ctx.storage.sql
      .exec("INSERT INTO kv(k, v) VALUES ('n', ?) ON CONFLICT(k) DO UPDATE SET v = v + excluded.v RETURNING v", by)
      .one().v as number;
  }

  get(): number {
    const rows = this.ctx.storage.sql.exec("SELECT v FROM kv WHERE k = 'n'").toArray();
    return rows.length ? (rows[0].v as number) : 0;
  }
}

// Worker-side helpers for a sharded counter: writes pick a random shard, reads fan in all shards.
export async function shardedIncrement(env: Env, name: string, shards: number): Promise<number> {
  const shard = Math.floor(Math.random() * shards);
  return env.COUNTER.getByName(`${name}#${shard}`).increment(1);
}

export async function shardedRead(env: Env, name: string, shards: number): Promise<{ total: number; perShard: number[] }> {
  const perShard = await Promise.all(
    Array.from({ length: shards }, (_, i) => env.COUNTER.getByName(`${name}#${i}`).get()),
  );
  return { total: perShard.reduce((a, b) => a + b, 0), perShard };
}

// RateLimiter: one object per (tenant or API key). Token bucket persisted in the object's SQLite.
import { DurableObject } from "cloudflare:workers";
import type { Env } from "./index";

export type Decision = { allowed: boolean; remaining: number; retryAfterMs: number };

export class RateLimiter extends DurableObject<Env> {
  private tokens = -1;  // in-memory copy; -1 means "not loaded yet" (fresh instance or after eviction)
  private updated = 0;

  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
    ctx.blockConcurrencyWhile(async () => {
      ctx.storage.sql.exec("CREATE TABLE IF NOT EXISTS bucket(id INTEGER PRIMARY KEY CHECK(id = 1), tokens REAL, updated INTEGER)");
      const row = ctx.storage.sql.exec("SELECT tokens, updated FROM bucket").toArray()[0];
      if (row) { this.tokens = row.tokens as number; this.updated = row.updated as number; }
    });
  }

  // capacity = burst size, refillPerSec = sustained rate. Whole method is synchronous => atomic per key.
  take(capacity: number, refillPerSec: number, cost = 1): Decision {
    const now = Date.now();
    if (this.tokens < 0) { this.tokens = capacity; this.updated = now; }
    // Refill based on elapsed time, capped at capacity.
    this.tokens = Math.min(capacity, this.tokens + ((now - this.updated) / 1000) * refillPerSec);
    this.updated = now;
    let allowed = false;
    if (this.tokens >= cost) { this.tokens -= cost; allowed = true; }
    // Persist so an eviction cannot hand a client a fresh full bucket. Output gate holds the reply until durable.
    this.ctx.storage.sql.exec(
      "INSERT INTO bucket(id, tokens, updated) VALUES (1, ?, ?) ON CONFLICT(id) DO UPDATE SET tokens = excluded.tokens, updated = excluded.updated",
      this.tokens, this.updated);
    const deficit = Math.max(0, cost - this.tokens);
    return { allowed, remaining: Math.floor(this.tokens), retryAfterMs: allowed ? 0 : Math.ceil((deficit / refillPerSec) * 1000) };
  }
}

# SIMULATION (not Cloudflare code) of write coalescing + the output gate.
# Writes go to a buffer; consecutive writes with no await in between commit as ONE atomic batch.
# Outgoing messages (the reply) are held until the batch is durable. A failed flush discards the reply.
class Obj:
    def __init__(self):
        self.disk, self.buffer, self.batches, self.outbox = {}, {}, [], []
    def put(self, k, v):                # like ctx.storage.sql.exec / storage.put without await
        self.buffer[k] = v
    def await_point(self, crash=False): # any await: the pending batch is flushed (or lost on crash)
        if self.buffer:
            if crash:
                self.buffer.clear(); self.outbox.clear()   # object resets, held replies are replaced by errors
                return "CRASH: batch lost, held reply discarded, caller gets an error"
            self.disk.update(self.buffer); self.batches.append(dict(self.buffer)); self.buffer.clear()
        return "flushed"
    def reply(self, msg):               # held by the output gate until the next flush succeeds
        self.outbox.append(msg)
    def deliver(self):
        if self.buffer:                 # a write is still pending: output gate is closed
            return []
        sent, self.outbox = self.outbox, []
        return sent

print("A) transfer, 3 writes, no await between them")
o = Obj(); o.disk = {"alice": 100, "bob": 0}
o.put("alice", 70); o.put("bob", 30); o.put("log:1", "alice->bob 30"); o.reply("200 transfer ok")
print("   before flush, client sees:", o.deliver() or "nothing yet (output gate closed)")
print("  ", o.await_point(), "| batches:", len(o.batches), "| client sees:", o.deliver(), "| disk:", o.disk)

print("B) same transfer, machine dies before the batch is durable")
o = Obj(); o.disk = {"alice": 100, "bob": 0}
o.put("alice", 70); o.put("bob", 30); o.reply("200 transfer ok")
print("  ", o.await_point(crash=True), "| disk:", o.disk, "| client sees:", o.deliver())

print("C) legacy style: await between debit and credit (two batches)")
o = Obj(); o.disk = {"alice": 100, "bob": 0}
o.put("alice", 70); print("   debit:", o.await_point())
o.put("bob", 30);   print("   credit:", o.await_point(crash=True))
print("   disk after crash:", o.disk, "<- debit committed, credit lost: money vanished")

A) transfer, 3 writes, no await between them
   before flush, client sees: nothing yet (output gate closed)
   flushed | batches: 1 | client sees: ['200 transfer ok'] | disk: {'alice': 70, 'bob': 30, 'log:1': 'alice->bob 30'}
B) same transfer, machine dies before the batch is durable
   CRASH: batch lost, held reply discarded, caller gets an error | disk: {'alice': 100, 'bob': 0} | client sees: []
C) legacy style: await between debit and credit (two batches)
   debit: flushed
   credit: CRASH: batch lost, held reply discarded, caller gets an error
   disk after crash: {'alice': 70, 'bob': 0} <- debit committed, credit lost: money vanished

// Point-in-time recovery for ONE SQLite-backed object (type-checked; PITR is a production feature).
// Call from an admin-only RPC method on the object, e.g. `await restoreTo(this.ctx, Date.now() - 15 * 60e3)`.
export async function restoreTo(ctx: DurableObjectState, when: number): Promise<{ target: string; undo: string }> {
  const target = await ctx.storage.getBookmarkForTime(when);        // must be within the last 30 days
  const undo = await ctx.storage.onNextSessionRestoreBookmark(target); // bookmark to roll the restore back
  // Store `undo` somewhere OUTSIDE this object (log, KV, another DO) before restarting.
  ctx.abort("restoring to bookmark " + target);                     // restart; the new session sees restored data
  return { target, undo };                                          // not reached: abort() resets the object
}

// SIMULATION (not Cloudflare code) of point-in-time recovery with lexically comparable bookmarks.
// The real API: getCurrentBookmark(), getBookmarkForTime(ts), onNextSessionRestoreBookmark(b) + ctx.abort().
type Entry = { bookmark: string; at: number; op: string; apply: (db: Map<string, number>) => void };

const log: Entry[] = [];
let seq = 0;
const bm = (n: number) => n.toString(16).padStart(8, "0") + "-" + "c0ffee"; // sortable like the real ones
function write(at: number, op: string, apply: Entry["apply"]) { log.push({ bookmark: bm(++seq), at, op, apply }); }
function replay(upTo: string): Map<string, number> {
  const db = new Map<string, number>();
  for (const e of log) if (e.bookmark <= upTo) e.apply(db);   // plain string comparison orders bookmarks
  return db;
}
const bookmarkForTime = (t: number) => [...log].reverse().find((e) => e.at <= t)!.bookmark;

write(1000, "INSERT order 1", (db) => db.set("order:1", 40));
write(2000, "INSERT order 2", (db) => db.set("order:2", 25));
write(3000, "UPDATE order 1", (db) => db.set("order:1", 45));
write(4000, "DELETE FROM orders  -- forgot WHERE", (db) => db.clear());
write(4500, "INSERT order 3 (after the bug)", (db) => db.set("order:3", 10));

const head = log[log.length - 1].bookmark;
console.log("current state  @", head, Object.fromEntries(replay(head)));
const target = bookmarkForTime(3999);              // just before the bad DELETE
const undo = head;                                 // real API returns an 'undo' bookmark for the restore
console.log("restore target @", target, Object.fromEntries(replay(target)));
console.log("lost by restoring:", JSON.stringify(log.filter((e) => e.bookmark > target).map((e) => e.op)));
console.log("undo bookmark  @", undo, "(restore again to this to roll back the recovery)");
console.log("bookmarks sortable:", bm(9) < bm(10), "| PITR window in the docs: 30 days");

current state  @ 00000005-c0ffee { 'order:3': 10 }
restore target @ 00000003-c0ffee { 'order:1': 45, 'order:2': 25 }
lost by restoring: ["DELETE FROM orders  -- forgot WHERE","INSERT order 3 (after the bug)"]
undo bookmark  @ 00000005-c0ffee (restore again to this to roll back the recovery)
bookmarks sortable: true | PITR window in the docs: 30 days