kv: harden in-process backend — atomic getdel, active expiry, multi-replica doc (#5)

Three RED-blessed non-blocking follow-ups on the KV-optional work.

1. Drop the non-atomic getdel() shim in lib/redis.ts. It shadowed BOTH backends'
   native atomic GETDEL with a get->del pipeline (other commands interleave
   between GET and DEL), reintroducing the one-time login-code double-use race
   that fetchAndDeleteLoginCodeData relies on GETDEL to prevent. ioredis maps
   getdel to the single-round-trip Redis GETDEL command; MemoryKV.getdel now
   reads+deletes in one un-yielded step (was `await this.get()` then delete — the
   await yielded the event loop, letting two racers both observe the value).

2. Bound MemoryKV growth. Lazy on-read eviction never fires for write-once-
   never-read keys (rl:<ip> rate-limit counters, one-shot tokens), so the map
   grew unbounded until the pod OOMKilled. Add sweepExpired() (one O(n) active-
   expire pass) run on an unref'd 60s timer, started by lib/redis.ts for the
   server singleton only (edge-gated; no test imports that module, so unit runs
   stay timer-free). MemoryKV stays pure (zero imports).

3. Document in .env.example that re-enabling multi-replica REQUIRES KV_URL
   (single-replica-by-construction otherwise: SQLite-on-RWO + in-process KV).
   The one-time non-silent in-process warning already exists in lib/kv/select.ts;
   no metrics surface exists, so log + doc is sufficient.

Tests: +3 cases (getdel atomicity under concurrent race; 100k-key sweep reclaims
to 0; sweep retains live/no-TTL keys). 28 pass, 3 skip (select.ts skips without
ioredis in a bare checkout), 0 fail.

Co-authored-by: Hanzo AI <ai@hanzo.ai>

.env.example

@@ -79,6 +79,12 @@ NEXT_PRIVATE_DOCUMENT_PASSWORD_KEY=my-superstrong-document-secret
 # an external Hanzo KV instance for multi-replica HA (shared state across pods).
 # Accepts the Hanzo KV brand scheme kv:// (kvs:// for TLS) or a redis:// DSN.
 # A malformed value fails CLOSED (the app throws rather than silently degrade).
+#
+# Re-enabling multi-replica (replicas > 1) REQUIRES KV_URL. dataroom is
+# single-replica-by-construction otherwise: SQLite on a ReadWriteOnce volume plus
+# the in-process KV. Without KV_URL each pod owns its OWN map, so sessions,
+# rate-limit windows and tus upload locks split-brain across replicas — scaling
+# out needs a shared DB AND KV_URL set.
 #   KV_URL=kv://:password@hanzo-kv:6379
 KV_URL=
 

lib/kv/memory-kv.test.ts

@@ -116,6 +116,23 @@ test("getdel returns value then removes it", async () => {
   assert.equal(await kv.get("state"), null);
 });
 
+test("SECURITY — getdel is atomic: two racing getdel, exactly one sees the value", async () => {
+  // The one-time login-code guard (lib/emails/send-verification-request.ts,
+  // fetchAndDeleteLoginCodeData) relies on GETDEL atomicity to stop a login code
+  // being used twice. A get-then-delete with an await BETWEEN the read and the
+  // delete lets both racers observe the value (TOCTOU); getdel must read+delete
+  // in one un-yielded step so only one caller ever wins.
+  const kv = new MemoryKV();
+  await kv.set("login_code:once", "grant");
+  const [a, b] = await Promise.all([
+    kv.getdel("login_code:once"),
+    kv.getdel("login_code:once"),
+  ]);
+  const winners = [a, b].filter((v) => v === "grant");
+  assert.equal(winners.length, 1, "exactly one racer may observe the value");
+  assert.equal(await kv.get("login_code:once"), null); // key is gone either way
+});
+
 test("call('SET', ...) dispatches to set (the set-with-options shim path)", async () => {
   const kv = new MemoryKV();
   assert.equal(await kv.call("SET", "k", "v", "EX", 60, "NX"), "OK");
@@ -152,7 +169,7 @@ test("pipeline: notification-queue pattern (rpush/expire/sadd)", async () => {
   assert.deepEqual(await kv.smembers("viewers:daily"), ["v:d:t"]);
 });
 
-test("pipeline: get+del (the getdel shim relies on results[0][1])", async () => {
+test("pipeline: get+del composes in order (results[0][1] = GET result before DEL)", async () => {
   const kv = new MemoryKV();
   await kv.set("g", "val");
   const res = await kv.pipeline().get("g").del("g").exec();
@@ -168,3 +185,31 @@ test("SECURITY CONTRACT — two MemoryKV instances DO NOT share state", async ()
   await a.set("session:1", "alice");
   assert.equal(await b.get("session:1"), null);
 });
+
+test("active expiry: sweepExpired reclaims write-once-never-read keys (bounds memory)", async () => {
+  // rl:<ip> rate-limit keys and one-shot session/token keys are written with a
+  // TTL and never read again, so the lazy on-read eviction never fires for them.
+  // Without an active sweep the map grows unbounded until the pod OOMKills.
+  // sweepExpired() reclaims every elapsed-TTL entry in one pass.
+  const kv = new MemoryKV();
+  const N = 100_000;
+  const past = Date.now() - 1; // already expired at write time (deterministic)
+  for (let i = 0; i < N; i++) {
+    await kv.set(`rl:${i}`, "1", "PXAT", past);
+  }
+  assert.equal(kv.size, N); // resident despite expiry — never read, so never lazily evicted
+  const evicted = kv.sweepExpired();
+  assert.equal(evicted, N);
+  assert.equal(kv.size, 0); // resident memory reclaimed
+});
+
+test("active expiry: sweepExpired keeps live keys, drops only expired ones", async () => {
+  const kv = new MemoryKV();
+  await kv.set("live", "1", "PX", 60_000); // far-future TTL
+  await kv.set("nottl", "1"); // no TTL at all
+  await kv.set("dead", "1", "PXAT", Date.now() - 1); // expired
+  assert.equal(kv.sweepExpired(), 1); // only "dead"
+  assert.equal(kv.size, 2);
+  assert.equal(await kv.get("live"), "1");
+  assert.equal(await kv.get("nottl"), "1");
+});

lib/kv/memory-kv.ts

@@ -39,6 +39,7 @@ type PipelineOp = () => Promise<[Error | null, unknown]>;
 
 export class MemoryKV {
   private store = new Map<string, Entry>();
+  private sweepTimer?: ReturnType<typeof setInterval>;
 
   // ── expiry + typed accessors ──────────────────────────────────────────────
 
@@ -53,6 +54,32 @@ export class MemoryKV {
     return e;
   }
 
+  /** Resident key count (live + not-yet-swept). Redis DBSIZE analogue — used to
+   *  assert active-expiry reclamation. */
+  get size(): number {
+    return this.store.size;
+  }
+
+  /**
+   * sweepExpired — active-expiry pass. Deletes every entry whose TTL has elapsed
+   * and returns the count reclaimed. The lazy on-read eviction in live() never
+   * fires for write-once-never-read keys (rl:<ip> rate-limit counters, one-shot
+   * session/token keys), so without this sweep the map grows unbounded until the
+   * pod OOMKills. One O(n) pass over the in-process map, run off the hot path on
+   * an unref'd timer (see startActiveExpiry). Pure + synchronous → unit-testable
+   * with no timer.
+   */
+  sweepExpired(now: number = Date.now()): number {
+    let evicted = 0;
+    for (const [k, e] of this.store) {
+      if (e.expireAt !== undefined && e.expireAt <= now) {
+        this.store.delete(k);
+        evicted++;
+      }
+    }
+    return evicted;
+  }
+
   private zsetOf(key: string): Zset {
     const e = this.live(key);
     if (e?.zset) return e.zset;
@@ -151,7 +178,12 @@ export class MemoryKV {
   }
 
   async getdel(key: string): Promise<string | null> {
-    const v = await this.get(key);
+    // Atomic read+delete: NO await between the read and the delete, so two racing
+    // getdel calls can never both observe the value. The one-time login-code guard
+    // in lib/emails/send-verification-request.ts (fetchAndDeleteLoginCodeData)
+    // depends on this to stop a code being used twice — see memory-kv.test.ts.
+    const e = this.live(key);
+    const v = e?.str ?? null;
     this.store.delete(key);
     return v;
   }
@@ -373,6 +405,31 @@ export class MemoryKV {
     return new MemoryPipeline(this);
   }
 
+  // ── active expiry (bounds resident memory) ─────────────────────────────────
+
+  /**
+   * startActiveExpiry — run sweepExpired() every intervalMs. Idempotent. The
+   * timer handle is .unref()'d so it NEVER holds the Node event loop open (the
+   * process may still exit with a sweep pending). The composition root
+   * (lib/redis.ts) starts this for the long-lived server singleton, gated OUT of
+   * the edge runtime; the pure unit tests never call it, so they stay timer-free.
+   */
+  startActiveExpiry(intervalMs: number = 60_000): void {
+    if (this.sweepTimer !== undefined) return;
+    const t = setInterval(() => this.sweepExpired(), intervalMs);
+    // unref where supported (Node Timeout); harmless no-op under a numeric handle.
+    (t as unknown as { unref?: () => void }).unref?.();
+    this.sweepTimer = t;
+  }
+
+  /** Stop the active-expiry timer (parity with quit/disconnect teardown). */
+  stopActiveExpiry(): void {
+    if (this.sweepTimer !== undefined) {
+      clearInterval(this.sweepTimer);
+      this.sweepTimer = undefined;
+    }
+  }
+
   // ── lifecycle no-ops (ioredis parity; MemoryKV needs no connection) ────────
 
   get status(): string {

lib/redis.ts

@@ -14,12 +14,25 @@
  *
  * Environment: KV_URL (e.g. kv://:password@hanzo-kv:6379 or a redis:// DSN).
  */
+import { MemoryKV } from "./kv/memory-kv";
 import { createKvClient } from "./kv/select";
 
 export const redis = createKvClient();
 
 export const lockerRedisClient = createKvClient();
 
+// In-process backend (KV_URL unset): actively expire keys so write-once-never-read
+// keys (rate-limit rl:<ip>, one-shot tokens) can't accrete to an OOM. Server runtime
+// only — the edge runtime has no long-lived process to sweep, and active expiry is
+// moot against the external ioredis client (Hanzo KV expires its own keys). The timer
+// is unref'd (MemoryKV.startActiveExpiry) so it never holds the process open; no test
+// imports this module, so unit runs stay timer-free.
+if (process.env.NEXT_RUNTIME !== "edge") {
+  for (const client of [redis, lockerRedisClient]) {
+    if (client instanceof MemoryKV) client.startActiveExpiry();
+  }
+}
+
 // Upstash-compatible set() shim — translates options-object calls to ioredis positional args
 const originalSet = redis.set.bind(redis);
 (redis as any).set = async function (
@@ -58,14 +71,11 @@ const originalZrangebyscore = redis.zrangebyscore.bind(redis);
   return originalZrange(key, start as number, stop as number);
 };
 
-// Upstash-compatible getdel() shim — atomic GET + DEL
-(redis as any).getdel = async function (key: string) {
-  const pipeline = redis.pipeline();
-  pipeline.get(key);
-  pipeline.del(key);
-  const results = await pipeline.exec();
-  return results?.[0]?.[1] ?? null;
-};
+// No getdel() shim: GETDEL is atomic natively on BOTH backends — ioredis maps it
+// to the single-round-trip Redis GETDEL command, and MemoryKV.getdel reads+deletes
+// in one un-yielded step. A get→del pipeline here would NOT be atomic (other
+// commands interleave between GET and DEL) and would reintroduce the one-time
+// login-code double-use race. Do not add one.
 
 // Upstash-compatible get() shim — try to auto-parse JSON
 const originalGet = redis.get.bind(redis);