A durable, distributed gen_server backed by a pluggable storage backend.
A dgen_server is an abstract entity composed of durable state and operations on
that state. The state lives in the configured backend (default: FoundationDB)
and the operations are defined by a callback module implementing the
dgen_server behaviour. This allows a gen_server to outlive any single Erlang
process, node, or cluster.
Zero or more Erlang processes may act on a dgen_server at any time. Processes
with consume enabled consume messages from the durable queue and invoke
callbacks; processes without it only publish messages.
Options
The following options may be passed via the Opts proplist:
tenant(required) -{DbHandle, Dir}pair identifying the backend subspace.consume(defaulttrue) - whether this process consumes messages from the queue.reset(defaultfalse) - whentrue, re-initialise the durable state even if it already exists.dead_letter_threshold(defaultinfinity) - 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}(forcallmessages), and the optionalhandle_dead_letter/2callback is invoked.infinity(the default) disables dead-lettering entirely.consume_k(default1) - the maximum number of queued messages a consumer processes per transaction (the batch size). Each consume cycle peeks up toconsume_kmessages, processes them one at a time while carrying the durable state in memory, and commits the whole batch in a single transaction (the state is read once and written once). It then immediately re-arms itself, so a consumer that has work keeps draining batches.Raising
consume_kamortises the fixed per-transaction cost (read version, state read/write, commit) across more messages, increasing throughput when the per-message callback work is small. The cost is larger transactions and a coarser unit of progress (a batch is all-or-nothing; on a conflict the whole batch retries). Because a busy consumer keeps draining, a largerconsume_kalso tends to keep a single node as the active consumer, which is useful when the callback's identity matters — for example an election-style callback where you want to minimise churn in who processes successive messages.See "consume_k and inlining" below for how
consume_k > 1changes call handling.lock_timeout(defaultinfinity) - maximum number of milliseconds a distributed lock may be held before other consumers treat it as stale. When a callback returns{lock, State},dgen_serversets a timestamped lock key in the backend that pauses all other consumers whilehandle_locked/4runs. Under normal operation the lock holder clears the key itself beforehandle_lockedreturns. If the holder is killed (SIGKILL, VM abort) the lock persists until another consumer detects staleness: it re-checkslock_timeoutms after the lock was set and clears it if that deadline has passed.infinity(the default) disables stale-lock detection entirely — a dead holder will permanently block all consumers. Set this to a value safely larger than the worst-casehandle_lockedduration for your callback.
consume_k and inlining
A message can reach a callback by two routes:
- Through the durable queue.
cast/2and (normally)call/2enqueue the message; a consumer later peeks it as part of aconsume_k-sized batch and processes it inside the batch's single transaction. This is the ordered, durable path, and the only path whenconsume_k > 1. - Inline. As a latency optimisation, when
consume_k =:= 1acall/2whose queue is currently empty and unlocked is processed immediately, in the caller's own request transaction, instead of being enqueued and waited on. The result is identical to processing it through the queue; it just skips the enqueue/await round-trip.
When consume_k > 1, inlining is disabled: every call/2 goes through the
queue and the batched consume loop, so consume_k is always in effect. Use this
when you want all processing funnelled through the batched, single-consumer loop —
for example to keep a stable consumer identity (an election-style callback that
would otherwise see successive messages committed by different nodes). With the
default consume_k =:= 1, inlining is enabled and a contention-free call/2 is
served on the fast path.
Independently of consume_k, priority_call/2 and priority_cast/2 always
bypass the queue (and locks) and are handled immediately. They are an explicit
escape hatch for urgent, usually read-only work; they trade away ordering with
respect to queued messages, so prefer call/2/cast/2 for anything that must be
ordered 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 executed after the transaction commits. The
argument is the module state as committed by that transaction — note that under
batched consumption (consume_k > 1) all of a batch's actions run after the
batch's single commit and each receives the state as of the end of the batch,
not the state at the message that produced the action.
Module State
The module that implements the dgen_server behaviour may define any term to
serve as the State. dgen_server will encode this state for writing to the
database. We encourage you to structure your state to fit in with this
encoding scheme, which will yield performance benefits.
- term (fallback) -
term_to_binary, chunked into 100KB values. - assigns map - map with all atom keys; each entry stored at a
separate DB key using
atom_to_binary(Key)in the path. - component list - list of maps where every item has an atom
idkey with a binary value; each item 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.
Summary
Functions
Sends a synchronous call request via the durable queue. Default timeout 5000ms.
Sends a synchronous call request via the durable queue.
Sends an asynchronous cast request to the dgen_server's durable queue.
Sends a batch of cast requests to the dgen_server's durable queue atomically.
Kills the dgen_server, deleting all durable state, queue items, and waiting
call keys. The process exits with Reason.
Returns a closure for atomically casting a message from within the caller's own backend transaction.
Sends a call that bypasses the durable queue and is handled immediately.
Like priority_call/2 but with an explicit timeout.
Sends a cast that bypasses the durable queue and is handled immediately.
Starts a dgen_server process without linking.
Starts a dgen_server process without linking, registered as Reg.
Starts a dgen_server process linked to the calling process.
Starts a dgen_server process linked to the calling process, registered as Reg.
Types
-type action() :: fun().
-type db_ctx() :: #{db := dgen_backend:tenant(), tuid := tuid()}.
-type event_type() :: {call, from()} | cast | info.
-type from() :: term().
-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}.
-type lock_ret() :: {lock, state()}.
-type option() :: {tenant, dgen_backend:tenant()} | {consume, boolean()} | {reset, boolean()} | {cache, boolean()} | {dead_letter_threshold, pos_integer() | infinity} | {lock_timeout, pos_integer() | infinity} | gen_server:start_opt().
-type options() :: [option()].
-type server() :: gen_server:server_ref().
-type start_ret() :: gen_server:start_ret().
-type state() :: term().
-type tuid() :: tuple().
-type tx_ctx() :: #{td := dgen_backend:tenant(), tuid := tuid()}.
Callbacks
-callback handle_cast(Msg :: term(), State :: state()) -> noreply_ret() | lock_ret() | stop_ret().
-callback handle_dead_letter(Msg :: term(), AttemptCount :: non_neg_integer()) -> any().
-callback handle_info(Info :: term(), State :: state()) -> noreply_ret() | stop_ret().
-callback handle_info_tx(TxCtx :: tx_ctx(), Info :: term(), State :: state()) -> noreply_ret() | stop_ret().
-callback handle_locked(DbCtx :: db_ctx(), EventType :: event_type(), Msg :: term(), State :: state()) -> reply_ret() | noreply_ret() | stop_ret().
Functions
Sends a synchronous call request via the durable queue. Default timeout 5000ms.
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: Default5000. Timeout in milliseconds, orinfinity.
Sends an asynchronous cast request to the dgen_server's durable queue.
Sends a batch of cast requests to the dgen_server's durable queue atomically.
-spec code_change(term(), internalstate(), term()) -> {ok, internalstate()}.
-spec handle_call(term(), gen_server:from(), internalstate()) -> {reply, term(), internalstate()}.
-spec handle_cast(term(), internalstate()) -> {noreply, internalstate()} | {stop, term(), internalstate()}.
-spec handle_info(term(), internalstate()) -> {noreply, internalstate()} | {stop, term(), internalstate()}.
-spec init(term()) -> {ok, internalstate()} | {error, term()}.
Kills the dgen_server, deleting all durable state, queue items, and waiting
call keys. The process exits with Reason.
-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.
-spec outbox_cast(server(), timeout()) -> fun((dgen_backend:tx(), term()) -> ok).
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.
Like priority_call/2 but with an explicit timeout.
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.
Starts a dgen_server process without linking.
See start_link/3 for details on Mod, Arg, and Opts.
-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.
Starts a dgen_server process linked to the calling process.
Modis the callback module implementing thedgen_serverbehaviour.Argis passed toMod:init/1.Optsis a proplist that must include{tenant, {Db, Dir}}and may includeconsumeandreset.
-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.
-spec terminate(term(), internalstate()) -> ok.