<SYSTEM>This is the full developer documentation for Benni</SYSTEM>

# Blocking Operations

> Block on lists, sorted sets, and streams from a session with a required timeout and typed multi-key attribution.

Blocking commands park a connection until an entry is available or the timeout elapses. Because they monopolize a connection, they live only on the [session](/benni/advanced/sessions/) accessors (`session.list(x)`, `session.zset(x)`, and `session.stream(x)`), never on the shared client. Calling one on `redis.list(x)` is a compile error.

## The Timeout Is Required

[Section titled “The Timeout Is Required”](#the-timeout-is-required)

Every blocking method takes a `{ timeoutSeconds }` option. There is no default: forgetting it is a type error, not a silent block-forever.

```ts
await using s = await redis.session();
const job = await s.list(jobs).blpop("pending", { timeoutSeconds: 5 });
```

The unit is in the name, and fractional values are allowed (`{ timeoutSeconds: 0.1 }`). `0`, negatives, `NaN`, and `Infinity` throw a `TypeError`. Redis treats a `0` timeout as block-forever, and arriving there through arithmetic is a shutdown hazard.

To block forever, spell it out with the literal `{ timeoutSeconds: "forever" }`:

```ts
const job = await s.list(jobs).blpop("pending", { timeoutSeconds: "forever" });
//    ^? Job, never null: the call either resolves a value or rejects on close
```

`{ timeoutSeconds: "forever" }` is a visible, greppable literal rather than a stray `0`. As a bonus it narrows the return type: a forever call cannot time out, so `null` drops from the result. `close()` still rejects a forever-blocked call promptly, so `await using` never hangs.

## Lists

[Section titled “Lists”](#lists)

Session list stores add the blocking pops and blocking move on top of the [regular list methods](/benni/data-structures/sets-and-lists/):

```ts
await using s = await redis.session();
const queue = s.list(jobs);


const job = await queue.blpop("pending", { timeoutSeconds: 5 });   // BLPOP  -> Job | null
const tail = await queue.brpop("pending", { timeoutSeconds: 5 });  // BRPOP  -> Job | null


// BLMOVE: pop from one end of source, push to one end of destination
const moved = await queue.blmove(
  "pending", "processing", "left", "right", { timeoutSeconds: 5 }
); // -> Job | null
```

A single-key pop returns the decoded value or `null` on timeout.

## Sorted Sets

[Section titled “Sorted Sets”](#sorted-sets)

Session sorted-set stores add blocking pops of the lowest or highest scoring member:

```ts
const s = await redis.session();
const min = await s.zset(priorities).bzpopmin("queue", { timeoutSeconds: 5 });
//    ^? { member: string; score: number } | null   (BZPOPMIN)
const max = await s.zset(priorities).bzpopmax("queue", { timeoutSeconds: 5 }); // BZPOPMAX
await s.close();
```

Each returns a `{ member, score }` entry or `null` on timeout.

## Streams

[Section titled “Streams”](#streams)

Session stream stores add a blocking read for entries newer than an id:

```ts
const batch = await s.stream(auditEvents).xread(
  "login", lastSeenId, { timeoutSeconds: 5, count: 100 }
); // XREAD BLOCK -> StreamEntry[]  ([] on timeout)
```

`xread` with a `timeoutSeconds` wraps `XREAD BLOCK` and returns an empty array on timeout, matching non-blocking `xread`’s null-to-`[]` convention. The `afterEntryId` accepts a concrete entry id or `"$"`; track the last seen id across iterations, because `"$"` re-arms “from now” on each call and can miss entries that arrive between calls. For at-least-once delivery across many workers, use [consumer groups](/benni/data-structures/consumer-groups/) instead.

## Typed Id Attribution Across Keys

[Section titled “Typed Id Attribution Across Keys”](#typed-id-attribution-across-keys)

Passing an **array** of keys blocks across several keys at once and tells you which key answered, with the id typed to exactly the keys you passed:

```ts
const hit = await s.list(jobs).blpop(["urgent", "pending"], { timeoutSeconds: 5 });
if (hit) {
  console.log(hit.id, hit.value);
  //         ^? "urgent" | "pending"    ^? Job
}
```

The answering key from the reply is reverse-mapped back to your typed id, so literal id types survive the round trip. The sorted-set forms mirror this shape:

```ts
const hit = await s.zset(priorities).bzpopmin(["high", "low"], { timeoutSeconds: 5 });
//    ^? { id: "high" | "low"; entry: { member: string; score: number } } | null
// brpop and bzpopmax are the mirror-image variants.
```

## Non-Blocking Multi-Key Pops

[Section titled “Non-Blocking Multi-Key Pops”](#non-blocking-multi-key-pops)

`LMPOP` and `ZMPOP` never block, so they land on the **shared** store with the same typed attribution shape, no session needed:

```ts
const hit = await redis.list(jobs).lmpop(["urgent", "pending"], { direction: "left", count: 10 });
//    ^? { id: "urgent" | "pending"; values: Job[] } | null


const scored = await redis.zset(priorities).zmpop(["high", "low"], { min: true, count: 10 });
//    ^? { id: "high" | "low"; entries: Array<{ member: string; score: number }> } | null
// lmpop with { direction: "right" } and zmpop with { max: true } are the mirror-image variants.
```

These check the keys in order and return `null` only when all of them are empty. A session inherits them too, since it reuses the same stores, but reach for the blocking `blpop`/`brpop`/`blmove`/`blmpop` when you want to wait for work rather than poll.

## Blocking Counted Multi-Key Pops

[Section titled “Blocking Counted Multi-Key Pops”](#blocking-counted-multi-key-pops)

`BLMPOP` and `BZMPOP` are the blocking counterparts of `LMPOP`/`ZMPOP`: they pop up to `count` entries from the first non-empty of several keys, blocking until one has data. Like the other blocking commands they are **session-only**, and they carry the same typed attribution:

```ts
await using s = await redis.session();


const hit = await s.list(jobs).blmpop(["urgent", "pending"], {
  direction: "left",
  timeoutSeconds: 5,
  count: 10
});
//    ^? { id: "urgent" | "pending"; values: Job[] } | null


const scored = await s.zset(priorities).bzmpop(["high", "low"], { min: true, count: 10 }, {
  timeoutSeconds: "forever"
});
//    ^? { id: "high" | "low"; entries: Array<{ member: string; score: number }> }
// blmpop with { direction: "right" } and bzmpop with { max: true } are the mirror-image variants.
```

Mind the distinction from `blpop` on a key array (`BLPOP`) above: `blmpop` (`BLMPOP`) pops a **counted batch** from the first non-empty key and returns `values`/`entries` arrays, while `blpop` on a key array (`BLPOP`) pops a **single** item and returns one `value`/`entry`. As with every blocking call, `{ timeoutSeconds: "forever" }` drops `null` from the return type.

## A Reliable Worker Queue

[Section titled “A Reliable Worker Queue”](#a-reliable-worker-queue)

A `BLPOP` reply served onto a connection that dies before the reply is read is lost. The durable pattern is `blmove` (`BLMOVE`) into a per-worker processing list: the move is atomic server-side, so a crash leaves the job recoverable in the processing list. On startup, drain that list before blocking for new work.

schema.ts

```ts
import { list, json } from "benni/schema";
type Job = { id: string; kind: "email" | "report" };
export const jobs = list("jobs", json<Job>());


// worker.ts: reliable BLMOVE loop, shutdown- and crash-safe
const processing = `worker-${process.pid}`;
const stop = new AbortController();
process.on("SIGTERM", () => stop.abort());


// startup recovery: drain this worker's processing list from a previous crash
for (
  let job = await redis.list(jobs).lpop(processing);
  job;
  job = await redis.list(jobs).lpop(processing)
) {
  await handle(job);
}


while (!stop.signal.aborted) {
  await using s = await redis.session();          // one extra connection, owned here
  const abort = () => void s.close();          // close() rejects an in-flight block in ~ms
  stop.signal.addEventListener("abort", abort, { once: true });
  try {
    const queue = s.list(jobs);
    while (!stop.signal.aborted) {
      // redis.list(jobs).blmove(...) would be a compile error; session only.
      const job = await queue.blmove(
        "pending", processing, "left", "right", { timeoutSeconds: 5 }
      );
      if (job === null) continue;              // heartbeat tick: re-check the stop signal
      await handle(job);
      await redis.list(jobs).lrem(processing, 1, job); // ack via the shared client
    }
  } catch (error) {
    if (s.closed) break;                       // shutdown or dropped connection
    await sleep(1000);                          // recover: the outer loop opens a fresh session
  } finally {
    stop.signal.removeEventListener("abort", abort);
  }
}
```

The `{ timeoutSeconds: 5 }` block doubles as a heartbeat: it returns `null` every five seconds so the loop can re-check the stop signal even when the queue is idle. Closing the session on `SIGTERM` rejects the in-flight block within milliseconds rather than waiting out the timeout.

# Redis Cluster

> Declare where a schema's hash tag goes so multi-key commands stay in one slot, and catch cross-slot mistakes at compile time and before they are sent.

Benni models slot **co-location**, not cluster topology. Declare where a schema puts its hash tag and Benni will keep multi-key commands inside one slot, reject the ones that cannot be, and tell you which layout fixes it.

## Benni Does Not Route

[Section titled “Benni Does Not Route”](#benni-does-not-route)

Routing is your driver’s job. Benni has no transport of its own: it binds to whatever `RedisClient` an adapter hands it, so cluster routing comes from the client underneath. Today that means adopting a cluster-aware ioredis instance through [`benni/ioredis`](/benni/runtime/ioredis/); `benni/node` builds its own single-node `createClient()` and cannot be handed a `createCluster()` one.

Topology discovery, `MOVED`/`ASK` redirects, per-node pools, and failover all stay in the driver, which has had a decade to get them right. What no driver can do for you is know, before you send, that a command’s keys belong together. That is the part Benni owns, because Benni is the only client where keys come from schemas rather than string concatenation.

## The Problem

[Section titled “The Problem”](#the-problem)

Redis routes a key by CRC16 of the substring between the first `{` and the first `}` after it, or of the whole key when there is no such pair. Every key in a single command must land on the same slot, or the server answers `CROSSSLOT Keys in request don't hash to the same slot`.

With the default `prefix:id` layout, two ids essentially never share a slot:

```ts
const carts = kv("cart", json<Cart>());


carts.key("u1"); // "cart:u1"  -> slot 13083
carts.key("u2"); // "cart:u2"  -> slot 888
```

So `mget(["u1", "u2"])`, `sunionstore`, `zmpop`, `bitop`, `pfmerge`, `lmove`, and every other multi-key method is unusable on a cluster. This works perfectly on a single node, which is exactly why it is discovered in production.

## The Three Layouts

[Section titled “The Three Layouts”](#the-three-layouts)

`hashTag` is an opt-in option on every keyed schema. Omitting it changes nothing.

| Option              | Key         | What it buys                                                                             |
| ------------------- | ----------- | ---------------------------------------------------------------------------------------- |
| omitted             | `cart:u1`   | Today’s layout. One slot per id.                                                         |
| `hashTag: "prefix"` | `{cart}:u1` | The whole keyspace shares one slot, so every multi-key method over this schema is legal. |
| `hashTag: "id"`     | `cart:{u1}` | Keys stay spread, but the same id co-locates across schemas.                             |

```ts
import { json, kv, zset } from "benni/schema";


// Bounded keyspace, needs within-schema set algebra: pin the whole thing.
const featureFlags = zset("flags", string(), { hashTag: "prefix" });
await redis.zset(featureFlags).zunionstore("all", "beta", ["internal"]);


// Unbounded keyspace, needs per-user co-location: tag the id.
const carts = kv("cart", json<Cart>(), { hashTag: "id" });
const orders = kv("order", json<Order>(), { hashTag: "id" });
// "cart:{u1}" and "order:{u1}" share a slot; "cart:{u2}" does not.
```

### Choosing Between Them

[Section titled “Choosing Between Them”](#choosing-between-them)

This is a co-location decision, not a compatibility flag.

`hashTag: "prefix"` makes multi-key commands legal by putting an entire keyspace on one node. That is right for a leaderboard, a feature-flag set, or a per-tenant index. It is a production incident for your main user keyspace, which will then be served by one node no matter how many you run.

`hashTag: "id"` keeps the distribution and co-locates everything about one entity. Reach for it whenever the thing you want in a single command is “all the data for user X” rather than “all the users”.

Neither is free to adopt later: both change the key format, so turning one on orphans existing data.

A `hashTag: "id"` prefix may not contain `{`. Redis reads the tag from the first `{` in the whole key, so a prefix like `cart{v2}` would take the tag away from the id and quietly undo the co-location the layout exists for. The schema builder rejects it when you declare the schema.

## Compile-Time Checking

[Section titled “Compile-Time Checking”](#compile-time-checking)

Because the tag is part of the key’s template-literal type, Benni can reject cross-slot combinations before you run anything. This covers `script().run()`, `redis.watch()`, and the transaction key declaration:

```ts
await redis.script(moveItem).run({
  keys: { from: carts.key("u1"), to: orders.key("u2") },
  //                                  ^ Type '"order:{u2}"' is not assignable to type
  //                                    'KeysMustShareOneHashSlot<"order:{u2}", "u1">'
  args: { amount: 1 }
});
```

The alias name is the error message: these keys must share one hash slot, and the tag they had to match was `u1`.

**A passing check means “no provable conflict”, not “provably co-located.”** Benni rejects only pairs whose hash tags are distinct string literals. Three things pass silently:

* Untagged keys, so adopting `hashTag` on one schema never breaks unrelated call sites.
* Keys built from runtime ids. `carts.key(userId)` has type `` `cart:{${string}}` ``, and no type system can tell whether two of those hold the same value.
* Raw `multi().add()` commands and the within-schema multi-key methods, which are checked at runtime instead.

That last group is why the runtime guard exists.

## Runtime Checking

[Section titled “Runtime Checking”](#runtime-checking)

Install the guard and every multi-key command is verified before it is sent:

```ts
import { assertSameSlot } from "benni/cluster";


const redis = benni(client, { cluster: assertSameSlot });


await redis.set(sessions).sunion("a1", ["b7"]);
// CrossSlotError: SUNION spans two Redis Cluster hash slots, which the server
// rejects with CROSSSLOT.
//   "sessions:a1" hashes to slot 9716
//   "sessions:b7" hashes to slot 4193
// Every key in one command must hash to the same slot. Declare the schema with
// hashTag: "prefix" so its keys become "{sessions}:<id>" and the whole keyspace
// shares one slot, or with hashTag: "id" so its keys become "sessions:{<id>}"
// and stay spread while co-locating one id across schemas.
```

`CrossSlotError` extends `ValidationError`, carries `command`, `keys`, and `slots`, and is thrown before anything reaches the socket. It is exported from `benni/cluster` alongside `slotOf` and `hashTagOf`.

The guard is **off by default** and should be: cross-slot multi-key commands are perfectly legal on a single-node Redis, and plenty of code relies on that. Turn it on in development and CI, where you want the mistake to surface.

### Why You Pass The Checker, Not `true`

[Section titled “Why You Pass The Checker, Not true”](#why-you-pass-the-checker-not-true)

`benni()` has to reference the guard in order to install it, so a `cluster: true` boolean would mean the root entry names it, and no bundler could then drop it. The CRC16 table and the error’s fix-hint prose would ship in every app, including every app that never turns the check on. Taking the function as a value moves all of it into `benni/cluster`, which only an app that imports it ever pays for. That is about 1.4 KB gzipped, roughly 15% of the default root entry.

The cost when it *is* installed is close to nothing. The guard compares hash tags as strings first, so a correctly configured schema never runs a CRC at all. When it is absent, each check is an optional call on an undefined function, and an optional call short-circuits its argument evaluation, so the key arrays are never even built.

### Declaring Transaction Keys

[Section titled “Declaring Transaction Keys”](#declaring-transaction-keys)

`multi()` queues raw command tuples, so Benni cannot see which keys they touch. Declare them:

```ts
await redis
  .multi()
  .keys([carts.key("u1"), orders.key("u1")])
  .add(["INCR", carts.key("u1")], numberReply)
  .add(["SADD", orders.key("u1"), "x"], numberReply)
  .exec();
```

This is a declaration, not a derivation. A key you queue but do not declare is not checked. Keys accumulate across calls, and each call is checked against the first tag seen, including across an intervening `add()`.

## What Is Not Covered

[Section titled “What Is Not Covered”](#what-is-not-covered)

Be clear-eyed about the boundary. A green build is not a cluster-safe build:

* Benni does not route. `MOVED`, `ASK`, topology, and failover are the driver’s.
* Sharded pub/sub (`SSUBSCRIBE`/`SPUBLISH`) is not modeled.
* `redis.raw.send()` and `redis.raw.pipeline()` bypass every check by design.
* Each surface validates itself. The keys you `watch()` are not checked against the keys the body’s transaction declares.

## Primitives

[Section titled “Primitives”](#primitives)

The bundled primitives are already slot-safe. `lock` and `ratelimit` touch one key per call. `queue` hash-tags every key into one slot, which is also what lets its Lua derive per-job key names. `cache` tags the id, so an entry and its own fill lock share a node while the cache itself stays spread.

# Optimistic Transactions

> Use redis.watch() to run a WATCH/MULTI/EXEC check-and-set that retries on conflict.

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](/benni/advanced/sessions/) connection, because `WATCH` state belongs to one connection.

## `redis.watch(keys, body, options?)`

[Section titled “redis.watch(keys, body, options?)”](#rediswatchkeys-body-options)

```ts
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 `UNWATCH`es 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()`](/benni/advanced/transactions/), and whose `exec()` resolves the tuple or `null` on abort.

Everything [`redis.multi()`](/benni/advanced/transactions/#encode-values-with-the-schemas-codec) 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](/benni/advanced/transactions/#the-decoder-is-not-checked-against-the-command) 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

[Section titled “Check-And-Set Example”](#check-and-set-example)

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

```ts
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

[Section titled “Balance Transfer”](#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:

```ts
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

[Section titled “The Write Side Is Not Typed The Way Stores Are”](#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](/benni/advanced/transactions/#the-decoder-is-not-checked-against-the-command).
* **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](/benni/advanced/scripts/) is the alternative: `script()` names its keys and types its args, so the call site is checked even though the body is Lua.

## Options

[Section titled “Options”](#options)

```ts
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

[Section titled “Abort, Opt-Out, And Exhaustion”](#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 `UNWATCH`es 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.

```ts
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

[Section titled “Manual Form”](#manual-form)

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

```ts
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

[Section titled “When To Reach For Lua Instead”](#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](/benni/advanced/scripts/): 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.

# Scans

> Use redis.scan to iterate keys and collection members incrementally without blocking Redis.

Use `redis.scan` to iterate keys and collection members incrementally without blocking Redis. Each scan returns an async iterable that pages through Redis cursors behind the scenes.

## Scan All Keys

[Section titled “Scan All Keys”](#scan-all-keys)

```ts
for await (const key of redis.scan.keys({ match: "user:*", count: 500 })) {
  console.log(key);
}
```

Options:

* `match` filters keys server-side with a glob pattern.
* `count` is a page-size hint per round trip, not a result limit.
* `type` restricts results to one Redis type such as `"hash"` or `"string"`.

## Scan A Schema’s Keys

[Section titled “Scan A Schema’s Keys”](#scan-a-schemas-keys)

```ts
for await (const key of redis.scan.kv(profiles)) {
  // key is "profile:<id>"
}
```

`redis.scan.kv` defaults `match` to the schema prefix (`profile:*`), so you only see keys that belong to that schema. Pass your own `match` to narrow it further.

Scanning lives on `redis.scan`, not on the stores: there is no `redis.kv(profiles).scan()` or `redis.query.profiles.scan()`. A scan yields keys and members rather than records, and it is a cursor loop rather than a single command, so it stays a namespace of its own instead of a method on every store.

## Scan Collection Members

[Section titled “Scan Collection Members”](#scan-collection-members)

```ts
for await (const member of redis.scan.set(teamMembers, "engineering")) {
  console.log(member);
}


for await (const entry of redis.scan.hash(users, "42")) {
  // entry is { field: "name"; value: string } | { field: "score"; value: number }
}


for await (const { member, score } of redis.scan.zset(leaderboards, "global")) {
  console.log(member, score);
}
```

Member scans wrap `SSCAN`, `HSCAN`, and `ZSCAN` and accept `{ match, count }`. Values decode through the schema’s codecs, and hash scans skip fields that are not declared in the schema.

## Early Break Is Safe

[Section titled “Early Break Is Safe”](#early-break-is-safe)

```ts
for await (const key of redis.scan.keys({ match: "session:*" })) {
  if (await handle(key)) break;
}
```

Redis scan cursors are stateless on the server, so breaking out of the loop simply stops issuing `SCAN` calls. There is nothing to close or clean up.

## Guarantees

[Section titled “Guarantees”](#guarantees)

`SCAN` offers a weak snapshot: keys that exist for the whole scan are returned at least once, but keys created or deleted mid-scan may or may not appear, and a key can be returned more than once. Deduplicate when your use case needs exactly-once handling.

# Scripts

> Use the script() builder to define Lua scripts with named keys, typed args, and a typed return value.

Use the `script()` builder to define Lua scripts with named keys, typed arguments, and a typed return value.

## Define A Script

[Section titled “Define A Script”](#define-a-script)

```ts
import { number, script } from "benni/schema";


export const rateLimit = script("rate-limit", {
  keys: ["counter"],
  args: { limit: number(), windowSeconds: number() },
  returns: number(),
  lua: `
    local current = redis.call("INCR", KEYS[1])
    if current == 1 then
      redis.call("EXPIRE", KEYS[1], ARGV[2])
    end
    if current > tonumber(ARGV[1]) then
      return 0
    end
    return current
  `
});
```

Keys are named and map to `KEYS[1..n]` in declared order. Args encode through their codecs and map to `ARGV[1..n]` in declared order.

## Run A Script

[Section titled “Run A Script”](#run-a-script)

```ts
const current = await redis.script(rateLimit).run({
  keys: { counter: "rate:user:42" },
  args: { limit: 100, windowSeconds: 60 }
});
//    ^? number
```

Both `keys` and `args` are checked at compile time: missing or misspelled names are type errors.

## EVALSHA Caching

[Section titled “EVALSHA Caching”](#evalsha-caching)

The first run loads the script with `SCRIPT LOAD` and caches its SHA per client. Later runs send only the hash via `EVALSHA`. When Redis replies with `NOSCRIPT` (after a server restart or `SCRIPT FLUSH`), Benni reloads the script and retries automatically, so callers never see the error.

A script can also return a `NOSCRIPT`-coded error of its own, and Redis passes it through byte for byte, so the message alone cannot tell the two apart. Before reloading a cached SHA, Benni asks the server with `SCRIPT EXISTS`: if the script is still there, the error came from the script and is raised as-is, rather than re-running side effects the script has already applied.

## Scalar Returns Only

[Section titled “Scalar Returns Only”](#scalar-returns-only)

`returns` decodes scalar replies (strings and numbers) through its codec. A script that returns a table or nil throws `TypeError: Expected Redis script reply to decode from scalar`.

For structured replies, drop to `defineScript` from the main entrypoint and decode the raw reply yourself:

```ts
import { createScriptRunner, defineScript } from "benni";


const topTwo = defineScript<[], string[]>({
  lua: `return redis.call("ZREVRANGE", KEYS[1], 0, 1)`,
  keyCount: 1,
  decode(reply) {
    if (!Array.isArray(reply)) {
      throw new TypeError("Expected Redis script reply to return array");
    }
    return reply as string[];
  }
});


const runner = createScriptRunner(client);
const top = await runner.run(topTwo, ["leaderboard:global"], []);
```

`defineScript` uses positional keys and args instead of named ones, but the same `EVALSHA` caching and `NOSCRIPT` recovery apply.

# Connection Sessions

> Use redis.session() to lease a dedicated connection for blocking commands and WATCH transactions.

Use `redis.session()` to lease a dedicated connection from the client. A session is shaped like the Benni handle (the same store accessors, bound to a private connection), plus the operations that are only safe when one caller owns the connection: blocking pops, blocking stream reads, and `WATCH`.

## Why A Session Is A Dedicated Connection

[Section titled “Why A Session Is A Dedicated Connection”](#why-a-session-is-a-dedicated-connection)

Two Redis workloads monopolize a connection. A blocking command (`BLPOP`, `BLMOVE`, `BZPOPMIN`, `XREAD`/`XREADGROUP` with `BLOCK`) parks the connection until an entry arrives or the timeout elapses; nothing else can use it meanwhile. A `WATCH`/`MULTI`/`EXEC` transaction arms optimistic-locking state that belongs to one connection and must not be interleaved with unrelated traffic.

Running either on the shared client would stall every other query. So Benni puts them behind a session: one session is one connection is one logical worker. There is no pooling: an app that needs N workers blocked at once opens N sessions, and the connection cost is explicit (the shared client is one connection; each live session is exactly one more).

Because the blocking and `WATCH` methods live only on the session-flavored accessors, calling them on the shared client is a compile error, not a runtime surprise:

```ts
await using session = await redis.session();
session.list(jobs).blpop("pending", { timeoutSeconds: 5 }); // exists
redis.list(jobs).blpop("pending", { timeoutSeconds: 5 });       // compile error
```

## Open A Session

[Section titled “Open A Session”](#open-a-session)

`redis.session()` has two forms. The scoped form takes a callback and closes the session for you when the callback settles. This is the recommended shape for a bounded unit of work:

```ts
const job = await redis.session(async (s) => {
  return s.list(jobs).blpop("pending", { timeoutSeconds: 5 });
});
```

The bare form returns the session and hands you the `close()` obligation. Pair it with `await using` (TypeScript 5.2 explicit resource management) so it closes at the end of the scope even on an early return or a throw:

```ts
await using session = await redis.session();
const job = await session.list(jobs).blpop("pending", { timeoutSeconds: 5 });
// session.close() runs automatically when the block exits
```

Without `await using`, you own `close()` and must call it in a `finally`:

```ts
const session = await redis.session();
try {
  const job = await session.list(jobs).blpop("pending", { timeoutSeconds: 5 });
} finally {
  await session.close();
}
```

## What A Session Exposes

[Section titled “What A Session Exposes”](#what-a-session-exposes)

A session carries every data-store accessor from the Benni handle (`kv`, `hash`, `list`, `set`, `zset`, `stream`, `counter`, `string`, `bitmap`, `geo`, `hll`), each bound to the private connection. The `list`, `zset`, and `stream` accessors are supersets that also expose their blocking variants (see [Blocking Operations](/benni/advanced/blocking-operations/)) and the blocking consumer-group read (see [Consumer Groups](/benni/data-structures/consumer-groups/)).

On top of the stores, a session adds the `WATCH` primitives:

```ts
session.watch(keys);   // WATCH k1 k2…; throws on an empty list
session.unwatch();     // UNWATCH
session.multi();        // abort-aware transaction builder; exec() resolves the tuple or null
```

See [Optimistic Transactions](/benni/advanced/optimistic-transactions/) for the retrying `redis.watch(...)` helper built on these.

`scan`, `pubsub`, and `script` are intentionally absent from a session: they have no session-specific semantics, and the smaller surface keeps a session’s purpose legible: block, or watch-then-commit. For raw commands there is `session.raw`, the underlying adapter session.

## Closed And Close Semantics

[Section titled “Closed And Close Semantics”](#closed-and-close-semantics)

`close()` tears the connection down immediately. It rejects any in-flight command and does **not** wait out a server-side blocking timeout, so a session parked on `{ timeoutSeconds: "forever" }` still exits promptly when you close it. `close()` is idempotent, and `[Symbol.asyncDispose]` is an alias of it, which is what makes `await using` safe.

There is no reconnection. A session that you close, or one whose connection drops, is dead: every in-flight and subsequent call rejects. Recovery is a new session, never a reconnect: a silent reconnect would drop `WATCH` state and blocked reads and turn visible failures into correctness bugs.

Check `session.closed` to tell a shutdown or dropped connection apart from an application error inside a worker loop:

```ts
try {
  const job = await session.list(jobs).blmove(
    "pending", "processing", "left", "right", { timeoutSeconds: 5 }
  );
} catch (error) {
  if (session.closed) return; // shutdown or dropped connection, stop the loop
  throw error;                 // a real error, surface it
}
```

A boolean check is robust where cross-adapter error-class mapping is fragile; a use-after-close otherwise rejects with `SessionClosedError`.

## Requirements

[Section titled “Requirements”](#requirements)

The bound client must implement the optional `session` method of the `RedisClient` interface (the Node and Bun adapters do). Otherwise `redis.session()` throws `TypeError: Redis client does not support sessions`, the same style as the `transaction` guard. Deno uses the Node adapter through npm compatibility, so session support follows the Node adapter there. The edge HTTP adapter omits sessions because it has no persistent connection.

Sessions pin real connections. Prefer the scoped callback form or `await using` so a session cannot outlive its work; as a backstop the parent client tracks live sessions and force-closes any survivors when you close the client.

# Transactions

> Use redis.multi() to run commands atomically with MULTI/EXEC and decode replies into a typed tuple.

Use `redis.multi()` to run several commands atomically with `MULTI`/`EXEC` and decode the replies into a typed tuple. It is the same builder a [session](/benni/advanced/sessions/) exposes as `s.multi()`.

## Build And Execute

[Section titled “Build And Execute”](#build-and-execute)

```ts
import { numberReply, okReply, stringOrNullReply } from "benni";
import { kv, number, string } from "benni/schema";


const names = kv("name", string());
const hits = kv("hits", number());
const drafts = kv("draft", string());


const [, visits, draft] = await redis
  .multi()
  .add(["SET", names.key("42"), names.encode("Ada")], okReply)
  .add(["INCR", hits.key("42")], numberReply)
  .add(["GETDEL", drafts.key("42")], stringOrNullReply)
  .exec();
//    ^? [void, number, string | null]
```

Each `.add(command, decoder)` queues one command and extends the result tuple type. `exec()` sends everything as one transaction and returns the decoded tuple, so results stay position-typed.

Note the argument encoding: values go through the schema’s own codec (`names.encode("Ada")`), never through `String(...)` or `JSON.stringify(...)`. That is the point of the next section.

## Encode Values With The Schema’s Codec

[Section titled “Encode Values With The Schema’s Codec”](#encode-values-with-the-schemas-codec)

`.add()` takes a raw command tuple, so Benni cannot encode arguments for you the way `redis.kv(schema).set()` does. It does not follow that you have to hand-encode them. Every schema exposes the codec its own store uses:

* Value-carrying keyspaces (`kv`, `set`, `list`, `zset`, `geo`) expose `encode(value)` and `decode(stored)` directly on the schema. A `hll` schema exposes `encode` only, since a HyperLogLog cannot be read back.
* `hash` and `stream` schemas expose `fields`, where each entry is a `Codec` with its own `.encode()` and `.decode()`.

```ts
import { hash, kv, number, string } from "benni/schema";


const links = hash("link", { url: string(), createdAt: number() });
const clicks = kv("clicks", number());


await redis
  .multi()
  .add(
    [
      "HSET",
      links.key("typed"),
      "url",
      links.fields.url.encode("https://typed.example"),
      "createdAt",
      links.fields.createdAt.encode(Date.now())
    ],
    numberReply
  )
  .add(["SET", clicks.key("typed"), clicks.encode(0)], okReply)
  .exec();
```

Reading those keys back through the typed stores returns `createdAt` as a real `number` and `clicks` as a real `number`, because the transaction wrote exactly the bytes the stores decode.

Why bother, when `String(Date.now())` produces the same string today? Because the codec is the same one the typed store uses, so a value the store would accept is the value that lands in Redis. A hand-rolled `String(...)` agrees with the codec by coincidence, and stops agreeing the moment the codec differs from plain stringification: `boolean()` stores `"1"`, not `"true"`; `json()` refuses non-finite numbers that `JSON.stringify` turns into `null`; a [Zod codec](/benni/integrations/zod/) may normalize the value on the way in. When the schema changes, codec-encoded writes follow it and hand-encoded ones silently do not.

It also moves failures to the write site. `links.fields.createdAt.encode("2026-08-02")` does not compile, because the field’s input type is `number`. A bad value that reaches the codec at runtime fails there (`ValidationError: number codec cannot encode a non-finite value`) instead of being stored and then poisoning a later read with a `ReplyShapeError` from a different part of the codebase.

## Reply Decoders

[Section titled “Reply Decoders”](#reply-decoders)

```ts
okReply;            // asserts "OK", returns void
numberReply;        // number
stringReply;        // string
stringOrNullReply;  // string | null
booleanNumberReply; // 1 -> true, 0 -> false
```

Decoders throw a `ReplyShapeError` (a `TypeError` subclass) when the reply shape does not match, so a wrong decoder fails loudly instead of leaking untyped values. Any `(reply: RedisReply) => T` function works as a decoder for other shapes.

`okReply` yields `void`, which is `undefined` at runtime, so a successful `SET` reads as `undefined` in the result tuple. A tuple of `[3, undefined]` is a transaction where **both** commands succeeded: an `OK` carries no information worth typing, so the slot is `void` rather than a `true` nobody would check. A failed command does not produce `undefined`, it throws.

## The Decoder Is Not Checked Against The Command

[Section titled “The Decoder Is Not Checked Against The Command”](#the-decoder-is-not-checked-against-the-command)

Here is the guarantee `.add()` cannot give you. A decoder is a plain `(reply: RedisReply) => T` function, and nothing ties it to the command string sitting beside it. Pair `numberReply` with a `SET` and it compiles:

```ts
await redis
  .multi()
  .add(["SET", names.key("42"), names.encode("Ada")], numberReply) // compiles, wrong
  .exec();
// ReplyShapeError: Expected Redis transaction reply to return number, got string "OK"
```

The error is precise, but it arrives at `exec()`, one round trip after the compiler could have caught it, and after the write committed. Encoding is type-checked; the decoder pairing is the one thing in a transaction you still have to get right by reading. The common pairings:

| Command                                                 | Decoder              | Slot type        |
| ------------------------------------------------------- | -------------------- | ---------------- |
| `SET`, `MSET`, `RENAME`                                 | `okReply`            | `void`           |
| `INCR`, `INCRBY`, `HSET`, `DEL`, `SADD`, `ZADD`, `LLEN` | `numberReply`        | `number`         |
| `GET`, `GETDEL`, `HGET`, `LPOP`                         | `stringOrNullReply`  | `string \| null` |
| `EXPIRE`, `SETNX`, `SISMEMBER`                          | `booleanNumberReply` | `boolean`        |

To have the compiler check the whole operation, reach for a [Lua script](/benni/advanced/scripts/) instead. `script()` declares its `keys`, `args`, and `returns` codec alongside the body, so nothing about it is paired by convention.

## Builders Are Immutable

[Section titled “Builders Are Immutable”](#builders-are-immutable)

Each `.add` returns a new builder, so partial transactions can be shared and branched without affecting each other:

```ts
const base = redis.multi().add(["INCR", "hits"], numberReply);


const withA = base.add(["GET", "a"], stringOrNullReply);
const withB = base.add(["GET", "b"], stringOrNullReply);
```

An empty transaction resolves to `[]` without contacting Redis.

## Declaring Keys

[Section titled “Declaring Keys”](#declaring-keys)

`.add()` queues raw command tuples, so Benni cannot tell which keys they touch. On Redis Cluster that matters, because every key in one transaction must hash to the same slot. Declare them with `.keys()`:

```ts
await redis
  .multi()
  .keys([carts.key("u1"), orders.key("u1")])
  .add(["INCR", carts.key("u1")], numberReply)
  .add(["SADD", orders.key("u1"), orders.encode("x")], numberReply)
  .exec();
```

Keys accumulate across calls, and the declared set is checked at compile time and, under a cluster guard such as `benni(client, { cluster: assertSameSlot })`, again before `EXEC` is sent. This is a declaration rather than a derivation: a key you queue but never declare is not checked. On a single-node Redis you can skip it entirely. See [Redis Cluster](/benni/advanced/cluster/).

## Requirements And Limits

[Section titled “Requirements And Limits”](#requirements-and-limits)

The bound client must implement the optional `transaction` method of the `RedisClient` interface (the Node and Bun adapters do). Otherwise `exec()` throws `TypeError: Redis client does not support transactions`.

For check-and-set logic that reads before it writes, use [optimistic transactions](/benni/advanced/optimistic-transactions/): `redis.watch()` runs a `WATCH`/`MULTI`/`EXEC` loop that retries on conflict. Very hot keys are still better served by a [Lua script](/benni/advanced/scripts/), which runs atomically on the server without a retry loop.

# Benni Client

> Create a Benni client by passing a Redis adapter to benni(), then reach every schema through typed data-structure accessors.

Create a Benni client by passing a Redis adapter to `benni`. It takes either shape, and they are the same call:

```ts
import { benni } from "benni";
import { node } from "benni/node";


// One object, no top-level await: the promise resolves on the first command.
const redis = benni({ client: node({ url }), schema });


// Or a client you already have.
const redis = benni(client, { schema });
```

The `client` accepts a connected `RedisClient`, a promise of one, a factory returning either, or another Benni handle. The cost of either lazy form is that a bad `REDIS_URL` surfaces at the first command instead of at startup.

The two lazy forms differ in one way worth knowing:

| Source                            | When it connects                    | `close()` before the first command | After a failed connect                                                       |
| --------------------------------- | ----------------------------------- | ---------------------------------- | ---------------------------------------------------------------------------- |
| A promise (`node({ url })`)       | Already connecting when you pass it | Closes the client it opened        | Every command reports that same failure: a settled promise cannot be retried |
| A factory (`() => node({ url })`) | On the first command                | Opens nothing                      | The next command calls the factory again                                     |

Reach for the factory when a module is loaded in a context that must not connect at all, which is why `benni/next`’s `cacheHandler` documents one: Next.js loads `cache-handler.mjs` at build time.

`close()` is final for both, and for the factory that is the point: a request that lands after shutdown rejects with `Redis client is closed` instead of calling the factory and opening a connection nothing is left to close. Calling `close()` more than once is a no-op.

`BenniOptions` has three fields, all optional:

| Option          | Effect                                                                                                                                                                                                    |
| --------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `schema`        | The schema module that backs [`redis.query`](#redisquery).                                                                                                                                                |
| `onPubSubError` | Called when a Pub/Sub handler throws (see [`redis.pubsub`](#redispubsub)). Without it, the error is rethrown asynchronously rather than swallowed.                                                        |
| `cluster`       | Check, before sending, that every key in a multi-key command hashes to one Redis Cluster slot, throwing `CrossSlotError` when it does not. Off by default. See [Redis Cluster](/benni/advanced/cluster/). |

Everything else a client can do follows from the adapter you pass in.

Every data-structure accessor exposes the store’s methods plus `key(id)` for the full Redis key and `del(id)`.

## Registering The Schema Module

[Section titled “Registering The Schema Module”](#registering-the-schema-module)

Declare the schema module once through the `Register` interface and the bare `Benni` type is the fully typed handle everywhere, so no signature has to repeat `typeof schema`:

redis.ts

```ts
import * as schema from "./schema";


export const redis = benni({ client: node({ url }), schema });


declare module "benni" {
  interface Register {
    schema: typeof schema;
  }
}
```

```ts
import type { Benni } from "benni";


export function makeHandlers(redis: Benni) {
  // redis.query.users, redis.hash(...), ... all fully typed
}
```

Registration is optional and changes nothing else. Without it, `Benni` stays generic over the open schema type and you name the module explicitly:

```ts
export function makeHandlers(redis: Benni<typeof schema>) { /* ... */ }
```

Pass the generic explicitly for a second handle bound to a different module, too: the registration sets the default, not a ceiling.

## `redis.query`

[Section titled “redis.query”](#redisquery)

The schema registry. When a `{ schema }` module is bound, `redis.query.<exportName>` resolves each schema to its typed resource, dispatched by the schema’s `kind`:

```ts
await redis.query.users.hset("42", { name: "Ada", score: 10 });


const user = await redis.query.users.hget("42");
//    ^? { name: string; score: number } | null


await redis.query.leaderboard.zadd("daily", [{ member: "ada", score: 100 }]);
```

`redis.query.<name>` returns the same resource as the matching `redis.<kind>(schema)` accessor. It covers the twelve data kinds (`kv`, `hash`, `set`, `list`, `zset`, `stream`, `bitmap`, `geo`, `hll`, pub/sub channels and patterns, and scripts) and the seven primitives (`cache`, `ratelimit`, `queue`, `lock`, `semaphore`, `idempotency`, `budget`):

```ts
await redis.query.userEvents.publish({ id: "42", action: "created" });
await redis.query.rateLimit.run({ keys: { counter: "user:42" }, args: { limit: 100 } });


// A primitive declared in the same module, reached the same way.
const profile = await redis.query.profiles.get(userId, () => db.load(userId));
```

Counter and string stores are not separate kinds, so a `kv` schema always maps to the `kv` resource. `redis.query.<name>` on a `kv(prefix, number())` therefore has `get` / `set` / `del` but no `incr`: reach for `redis.counter(schema)` for the counter commands and `redis.string(schema)` for the string ones. Both work on the same keys as the `kv` resource, so mixing them on one schema is fine.

```ts
await redis.query.clicks.set("home", 0);          // kv resource
await redis.counter(clicks).incr("home");         // counter view, not on redis.query
```

Non-schema exports (types, helpers) are dropped, and `redis.query` is `{}` when no schema is bound. See [Schema Registry](/benni/core-concepts/schema-registry/).

## `redis.kv(schema)`

[Section titled “redis.kv(schema)”](#rediskvschema)

Typed Redis string values:

```ts
await redis.kv(profiles).set("42", profile, { ttlSeconds: 3600 });
const loaded = await redis.kv(profiles).get("42");
await redis.kv(profiles).del("42");
```

`set` returns `Promise<void>` for plain writes. With `{ nx: true }` (only create) or `{ xx: true }` (only update) it returns `Promise<boolean>` indicating whether the write happened:

```ts
const created = await redis.kv(profiles).set("42", profile, { nx: true });
const updated = await redis.kv(profiles).set("42", profile, { xx: true });
```

## `redis.string(schema)`

[Section titled “redis.string(schema)”](#redisstringschema)

String operations for `kv` schemas with a `string()` codec:

```ts
const drafts = kv("draft", string());


await redis.string(drafts).append("42", " more text");
const slice = await redis.string(drafts).getrange("42", 0, 4);
const length = await redis.string(drafts).strlen("42");
const value = await redis.string(drafts).getex("42", 3600);


// LCS: longest common subsequence of two keys in the same schema.
const sub = await redis.string(drafts).lcs("42", "43"); // the subsequence string
const len = await redis.string(drafts).lcs("42", "43", { len: true }); // its length
const idx = await redis.string(drafts).lcs("42", "43", {
  idx: true,
  withMatchLen: true
});
//    ^? { matches: { a: [number, number]; b: [number, number]; length?: number }[]; length: number }
```

`getrange`, `setrange`, and `strlen` work in **bytes**, not string indices, because that is how Redis indexes a string. For ASCII the two are the same. For anything else they are not: `"café"` is 5 bytes and 4 characters. A range boundary that falls inside a multi-byte character decodes to the replacement character, so read the whole value with `getrange(id, 0, -1)`, or split your chunks on byte boundaries you computed yourself.

## `redis.counter(schema)`

[Section titled “redis.counter(schema)”](#rediscounterschema)

Atomic counters for `kv` schemas with a `number()` codec:

```ts
const hits = kv("hits", number());


const total = await redis.counter(hits).incr("42");
await redis.counter(hits).incrby("42", 10);
await redis.counter(hits).decrby("42", 3);
```

Redis counters are 64-bit. Once a counter passes `Number.MAX_SAFE_INTEGER` its value can no longer be represented exactly as a JavaScript number, so the integer commands throw a `ReplyShapeError` rather than resolve a rounded one. The same applies to `BITFIELD` reads of the wide encodings (`i64`, `u63`).

## `redis.hash(schema)`

[Section titled “redis.hash(schema)”](#redishashschema)

Typed Redis hashes:

```ts
await redis.hash(users).hset("42", { name: "Ada", score: 10 });
await redis.hash(users).hset("42", "score", 11);
const user = await redis.hash(users).hget("42");
const field = await redis.hash(users).hrandfield("42");
```

## `redis.set(schema)`

[Section titled “redis.set(schema)”](#redissetschema)

Typed Redis sets:

```ts
await redis.set(teamMembers).sadd("engineering", ["ada"]);
const members = await redis.set(teamMembers).smembers("engineering");
```

## `redis.list(schema)`

[Section titled “redis.list(schema)”](#redislistschema)

Typed Redis lists:

```ts
await redis.list(events).rpush("user:42", [event]);
const recent = await redis.list(events).lrange("user:42", 0, 9);
```

## `redis.zset(schema)`

[Section titled “redis.zset(schema)”](#rediszsetschema)

Typed Redis sorted sets:

```ts
await redis.zset(leaderboards).zadd("weekly", [
  { member: "user:42", score: 100 }
]);


const top = await redis.zset(leaderboards).zrange("weekly", {
  start: 0,
  stop: 9,
  rev: true
});
```

When members share a score, `zrange` with `{ byLex: true }` ranges over them lexically, as do `zlexcount`, `zremrangebylex`, and `zrangestore` with `{ byLex: true }`:

```ts
const names = await redis.zset(nameIndex).zrange("directory", {
  byLex: true,
  min: { value: "ada" },
  max: "+"
});
```

See [Lexicographic Ranges](/benni/data-structures/sorted-sets/#lexicographic-ranges).

## `redis.hll(schema)`

[Section titled “redis.hll(schema)”](#redishllschema)

Typed Redis HyperLogLog values:

```ts
await redis.hll(pageViews).pfadd("2026-07-04", ["user:42"]);
const count = await redis.hll(pageViews).pfcount("2026-07-04");
```

## `redis.stream(schema)`

[Section titled “redis.stream(schema)”](#redisstreamschema)

Typed Redis streams:

```ts
await redis.stream(activity).xadd("42", { action: "login", points: 5 });
const entries = await redis.stream(activity).xrange("42", { count: 10 });
```

`.group(name)` opens a consumer group on the stream for at-least-once delivery across workers:

```ts
const group = redis.stream(activity).group("processors");
await group.create("42", { from: "start" });
const batch = await group.consumer("w-1").xreadgroup("42", { count: 10 });
```

See [Consumer Groups](/benni/data-structures/consumer-groups/).

## `redis.bitmap(schema)`

[Section titled “redis.bitmap(schema)”](#redisbitmapschema)

Typed Redis bitmaps:

```ts
await redis.bitmap(dailyActive).setbit("2026-07-04", 42, true);
const total = await redis.bitmap(dailyActive).bitcount("2026-07-04");


// Packed integer fields via BITFIELD; the result tuple is typed to the chain.
const [visits] = await redis.bitmap(dailyActive)
  .bitfield("2026-07-04")
  .incrby("u32", 0, 1)
  .exec();
```

## `redis.geo(schema)`

[Section titled “redis.geo(schema)”](#redisgeoschema)

Typed Redis geospatial indexes:

```ts
await redis.geo(stores).geoadd("berlin", [
  { member: "store:1", longitude: 13.405, latitude: 52.52 }
]);


const nearby = await redis.geo(stores).geosearch("berlin", {
  from: { longitude: 13.4, latitude: 52.52 },
  by: { radius: 5, unit: "km" }
});
```

## `redis.scan`

[Section titled “redis.scan”](#redisscan)

Async-iterable scans over keys and collection members:

```ts
for await (const key of redis.scan.keys({ match: "user:*" })) {
  console.log(key);
}


for await (const key of redis.scan.kv(profiles)) { /* profile:* keys */ }
for await (const member of redis.scan.set(teamMembers, "engineering")) { /* ... */ }
for await (const entry of redis.scan.hash(users, "42")) { /* { field, value } */ }
for await (const entry of redis.scan.zset(leaderboards, "global")) { /* { member, score } */ }
```

See [Scans](/benni/advanced/scans/) for options and iteration guarantees.

## `redis.pubsub`

[Section titled “redis.pubsub”](#redispubsub)

Typed publish and subscribe. `PUBLISH` is a stateless command, so publishing rides the bound client and works on every adapter; it returns the number of subscribers Redis delivered to:

```ts
const receivers = await redis.pubsub.channel(userEvents).publish({
  id: "42",
  action: "created"
});
```

`subscribe` takes just a handler and returns a subscription with `unsubscribe()`. The first subscription lazily leases one subscriber connection from the client and every channel and pattern is multiplexed onto it; it closes when the last subscription goes away:

```ts
const subscription = await redis.pubsub.channel(userEvents).subscribe((message) => {
  // message is the channel's decoded output type
});


await subscription.unsubscribe();
```

`redis.pubsub.pattern(...).subscribe(handler)` receives every matching channel, and the handler’s second argument is the concrete channel name:

```ts
const patternSubscription = await redis.pubsub
  .pattern(userEventPattern)
  .subscribe((message, channelName) => { /* ... */ });
```

`stream(options?)` is the async-iterator form of the same subscription. A channel stream yields decoded messages; a pattern stream yields `{ message, channel }`. Aborting `options.signal` (or leaving the loop) ends iteration and releases the subscription:

```ts
const controller = new AbortController();


for await (const message of redis.pubsub
  .channel(userEvents)
  .stream({ signal: controller.signal })) {
  // ...
}
```

`redis.pubsub.close()` drops every subscription and closes the leased connection. Publishing keeps working afterwards, and the next `subscribe` leases a fresh connection:

```ts
await redis.pubsub.close();
```

Subscribing requires a client that can hold a connection. An adapter advertises this with the optional `subscriber?()` method on the `RedisClient` contract, the pub/sub counterpart to `session?()`:

```ts
import type { RedisClient, RedisSubscriber } from "benni";


declare const client: RedisClient;
//    ^? { send, pipeline, transaction?, session?, subscriber?, close }


declare function open(): Promise<RedisSubscriber>;
//    ^? { subscribe, unsubscribe, psubscribe?, punsubscribe?, closed, close }
```

Benni leases at most one subscriber per client, so adapters do no bookkeeping. When `subscriber` is undefined (the HTTP adapter), `subscribe` throws `TypeError` at call time, the same style as the session guard. `psubscribe`/`punsubscribe` are optional in turn, which is how the Bun adapter reports patterns as unsupported instead of hanging. Pass `onPubSubError` to `benni()` to route a handler that throws; without it the error is rethrown asynchronously rather than swallowed. See [Pub/Sub](/benni/data-structures/pubsub/).

## `redis.session`

[Section titled “redis.session”](#redissession)

Lease a dedicated connection for blocking commands and `WATCH` transactions. The scoped form closes the session when the callback settles; the bare form returns it and hands you the `close()` obligation (pair with `await using`):

```ts
const job = await redis.session(async (s) => {
  return s.list(jobs).blpop("pending", { timeoutSeconds: 5 });
});


await using session = await redis.session();
```

A session carries the same store accessors as the Benni handle, bound to its private connection, where `list`, `zset`, and `stream` are supersets that add the blocking variants and the blocking consumer-group read. It also adds `session.watch(keys)`, `session.unwatch()`, and `session.multi()`, plus `session.raw`, `session.closed`, and `session.close()`. It throws `TypeError` if the client does not support sessions. See [Sessions](/benni/advanced/sessions/), [Blocking Operations](/benni/advanced/blocking-operations/), and [Consumer Groups](/benni/data-structures/consumer-groups/).

## `redis.watch`

[Section titled “redis.watch”](#rediswatch)

Retrying optimistic transaction (`WATCH`/`MULTI`/`EXEC`), discoverable next to `redis.multi()`:

```ts
const result = await redis.watch(
  views.key("home"),
  async (s) => {
    const current = (await s.kv(views).get("home")) ?? 0;
    return s.multi().add(["SET", views.key("home"), String(current + 1)], okReply);
  },
  { attempts: 5, onAbort: ({ attempt }) => metrics.increment("cas.conflict", { attempt }) }
);
//    ^? [void] | null   (null = the body opted out)
```

Each attempt watches the keys, runs the body, and commits the transaction it returns; a conflict retries, a `null` body opts out, and exhausted attempts throw `WatchRetriesExceededError`. See [Optimistic Transactions](/benni/advanced/optimistic-transactions/).

## `redis.raw`

[Section titled “redis.raw”](#redisraw)

Direct Redis access:

```ts
await redis.raw.send(["PING"]);
await redis.raw.pipeline([
  ["SET", "a", "1"],
  ["GET", "a"]
]);
```

# Errors

> Every error type Benni throws, what makes it fire, and the structured properties it carries so you can branch on a field instead of matching message text.

Benni never casts a bad value and moves on. When something is wrong it throws, and it throws a named class carrying the structured detail you need to decide what to do next. This page is the full public error surface.

## Which Error Should I Catch?

[Section titled “Which Error Should I Catch?”](#which-error-should-i-catch)

Four questions cover almost every case:

| The failure is                                 | Catch                     | Where it comes from |
| ---------------------------------------------- | ------------------------- | ------------------- |
| I passed bad input, nothing was sent           | `ValidationError`         | `benni`             |
| Redis answered with an error reply             | `RedisServerError`        | `benni`             |
| A reply or stored value was the wrong shape    | `ReplyShapeError`         | `benni`             |
| A primitive could not give me what I asked for | the primitive’s own class | `benni/primitives`  |

The first three are the important distinction, and they are mutually exclusive:

* **`ValidationError`** means the command never left the process. Benni refused your arguments.
* **`RedisServerError`** means the command reached Redis and Redis refused it.
* **`ReplyShapeError`** means the command *succeeded* and the value that came back did not match what your schema declared.

Everything else is either a subclass of one of those or a primitive-specific outcome.

One class is deliberately absent: connection and transport failures. A closed socket, a DNS failure, an ioredis `MaxRetriesPerRequestError`, an Upstash HTTP 502, a non-JSON REST body. Benni passes those through from the underlying client untouched, because it has nothing to add and wrapping them would only hide what the client already told you.

## Core Errors

[Section titled “Core Errors”](#core-errors)

These are exported from the root `benni` entrypoint (and from `benni/core`), with one exception noted below: `CrossSlotError` lives in `benni/cluster`.

### `ValidationError`

[Section titled “ValidationError”](#validationerror)

Extends `TypeError`. Thrown when caller-supplied input fails validation **before any command is sent to Redis**: an out-of-range count, a non-finite number, a blocking timeout of `0`, a contradictory option combination.

It extends `TypeError` so existing `instanceof TypeError` handling keeps working. Catch `ValidationError` specifically to tell “I passed bad input” apart from a protocol-level failure.

Properties: none beyond `message`. A `ValidationError` is a programming mistake, so the message names the argument and the constraint.

### `ReplyShapeError`

[Section titled “ReplyShapeError”](#replyshapeerror)

Extends `TypeError`. Thrown when a Redis reply, or a stored value handed to a codec, does not match the shape a decoder expected.

| Property | Type      | Meaning                                                               |
| -------- | --------- | --------------------------------------------------------------------- |
| `reply`  | `unknown` | The raw value received, so you can inspect or log it programmatically |

Most messages read `Expected Redis <COMMAND> to return <expected>, got <actual>`. Two common sources:

* A `number()` or `json()` codec decoding a value some other writer stored in a shape the codec cannot read.
* A failing **`json(validator)`** read. When the stored JSON does not satisfy the validator, the read throws `ReplyShapeError` naming the validator vendor and the issues, with the offending value on `.reply`. This is the whole reason to prefer `json(validator)` over `json<T>()`: see [JSON values](/benni/data-structures/json-values/).

`ReplyShapeError` is not a server error. The command succeeded; the *data* was wrong.

### `PartialRecordError`

[Section titled “PartialRecordError”](#partialrecorderror)

Extends `ReplyShapeError`. Thrown when a whole-record hash read (`hget(id)` on a `hash()` schema) finds some, but not all, of the declared fields.

| Property  | Type                | Meaning                              |
| --------- | ------------------- | ------------------------------------ |
| `missing` | `readonly string[]` | The declared fields that were absent |
| `reply`   | `unknown`           | Inherited: the raw `HMGET` array     |

The reply is well formed here, so this is not a protocol or adapter fault. It means the stored record is incomplete, most often because individual fields were given their own TTLs with `hexpire` and some have since lapsed, or because `hdel` removed a declared field.

It extends `ReplyShapeError` so code that already catches that keeps working. Catch `PartialRecordError` specifically to tell an ordinary incomplete record apart from a genuine shape violation, and reach for `hgetall` (which types its result as `Partial`) when incompleteness is expected. See [Hashes](/benni/data-structures/hashes/).

### `UnsupportedCapabilityError`

[Section titled “UnsupportedCapabilityError”](#unsupportedcapabilityerror)

Extends `TypeError`. Thrown when the client behind the call does not implement the optional capability the call needed: `transaction` (MULTI/EXEC), `session` (a borrowed connection, for `WATCH` and blocking reads), or `subscriber` (Pub/Sub).

| Property     | Type                                         | Meaning                                 |
| ------------ | -------------------------------------------- | --------------------------------------- |
| `capability` | `"transaction" \| "session" \| "subscriber"` | Which one the client turned out to lack |

Every built-in adapter implements all three except `benni/upstash`, which is stateless HTTP and so has no `session` or `subscriber`. In practice you meet this class with a hand-written client, or when you `subscribe` over Upstash.

A connected client advertises its optional methods by having them defined, so Benni feature-detects and picks a fallback before calling. A client passed as a **promise or a factory** cannot be interrogated at bind time, so the facade over it defines all three and reports the gap from inside the call with this error instead. That is what the class is for: it keeps `benni(node({ url }))` and `benni(await node({ url }))` behaving identically on a client that is missing a capability, because a caller with a legitimate fallback can catch it and take that fallback either way.

The message is the same text a connected client’s own guard uses, and `TypeError` is still the base class, so `instanceof TypeError` and message matching that predate this class keep working.

What it never does is downgrade something silently. `redis.multi()` exists for MULTI/EXEC atomicity, so on a client without `transaction` it throws rather than quietly becoming a pipeline. Only a call site that is correct without the atomicity falls back, and `hset(id, value, { ttlSeconds })` is the one that does: it prefers `HSET` plus `EXPIRE` in a transaction and settles for a pipeline.

### `RedisServerError`

[Section titled “RedisServerError”](#redisservererror)

Extends `Error`. Thrown when the Redis **server** answered with an error reply: `WRONGTYPE` on a key holding another type, `NOSCRIPT`, `OOM`, `READONLY`, `NOAUTH`, or a Lua script’s own `redis.error_reply(...)`.

| Property  | Type                  | Meaning                                                                                    |
| --------- | --------------------- | ------------------------------------------------------------------------------------------ |
| `code`    | `string \| undefined` | The uppercase code the reply opens with, parsed for you                                    |
| `command` | `string \| undefined` | Uppercased name of the command that drew the error, when the throw site could attribute it |
| `cause`   | `unknown`             | The adapter-native error (or raw payload), set whenever there was one                      |

#### Why It Exists

[Section titled “Why It Exists”](#why-it-exists)

This is the one error type **every adapter normalizes to**. Before it, the same `WRONGTYPE` reached you as node-redis’s `SimpleError` on `benni/node`, ioredis’s `ReplyError` on `benni/ioredis`, Bun’s `RedisError` on `benni/bun`, and a bare `Error` built from the REST payload on `benni/upstash`: one taxonomy per runtime, so a `catch` block written against one adapter silently stopped matching after a move to another. That is the opposite of what one typed API across runtimes is supposed to buy.

Now `error instanceof RedisServerError` means the same thing on all four.

#### Branch On `.code`, Not On The Message

[Section titled “Branch On .code, Not On The Message”](#branch-on-code-not-on-the-message)

`code` is the parsed leading token of the reply: `WRONGTYPE`, `NOSCRIPT`, `NOAUTH`, `OOM`, `READONLY`, `MOVED`, `ASK`, `BUSYGROUP`, `EXECABORT`, and the rest of Redis’s vocabulary. Branch on it rather than matching substrings of prose that Redis is free to reword.

```ts
import { RedisServerError } from "benni";


try {
  await redis.query.leaderboard.zadd("global", { member: "ada", score: 1 });
} catch (error) {
  if (!(error instanceof RedisServerError)) throw error;


  switch (error.code) {
    case "WRONGTYPE":
      // that key holds a hash, not a sorted set: a schema or key-prefix bug
      throw new Error(`${error.command} hit the wrong type`, { cause: error });
    case "OOM":
    case "READONLY":
      // the server cannot accept writes right now; shed load and retry later
      return { retryable: true };
    case "NOAUTH":
      // credentials are wrong or missing: not worth retrying
      throw error;
    default:
      throw error;
  }
}
```

`code` is `undefined` when the server’s text carries no code, which in practice means a Lua script returned a bare `redis.error_reply("some text")`. Handle that in your `default` branch rather than assuming a code is always present.

`message` is the server’s text **verbatim, code first**, so message matching that predates this class keeps working. `cause` holds the adapter-native error, so nothing the underlying client attached is lost.

#### What Is Not A `RedisServerError`

[Section titled “What Is Not A RedisServerError”](#what-is-not-a-redisservererror)

Transport failures are deliberately excluded, because nothing about them came from Redis:

* an Upstash HTTP 502, or a body that is not JSON
* a closed socket, a connection reset, a client shut down mid-command
* an ioredis `MaxRetriesPerRequestError`, a node-redis `ClientClosedError`

Those reach you as whatever the underlying client threw. A `RedisServerError` always means Redis itself formed an answer and that answer was an error.

Cluster redirections (`MOVED`, `ASK`) are followed by the cluster-aware client underneath and normally never surface. One that does reach you is a real failure, and its code is parsed like any other.

### `redisErrorCode(message)`

[Section titled “redisErrorCode(message)”](#rediserrorcodemessage)

```ts
function redisErrorCode(message: string): string | undefined;
```

The classifier `RedisServerError` uses on its own message, exported so a caller holding a raw message (from a nested reply, a log line, a script’s own output) can classify it the same way. Returns the leading uppercase code, or `undefined` when there is none.

It requires at least three characters, on purpose: the shortest real codes are `ERR`, `OOM`, and `ASK`, and the three-character floor keeps a script’s `redis.error_reply("A bad thing")` from reporting `A` as an error code.

### `redisServerError(source, command?)`

[Section titled “redisServerError(source, command?)”](#redisservererrorsource-command)

```ts
function redisServerError(source: unknown, command?: string): RedisServerError;
```

The normalizer every built-in adapter runs a server error reply through. It preserves the message verbatim, keeps the original as `cause`, and passes an already-normalized `RedisServerError` through untouched, so re-wrapping on the way out of a nested call cannot double-wrap or break identity comparisons.

You only need this if you are **writing an adapter**. Deciding *whether* something is a server error reply stays with each adapter, which knows its client’s taxonomy; this function only does the conversion.

`RedisServerErrorOptions` is the exported shape of its second argument on the constructor: `{ command?: string; cause?: unknown }`.

### `SessionClosedError`

[Section titled “SessionClosedError”](#sessionclosederror)

Extends `Error`. Thrown by the session command gate for any use of a session after `close()`.

Properties: none. Prefer the boolean `session.closed` for worker loops: a flag is robust where cross-adapter error-class mapping is fragile, and in-flight rejections during a connection drop keep the adapter-native error rather than becoming this. See [Connection Sessions](/benni/advanced/sessions/).

### `WatchRetriesExceededError`

[Section titled “WatchRetriesExceededError”](#watchretriesexceedederror)

Extends `Error`. Thrown by `redis.watch()` when every attempt aborted because a `WATCH`ed key kept changing under it.

| Property   | Type     | Meaning                           |
| ---------- | -------- | --------------------------------- |
| `attempts` | `number` | The total number of attempts made |

Exhaustion is exceptional, so it throws rather than returning `null`, which keeps the happy path ceremony-free. See [Optimistic Transactions](/benni/advanced/optimistic-transactions/).

### `CrossSlotError`

[Section titled “CrossSlotError”](#crosssloterror)

Extends `ValidationError` (so `instanceof TypeError` still holds). Thrown when a command’s keys span two Redis Cluster hash slots, caught **before** the command is sent.

| Property  | Type                        | Meaning                                          |
| --------- | --------------------------- | ------------------------------------------------ |
| `command` | `string`                    | The command that would have been sent            |
| `keys`    | `readonly [string, string]` | The first key and the key that disagreed with it |
| `slots`   | `readonly [number, number]` | Their two slots                                  |

Exported from **`benni/cluster`**, not from the root entrypoint: naming it from the root would put the CRC16 table and the fix-hint prose into every bundle, including the ones that never enable the check. See [Redis Cluster](/benni/advanced/cluster/).

## Primitive Errors

[Section titled “Primitive Errors”](#primitive-errors)

Exported from `benni/primitives`.

### Lock

[Section titled “Lock”](#lock)

| Error                  | Properties | Thrown when                                                                                                                                            |
| ---------------------- | ---------- | ------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `LockNotAcquiredError` | `key`      | `lock().run()` could not acquire the lock. Acquisition is fail-fast by default (`retries: 0`), so a caller that finds the lock held throws immediately |
| `LockLeaseLostError`   | `key`      | The lock was lost while `fn` was still running: renewal found the key gone or owned by another token                                                   |

`run()` rejects with `LockLeaseLostError` **even when `fn` itself resolved**, because a body that completed without the lock did not complete under the guarantee it was written against. The same error is the abort reason on `handle.signal`. [Distributed Lock](/benni/primitives/lock/#when-the-lock-is-lost) covers the two-pronged detection and how to size `ttlMs` and `heartbeatMs`.

### Semaphore

[Section titled “Semaphore”](#semaphore)

| Error                       | Properties     | Thrown when                                                                                                    |
| --------------------------- | -------------- | -------------------------------------------------------------------------------------------------------------- |
| `SemaphoreNotAcquiredError` | `key`, `limit` | `semaphore().run()` found no slot free                                                                         |
| `SemaphoreLeaseLostError`   | `key`, `limit` | The slot was lost while `fn` was still running: renewal found the lease gone, so it had already been reclaimed |

A lost lock means two callers collided on one key; a lost slot means the semaphore **over-admits**, so a `limit: 20` pool guarding a provider quota quietly runs 21 in flight. As with the lock, `run()` rejects even when `fn` resolved, and the error is the abort reason on `held.signal`. [Semaphore](/benni/primitives/semaphore/#when-the-slot-is-lost) has the detail.

### Queue

[Section titled “Queue”](#queue)

| Error               | Properties | Thrown when                                                                                                                                                                                                                                                                                    |
| ------------------- | ---------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `JobNotFoundError`  | `jobId`    | A job id is not in Redis: it never existed, or it finished and its `resultTtlMs` elapsed                                                                                                                                                                                                       |
| `JobLeaseLostError` | `jobId`    | Thrown *inside a handler* when this worker no longer owns the job. Its lease expired and another worker reclaimed it, so `emit()`, `progress()`, and the automatic heartbeat all abort the job’s signal and throw this rather than let you burn tokens on a run whose result will be discarded |

Two more are errors **you throw**, from inside a handler, to steer the retry machinery:

| Error              | Signature                           | Effect                                                                                                                                                            |
| ------------------ | ----------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `TerminalJobError` | `(message, options?)`               | Fail the job immediately, no further attempts. For anything a retry would reproduce verbatim: a malformed request, a content-policy refusal, an unsupported model |
| `RetryJobError`    | `(message, retryAfterMs, options?)` | Retry after an explicit delay, overriding the configured backoff. Built for a provider’s `Retry-After` header                                                     |

`RetryJobError` carries `retryAfterMs` and rejects a non-finite value with a `ValidationError` from its own constructor. A `Retry-After` header parsed straight through can be `NaN`, Redis will not accept that as a sorted-set score, and by then the retry path has already dropped the lease, so the job would be stranded. Refusing it at construction leaves the worker free to fall back to ordinary backoff. See [AI Job Queue](/benni/primitives/queue/).

### Idempotency

[Section titled “Idempotency”](#idempotency)

| Error                            | Properties     | Thrown when                                                |
| -------------------------------- | -------------- | ---------------------------------------------------------- |
| `IdempotencyConflictError`       | `key`          | Another caller holds the key and `onConflict` is `"throw"` |
| `IdempotencyTimeoutError`        | `key`          | `onConflict: "wait"` gave up before the holder finished    |
| `IdempotencyNotRecordedError<T>` | `key`, `value` | The handler succeeded but its result could not be stored   |

`IdempotencyNotRecordedError` is the one that needs care. **The side effect happened.** What failed is the record of it, so the running marker will lapse and a later caller with the same key will run the handler again. Treat it as indeterminate rather than as a failure: `value` carries the result if you can still use it (return it to the client, write it somewhere durable), but do not assume a retry is safe. The usual causes are a codec that cannot encode the result, or a Redis blip between finishing the work and recording it. The underlying failure is on `cause`. See [Idempotency](/benni/primitives/idempotency/).

### Budget

[Section titled “Budget”](#budget)

| Error                     | Properties | Thrown when                                |
| ------------------------- | ---------- | ------------------------------------------ |
| `BudgetWindowRolledError` | `id`       | The window rolled over under every attempt |

That takes a process stalled for longer than `windowMs` between building the keys and the script running. Nothing was applied, so the call is safe to retry, and a hold whose `settle` throws this is still usable. See [Budget](/benni/primitives/budget/).

## Inheritance At A Glance

[Section titled “Inheritance At A Glance”](#inheritance-at-a-glance)

```text
TypeError
├── ValidationError
│   └── CrossSlotError
├── UnsupportedCapabilityError
└── ReplyShapeError
    └── PartialRecordError


Error
├── RedisServerError
├── SessionClosedError
├── WatchRetriesExceededError
├── LockNotAcquiredError / LockLeaseLostError
├── SemaphoreNotAcquiredError / SemaphoreLeaseLostError
├── JobNotFoundError / JobLeaseLostError / TerminalJobError / RetryJobError
├── IdempotencyConflictError / IdempotencyTimeoutError / IdempotencyNotRecordedError
└── BudgetWindowRolledError
```

Order your `catch` branches most specific first: `PartialRecordError` before `ReplyShapeError`, `CrossSlotError` before `ValidationError`.

## See Also

[Section titled “See Also”](#see-also)

* [JSON values](/benni/data-structures/json-values/) for why `json(validator)` throws where `json<T>()` stays quiet
* [Hashes](/benni/data-structures/hashes/#missing-declared-fields-throw) for `PartialRecordError` and the tolerant `hgetall` read
* [Optimistic Transactions](/benni/advanced/optimistic-transactions/) for `WatchRetriesExceededError` and the retry loop around it
* [Redis Cluster](/benni/advanced/cluster/) for `CrossSlotError` and the slot guard
* [Philosophy](/benni/getting-started/philosophy/) for the “nothing is silent” rule these classes implement

# API Overview

> The three imports most applications need, plus where the lower-level core builders live.

Most applications use three imports:

```ts
import { benni } from "benni";
import { node } from "benni/node";
import { hash, hll, json, kv, number, string, zset } from "benni/schema";
```

## Schema API

[Section titled “Schema API”](#schema-api)

Use `benni/schema` to define Redis key families:

```ts
kv("profile", json<Profile>());
hash("user", { name: string(), score: number() });
set("team-members", string());
list("events", json<Event>());
zset("leaderboard", string());
hll("page-views", string());
channel("events:user", json<UserEvent>());
```

## Client API

[Section titled “Client API”](#client-api)

Bind a Redis client:

```ts
const redis = benni(client, { schema });
```

Use data-structure resources:

```ts
redis.kv(profiles);
redis.hash(users);
redis.set(teamMembers);
redis.list(events);
redis.zset(leaderboards);
redis.hll(pageViews);
redis.pubsub.channel(userEvents);
```

Use `redis.raw` for direct Redis commands:

```ts
await redis.raw.send(["PING"]);
```

## Lower-Level Core API

[Section titled “Lower-Level Core API”](#lower-level-core-api)

The `benni/core` entrypoint exposes the building blocks the client is made of (`defineKeyspace`, `createKeyValueStore`, `createHashStore`, and the other store builders) for adapter authors and advanced integrations.

Application code should prefer the schema-first API shown in the guide; every accessor is documented in the [Benni Client reference](/benni/api/benni-client/).

# Schema Builders

> Schema builders are exported from benni/schema.

Schema builders are exported from `benni/schema`.

## Codecs

[Section titled “Codecs”](#codecs)

```ts
string();
number();
boolean();
json<T>();
json(validator);
bytes();
enumOf(["pending", "active", "done"]);
```

A codec controls how Benni writes values to Redis and decodes values returned by Redis. `bytes()` stores `Uint8Array` values as base64-encoded strings in Redis. `enumOf([...])` constrains a field to a fixed set of string literals, stored as the plain string (no JSON overhead) and validated on decode, inferring the union of the values (`"pending" | "active" | "done"`).

`json` has two forms, and the validating one is the default to reach for. `json(validator)` accepts any [Standard Schema](https://standardschema.dev) validator (Zod, Valibot, ArkType, …): every read is validated at runtime, and the value type is inferred from the validator, with no type parameter needed. Benni stays zero-dependency; the Standard Schema interface is inlined.

`json<T>()` is the escape hatch: a pure cast, with no runtime validation at all. `JSON.parse` runs and its result is asserted to be `T`. A stored value missing required fields, or carrying a field of the wrong type, is handed back typed as a complete `T` and nothing throws. Use it only where you own every writer of the key.

```ts
import { z } from "zod";


const users = kv("user", json(z.object({ name: z.string() })));     // validated
//    reads infer { name: string } | null from the Zod schema
const profiles = kv("profile", json<Profile>());                    // cast, unchecked
```

Invalid stored data throws a `ReplyShapeError` naming the validation issues. Async validators (schemas with async refinements) throw a clear error; `json(validator)` requires a synchronous validator.

Standard Schema defines only the read direction, so `json(validator)` cannot validate writes. The optional [`benni/zod`](/benni/integrations/zod/) subpath runs [Zod codecs](https://zod.dev/codecs) in both directions: `zodCodec(schema)` for string-stored fields (rich types like `Date` that round-trip) and `zodJson(schema)` as a write-validating `json(validator)`.

`number()` rejects non-finite input (`NaN`/`Infinity`) at write time, so a bad value fails at the `set` rather than poisoning a later `get`. Decode failures (a malformed stored value, an out-of-set enum, a wrong reply shape) throw a `ReplyShapeError` (which carries the offending `.reply`); invalid caller input throws a `ValidationError`. Both extend `TypeError`, so existing `catch` blocks keep working while you can now discriminate the two.

When your app needs a codec Benni does not ship, pass a plain `Codec` object, anything with `encode`/`decode`:

```ts
import type { Codec } from "benni";


const uppercase: Codec<string, string> = {
  encode(value) {
    return value.toUpperCase();
  },
  decode(stored) {
    return stored;
  }
};
```

## Key-Value

[Section titled “Key-Value”](#key-value)

```ts
const profiles = kv("profile", json<Profile>());
```

Use with:

```ts
redis.kv(profiles);
```

## Hash

[Section titled “Hash”](#hash)

```ts
const users = hash("user", {
  name: string(),
  score: number()
});
```

Use with:

```ts
redis.hash(users);
```

## Collections

[Section titled “Collections”](#collections)

```ts
const tags = set("tags", string());
const events = list("events", json<Event>());
const leaderboard = zset("leaderboard", string());
const pageViews = hll("page-views", string());
```

Use with:

```ts
redis.set(tags);
redis.list(events);
redis.zset(leaderboard);
redis.hll(pageViews);
```

## Stream

[Section titled “Stream”](#stream)

```ts
const activity = stream("activity", {
  action: string(),
  points: number()
});
```

Use with:

```ts
redis.stream(activity);
```

## Bitmap

[Section titled “Bitmap”](#bitmap)

```ts
const dailyActive = bitmap("daily-active");
```

Bitmaps take no codec; bits are addressed by offset and exposed as booleans. Use with:

```ts
redis.bitmap(dailyActive);
```

## Geo

[Section titled “Geo”](#geo)

```ts
const stores = geo("stores", string());
```

Use with:

```ts
redis.geo(stores);
```

## Script

[Section titled “Script”](#script)

```ts
const rateLimit = script("rate-limit", {
  keys: ["counter"],
  args: { limit: number(), windowSeconds: number() },
  returns: number(),
  lua: `return redis.call("INCR", KEYS[1])`
});
```

Use with:

```ts
redis.script(rateLimit);
```

## Pub/Sub

[Section titled “Pub/Sub”](#pubsub)

```ts
const userEvents = channel("events:user", json<UserEvent>());
const userEventPattern = pattern("events:user:*", json<UserEvent>());
```

Use with:

```ts
redis.pubsub.channel(userEvents);
redis.pubsub.pattern(userEventPattern);
```

# Benni vs ioredis

> An honest comparison of Benni and ioredis for TypeScript projects: what each one types, which features only ioredis has, and how to choose.

Short version: **ioredis is a Redis client. Benni is a typed layer over one.** They are not competing for the same job, and in fact [`benni/ioredis`](/benni/runtime/ioredis/) runs the whole typed API *on* ioredis, including an instance you already have. So the real question is not “which one” but “do I want my data typed on top of the client I already run”.

## Choose by what you need

[Section titled “Choose by what you need”](#choose-by-what-you-need)

**Reach for plain ioredis when** you want nothing between you and the commands, or you need an ecosystem package built specifically on it (BullMQ being the big one), or you rely on transport features Benni does not model, such as Sentinel failover and sharded Pub/Sub.

**Reach for Benni when** your application stores values with a shape you care about, and you want the declared type to survive the round trip, plus one API that runs unchanged on Node, Bun, Deno, and edge/serverless runtimes.

**You do not have to choose the transport.** Benni’s ioredis adapter means adopting Benni is not a client migration: keep ioredis, keep your connection, and add types on top.

## The core difference: command types vs data types

[Section titled “The core difference: command types vs data types”](#the-core-difference-command-types-vs-data-types)

ioredis is fully typed, but only at the *command surface*. The type describes Redis’s wire shape, not your value:

```ts
// plain ioredis: typed, as a string
const raw = await ioredisClient.get("profile:42"); // string | null
const profile = JSON.parse(raw!) as Profile; // hand-cast; the compiler never checked it


const h = await ioredisClient.hgetall("user:42"); // Record<string, string>
const score = Number(h.score); // coerce by hand, every time
```

The type is gone the moment your data crosses the Redis edge, and the cast you write to get it back is unchecked. Benni moves the declaration up front:

```ts
const users = hash("user", { name: string(), score: number() });


const user = await redis.query.users.hget("42");
//    ^? { name: string; score: number } | null
```

`score` is a `number` because the schema says so, with no coercion and no cast. Keys are derived from the schema too (`redis.query.users.key("42")` is `"user:42"`), so prefix strings stop being scattered across the codebase.

## What ioredis has that Benni does not

[Section titled “What ioredis has that Benni does not”](#what-ioredis-has-that-benni-does-not)

Stated plainly, because these are real reasons to keep reaching for ioredis directly. Note that the first two are about *transport*, so [`benni/ioredis`](/benni/runtime/ioredis/) lets you keep them and still get typed data on top.

* **Redis Cluster routing.** ioredis has `Redis.Cluster` with slot awareness, `MOVED`/`ASK` handling, and multi-node routing. Benni has no transport of its own and does not route. What it does add is slot *co-location*: schemas declare where their hash tag goes, and cross-slot commands are caught at compile time and (opt-in) before they are sent. See [Redis Cluster](/benni/advanced/cluster/).
* **Sentinel.** High-availability failover via Sentinel is ioredis’s, not Benni’s, though an adopted ioredis client keeps it.
* **Sharded Pub/Sub.** `SSUBSCRIBE`/`SPUBLISH` are not modelled by Benni’s typed Pub/Sub.
* **Ecosystem lock-in, in the good sense.** BullMQ, `ioredis-mock`, rate-limiter libraries, and many framework session stores expect an ioredis instance.
* **Maturity.** ioredis has been the default for years, with the battle-testing that implies.

## What Benni has that ioredis does not

[Section titled “What Benni has that ioredis does not”](#what-benni-has-that-ioredis-does-not)

* **End-to-end typed data**, as above.
* **One API across runtimes.** The same typed calls run on Node and Bun over TCP, and on Cloudflare Workers / Vercel Edge over HTTP. Only the adapter import changes. Practically, this means local development against real Redis over TCP and production over HTTP from the same code.
* **Schema-derived keys and TTLs** instead of hand-built strings.
* **Correct primitives included.** A distributed lock that never frees a lock that expired and was re-acquired, a sliding-window rate limiter in one atomic round trip, a stampede-proof read-through cache, and an AI-shaped job queue with resumable output streams.
* **Validator integration.** `json(zodSchema)` gives runtime-validated, inferred reads through any [Standard Schema](https://standardschema.dev) validator (Zod, Valibot, ArkType) with no added dependency.

## Side by side

[Section titled “Side by side”](#side-by-side)

|                                               | ioredis                        | Benni                                      |
| --------------------------------------------- | ------------------------------ | ------------------------------------------ |
| Raw command access                            | Yes                            | Yes, via `redis.raw.send([...])`           |
| Command-level types                           | Yes                            | Yes                                        |
| Your data’s types survive a read              | No, you cast                   | Yes                                        |
| Schema-derived keys                           | Manual                         | Yes                                        |
| Runtimes                                      | Node (and Bun/Deno via compat) | Node, Bun, Deno, edge/serverless           |
| Redis Cluster routing                         | Yes                            | No, but slot co-location is typed          |
| Sentinel                                      | Yes                            | Via an adopted ioredis client              |
| Sharded Pub/Sub                               | Yes                            | No                                         |
| Distributed lock / rate limit / cache / queue | Via ecosystem packages         | Built in                                   |
| Transport                                     | Its own                        | ioredis, node-redis, Bun’s client, or HTTP |

## Migration cost, honestly

[Section titled “Migration cost, honestly”](#migration-cost-honestly)

There is no client migration. `benni/ioredis` takes a URL, ioredis options, or an ioredis instance you already built and tuned:

```ts
import Redis from "ioredis";
import { benni } from "benni";
import { ioredis } from "benni/ioredis";
import * as schema from "./schema";


const existing = new Redis(process.env.REDIS_URL ?? "redis://127.0.0.1:6379");


const client = await ioredis(existing);
export const redis = benni(client, { schema });
```

An adopted client is borrowed, not taken over: `client.close()` reaps only the sessions and subscriber connections Benni leased from it, and leaves your client open. See [the ioredis adapter](/benni/runtime/ioredis/) for the details.

That also means ioredis-dependent packages in the same process (BullMQ, for instance) keep working on the very same connection. Nothing about Benni’s schemas prevents another client from reading or writing the same keys: schemas describe keys, they do not own them.

## Adopting incrementally

[Section titled “Adopting incrementally”](#adopting-incrementally)

Benni does not require a migration. Schemas are plain values that create no keys and run no migrations, so you can declare one key family, use it, and leave the rest of your Redis access exactly as it is. `redis.raw.send([...])` is always available for commands you have not modelled.

# Benni vs @upstash/redis

> An honest comparison of Benni and @upstash/redis for edge and serverless TypeScript: where they overlap, what each types, and when vendor independence matters.

These two overlap more than [Benni and ioredis](/benni/comparisons/ioredis/) do: both run on edge and serverless runtimes, and both can talk to Upstash over HTTP. In fact Benni’s edge adapter speaks the **same Upstash REST protocol**: `benni/upstash` is a client for it, not an alternative to it.

The real difference is what gets typed, and whether your application code is tied to one transport.

## Choose by what you need

[Section titled “Choose by what you need”](#choose-by-what-you-need)

**Reach for `@upstash/redis` when** you are all-in on Upstash, want the officially supported client, or depend on its ecosystem (`@upstash/ratelimit`, `@upstash/vector`, and the Upstash-specific conveniences).

**Reach for Benni when** you want your declared types to survive the round trip, and you want the same code to run over TCP in development and HTTP in production without a rewrite.

## Types you assert vs types you declare

[Section titled “Types you assert vs types you declare”](#types-you-assert-vs-types-you-declare)

`@upstash/redis` is typed, and it lets you pass a type parameter on a read:

```ts
const profile = await redis.get<Profile>("profile:42");
//    ^? Profile | null
```

That is a genuine convenience, but the type is an **assertion at the call site**, not a derivation. Nothing checks that the value written to `profile:42` was ever a `Profile`, nothing stops a different call site from asserting a different type for the same key, and there is no single place that says what lives there. It is a tidier cast.

Benni inverts it. You declare the key family once, and both directions are checked against that declaration:

```ts
// schema.ts: the single source of truth
export const profiles = kv("profile", json<Profile>());


// writes are checked against the schema…
await redis.query.profiles.set("42", profile);


// …and reads derive their type from it
const loaded = await redis.query.profiles.get("42");
//    ^? Profile | null
```

Swap `json<Profile>()` for `json(profileZodSchema)` and reads are validated at runtime too, through any [Standard Schema](https://standardschema.dev) validator (Zod, Valibot, ArkType) with no extra dependency. An asserted generic cannot do that, because there is nothing to validate against.

## One API, both transports

[Section titled “One API, both transports”](#one-api-both-transports)

This is the practical reason to prefer Benni even on Upstash. `@upstash/redis` is an HTTP client, so the stateless surface is all you get. Benni’s typed API is identical across adapters:

```ts
// development: real Redis over TCP
import { node } from "benni/node";
const client = await node({ url: process.env.REDIS_URL });


// production: Upstash over HTTP
import { upstash } from "benni/upstash";
const client = upstash({
  url: process.env.UPSTASH_REDIS_REST_URL as string,
  token: process.env.UPSTASH_REDIS_REST_TOKEN as string
});
```

Everything after that line, every schema, every query, every primitive, is unchanged. You can develop against a local Redis in Docker, deploy to the edge, and move a workload back to a long-running Node process later without touching application code.

It also means you are not locked to one vendor. `benni/upstash` works against any Upstash-REST-compatible server, including self-hosted [`serverless-redis-http`](https://github.com/hiett/serverless-redis-http) in front of your own Redis.

## What Benni does not do

[Section titled “What Benni does not do”](#what-benni-does-not-do)

* **No Pub/Sub subscribing over HTTP.** This is a protocol limit, not a Benni choice: subscribing needs a persistent connection, so it requires `benni/node` or `benni/bun`. Publishing is a single stateless `PUBLISH` and works fine on the edge.
* **No blocking commands, sessions, or `WATCH` transactions over HTTP**, for the same reason. Pipelines and atomic `MULTI`/`EXEC` do work: they map onto the REST `/pipeline` and `/multi-exec` endpoints.
* **No binary command arguments over REST.** Use the `bytes()` codec, which stores base64 strings, or a TCP adapter.
* **Not officially supported by Upstash.** If you need a vendor support channel for client bugs, the official client is the safer pick.
* **No Upstash-specific extras.** Benni models Redis, not the Upstash platform.

## Side by side

[Section titled “Side by side”](#side-by-side)

|                                   | `@upstash/redis`                 | Benni                                  |
| --------------------------------- | -------------------------------- | -------------------------------------- |
| Transport                         | HTTP/REST                        | TCP *and* HTTP, same API               |
| Runs on edge/serverless           | Yes                              | Yes, via `benni/upstash`               |
| Command-level types               | Yes                              | Yes                                    |
| Read types                        | Asserted per call (`get<T>`)     | Derived from one schema                |
| Write types checked               | No                               | Yes                                    |
| Runtime validation                | No                               | Yes, via any Standard Schema validator |
| Schema-derived keys               | Manual                           | Yes                                    |
| Vendor independence               | Upstash                          | Any Upstash-REST-compatible server     |
| Rate limit / lock / cache / queue | `@upstash/ratelimit` and friends | Built into `benni/primitives`          |
| Officially supported by Upstash   | Yes                              | No                                     |
| Dependencies                      | Zero                             | Zero on the edge adapter               |

## Using both

[Section titled “Using both”](#using-both)

Nothing stops you. Schemas describe keys, they do not own them, so `@upstash/ratelimit` and Benni can share the same database without interfering. `redis.raw.send([...])` is also always available if you want to issue a command Benni has not modelled rather than reaching for a second client.

# Defining Schemas

> A Benni schema describes one Redis key family.

A Benni schema describes one Redis key family.

```ts
import { z } from "zod";
import { hash, json, kv, number, string } from "benni/schema";


export const users = hash("user", {
  name: string(),
  score: number()
});


export const profiles = kv(
  "profile",
  json(z.object({ name: z.string(), score: z.number() }))
);
```

`json(validator)` takes any [Standard Schema](https://standardschema.dev) validator (Zod, Valibot, ArkType, …), infers the value type from it, and validates every read. Prefer it. The bare-type form, `json<{ name: string; score: number }>()`, is a cast with no runtime check: a stored value missing a required field still reads back typed as if it were complete. Reach for it only when you own every writer of that key and accept that. See [JSON values](/benni/data-structures/json-values/).

The `users` schema describes keys like:

```txt
user:42
user:123
user:ada
```

It also gives you typed access to those keys:

```ts
await redis.hash(users).hset("42", { name: "Ada", score: 10 });
const user = await redis.hash(users).hgetall("42");
```

Schemas are not database schemas in the migration sense. They are plain TypeScript values.

* They do not create Redis keys.
* They do not require migrations.
* They do not block raw Redis access.
* They can live next to the application code that owns the data.

## Builders

[Section titled “Builders”](#builders)

Use schema builders from `benni/schema`:

```ts
import {
  boolean,
  channel,
  hash,
  json,
  kv,
  list,
  number,
  pattern,
  set,
  zset,
  string
} from "benni/schema";
```

When your app has a codec Benni does not ship, pass a plain `Codec` object, anything with `encode`/`decode`:

```ts
import type { Codec } from "benni";


const dateString: Codec<Date, Date> = {
  encode(value) {
    return value.toISOString();
  },
  decode(stored) {
    return new Date(stored);
  }
};
```

# Keys And Prefixes

> Every Benni schema has a prefix. Benni combines the prefix with an id to produce a Redis key.

Every Benni schema has a prefix. Benni combines the prefix with an id to produce a Redis key.

```ts
export const users = hash("user", {
  name: string(),
  score: number()
});


const key = redis.hash(users).key("42");
// "user:42"
```

Use prefixes as stable names for Redis key families:

```ts
kv("session", json<Session>());
hash("user", { name: string(), score: number() });
zset("leaderboard", string());
```

If an id comes from a route, database row, token, or Redis itself, pass it as a normal `string`, `number`, or `bigint`.

```ts
await redis.hash(users).hset(userId, {
  name: "Ada",
  score: 10
});
```

When ids are known at compile time, pass them to the schema for editor autocomplete:

```ts
export const demo = kv("demo", string(), {
  ids: ["test1", "test2"]
});


demo.key("test1");
// "demo:test1"
```

Prefixes should describe the data family, not the Redis command used to store it. Prefer `user`, `session`, and `feature-flag` over names like `hash-user`.

## Hash Tags

[Section titled “Hash Tags”](#hash-tags)

That same options bag takes `hashTag`, which moves braces into the key so Redis Cluster routes it deliberately rather than by accident:

```ts
kv("profile", string()); //                         "profile:42"
kv("profile", string(), { hashTag: "prefix" }); //  "{profile}:42"
kv("cart", string(), { hashTag: "id" }); //         "cart:{42}"
```

A cluster hashes only the text between the first `{` and the first `}`, so `"prefix"` pins a whole keyspace to one slot and `"id"` co-locates the same id across every schema tagged that way. On a single-node Redis it changes nothing but the key text. See [Redis Cluster](/benni/advanced/cluster/) for how to choose, and for the compile-time and runtime checks that come with it.

# Raw Redis Access

> Benni does not try to hide Redis or cover every command with a typed abstraction.

Benni does not try to hide Redis or cover every command with a typed abstraction.

For commands that are not typed yet, advanced Redis usage, debugging, or one-off operations, use the underlying Redis client directly:

```ts
await redis.raw.send(["PING"]);
await redis.raw.send(["SET", "custom:key", "value"]);
await redis.raw.send(["ZADD", "custom:leaderboard", 10, "user:42"]);
```

Use typed schemas where they reduce repeated app code:

```ts
await redis.kv(profiles).set("42", profile, {
  ttlSeconds: 3600
});
```

Use raw Redis where Redis itself is the clearest API:

```ts
await redis.raw.send(["CLIENT", "INFO"]);
```

Typed Benni keys are useful even when you drop down to raw Redis:

```ts
const key = redis.hash(users).key("42");
await redis.raw.send(["EXISTS", key]);
```

The raw client accepts Redis command arguments as strings, numbers, bigints, and byte arrays. Replies are returned in the adapter’s Redis reply shape.

# Schema Registry

> Declare schemas once, bind the module, and reach every store by name through redis.query.

Declare schemas once, bind the module, and reach every store by name through `redis.query`.

schema.ts

```ts
import { hash, kv, zset, json, number, string } from "benni/schema";


export const users = hash("user", {
  name: string(),
  score: number()
});


export const profiles = kv("profile", json<{ tier: string }>());
export const leaderboard = zset("leaderboard", string());
```

Bind the module once when you create the client:

redis.ts

```ts
import { benni } from "benni";
import { node } from "benni/node";
import * as schema from "./schema";


export const redis = benni(await node(), { schema });
```

Then reach each store by its export name, with full inference:

app.ts

```ts
import { redis } from "./redis";


await redis.query.users.hset("42", { name: "Ada", score: 10 });


const user = await redis.query.users.hgetall("42");
//    ^? { name: string; score: number } | null


await redis.query.leaderboard.zadd("daily", [{ member: "ada", score: 100 }]);
```

## How It Works

[Section titled “How It Works”](#how-it-works)

Each schema builder stamps a `kind` discriminant: one of the twelve data kinds (`kv`, `hash`, `set`, `list`, `zset`, `stream`, `bitmap`, `geo`, `hll`, `channel`, `pattern`, `script`) or one of the seven primitives (`cache`, `ratelimit`, `queue`, `lock`, `semaphore`, `idempotency`, `budget`). `redis.query.<name>` dispatches on that `kind` and resolves each schema to exactly the store `redis.<kind>(schema)` would return: same methods, same inference.

* Entries in the bound module that are not schemas (a re-exported type, a helper function, a Zod or Valibot validator you pass to `json()`) are dropped from the registry. Being a schema means carrying the store binding a builder attaches, not merely having a `kind` property of your own.
* `redis.query` is `{}` when no `{ schema }` is bound.
* Counter and string operations are the exception to “prefer `redis.query`”, because they are not kinds. See [The Counter And String Exception](#the-counter-and-string-exception) below before you write your first `incr`.

## Kind To Resource

[Section titled “Kind To Resource”](#kind-to-resource)

| `kind`        | Resource                       | Access                                              |
| ------------- | ------------------------------ | --------------------------------------------------- |
| `kv`          | `redis.kv(schema)`             | `redis.query.<name>.get` / `.set`                   |
| `hash`        | `redis.hash(schema)`           | `redis.query.<name>.hget` / `.hset`                 |
| `set`         | `redis.set(schema)`            | `redis.query.<name>.sadd` / `.smembers`             |
| `list`        | `redis.list(schema)`           | `redis.query.<name>.rpush` / `.lrange`              |
| `zset`        | `redis.zset(schema)`           | `redis.query.<name>.zadd` / `.zrange`               |
| `stream`      | `redis.stream(schema)`         | `redis.query.<name>.xadd` / `.group`                |
| `bitmap`      | `redis.bitmap(schema)`         | `redis.query.<name>.setbit` / `.bitcount`           |
| `geo`         | `redis.geo(schema)`            | `redis.query.<name>.geoadd` / `.geosearch`          |
| `hll`         | `redis.hll(schema)`            | `redis.query.<name>.pfadd` / `.pfcount`             |
| `channel`     | `redis.pubsub.channel(schema)` | `redis.query.<name>.publish` / `.subscribe`         |
| `pattern`     | `redis.pubsub.pattern(schema)` | `redis.query.<name>.subscribe`                      |
| `script`      | `redis.script(schema)`         | `redis.query.<name>.run({ keys, args })`            |
| `cache`       | `cache(client, options)`       | `redis.query.<name>.get` / `.peek` / `.del`         |
| `ratelimit`   | `ratelimit(client, options)`   | `redis.query.<name>.check`                          |
| `queue`       | `queue(client, options)`       | `redis.query.<name>.enqueue` / `.worker` / `.watch` |
| `lock`        | `lock(client, options)`        | `redis.query.<name>.run` / `.acquire`               |
| `semaphore`   | `semaphore(client, options)`   | `redis.query.<name>.run` / `.acquire`               |
| `idempotency` | `idempotency(client, options)` | `redis.query.<name>.run`                            |
| `budget`      | `budget(client, options)`      | `redis.query.<name>.reserve` / `.spend`             |

## Primitives In The Registry

[Section titled “Primitives In The Registry”](#primitives-in-the-registry)

The [primitives](/benni/primitives/cache/) declare themselves the same way the data structures do, so a cache or a queue is a schema value that lands in `redis.query` and carries its own configuration. Import them from `benni/schema` and they sit next to the stores they belong beside:

schema.ts

```ts
import {
  budget, cache, hash, json, number, queue, ratelimit, string
} from "benni/schema";
import { z } from "zod";


export const users = hash("user", { name: string(), score: number() });


const profile = z.object({ name: z.string(), score: z.number() });
export const profiles = cache("profile", { ttlMs: 60_000, codec: json(profile) });
export const apiLimit = ratelimit("api", { limit: 10, windowMs: 60_000 });
export const generate = queue<{ prompt: string }, string>("generate");
export const tokens = budget("tokens", { limit: 1_000_000, windowMs: 86_400_000 });
```

app.ts

```ts
const profile = await redis.query.profiles.get(userId, () => db.load(userId));
const { success } = await redis.query.apiLimit.check(userId);
const { id } = await redis.query.generate.enqueue({ prompt });
```

The first argument is the key prefix, exactly as it is for `hash` or `kv`; everything else is the same options bag the client-taking form takes. Nothing is imported that you do not declare: each schema carries its own store binding, so a bundle only pulls in the primitives that appear in the module.

`benni/primitives` keeps the client-taking form (`cache(client, options)`) for code that holds a client but no handle, such as a middleware factory. The two produce the same store over the same keys.

## The Counter And String Exception

[Section titled “The Counter And String Exception”](#the-counter-and-string-exception)

`redis.query` covers those kinds and nothing else, and there is one gap worth knowing before you meet it. Counters and strings are not kinds of their own: they are alternate views over a plain `kv` keyspace, so a `kv` schema always resolves to the `kv` resource in the registry, whatever its codec.

That means `redis.query.<name>` gives you `get` / `set` / `del` but **no `incr`**, even when the schema is a `kv(prefix, number())` that exists only to be incremented:

schema.ts

```ts
export const clicks = kv("clicks", number());
```

app.ts

```ts
await redis.query.clicks.set("home", 0); // fine: the kv resource
await redis.query.clicks.incr("home");   // does not compile: kv has no incr


const total = await redis.counter(clicks).incr("home"); // reach for the counter view
```

The same holds for the string view: `append`, `getrange`, `strlen`, and friends live on `redis.string(schema)`, not on `redis.query.<name>`.

So the “prefer `redis.query`” rule has exactly two exceptions, and they are both on `kv`:

| Want                                              | Use                                      |
| ------------------------------------------------- | ---------------------------------------- |
| `get`, `set`, `del`, `expire`, …                  | `redis.query.<name>` (the `kv` resource) |
| `incr`, `incrby`, `incrbyfloat`, `decr`, `decrby` | `redis.counter(schema)`                  |
| `append`, `getrange`, `setrange`, `strlen`        | `redis.string(schema)`                   |

Both accessors take the schema value, so a counter-heavy module tends to import its schemas directly rather than going through the registry for those calls. They read and write the same keys as the `kv` resource, so mixing them on one schema is normal: `redis.query.clicks.set("home", 0)` to seed and `redis.counter(clicks).incr("home")` to bump.

## Relationship To Explicit Accessors

[Section titled “Relationship To Explicit Accessors”](#relationship-to-explicit-accessors)

The registry is sugar over the explicit accessors. `redis.query.users` returns the same resource as `redis.hash(users)`, so you can mix the two styles freely:

```ts
// These are equivalent
await redis.query.users.hset("42", { name: "Ada", score: 10 });
await redis.hash(users).hset("42", { name: "Ada", score: 10 });
```

The explicit `redis.kv(schema)` / `redis.hash(schema)` accessors still exist unchanged. Use them when a schema is not part of a bound module, or when you prefer passing the schema value directly.

## A Multi-Kind Module

[Section titled “A Multi-Kind Module”](#a-multi-kind-module)

A single schema module can mix every kind. Each export becomes a registry entry:

schema.ts

```ts
import {
  hash, kv, zset, channel, script,
  json, number, string
} from "benni/schema";


export type UserEvent = { id: string; action: string };


export const users = hash("user", { name: string(), score: number() });
export const profiles = kv("profile", json<{ tier: string }>());
export const leaderboard = zset("leaderboard", string());
export const userEvents = channel("events:user", json<UserEvent>());
export const rateLimit = script("rate-limit", {
  keys: ["counter"],
  args: { limit: number() },
  returns: number(),
  lua: `return redis.call("INCR", KEYS[1])`
});
```

app.ts

```ts
await redis.query.profiles.set("42", { tier: "pro" }, { ttlSeconds: 3600 });
await redis.query.leaderboard.zadd("daily", [{ member: "ada", score: 100 }]);
await redis.query.userEvents.publish({ id: "42", action: "created" });


const allowed = await redis.query.rateLimit.run({
  keys: { counter: "user:42" },
  args: { limit: 100 }
});
//    ^? number
```

The `UserEvent` type export is dropped from the registry; only the schemas resolve to stores.

# TTL And Expiration

> Use ttl when a value should expire automatically.

Use `ttl` when a value should expire automatically.

```ts
await redis.kv(sessions).set(
  sessionId,
  {
    userId: "42",
    createdAt: new Date().toISOString()
  },
  { ttlSeconds: 60 * 60 * 24 * 7 }
);
```

`ttl` is measured in seconds and maps to Redis expiration commands.

For hashes, Benni writes the fields and then applies expiration to the Redis key:

```ts
await redis.hash(users).hset(
  "42",
  { name: "Ada", score: 10 },
  { ttlSeconds: 60 * 60 }
);
```

Every keyed store (kv, hash, set, list, sorted set, stream, geo, bitmap, HyperLogLog, string, and counter) exposes the same key-level lifecycle helpers: `exists`, `ttl`, `expire`, and `persist`:

```ts
const ttl = await redis.kv(sessions).ttl(sessionId);


await redis.kv(sessions).expire(sessionId, 60 * 15);
await redis.kv(sessions).persist(sessionId);
```

So a hash can read back the TTL it set via `hset`:

```ts
await redis.hash(users).hset(
  "42",
  { name: "Ada", score: 10 },
  { ttlSeconds: 60 * 60 }
);


const remaining = await redis.hash(users).ttl("42"); // > 0
await redis.hash(users).expire("42", 60 * 60 * 24); // extend
await redis.hash(users).persist("42"); // clear the TTL
await redis.hash(users).exists("42"); // true
```

`ttl` returns the remaining seconds, `-1` for a key with no expiry, and `-2` for a missing key.

Use `nx` when setting a key only if it does not already exist:

```ts
await redis.kv(sessions).set(sessionId, nextSession, {
  nx: true,
  ttlSeconds: 60 * 60
});
```

Use `xx` when replacing a key only if it already exists:

```ts
await redis.kv(sessions).set(sessionId, nextSession, {
  xx: true,
  ttlSeconds: 60 * 60
});
```

`nx` and `xx` cannot be combined, and neither can `ttlSeconds` with `keepTtl`; both invalid pairs are compile errors, not runtime throws. See [Type Safety](/benni/core-concepts/type-safety/).

# Type Safety

> Benni types come from codecs and schemas.

Benni types come from codecs and schemas.

```ts
export const users = hash("user", {
  name: string(),
  score: number(),
  active: boolean()
});
```

Writes must match the schema:

```ts
await redis.hash(users).hset("42", {
  name: "Ada",
  score: 10,
  active: true
});
```

Reads return decoded values:

```ts
const user = await redis.hash(users).hgetall("42");
//    ^? { name: string; score: number; active: boolean } | null
```

Hash field methods are typed by field name:

```ts
await redis.hash(users).hset("42", "score", 11);
const score = await redis.hash(users).hget("42", "score");
//    ^? number | null
```

For JSON values, the TypeScript type is supplied by the app:

```ts
type Session = {
  userId: string;
  createdAt: string;
};


export const sessions = kv("session", json<Session>());
```

## Inferring Types From Schemas

[Section titled “Inferring Types From Schemas”](#inferring-types-from-schemas)

Every schema carries type-only `$inferInput` / `$inferOutput` anchors, plus the `InferInput<T>` / `InferOutput<T>` utility types exported from `benni/schema`. Name a schema’s value types anywhere without redeclaring them:

```ts
import { hash, json, kv, number, string } from "benni/schema";
import type { InferInput, InferOutput } from "benni/schema";


export const users = hash("user", {
  name: string(),
  score: number()
});
export const profiles = kv("profile", json<Profile>());


type NewUser = InferInput<typeof users>;
//   ^? { name: string; score: number }


type StoredProfile = typeof profiles.$inferOutput;
//   ^? Profile
```

`InferInput` is the write-side type (what `hset`/`set` accept) and `InferOutput` the read-side type (what `hgetall`/`get` return, before the `| null`). They differ when a codec transforms values on the way through. The `$infer*` properties are type-only phantoms; they never exist at runtime, so only use them in type positions (`typeof users.$inferInput`).

## Runtime Validation With Standard Schema

[Section titled “Runtime Validation With Standard Schema”](#runtime-validation-with-standard-schema)

`json(validator)` accepts any [Standard Schema](https://standardschema.dev) validator (Zod, Valibot, ArkType, …). Reads are validated at runtime and the value type is inferred from the validator, with no explicit type parameter needed. See [schema builders](/benni/api/schema-builders/) for details.

```ts
import { z } from "zod";


const Profile = z.object({ name: z.string(), score: z.number() });
export const profiles = kv("profile", json(Profile));


const profile = await redis.kv(profiles).get("42");
//    ^? { name: string; score: number } | null (validated at runtime)
```

With the plain `json<T>()` form, `T` is trusted, not validated: Benni validates command reply shapes and decodes stored values, but does not check arbitrary JSON against your type. If untrusted code writes to the same Redis keys, pass a validator or validate at your application boundary.

Standard Schema validates reads only; it has no encode direction. To validate writes too, and to store rich types like `Date` or `bigint` that round-trip, use [Zod codecs via `benni/zod`](/benni/integrations/zod/).

## Typed Keys

[Section titled “Typed Keys”](#typed-keys)

Keys keep their literal types. `redis.query.users.key("42")` (and `redis.hash(users).key("42")`) has the type `"user:42"`, not `string`; template-literal key types survive the accessors and the query registry, so key-shaped APIs like `redis.watch([...])` stay precise.

## Illegal Option Combinations Don’t Compile

[Section titled “Illegal Option Combinations Don’t Compile”](#illegal-option-combinations-dont-compile)

Mutually exclusive command options are modeled in the types, so an invalid combination is a compile error rather than a runtime throw:

```ts
await redis.kv(profiles).set("42", value, { nx: true, xx: true });        // compile error
await redis.kv(profiles).set("42", value, { ttlSeconds: 60, keepTtl: true }); // compile error
await redis.zset(board).zadd("global", entry, { nx: true, gt: true });    // compile error
await redis.hash(users).hsetex("42", fields, { fnx: true, fxx: true });   // compile error
```

The same applies to `hsetex`’s expiry modes (at most one of `ttlSeconds` / `ttlMilliseconds` / `expireAtSeconds` / `expireAtMilliseconds` / `keepTtl`) and `geoadd`’s `nx`/`xx`.

# Bitmaps

> Use bitmaps for dense boolean flags addressed by integer offsets.

Use bitmaps for dense boolean flags addressed by integer offsets.

## Define A Bitmap

[Section titled “Define A Bitmap”](#define-a-bitmap)

```ts
import { bitmap } from "benni/schema";


export const dailyActive = bitmap("daily-active");
```

Bitmaps take no value codec. Each bit is addressed by a non-negative integer offset and exposed as a `boolean`.

## Set And Get Bits

[Section titled “Set And Get Bits”](#set-and-get-bits)

```ts
const previous = await redis.bitmap(dailyActive).setbit("2026-07-04", 42, true);
//    ^? boolean (the bit's previous value)


const active = await redis.bitmap(dailyActive).getbit("2026-07-04", 42);
```

## Count Set Bits

[Section titled “Count Set Bits”](#count-set-bits)

```ts
const total = await redis.bitmap(dailyActive).bitcount("2026-07-04");


const inFirstKilobyte = await redis.bitmap(dailyActive).bitcount("2026-07-04", {
  start: 0,
  end: 1023,
  unit: "BYTE"
});
```

The optional range takes `start`, `end`, and a `unit` of `"BYTE"` (the Redis default) or `"BIT"`.

## Find The First Bit

[Section titled “Find The First Bit”](#find-the-first-bit)

```ts
const first = await redis.bitmap(dailyActive).bitpos("2026-07-04", true);
//    ^? number | null
```

`bitpos` wraps `BITPOS` and returns `null` when no matching bit exists. Pass `start`, `end`, and `unit` to limit the search; `end` requires `start`, and `unit` requires both.

## Combine Bitmaps

[Section titled “Combine Bitmaps”](#combine-bitmaps)

```ts
const sizeInBytes = await redis.bitmap(dailyActive).bitop("2026-week-27", "OR", [
  "2026-07-01",
  "2026-07-02",
  "2026-07-03"
]);
```

`bitop(destination, operation, sources)` runs `BITOP` with `"AND"`, `"OR"`, `"XOR"`, or `"NOT"` and stores the result under the destination ID. `"NOT"` requires exactly one source ID.

## Packed Integer Fields

[Section titled “Packed Integer Fields”](#packed-integer-fields)

`bitfield` treats a key as a row of arbitrary-width integers packed at bit offsets, ideal for compact per-entity counters. Chain operations and call `exec`; the result is a **tuple typed to match the chain**:

```ts
const [views, previous, level] = await redis.bitmap(dailyActive)
  .bitfield("2026-07-04")
  .get("u32", 0) //           number       (read a 32-bit unsigned field)
  .set("u32", 0, 100) //      number | null (returns the previous value)
  .overflow("sat") //         mode for the ops that follow it
  .incrby("u8", "#8", 1) //   number | null (the new value)
  .exec();
```

* **Encoding**: `u1`–`u63` (unsigned) or `i1`–`i64` (signed).
* **Offset**: an absolute bit offset (`0`), or `#n` to address the *nth* field of that width (`"#8"` = the 9th `u8`, i.e. bit offset 64).
* **`get`** yields a `number`; **`set`** and **`incrby`** yield `number | null`.
* **`overflow`** sets the mode for the operations *after* it: `"wrap"` (the default, modular), `"sat"` (clamp to the type’s min/max), or `"fail"` (leave the field unchanged and return `null`). That `null` is why `set`/`incrby` are nullable.

```ts
// Atomic per-user counters clamped to a byte, no read-modify-write race.
const [clamped] = await redis.bitmap(quotas)
  .bitfield(userId)
  .overflow("sat")
  .incrby("u8", 0, 1)
  .exec();
```

## Delete

[Section titled “Delete”](#delete)

```ts
await redis.bitmap(dailyActive).del("2026-07-04");
```

## Raw Redis Equivalent

[Section titled “Raw Redis Equivalent”](#raw-redis-equivalent)

```ts
await nodeRedis.setBit("daily-active:2026-07-04", 42, 1);
const total = await nodeRedis.bitCount("daily-active:2026-07-04");
```

Use bitmaps for daily-active tracking, feature rollouts keyed by numeric user ID, and any dense set of boolean flags where offsets map to entities. Use sets when members are sparse strings rather than dense integers.

# Consumer Groups

> Use stream consumer groups for at-least-once delivery across many workers, with pending tracking and crash recovery.

Consumer groups let several workers share a [stream](/benni/data-structures/streams/) with at-least-once delivery. Redis tracks which entries each consumer has received but not yet acknowledged (the pending entries list, or PEL), so a crashed worker’s in-flight entries can be recovered by another.

Groups hang off the stream **store**, not the schema: group topology changes at deploy time, while the schema stays about data shape. You bind a group by name, then a consumer by name, and the stream id (the schema key id) stays the first argument of every call:

```ts
const group = redis.stream(auditEvents).group("processors");
const me = group.consumer(`c-${process.pid}`);
```

Everything except the blocking group read is non-blocking and runs on the shared client. The blocking read requires a [session](/benni/advanced/sessions/).

## Create A Group

[Section titled “Create A Group”](#create-a-group)

```ts
const created = await group.create("login", { from: "start" });
//    ^? boolean (true = created, false = the group already existed)
```

`create` is idempotent: it returns `false` when the group already exists rather than throwing. `from` is **required** and says where a new group starts reading, because defaulting would silently choose between replaying all history and skipping it:

* `"start"`: deliver the stream’s full history to the group.
* `"end"`: deliver only entries added after the group is created.
* `{ entryId: "1720094400000-0" }`: deliver everything after a specific id.

`create` also creates the stream if it is missing (`MKSTREAM`); pass `{ from, mkstream: false }` to require the stream to exist first.

## Read And Ack

[Section titled “Read And Ack”](#read-and-ack)

A consumer reads new deliveries with `xreadgroup` and acknowledges them with `xack`. New deliveries (`>`) never include tombstones:

```ts
const batch = await me.xreadgroup("login", { count: 20 }); // XREADGROUP > -> StreamEntry[]
for (const entry of batch) {
  await handleEntry(entry.value); // ^? Partial<{ type: string; userId: string }>
}
if (batch.length > 0) {
  await group.xack("login", batch.map((e) => e.id)); // XACK -> number acknowledged
}
```

`xack` also exists on the consumer (`me.xack("login", ids)`) as a convenience mirror, so worker code never has to reach back up to the group.

## Tombstones

[Section titled “Tombstones”](#tombstones)

An entry that is `XDEL`ed from the stream while it is still in a consumer’s PEL decodes as a **tombstone**: its `value` is `null`. Tombstones only ever appear on history and claim paths (`xreadgroup` with `after`, `xclaim`, `xautoclaim`), never on a live `xreadgroup`. You still have to ack a tombstone to clear it from the PEL; there is nothing left to process, so ack and move on:

```ts
for (const entry of await me.xreadgroup("login", { after: "0", count: 100 })) {
  if (entry.value === null) {
    await group.xack("login", [entry.id]); // tombstone: deleted upstream, just clear it
    continue;
  }
  await handleEntry(entry.value);
  await group.xack("login", [entry.id]);
}
```

## Crash Recovery With autoClaim

[Section titled “Crash Recovery With autoClaim”](#crash-recovery-with-autoclaim)

When a worker dies, its unacked entries sit idle in its PEL. `xautoclaim` (`XAUTOCLAIM`) steals entries idle longer than `minIdleMs` and reassigns them to the calling consumer. It scans with a cursor: `"0-0"` starts a scan, and a returned cursor of `"0-0"` means the scan is complete.

```ts
let cursor = "0-0";
do {
  const res = await me.xautoclaim("login", {
    minIdleMs: 60_000, // only steal entries idle > 60s (assume the owner is dead)
    start: cursor,
    count: 50
  });
  //   ^? { cursor: string; entries: PendingStreamEntry[]; deletedIds: string[] }
  for (const entry of res.entries) {
    if (entry.value === null) {
      await group.xack("login", [entry.id]); // tombstone
      continue;
    }
    await handleEntry(entry.value);
    await group.xack("login", [entry.id]);
  }
  cursor = res.cursor;
} while (cursor !== "0-0");
```

On Redis 7+, entry ids that `xautoclaim` finds already deleted from the stream are dropped from the PEL by Redis itself and reported in `deletedIds`; nothing to ack for those. To recover a **specific** set of entries by id, use `xclaim` (`XCLAIM`) with the same `minIdleMs` guard.

## Inspect Pending Work

[Section titled “Inspect Pending Work”](#inspect-pending-work)

For dashboards and janitors, `xpending` gives a summary and `xpending` with options lists individual pending entries, both non-blocking on the shared client:

```ts
const summary = await group.xpending("login");
//    ^? { count; minEntryId; maxEntryId; consumers: [{ consumer; count }] }


const stuck = await group.xpending("login", { count: 10, minIdleMs: 300_000 });
for (const row of stuck) {
  //  ^? { entryId; consumer; idleMs; deliveries }
  if (row.deliveries > 5) {
    await deadLetter(row.entryId); // the delivery counter is a poison-pill detector
  }
}
```

`count` is required on the `xpending` extended form because Redis requires it. Idle thresholds are milliseconds and carry a `...Ms` suffix (`minIdleMs`, `idleMs`): the PEL speaks milliseconds, while [blocking timeouts](/benni/advanced/blocking-operations/) speak seconds (`timeoutSeconds`), and the suffixes keep the units unambiguous at call sites.

`deleteConsumer` removes a consumer and destroys its PEL entries; `destroy` removes the whole group.

## Blocking Group Read (Session Only)

[Section titled “Blocking Group Read (Session Only)”](#blocking-group-read-session-only)

A live worker that wants to wait for new deliveries uses `xreadgroup` with a `timeoutSeconds`, which is only reachable through a session, since it parks the connection like any other [blocking operation](/benni/advanced/blocking-operations/):

```ts
await redis.session(async (s) => {
  const live = s.stream(auditEvents).group("processors").consumer(`c-${process.pid}`);
  while (!shutdown.signal.aborted) {
    const batch = await live.xreadgroup("login", { timeoutSeconds: 5, count: 20 });
    for (const entry of batch) await handleEntry(entry.value);
    if (batch.length > 0) {
      await group.xack("login", batch.map((e) => e.id)); // ack via the shared client
    }
  }
});
```

Blocking `xreadgroup` always reads `>` (new deliveries), because Redis only honors `BLOCK` for new entries, so there is no id parameter, and no tombstones. It returns an empty array on timeout, which the `{ timeoutSeconds: 5 }` loop above treats as a heartbeat.

## Full Worker Lifecycle

[Section titled “Full Worker Lifecycle”](#full-worker-lifecycle)

The pieces above compose into a worker: recover this consumer’s own history, steal from dead peers, then loop on the blocking read.

```ts
const group = redis.stream(auditEvents).group("processors");
const me = group.consumer(`c-${process.pid}`);


await group.create("login", { from: "start" }); // idempotent bootstrap


// (a) recover this consumer's own unacked work from a previous crash
for (const entry of await me.xreadgroup("login", { after: "0", count: 100 })) {
  if (entry.value === null) { await group.xack("login", [entry.id]); continue; }
  await handleEntry(entry.value);
  await group.xack("login", [entry.id]);
}


// (b) steal work abandoned by dead consumers (idle > 60s)
let cursor = "0-0";
do {
  const res = await me.xautoclaim("login", { minIdleMs: 60_000, start: cursor, count: 50 });
  for (const entry of res.entries) {
    if (entry.value === null) { await group.xack("login", [entry.id]); continue; }
    await handleEntry(entry.value);
    await group.xack("login", [entry.id]);
  }
  cursor = res.cursor;
} while (cursor !== "0-0");


// (c) live loop: the blocking group read is only reachable through a session
await redis.session(async (s) => {
  const live = s.stream(auditEvents).group("processors").consumer(`c-${process.pid}`);
  while (!shutdown.signal.aborted) {
    const batch = await live.xreadgroup("login", { timeoutSeconds: 5, count: 20 });
    for (const entry of batch) await handleEntry(entry.value);
    if (batch.length > 0) await group.xack("login", batch.map((e) => e.id));
  }
});
```

`XINFO`, `XGROUP SETID`/`CREATECONSUMER`, `NOACK`, and multi-stream group reads are not wrapped yet; use [`redis.raw`](/benni/core-concepts/raw-redis-access/) for those.

# Geospatial

> Use geo sets to index members by coordinates and query them by radius or box.

Use geo sets to index members by coordinates and query them by radius or box.

## Define A Geo Set

[Section titled “Define A Geo Set”](#define-a-geo-set)

```ts
import { geo, string } from "benni/schema";


export const stores = geo("stores", string());
```

## Add Members

[Section titled “Add Members”](#add-members)

```ts
await redis.geo(stores).geoadd("berlin", [
  { member: "store:1", longitude: 13.405, latitude: 52.52 },
  { member: "store:2", longitude: 13.3888, latitude: 52.517 }
]);
```

Coordinates are validated before the command is sent: longitude must be between -180 and 180, latitude between -85.05112878 and 85.05112878 (the Redis geohash limits). Out-of-range values throw a `TypeError` instead of a server error.

`geoadd` accepts `{ nx: true }` to only add new members, `{ xx: true }` to only update existing ones, and `{ ch: true }` to include updated members in the returned count, the same tokens `zadd` takes. `nx` and `xx` are mutually exclusive; combining them is a compile error.

## Positions And Distances

[Section titled “Positions And Distances”](#positions-and-distances)

```ts
const positions = await redis.geo(stores).geopos("berlin", ["store:1"]);
//    ^? Array<{ longitude: number; latitude: number } | null>


const meters = await redis.geo(stores).geodist("berlin", "store:1", "store:2");
const km = await redis.geo(stores).geodist("berlin", "store:1", "store:2", "km");
```

`geodist` returns `null` when either member is missing. Units are `"m"` (the default), `"km"`, `"mi"`, and `"ft"`.

## Geohashes

[Section titled “Geohashes”](#geohashes)

```ts
const hashes = await redis.geo(stores).geohash("berlin", ["store:1", "store:2"]);
//    ^? Array<string | null>
```

## Search

[Section titled “Search”](#search)

```ts
const nearby = await redis.geo(stores).geosearch("berlin", {
  from: { longitude: 13.4, latitude: 52.52 },
  by: { radius: 5, unit: "km" },
  order: "asc",
  count: { count: 10 },
  withDistance: true,
  withCoordinates: true
});
//    ^? Array<{ member: string; distance?: number; coordinates?: { ... } }>
```

`from` is either a coordinate pair or `{ member }` to search around an existing member. `by` is either `{ radius, unit }` for a circle or `{ width, height, unit }` for a box. `distance` and `coordinates` appear on results only when requested.

## Store Search Results

[Section titled “Store Search Results”](#store-search-results)

```ts
await redis.geo(stores).geosearchstore("berlin-center", "berlin", {
  from: { member: "store:1" },
  by: { radius: 2, unit: "km" }
});
```

`geosearchstore` writes matches into the destination key. Pass `{ storeDistance: true }` to store each member’s distance as its sorted-set score.

## Delete

[Section titled “Delete”](#delete)

```ts
await redis.geo(stores).del("berlin");
```

## Raw Redis Equivalent

[Section titled “Raw Redis Equivalent”](#raw-redis-equivalent)

```ts
await nodeRedis.geoAdd("stores:berlin", [
  { member: "store:1", longitude: 13.405, latitude: 52.52 }
]);
```

Use geo sets for store locators, delivery zones, and nearby-entity queries. Redis stores them as sorted sets under the hood, so sorted-set commands also work on the same keys.

# Hashes

> Use hashes when you want to store object-like data under a Redis key.

Use hashes when you want to store object-like data under a Redis key.

## Define A Hash

[Section titled “Define A Hash”](#define-a-hash)

```ts
import { hash, number, string } from "benni/schema";


export const users = hash("user", {
  name: string(),
  score: number()
});
```

## Write A Hash

[Section titled “Write A Hash”](#write-a-hash)

```ts
await redis.hash(users).hset("42", {
  name: "Ada",
  score: 10
});
```

## Read A Hash

[Section titled “Read A Hash”](#read-a-hash)

```ts
const user = await redis.hash(users).hgetall("42");
//    ^? { name: string; score: number } | null
```

## Update Fields

[Section titled “Update Fields”](#update-fields)

```ts
await redis.hash(users).hset("42", "score", 11);
await redis.hash(users).hincrby("42", "score", 1);
```

## Read Fields

[Section titled “Read Fields”](#read-fields)

One `hget`, two jobs: pass a field name to read that one field, or nothing to read the whole record. There is no `hgetField` or `hgetOne`.

```ts
const score = await redis.hash(users).hget("42", "score");
//    ^? number | null      (one field)


const user = await redis.hash(users).hget("42");
//    ^? { name: string; score: number } | null   (the whole record)


const fields = await redis.hash(users).hmget("42", ["name", "score"]);
//    ^? { name?: string | null; score?: number | null }
```

The single-field form returns the field’s decoded type, so `hget("42", "score")` is a `number | null` and not a string you have to parse.

## Missing Declared Fields Throw

[Section titled “Missing Declared Fields Throw”](#missing-declared-fields-throw)

A hash under `hash("user", …)` is a record your schema owns, so the whole-record read insists on it. `hget("42")` needs every declared field and throws a `PartialRecordError` when one is gone (deleted with `hdel`, or expired by a per-field TTL). `hgetall` is the tolerant read for exactly that case, and types its result as `Partial`:

```ts
const strict = await redis.hash(users).hget("42");
//    ^? { name: string; score: number } | null   (throws PartialRecordError if incomplete)


const tolerant = await redis.hash(users).hgetall("42");
//    ^? { name?: string; score?: number } | null
```

This is the opposite of how [stream](/benni/data-structures/streams/) entry values behave, which are always `Partial` and never throw. The difference is who writes the key: a hash is a record you own, while a stream is an append log any producer can write to. See [Entry Values Are Partial](/benni/data-structures/streams/#entry-values-are-partial).

## Random Fields

[Section titled “Random Fields”](#random-fields)

Pick field names at random with `HRANDFIELD`. `hrandfield` with no count returns a single field name, or `null` when the key is missing:

```ts
const field = await redis.hash(users).hrandfield("42");
//    ^? string | null
```

Pass a nonzero `count`. A positive count returns that many **distinct** field names (capped at the hash’s size); a negative count allows repeats and always returns `|count|` names:

```ts
const distinct = await redis.hash(users).hrandfield("42", { count: 2 });
//    ^? string[]   (up to 2 distinct field names)


const withRepeats = await redis.hash(users).hrandfield("42", { count: -5 });
//    ^? string[]   (exactly 5 names, repeats allowed)
```

Both forms return raw field names; like `hkeys`, the result may include fields not declared in the schema. The value-bearing form (`HRANDFIELD ... WITHVALUES`) is intentionally not provided: a random field’s value cannot be soundly decoded without knowing which codec it belongs to, the same reason there is no bare `HVALS` accessor.

## Field Expiration

[Section titled “Field Expiration”](#field-expiration)

Redis 7.4+ can expire individual hash fields, and Redis 8 adds get/set variants that touch field TTLs atomically.

Set a per-field TTL with `hexpire`. Pass a number for a relative TTL in seconds, or an options object to choose the unit and whether the value is a relative duration or an absolute Unix time:

```ts
await redis.hash(users).hexpire("42", ["score"], 3600); // HEXPIRE (seconds)
await redis.hash(users).hexpire("42", ["score"], { ttlMilliseconds: 500 }); // HPEXPIRE
await redis.hash(users).hexpire("42", ["score"], { expireAtSeconds: 1893456000 }); // HEXPIREAT
```

Read the remaining TTL or the absolute expiry time (each in seconds by default, or milliseconds with `{ milliseconds: true }`), and clear TTLs with `hpersist`:

```ts
await redis.hash(users).httl("42", "score"); // HTTL (seconds)
await redis.hash(users).httl("42", "score", { milliseconds: true }); // HPTTL
await redis.hash(users).hexpiretime("42", "score"); // HEXPIRETIME
await redis.hash(users).hpersist("42", ["score"]); // HPERSIST
```

Get, set, and delete fields while touching their TTL in a single round trip:

```ts
// HGETEX: read fields and (optionally) reset their TTL.
const seen = await redis.hash(users).hgetex("42", ["name"], { ttlSeconds: 60 });


// HSETEX: set fields with a TTL atomically; fnx writes only if no field exists,
// fxx only if all do (the Redis FNX/FXX tokens); combining them is a compile error.
const wrote = await redis.hash(users).hsetex(
  "42",
  { name: "Ada", score: 10 },
  { ttlSeconds: 3600 }
);


// HGETDEL: read fields and delete them (the key is removed once its last field goes).
const removed = await redis.hash(users).hgetdel("42", ["name", "score"]);
```

`hsetex` writes only the fields you pass, and it rejects a field whose value is `undefined` rather than storing the string `"undefined"`: omit the key to leave that field alone. `hgetex` with an empty field list rejects too when you pass an expiry, because there is no field to apply it to.

A lapsed field TTL leaves the hash partially populated, and so does `hdel` or `hgetdel` on a declared field. That is precisely the case [`hgetall` exists for](#missing-declared-fields-throw): a record with per-field TTLs should be read with `hgetall`, since `hget("42")` throws a `PartialRecordError` the moment one declared field has gone.

## Delete Fields Or The Hash

[Section titled “Delete Fields Or The Hash”](#delete-fields-or-the-hash)

`hdel` takes one field or an array and returns the count removed:

```ts
await redis.hash(users).hdel("42", "score");
await redis.hash(users).hdel("42", ["name", "score"]);
await redis.hash(users).del("42");
```

## With TTL

[Section titled “With TTL”](#with-ttl)

```ts
await redis.hash(users).hset(
  "42",
  { name: "Ada", score: 10 },
  { ttlSeconds: 3600 }
);
```

## Raw Redis Equivalent

[Section titled “Raw Redis Equivalent”](#raw-redis-equivalent)

```ts
await nodeRedis.hSet("user:42", {
  name: "Ada",
  score: "10"
});
```

Use hashes for users, profiles, counters, session metadata, and object-like data where fields may be read or updated independently. Prefer a JSON key-value schema when the whole object is usually stored and read as one blob.

# HyperLogLog

> Use HyperLogLog when you need approximate cardinality counts with low memory usage.

Use HyperLogLog when you need approximate cardinality counts with low memory usage.

```ts
import { hll, string } from "benni/schema";


export const pageViews = hll("page-views", string());
```

Add values:

```ts
await redis.hll(pageViews).pfadd("2026-07-04", [
  "user:42",
  "user:7"
]);
```

`pfadd` always takes an array, so a single value is `pfadd(id, [value])`; there is no single-value overload. An empty array is a no-op rather than a command that would create the key.

Count unique values:

```ts
const uniqueVisitors = await redis.hll(pageViews).pfcount("2026-07-04");
```

Count across multiple keys:

```ts
const weeklyVisitors = await redis.hll(pageViews).pfcount([
  "2026-07-01",
  "2026-07-02",
  "2026-07-03"
]);
```

Merge keys:

```ts
await redis.hll(pageViews).pfmerge("2026-week-27", [
  "2026-07-01",
  "2026-07-02",
  "2026-07-03"
]);
```

Raw Redis equivalent:

```ts
await nodeRedis.pfAdd("page-views:2026-07-04", ["user:42", "user:7"]);
const uniqueVisitors = await nodeRedis.pfCount("page-views:2026-07-04");
```

Use HyperLogLog for approximate unique counts such as daily visitors, active users, unique IPs, and event reach. Use sets when you need exact membership checks or exact members back.

# JSON Values

> Benni JSON values are encoded into normal Redis string values with JSON.stringify and decoded with JSON.parse.

Benni JSON values are encoded into normal Redis string values with `JSON.stringify` and decoded with `JSON.parse`.

`json` has two forms. Reach for the validating one by default:

```ts
import { z } from "zod";
import { json, kv } from "benni/schema";


const Settings = z.object({
  theme: z.enum(["light", "dark"]),
  emailNotifications: z.boolean()
});


export const settings = kv("settings", json(Settings));
```

`json(validator)` accepts any [Standard Schema](https://standardschema.dev) validator (Zod, Valibot, ArkType, …), infers the value type from it (no type parameter needed), and checks **every read** at runtime.

Write JSON:

```ts
await redis.kv(settings).set("user:42", {
  theme: "dark",
  emailNotifications: true
});
```

Read JSON:

```ts
const value = await redis.kv(settings).get("user:42");
//    ^? { theme: "light" | "dark"; emailNotifications: boolean } | null (validated)
```

Data that does not match throws a `ReplyShapeError` naming the failing paths, at the read that found it.

With Redis directly:

```ts
await nodeRedis.set(
  "settings:user:42",
  JSON.stringify({
    theme: "dark",
    emailNotifications: true
  })
);


const raw = await nodeRedis.get("settings:user:42");
const value = raw === null ? null : JSON.parse(raw);
```

## `json<T>()` Is The Unchecked Escape Hatch

[Section titled “json\<T>() Is The Unchecked Escape Hatch”](#jsont-is-the-unchecked-escape-hatch)

The second form takes a TypeScript type instead of a validator:

```ts
type Settings = {
  theme: "light" | "dark";
  emailNotifications: boolean;
};


export const settings = kv("settings", json<Settings>()); // no runtime check
```

`json<T>()` is a pure cast. `T` is asserted, never verified: `JSON.parse` runs, the result is handed back as `T`, and nothing compares the two. Be clear about what that means:

* A stored value missing required fields reads back as a complete `T`. Store `{"slug":"a"}` under a `json<ClickEvent>()` where `ClickEvent` has three required fields, and the read resolves to `{ slug: "a" }` typed as a full `ClickEvent`. Nothing throws; the `undefined`s surface much later, somewhere else.
* A field of the wrong type reads back as the declared type.
* Extra fields are kept and invisible to the types.
* Only genuinely malformed JSON throws, and only because `JSON.parse` refuses it.

Use it when the value’s provenance is genuinely beyond doubt and you accept that cost: a cache entry your own code wrote in the same deploy, a value you are about to revalidate anyway, or a hot path where you have measured the validator and decided against it. Anything that crosses a version boundary, a service boundary, or a schema change (which is most stored data, since Redis outlives your process) wants `json(validator)`.

Note which direction each form covers. Standard Schema defines reads only, so `json(validator)` validates reads and trusts writes. To validate both, and to store rich types like `Date` that genuinely round-trip, use a [Zod codec](/benni/integrations/zod/) via `zodJson(schema)`.

Both forms work in every value-carrying schema, not just `kv`: `list("events", json(ClickEvent))` validates each element it decodes, while `list("events", json<ClickEvent>())` does not.

See [schema builders](/benni/api/schema-builders/) for the full codec list.

Use JSON key-value schemas when your app treats the value as a document. Use hashes when individual fields need independent Redis operations.

# Key Values

> Use key-value schemas when one Redis key stores one scalar or serialized value.

Use key-value schemas when one Redis key stores one scalar or serialized value.

## Define A Key-Value Schema

[Section titled “Define A Key-Value Schema”](#define-a-key-value-schema)

```ts
import { json, kv } from "benni/schema";


type UserProfile = {
  name: string;
  score: number;
};


export const profiles = kv("profile", json<UserProfile>());
```

## Write

[Section titled “Write”](#write)

```ts
await redis.kv(profiles).set("42", {
  name: "Ada",
  score: 10
});
```

## Read

[Section titled “Read”](#read)

```ts
const profile = await redis.kv(profiles).get("42");
//    ^? UserProfile | null
```

## With TTL

[Section titled “With TTL”](#with-ttl)

```ts
await redis.kv(profiles).set(
  "42",
  { name: "Ada", score: 10 },
  { ttlSeconds: 3600 }
);
```

## Conditional Writes

[Section titled “Conditional Writes”](#conditional-writes)

```ts
const created = await redis.kv(profiles).set("42", profile, {
  nx: true,
  ttlSeconds: 3600
});


const updated = await redis.kv(profiles).set("42", profile, {
  xx: true
});
```

## Raw Redis Equivalent

[Section titled “Raw Redis Equivalent”](#raw-redis-equivalent)

```ts
await nodeRedis.set("profile:42", JSON.stringify(profile), {
  EX: 3600
});
```

Use key-value schemas for sessions, feature flags, cached API responses, and values that are usually read or written as a whole.

# Pub/Sub

> Use typed channels when publishers and subscribers should share a message shape.

Use typed channels when publishers and subscribers should share a message shape.

```ts
import { channel, json } from "benni/schema";


export const userEvents = channel(
  "events:user",
  json<{ id: string; action: "created" | "deleted" }>()
);
```

There is nothing to configure. Bind a client the usual way and reach the channel through `redis.pubsub`:

```ts
import { benni } from "benni";
import { node } from "benni/node";
import * as schema from "./schema";


const client = await node();
export const redis = benni(client, { schema });
```

## Publishing

[Section titled “Publishing”](#publishing)

`PUBLISH` is one stateless command, so publishing rides the bound client and works on every adapter, including [`benni/upstash`](/benni/runtime/edge/) on the edge:

```ts
const receivers = await redis.pubsub.channel(userEvents).publish({
  id: "42",
  action: "created"
});
//    ^? number (how many subscribers Redis delivered to)
```

## Subscribing

[Section titled “Subscribing”](#subscribing)

`subscribe` takes only a handler. The first subscription lazily leases one subscriber connection from the bound client; every later channel and pattern is multiplexed onto that same connection, and it closes again when the last subscription goes away. So there are no idle connections to manage, and no second object to pass around:

```ts
const subscription = await redis.pubsub.channel(userEvents).subscribe((message) => {
  // message is { id: string; action: "created" | "deleted" }
  console.log(message.action);
});


await subscription.unsubscribe();
```

Subscribing to the same channel twice costs one Redis subscription, not two: Benni registers a single listener per channel name and fans out to your handlers. Each `subscribe` call gets its own `unsubscribe`, and the channel is dropped from Redis when the last handler for it leaves.

To tear everything down at once (on shutdown, or between tests), use `close`:

```ts
await redis.pubsub.close();
```

That drops every subscription and closes the leased connection. It also ends any `stream()` loop that is still running, including one you started without a signal, so a shutdown path can await its consumers. Publishing keeps working afterwards, and a later `subscribe` simply leases a fresh connection.

## Patterns

[Section titled “Patterns”](#patterns)

Use a pattern when one handler should receive several channels. The handler also gets the concrete channel the message arrived on:

```ts
import { json, pattern } from "benni/schema";


export const userEventPattern = pattern(
  "events:user:*",
  json<{ id: string; action: string }>()
);


const subscription = await redis.pubsub
  .pattern(userEventPattern)
  .subscribe((message, channelName) => {
    console.log(channelName, message.action);
  });


await subscription.unsubscribe();
```

Patterns share the same leased connection and the same ref-counting as channels.

## One channel per entity

[Section titled “One channel per entity”](#one-channel-per-entity)

Most Pub/Sub is per something: one channel per chat room, per user, per job. Pass an id to `channel()` and Benni derives `name:<id>` the same way a keyspace derives `prefix:<id>`:

```ts
import { channel, json } from "benni/schema";


export const roomEvents = channel(
  "chat:room",
  json<{ from: string; text: string }>()
);
```

```ts
// PUBLISH chat:room:42
await redis.pubsub
  .channel(roomEvents, "42")
  .publish({ from: "ada", text: "hi" });


// SUBSCRIBE chat:room:42
const subscription = await redis.pubsub
  .channel(roomEvents, "42")
  .subscribe((message) => {
    console.log(message.text);
  });
```

The id is optional, and leaving it off is unchanged: `redis.pubsub.channel(roomEvents)` still addresses `chat:room` itself and nothing else. So one schema can carry both a per-room feed and a channel for everyone.

Ids are typed like keyspace ids (a string, a number, or a bigint), and `ids` narrows them to a known set for autocomplete and a compile-time check:

```ts
export const jobEvents = channel(
  "jobs",
  json<{ state: "queued" | "running" | "done" }>(),
  { ids: ["import", "export"] }
);


await redis.pubsub.channel(jobEvents, "import").publish({ state: "done" });
// redis.pubsub.channel(jobEvents, "nope") does not compile
```

You never have to build the channel string yourself. `channelName` resolves it, on the schema and on the resource, the way `key` does for a keyspace:

```ts
roomEvents.channelName("42"); // "chat:room:42"
roomEvents.channelName(); // "chat:room"
redis.pubsub.channel(roomEvents, "42").channelName(); // "chat:room:42"
```

A schema reached through the [registry](/benni/core-concepts/schema-registry/) scopes with `at(id)`, which is what `redis.pubsub.channel(schema, id)` calls underneath:

```ts
await redis.query.roomEvents.at("42").publish({ from: "ada", text: "hi" });
```

There is no `hashTag` option on a channel, because a channel is not a key: plain Pub/Sub is broadcast across a cluster rather than routed by slot, so there is no co-location to arrange.

### Pairing with a pattern

[Section titled “Pairing with a pattern”](#pairing-with-a-pattern)

Per-entity channels and patterns are two halves of the same shape: publish to one room, subscribe to all of them. The id is derived by exactly the rule a keyspace uses, so a pattern over the prefix matches every channel the schema can produce:

```ts
import { json, pattern } from "benni/schema";


export const anyRoom = pattern(
  "chat:room:*",
  json<{ from: string; text: string }>()
);


await redis.pubsub.pattern(anyRoom).subscribe((message, channelName) => {
  console.log(channelName, message.text); // "chat:room:42 hi"
});


await redis.pubsub
  .channel(roomEvents, "42")
  .publish({ from: "ada", text: "hi" });
```

Ids are joined on verbatim, exactly as a keyspace joins them, so nothing about an id is interpreted: a channel subscribe is a literal name, never a glob, and only `pattern()` reads `*`, `?`, and `[...]` as wildcards. An id containing a colon simply nests one level further, and `chat:room:*` still matches it, because a Redis glob `*` spans colons too.

## Consuming as an async iterator

[Section titled “Consuming as an async iterator”](#consuming-as-an-async-iterator)

Callbacks are awkward when the consumer is a loop: an SSE response, a worker that processes one message at a time. `stream()` gives you the same subscription as an async iterable, and releases it when iteration ends:

```ts
const controller = new AbortController();


for await (const message of redis.pubsub
  .channel(userEvents)
  .stream({ signal: controller.signal })) {
  console.log(message.action);
}
```

Aborting the signal ends the loop; so does `break`ing out of it or `return`ing from the enclosing function. Either way the subscription is released on the way out, which means the leased connection closes too if nothing else is subscribed.

The pattern form yields the channel alongside the message, because with a pattern you usually need to know which channel matched:

```ts
for await (const { message, channel: channelName } of redis.pubsub
  .pattern(userEventPattern)
  .stream({ signal: controller.signal })) {
  console.log(channelName, message.action);
}
```

Messages that arrive while your loop body is busy are buffered in memory, so a slow consumer does not drop messages, but it also does not apply backpressure to Redis, which has none for Pub/Sub. If your consumer can fall behind indefinitely, use a [stream](/benni/data-structures/streams/) instead: Pub/Sub is fire-and-forget and has no replay.

## When a handler throws

[Section titled “When a handler throws”](#when-a-handler-throws)

Delivery continues to the other handlers no matter what one of them does. By default a handler that throws or rejects is rethrown asynchronously, so the failure surfaces as an unhandled error instead of being swallowed. Pass `onPubSubError` when you would rather route it somewhere:

```ts
const redis = benni(client, {
  schema,
  onPubSubError: (error) => logger.error({ error }, "pubsub handler failed")
});
```

## Adapter support

[Section titled “Adapter support”](#adapter-support)

Subscribing needs a connection the adapter can hold open, which is the one thing HTTP cannot do:

| Adapter                                     | Publish | Channel subscribe      | Pattern subscribe                                  |
| ------------------------------------------- | ------- | ---------------------- | -------------------------------------------------- |
| [`benni/node`](/benni/runtime/node/)        | Yes     | Yes                    | Yes                                                |
| [`benni/bun`](/benni/runtime/bun-and-deno/) | Yes     | Yes                    | No (`psubscribe` is broken upstream in Bun 1.3.14) |
| [`benni/upstash`](/benni/runtime/edge/)     | Yes     | No (HTTP is stateless) | No                                                 |

Both gaps fail loudly rather than hanging. Subscribing on an adapter that cannot lease a connection throws `TypeError`, and so does `pattern(...).subscribe(...)` on Bun. An adapter advertises the capability by implementing the optional `subscriber?()` method on the [client contract](/benni/api/benni-client/#redispubsub), the same way `session?()` advertises sessions.

# Sets And Lists

> Sets and lists model collections under Redis keys while preserving member types.

Sets and lists model collections under Redis keys while preserving member types.

## Sets

[Section titled “Sets”](#sets)

Use sets for unique membership.

```ts
import { set, string } from "benni/schema";


export const teamMembers = set("team-members", string());


await redis.set(teamMembers).sadd("engineering", ["ada", "grace"]);


const hasAda = await redis.set(teamMembers).sismember("engineering", "ada");
const members = await redis.set(teamMembers).smembers("engineering");
```

Raw Redis equivalent:

```ts
await nodeRedis.sAdd("team-members:engineering", ["ada", "grace"]);
const members = await nodeRedis.sMembers("team-members:engineering");
```

## Lists

[Section titled “Lists”](#lists)

Use lists for ordered queues, recent items, and bounded histories.

```ts
import { json, list } from "benni/schema";


type Event = {
  type: string;
  at: string;
};


export const events = list("events", json<Event>());


await redis.list(events).rpush("user:42", [
  { type: "login", at: new Date().toISOString() }
]);


const recent = await redis.list(events).lrange("user:42", 0, 9);
```

Raw Redis equivalent:

```ts
await nodeRedis.rPush("events:user:42", JSON.stringify(event));
const recent = await nodeRedis.lRange("events:user:42", 0, 9);
```

## Members Are Always Passed As An Array

[Section titled “Members Are Always Passed As An Array”](#members-are-always-passed-as-an-array)

The variadic writers (`sadd`, `srem`, `lpush`, `rpush`, and `pfadd` on a [HyperLogLog](/benni/data-structures/hyperloglog/)) take an array, even for a single member. There is no single-value overload, so `lpush(id, value)` does not compile:

```ts
await redis.list(events).lpush("user:42", [event]);  // one member
await redis.list(events).lpush("user:42", [a, b]);   // many
```

One shape means a loop that pushes one item and a batch that pushes a hundred are the same call, and an empty array is a no-op rather than a command that would create the key. The readers that return a slice are positional, matching Redis: `lrange(id, start, stop)`.

# Sorted Sets

> Use sorted sets for ranked values: leaderboards, priorities, timestamps, and scored indexes.

Use sorted sets for ranked values: leaderboards, priorities, timestamps, and scored indexes.

```ts
import { zset, string } from "benni/schema";


export const leaderboards = zset("leaderboard", string());
```

Add members with scores; `zadd` takes a single entry or an array:

```ts
await redis.zset(leaderboards).zadd("global", { member: "user:42", score: 100 });


await redis.zset(leaderboards).zadd("global", [
  { member: "user:42", score: 100 },
  { member: "user:7", score: 80 }
]);
```

Conditions mirror the Redis tokens: `nx` (only add new members), `xx` (only update existing), `gt`/`lt` (only move a score up/down), and `ch` (count changed members instead of only added ones). Illegal combinations (`nx` with `xx`, `gt`, or `lt`, and `gt` with `lt`) are compile errors:

```ts
await redis.zset(leaderboards).zadd("global", entries, { gt: true, ch: true });
```

Read the top members:

```ts
const top = await redis.zset(leaderboards).zrange("global", { start: 0, stop: 9, rev: true });
```

Read members with scores:

```ts
const entries = await redis.zset(leaderboards).zrange("global", { start: 0, stop: 9, withScores: true });
//    ^? Array<{ member: string; score: number }>
```

`withScores` decides the return type, so it has to be a literal `true` or `false`, never a variable of type `boolean`. The same holds for `zdiff`, `zunion`, `zinter`, and `zrandmember`. If the flag is computed, branch on it and make two calls: the compiler cannot pick a reply shape it does not know yet.

Note the call shape: `zrange` takes an **options object**, not positional bounds, and the range, direction, and score flag all live in it (`{ start, stop, rev, withScores }`). There is no separate `zrangeWithScores` method. One signature is what lets `byScore` and `byLex` (below) reuse the same call instead of multiplying into a method per Redis token.

Increment a score:

```ts
await redis.zset(leaderboards).zincrby("global", 5, "user:42");
```

Scores may be `Infinity` or `-Infinity`, which Redis stores as `+inf` and `-inf`. `zscore` reads them back as the JavaScript infinities, and every score bound (`byScore` ranges, `zcount`, `zremrangebyscore`) accepts the same values, so a score can go straight back in as a bound. Only `NaN` is rejected.

Raw Redis equivalent:

```ts
await nodeRedis.zAdd("leaderboard:global", [
  { value: "user:42", score: 100 },
  { value: "user:7", score: 80 }
]);
```

Sorted sets are a good fit for leaderboards, ranking search candidates, rate-limit windows, delayed jobs, and anything where score ordering matters.

## Lexicographic Ranges

[Section titled “Lexicographic Ranges”](#lexicographic-ranges)

When every member in a sorted set shares the same score, Redis orders them lexically by member value. That turns a sorted set into a sorted index, handy for autocomplete, prefix search, or any alphabetized listing. Calling `zrange` with `byLex: true` exposes Redis’s `BYLEX` family over that ordering.

```ts
import { zset, string } from "benni/schema";


export const names = zset("name-index", string());
```

Add every member with the **same score** so ordering is purely lexical:

```ts
await redis.zset(names).zadd("directory", [
  { member: "adam", score: 0 },
  { member: "ada", score: 0 },
  { member: "ben", score: 0 },
  { member: "bella", score: 0 },
  { member: "cara", score: 0 }
]);
```

Range over members between two bounds. A bound is either the `"-"` / `"+"` sentinel (lowest / highest possible member) or `{ value }`, which is inclusive by default:

```ts
const aToB = await redis.zset(names).zrange("directory", {
  byLex: true,
  min: { value: "ada" },
  max: { value: "ben" }
});
//    ^? string[]   → ["ada", "adam", "ben"]
```

Set `inclusive: false` on a bound to make it exclusive:

```ts
const openEnded = await redis.zset(names).zrange("directory", {
  byLex: true,
  min: { value: "ada", inclusive: false },
  max: { value: "ben", inclusive: false }
});
//    → ["adam"]
```

Use the `"-"` and `"+"` sentinels for open ranges; this reads every member, in order. `offset` and `count` apply a `LIMIT` and must be provided together:

```ts
const firstThree = await redis.zset(names).zrange("directory", {
  byLex: true,
  min: "-",
  max: "+",
  offset: 0,
  count: 3
});
//    → ["ada", "adam", "bella"]
```

Set `rev: true` to walk the range high-to-low. The `min`/`max` bounds still describe the low and high ends of the range; only the result order flips:

```ts
const reversed = await redis.zset(names).zrange("directory", {
  byLex: true,
  min: "-",
  max: "+",
  rev: true
});
//    → ["cara", "bella", "ben", "adam", "ada"]
```

Count the members in a lex range without materializing them:

```ts
const inRange = await redis.zset(names).zlexcount(
  "directory",
  { value: "ada" },
  { value: "ben" }
);
//    ^? number   → 3
```

Remove every member in a lex range:

```ts
const removed = await redis.zset(names).zremrangebylex(
  "directory",
  { value: "ada" },
  { value: "adam" }
);
//    ^? number   (members deleted)
```

Store a lex slice into another key with `zrangestore` and `byLex: true`. It accepts the same `min`, `max`, `rev`, `offset`, and `count` options as a `byLex` `zrange` and returns the number of members written:

```ts
const stored = await redis.zset(names).zrangestore("b-names", "directory", {
  byLex: true,
  min: { value: "b" },
  max: { value: "c", inclusive: false }
});
//    ^? number   → 2   ("bella", "ben" written to the "b-names" key)
```

The member `value` in a bound is encoded through the schema’s codec, exactly like a member passed to `zadd`. Lex ranges assume **all scores are equal**. This is standard Redis `BYLEX` behavior, and results are undefined when scores differ. Because Redis rejects `WITHSCORES` alongside `BYLEX`, you cannot combine `byLex: true` with `withScores: true`; use `zrange` with `{ byScore: true, withScores: true }` when you need scores back.

Raw Redis equivalent:

```ts
await nodeRedis.sendCommand([
  "ZRANGE",
  "name-index:directory",
  "[ada",
  "[ben",
  "BYLEX"
]);
```

# Streams

> Use streams for append-only event logs with ordered, ID-addressable entries.

Use streams for append-only event logs with ordered, ID-addressable entries.

## Define A Stream

[Section titled “Define A Stream”](#define-a-stream)

```ts
import { number, stream, string } from "benni/schema";


export const activity = stream("activity", {
  action: string(),
  points: number()
});
```

## Add Entries

[Section titled “Add Entries”](#add-entries)

```ts
const entryId = await redis.stream(activity).xadd("42", {
  action: "login",
  points: 5
});
// "1720094400000-0"
```

`xadd` accepts options for the entry ID, stream creation, and trimming on write:

```ts
await redis.stream(activity).xadd(
  "42",
  { action: "login", points: 5 },
  {
    nomkstream: true,
    maxLen: { count: 1000, approximate: true }
  }
);
```

With `nomkstream: true`, Redis skips missing streams (`NOMKSTREAM`) and `xadd` returns `null` instead of an entry ID. Any call that spells the flag out, including a computed `nomkstream: someBoolean`, types as `Promise<string | null>`; a call that leaves it off is `Promise<string>`. Passing an options object typed as `StreamAddOptions`, where the flag is optional, does not compile: the reply shape depends on the flag, so it has to be visible at the call site. `maxLen` trims while adding, with the same shape as `xtrim`’s `maxLen`. Pass `entryId` to set an explicit ID instead of the default `*`.

## Read Ranges

[Section titled “Read Ranges”](#read-ranges)

```ts
const entries = await redis.stream(activity).xrange("42", { count: 10 });
//    ^? Array<{ id: string; value: Partial<{ action: string; points: number }> }>


const newest = await redis.stream(activity).xrevrange("42", { count: 10 });
```

`start` and `end` default to the full stream (`-` to `+`). Fields not declared in the schema are skipped.

## Entry Values Are Partial

[Section titled “Entry Values Are Partial”](#entry-values-are-partial)

Every read shape that carries a stream entry value (`xrange`, `xrevrange`, `xread`, and the consumer-group reads) types it as `Partial<...>`, so a field declared as `action: string()` reads back as `string | undefined` and needs a fallback:

```ts
for (const entry of await redis.stream(activity).xrange("42")) {
  const action = entry.value.action ?? "(unknown)";
  const points = entry.value.points ?? 0;
}
```

This is deliberate, and it is worth knowing that it is the **opposite** policy from hashes, because the two use the same declared-fields concept:

|                                               | Missing declared field                        |
| --------------------------------------------- | --------------------------------------------- |
| Stream entry (`xrange`, `xread`, group reads) | Reads as `undefined`; you supply the fallback |
| Hash whole-record read (`hget(id)`)           | Throws `PartialRecordError`                   |
| Hash tolerant read (`hgetall(id)`)            | Reads as `undefined` (also `Partial`)         |

The difference follows from who writes the key. A hash under `hash("user", …)` is a record your schema owns, so a declared field that has gone missing is a bug worth a loud `PartialRecordError` from `hget`, with `hgetall` as the explicit tolerant read for records that use per-field TTLs. See [Hashes](/benni/data-structures/hashes/).

A stream is an append-only log, and any producer can append to it: an older service, a `redis-cli XADD`, a version of your code that predates the field you just added to the schema. Entries already written are immutable, so a schema can never be retrofitted onto them. Typing entry values as complete records would be a claim about every past and future writer that Benni cannot check, so it stays a `Partial` and the fallback stays visible at the read.

The write side has no such doubt. `xadd` requires every declared field, so entries your own code appends are always complete.

A [consumer group](/benni/data-structures/consumer-groups/) re-reading its pending list goes one step further: there the whole `value` can be `null`, meaning the entry was deleted upstream and there is nothing left to decode.

## Read After An Entry ID

[Section titled “Read After An Entry ID”](#read-after-an-entry-id)

```ts
const next = await redis.stream(activity).xread("42", "1720094400000-0", {
  count: 100
});
```

`xread` returns entries newer than the given entry ID, or an empty array when there is nothing new. Use `"0"` to read from the beginning.

## Trim

[Section titled “Trim”](#trim)

```ts
await redis.stream(activity).xtrim("42", { maxLen: { count: 1000, approximate: true } });
await redis.stream(activity).xtrim("42", { minId: { value: "1720094400000-0" } });
```

Both return the number of removed entries. `approximate: true` lets Redis trim in whole macro nodes, which is faster. `{ maxLen: { count: 0 } }` empties the stream but keeps the key, so its consumer groups and their pending lists survive; `del` deletes the groups along with the stream.

## Remove, Count, Delete

[Section titled “Remove, Count, Delete”](#remove-count-delete)

```ts
await redis.stream(activity).xdel("42", ["1720094400000-0"]);
const size = await redis.stream(activity).xlen("42");
await redis.stream(activity).del("42");
```

## Raw Redis Equivalent

[Section titled “Raw Redis Equivalent”](#raw-redis-equivalent)

```ts
await nodeRedis.xAdd("activity:42", "*", {
  action: "login",
  points: "5"
});
```

For at-least-once delivery across many workers, use [consumer groups](/benni/data-structures/consumer-groups/) (`XGROUP`, `XREADGROUP`, `XACK`) via `redis.stream(activity).group(name)`. Use `xread` for single-consumer polling. To block a worker until an entry arrives, [`xread` with a `timeoutSeconds`](/benni/advanced/blocking-operations/) and the blocking group read run on a [session](/benni/advanced/sessions/).

Use streams for activity feeds, audit logs, and event pipelines where entries need stable IDs and time ordering.

# Examples

> Copy-pasteable examples for the schema-first Benni API, one data structure at a time.

Every example on this page uses the same shape: declare schemas once, bind a client once, then reach each data structure through the bound handle.

schema.ts

```ts
import {
  bitmap,
  channel,
  geo,
  hash,
  hll,
  json,
  kv,
  list,
  number,
  pattern,
  script,
  set,
  stream,
  string,
  zset
} from "benni/schema";


type UserProfile = {
  name: string;
  score: number;
};


export const profiles = kv("profile", json<UserProfile>());
export const counters = kv("counter", number());
export const texts = kv("text", string());


export const users = hash("user", {
  name: string(),
  score: number()
});


export const roles = set("roles", string());
export const jobs = list("jobs", json<{ id: string; kind: "email" | "report" }>());
export const leaderboard = zset("leaderboard", string());


export const events = stream("events", {
  type: string(),
  userId: string()
});


export const activity = bitmap("activity");
export const cities = geo("cities", string());
export const visitors = hll("visitors", string());


export const userEvents = channel(
  "events:user",
  json<{ id: string; action: "created" | "deleted" }>()
);
export const userEventPattern = pattern(
  "events:user:*",
  json<{ id: string; action: string }>()
);


export const incrementBy = script("increment-by", {
  keys: ["counter"],
  args: { amount: number() },
  returns: number(),
  lua: "return redis.call('INCRBY', KEYS[1], ARGV[1])"
});
```

redis.ts

```ts
import { benni } from "benni";
import { node } from "benni/node";
import * as schema from "./schema";


const client = await node({
  url: process.env.REDIS_URL ?? "redis://127.0.0.1:6379"
});


export const redis = benni(client, { schema });
```

The sections below assume these two files. The lower-level building blocks the client is made of (`defineKeyspace`, `createHashStore`, …) live under `benni/core` for adapter authors and advanced integrations; see the [API overview](/benni/api/overview/).

## Typed JSON Key-Value

[Section titled “Typed JSON Key-Value”](#typed-json-key-value)

```ts
import { profiles } from "./schema";


await redis.kv(profiles).set("42", { name: "Ada", score: 10 }, { ttlSeconds: 60 });


const profile = await redis.kv(profiles).get("42");
// profile is UserProfile | null


await redis.kv(profiles).mset([
  ["43", { name: "Grace", score: 12 }],
  ["44", { name: "Linus", score: 8 }]
]);


const many = await redis.kv(profiles).mget(["42", "43", "missing"]);
// many is Array<UserProfile | null>


await redis.kv(profiles).del("42");
```

## Known IDs For Autocomplete

[Section titled “Known IDs For Autocomplete”](#known-ids-for-autocomplete)

When IDs are known at compile time, pass them into the schema. Editors then autocomplete IDs such as `"test1"` and full key strings such as `"demo:test1"`.

```ts
import { type RedisKey } from "benni";
import { kv, string } from "benni/schema";


const demos = kv("demo", string(), {
  ids: ["test1", "test2"]
});


await redis.kv(demos).set("test1", "value");


const key = redis.kv(demos).key("test1");
// key is "demo:test1"


type DemoKey = RedisKey<"demo", "test1" | "test2">;
// DemoKey is "demo:test1" | "demo:test2"
```

If IDs come from users, databases, or Redis itself, leave `ids` out and Benni accepts normal `string | number | bigint` IDs.

## Integer Counter

[Section titled “Integer Counter”](#integer-counter)

```ts
import { counters } from "./schema";


await redis.counter(counters).incr("page-views");
await redis.counter(counters).incrby("page-views", 5);
await redis.counter(counters).decr("page-views");
```

## String Commands

[Section titled “String Commands”](#string-commands)

`redis.string()` exposes the Redis string commands that only make sense for plain string values.

```ts
import { texts } from "./schema";


await redis.string(texts).append("welcome", "hello");
await redis.string(texts).append("welcome", " world");


const firstWord = await redis.string(texts).getrange("welcome", 0, 4);
const length = await redis.string(texts).strlen("welcome");
const value = await redis.string(texts).getex("welcome", 60);
```

## Typed Hash

[Section titled “Typed Hash”](#typed-hash)

```ts
import { users } from "./schema";


await redis.hash(users).hset("42", { name: "Ada", score: 10 }, { ttlSeconds: 300 });


const user = await redis.hash(users).hget("42");
// user is { name: string; score: number } | null


await redis.hash(users).hset("42", "name", "Grace");
const score = await redis.hash(users).hincrby("42", "score", 1);
const hasName = await redis.hash(users).hexists("42", "name");


await redis.hash(users).del("42");
```

## Typed Set

[Section titled “Typed Set”](#typed-set)

```ts
import { roles } from "./schema";


await redis.set(roles).sadd("user:42", ["admin", "editor"]);


const isAdmin = await redis.set(roles).sismember("user:42", "admin");
const allRoles = await redis.set(roles).smembers("user:42");


await redis.set(roles).srem("user:42", ["editor"]);
await redis.set(roles).del("user:42");
```

## Typed List

[Section titled “Typed List”](#typed-list)

```ts
import { jobs } from "./schema";


await redis.list(jobs).rpush("pending", [
  { id: "job-1", kind: "email" },
  { id: "job-2", kind: "report" }
]);


const nextJob = await redis.list(jobs).lpop("pending");
// nextJob is { id: string; kind: "email" | "report" } | null


const remaining = await redis.list(jobs).lrange("pending", 0, -1);
// remaining is Array<{ id: string; kind: "email" | "report" }>


await redis.list(jobs).del("pending");
```

## Typed Sorted Set

[Section titled “Typed Sorted Set”](#typed-sorted-set)

```ts
import { leaderboard } from "./schema";


await redis.zset(leaderboard).zadd("daily", [
  { member: "alice", score: 10 },
  { member: "bob", score: 20 }
]);


const top = await redis.zset(leaderboard).zrange("daily", {
  start: 0,
  stop: -1,
  withScores: true
});
// top is Array<{ readonly member: string; readonly score: number }>


await redis.zset(leaderboard).zincrby("daily", 5, "alice");
const aliceScore = await redis.zset(leaderboard).zscore("daily", "alice");


await redis.zset(leaderboard).del("daily");
```

## Typed Stream

[Section titled “Typed Stream”](#typed-stream)

```ts
import { events } from "./schema";


const entryId = await redis.stream(events).xadd("audit", {
  type: "login",
  userId: "42"
});


const latest = await redis.stream(events).xread("audit", "0-0", { count: 10 });
const history = await redis.stream(events).xrange("audit", { count: 10 });


await redis.stream(events).del("audit");
```

An entry is `{ id, value }`, and `value` is a `Partial` of the declared fields, because a stream entry can legally carry any subset of them. Read fields off `value`, with a fallback for the ones you require:

```ts
for (const entry of history) {
  const type = entry.value.type ?? "(unknown)";
  console.log(entry.id, type, entry.value.userId);
  //                          ^? string | undefined
}
```

## Typed Bitmap

[Section titled “Typed Bitmap”](#typed-bitmap)

```ts
import { activity } from "./schema";


await redis.bitmap(activity).setbit("2026-07-04", 42, true);


const active = await redis.bitmap(activity).getbit("2026-07-04", 42);
const activeCount = await redis.bitmap(activity).bitcount("2026-07-04");


await redis.bitmap(activity).del("2026-07-04");
```

## Typed Geo

[Section titled “Typed Geo”](#typed-geo)

```ts
import { cities } from "./schema";


await redis.geo(cities).geoadd("europe", [
  { member: "Berlin", longitude: 13.405, latitude: 52.52 },
  { member: "Paris", longitude: 2.3522, latitude: 48.8566 }
]);


const nearby = await redis.geo(cities).geosearch("europe", {
  from: { longitude: 13.405, latitude: 52.52 },
  by: { radius: 1000, unit: "km" },
  withDistance: true,
  withCoordinates: true
});


await redis.geo(cities).del("europe");
```

## Typed HyperLogLog

[Section titled “Typed HyperLogLog”](#typed-hyperloglog)

```ts
import { visitors } from "./schema";


await redis.hll(visitors).pfadd("today", ["user:1", "user:2", "user:1"]);


const approximateVisitors = await redis.hll(visitors).pfcount("today");


await redis.hll(visitors).del("today");
```

## Cursor Scans

[Section titled “Cursor Scans”](#cursor-scans)

```ts
import { leaderboard, profiles } from "./schema";


for await (const key of redis.scan.kv(profiles, { count: 100 })) {
  // key is a Redis key matching profile:*
}


for await (const entry of redis.scan.zset(leaderboard, "daily")) {
  // entry is { member: string; score: number }
}
```

## Pub/Sub

[Section titled “Pub/Sub”](#pubsub)

Subscribing needs no extra setup: the first subscription leases one subscriber connection from the bound client and closes it again when the last subscription goes away.

```ts
const subscription = await redis.pubsub.channel(schema.userEvents).subscribe(
  (message) => {
    // message is { id: string; action: "created" | "deleted" }
    console.log(message);
  }
);


await redis.pubsub.channel(schema.userEvents).publish({
  id: "42",
  action: "created"
});


await subscription.unsubscribe();
```

Use a typed pattern when one handler should receive several matching channels:

```ts
const patternSubscription = await redis.pubsub
  .pattern(schema.userEventPattern)
  .subscribe((message, channel) => {
    // message is decoded; channel is the concrete channel name
  });


await patternSubscription.unsubscribe();
```

Or consume a subscription as an async iterator, which releases it when the loop ends:

```ts
const controller = new AbortController();


for await (const message of redis.pubsub
  .channel(schema.userEvents)
  .stream({ signal: controller.signal })) {
  console.log(message.action);
}
```

Publishing is one stateless `PUBLISH` on the bound client, so it works on every adapter including [`benni/upstash`](/benni/runtime/edge/). Subscribing needs a connection the adapter can hold, and `redis.pubsub.close()` drops every subscription at once. See [Pub/Sub](/benni/data-structures/pubsub/).

## Typed Transaction

[Section titled “Typed Transaction”](#typed-transaction)

`redis.multi()` builds a `MULTI`/`EXEC` transaction whose result is a position-typed tuple:

```ts
import { booleanNumberReply, okReply, stringOrNullReply } from "benni";


const [setResult, stored, exists] = await redis
  .multi()
  .add(["SET", "tx:key", "value"], okReply)
  .add(["GET", "tx:key"], stringOrNullReply)
  .add(["EXISTS", "tx:key"], booleanNumberReply)
  .exec();
```

For `WATCH`-based optimistic transactions, see [Optimistic Transactions](/benni/advanced/optimistic-transactions/).

## Typed Lua Script

[Section titled “Typed Lua Script”](#typed-lua-script)

The `script()` schema names its keys and types its args; the first run loads the script and later runs send cached `EVALSHA`:

```ts
import { incrementBy } from "./schema";


const value = await redis.script(incrementBy).run({
  keys: { counter: "script:counter" },
  args: { amount: 5 }
});
// value is number
```

## Raw Command Fallback

[Section titled “Raw Command Fallback”](#raw-command-fallback)

Use raw commands when a typed helper does not exist yet.

```ts
const reply = await redis.raw.send(["SET", "raw:key", "value"]);
if (reply !== "OK") {
  throw new TypeError("SET failed");
}


const value = await redis.raw.send(["GET", "raw:key"]);
```

## Test With A Fake Client

[Section titled “Test With A Fake Client”](#test-with-a-fake-client)

The `RedisClient` contract is three required methods: `send`, `pipeline`, and `close` (`transaction` and `session` are optional), so unit tests can drive the whole typed API with a scripted fake:

```ts
import {
  benni,
  type RedisClient,
  type RedisCommand,
  type RedisReply
} from "benni";
import { json, kv } from "benni/schema";


function fakeClient(commands: RedisCommand[], replies: RedisReply[]): RedisClient {
  return {
    async send(command) {
      commands.push(command);
      const reply = replies.shift();
      if (reply === undefined) throw new Error("No fake Redis reply queued");
      return reply;
    },
    async pipeline(pipelineCommands) {
      commands.push(...pipelineCommands);
      return replies.splice(0, pipelineCommands.length);
    },
    async close() {}
  };
}


const commands: RedisCommand[] = [];
const profiles = kv("user", json<{ name: string }>());
const redis = benni(fakeClient(commands, ["OK", "{\"name\":\"Ada\"}"]), {
  schema: { profiles }
});


await redis.kv(profiles).set("42", { name: "Ada" });
const user = await redis.kv(profiles).get("42");


console.log(commands);
console.log(user);
```

# Installation

> Install Benni, pick the peer dependency your runtime needs, and check which Redis servers are supported.

Install Benni and the Redis client used by the Node.js adapter:

```sh
pnpm add benni redis
```

Benni is an ESM package. The primary imports are:

```ts
import { benni } from "benni";
import { node } from "benni/node";
import { hash, json, kv, number, string } from "benni/schema";
```

The Node.js adapter uses the [`redis`](https://www.npmjs.com/package/redis) package (node-redis, the officially recommended Node client). Benni declares it as an **optional peer dependency**, so you install it alongside Benni only when you use `benni/node`. That keeps the install tiny for other runtimes. Bun is supported through Bun’s built-in Redis client and needs no extra package:

```sh
bun add benni
```

Benni has four optional peer dependencies, all opt-in; install one only when you import the subpath that needs it:

| Peer                                               | Needed by       | Version   |
| -------------------------------------------------- | --------------- | --------- |
| [`redis`](https://www.npmjs.com/package/redis)     | `benni/node`    | `^6.1.0`  |
| [`ioredis`](https://www.npmjs.com/package/ioredis) | `benni/ioredis` | `^5.0.0`  |
| [`hono`](https://hono.dev)                         | `benni/hono`    | `>=4.0.0` |
| [`zod`](https://zod.dev)                           | `benni/zod`     | `^4.1.0`  |

Already running ioredis? [`benni/ioredis`](/benni/runtime/ioredis/) gives the same typed API and can adopt the client you already have, so adopting Benni is not a client migration:

```sh
pnpm add benni ioredis
```

Deno needs no separate adapter: it runs node-redis directly through npm compatibility, so Deno users import the Node adapter (`npm:benni/node`) and `npm:redis`. There is no `benni/deno` entrypoint: Benni ships one runtime-agnostic core plus thin client adapters, not per-runtime builds.

## Server Compatibility

[Section titled “Server Compatibility”](#server-compatibility)

CI runs the integration suite against Redis 8 and an Upstash-REST-compatible endpoint. The other rows are verified manually against the same suite:

| Server    | Coverage                                                                                                                                                                                                            |
| --------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Redis 8   | Full surface.                                                                                                                                                                                                       |
| Redis 7.4 | Everything except `hsetex`/`hgetex`/`hgetdel` (Redis 8 commands).                                                                                                                                                   |
| Redis 7.2 | Additionally no hash field TTLs (`hexpire`/`httl`/…, introduced in 7.4).                                                                                                                                            |
| Valkey 8  | Same profile as Redis 7.2 (Valkey forked pre-7.4).                                                                                                                                                                  |
| Dragonfly | The common surface works (kv, hashes, sets, lists, sorted sets, streams, geo, HyperLogLog, bitmaps, Pub/Sub, transactions, scripts); `LCS`, `GEOSEARCHSTORE`, and hash field TTLs are not implemented by Dragonfly. |

Everything else (streams, sorted sets, `lmpop`, `sintercard`, geo, bitfields) works from Redis 7.2 up.

Upstash and other serverless endpoints are covered through the [HTTP adapter](/benni/runtime/edge/).

For local development, run Redis with Docker:

```sh
pnpm redis:build
pnpm redis:run
```

Then point Benni at Redis:

```ts
const client = await node({
  url: process.env.REDIS_URL ?? "redis://127.0.0.1:6379"
});
```

# Introduction

> The end-to-end typed Redis client for TypeScript: one API across Node, Bun, Deno, and the edge.

Benni is the **end-to-end typed Redis client for TypeScript**: one typed API across Node, Bun, Deno, and the edge (Upstash/HTTP).

`node-redis` and `ioredis` are already typed, but only at the *command surface*: a reply comes back as Redis’s generic wire shape (`string | null`, `Record<string, string>`), so your type is gone the moment data crosses the Redis edge and you re-parse it by hand. Benni declares your data model once with typed codecs and carries those types from write to read. **They type the commands; Benni types your data.**

```ts
import { hash, number, string } from "benni/schema";


export const users = hash("user", {
  name: string(),
  score: number()
});


await redis.hash(users).hset("42", {
  name: "Ada",
  score: 10
});


const user = await redis.hash(users).hget("42");
//    ^? { name: string; score: number } | null
```

Benni is a typed **client**, not an ORM. It does not create tables, run migrations, or hide Redis. A Benni schema is a plain TypeScript description of a Redis key family, and raw access is always one call away.

The mental model is small:

```txt
Schema = how a Redis key family is shaped
Client = typed Redis access bound to your schemas
Raw = direct Redis access when that is clearer
```

Use Benni where schemas help. Use raw Redis where Redis itself is the clearest API.

# Philosophy

> The seven principles behind Benni's API: command names stay, schemas are plain values, one round trip whenever Redis allows one, and nothing is hidden or silent.

Benni is built on a small set of opinions. They explain most of the API, including the parts that look like omissions.

## Command Names Stay

[Section titled “Command Names Stay”](#command-names-stay)

Benni adds a type layer, not a query language. Methods are named after the Redis commands they run, so `hgetall` runs `HGETALL` and `zincrby` runs `ZINCRBY`. Every Redis doc page, every StackOverflow answer, and every `MONITOR` line still applies.

```ts
await redis.hash(users).hincrby("42", "score", 1); // HINCRBY user:42 score 1
```

Where Redis itself has folded old commands into a newer one, Benni follows Redis rather than the history: `zrange(id, { byScore: true, min, max })` mirrors modern `ZRANGE ... BYSCORE` instead of reviving the deprecated `ZRANGEBYSCORE`.

The argument list follows from the same rule, which is worth knowing before you reach for the docs on a method you have not called yet. A command with one fixed form takes its arguments positionally, in the command’s own order:

```ts
await redis.zset(board).zremrangebyscore("daily", "-inf", cutoff); // ZREMRANGEBYSCORE board:daily -inf <cutoff>
await redis.zset(board).zincrby("daily", 5, "alice");              // ZINCRBY board:daily 5 alice
```

A command with modifiers, or with several forms that give the same slot different meanings, takes one options object instead, so the call site reads like the modifiers it is choosing:

```ts
await redis.zset(board).zrange("daily", { start: 0, stop: 9, rev: true, withScores: true });
```

That is why `zrange` puts even its bounds in the object while `zremrangebyscore` does not: `ZRANGE`’s bounds are indexes, scores, or lex bounds depending on the modifier chosen alongside them, and `ZREMRANGEBYSCORE` only ever takes a score range.

This is deliberate altitude. An entity/document API would be further from Redis and would have to re-teach you everything you already know. Staying at command level means the only new thing to learn is the schema.

## Schemas Are Values, Not Migrations

[Section titled “Schemas Are Values, Not Migrations”](#schemas-are-values-not-migrations)

A schema is a plain TypeScript value describing the shape of a key family. Declaring one creates no keys, opens no connection, and runs nothing at import time.

```ts
export const users = hash("user", { name: string(), score: number() });
```

That is why Benni has no CLI, no codegen step, and no migration story: there is nothing to generate and nothing to migrate. Delete a schema and Redis is untouched. Adopt Benni for one key family and leave the other twelve on your raw client.

## One Round Trip Whenever Redis Allows One

[Section titled “One Round Trip Whenever Redis Allows One”](#one-round-trip-whenever-redis-allows-one)

Where a single command can do the job, Benni sends a single command:

* `hget(id)` with no field list is one `HMGET`, not a pipeline of `HGET`s. A pipeline can interleave with another client’s write and hand you a torn record.
* `hgetex` reads fields and slides the TTL in one command.
* `ratelimit.check()` is one atomic Lua evaluation: expire, count, admit.
* Calls with empty input resolve without touching the network at all.

This matters most on [the edge](/benni/runtime/edge/), where every command is an HTTP request and a saved round trip is tens of milliseconds, not microseconds. Benni will not quietly turn one of your calls into four.

## Nothing Is Hidden

[Section titled “Nothing Is Hidden”](#nothing-is-hidden)

There is no lazy loading, no identity map, no background fetching. A call sends the commands its name says it sends, and nothing else. When Benni cannot express something, it hands you the key instead of inventing an abstraction:

```ts
const key = redis.query.users.key("42"); // "user:42"
await redis.raw.send(["OBJECT", "ENCODING", key]);
```

## Nothing Is Silent

[Section titled “Nothing Is Silent”](#nothing-is-silent)

A layer that guesses is worse than no layer. When reality does not match the declared type, Benni throws where it happened:

* A reply that does not match the expected shape throws `ReplyShapeError`, carrying the raw reply on `.reply`. It never casts and moves on.
* Bad caller input throws `ValidationError` before anything is sent to Redis.
* An adapter that cannot do something says so. Calling `redis.session()` on the HTTP adapter throws, because the edge has no long-lived connection and pretending otherwise would only move the failure somewhere harder to debug.

`ReplyShapeError` and `ValidationError` both extend `TypeError`, so existing error handling keeps working.

## One API, Every Runtime, Honest About The Differences

[Section titled “One API, Every Runtime, Honest About The Differences”](#one-api-every-runtime-honest-about-the-differences)

The same typed API runs on Node, Bun, Deno, and the edge. Benni does not polyfill away real platform differences: sessions, blocking commands, `WATCH`, and Pub/Sub subscribing all need a persistent connection, so the HTTP adapter omits them rather than emulating them by polling behind your back. Pipelines and `MULTI`/`EXEC`, which HTTP *can* do, are there. See [what works and what doesn’t](/benni/runtime/edge/) on the edge.

## Batteries Only For What Is Easy To Get Wrong

[Section titled “Batteries Only For What Is Easy To Get Wrong”](#batteries-only-for-what-is-easy-to-get-wrong)

[Primitives](/benni/primitives/lock/) exist for the handful of patterns that are subtly hard: a correct distributed lock, an accurate sliding window, a stampede-proof cache, a job queue with leases and resumable output. These are worth shipping because most hand-rolled versions have a race in them.

Benni deliberately does not ship a secondary-index manager, a full-text search layer, or a general job framework. Those belong to Redis Search, or to BullMQ, or to your application, and shipping a mediocre version of each is how a client turns into an ORM.

# Quick Start

> Declare two Redis key families, bind a client, and read your own types back: the five-minute path from install to first typed round trip.

This example defines two Redis key families: one hash for user metadata and one JSON key-value store for full profiles.

First install Benni and the Node client (see [Installation](/benni/getting-started/installation/) for other runtimes):

```sh
pnpm add benni redis zod
```

[`zod`](https://zod.dev) is here because the JSON store below validates its reads. Any [Standard Schema](https://standardschema.dev) validator works (Zod, Valibot, ArkType) and Benni depends on none of them; the validator you already use is the one it will use.

## Define Schemas

[Section titled “Define Schemas”](#define-schemas)

schema.ts

```ts
import { hash, json, kv, number, string } from "benni/schema";
import { z } from "zod";


export const users = hash("user", {
  name: string(),
  score: number()
});


// json(validator) infers the value type from the validator and checks every
// read against it at runtime. This is the form to reach for.
const profile = z.object({
  name: z.string(),
  score: z.number()
});


export type UserProfile = z.infer<typeof profile>;
export const profiles = kv("profile", json(profile));
```

Schemas are plain TypeScript values. They do not create Redis keys or require migrations.

One thing to know about that JSON store before you build on it. `json(profile)` validates every read: if the stored JSON does not match, the read throws [`ReplyShapeError`](/benni/api/errors/#replyshapeerror) with the offending value attached rather than handing back something that lies about its type. There is a second form, `json<UserProfile>()`, which is the **unchecked escape hatch**:

```ts
// no validator, no runtime check: JSON.parse plus an assertion of the type
export type UserProfile = { name: string; score: number };
export const profiles = kv("profile", json<UserProfile>());
```

That is a pure cast. A value written by an older deploy, by another service, or by hand in `redis-cli` still types as a complete `UserProfile` even when fields are missing, and nothing throws. Reach for it only when you own every writer and the value has no shape worth checking. Prefer `json(validator)` everywhere else, and especially anywhere the data outlives the code that wrote it, which in Redis is most data. See [JSON values](/benni/data-structures/json-values/) for the full comparison.

## Create A Client

[Section titled “Create A Client”](#create-a-client)

redis.ts

```ts
import { benni } from "benni";
import { node } from "benni/node";
import * as schema from "./schema";


export const redis = benni({
  client: node({ url: process.env.REDIS_URL ?? "redis://127.0.0.1:6379" }),
  schema
});
```

`node()` returns a promise, and `benni()` takes it unawaited: the connection opens on the first command instead of at module scope, so this file needs no top-level `await` and drops into a Next.js route or an edge bundle unchanged. The trade is that a connection failure surfaces at the first command rather than at construction. Pass a client you already awaited when you would rather find out at startup:

```ts
const client = await node({ url: process.env.REDIS_URL });
export const redis = benni(client, { schema });
```

Both forms take the same options. `benni(client, options)` and `benni({ client, ...options })` are the same call.

To pass the bound handle around, register the schema module once and the exported `Benni` type is already the fully typed handle:

```ts
// redis.ts, next to the code above
declare module "benni" {
  interface Register {
    schema: typeof schema;
  }
}
```

```ts
import type { Benni } from "benni";


export function makeHandlers(redis: Benni) { /* ... */ }
```

Without the registration nothing breaks: `Benni` stays generic and `Benni<typeof schema>` still names the handle. Every accessor it exposes is listed in the [Benni Client reference](/benni/api/benni-client/).

The client owns a connection, so close it when your process or test finishes, otherwise Node never exits:

```ts
await redis.raw.close();
```

## Read And Write

[Section titled “Read And Write”](#read-and-write)

Because the schema module is bound to the client, reach each store by its export name through `redis.query`:

app.ts

```ts
import { redis } from "./redis";


await redis.query.users.hset("42", {
  name: "Ada",
  score: 10
});


const user = await redis.query.users.hget("42");
//    ^? { name: string; score: number } | null


await redis.query.profiles.set(
  "42",
  { name: "Ada", score: 10 },
  { ttlSeconds: 60 * 60 }
);
```

The explicit `redis.hash(schema)` accessors remain available and return the same store; see the [Schema Registry](/benni/core-concepts/schema-registry/).

## Drop To Redis

[Section titled “Drop To Redis”](#drop-to-redis)

```ts
const userKey = redis.query.users.key("42");
// "user:42"


const pong = await redis.raw.send(["PING"]);
```

The typed API handles repeated app patterns. The raw client stays available for commands that are not typed yet or when direct Redis is simpler.

# Why Benni?

> Why add a typed schema layer when node-redis and ioredis are already typed? Because they type the commands, not your data.

Redis is often used like this, here with raw `node-redis`:

```ts
await nodeRedis.hSet(`user:${id}`, {
  name: user.name,
  score: String(user.score)
});


const raw = await nodeRedis.hGetAll(`user:${id}`);
const loadedUser = {
  name: raw.name,
  score: Number(raw.score)
};
```

`node-redis` even types that `hGetAll` call, as `Record<string, string>`. The types are *present*, but they describe Redis’s wire shape, not your data: `score` comes back a `string`, and you coerce it by hand. This works, but over time it creates problems:

* Key names are spread across the codebase.
* Values are manually serialized and parsed.
* Return types are not obvious.
* Data structures are implicit.
* Refactoring is risky.
* Raw Redis commands are powerful but easy to misuse.

Benni keeps Redis explicit, but adds a typed schema layer:

```ts
export const users = hash("user", {
  name: string(),
  score: number()
});


await redis.hash(users).hset(id, {
  name: "Ada",
  score: 10
});


const user = await redis.hash(users).hget(id);
```

You still use Redis. You still understand what happens. You just stop scattering strings and parsers across your app.

## Benni And Redis Clients

[Section titled “Benni And Redis Clients”](#benni-and-redis-clients)

Benni is not a replacement for Redis. It is a typed layer on top of a Redis client: the client types the commands, Benni types your data.

| Feature                  | node-redis / ioredis | Benni                            |
| ------------------------ | -------------------- | -------------------------------- |
| Raw Redis commands       | Yes                  | Yes                              |
| Command-level types      | Yes (wire shape)     | Yes                              |
| Typed schemas            | Manual               | Yes                              |
| Typed hash / JSON values | Manual               | Yes                              |
| Key prefixes             | Manual               | Schema-based                     |
| Runtime reach            | Node only            | Node, Bun, Deno, edge/serverless |
| Escape hatch             | Native               | `redis.raw`                      |

Use a raw client when you want direct command access everywhere. Use Benni when your application has repeated Redis data patterns and you want your declared types to survive the round-trip, safer keys, and better refactoring, across every runtime.

## What It Measures Out To

[Section titled “What It Measures Out To”](#what-it-measures-out-to)

Three apps built twice against Redis 8, once through Benni and once through raw `node-redis`, feature for feature. Lines of implementation code, blank lines excluded:

| App                                                                           | Benni | Raw |
| ----------------------------------------------------------------------------- | ----- | --- |
| URL shortener (hash, counter, sorted set, stream, cache, rate limit)          | 97    | 171 |
| AI generation service (`queue`: resumable stream, cancel, retries)            | 45    | 437 |
| Realtime presence and payouts (sessions, leaderboard, pub/sub, `WATCH`, lock) | 103   | 197 |

Plain typed reads and writes come out about even. What Benni saves is the code around them: the raw column carries a sliding-window limiter, a read-through cache with single-flight, a token-fenced lock, and a queue with heartbeat leases, six hand-written Lua scripts in total. Reach for BullMQ and a limiter package instead and the counts converge again, at the price of several more dependencies that still hand your data back as `string | null`.

Nine ordinary Redis bugs planted in both versions (a typo’d hash field, a wrong value type, a missing required field, a read of an undeclared field, a nullable read treated as non-null, a counter reply used as a string, the wrong store kind, an undeclared event shape published to a typed channel, and a number member in a string-member sorted set) were **nine compile errors through Benni and four through the raw version**, which had a hand-written typed edge of its own. The five it missed were the quiet ones: the typo added a second field rather than replacing one, a date string in a numeric slot read back as `NaN`, a partial write left a partial record, and a field nobody writes read as `undefined`. Only the wrong store kind threw.

## What The Types Cost

[Section titled “What The Types Cost”](#what-the-types-cost)

Nothing you can measure at runtime: 2,000 sequential ops against a local Redis, seven interleaved reps, medians of 166 ms for Benni’s `hset` against 162 ms for `node-redis`’s `hSet`, and 171 ms against 161 ms for the read. Three to six percent, inside the run-to-run spread, on a loopback with no real round trip to hide behind.

At compile time it is cheaper than not using it. The same three apps under `tsc --extendedDiagnostics`:

|                           | Types  | Instantiations | Check time |
| ------------------------- | ------ | -------------- | ---------- |
| Benni versions            | 8,766  | 13,774         | 0.15s      |
| Raw `node-redis` versions | 33,695 | 198,061        | 0.54s      |

A schema layer sounds like something that slows an editor down. `node-redis`’s own command generics cost roughly 14 times the type instantiations that Benni’s typed surface does, so in practice the schema layer is the cheap part.

# Hono

> Rate limiting, response caching, and sessions as Hono middleware: one stack that runs on Node, Bun, Deno, and Cloudflare Workers.

`benni/hono` packages the [primitives](/benni/primitives/ratelimit/) as drop-in [Hono](https://hono.dev) middleware: rate limiting, response caching, and sessions. Because they take any `RedisClient`, the same middleware stack runs everywhere Hono does: Node, Bun, Deno, and Cloudflare Workers. On Workers, pair it with [`benni/upstash`](/benni/runtime/edge/).

```ts
import { Hono } from "hono";
import { upstash } from "benni/upstash";
import { ratelimit } from "benni/hono";


const client = upstash({
  url: process.env.UPSTASH_REDIS_REST_URL as string,
  token: process.env.UPSTASH_REDIS_REST_TOKEN as string
});


const app = new Hono();
app.use(
  "*",
  ratelimit({ client, limit: 100, windowMs: 60_000, key: (c) => c.get("userId") })
);
```

Every middleware accepts `client` as a `RedisClient`, a `Promise<RedisClient>`, or a `() => Promise<RedisClient>` factory, awaited once on first use and cached, so lazy connection setups just work.

## Rate limiting

[Section titled “Rate limiting”](#rate-limiting)

Sliding-window rate limiting, one atomic Lua round trip per request, the [`ratelimit` primitive](/benni/primitives/ratelimit/) behind a middleware. Allowed requests carry `X-RateLimit-Limit`, `X-RateLimit-Remaining`, and `X-RateLimit-Reset` (epoch seconds); denied requests get a JSON `429` with `Retry-After`.

```ts
import { Hono } from "hono";
import { ratelimit } from "benni/hono";


const app = new Hono();


app.use(
  "/api/*",
  ratelimit({
    client,
    limit: 100,
    windowMs: 60_000,
    key: (c) => c.req.header("x-api-key") ?? "anonymous"
  })
);
```

| Option     | Default       | Meaning                                                                          |
| ---------- | ------------- | -------------------------------------------------------------------------------- |
| `limit`    | -             | Maximum requests allowed within the window.                                      |
| `windowMs` | -             | Window length in milliseconds.                                                   |
| `prefix`   | `"ratelimit"` | Key namespace; keys are `<prefix>:<id>`.                                         |
| `key`      | -             | `(c) => string \| Promise<string>`, the rate-limit subject. Required: see below. |

`key` is required on purpose. There is no request property a limiter can trust without knowing the deployment: `x-forwarded-for` and `cf-connecting-ip` are set by the client on a direct deploy, and appended to rather than replaced by many proxies, so a default built on either would let a caller pick its own identity and nullify the limit by varying one header. Pass the value your deployment actually verifies: an authenticated user or API key id where you have one, otherwise the client address your platform exposes.

```ts
// Behind a proxy you control, which overwrites the header:
key: (c) => c.req.header("x-forwarded-for")?.split(",")[0]?.trim() ?? "anonymous"
// On Cloudflare Workers:
key: (c) => c.req.header("cf-connecting-ip") ?? "anonymous"
```

## Response caching

[Section titled “Response caching”](#response-caching)

Read-through caching for `GET`/`HEAD` responses (other methods pass through). On a hit the stored response is replayed with an `X-Benni-Cache: hit` header; on a miss the handler runs and successful responses are stored with `SET PX ttlMs`. **Every Redis failure fails open**: the request always runs.

The cache key is the full URL (`method:origin+path+query`) plus any `vary` headers, so it is a *shared* cache, and one app bound to several hostnames keeps one entry per host. A response is never stored when any of these hold:

* the request carried an `Authorization` or a `Range` header;
* the response is anything but a plain `200`;
* the handler or an inner middleware set a cookie on the response;
* the response carries a `no-store`, `no-cache`, or `private` `Cache-Control`, or a `Vary` naming a header you did not list in `vary` (`Vary: *` is never storable);
* the handler read or wrote the [`session`](#sessions) in any way, including reading `id` or `isNew` (`cache()` asks the session bag directly, so this holds whichever order the two middlewares are composed in).

That last rule is what keeps a per-user route safe. Note the cookie check alone would not: a returning visitor already has their `sid`, so `session()` emits no `Set-Cookie` and there is nothing for a cookie check to see. If a route varies by anything the cache cannot observe (a header you did not list in `vary` and the response does not declare in `Vary`, a value read straight from `c.req.header("Cookie")`), do not put `cache()` on it, or give it a `key` that includes the distinguishing value.

Stored entries keep `content-type`, `cache-control`, `vary`, `etag`, and `last-modified`, so a replay stays honest to the browser and to any CDN in front of you. Every other response header is dropped.

```ts
import { Hono } from "hono";
import { cache } from "benni/hono";


const app = new Hono();


app.get(
  "/report",
  cache({ client, ttlMs: 30_000, vary: ["accept-language"] }),
  async (c) => c.json(await buildExpensiveReport())
);
```

| Option   | Default                                | Meaning                                   |
| -------- | -------------------------------------- | ----------------------------------------- |
| `ttlMs`  | -                                      | Entry lifetime in milliseconds.           |
| `prefix` | `"hono-cache"`                         | Key namespace; keys are `<prefix>:<key>`. |
| `key`    | `method + ":" + origin + path + query` | `(c) => string`, the cache key.           |
| `vary`   | `[]`                                   | Header names folded into the key.         |

Bodies are stored as text (`{ status, headers, body }` JSON), so this is for text-ish responses (JSON, HTML), not streaming or binary payloads.

## Sessions

[Section titled “Sessions”](#sessions)

These are **cookie-backed user sessions**, not [`redis.session()`](/benni/advanced/sessions/) connection leases, so they work on Workers and other edge runtimes with `benni/upstash`.

Redis-backed sessions behind a `sid` cookie. The record is a JSON object under `<prefix>:<id>`, loaded before your handler and persisted after it, but only when the handler actually wrote something (`SET ... EX ttlSeconds`, so the TTL rolls on every write). New sessions get a `crypto.randomUUID()` id and a `Set-Cookie` header; `clear()` deletes the stored record.

A write back to a record this request loaded is conditional (`SET ... XX`), so a request that was already in flight when a concurrent `clear()` deleted the record cannot resurrect it. Beyond that, writes are last-writer-wins over the whole record: two overlapping requests that each set a different key can still lose one of the two writes.

Call `regenerate()` on login and on any privilege change. It mints a fresh id, carries the data over, deletes the record under the old id, and issues a new `Set-Cookie`. This is the defence against session fixation: without it, a session id an attacker planted in the victim’s browser stays the id the authenticated data lives under, and replaying that cookie is enough to become the victim.

```ts
import { Hono } from "hono";
import { getSession, session } from "benni/hono";


const app = new Hono();
app.use("*", session({ client, ttlSeconds: 86_400 }));


app.post("/login", async (c) => {
  const user = await authenticate(c);
  const bag = getSession(c);
  // Never keep the pre-login id once the session becomes privileged.
  bag.regenerate();
  bag.set("userId", user.id);
  return c.json({ ok: true });
});


app.get("/me", (c) => {
  const userId = getSession(c).get<string>("userId");
  return userId ? c.text(userId) : c.text("anonymous", 401);
});


app.post("/logout", (c) => {
  getSession(c).clear();
  return c.text("bye");
});
```

| Option       | Default                                                           | Meaning                                                |
| ------------ | ----------------------------------------------------------------- | ------------------------------------------------------ |
| `ttlSeconds` | `86400`                                                           | Session lifetime in seconds; refreshed on every write. |
| `prefix`     | `"hono-session"`                                                  | Key namespace; keys are `<prefix>:<id>`.               |
| `cookieName` | `"sid"`                                                           | Session cookie name.                                   |
| `cookie`     | `path: "/"`, `httpOnly: true`, `sameSite: "Lax"`, `secure: false` | Cookie attributes; enable `secure` in production.      |

Session values are `unknown` per key; `get<T>` is a convenience assertion, not a validation. The session is a convenience bag; codec-level typing belongs to your [Benni schemas](/benni/core-concepts/defining-schemas/).

## Putting it together

[Section titled “Putting it together”](#putting-it-together)

```ts
import { Hono } from "hono";
import { upstash } from "benni/upstash";
import { cache, getSession, ratelimit, session } from "benni/hono";


const client = upstash({
  url: process.env.UPSTASH_REDIS_REST_URL as string,
  token: process.env.UPSTASH_REDIS_REST_TOKEN as string
});


const app = new Hono();


app.use(
  "*",
  ratelimit({ client, limit: 100, windowMs: 60_000, key: (c) => c.get("userId") })
);
app.use("*", session({ client }));


app.get("/pricing", cache({ client, ttlMs: 60_000 }), (c) =>
  c.json({ plans: ["free", "pro"] })
);


app.post("/login", (c) => {
  const bag = getSession(c);
  bag.regenerate();
  bag.set("userId", "u1");
  return c.json({ ok: true });
});


export default app;
```

The same file deploys to a Node server, a Bun process, Deno Deploy, or a Cloudflare Worker; only the adapter changes.

# Next.js

> Redis-backed ISR caching and edge-ready rate limiting for Next.js: a custom cache handler and a middleware limiter in one import.

`benni/next` connects Next.js to Redis in the two places it matters: a **custom cache handler** so ISR/App-Router cache entries survive deploys and are shared across instances, and a **rate limiter** for middleware, route handlers, and Server Actions.

Both work over every adapter. On Vercel and other edge runtimes, pair them with [`benni/upstash`](/benni/runtime/edge/): middleware has no TCP, but the Upstash adapter needs nothing beyond `fetch`.

## Cache handler

[Section titled “Cache handler”](#cache-handler)

Next.js caches pages, route handler output, and `fetch` data in local files by default, so each instance has its own cache and a deploy wipes it. A custom cache handler moves that storage to Redis.

cache-handler.mjs

```ts
import { cacheHandler } from "benni/next";
import { upstash } from "benni/upstash";


export default cacheHandler({
  client: () =>
    upstash({
      url: process.env.UPSTASH_REDIS_REST_URL as string,
      token: process.env.UPSTASH_REDIS_REST_TOKEN as string
    })
});
```

next.config.ts

```ts
const nextConfig = {
  cacheHandler: require.resolve("./cache-handler.mjs"),
  cacheMaxMemorySize: 0 // disable the per-instance in-memory cache
};


export default nextConfig;
```

`cacheHandler(options)` returns a class; Next.js instantiates the module’s default export itself. Pass `client` as a lazy factory (as above) so no connection is opened when Next.js loads the module at build time; the factory is awaited once and cached.

| Option              | Default        | Meaning                                                              |
| ------------------- | -------------- | -------------------------------------------------------------------- |
| `client`            | -              | A `RedisClient`, a promise of one, or a lazy factory (awaited once). |
| `prefix`            | `"next-cache"` | Key namespace.                                                       |
| `defaultTtlSeconds` | -              | Safety-cap TTL for entries without a `revalidate` period.            |

The handler matches the cache-handler shape of Next.js 14.1+; the Next.js 15 `resetRequestCache()` hook is a no-op on this handler (it keeps no request-local state). Reads fail open: an entry that does not decode is treated as a miss, never an error.

### How tags map to Redis keys

[Section titled “How tags map to Redis keys”](#how-tags-map-to-redis-keys)

Each entry is stored as JSON under `<prefix>:entry:<key>`, with `SET ... EX <revalidate>` when the page declares a numeric `revalidate`, without a TTL when it opts out (`revalidate: false`), unless `defaultTtlSeconds` caps it. Each tag keeps a set of the keys written under it:

```plaintext
next-cache:entry:/blog          -> { value, lastModified, tags }
next-cache:tag:posts            -> SMEMBERS { "/blog", "/blog/post-1" }
```

A tag set is expired alongside the entries it names: every write extends the set to the entry’s TTL, never shortens it, and an entry that never expires makes the set permanent. So a tag set is reclaimed once its last member has gone, instead of growing for the life of the deployment.

`revalidateTag("posts")` is then one `SMEMBERS` per tag plus a chunked `DEL` of the matching entries, followed by an `SREM` of exactly the members it saw, with no scans. Only the tags Next.js passes on `set()` (`ctx.tags`) feed the index.

## Rate limiting

[Section titled “Rate limiting”](#rate-limiting)

`rateLimit(options)` wraps the [`ratelimit`](/benni/primitives/ratelimit/) primitive (an exact sliding window, one atomic Lua round trip per check) in a web-standard shape: give it a `Request`, get back `null` (allowed) or a finished `429 Response`.

middleware.ts

```ts
import { rateLimit } from "benni/next";
import { upstash } from "benni/upstash";


const limiter = rateLimit({
  client: () =>
    upstash({
      url: process.env.UPSTASH_REDIS_REST_URL as string,
      token: process.env.UPSTASH_REDIS_REST_TOKEN as string
    }),
  limit: 20,
  windowMs: 10_000,
  identify: (request) =>
    request.headers.get("x-forwarded-for")?.split(",")[0]?.trim() ?? "anonymous"
});


export async function middleware(request: Request) {
  const denied = await limiter(request);
  if (denied) return denied;
}


export const config = { matcher: "/api/:path*" };
```

The denial response carries `Retry-After` (seconds) plus `X-RateLimit-Limit`, `X-RateLimit-Remaining`, and `X-RateLimit-Reset` (epoch seconds).

`identify` is required on purpose. There is no request property a limiter can trust without knowing the deployment: on a self-hosted Next.js, or behind a proxy that appends rather than replaces, `x-forwarded-for` is attacker-controlled, so a default built on it would let a caller vary one header to bypass the limit and mint a fresh Redis key every time. The snippet above is the right form on a platform whose edge overwrites the header, such as Vercel. Better still is an identity you authenticated yourself:

```ts
const limiter = rateLimit({
  client,
  limit: 100,
  windowMs: 60_000,
  identify: (request) => request.headers.get("x-api-key") ?? "anonymous"
});
```

### Server Actions

[Section titled “Server Actions”](#server-actions)

A Server Action has no `Request`. The limiter also exposes `.check(identity)`, which returns the raw [`RatelimitResult`](/benni/primitives/ratelimit/):

```ts
"use server";


export async function submitComment(formData: FormData) {
  const { success, resetMs } = await limiter.check(await getUserId());
  if (!success) {
    return { error: "Too many comments. Try again shortly.", resetMs };
  }
  // ...
}
```

## Which adapter where

[Section titled “Which adapter where”](#which-adapter-where)

* **Edge middleware**: [`benni/upstash`](/benni/runtime/edge/). The edge runtime has no TCP sockets; the Upstash adapter speaks HTTP with zero dependencies.
* **Route handlers / Server Actions on Node, and self-hosted deploys**: [`benni/node`](/benni/runtime/node/) for pooled TCP connections; `benni/upstash` also works if you are already on Upstash.
* **The cache handler** runs wherever your Next.js server runs and only needs `send`/`pipeline`, so either adapter fits.

# Zod

> Bidirectional Zod codecs as Benni field codecs: writes validated with z.encode, reads validated with z.decode, and rich types (Date, bigint, URL) that genuinely round-trip.

Benni’s core already accepts any [Standard Schema](https://standardschema.dev) validator via [`json(schema)`](/benni/api/schema-builders/), but Standard Schema only defines *one* direction, so that validates **reads only**, and writes are a blind `JSON.stringify`. [Zod codecs](https://zod.dev/codecs) (Zod 4.1+) define both directions, and `benni/zod` runs them both:

* **Writes are validated.** A bad value throws `ValidationError` at the `set`, before anything is sent, not at some later read in another process.
* **Rich types round-trip.** `Date`, `bigint`, `URL`, custom classes, stored in their string / JSON-safe form, revived on read. With plain `json<T>()`, a `Date` field silently comes back as a `string`.

```ts
import * as z from "zod";
import { kv } from "benni/schema";
import { zodCodec, zodJson } from "benni/zod";


const isoDate = z.codec(z.iso.datetime(), z.date(), {
  decode: (iso) => new Date(iso),
  encode: (date) => date.toISOString()
});


const user = z.object({ name: z.string(), created: isoDate });
export const users = kv("user", zodJson(user));


await redis.kv(users).set("u1", { name: "ada", created: new Date() });
const found = await redis.kv(users).get("u1");
//    ^? { name: string; created: Date } | null (created is a real Date)
```

Zod is an **optional peer dependency** (`zod@^4.1.0`); only the `benni/zod` subpath imports it. The adapter is built against `zod/v4/core`, so schemas from both `zod` and `zod/mini` work.

## `zodCodec(schema)`: string-stored fields

[Section titled “zodCodec(schema): string-stored fields”](#zodcodecschema-string-stored-fields)

Takes any Zod schema or codec whose *encoded* (input) side is a string and returns a Benni `Codec`. Use it anywhere a codec is accepted: kv values, hash fields, list items, set and sorted-set members, stream fields, pub/sub messages.

```ts
import { hash, string } from "benni/schema";
import { zodCodec } from "benni/zod";


export const sessions = hash("session", {
  userId: string(),
  expiresAt: zodCodec(isoDate) // Date in your code, ISO string in Redis
});
```

A plain string schema works too: `zodCodec(z.email())` stores the string as-is and validates it in both directions. Passing a schema whose encoded side isn’t a string (`z.number()`, `z.date()`, …) is a compile error.

## `zodJson(schema)`: JSON-stored values

[Section titled “zodJson(schema): JSON-stored values”](#zodjsonschema-json-stored-values)

A stronger [`json(schema)`](/benni/data-structures/json-values/): writes run `z.encode` (validated, codec fields converted to their JSON-safe form) then `JSON.stringify`; reads run `JSON.parse` then `z.decode` (validated, codec fields revived).

Fields that aren’t JSON-safe need a codec to a JSON-safe form, like `isoDate` above. A bare `z.date()` field would stringify on write but fail loudly on read, which is still better than `json<T>()`’s silent type lie, but a codec is the actual fix.

## Useful codecs

[Section titled “Useful codecs”](#useful-codecs)

Zod doesn’t ship codec presets; its [codecs page](https://zod.dev/codecs) maintains copy-paste implementations for the common ones: `Date` ↔ ISO string (above), plus:

```ts
const bigintString = z.codec(z.string().regex(/^-?\d+$/), z.bigint(), {
  decode: (s) => BigInt(s),
  encode: (b) => b.toString()
});


const urlString = z.codec(z.url(), z.instanceof(URL), {
  decode: (s) => new URL(s),
  encode: (url) => url.href
});
```

## Errors

[Section titled “Errors”](#errors)

The adapter maps into Benni’s unified error classes:

* Encode failures throw [`ValidationError`](/benni/api/schema-builders/), a caller mistake; nothing is sent to Redis.
* Decode failures throw `ReplyShapeError` with the stored string attached as `.reply`, and the message names the failing paths (`created: Invalid ISO datetime`).
* Async schemas (`.refine(async …)`) can’t run in a synchronous codec; both directions throw `ValidationError` telling you so. If such a refinement *rejects* instead of just failing, zod discards that promise internally, so the rejection also surfaces as an unhandled rejection Benni cannot claim. Keep async work out of the schema.
* `zodJson` refuses values JSON cannot carry faithfully: `NaN`, `Infinity`, `BigInt`, and circular structures all throw `ValidationError` before the write, exactly as the plain `json()` codec does. A non-finite number would otherwise be stored as `null` and read back as if the key were missing.
* `zodCodec` needs a schema whose encoded side really is a string. `z.any()` satisfies the type constraint without checking anything, so a non-string encode result throws `ValidationError` rather than reaching Redis as `[object Object]`.

# AI Apps

> Redis recipes for LLM apps: chat memory on streams, token budgets, response caching by prompt hash, and resumable generations, fully typed.

LLM backends are state-heavy: conversation history, per-user budgets, response caches, in-flight generation tracking. Redis is the natural home for all of it, and every recipe on this page is fully typed end to end. They all assume the client binding from the [Quick Start](/benni/getting-started/quick-start/), and every one of them (streams, counters, and all three primitives) runs unchanged on the [edge adapter](/benni/runtime/edge/).

## Chat Memory On A Stream

[Section titled “Chat Memory On A Stream”](#chat-memory-on-a-stream)

A conversation is an append-only log, which is exactly what a Redis stream is: ordered entries with stable IDs, one stream per conversation id. Trimming on write with `maxLen` keeps memory bounded per conversation: no cron job, no unbounded keys.

```ts
import { enumOf, stream, string } from "benni/schema";


export const chat = stream("chat", {
  role: enumOf(["user", "assistant", "system"]),
  content: string()
});
```

Append each turn, trimming to the last \~200 as you write:

```ts
await redis.stream(chat).xadd(
  conversationId,
  { role: "user", content },
  { maxLen: { count: 200, approximate: true } }
);
```

Load the prompt window with `xrevrange` (newest first, so `count` caps the read), reverse back into chronological order, and map entries straight onto an AI SDK `messages` array:

```ts
import { generateText } from "ai";


const recent = await redis
  .stream(chat)
  .xrevrange(conversationId, { count: 20 });


const messages = recent
  .reverse()
  .flatMap(({ value }) =>
    value.role !== undefined && value.content !== undefined
      ? [{ role: value.role, content: value.content }]
      : []
  );
//    ^? Array<{ role: "user" | "assistant" | "system"; content: string }>


const { text } = await generateText({ model: openai("gpt-4o-mini"), messages });
```

Entry values are `Partial` because Redis does not enforce stream entry shapes; the `flatMap` guard both narrows the types and skips malformed entries. For abandoned conversations, arm a per-conversation TTL after writing:

```ts
await redis.stream(chat).expire(conversationId, 60 * 60 * 24 * 30); // 30 days
```

In production, size `maxLen` to your model’s context budget, not your UI’s history length, and remember `approximate: true` trims in whole macro nodes, so the stream may briefly hold a few more entries than the count. See [Streams](/benni/data-structures/streams/) for the full store API.

## Token Budgets For LLM Endpoints

[Section titled “Token Budgets For LLM Endpoints”](#token-budgets-for-llm-endpoints)

Requests-per-minute alone does not protect an LLM endpoint: twenty small requests and twenty 100k-token requests cost wildly different amounts. Layer two checks: a sliding-window request limit via the [`ratelimit` primitive](/benni/primitives/ratelimit/), and a daily token budget in a plain counter keyed by user and date.

```ts
import { kv, number } from "benni/schema";
import { ratelimit } from "benni/primitives";


export const dailyTokens = kv("tokens", number());


const limiter = ratelimit(client, { limit: 20, windowMs: 60_000, prefix: "llm" });
const DAILY_TOKEN_BUDGET = 200_000;


export async function POST(request: Request): Promise<Response> {
  const { userId, prompt } = (await request.json()) as {
    userId: string;
    prompt: string;
  };


  // Layer 1: requests per minute, sliding window.
  const { success, resetMs } = await limiter.check(userId);
  if (!success) {
    return new Response("Too Many Requests", {
      status: 429,
      headers: { "Retry-After": String(Math.ceil((resetMs - Date.now()) / 1000)) }
    });
  }


  // Layer 2: tokens per day.
  const day = new Date().toISOString().slice(0, 10); // "2026-07-12"
  const budgetId = `${userId}:${day}`;
  const used = (await redis.kv(dailyTokens).get(budgetId)) ?? 0;
  if (used >= DAILY_TOKEN_BUDGET) {
    return new Response("Daily token budget exhausted", { status: 429 });
  }


  const result = await generateText({ model: openai("gpt-4o-mini"), prompt });


  // Record usage; the increment that creates the key arms its TTL.
  const total = await redis
    .counter(dailyTokens)
    .incrby(budgetId, result.usage.totalTokens);
  if (total === result.usage.totalTokens) {
    await redis.counter(dailyTokens).expire(budgetId, 60 * 60 * 24 * 2);
  }


  return Response.json({ text: result.text, tokensUsedToday: total });
}
```

The sliding window matters here: a fixed window resets all at once, so a caller can burn a full limit at 11:59 and again at 12:00, a 2x burst exactly when abuse scripts hammer the boundary. The sliding-window log admits at most `limit` requests in *any* 60-second span. The date in the budget key does the real expiry work; the TTL is just cleanup, so its precision never affects correctness. Note the budget check runs *before* the call but records *after*, so concurrent requests can overshoot the budget by one generation each, which is the usual, acceptable trade for not holding a reservation across a model call.

## Cache Responses By Prompt Hash

[Section titled “Cache Responses By Prompt Hash”](#cache-responses-by-prompt-hash)

Identical prompts arrive in bursts (the same trending question, the same retried classification), and every duplicate model call costs real money and seconds of latency. The [`cache` primitive](/benni/primitives/cache/) is single-flight: on a miss, exactly one caller runs the loader while concurrent identical prompts wait for the filled value, so a burst of the same prompt becomes one model call. Key it by a SHA-256 over everything that determines the output: model, system prompt, and user input.

```ts
import { cache } from "benni/primitives";


const responses = cache<string>(client, {
  ttlMs: 24 * 60 * 60 * 1000,
  prefix: "llm-response"
});


// Web Crypto: works on Node, Bun, Deno, and every edge runtime.
async function promptHash(model: string, system: string, input: string): Promise<string> {
  const bytes = new TextEncoder().encode(JSON.stringify([model, system, input]));
  const digest = await crypto.subtle.digest("SHA-256", bytes);
  return [...new Uint8Array(digest)]
    .map((byte) => byte.toString(16).padStart(2, "0"))
    .join("");
}


const id = await promptHash("gpt-4o-mini", SYSTEM_PROMPT, userInput);
const text = await responses.get(id, async () => {
  const result = await generateText({
    model: openai("gpt-4o-mini"),
    system: SYSTEM_PROMPT,
    prompt: userInput
  });
  return result.text;
});
```

Only cache calls that are deterministic enough to reuse: classification, extraction, and RAG-style answers at low temperature, not open-ended chat. Anything that changes the output belongs in the hash (temperature, retrieval context, output schema version), and set `lockTtlMs` above your slowest generation so waiters don’t fail open into a duplicate model call mid-load.

## Resumable Generation State

[Section titled “Resumable Generation State”](#resumable-generation-state)

Streamed responses die with the connection: a mobile client drops mid-generation and has to start (and you have to pay) from scratch. Append chunks to a stream as the model produces them, and a reconnecting client replays everything after the last entry ID it saw.

If the generation should also survive the *server* going away (a deploy, a crash, a serverless invocation timing out), reach for the [`queue` primitive](/benni/primitives/queue/) instead. It runs the generation on a worker, gives every job this same resumable stream, and adds the lifecycle this recipe leaves to you: retries, cancellation, and dead-lettering.

```ts
export const generation = stream("generation", { chunk: string() });


// Producer: append chunks as the model streams them.
for await (const delta of textStream) {
  await redis.stream(generation).xadd(generationId, { chunk: delta });
}
await redis.stream(generation).expire(generationId, 60 * 60);


// Reconnecting client: replay everything after the last seen entry ID.
const missed = await redis.stream(generation).xread(generationId, lastSeenEntryId);
```

`xread` returns entries newer than the given ID (use `"0"` for a full replay). On a long-lived server, a [session](/benni/advanced/sessions/)’s blocking `xread` with `{ timeoutSeconds }` turns the replay loop into a live tail. Writing one entry per token is chatty, so batch a few chunks per `xadd` under load. See [Streams](/benni/data-structures/streams/) for ranges, trimming, and consumer groups.

Use a stream, not [Pub/Sub](/benni/data-structures/pubsub/), for anything a client might reconnect to: Pub/Sub is fire-and-forget, so chunks published while the client was offline are simply gone. Pub/Sub earns its place when you need *live fan-out* to several watchers of the same generation (a shared session, an ops dashboard), and each one only cares about what arrives from now on. That subscriber side needs a held connection, so it runs on Node or Bun rather than the edge, and `stream()` makes it a plain loop you can pipe into an SSE response:

```ts
import { channel, json } from "benni/schema";


export const generationFeed = channel("feed:generation", json<{ chunk: string }>());


// Long-lived server: one watcher per connected viewer.
const controller = new AbortController();
for await (const { chunk } of redis.pubsub
  .channel(generationFeed)
  .stream({ signal: controller.signal })) {
  writeSse(chunk);
}
```

Publishing is one stateless `PUBLISH`, so the producer side of that fan-out still runs anywhere: an edge handler can publish chunks that long-lived servers subscribe to.

## Deduplicate In-Flight Generations

[Section titled “Deduplicate In-Flight Generations”](#deduplicate-in-flight-generations)

Retries and double-clicks are the other way to pay twice for one answer. Wrap the generation in the [`lock` primitive](/benni/primitives/lock/) keyed by a client-supplied request ID: the first request generates, and any duplicate that arrives while it is running gets a `409` instead of a second model call.

For a queued generation, prefer the [`queue` primitive](/benni/primitives/queue/)’s `idempotencyKey`, which returns the *original job* rather than a `409`, so the duplicate request can watch or await the answer the first one is already producing.

```ts
import { lock, LockNotAcquiredError } from "benni/primitives";


const generating = lock(client, { ttlMs: 60_000, prefix: "generating" });


try {
  return await generating.run(requestId, async () => {
    const { text } = await generateText({ model: openai("gpt-4o-mini"), prompt });
    return Response.json({ text });
  });
} catch (error) {
  if (error instanceof LockNotAcquiredError) {
    return new Response("Generation already in flight", { status: 409 });
  }
  throw error;
}
```

Set `ttlMs` above your worst-case generation time (or `extend()` the handle for long jobs); if the holder crashes, the TTL frees the lock instead of deadlocking the request ID. The lock deduplicates *in-flight* work; pair it with the response cache above so a retry that lands *after* completion gets the finished answer instead of a 409.

## Works Everywhere

[Section titled “Works Everywhere”](#works-everywhere)

Everything on this page (streams, counters, `ratelimit`, `cache`, and `lock`) runs on the same typed API across Node, Bun, and Deno, and over [`benni/upstash`](/benni/runtime/edge/) on Cloudflare Workers, Vercel Edge, and Deno Deploy. The one exception is blocking stream reads, which need a persistent connection; the polling `xread` shown here works on every adapter.

# Caching

> Use a JSON key-value schema for cached responses.

Tip

For most apps, reach for the first-class [`cache` primitive](/benni/primitives/cache/), a read-through cache with stampede protection built in. This page shows the underlying key-value pattern if you want to roll your own.

Use a JSON key-value schema for cached responses.

```ts
import { json, kv } from "benni/schema";


type ProductSummary = {
  id: string;
  name: string;
  priceCents: number;
};


export const productCache = kv(
  "cache:product",
  json<ProductSummary>()
);
```

Read-through caching:

```ts
async function getProduct(id: string) {
  const cached = await redis.kv(productCache).get(id);
  if (cached) return cached;


  const product = await fetchProductFromDatabase(id);


  await redis.kv(productCache).set(id, product, {
    ttlSeconds: 60 * 5
  });


  return product;
}
```

Invalidate when the source of truth changes:

```ts
await redis.kv(productCache).del(id);
```

Raw Redis equivalent:

```ts
await nodeRedis.set(
  `cache:product:${id}`,
  JSON.stringify(product),
  { EX: 60 * 5 }
);
```

# Worked Examples

> These examples show Benni as application Redis code, not isolated method calls.

These examples show Benni as application Redis code, not isolated method calls.

## User Profiles

[Section titled “User Profiles”](#user-profiles)

```ts
import { hash, json, kv, number, string } from "benni/schema";


export const users = hash("user", {
  name: string(),
  score: number()
});


export const profiles = kv(
  "profile",
  json<{
    bio: string;
    links: string[];
  }>()
);


await redis.hash(users).hset("42", {
  name: "Ada",
  score: 10
});


await redis.kv(profiles).set("42", {
  bio: "First programmer",
  links: ["https://example.com"]
});
```

## Feature Flags

[Section titled “Feature Flags”](#feature-flags)

```ts
import { boolean, kv } from "benni/schema";


export const flags = kv("feature-flag", boolean());


await redis.kv(flags).set("new-dashboard", true);


if (await redis.kv(flags).get("new-dashboard")) {
  // enable the feature
}
```

## Raw Escape Hatch

[Section titled “Raw Escape Hatch”](#raw-escape-hatch)

```ts
const key = redis.hash(users).key("42");
const exists = await redis.raw.send(["EXISTS", key]);
```

For a copy-pasteable example of every data structure in turn, see [Examples](/benni/examples/). The lower-level store builders live under `benni/core`; see the [API Overview](/benni/api/overview/).

# Leaderboards

> Use a sorted set when Redis should rank members by score.

Use a sorted set when Redis should rank members by score.

```ts
import { zset, string } from "benni/schema";


export const leaderboards = zset("leaderboard", string());
```

Record scores:

```ts
await redis.zset(leaderboards).zadd("weekly", [
  { member: "user:42", score: 1200 },
  { member: "user:7", score: 950 }
]);
```

Increment a score:

```ts
await redis.zset(leaderboards).zincrby("weekly", 25, "user:42");
```

Read the top 10:

```ts
const top = await redis
  .zset(leaderboards)
  .zrange("weekly", { start: 0, stop: 9, rev: true });
```

Read scores with members:

```ts
const ranked = await redis
  .zset(leaderboards)
  .zrange("weekly", { start: 0, stop: 9, withScores: true });
```

Raw Redis equivalent:

```ts
await nodeRedis.zIncrBy("leaderboard:weekly", 25, "user:42");
const top = await nodeRedis.zRange("leaderboard:weekly", 0, 9, {
  REV: true
});
```

# Rate Limiting From Scratch

> Cap requests per window with an atomic INCR + EXPIRE Lua script.

Tip

For most apps, reach for the first-class [`ratelimit` primitive](/benni/primitives/ratelimit/), a sliding-window limiter in one atomic call. This page shows the underlying fixed-window pattern if you want to roll your own.

Rate limiting caps how many actions a caller may take in a time window. The classic Redis approach is a **fixed window** counter: `INCR` a per-caller key and set its TTL to the window on the first hit. Doing both inside one `script()` keeps the increment and the expiry atomic, so a burst can never leave a counter without a TTL, the bug that turns a rate limiter into a permanent lockout.

## Define the limiter script

[Section titled “Define the limiter script”](#define-the-limiter-script)

```ts
import { number, script } from "benni/schema";


export const rateLimit = script("rate-limit", {
  keys: ["counter"],
  args: { windowSeconds: number() },
  returns: number(),
  lua: `
    local current = redis.call("INCR", KEYS[1])
    if current == 1 then
      redis.call("EXPIRE", KEYS[1], ARGV[1])
    end
    return current
  `
});
```

Add it to your bound `{ schema }` to reach it as `redis.query.rateLimit`, or call it directly with `redis.script(rateLimit)`.

## Check a limit

[Section titled “Check a limit”](#check-a-limit)

```ts
const WINDOW_SECONDS = 60;
const MAX_PER_WINDOW = 100;


async function allow(userId: string): Promise<boolean> {
  const count = await redis.script(rateLimit).run({
    keys: { counter: `ratelimit:${userId}` },
    args: { windowSeconds: WINDOW_SECONDS }
  });
  return count <= MAX_PER_WINDOW;
}
```

`INCR` returns the running count for the window. The first call in a window creates the key and arms its TTL; the window resets when the key expires, so there is nothing to clean up.

## Use it in a request handler

[Section titled “Use it in a request handler”](#use-it-in-a-request-handler)

```ts
async function handle(request: Request, userId: string): Promise<Response> {
  if (!(await allow(userId))) {
    return new Response("Too Many Requests", { status: 429 });
  }
  return serve(request);
}
```

## Report the remaining budget

[Section titled “Report the remaining budget”](#report-the-remaining-budget)

`script()` decodes a single scalar, so the counter comes back as a number. Derive the rest on the client and surface it in headers:

```ts
const count = await redis.script(rateLimit).run({
  keys: { counter: `ratelimit:${userId}` },
  args: { windowSeconds: WINDOW_SECONDS }
});
const remaining = Math.max(0, MAX_PER_WINDOW - count);
const allowed = count <= MAX_PER_WINDOW;
```

## Beyond fixed windows

[Section titled “Beyond fixed windows”](#beyond-fixed-windows)

* **Sliding window**: key on a rolling bucket (`Math.floor(Date.now() / 1000 / WINDOW)`) and sum the current and previous buckets, weighted by how far into the window you are. Same `script()` shape, a little more Lua.
* **Token bucket**: store the token count and last-refill timestamp in a `hash`, refilling on read. Combine [per-field TTL](/benni/data-structures/hashes/#field-expiration) with a `script()` for the atomic refill-and-take.

The fixed window is the simplest correct choice and the right default; reach for the others only when smoothing bursts across the boundary actually matters.

# User Session Store

> Sessions are a natural fit for JSON key-value entries with TTL.

Sessions are a natural fit for JSON key-value entries with TTL.

Note

This page is about **user login sessions**, a record you store in Redis. It is unrelated to [`redis.session()`](/benni/advanced/sessions/), which leases a dedicated Redis *connection* for blocking commands and `WATCH`. Cookie-backed login sessions work on every adapter, including the edge.

```ts
import { json, kv } from "benni/schema";


type Session = {
  userId: string;
  createdAt: string;
};


export const sessions = kv("session", json<Session>());
```

Create a session:

```ts
await redis.kv(sessions).set(
  sessionId,
  {
    userId: "42",
    createdAt: new Date().toISOString()
  },
  {
    ttlSeconds: 60 * 60 * 24 * 7
  }
);
```

Read a session:

```ts
const session = await redis.kv(sessions).get(sessionId);
```

Extend a session:

```ts
await redis.kv(sessions).expire(sessionId, 60 * 60 * 24 * 7);
```

Delete a session:

```ts
await redis.kv(sessions).del(sessionId);
```

Raw Redis equivalent:

```ts
await nodeRedis.set(`session:${sessionId}`, JSON.stringify(session), {
  EX: 60 * 60 * 24 * 7
});
```

## Field-level expiry with hashes

[Section titled “Field-level expiry with hashes”](#field-level-expiry-with-hashes)

When parts of a session expire on different schedules (say a short-lived CSRF token alongside a week-long identity), model it as a `hash` and give each field its own TTL. Redis 8 sets the values and their expiry atomically with `HSETEX`:

```ts
import { hash, number, string } from "benni/schema";


export const sessionData = hash("session", {
  userId: string(),
  csrfToken: string(),
  lastSeen: number()
});
```

```ts
// Identity lives for a week; the CSRF token for an hour.
await redis.hash(sessionData).hsetex(
  sessionId,
  { userId: "42", lastSeen: Date.now() },
  { ttlSeconds: 60 * 60 * 24 * 7 }
);
await redis.hash(sessionData).hsetex(
  sessionId,
  { csrfToken: token },
  { ttlSeconds: 60 * 60 }
);


// Read the identity and slide its TTL in one round trip (HGETEX).
const identity = await redis.hash(sessionData).hgetex(
  sessionId,
  ["userId", "lastSeen"],
  { ttlSeconds: 60 * 60 * 24 * 7 }
);
```

The CSRF token expires on its own an hour in while the identity fields keep the session alive for a week: no separate keys, and each field carries its own clock. See [Field Expiration](/benni/data-structures/hashes/#field-expiration).

# Budget

> Cost-weighted spend limits: cap a user at tokens or cents per window, with reservations that hold an estimate while a model call is in flight.

Rate limits count requests. Model calls are not priced by the request, so counting them caps nothing you actually care about.

One call with a 200k-token context costs what fifty 4k-token calls cost. “100 requests per minute” lets a single user spend fifty times more than another while both stay inside the limit. `budget` counts the unit you are billed in: tokens, cents, credits.

## The Simple Case

[Section titled “The Simple Case”](#the-simple-case)

When you know the cost before you spend it:

schema.ts

```ts
import { budget } from "benni/schema";


export const tokens = budget("tokens", {
  limit: 2_000_000,       // tokens
  windowMs: 86_400_000    // per day
});
```

Declared as a schema value it lands in [`redis.query`](/benni/core-concepts/schema-registry/) and needs no client of its own. `benni/primitives` exports the same budget in its client-taking form for code that holds a client but no handle: `budget({ client, limit, windowMs })`.

app.ts

```ts
const { ok, remaining, retryAfterMs } = await redis.query.tokens.charge(userId, promptTokens);
if (!ok) {
  return Response.json(
    { error: "Daily token budget exhausted", remaining },
    { status: 429, headers: { "Retry-After": String(Math.ceil(retryAfterMs / 1000)) } }
  );
}
```

One atomic round trip. Nothing is charged when it does not fit.

Amounts must be whole numbers, because the counters underneath are Redis integers. Budget in the smallest unit you meter: tokens are already whole, and money should be cents, or micro-cents when per-token prices need the resolution.

## Reservations

[Section titled “Reservations”](#reservations)

The hard part is that you do not know a call’s real cost until it returns. Check-then-spend is a race: ten concurrent requests all see room, all proceed, and the budget is blown by the time any of them reports usage.

Hold an estimate first, then reconcile:

```ts
const hold = await budgets.reserve(userId, 8_000);
if (!hold) return new Response("Budget exhausted", { status: 429 });


try {
  const result = await callModel(prompt);
  await hold.settle(result.usage.totalTokens);   // charge what was really used
} catch {
  await hold.release();                           // charge nothing
}
```

The estimate counts against everyone else from the moment it is taken, so concurrent callers see it. On `settle` the hold is replaced by the real number, which is usually smaller, and the difference goes straight back to the budget.

A hold is a **lease, not a lock**. If the process holding it dies, the hold lapses on its own and stops counting; there is no sweeper to run and nothing to clean up. For calls that outlive `holdTtlMs` (two minutes by default), heartbeat with `hold.extend()`.

### Settle Semantics

[Section titled “Settle Semantics”](#settle-semantics)

Two cases look identical from the outside and need opposite answers, so it is worth being precise:

* **Settling twice charges once**, whether or not the hold was still there. Double-billing one call silently under-serves a paying user, which is the worse failure.
* **Settling after the hold has lapsed still charges.** The money was spent. A budget that forgets real spend is not a budget, even though this can briefly push usage over the limit.

Both hold together because settling is deduplicated in Redis, on the reservation token. The first settle for a token claims a small marker key that lives for `holdTtlMs`; any settle that finds the marker already claimed charges nothing. The handle short-circuits a repeat settle before it costs a round trip, but the marker is the actual guarantee.

That distinction matters for one failure in particular. If a settle’s reply is lost on the way back, a socket reset, a command timeout, a failover, the call rejects and you cannot tell whether Redis applied it. So retry it:

```ts
try {
  await hold.settle(usedTokens);
} catch {
  await hold.settle(usedTokens);   // safe: the token is charged at most once
}
```

A settle that rejects always leaves the hold usable, and a retry that reaches a server which already charged is a no-op. Handle-local bookkeeping cannot make that call, because the handle never learns what the server did.

The second case above is why `extend()` exists: keep long calls inside their own hold and it never arises.

The marker is its own key rather than an entry in the reservation set, so it never lengthens the scan that summing live holds pays for.

## Reading Without Spending

[Section titled “Reading Without Spending”](#reading-without-spending)

```ts
const { remaining, retryAfterMs } = await budgets.check(userId);
await budgets.reset(userId);   // clear spend and holds outright
```

`check` includes live holds, so it reflects what a caller would actually be allowed right now.

## Accuracy

[Section titled “Accuracy”](#accuracy)

The window is a two-bucket sliding estimate: the previous window’s spend decays out linearly as the current one fills. That means usage can drift slightly over the limit near a bucket boundary.

This is deliberate. The exact alternative is a log with one entry per request, and for a daily token budget that keeps every request of the last 24 hours alive in memory just to add up numbers. A counter is O(1) and never grows. If you need a hard ceiling rather than a spend guardrail, enforce it at the billing layer, not here.

The one place cost is not O(1) is summing live reservations, which walks the reservation set. That is bounded by *concurrent in-flight calls for a single id*, normally single digits. The limit does most of that bounding on its own, since every hold consumes headroom, but a hold for `0` consumes none, so `maxHolds` (10000 by default) puts a ceiling on the set regardless. Past it `reserve` returns `null` like any other denial.

`retryAfterMs` is the time until enough units decay out of the window for that exact spend, computed server-side. It is not the time to the next bucket boundary, which frees nothing: the two-bucket estimate is continuous across the roll.

Which bucket a call lands in is decided by the server’s clock, so a call that crosses a boundary is re-run against the bucket the server named. If this process is stalled for longer than a whole window in between, every attempt misses and the call throws `BudgetWindowRolledError` (exported from `benni/primitives`) rather than inventing an answer. Nothing was applied, so it is safe to retry, and a hold whose `settle` throws it is still usable.

## Cluster Safety

[Section titled “Cluster Safety”](#cluster-safety)

Each id’s two window buckets, its reservation set, and its settle markers share a `{<id>}` hash tag, so they live on one node and the scripts can touch them together. Different ids still spread across the keyspace. See [Redis Cluster](/benni/advanced/cluster/).

## Options

[Section titled “Options”](#options)

| Option      | Default    | What it does                                      |
| ----------- | ---------- | ------------------------------------------------- |
| `limit`     | required   | Units allowed per window. Must be a whole number. |
| `windowMs`  | required   | Window length in milliseconds.                    |
| `prefix`    | `"budget"` | Key namespace.                                    |
| `holdTtlMs` | `120000`   | How long a reservation counts before lapsing.     |
| `maxHolds`  | `10000`    | Most reservations one id may hold at once.        |

## When You Don’t Need This

[Section titled “When You Don’t Need This”](#when-you-dont-need-this)

* **You are limiting request rate, not spend.** Use [`ratelimit`](/benni/primitives/ratelimit/); it is exact and cheaper.
* **You are limiting concurrency.** “At most 20 calls in flight” is [`semaphore`](/benni/primitives/semaphore/).
* **You need per-request billing records.** This is a guardrail, not a ledger. Keep the ledger in your database and use this to stop runaway spend before it happens.

## See Also

[Section titled “See Also”](#see-also)

* [Rate Limiting](/benni/primitives/ratelimit/) for request-count limits
* [Semaphore](/benni/primitives/semaphore/) for concurrency limits
* [AI Apps](/benni/patterns/ai-apps/) for how these compose

# Cache

> A read-through cache with stampede protection: one loader call per miss, no matter how many concurrent readers.

`cache` is a read-through cache with **stampede protection**: on a miss, exactly one caller runs the loader (single-flight via the [distributed lock](/benni/primitives/lock/)); every other concurrent reader waits for the filled value instead of hammering your backend.

schema.ts

```ts
import { cache, json } from "benni/schema";
import { z } from "zod";


const profile = z.object({ name: z.string(), score: z.number() });


export const profiles = cache("profile", { ttlMs: 60_000, codec: json(profile) });
```

app.ts

```ts
const profile = await redis.query.profiles.get(userId, () => db.loadProfile(userId));
```

The classic failure this prevents: a hot key expires, 500 requests miss at once, and all 500 hit the database together. With `cache`, one of them loads; the other 499 poll Redis for the filled entry.

Declared as a schema value it lands in [`redis.query`](/benni/core-concepts/schema-registry/) and needs no client of its own. Where you hold a client but no handle, `benni/primitives` exports the same cache in its client-taking form, over the same keys:

```ts
import { cache } from "benni/primitives";


const profiles = cache<Profile>({ client, ttlMs: 60_000 });
const profile = await profiles.get(userId, () => db.loadProfile(userId));
```

`client` accepts a `RedisClient`, a promise of one, a factory, or a Benni handle, so it works over every adapter, including [`benni/upstash`](/benni/runtime/edge/) on the edge.

## API

[Section titled “API”](#api)

```ts
const store = cache<T>({ client, ...options });


await store.get(id, loader); // read; run loader once on a miss
await store.peek(id);        // read without loading (T | null)
await store.set(id, value);  // write directly (with the configured TTL)
await store.del(id);         // drop; returns the deleted count; the next get reloads
```

Values are encoded with `codecs.json<T>()` by default; pass `codec` to store anything else.

## Invalidation beats an in-flight load

[Section titled “Invalidation beats an in-flight load”](#invalidation-beats-an-in-flight-load)

A loader publishes its result **only while it still holds the fill lock**, so the canonical write-through order is safe:

```ts
await db.updateProfile(userId, patch);
await profiles.del(userId); // also breaks any fill lock in flight
```

`del` drops the entry and the fill lock together. A loader that read its value before the `del` finds its lock gone, so it returns that value to its own caller but does not cache it, and the next `get` reloads. The same fence stops a slow loader from overwriting a fresher entry published after its lock expired.

## Failure behavior (fail open, never deadlock)

[Section titled “Failure behavior (fail open, never deadlock)”](#failure-behavior-fail-open-never-deadlock)

If the caller holding the fill lock dies mid-load, its lock expires after `lockTtlMs` and waiting readers **load for themselves**. The worst case under failure is a brief duplicate load, never an error, never a deadlock. Waiters watch the lock rather than just the value, so when a lease is handed to a new loader they wait for that loader instead of all giving up at once; the total wait is capped at three lock lifetimes.

The other side of the fence: a load that takes longer than `lockTtlMs` no longer publishes, because by then its result may be older than whatever replaced it. The value still reaches the caller that asked for it, but it is not cached. Set `lockTtlMs` above your slowest load.

## Options

[Section titled “Options”](#options)

| Option      | Default            | Meaning                                                                                                                        |
| ----------- | ------------------ | ------------------------------------------------------------------------------------------------------------------------------ |
| `ttlMs`     | -                  | Entry lifetime.                                                                                                                |
| `prefix`    | `"cache"`          | Key namespace; entries live at `<prefix>:<id>`, fill locks at `<prefix>:lock:<id>`.                                            |
| `codec`     | `codecs.json<T>()` | Value codec.                                                                                                                   |
| `lockTtlMs` | `10000`            | How long one loader may hold the fill lock before waiters fail open and it can no longer publish. Set above your slowest load. |
| `pollMs`    | `50`               | Poll interval while waiting on another caller’s load.                                                                          |

See [Caching patterns](/benni/patterns/caching/) for the underlying Redis approach if you want to roll your own.

# Idempotency

> Exactly-once side effects keyed by a client-supplied Idempotency-Key, replaying the original response to retries instead of running the effect twice.

A retried POST must not charge the card twice, and it must return the *first* response rather than a fresh one. That is the [Stripe `Idempotency-Key`](https://docs.stripe.com/api/idempotent_requests) contract, and clients retry far more often than you would like: double-clicks, mobile reconnects, proxy timeouts, and every SDK with automatic retries.

schema.ts

```ts
import { idempotency } from "benni/schema";


export const charges = idempotency<Receipt>("charge");
```

app.ts

```ts
export async function POST(request: Request) {
  const { value, replayed } = await redis.query.charges.run(
    request.headers.get("Idempotency-Key"),
    () => chargeCard(order)
  );
  return Response.json(value, { headers: { "Idempotent-Replay": String(replayed) } });
}
```

The first caller runs the handler and stores its result. Every later caller with that key gets the stored result back, without the handler running again.

Declared as a schema value it lands in [`redis.query`](/benni/core-concepts/schema-registry/) and needs no client of its own. `benni/primitives` exports the same runner in its client-taking form for code that holds a client but no handle: `idempotency<Receipt>({ client })`.

## Not A Cache

[Section titled “Not A Cache”](#not-a-cache)

The two look alike and behave differently in the way that matters. A cache may recompute a pure read whenever it likes; a miss costs latency. Here a “miss” costs a second charge on someone’s card, so the effect must run exactly once and the *stored* outcome must be replayed even if recomputing would be cheap.

Which is why [`cache`](/benni/primitives/cache/) is keyed by what you are reading, and this is keyed by the request the client made.

## Concurrent Duplicates

[Section titled “Concurrent Duplicates”](#concurrent-duplicates)

A double-click sends two requests before either finishes. The loser waits for the winner’s result and returns the same receipt:

```ts
// Both calls return { id: "rcpt_1" }. chargeCard runs once.
const [a, b] = await Promise.all([
  once.run("key-1", () => chargeCard(order)),
  once.run("key-1", () => chargeCard(order))
]);
```

If you would rather reject than wait, `onConflict: "throw"` raises `IdempotencyConflictError` while another caller holds the key. Waiting gives up after `waitTimeoutMs` with `IdempotencyTimeoutError` rather than hanging forever.

## Optional Keys

[Section titled “Optional Keys”](#optional-keys)

Passing `null`, `undefined`, or `""` runs the handler unguarded and reports `replayed: false`, so you can forward an optional header straight through:

```ts
// No branching on whether the client sent a key.
await once.run(request.headers.get("Idempotency-Key"), handler);
```

## Failures Release The Key

[Section titled “Failures Release The Key”](#failures-release-the-key)

**If the handler throws, the record is deleted so the operation can be retried.** That is right for the failures you actually see, a timeout or a 503, where the client should be able to try again with the same key.

It also means a handler that fails *after* a partial side effect will repeat that part. This is an idempotency key, not a transaction. Either make the effect safe to repeat, or record progress inside it:

```ts
await once.run(key, async () => {
  const charge = await stripe.charges.create(
    { amount, currency: "usd" },
    { idempotencyKey: key }   // pass it downstream too
  );
  await db.orders.markPaid(order.id, charge.id);
  return toReceipt(charge);
});
```

Forwarding the same key to the downstream provider is the belt-and-braces version, and worth doing whenever the provider supports it.

## When The Result Cannot Be Stored

[Section titled “When The Result Cannot Be Stored”](#when-the-result-cannot-be-stored)

If the handler succeeds but storing its result fails, `run` throws `IdempotencyNotRecordedError` rather than returning normally. That is deliberate: the side effect happened, but nothing was recorded, so the running marker will lapse and a later call with the same key will run the handler again. Reporting plain success would hide exactly the guarantee you came here for.

Treat it as indeterminate rather than as a failure. The work is done, and the error carries the result so you can still use it:

```ts
try {
  const { value } = await once.run(key, () => chargeCard(order));
  return Response.json(value);
} catch (error) {
  if (error instanceof IdempotencyNotRecordedError) {
    // The charge went through; only the record of it did not. Return it, and
    // do not let the client retry blind.
    return Response.json(error.value, { status: 200 });
  }
  throw error;
}
```

The usual causes are a codec that cannot encode the result, or a Redis blip between finishing the work and recording it.

## Inspecting

[Section titled “Inspecting”](#inspecting)

```ts
await once.peek("key-1");    // the stored result, or null if absent or running
await once.forget("key-1");  // drop it so the next call runs again
```

## Options

[Section titled “Options”](#options)

| Option          | Default            | What it does                                                       |
| --------------- | ------------------ | ------------------------------------------------------------------ |
| `ttlMs`         | `86400000`         | How long a result stays replayable (24h, matching Stripe).         |
| `prefix`        | `"idem"`           | Key namespace.                                                     |
| `codec`         | `codecs.json<T>()` | How the result is stored.                                          |
| `runningTtlMs`  | `waitTimeoutMs`    | How long one caller may hold the key before others assume it died. |
| `onConflict`    | `"wait"`           | `"wait"` for the holder’s result, or `"throw"`.                    |
| `waitTimeoutMs` | `30000`            | How long to wait under `"wait"`.                                   |
| `pollMs`        | `50`               | Poll interval while waiting.                                       |

Size `runningTtlMs` to your slowest handler. Too short and a second caller assumes the first died and runs the effect again, which is the failure this primitive exists to prevent.

## When You Don’t Need This

[Section titled “When You Don’t Need This”](#when-you-dont-need-this)

* **The operation is naturally idempotent.** A `PUT` that sets a value needs no key.
* **You are caching a read.** Use [`cache`](/benni/primitives/cache/); it has stampede protection and no exactly-once bookkeeping to pay for.
* **The work is long-running.** Hand it to the [queue](/benni/primitives/queue/), which takes an `idempotencyKey` of its own and gives you a job to poll.

## See Also

[Section titled “See Also”](#see-also)

* [Cache](/benni/primitives/cache/)
* [AI Job Queue](/benni/primitives/queue/), which has idempotency built in
* [Next.js integration](/benni/integrations/nextjs/)

# Distributed Lock

> A correct distributed lock over Redis: acquire with SET NX PX, renew the lease while your critical section runs, release atomically so you never free someone else's lock.

`lock` is a distributed lock built the correct way: acquire with `SET key token NX PX ttl`, and release with an atomic check-and-delete Lua so a caller can **never** delete a lock that already expired and was re-acquired by someone else, the classic footgun of a naive `DEL`.

schema.ts

```ts
import { lock } from "benni/schema";


export const orderLocks = lock("order", { ttlMs: 10_000 });
```

app.ts

```ts
await redis.query.orderLocks.run("42", async () => {
  // critical section: the lock is renewed while this runs, and released
  // automatically, even if this throws
});
```

Declared as a schema value it lands in [`redis.query`](/benni/core-concepts/schema-registry/) and needs no client of its own. Where you hold a client but no handle, `benni/primitives` exports the same lock in its client-taking form, over the same keys:

```ts
import { lock } from "benni/primitives";


const locks = lock({ client, prefix: "order", ttlMs: 10_000 });
await locks.run("42", async () => { /* ... */ });
```

`client` accepts a `RedisClient`, a promise of one, a factory, or a Benni handle, so it works over every adapter, including [`benni/upstash`](/benni/runtime/edge/) on the edge (it needs only `SET` and `EVALSHA`, no persistent connection).

Two defaults decide how it behaves under pressure, and both are worth reading before you ship: acquisition **fails fast**, and `run` **renews the lease** while your body is in flight.

## Acquiring Fails Fast

[Section titled “Acquiring Fails Fast”](#acquiring-fails-fast)

`retries` defaults to `0`. A caller that finds the lock held does not wait: `acquire` resolves `null` and `run` throws `LockNotAcquiredError` immediately.

Concretely, six concurrent callers on the same id means one runs and **five throw**. That is the right default for a request handler (return 409 rather than pile up requests behind a lock), and the wrong one if what you meant was to serialize concurrent work.

Catch the error when “someone else is doing it” is a real answer:

```ts
import { LockNotAcquiredError } from "benni/primitives";


try {
  await locks.run("order:42", processOrder);
} catch (error) {
  if (error instanceof LockNotAcquiredError) {
    // someone else holds error.key, so back off, reschedule, or return 409
    return new Response("Already processing", { status: 409 });
  }
  throw error;
}
```

Pass retries when every caller must eventually run:

```ts
// Each caller waits its turn behind the holder: all six run, one at a time.
await locks.run("order:42", processOrder, {
  retries: 100,
  retryDelayMs: 50
});
```

Retries are a bounded spin, not a fair queue: callers do not get the lock in arrival order, and a heavily contended lock can starve an unlucky one. If strict ordering matters, that is a job for the [queue](/benni/primitives/queue/).

## Lease Renewal

[Section titled “Lease Renewal”](#lease-renewal)

The TTL is a safety net for crashes: if your process dies mid-section, the lock expires instead of deadlocking forever. But a TTL that expires while your body is still running is not a safety net, it is a silent correctness bug. The key lapses, another caller acquires it, and your body keeps running as though it were still exclusive. Two writers, one critical section, no error anywhere.

So `run` renews the lock while `fn` is in flight, every `heartbeatMs`:

```ts
const locks = lock(client, { ttlMs: 10_000 }); // renewed every 2.5s


await locks.run("report:nightly", async () => {
  await generateReport(); // may take minutes; the lock is held throughout
});
```

`heartbeatMs` defaults to a quarter of the effective `ttlMs` (`Math.max(1, Math.floor(ttlMs / 4))`, the same ratio the [queue](/benni/primitives/queue/) uses for job leases). A quarter means three renewals in a row can fail outright before the lock could lapse, so a blip on the wire is survivable rather than fatal.

Set it yourself when you want a different margin, and pass `false` to opt out entirely:

```ts
// Renew more often: a tighter margin against a flaky connection.
await locks.run("order:42", processOrder, { heartbeatMs: 1_000 });


// Opt out: the lock expires ttlMs after it was taken, whatever fn is doing.
// This is the pre-renewal behaviour, kept reachable for short bodies that
// genuinely cannot outlive their TTL.
await locks.run("order:42", processOrder, { heartbeatMs: false });
```

A `heartbeatMs` you pass yourself must be **at most half of `ttlMs`**, otherwise the call throws `ValidationError` before the lock is taken. At a half, one renewal can still fail before the lock could lapse; above it, the first tick can arrive at or after expiry and the lock lapses before renewal ever runs. That misconfiguration used to be silent and load dependent: a short body finished before the first tick and looked fine, while a long one failed with `LockLeaseLostError` on an uncontended lock. The check is on the value you supply, not on the derived default, which stays intact for a `ttlMs` so small that no ratio could hold.

```ts
// Throws: ValidationError, lock heartbeatMs must be at most half of ttlMs (1000)
await locks.run("order:42", processOrder, { ttlMs: 1_000, heartbeatMs: 2_000 });
```

A body that finishes inside the first interval costs no extra round trips, so renewal is free for the short critical sections that never needed it.

## When The Lock Is Lost

[Section titled “When The Lock Is Lost”](#when-the-lock-is-lost)

Renewal can fail for a real reason: the lock expired and someone else took it, or a human ran `DEL`. When that happens `run` rejects with `LockLeaseLostError`, which carries the contested `.key`.

It rejects **even when `fn` resolved**. A body that finished without the lock did not finish under the mutual exclusion it was written against, and resolving would hide exactly that:

```ts
import { LockLeaseLostError } from "benni/primitives";


try {
  const receipt = await locks.run("order:42", chargeCard);
  // reached only if the lock was held for the whole call
  return receipt;
} catch (error) {
  if (error instanceof LockLeaseLostError) {
    // chargeCard may have completed, but not exclusively: reconcile rather
    // than assume either outcome
    return reconcile(error.key);
  }
  throw error;
}
```

### Detection Is Two-Pronged

[Section titled “Detection Is Two-Pronged”](#detection-is-two-pronged)

A lost lease is not the same thing as a failed round trip, and conflating them would make every network hiccup fatal. `run` declares the lock lost only when:

1. **`extend` reports it is gone.** The renewal Lua checks the token, so a `0` reply means the key is missing or now owned by somebody else. This is immediate and definitive.
2. **A full `ttlMs` has passed with no successful renewal.** This catches the cases the first prong cannot see: renewals that keep rejecting, and a renewal that hangs and never answers at all. Silence is treated as loss, because after `ttlMs` the key has demonstrably lapsed.

That deadline is read both from the renewal tick and again in the same turn your body finishes, which matters because a tick is not guaranteed to run. A body that blocks the event loop (synchronous CPU work, a blocking native call) past its TTL starves the interval entirely, and a timer is a macrotask while resuming from `await fn(handle)` is a microtask, so the completion check would otherwise win the race and report success for a lock that had already expired. The final `release` is consulted for the same reason: it runs the same token check `extend` does, so a `false` reply is Redis saying the lock had already moved on.

One failed renewal is not a loss. The next tick simply retries. To see those failures, pass `onRenewError`:

```ts
await locks.run("order:42", processOrder, {
  onRenewError: (error) => {
    // A renewal round trip failed: a dropped connection, a timeout. The next
    // tick retries and the lock may well survive, so this is telemetry, not a
    // failure. Without the hook these errors are swallowed.
    logger.warn({ error }, "lock renewal round trip failed");
  }
});
```

### `handle.signal`

[Section titled “handle.signal”](#handlesignal)

Rejecting after the fact is a correct report, but it is late: the body already did the work. `handle.signal` is an `AbortSignal` that aborts with the `LockLeaseLostError` the moment the lock is known to be gone, so the work can stop instead of finishing unprotected.

It composes with anything that takes a signal:

```ts
await locks.run("order:42", async (handle) => {
  // fetch rejects as soon as the lock is lost
  const res = await fetch(url, { signal: handle.signal });


  // so does an AI SDK call
  const { text } = await generateText({
    model,
    prompt,
    abortSignal: handle.signal
  });


  return { res, text };
});
```

If `fn` rejects with the abort reason itself (as `fetch` does), that error propagates unchanged rather than being replaced.

## Acquire And Release Manually

[Section titled “Acquire And Release Manually”](#acquire-and-release-manually)

```ts
const handle = await locks.acquire("order:42");
if (handle) {
  try {
    // ... work ...
  } finally {
    await handle.release(); // resolves true only if we still held it
  }
}
```

`acquire` resolves `null` when the lock is already held. `release()` and `extend()` resolve `true` only when your token still owns the key; both run the atomic Lua, so they are safe under expiry races.

**An `acquire`d handle is not renewed in the background.** Nothing watches it on your behalf: if the work can outlive `ttlMs`, you have to call `extend()` yourself. That also means `handle.signal` cannot fire unless you do, because your own `extend()` resolving `false` is the only thing that can abort it. If you want renewal, use `run`.

```ts
const handle = await locks.acquire("report", { ttlMs: 30_000 });
// ... halfway through a long job, keep the lock alive:
const stillOurs = await handle?.extend(30_000);
if (stillOurs === false) {
  // we overran: the lock is gone and handle.signal has aborted
}
```

## Options

[Section titled “Options”](#options)

| Option         | Where                      | Default     | Meaning                                                                                                          |
| -------------- | -------------------------- | ----------- | ---------------------------------------------------------------------------------------------------------------- |
| `prefix`       | `lock(client, …)`          | `"lock"`    | Key namespace; keys are `<prefix>:<id>`.                                                                         |
| `ttlMs`        | `lock` / `acquire` / `run` | `30000`     | Lock lifetime. It is the crash backstop, and with `run` it is also the renewal window.                           |
| `retries`      | `acquire` / `run`          | `0`         | Attempts when the lock is held. `0` fails fast.                                                                  |
| `retryDelayMs` | `acquire` / `run`          | `100`       | Delay between retries.                                                                                           |
| `heartbeatMs`  | `run`                      | `ttlMs / 4` | Renewal interval while `fn` runs. Must be at most half of `ttlMs` when set explicitly. `false` disables renewal. |
| `onRenewError` | `run`                      | none        | Called when a renewal round trip fails. Not a lost lock.                                                         |

## Relationship To `semaphore`

[Section titled “Relationship To semaphore”](#relationship-to-semaphore)

A lock lets one caller through. [`semaphore`](/benni/primitives/semaphore/) lets `N` through, and follows the same lease policy: `run` renews in the background, a lost slot surfaces as a typed error (`SemaphoreLeaseLostError`) rather than a silent success, and its handle carries the same `signal`. The difference is what a lost lease costs you: for a lock it means two writers collided, and for a semaphore it means the pool over-admits. Everything else you know here transfers.

## See Also

[Section titled “See Also”](#see-also)

* [Semaphore](/benni/primitives/semaphore/) for bounded concurrency rather than one-at-a-time
* [Idempotency](/benni/primitives/idempotency/) when the goal is “exactly once”, not “one at a time”
* [Queue](/benni/primitives/queue/) when callers must all run, in order

# AI Job Queue

> Run model calls as background jobs that survive refreshes, deploys, and crashes, with a resumable output stream and a Stop button that actually stops the bill.

`queue` runs expensive model calls as background jobs, so a generation survives the user refreshing the page, your server deploying, and the request timing out.

schema.ts

```ts
import { queue } from "benni/schema";


export const generate = queue<{ prompt: string }, string>("generate");
```

app.ts

```ts
const { id } = await redis.query.generate.enqueue({ prompt });
```

Declared as a schema value it lands in [`redis.query`](/benni/core-concepts/schema-registry/) and needs no client of its own. Where you hold a client but no handle, `benni/primitives` exports the same queue in its client-taking form, over the same keys:

```ts
import { queue } from "benni/primitives";


const jobs = queue<{ prompt: string }, string>({ client, prefix: "generate" });
```

## The five bugs you hit in order

[Section titled “The five bugs you hit in order”](#the-five-bugs-you-hit-in-order)

Every app with a **Generate** button discovers these in the same sequence:

1. **The request times out.** A long generation outlives your platform’s request limit, and the user gets a 504 after paying for 45 seconds of tokens.
2. **A refresh loses everything.** The stream lived in one HTTP response. The tab reloads, the answer is gone, and you generate it again, at full price.
3. **Deploys eat in-flight work.** Every generation running when the container recycles just… disappears. Nobody knows which ones.
4. **Stop doesn’t stop.** The user hits Stop, the UI clears, and your server keeps streaming tokens from the provider into a void you are still billed for.
5. **One click bills twice.** A double-click, a client retry, or an at-least-once webhook fires two identical generations, and both run.

Each has a well-known fix. The catch is that they’re *five different fixes* (a job queue, a stream buffer, a lease, a cancellation channel, an idempotency store), and they have to agree with each other. Wire them separately and they fight: the classic version of this is a resumable-stream layer that can’t tell a user pressing **Stop** apart from a dropped connection, so it dutifully resumes a generation the user cancelled.

`queue` is those five fixes as one thing.

## The whole loop

[Section titled “The whole loop”](#the-whole-loop)

**Producer**: one atomic round trip. Runs anywhere, including [the edge](/benni/runtime/edge/):

```ts
const { id } = await jobs.enqueue({ prompt }, { idempotencyKey: requestId });
```

**Worker**: a long-lived process:

```ts
import { openai } from "@ai-sdk/openai";
import { streamText } from "ai";


jobs.worker(
  async (job) => {
    const { textStream } = streamText({
      model: openai("gpt-4o-mini"),
      prompt: job.payload.prompt,
      abortSignal: job.signal // Stop actually stops the provider
    });


    let text = "";
    for await (const delta of textStream) {
      text += delta;
      await job.emit(delta); // stream to watchers, and stay alive
    }
    return text;
  },
  { concurrency: 8 }
);
```

**Consumer**: an endpoint that survives reconnects:

```ts
for await (const event of jobs.watch(id, { after: lastSeenEventId })) {
  if (event.type === "chunk") send(event.data, event.id);
  if (event.type === "restarted") clear();
  if (event.type === "completed") return event.result;
}
```

That’s the whole thing. All five bugs are gone: the work outlives the request, `after` resumes it, a dead worker’s job is reclaimed, `job.signal` propagates Stop, and `idempotencyKey` collapses the duplicate.

The `watch` loop needs no break condition; the iterator ends itself after the job’s terminal event.

## What it replaces

[Section titled “What it replaces”](#what-it-replaces)

| The bug                    | What you’d otherwise wire                       | Here                                      |
| -------------------------- | ----------------------------------------------- | ----------------------------------------- |
| Request times out          | A job queue + a worker                          | `enqueue` / `worker`                      |
| Refresh loses the stream   | A separate stream buffer keyed by generation id | Every job *has* an output stream          |
| Deploy eats in-flight work | Visibility timeouts, stalled-job sweepers       | Heartbeat leases, reclaimed automatically |
| Stop doesn’t stop          | A cancellation channel the worker polls         | `cancel()` → `job.signal` aborts          |
| Double-billed clicks       | An idempotency table with its own TTL rules     | `idempotencyKey`                          |
| A 429 you retried too fast | Backoff logic per provider                      | `RetryJobError(msg, retryAfterMs)`        |

## Three ideas worth knowing

[Section titled “Three ideas worth knowing”](#three-ideas-worth-knowing)

Everything above rests on these, and they’re what make it feel different in use.

### Streaming a token *is* the heartbeat

[Section titled “Streaming a token is the heartbeat”](#streaming-a-token-is-the-heartbeat)

`job.emit(token)` appends to the job’s stream **and** renews the lease **and** checks for cancellation: one round trip, no separate keepalive to remember.

This is why a ten-minute generation is ordinary here rather than something you tune around. Most queues detect a dead worker by *idleness*, which is precisely wrong for a worker legitimately blocked on a slow model. A worker here says “still alive” by doing its actual job.

If your handler doesn’t stream, an automatic heartbeat covers it; you don’t have to call anything.

### Stop stops the bill

[Section titled “Stop stops the bill”](#stop-stops-the-bill)

```ts
await jobs.cancel(id); // true if the job will not produce a result
```

A job that hasn’t started is removed outright. A running job is flagged, and its worker aborts `job.signal` on the next `emit()`, so a `fetch` or AI SDK call wired to that signal tears down mid-stream and you stop paying for a cancelled answer. Either way the job settles `cancelled`, and watchers get a `cancelled` event instead of hanging forever.

Cancelling can’t race the worker into a double-settle: only the worker holding the current lease may write a result. Nor can it lose to one. If the handler finishes, or throws a retryable error, after `cancel()` returned `true`, the job still settles `cancelled`: the result is discarded and no further attempt is scheduled, rather than the queue paying for a generation the caller already stopped.

### A retried generation restarts its stream

[Section titled “A retried generation restarts its stream”](#a-retried-generation-restarts-its-stream)

If attempt 1 dies halfway through `"The capital of"`, attempt 2 starts over. The partial tokens are dropped and a `restarted` event is written first, so a client resuming from a cursor clears its buffer instead of rendering `"The capital ofThe capital of France is Paris"`. The marker is always written above every entry id the previous attempt used, so a resuming cursor can’t skip past it.

That’s the one event type worth handling deliberately.

## Paying once for duplicate work

[Section titled “Paying once for duplicate work”](#paying-once-for-duplicate-work)

```ts
const first = await jobs.enqueue({ prompt }, { idempotencyKey: requestId });
const again = await jobs.enqueue({ prompt }, { idempotencyKey: requestId });
// again.id === first.id, again.deduplicated === true, no second model call
```

The key stays bound to the job for as long as it runs *and after it completes*, so a retry arriving late gets the finished answer rather than starting over. `idempotencyTtlMs` is the retention after completion, not a countdown from enqueue: a job that sits in a backlog for an hour and then streams for ten minutes still holds its key throughout. A job that fails or is cancelled releases its key: there’s no answer to hand out, so a genuine retry should be allowed.

## Retries that match how providers actually fail

[Section titled “Retries that match how providers actually fail”](#retries-that-match-how-providers-actually-fail)

Everything retries with exponential backoff and full jitter, except what you mark otherwise:

```ts
import { RetryJobError, TerminalJobError } from "benni/primitives";


jobs.worker(async (job) => {
  const response = await fetch(providerUrl, {
    method: "POST",
    body: JSON.stringify(job.payload),
    signal: job.signal
  });


  // 429: the provider told us exactly when to come back. Believe it.
  if (response.status === 429) {
    const retryAfter = Number(response.headers.get("retry-after") ?? 1);
    throw new RetryJobError("rate limited", retryAfter * 1000);
  }


  // 400: a retry reproduces this verbatim. Don't waste three attempts.
  if (response.status === 400) {
    throw new TerminalJobError(`malformed request: ${await response.text()}`);
  }


  return (await response.json()).text;
});
```

Pass `isRetryable` to `worker()` to replace the classification wholesale.

## Awaiting a result

[Section titled “Awaiting a result”](#awaiting-a-result)

When you just want the answer and don’t care about tokens:

```ts
const text = await jobs.wait(id);
```

It checks the record first, so a job that already finished returns immediately instead of waiting for an event that has passed. Unknown ids (and jobs whose `resultTtlMs` has elapsed) reject with `JobNotFoundError`.

## When a worker dies

[Section titled “When a worker dies”](#when-a-worker-dies)

Nothing to configure. The dead worker’s lease expires, the next `reserve` reclaims the job, and it goes back to the ready set, or straight to the dead letter set if it’s out of attempts. Because attempts are counted when a job is *reserved*, a handler that reliably crashes the process dead-letters instead of looping forever.

A zombie worker that wakes up later can’t clobber the job that replaced it: `emit()`, `progress()`, and the heartbeat all throw `JobLeaseLostError` once the lease token is stale, and settling is refused.

```ts
const worker = jobs.worker(handler, {
  concurrency: 8,
  leaseMs: 120_000, // longer than your slowest generation
  heartbeatMs: 15_000 // comfortably inside the lease
});


// Graceful shutdown: stop taking new work, let in-flight jobs finish.
process.on("SIGTERM", () => void worker.stop());
```

`stop()` never kills a running job: in-flight work keeps its lease and finishes, so nothing is double-run.

## Inspecting a job

[Section titled “Inspecting a job”](#inspecting-a-job)

`jobs.get(id)` returns the whole job record, and its current state is the `status` **property** on that record. There is no `jobs.status(id)`, and the field is `status`, not `state`:

```ts
const job = await jobs.get(id);
//    ^? Job<{ prompt: string }, string> | null


if (job?.status === "completed") {
  const text = job.result; // present only on "completed"
}
```

The record also carries `attempt` / `maxAttempts`, `progress` (`0`-`1`), `error` (the last failure message, kept across retries), `priority`, the `createdAt` / `updatedAt` / `startedAt` / `finishedAt` timestamps, `idempotencyKey`, and `cancelRequested` (true from the moment `cancel()` is called, even while the job is still running).

Two names worth spelling out, because they do not match each other: a `Worker` is shut down with `worker.stop()`, while the Redis client is shut down with `client.close()`. The worker is not a connection, so it does not get `close()`.

## Operating it

[Section titled “Operating it”](#operating-it)

```ts
await jobs.stats(); // { waiting, scheduled, active, dead }


for (const id of await jobs.dead({ count: 20 })) {
  await jobs.retryDead(id); // back to the queue with a fresh attempt count
}
```

`retryDead` also clears the failed attempt’s output, so a watcher doesn’t stop on the stale `failed` event.

Jobs can also be delayed and prioritised, which is how interactive work stays ahead of batch work:

```ts
await jobs.enqueue(payload, { priority: 9 }); // a user is waiting
await jobs.enqueue(payload, { priority: 0, delayMs: 60_000 }); // nightly backfill
```

Passing your own `id` reuses it: once that job has finished, re-enqueuing the id starts a clean generation, dropping the old record, its dead-letter entry, and its output stream so a watcher can’t stop on last time’s terminal event. Reusing an id that hasn’t finished throws instead, because there is no honest way to have one id be two live jobs.

## When you don’t need this

[Section titled “When you don’t need this”](#when-you-dont-need-this)

Be honest about the fit: it’s a worker process to run and monitor:

* **The call is fast and the user is watching.** Under a few seconds, stream it from the request and skip all of this.
* **You’ve already answered this prompt.** [`cache`](/benni/primitives/cache/) is cheaper than any queue, so check it first and queue only on a miss.
* **You need durable *execution*.** On a retry, your handler re-runs from the top. There is no checkpointing of a half-finished agent loop and no resuming mid-function. If you need “the agent completed 3 of 7 tool calls, resume at 4”, model those steps as separate jobs, or use a durable-execution engine.
* **You have no long-lived process.** `worker()` needs one. Producing and watching work fine at the edge; running doesn’t.

## How it’s built

[Section titled “How it’s built”](#how-its-built)

Job *lifecycle* lives in sorted sets; job *output* lives in a stream. Each is good at exactly one of those.

| Key                    | Type   | Holds                                                    |
| ---------------------- | ------ | -------------------------------------------------------- |
| `{prefix}:ready`       | zset   | Runnable jobs, priority-major and FIFO within a priority |
| `{prefix}:scheduled`   | zset   | Delayed jobs and backoff retries, scored by ready time   |
| `{prefix}:leases`      | zset   | Owned jobs, scored by lease expiry                       |
| `{prefix}:dead`        | zset   | Dead-lettered jobs                                       |
| `{prefix}:job:<id>`    | hash   | The job record                                           |
| `{prefix}:events:<id>` | stream | That job’s output                                        |
| `{prefix}:signal`      | list   | A doorbell, so idle workers block instead of polling     |

A stream consumer group would hand you recovery via `XAUTOCLAIM`, but it reclaims by *idle time*, the wrong signal for a worker blocked on a slow model, and it gives you no delays, priority, backoff, or dead-lettering. Sorted sets give all four; the stream does what streams are good at.

Every key shares one hash tag, so a queue occupies a single Redis Cluster slot. Every state change is a single Lua script, so there is no window where a job is in two places or none.

## Options

[Section titled “Options”](#options)

| Option                       | Default          | Notes                                                                                                                    |
| ---------------------------- | ---------------- | ------------------------------------------------------------------------------------------------------------------------ |
| `prefix`                     | `"queue"`        | Key namespace; also the Cluster hash tag                                                                                 |
| `codec` / `resultCodec`      | `codecs.json()`  | Any [codec](/benni/api/schema-builders/#codecs), including [`zodCodec`](/benni/integrations/zod/) for validated payloads |
| `leaseMs`                    | `60000`          | Ownership without a heartbeat, sized for model calls                                                                     |
| `maxAttempts`                | `3`              | Attempts before dead-lettering                                                                                           |
| `backoffMs` / `maxBackoffMs` | `1000` / `60000` | Exponential curve with full jitter                                                                                       |
| `resultTtlMs`                | `3600000`        | How long a finished record and its stream survive                                                                        |
| `eventsMaxLen`               | `10000`          | Retained events per job, so token streams stay bounded                                                                   |

Worker options: `concurrency`, `leaseMs`, `heartbeatMs`, `pollMs`, `isRetryable`, `onError`.

## Runtime support

[Section titled “Runtime support”](#runtime-support)

`enqueue`, `get`, `cancel`, `wait`, `watch`, `stats`, and `dead` need only `EVALSHA` and stream reads, so they run on every adapter, including [`benni/upstash`](/benni/runtime/edge/) on Cloudflare Workers and Vercel Edge. That’s the shape most AI apps want: enqueue from an edge route, run the model on a worker.

`worker()` needs a persistent process. Where the adapter offers a dedicated connection it blocks on the doorbell list, so a job starts a round trip after it’s enqueued; otherwise it polls at `pollMs`. `watch()` degrades the same way: blocking `XREAD` on Node and Bun, polling on the edge.

One cost to size for: each live `watch()` holds a connection while it iterates, and each `worker()` holds one for its doorbell. Fine for a worker fleet and a handful of dashboards; an endpoint fanning one generation out to thousands of concurrent viewers would want a connection per viewer. Until a shared tail exists, pass a `pollMs` for those watchers, or fan out from a single server-side `watch()` to your own subscribers.

## See also

[Section titled “See also”](#see-also)

* [AI Apps](/benni/patterns/ai-apps/): chat memory, token budgets, and response caching around this queue
* [Cache](/benni/primitives/cache/): answer a repeat prompt without queueing anything
* [Rate Limiting](/benni/primitives/ratelimit/): cap what reaches the queue per user
* [Streams](/benni/data-structures/streams/): the typed API under the job output stream

# Rate Limiting (primitive)

> A sliding-window rate limiter over Redis: one atomic round trip per check, accurate, and edge-ready.

`ratelimit` is a sliding-window rate limiter. Each `check(id)` is a single atomic Lua round trip that drops expired entries, counts the window, and admits the request if it is under the limit.

schema.ts

```ts
import { ratelimit } from "benni/schema";


export const apiLimit = ratelimit("api", { limit: 10, windowMs: 60_000 });
```

app.ts

```ts
const { success, remaining, resetMs } = await redis.query.apiLimit.check(userId);
if (!success) {
  throw new Response("Too Many Requests", {
    status: 429,
    headers: { "Retry-After": String(Math.ceil((resetMs - Date.now()) / 1000)) }
  });
}
```

Declared as a schema value it lands in [`redis.query`](/benni/core-concepts/schema-registry/) and needs no client of its own. Where you hold a client but no handle, such as inside a middleware factory, `benni/primitives` exports the same limiter in its client-taking form, over the same keys:

```ts
import { ratelimit } from "benni/primitives";


const limiter = ratelimit({ client, limit: 10, windowMs: 60_000 });
const { success } = await limiter.check(userId);
```

`client` accepts a `RedisClient`, a promise of one, a factory, or a Benni handle, so it runs over every adapter, including [`benni/upstash`](/benni/runtime/edge/) on the edge, which is where rate limiting is most often needed.

## The result

[Section titled “The result”](#the-result)

```ts
type RatelimitResult = {
  success: boolean;   // is this request allowed?
  limit: number;      // the configured limit
  remaining: number;  // requests left in the window (0 when denied)
  resetMs: number;    // epoch-ms when the window next frees a slot
};
```

## How it works

[Section titled “How it works”](#how-it-works)

The window is a **log of request timestamps in one sorted set** (a single key, so it is Redis Cluster safe). That makes the limit exact (no fixed-window boundary bursts) at the cost of storing up to `limit` entries per key. For typical API limits (tens to hundreds per window) that is ideal; for very high per-key rates, prefer a counter-based limiter.

## Options

[Section titled “Options”](#options)

| Option     | Default       | Meaning                                     |
| ---------- | ------------- | ------------------------------------------- |
| `limit`    | -             | Maximum requests allowed within the window. |
| `windowMs` | -             | Window length in milliseconds.              |
| `prefix`   | `"ratelimit"` | Key namespace; keys are `<prefix>:<id>`.    |

Use a stable `id` per subject: a user id, API key, or IP. Each id is limited independently.

See [Rate Limiting patterns](/benni/patterns/rate-limiting/) for the underlying Redis approach if you want to roll your own.

# Semaphore

> Bounded concurrency across processes: at most N callers in the critical section at once, with leases that renew while you work and reclaim slots from dead holders.

A lock lets one caller through. A semaphore lets `N` through. That number is usually what a provider actually enforces.

Rate and concurrency are different constraints, and model providers impose both: a rate limit protects their billing, a concurrency limit protects their capacity, and exceeding either gets you 429s. `p-limit` solves this inside one process; the moment you run two instances, the limit is per-instance and the provider sees the sum.

schema.ts

```ts
import { semaphore } from "benni/schema";


export const slots = semaphore("provider", { limit: 20, leaseMs: 60_000 });
```

app.ts

```ts
const answer = await redis.query.slots.run("openai", async () => callModel(prompt));
```

Declared as a schema value it lands in [`redis.query`](/benni/core-concepts/schema-registry/) and needs no client of its own. `benni/primitives` exports the same semaphore in its client-taking form for code that holds a client but no handle: `semaphore({ client, limit: 20 })`.

At most 20 callers are inside that body at once, across every process pointed at the same Redis. The lease is renewed while the body runs, so a slow call keeps its slot rather than losing it mid-flight.

## Acquiring Fails Fast

[Section titled “Acquiring Fails Fast”](#acquiring-fails-fast)

`retries` defaults to `0`. A caller that finds every slot taken does not wait: `acquire` returns `null` and `run` throws `SemaphoreNotAcquiredError`, which carries the `.key` and the `.limit`.

Concretely, 25 concurrent callers on a `limit: 20` semaphore means 20 run and **five throw**. That is load shedding, and it is usually what you want in a request handler.

```ts
const held = await slots.acquire("openai");
if (!held) return new Response("Busy, try again", { status: 503 });
try {
  await doWork();
} finally {
  await held.release();
}
```

`acquire` returns `null` rather than throwing when every slot is taken, so “no capacity” is an ordinary branch. `run` throws `SemaphoreNotAcquiredError` instead, since it has nowhere to put a null.

To queue instead of shedding:

```ts
await slots.run("openai", work, { retries: 100, retryDelayMs: 50 });
```

Retries are a bounded spin, not a fair queue: callers do not get slots in arrival order, and a heavily contended semaphore can starve an unlucky one. If strict ordering matters, that is a job for the [queue](/benni/primitives/queue/), which is built for it.

## Leases And Dead Holders

[Section titled “Leases And Dead Holders”](#leases-and-dead-holders)

A slot is held by a lease, not by a connection. Holders live in a sorted set scored by expiry, so reclaiming the slots of processes that crashed is just dropping the expired range, on the next acquire, with no sweeper to run.

That is what makes a crashed holder harmless. It is also what makes a *slow* holder dangerous: when your lease lapses while you are still working, the next acquire prunes you and admits somebody else. Nothing failed, nothing was logged, and the semaphore is now over its limit. A `limit: 20` guarding a provider quota quietly runs 21 in flight, which is precisely the 429 it existed to prevent.

This is the one place the semaphore differs from a [lock](/benni/primitives/lock/) in kind rather than in degree. A lost lock means two writers collided on one key. A lost slot means the pool **over-admits**: everyone else is behaving correctly and the ceiling is simply wrong.

## Lease Renewal

[Section titled “Lease Renewal”](#lease-renewal)

So `run` renews the lease while `fn` is in flight, every `heartbeatMs`:

```ts
const slots = semaphore(client, { limit: 20, leaseMs: 60_000 }); // 15s heartbeat


await slots.run("openai", async () => {
  // a streaming completion that runs for minutes keeps its slot throughout
  return callModel(prompt);
});
```

`heartbeatMs` defaults to a quarter of the effective `leaseMs` (`Math.max(1, Math.floor(leaseMs / 4))`, which at the default `leaseMs` is exactly the 15s the [queue](/benni/primitives/queue/) uses for job leases). A quarter means three renewals in a row can fail outright before the slot could lapse, so a blip on the wire is survivable rather than fatal.

Set it yourself for a different margin, and pass `false` to opt out entirely:

```ts
// Renew more often: a tighter margin against a flaky connection.
await slots.run("openai", work, { heartbeatMs: 5_000 });


// Opt out: the slot is reclaimable leaseMs after it was taken, whatever fn is
// doing. This is the pre-renewal behaviour, kept reachable for short bodies
// that genuinely cannot outlive their lease.
await slots.run("openai", work, { heartbeatMs: false });
```

A `heartbeatMs` you pass yourself must be **at most half of `leaseMs`**, otherwise the call throws `ValidationError` before a slot is taken. At a half, one renewal can still fail before the slot could lapse; above it, the first tick can arrive at or after expiry and the slot is reclaimable before renewal ever runs. That misconfiguration used to be silent and load dependent: a short body finished before the first tick and looked fine, while a long one failed with `SemaphoreLeaseLostError` even with the pool uncontended. The check is on the value you supply, not on the derived default, which stays intact for a `leaseMs` so small that no ratio could hold.

```ts
// Throws: ValidationError, semaphore heartbeatMs must be at most half of leaseMs (1000)
await slots.run("openai", work, { leaseMs: 1_000, heartbeatMs: 2_000 });
```

A body that finishes inside the first interval costs no extra round trips, so renewal is free for the short calls that never needed it.

## When The Slot Is Lost

[Section titled “When The Slot Is Lost”](#when-the-slot-is-lost)

If renewal finds the slot gone, `run` rejects with `SemaphoreLeaseLostError`, carrying the `.key` and the `.limit`.

It rejects **even when `fn` resolved**. A body that finished without a slot did not finish under the bound it was written against, and resolving would hide exactly the over-admission you added the semaphore to prevent:

```ts
import { SemaphoreLeaseLostError } from "benni/primitives";


try {
  return await slots.run("openai", () => callModel(prompt));
} catch (error) {
  if (error instanceof SemaphoreLeaseLostError) {
    // the call may have completed, but the pool was over its limit while it
    // ran: treat it as a capacity incident, and raise leaseMs if it recurs
    logger.error({ key: error.key, limit: error.limit }, "semaphore overran");
  }
  throw error;
}
```

### Detection Is Two-Pronged

[Section titled “Detection Is Two-Pronged”](#detection-is-two-pronged)

A lost lease is not the same thing as a failed round trip, and conflating them would make every network hiccup fatal. `run` declares the slot lost only when:

1. **`extend` reports it is gone.** The renewal Lua checks that our member is present *and* that its score is still in the future, because presence is not ownership: an expired member sits in the set until some acquire prunes it. A `false` result is immediate and definitive.
2. **A full `leaseMs` has passed with no successful renewal.** This catches what the first prong cannot see: renewals that keep rejecting, and a renewal that hangs and never answers at all. Silence is treated as loss, because after `leaseMs` the slot is demonstrably reclaimable.

That deadline is read both from the renewal tick and again in the same turn your body finishes, which matters because a tick is not guaranteed to run. A body that blocks the event loop (synchronous CPU work, a blocking native call) past its lease starves the interval entirely, and a timer is a macrotask while resuming from `await fn(handle)` is a microtask, so the completion check would otherwise win the race and report success for a slot that had already been reclaimed. The final `release` is consulted for the same reason: it runs the same ownership check `extend` does, so a `false` reply is Redis saying the slot had already moved on.

One failed renewal is not a loss. The next tick simply retries. To see those failures, pass `onRenewError`:

```ts
await slots.run("openai", work, {
  onRenewError: (error) => {
    // A renewal round trip failed: a dropped connection, a timeout. The next
    // tick retries and the slot may well survive, so this is telemetry, not a
    // failure. Without the hook these errors are swallowed.
    logger.warn({ error }, "semaphore renewal round trip failed");
  }
});
```

### `held.signal`

[Section titled “held.signal”](#heldsignal)

Rejecting after the fact is a correct report, but it is late: the call already went out over the limit. `held.signal` is an `AbortSignal` that aborts with the `SemaphoreLeaseLostError` the moment the slot is known to be gone, so the work can stop instead of running unaccounted for.

It composes with anything that takes a signal, which for a semaphore is usually the very call you are bounding:

```ts
await slots.run("openai", async (held) => {
  const { text } = await generateText({
    model,
    prompt,
    abortSignal: held.signal // stops the moment we are over the limit
  });
  return text;
});
```

```ts
await slots.run("openai", async (held) => {
  const res = await fetch(url, { signal: held.signal });
  return res.json();
});
```

If `fn` rejects with the abort reason itself (as `fetch` does), that error propagates unchanged rather than being replaced.

### Renewing By Hand

[Section titled “Renewing By Hand”](#renewing-by-hand)

**An `acquire`d handle is not renewed in the background.** Nothing watches the lease on your behalf: if the work can outlive `leaseMs`, call `extend()` yourself. That also means `held.signal` cannot fire unless you do, because your own `extend()` resolving `false` is the only thing that can abort it. If you want renewal, use `run`.

```ts
const held = await slots.acquire("openai", { leaseMs: 5_000 });
// ... 6 seconds pass ...
const stillOurs = await held?.extend();
if (stillOurs === false) {
  // our slot is gone and held.signal has aborted: someone else has it now
}
```

## Inspecting

[Section titled “Inspecting”](#inspecting)

```ts
await slots.count("openai");   // live holders, ignoring lapsed leases
```

## Relationship To `lock`

[Section titled “Relationship To lock”](#relationship-to-lock)

This is [`lock`](/benni/primitives/lock/) with a number: same handle shape, same `run`, same retry options, same lease renewal, same `signal`. Reach for `lock` when the answer is one and for this when it is a budget. Everything you know about one transfers, apart from what a lost lease costs: for `lock` it is two writers in one critical section, and here it is a pool that admits one caller too many.

## Options

[Section titled “Options”](#options)

| Option         | Where                           | Default       | What it does                                                                                                       |
| -------------- | ------------------------------- | ------------- | ------------------------------------------------------------------------------------------------------------------ |
| `limit`        | `semaphore(client, …)`          | required      | How many holders at once.                                                                                          |
| `prefix`       | `semaphore(client, …)`          | `"semaphore"` | Key namespace; keys are `<prefix>:<id>`.                                                                           |
| `leaseMs`      | `semaphore` / `acquire` / `run` | `60000`       | How long a slot is held without an `extend`. With `run` it is also the renewal window.                             |
| `retries`      | `acquire` / `run`               | `0`           | Attempts when every slot is taken. `0` fails fast.                                                                 |
| `retryDelayMs` | `acquire` / `run`               | `100`         | Delay between retries.                                                                                             |
| `heartbeatMs`  | `run`                           | `leaseMs / 4` | Renewal interval while `fn` runs. Must be at most half of `leaseMs` when set explicitly. `false` disables renewal. |
| `onRenewError` | `run`                           | none          | Called when a renewal round trip fails. Not a lost slot.                                                           |

## When You Don’t Need This

[Section titled “When You Don’t Need This”](#when-you-dont-need-this)

* **One caller at a time.** That is [`lock`](/benni/primitives/lock/).
* **Limiting request rate.** That is [`ratelimit`](/benni/primitives/ratelimit/). Concurrency and rate are independent; you often want both.
* **Concurrency inside one process.** `p-limit` is in-memory and free. This costs a round trip per acquire, which only buys you something when the limit spans processes.
* **Work that should be queued, not rejected.** If callers must eventually all run, in order, use the [queue](/benni/primitives/queue/) and set its worker concurrency.

## See Also

[Section titled “See Also”](#see-also)

* [Distributed Lock](/benni/primitives/lock/)
* [Rate Limiting](/benni/primitives/ratelimit/)
* [Budget](/benni/primitives/budget/) for spend limits

# Bun And Deno

> Bun uses Bun's built-in Redis client. Deno uses the Node adapter through npm compatibility.

Bun is supported through Bun’s built-in Redis client. Deno uses the Node adapter through npm compatibility.

## Bun

[Section titled “Bun”](#bun)

```ts
import { benni } from "benni";
import { bun } from "benni/bun";
import * as schema from "./schema";


const client = await bun({
  url: process.env.REDIS_URL ?? "redis://127.0.0.1:6379"
});


export const redis = benni(client, { schema });
```

[Pub/Sub](/benni/data-structures/pubsub/) needs no setup: the Bun adapter can lease a subscriber connection, so `redis.pubsub.channel(...).subscribe(...)` works on the bound client:

```ts
const subscription = await redis.pubsub
  .channel(schema.userEvents)
  .subscribe((message) => { /* ... */ });


await subscription.unsubscribe();
```

Channel subscriptions only, though. The Bun subscriber deliberately omits `psubscribe` because it is broken in Bun 1.3.14 (it hangs rather than resolving), so `redis.pubsub.pattern(...).subscribe(...)` throws `TypeError` on Bun instead of deadlocking. Subscribe to the individual channels until Bun ships a fix, or run pattern subscriptions on the [Node adapter](/benni/runtime/node/). Publishing is unaffected: it is one stateless `PUBLISH` on the bound client.

The Bun adapter supports [sessions](/benni/advanced/sessions/), so `redis.session()` and `redis.watch()` work: each session is a fresh Bun Redis client with reconnection and offline queueing disabled, and closing it rejects an in-flight blocking read promptly.

The Bun adapter runs the same Redis contract suite as the Node adapter against a real server:

```sh
BENNI_REDIS_URL=redis://127.0.0.1:6379 pnpm test:bun
```

## Deno

[Section titled “Deno”](#deno)

Deno needs no dedicated adapter: it runs node-redis directly through npm compatibility, which gives full Redis 8 command support and reuses the same adapter Node uses. Use the **Node adapter** with `npm:` specifiers:

```ts
// deno.json import map, or inline npm: specifiers
import { benni } from "npm:benni";
import { node } from "npm:benni/node";


const client = await node({ url: "redis://127.0.0.1:6379" });
export const redis = benni(client, { schema });
```

Deno resolves `redis` through its own `npm:` specifiers, so Benni’s optional `redis` peer dependency (an npm concern) does not apply. A Deno-native adapter over a JSR client such as `@redis/redis` is a possible future addition, but it would only be a different *engine* behind the same core.

If you prefer a different client entirely, the portable seam is the core `RedisClient` interface:

```ts
type RedisClient = {
  send(command: RedisCommand): Promise<RedisReply>;
  pipeline(commands: readonly RedisCommand[]): Promise<RedisReply[]>;
  transaction?(commands: readonly RedisCommand[]): Promise<RedisReply[]>;
  session?(): Promise<RedisSession>;
  subscriber?(): Promise<RedisSubscriber>;
  close(): Promise<void>;
};
```

If you already have a Deno Redis client, an adapter can implement that interface and then pass it to `benni(client)`. `transaction`, `session`, and `subscriber` are optional, and each one gates a feature rather than the whole client: an adapter that omits `session` still works, but `redis.session()` and `redis.watch()` throw `TypeError: Redis client does not support sessions` until it implements one, and an adapter that omits `subscriber` can still publish while `redis.pubsub.channel(...).subscribe(...)` throws. See [Sessions](/benni/advanced/sessions/) for the connection role a session fills, and [Pub/Sub](/benni/data-structures/pubsub/) for the subscriber one.

# Edge (Upstash / HTTP)

> Run the same typed Benni API on serverless and edge runtimes over Upstash's REST protocol, with nothing but fetch.

The `benni/upstash` adapter speaks the [Upstash REST protocol](https://upstash.com/docs/redis/features/restapi) over HTTP, so the **same typed Benni API** runs on serverless and edge runtimes (Cloudflare Workers, Vercel Edge, Fastly, Deno Deploy) with nothing but `fetch`. It has **zero dependencies**.

```ts
import { benni } from "benni";
import { upstash } from "benni/upstash";
import * as schema from "./schema";


const client = upstash({
  url: process.env.UPSTASH_REDIS_REST_URL as string,
  token: process.env.UPSTASH_REDIS_REST_TOKEN as string
});


export const redis = benni(client, { schema });
```

There is no connection to open, so `upstash` is synchronous (no `await`). Command arrays are `POST`ed directly to the REST endpoint; `pipeline` uses `/pipeline` and `redis.multi()` uses `/multi-exec` (atomic `MULTI`/`EXEC`).

## What works and what doesn’t

[Section titled “What works and what doesn’t”](#what-works-and-what-doesnt)

HTTP is stateless: one request, one response, no persistent exclusive connection. So the adapter serves the whole **command surface** but not the features that need a held connection:

| Works over HTTP                                                                                         | Not available over HTTP                                                                               |
| ------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------- |
| All typed data-structure stores (`hash`, `kv`, `set`, `list`, `zset`, `stream`, `bitmap`, `geo`, `hll`) | [Sessions](/benni/advanced/sessions/) via `redis.session()`                                           |
| `SCAN`/`HSCAN`/`SSCAN`/`ZSCAN` (the cursor rides in the command)                                        | Blocking commands (`BLPOP`, `BRPOP`, `BLMOVE`, `BZPOPMIN`/`MAX`, `XREAD BLOCK`)                       |
| Lua scripts, `BITFIELD`, geo, HyperLogLog                                                               | `WATCH`-based optimistic transactions via `redis.watch()`                                             |
| `redis.multi()` (atomic `/multi-exec`)                                                                  | [Pub/Sub](/benni/data-structures/pubsub/) **subscribing** (there is no subscriber connection to hold) |
| Pub/Sub **publishing** (`PUBLISH` is one stateless command)                                             |                                                                                                       |

`redis.session()` and `redis.watch()` throw a clear `TypeError` on this client, because the adapter deliberately omits `session`. `redis.pubsub.channel(...).subscribe(...)` throws the same way, because it omits `subscriber` for the same reason. When you need those, use a TCP adapter ([Node](/benni/runtime/node/) or [Bun](/benni/runtime/bun-and-deno/)) on a long-lived server.

Publishing is the useful half on the edge, and it needs nothing held open. An edge handler can fan an event out to long-lived workers that subscribe over TCP:

```ts
await redis.pubsub.channel(userEvents).publish({ id: "42", action: "created" });
```

Binary (`Uint8Array`) command arguments are not supported over REST; use the `bytes()` codec, which stores base64 strings, or a TCP adapter.

## Any Upstash-REST-compatible endpoint

[Section titled “Any Upstash-REST-compatible endpoint”](#any-upstash-rest-compatible-endpoint)

The adapter is not tied to Upstash’s hosted service. It works against anything that speaks the same protocol, including [`serverless-redis-http`](https://github.com/hiett/serverless-redis-http) (SRH), a self-hostable proxy you can run in front of a plain Redis for local development or CI:

```sh
docker run -p 8079:80 \
  -e SRH_MODE=env -e SRH_TOKEN=example_token \
  -e SRH_CONNECTION_STRING="redis://host.docker.internal:6379" \
  hiett/serverless-redis-http
```

```ts
const client = upstash({
  url: "http://127.0.0.1:8079",
  token: "example_token"
});
```

Benni runs the same shared client-contract suite that pins the Node and Bun adapters against SRH over HTTP, so the typed stores behave identically to a TCP connection (minus the session-only features above).

### A failed transaction may not carry a Redis error

[Section titled “A failed transaction may not carry a Redis error”](#a-failed-transaction-may-not-carry-a-redis-error)

One difference the contract suite does record, because it is the endpoint’s choice rather than Benni’s. Over REST a service sits in front of Redis and decides what a failed `MULTI`/`EXEC` looks like on the wire, and SRH answers with a 5xx carrying nothing:

```text
POST /pipeline    [["PING"],["ZADD","str","1","member"]]
  -> 200  [{"result":"PONG"},{"error":"WRONGTYPE Operation against a key..."}]


POST /multi-exec  [["PING"],["ZADD","str","1","member"]]
  -> 500  (no body)
```

With no reply to read, `redis.multi().exec()` rejects with a transport `Error` rather than a [`RedisServerError`](/benni/api/errors/). Benni will not invent a `.code` from a gateway’s status line, because that would hand you a `RedisServerError` for what might equally be an upstream outage.

What this does and does not change:

* A failed transaction **always rejects**, on every adapter. It never resolves as though it committed.
* Single commands and pipelines are unaffected: both carry `{ "error": ... }`, so both normalize to `RedisServerError` with the code parsed.
* Code that branches on `.code` should confirm the error is a `RedisServerError` first, which is the rule everywhere anyway:

```ts
try {
  await redis.multi().add(["ZADD", key, "1", "member"], numberReply).exec();
} catch (error) {
  if (error instanceof RedisServerError && error.code === "WRONGTYPE") {
    // Redis said no, and said why
  } else {
    // the transaction failed without an attributable reply: retry or surface it
  }
}
```

A hosted endpoint may well return a readable error where SRH does not. The contract suite asserts only what the transport can actually guarantee, so write the `catch` above and it is correct against both.

# ioredis

> Use Benni with the ioredis client you already run, including adopting an existing instance, so adopting Benni is not a client migration.

`benni/ioredis` runs the whole typed API on [ioredis](https://www.npmjs.com/package/ioredis), the most widely deployed Redis client for Node. If your app already uses ioredis, this is the adapter to pick: **you do not have to swap Redis clients to use Benni.**

```sh
pnpm add benni ioredis
```

```ts
import { benni } from "benni";
import { ioredis } from "benni/ioredis";
import * as schema from "./schema";


const client = await ioredis(process.env.REDIS_URL);


export const redis = benni(client, { schema });
```

## Three ways in

[Section titled “Three ways in”](#three-ways-in)

A URL:

```ts
const client = await ioredis("redis://127.0.0.1:6379");
```

Any ioredis options (`host`, `port`, `password`, `tls`, `sentinels`, …):

```ts
const client = await ioredis({
  host: process.env.REDIS_HOST,
  port: 6379,
  password: process.env.REDIS_PASSWORD
});
```

Or an ioredis instance you already have, which is the important one:

```ts
import Redis from "ioredis";


const existing = new Redis(process.env.REDIS_URL ?? "redis://127.0.0.1:6379"); // yours, already configured
const client = await ioredis(existing);
```

Adopting means Benni shares the connection you already tuned, monitor, and pool. There is no second client, no second connection budget, and no migration: you can start typing one keyspace and leave the rest of your app calling `existing` directly.

## Who owns the connection

[Section titled “Who owns the connection”](#who-owns-the-connection)

An adopted client is **borrowed**. `client.close()` shuts down the sessions and subscriber connections Benni leased, and leaves your client open, because you still own its lifetime:

```ts
await client.close();  // Benni's leases are gone
await existing.quit(); // you close yours, when you're ready
```

A client Benni created from a URL or options is **owned**, and `close()` quits it for you.

One consequence worth knowing: Benni attaches an `"error"` listener only to clients it created. An adopted client keeps whatever error handling you gave it, and Benni will not silently swallow errors on a client it does not own. Make sure yours has a listener, or an idle network blip will crash the process (that is ioredis behaviour, not Benni’s).

### `keyPrefix` is not supported

[Section titled “keyPrefix is not supported”](#keyprefix-is-not-supported)

`ioredis({ keyPrefix })`, and adopting a client that sets it, both throw. ioredis rewrites key *arguments* but leaves `SCAN`/`MATCH` patterns alone, so a prefixed client stores at `<prefix><key>` while `schema.key()` and every scan still say `<key>`. Scans would return nothing at all, without an error.

Benni’s schemas already own key naming, so put the prefix there instead:

```ts
const users = hash(prefix + "user", { name: string() });
```

## What’s supported

[Section titled “What’s supported”](#whats-supported)

Everything. ioredis speaks RESP2, whose flat reply shapes are exactly what the typed stores decode, so replies pass through with no normalization:

| Feature                                                               | Supported |
| --------------------------------------------------------------------- | --------- |
| Typed stores, transactions, scripts                                   | Yes       |
| [Sessions](/benni/advanced/sessions/): blocking commands, `WATCH`     | Yes       |
| [Pub/Sub](/benni/data-structures/pubsub/) subscribe                   | Yes       |
| Pattern subscriptions (`psubscribe`)                                  | Yes       |
| [Primitives](/benni/primitives/queue/): queue, cache, lock, ratelimit | Yes       |

Sessions duplicate the connection with reconnection disabled and the offline queue off, so a drop rejects in-flight and subsequent commands instead of silently reconnecting, which would lose `WATCH` state and blocked reads. Closing a session calls `disconnect()` rather than `quit()`, so an in-flight blocking read is rejected at once instead of waiting out its server-side timeout. The parent client tracks live sessions and subscribers and force-closes any survivors.

Pub/Sub delivers every subscription through one connection-level event, so the adapter routes by channel and pattern name internally. You just subscribe:

```ts
const subscription = await redis.pubsub.channel(userEvents).subscribe((message) => {
  console.log(message.action);
});
```

## ioredis or node-redis?

[Section titled “ioredis or node-redis?”](#ioredis-or-node-redis)

Both adapters expose the identical typed API and pass the same client-contract suite, so this is a question about your app, not about Benni:

* **Already on ioredis.** Use `benni/ioredis`, and adopt your existing instance. Zero migration.
* **Already on node-redis.** Use [`benni/node`](/benni/runtime/node/).
* **Greenfield.** Either works. `redis` (node-redis) is the officially maintained client and tracks new Redis 8 commands soonest; ioredis has the larger install base and richer cluster/sentinel configuration.

You can switch adapters later by changing one import; the schemas, stores, and primitives above it do not move.

## Cluster and Sentinel

[Section titled “Cluster and Sentinel”](#cluster-and-sentinel)

Sentinel configuration works, since it is just ioredis options:

```ts
const client = await ioredis({
  sentinels: [{ host: "localhost", port: 26379 }],
  name: "mymaster"
});
```

Cluster splits the responsibility. Adopt an `ioredis.Cluster` instance and ioredis does the routing (topology, `MOVED`/`ASK`, failover); Benni never had a transport of its own and does not try to. What Benni adds on top is slot **co-location**: schemas declare where their hash tag goes, the compiler rejects multi-key calls whose tags provably disagree, and `benni(client, { cluster: true })` catches the rest before they are sent. See [Redis Cluster](/benni/advanced/cluster/) for the layouts and the guard.

# Node.js Setup

> Node.js is supported through the redis package.

Node.js is supported through the `redis` package.

```ts
import { benni } from "benni";
import { node } from "benni/node";
import * as schema from "./schema";


const client = await node({
  url: process.env.REDIS_URL ?? "redis://127.0.0.1:6379"
});


export const redis = benni(client, { schema });
```

Close the client when your process or test is done:

```ts
await client.close();
```

`close()` is safe to call twice, and it is final: once it has run, `redis.session()` and a fresh Pub/Sub subscribe reject with “client is closed” rather than quietly opening a connection nothing will ever close. The same holds for `benni/ioredis` and `benni/bun`.

The Node adapter defaults to RESP2 replies because the typed stores validate Redis reply shapes such as arrays, maps, numbers, strings, and nulls. You can pass normal `redis` client options to `node`.

The Node adapter supports [sessions](/benni/advanced/sessions/), so `redis.session()` and `redis.watch()` work: each session duplicates the connection with reconnection disabled and closes by destroying the socket, which rejects an in-flight blocking read promptly rather than waiting out its timeout. The parent client tracks live sessions and force-closes any survivors when you close it.

[Pub/Sub](/benni/data-structures/pubsub/) needs no setup either: the adapter can also lease a subscriber connection, so `redis.pubsub.channel(...).subscribe(...)` and the pattern form both work out of the box. Benni leases that connection on the first subscribe and closes it when the last subscription goes away; the parent `client.close()` force-closes it too if you skip `redis.pubsub.close()`. Pattern subscriptions work here and on [`benni/ioredis`](/benni/runtime/ioredis/); Bun is the one adapter without them.

Already using ioredis instead? [`benni/ioredis`](/benni/runtime/ioredis/) gives the identical typed API and can adopt your existing client, so you do not have to switch Redis clients to use Benni.