dgen_server behaviour (DGen v0.4.0)

Copy Markdown View Source

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 (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 maximum number of queued messages a consumer processes per transaction (the batch size). Each consume cycle peeks up to consume_k messages, 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_k amortises 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 larger consume_k also 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 > 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 that pauses all other consumers while handle_locked/4 runs. Under normal operation the lock holder clears the key itself before handle_locked returns. If the holder is killed (SIGKILL, VM abort) the lock persists until another consumer detects staleness: it re-checks lock_timeout ms 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-case handle_locked duration for your callback.

consume_k and inlining

A message can reach a callback by two routes:

  1. Through the durable queue. cast/2 and (normally) call/2 enqueue the message; a consumer later peeks it as part of a consume_k-sized batch and processes it inside the batch's single transaction. This is the ordered, durable path, and the only path when consume_k > 1.
  2. Inline. As a latency optimisation, when consume_k =:= 1 a call/2 whose 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 id key 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

action()

-type action() :: fun().

db_ctx()

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

event_type()

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

from()

-type from() :: term().

init_ret()

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

internalstate()

-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_ret()

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

noreply_ret()

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

option()

-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().

options()

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

reply_ret()

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

server()

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

start_ret()

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

state()

-type state() :: term().

stop_ret()

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

tuid()

-type tuid() :: tuple().

tx_ctx()

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

Callbacks

handle_call(Request, From, State)

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

handle_call_tx(TxCtx, Request, From, State)

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

handle_cast(Msg, State)

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

handle_cast_tx(TxCtx, Msg, State)

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

handle_dead_letter(Msg, AttemptCount)

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

handle_info(Info, State)

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

handle_info_tx(TxCtx, Info, State)

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

handle_locked(DbCtx, EventType, Msg, State)

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

init(Args)

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

Functions

call(Server, Request)

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

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

call/3

-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(Server, Request)

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

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

cast_k(Server, Requests)

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

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

code_change(OldVsn, State, Extra)

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

get_quid(Tuple)

handle_call(Request, From, State)

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

handle_cast(Msg, State)

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

handle_info(Info, State)

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

init(Args)

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

kill(Server, Reason)

-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(Server)

-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(Server, Timeout)

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

priority_call(Server, Request)

-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(Server, Request, Timeout)

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

Like priority_call/2 but with an explicit timeout.

priority_cast(Server, Request)

-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(Mod, Arg, Opts)

-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(Reg, Mod, Arg, 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.

start_link(Mod, Arg, Opts)

-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(Reg, Mod, Arg, Opts)

-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(Reason, State)

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