Group.Transport.Outbox behaviour (Group v0.3.1)

Copy Markdown View Source

Lossy per-shard outboxes for sideband replica transports.

This module is an implementation helper, not a replacement for Group.Transport. Distribution can continue sending directly with :erlang.send_nosuspend/3. A sideband adapter delegates outgoing/5 to push/5, which performs only a local send/2 to the matching shard outbox.

Each outbox batches messages by target node outside the Group shard. Expired messages and batches rejected by the backend are deliberately dropped; anti-entropy repairs them. Backends may perform bounded blocking work in send_batch/4 because they run in the outbox rather than a Group process.

A backend using this helper implements:

@behaviour Group.Transport.Outbox

def init_outbox(group, shard, opts), do: {:ok, backend_state}

def send_batch(target_node, messages, deadline, backend_state) do
  # Return promptly once `deadline` has passed. It is safe to drop.
  {:ok, backend_state}
end

The backend must pass only complete logical messages to Group.Transport.incoming_batch/4.

Options

  • :outbox_batch_size - maximum logical messages collected per flush, default 64
  • :outbox_batch_bytes - approximate external-term bytes collected per flush, default 1_048_576
  • :outbox_flush_interval - maximum batching delay in milliseconds, default 1
  • :outbox_deadline - maximum useful residence time for an outgoing message in milliseconds, default 100
  • :outbox_max_messages - maximum accepted messages across the local mailbox, pending batch, and backend send, default 1_024

Admission bounds mailbox growth by message count. A backend must also put a finite bound on every socket enqueue or write it performs. Exact snapshot messages are independently bounded by :replicated_snapshot_chunk_target_bytes. Other logical messages or a whole batch may still exceed :outbox_batch_bytes; a transport with a smaller finite frame size must segment and completely reassemble those batches before local delivery.

Summary

Types

backend_state()
message()
outgoing_result()

Callbacks

init_outbox(group, shard, opts)
send_batch(target_node, messages, deadline, backend_state)

Functions

child_spec(opts)

Returns a supervisor child specification for one outbox per Group shard.

push(group, target_node, shard, message, opts)

Pushes a replica message into its shard's local outbox.

Types

backend_state()

@type backend_state() :: term()

message()

@type message() :: Group.Transport.message()

outgoing_result()

@type outgoing_result() :: Group.Transport.outgoing_result()

Callbacks

init_outbox(group, shard, opts)

@callback init_outbox(group :: atom(), shard :: non_neg_integer(), opts :: keyword()) ::
  {:ok, backend_state()}

send_batch(target_node, messages, deadline, backend_state)

@callback send_batch(
  target_node :: node(),
  messages :: [message()],
  deadline :: integer(),
  backend_state()
) ::
  {outgoing_result(), backend_state()}

Functions

child_spec(opts)

Returns a supervisor child specification for one outbox per Group shard.

:name, :num_shards, and :backend are required. All options are passed unchanged to backend.init_outbox/3.

push(group, target_node, shard, message, opts)

Pushes a replica message into its shard's local outbox.

This operation is local and nonblocking and performs no backend or socket work. :ok only means the outbox was available; the message may later expire or be dropped under transport pressure. A concurrently terminating outbox can also lose an accepted message, which anti-entropy repairs.