Ankusa.Intake behaviour (ankusa v0.7.0)

The intake log behind wal: :dual (ADR 0002): a replicated log an acked hook is also confirmed in, so a lost host does not lose it.

An adapter implements this behaviour (Ankusa.Intake.Memory is the test double; Ankusa.Intake.Kafka ships in ankusa_kafka). Everything else is core: the writer's dual commit (Ankusa.Queue.Writer), the pending set and breaker (Ankusa.Intake.Pending, Ankusa.Intake.Gate), the live reconciler that carries a half-persisted hook to completion (Ankusa.Intake.Reconciler), and boot recovery (Ankusa.Intake.Recovery).

A record is {seq, bin, meta}: the hook's hooks value, written once under the writer's pre-assigned seq. "Confirmed" is the adapter's promise that the log's replicas hold it (Kafka: acks=all with min.insync.replicas); the adapter makes one attempt per publish_batch/3 call and the reconciler owns every retry. Nothing here assumes exactly-once publishing: every reader applies a record through the seq-and-id rule of Ankusa.Intake.Recovery, so a duplicate or reordered record changes nothing.

An adapter's failure is one of the four classes of ground rule 2 (:not_found | :unavailable | :throttled | :permanent), with an optional retry_after in milliseconds: {:throttled, 500}.

Summary

Types

class()

failure()

log_record()

meta()

replayed()

Callbacks

assign_partition(instance, node_id)

The partition this node writes, chosen once at its first dual boot.

child_spec(keyword)

The adapter's own processes (its client), started before the store.

describe(instance)

The settings the startup sentence quotes, e.g. acks=all, min.insync.replicas=2, replicas=3.

earliest(instance)

The earliest position the log still holds, nil for an empty log.

publish_batch(instance, list, opts)

Blocks until the batch is confirmed or opts[:timeout] runs out. position is the log position of the batch's last record. opts also carries :node, :topic and :partition (the node's stamped partition, or nil).

ready?(instance)

A cheap local answer (client up, leader known); never a produce.

replay(instance, from, function, opts)

Calls fun for every record of this node (opts[:node]) from position from, in log order. The adapter skips records of other nodes and of a kind or version it does not know before calling fun, and reports each skipped unknown one with skipped/1. Returns the position of the last record passed.

verify(instance)

Boot check of the log's topology, retention and record size.

Functions

assign_partition(instance)

Asks the adapter for this node's partition; :none when it has no such notion.

describe(instance)

The settings the adapter quotes in the startup sentence.

durability(config)

The startup sentence of wal: :dual (ADR 0002, section 1). describe/1 of the adapter supplies the verified settings; with intake.verify_topic: false the sentence says they are unverified instead.

earliest(instance)

The earliest position the log holds, nil when empty.

enabled?(config)

Whether the config runs the dual commit.

info(instance)

The intake object of GET /v1/wal.

known?(kind, version)

Whether an adapter should pass a record of this kind and version to the replay function.

module(config)

The adapter module intake.type names.

publish_batch(instance, records)

One publish attempt with the configured deadline, measured. Never raises: {:ok, position} or {:error, class, retry_after_ms | nil}.

ready?(instance)

The adapter's cheap readiness answer; false if it cannot answer.

replay(instance, from, fun)

Replays this node's records from from (see replay/4).

skipped(instance)

A record of a kind or version this release does not know (adapters call it per skip).

state(instance)

:recovering while boot replay runs, :down while the breaker is open and the adapter still says it is not ready (the readiness probe drains the node), :up otherwise: with the breaker open but the adapter ready again, the node is ready for the one batch that probes it.

topic(config)

The log's name: intake.topic, else ankusa.intake.<fleet>.

validate_config!(config)

Boot check: the adapter's package is on the code path.

verify(instance)

The adapter's boot verification.

Types

class()

@type class() :: :not_found | :unavailable | :throttled | :permanent

failure()

@type failure() :: class() | {class(), pos_integer()}

log_record()

@type log_record() :: {seq :: non_neg_integer(), bin :: binary(), meta()}

meta()

@type meta() :: %{
  id: String.t(),
  tenant: String.t(),
  source: String.t(),
  incarnation: String.t()
}

replayed()

@type replayed() :: %{
  position: non_neg_integer(),
  node: String.t(),
  seq: non_neg_integer(),
  id: String.t(),
  kind: String.t(),
  version: String.t(),
  bin: binary()
}

Callbacks

assign_partition(instance, node_id)

(optional)

@callback assign_partition(instance :: atom(), node_id :: String.t()) ::
  {:ok, non_neg_integer()} | :none

The partition this node writes, chosen once at its first dual boot.

child_spec(keyword)

@callback child_spec(keyword()) :: Supervisor.child_spec()

The adapter's own processes (its client), started before the store.

describe(instance)

@callback describe(instance :: atom()) :: String.t()

The settings the startup sentence quotes, e.g. acks=all, min.insync.replicas=2, replicas=3.

earliest(instance)

@callback earliest(instance :: atom()) ::
  {:ok, non_neg_integer() | nil} | {:error, failure()}

The earliest position the log still holds, nil for an empty log.

publish_batch(instance, list, opts)

@callback publish_batch(instance :: atom(), [log_record()], opts :: keyword()) ::
  {:ok, %{position: non_neg_integer()}} | {:error, failure()}

ready?(instance)

@callback ready?(instance :: atom()) :: boolean()

A cheap local answer (client up, leader known); never a produce.

replay(instance, from, function, opts)

@callback replay(
  instance :: atom(),
  from :: non_neg_integer() | :earliest,
  (replayed() -> :ok | {:error, term()}),
  opts :: keyword()
) :: {:ok, non_neg_integer() | nil} | {:error, failure()}

verify(instance)

@callback verify(instance :: atom()) :: :ok | {:error, term()}

Boot check of the log's topology, retention and record size.

Functions

assign_partition(instance)

@spec assign_partition(atom()) :: {:ok, non_neg_integer()} | :none

Asks the adapter for this node's partition; :none when it has no such notion.

describe(instance)

@spec describe(atom()) :: String.t()

The settings the adapter quotes in the startup sentence.

durability(config)

@spec durability(Ankusa.Config.t()) :: String.t()

earliest(instance)

@spec earliest(atom()) :: {:ok, non_neg_integer() | nil} | {:error, class()}

The earliest position the log holds, nil when empty.

enabled?(config)

@spec enabled?(Ankusa.Config.t()) :: boolean()

Whether the config runs the dual commit.

info(instance)

@spec info(atom()) :: map()

The intake object of GET /v1/wal.

known?(kind, version)

@spec known?(String.t(), String.t()) :: boolean()

Whether an adapter should pass a record of this kind and version to the replay function.

module(config)

@spec module(Ankusa.Config.t()) :: module()

The adapter module intake.type names.

publish_batch(instance, records)

@spec publish_batch(atom(), [log_record()]) ::
  {:ok, non_neg_integer()} | {:error, class(), pos_integer() | nil}

One publish attempt with the configured deadline, measured. Never raises: {:ok, position} or {:error, class, retry_after_ms | nil}.

ready?(instance)

@spec ready?(atom()) :: boolean()

The adapter's cheap readiness answer; false if it cannot answer.

replay(instance, from, fun)

@spec replay(atom(), non_neg_integer() | :earliest, (replayed() ->
                                                 :ok | {:error, term()})) ::
  {:ok, non_neg_integer() | nil} | {:error, class()}

Replays this node's records from from (see replay/4).

skipped(instance)

@spec skipped(atom()) :: :ok

A record of a kind or version this release does not know (adapters call it per skip).

state(instance)

@spec state(atom()) :: :up | :down | :recovering

topic(config)

@spec topic(Ankusa.Config.t()) :: String.t()

The log's name: intake.topic, else ankusa.intake.<fleet>.

validate_config!(config)

@spec validate_config!(Ankusa.Config.t()) :: :ok

Boot check: the adapter's package is on the code path.

verify(instance)

@spec verify(atom()) :: :ok | {:error, term()}

The adapter's boot verification.