Optimistic Transactions

Use redis.watch() for check-and-set logic: read some keys, decide what to write, and commit atomically only if none of the watched keys changed underneath you. If a watched key changed, the commit aborts and the helper retries.

This is Redis optimistic locking (WATCH/MULTI/EXEC) wrapped as a retry loop. It runs on a session connection, because WATCH state belongs to one connection.

redis.watch(keys, body, options?)

await redis.watch("views:home", async (s) => {
  // read the watched keys through the session's typed stores, then return a
  // built, un-executed transaction to commit, or null to opt out
  return s.multi();
});

Per attempt the helper opens (or borrows) a session, sends WATCH keys, runs your body, and calls exec() on the transaction the body returns. Because the body hands back the built-but-not-executed transaction, you can neither forget to exec() nor double-exec().

• The commit succeeds → redis.watch resolves the decoded result tuple.
• A watched key changed → exec() aborts, onAbort fires, the helper backs off (if configured) and retries with a fresh WATCH.
• The body returns null → the helper UNWATCHes and resolves null; you opted out.
• Attempts run out → the helper throws WatchRetriesExceededError.

Read the watched keys through the session (s.kv(...), s.hash(...), …) so the reads happen on the same connection that holds the WATCH. Build the write with s.multi(), whose .add(command, decoder) extends a position-typed result tuple exactly like redis.multi(), and whose exec() resolves the tuple or null on abort.

Everything redis.multi() says about arguments applies here. Encode each value with the schema’s own codec, schema.encode(value) for a keyspace and schema.fields.<name>.encode(value) for a hash field, rather than String(...): the transaction then writes exactly what the typed store reads back, and a wrong type fails at the encode call instead of at some later read. The decoder you pair with each command is still unchecked against the command string, so keep the usual pairings in mind (SET to okReply, INCR and HSET to numberReply, GET to stringOrNullReply).

Check-And-Set Example

Cap a counter at a ceiling, retrying if a concurrent writer moves it:

import { okReply, numberReply, WatchRetriesExceededError } from "benni";
import { number, kv } from "benni/schema";

const views = kv("views", number());

const result = await redis.watch(
  views.key("home"),
  async (s) => {
    const current = (await s.kv(views).get("home")) ?? 0; // read on the watching connection
    if (current >= 1_000_000) return null;                      // opt out -> resolves null
    return s
      .multi()
      // views.encode is the codec the typed store uses, so the value the store
      // would accept is the value that lands in Redis
      .add(["SET", views.key("home"), views.encode(current + 1)], okReply)
      .add(["INCR", `${views.key("home")}:writes`], numberReply);
  },
  {
    attempts: 5,
    onAbort: ({ attempt }) => metrics.increment("views.cas_conflict", { attempt })
  }
);
//    ^? [void, number] | null   (null = the body opted out)

A committed result here logs as [undefined, 4]. The void slot is what okReply yields for the SET, and undefined is a success: an OK reply carries nothing worth typing. A command that actually failed throws rather than landing undefined in the tuple.

Balance Transfer

Move funds between two accounts atomically, aborting the whole operation if either balance shifts mid-flight, and opting out cleanly when the source lacks funds:

import { okReply } from "benni";
import { number, kv } from "benni/schema";

const balances = kv("balance", number());

async function transfer(from: string, to: string, amount: number) {
  return redis.watch(
    [balances.key(from), balances.key(to)], // watch both accounts
    async (s) => {
      const fromBalance = (await s.kv(balances).get(from)) ?? 0;
      if (fromBalance < amount) return null; // insufficient funds -> resolves null, no retry
      const toBalance = (await s.kv(balances).get(to)) ?? 0;
      return s
        .multi()
        .add(
          ["SET", balances.key(from), balances.encode(fromBalance - amount)],
          okReply
        )
        .add(
          ["SET", balances.key(to), balances.encode(toBalance + amount)],
          okReply
        );
    },
    { attempts: 10 }
  );
}

const outcome = await transfer("alice", "bob", 50);
if (outcome === null) {
  // either the body opted out (insufficient funds)…
}

The Write Side Is Not Typed The Way Stores Are

Say it plainly, because this is where balance-changing code lives: inside a watched transaction the reads go through the typed stores (s.kv(balances).get(from) hands back number | null), and the writes are hand-built command arrays. There is no s.kv(balances).set(...) that enrolls in the transaction. The schema still helps on both halves of every argument, and that is the whole of the help you get:

• balances.key(from) derives the key, so no key text is written by hand.
• balances.encode(value) encodes it with the same codec the typed store reads back, and users.fields.score.encode(value) does it per hash field.

What nothing checks:

• The command string against the schema’s kind. ["LPUSH", balances.key(from), …] on a string keyspace compiles, and fails at EXEC with WRONGTYPE.
• The decoder against the command. numberReply on a SET compiles and throws only when exec() decodes the reply. See Transactions.
• Arity and option order. ["SET", key, value, "EX"] with no seconds is a runtime error reply, not a type error.

So the compile-time guarantee you get elsewhere in Benni stops at .add(...). Three habits keep the rest honest, and they are worth making review rules for any transaction that moves money:

1. Never write a key string; always schema.key(id).
2. Never String(value); always schema.encode(value) or schema.fields.<name>.encode(value). A wrong type then fails at the encode call, above Redis, instead of landing a bad value that reads back as NaN.
3. Keep the body short enough to read in one screen, and pair each command with its decoder as you add it rather than afterwards.

When a check-and-set is hot enough or important enough that you want the whole thing checked and atomic in one place, a typed Lua script is the alternative: script() names its keys and types its args, so the call site is checked even though the body is Lua.

Options

redis.watch(keys, body, {
  attempts,   // total attempts, default 5, must be >= 1
  backoff,    // (attempt: number) => ms to wait before the next retry; default: no delay
  onAbort,    // ({ attempt }) => void, called on each conflict, before backoff
  session     // borrow a long-lived session for hot paths; the helper never closes it
});

onAbort fires on every conflict, so you can watch contention with metrics before it becomes an incident. backoff is opt-in: there is no hidden default sleep. session lets a hot path reuse one connection across many redis.watch calls; the helper closes only sessions it opened itself.

WATCH belongs to the connection, so two redis.watch calls sharing one borrowed session cannot run at the same time: the second waits for the first to reach its EXEC. Reuse a session to save connections on a hot path, and open separate sessions when you want the watches to overlap.

Abort, Opt-Out, And Exhaustion

Three distinct outcomes, three distinct signals:

• Abort (conflict). A watched key changed before exec(), so exec() resolves null internally. The helper retries with a fresh WATCH, and you never see this directly unless attempts runs out.
• Opt-out. Your body returns null. The helper UNWATCHes and redis.watch resolves null. Use it for “value already correct” or “insufficient funds”: a deliberate no-op, not a failure.
• Exhaustion. All attempts aborted. redis.watch throws WatchRetriesExceededError, which carries .attempts. Exhaustion is exceptional, so it throws rather than returning null, keeping the happy path ceremony-free.

try {
  const result = await redis.watch(/* … */);
  if (result === null) {
    // the body opted out
  }
} catch (error) {
  if (error instanceof WatchRetriesExceededError) {
    console.error(`gave up after ${error.attempts} conflicts`);
  }
}

A per-command runtime error inside a committed EXEC (for example a WRONGTYPE) rejects the promise. Note that Redis MULTI has no rollback: the other commands in that transaction still committed.

Manual Form

For a custom loop, drive the primitives on a session directly:

await using s = await redis.session();
await s.watch([views.key("home")]);
const current = (await s.kv(views).get("home")) ?? 0;
const outcome = await s
  .multi()
  .add(["SET", views.key("home"), views.encode(current + 1)], okReply)
  .exec();
//    ^? [void] | null   (the void slot reads as undefined; the SET succeeded)
if (outcome === null) {
  // a watched key changed, so re-WATCH and retry
}

An empty watched exec() throws a TypeError. Unlike redis.multi(), a watched transaction that commits nothing would leave WATCH armed on the connection, so it is banned. EXEC clears watch state server-side on both success and abort, so no UNWATCH is needed between retries.

When To Reach For Lua Instead

WATCH livelocks under heavy contention by design: many writers keep invalidating each other’s reads, and retries pile up. For very hot check-and-set keys, prefer a Lua script: scripts run atomically on the server, reading before they write without any retry loop. The onAbort metrics exist precisely because this failure mode is load-dependent and invisible in low-traffic testing.