# `dgen_server`
[🔗](https://github.com/foundationdb-beam/dgen/blob/main/src/dgen_server.erl#L1)

A durable, distributed gen_server backed by a pluggable storage backend.

A dgen_server is an abstract entity made of durable state and the operations on
that state. The state lives in the configured backend (default: FoundationDB),
and a callback module implementing the `dgen_server` behaviour defines the
operations. This lets a gen_server outlive any single Erlang process, node, or
cluster.

Zero or more Erlang processes may act on a dgen_server at any time. A process
with `consume` enabled takes messages off the durable queue and invokes the
callbacks. A process without it only publishes messages.

## Options

The following options may be passed via the `Opts` proplist:

- `tenant` (required) - `{DbHandle, Dir}` pair identifying the backend subspace.
- `consume` (default `true`) - whether this process consumes messages from
  the queue.
- `reset` (default `false`) - when `true`, re-initialise the durable state
  even if it already exists.
- `dead_letter_threshold` (default `infinity`) - number of consecutive
  processing failures before a message is treated as a dead letter. When a
  message has failed this many times it is moved to the dead-letter queue, the
  caller raises `{dead_letter, N}` (for `call` messages), and the optional
  `handle_dead_letter/2` callback is invoked. `infinity` (the default) disables
  dead-lettering entirely.
- `consume_k` (default `1`) - the batch size, meaning the maximum number of
  queued messages a consumer processes per transaction. Each consume cycle peeks
  up to `consume_k` messages and processes them one at a time, carrying the
  durable state in memory. It then commits the whole batch in a **single**
  transaction, so the state is read once and written once. The consumer then
  starts the next cycle right away, so a consumer that has work keeps draining
  batches.

  Raising `consume_k` spreads the fixed cost of a transaction (read version,
  state read and write, commit) across more messages. That raises throughput when
  each callback does little work. The cost is larger transactions and a coarser
  unit of progress. A batch is all-or-nothing, so a conflict retries the whole
  batch. Because a busy consumer keeps draining, a larger `consume_k` also tends
  to keep a **single** node as the active consumer. That helps when it matters
  which node runs the callback, for example in an election-style callback where
  you want the same node to keep processing successive messages.

  See "consume_k and inlining" below for how `consume_k > 1` changes call
  handling.
- `lock_timeout` (default `infinity`) - maximum number of milliseconds a
  distributed lock may be held before other consumers treat it as stale.  When
  a callback returns `{lock, State}`, `dgen_server` sets a timestamped lock key
  in the backend. The key pauses all other consumers while `handle_locked/4`
  runs. Normally the lock holder clears the key itself once `handle_locked`
  returns.  If the holder is killed (SIGKILL, VM abort), the lock stays until
  another consumer notices it is stale. That consumer re-checks the lock
  `lock_timeout` ms after it was set and clears it if the deadline has passed.
  We call this busting the lock.  `infinity` (the default) turns off stale-lock
  detection entirely, so a dead holder blocks all consumers forever.

  Busting is a guess, based only on time, that the holder is dead. Nothing stops
  a holder that is merely slow, so the guess can be wrong.  The lock is
  **fenced** against that. Each acquisition stamps the key with a unique token,
  and a holder may release the lock or commit the result of `handle_locked/4`
  only while the key still carries its token.  So if a holder is busted mid-run,
  its commit is refused instead of overwriting what its successor has written
  since, and it cannot release the successor's lock.

  A fenced section is not lost.  The transaction that refuses the commit also
  puts the message that took the lock back on the queue, and the process exits
  with `{lock_fenced, requeued}`. Whichever consumer holds the lock next retries
  the message. Because `handle_locked/4` runs outside a transaction, its
  **external side effects have already happened and will happen again on the
  retry**. It must be safe to run twice.

  A wrong guess now costs a retry rather than silent data loss. So set this to
  how quickly you want to recover from a dead holder, rather than above the
  worst-case `handle_locked` duration.

## consume_k and inlining

A message can reach a callback by two routes:

1. **Through the durable queue.** `cast/2`, and usually `call/2`, enqueue the
   message. A consumer later peeks it as part of a `consume_k`-sized batch and
   processes it inside that batch's single transaction. This is the ordered,
   durable path, and the only path when `consume_k > 1`.
2. **Inline.** To save latency when `consume_k =:= 1`, a `call/2` that finds
   its queue empty and unlocked is processed *immediately*, in the caller's own
   request transaction, instead of being enqueued and waited on. The result is
   the same as going through the queue. It just skips the round-trip of
   enqueueing and waiting.

When **`consume_k > 1`, inlining is disabled**. Every `call/2` goes through the
queue and the batched consume loop, so `consume_k` always applies. Use this
when you want all processing to go through the batched, single-consumer loop,
for example to keep the same node as consumer. An election-style callback would
otherwise see successive messages committed by different nodes. With the
default `consume_k =:= 1`, inlining is enabled and a `call/2` with no
contention takes the fast path.

Whatever `consume_k` is, `priority_call/2` and `priority_cast/2` **always**
bypass the queue and any lock, and are handled immediately. They are an
explicit escape hatch for urgent, usually read-only work. They give up ordering
relative to queued messages, so prefer `call/2` and `cast/2` for anything that
must stay in order with the rest of the stream.

## Callbacks

- `init/1` - return `{ok, State}` or `{ok, Tuid, State}`.
- `handle_call/3` - return `{reply, Reply, State}` or
  `{reply, Reply, State, Actions}`.
- `handle_cast/2` - return `{noreply, State}` or `{noreply, State, Actions}`.
- `handle_info/2` - return `{noreply, State}` or `{noreply, State, Actions}`.
- `handle_dead_letter/2` (optional) - called after a message is dead-lettered.
  Receives `(Msg, AttemptCount)`. Return value is ignored. Useful for custom
  alerting or metrics.

`Actions` is a list of 1-arity funs that run after the transaction commits.
Each one receives the module state as that transaction committed it. Under
batched consumption (`consume_k > 1`), all of a batch's actions run after its
single commit. Each receives the state as of the **end of the batch**, not the
state right after the message that produced the action.

## Module State

The module that implements the `dgen_server` behaviour may use any term as its
State. `dgen_server` encodes this state to write it to the database. If you
shape your state to fit the encodings below, updates get cheaper.

- **term** (fallback) - `term_to_binary`, chunked into 100KB values.
- **assigns map** - a map whose keys are all atoms. Each entry is stored at its
  own DB key, with `atom_to_binary(Key)` in the path.
- **component list** - a list of maps where every item has an atom `id` key
  with a binary value. Each item is stored separately, ordered by a fractional
  index embedded in the DB key.

The encoding is applied recursively. For example, an assigns map whose value
is a component list will nest both encodings in the key path.

*This documentation is LLM-generated. See the AI disclosure in `README.md`.*

# `action`

```erlang
-type action() :: fun().
```

# `db_ctx`

```erlang
-type db_ctx() :: #{db := dgen_backend:tenant(), tuid := tuid()}.
```

# `event_type`

```erlang
-type event_type() :: {call, from()} | cast | info.
```

# `from`

```erlang
-type from() :: term().
```

# `init_ret`

```erlang
-type init_ret() :: {ok, state()} | {ok, tuid(), state()} | {error, term()}.
```

# `internalstate`

```erlang
-type internalstate() ::
          #state{tenant :: dgen_backend:tenant(),
                 mod :: atom(),
                 tuid :: tuple(),
                 watch :: undefined | dgen_backend:future(),
                 cache :: boolean(),
                 mod_state_cache ::
                     undefined |
                     {dgen_backend:versionstamp() | dgen_backend:future(),
                      {ok, term()} | {error, not_found}},
                 cache_misses :: non_neg_integer(),
                 dead_letter_threshold :: pos_integer() | infinity,
                 consume_k :: pos_integer(),
                 lock_timeout :: pos_integer() | infinity,
                 lock_token :: undefined | binary(),
                 locked_envelope :: undefined | term()}.
```

# `lock_ret`

```erlang
-type lock_ret() :: {lock, state()}.
```

# `noreply_ret`

```erlang
-type noreply_ret() :: {noreply, state()} | {noreply, state(), [action()]}.
```

# `option`

```erlang
-type option() ::
          {tenant, dgen_backend:tenant() | function()} |
          {consume, boolean()} |
          {reset, boolean()} |
          {cache, boolean()} |
          {dead_letter_threshold, pos_integer() | infinity} |
          {lock_timeout, pos_integer() | infinity} |
          gen_server:start_opt().
```

# `options`

```erlang
-type options() :: [option()].
```

# `reply_ret`

```erlang
-type reply_ret() :: {reply, term(), state()} | {reply, term(), state(), [action()]}.
```

# `server`

```erlang
-type server() :: gen_server:server_ref().
```

# `start_ret`

```erlang
-type start_ret() :: gen_server:start_ret().
```

# `state`

```erlang
-type state() :: term().
```

# `stop_ret`

```erlang
-type stop_ret() :: {stop, term(), state()} | {stop, term(), state(), [action()]}.
```

# `tuid`

```erlang
-type tuid() :: tuple().
```

# `tx_ctx`

```erlang
-type tx_ctx() :: #{td := dgen_backend:tenant(), tuid := tuid()}.
```

# `handle_call`
*optional* 

```erlang
-callback handle_call(Request :: term(), From :: from(), State :: state()) ->
                         reply_ret() | lock_ret() | stop_ret().
```

# `handle_call_tx`
*optional* 

```erlang
-callback handle_call_tx(TxCtx :: tx_ctx(), Request :: term(), From :: from(), State :: state()) ->
                            reply_ret() | lock_ret() | stop_ret().
```

# `handle_cast`
*optional* 

```erlang
-callback handle_cast(Msg :: term(), State :: state()) -> noreply_ret() | lock_ret() | stop_ret().
```

# `handle_cast_tx`
*optional* 

```erlang
-callback handle_cast_tx(TxCtx :: tx_ctx(), Msg :: term(), State :: state()) ->
                            noreply_ret() | lock_ret() | stop_ret().
```

# `handle_dead_letter`
*optional* 

```erlang
-callback handle_dead_letter(Msg :: term(), AttemptCount :: non_neg_integer()) -> any().
```

# `handle_info`
*optional* 

```erlang
-callback handle_info(Info :: term(), State :: state()) -> noreply_ret() | stop_ret().
```

# `handle_info_tx`
*optional* 

```erlang
-callback handle_info_tx(TxCtx :: tx_ctx(), Info :: term(), State :: state()) -> noreply_ret() | stop_ret().
```

# `handle_locked`
*optional* 

```erlang
-callback handle_locked(DbCtx :: db_ctx(), EventType :: event_type(), Msg :: term(), State :: state()) ->
                           reply_ret() | noreply_ret() | stop_ret().
```

# `init`

```erlang
-callback init(Args :: term()) -> init_ret().
```

# `call`

```erlang
-spec call(server(), term()) -> term().
```

Sends a synchronous call request via the durable queue. Default timeout 5000ms.

# `call`

```erlang
-spec call(server(), term(), timeout() | list()) -> term().
```

Sends a synchronous call request via the durable queue.

The request is enqueued durably and the caller blocks until a consumer
processes it and writes the reply, or until `Timeout` milliseconds elapse.

## Options

- `timeout`: Default `5000`. Timeout in milliseconds, or `infinity`.

# `cast`

```erlang
-spec cast(server(), term()) -> ok.
```

Sends an asynchronous cast request to the dgen_server's durable queue.

# `cast_k`

```erlang
-spec cast_k(server(), [term()]) -> ok.
```

Sends a batch of cast requests to the dgen_server's durable queue atomically.

# `code_change`

```erlang
-spec code_change(term(), internalstate(), term()) -> {ok, internalstate()}.
```

# `get_quid`

# `handle_call`

```erlang
-spec handle_call(term(), gen_server:from(), internalstate()) -> {reply, term(), internalstate()}.
```

# `handle_cast`

```erlang
-spec handle_cast(term(), internalstate()) ->
                     {noreply, internalstate()} | {stop, term(), internalstate()}.
```

# `handle_info`

```erlang
-spec handle_info(term(), internalstate()) ->
                     {noreply, internalstate()} | {stop, term(), internalstate()}.
```

# `init`

```erlang
-spec init(term()) -> {ok, internalstate()} | {error, term()}.
```

# `kill`

```erlang
-spec kill(server(), term()) -> ok.
```

Kills the dgen_server, deleting all durable state, queue items, and waiting
call keys. The process exits with `Reason`.

# `outbox_cast`

```erlang
-spec outbox_cast(server()) -> fun((dgen_backend:tx(), term()) -> ok).
```

Returns a closure for atomically casting a message from within the caller's
own backend transaction.

Call this before opening the transaction as a preparatory step. Bind the
result to `Cast` and call `Cast(Tx, Message)` inside the transaction to
enqueue the message without going through the dgen_server process. The queue
directory and identifier are captured internally and not exposed to the
caller.

## Backend coupling

This function is intended for callers that are already operating directly
with a backend transaction — for example, when a message must be enqueued
atomically alongside other writes in the same transaction. Using it means
intentionally stepping outside the gen_server abstraction: the caller takes
responsibility for managing the transaction lifetime and is coupled to the
configured backend. If you do not need to compose the enqueue with other
backend writes, prefer `cast/2` instead.

# `outbox_cast`

```erlang
-spec outbox_cast(server(), timeout()) -> fun((dgen_backend:tx(), term()) -> ok).
```

# `priority_call`

```erlang
-spec priority_call(server(), term()) -> term().
```

Sends a call that bypasses the durable queue and is handled immediately.

Use with caution: this breaks ordering guarantees with respect to queued
messages and ignores locks. Can be useful for snapshot reads.

# `priority_call`

```erlang
-spec priority_call(server(), term(), timeout()) -> term().
```

Like `priority_call/2` but with an explicit timeout.

# `priority_cast`

```erlang
-spec priority_cast(server(), term()) -> ok.
```

Sends a cast that bypasses the durable queue and is handled immediately.

Use with caution: this breaks ordering guarantees with respect to queued
messages and ignores locks.

# `start`

```erlang
-spec start(module(), term(), options()) -> start_ret().
```

Starts a dgen_server process without linking.

See `start_link/3` for details on `Mod`, `Arg`, and `Opts`.

# `start`

```erlang
-spec start(gen_server:server_name(), module(), term(), options()) -> start_ret().
```

Starts a dgen_server process without linking, registered as `Reg`.

See `start_link/3` for details on `Mod`, `Arg`, and `Opts`.

# `start_link`

```erlang
-spec start_link(module(), term(), options()) -> start_ret().
```

Starts a dgen_server process linked to the calling process.

- `Mod` is the callback module implementing the `dgen_server` behaviour.
- `Arg` is passed to `Mod:init/1`.
- `Opts` is a proplist that must include `{tenant, {Db, Dir}}` and may
  include `consume` and `reset`.

# `start_link`

```erlang
-spec start_link(gen_server:server_name(), module(), term(), options()) -> start_ret().
```

Starts a dgen_server process linked to the calling process, registered as `Reg`.

See `start_link/3` for details on `Mod`, `Arg`, and `Opts`.

# `terminate`

```erlang
-spec terminate(term(), internalstate()) -> ok.
```

---

*Consult [api-reference.md](api-reference.md) for complete listing*