Raxol.Agent.ThreadLog behaviour (Raxol Agent v2.6.0)

Copy Markdown View Source

Behaviour for Raxol.Agent.ThreadEvent persistence.

Implementations decide the storage medium. Two adapters ship with raxol_agent today:

A Raxol.Agent.ThreadLog.Dets adapter for single-node durable logs is a deliberate follow-up; the Ets ↔ Postgrex pair covers the two most common shapes (single-node ephemeral, multi-node durable).

Sequence assignment

The adapter assigns sequence atomically per thread_id on append/4. Callers do not pick the number. Within a thread_id, sequences start at 0 and increase strictly monotonically. The {thread_id, sequence} pair is the unique key.

Concurrency contract

Adapters assume single-writer per thread_id. The expected pattern is one agent process owning one thread_id. Multi-writer scenarios (two agents writing to the same thread_id concurrently) may produce primary-key conflicts that the Postgrex adapter retries internally up to three times; under heavier contention, callers should serialize through a single owning process.

The Ets adapter uses :ets.update_counter for sequence allocation, which is atomic for arbitrary concurrent writers. Sequence ordering is preserved.

Querying

list/3 returns events in sequence order (ascending by default, override with :order). list_by_kind/4 filters; the kind set is open (see Raxol.Agent.ThreadEvent). latest/2 is sugar for the highest-sequence event. truncate/3 removes events with sequence < before, retaining the tail.

Configuration

When wired into an agent's thread_log/0 callback:

def thread_log do
  {Raxol.Agent.ThreadLog.Postgrex,
   %{conn: MyApp.Postgrex, table: "agent_threads"}}
end

Bare modules (no tuple) inherit %{}.

Summary

Types

Per-call configuration map passed to every callback.

Options accepted by list/3 and list_by_kind/4

Monotonic sequence within a thread; starts at 0.

Thread identifier; arbitrary binary.

Callbacks

Append an event. The adapter assigns the next sequence for the given thread_id atomically and returns the materialized event.

Return the highest-sequence event for thread_id.

Return all events for thread_id matching opts.

Return events of kind for thread_id, sequence-ordered per opts.

Remove events with sequence < before for thread_id. The tail (events at or after before) is retained. Useful for retention policies.

Functions

Dispatch append to the adapter. Returns {:ok, event} or {:error, reason}. Returns {:ok, :no_log} when the adapter tuple is nil so callers can treat "no thread log" as "successfully no-op" without branching.

Dispatch latest. Returns {:error, :not_found} for nil adapter.

Dispatch list to the adapter. Returns {:ok, []} for nil adapter.

Dispatch list_by_kind. Returns {:ok, []} for nil adapter.

Normalize the thread-log opt into {module, config} form.

Dispatch truncate. No-op for nil adapter.

Types

config()

@type config() :: map()

Per-call configuration map passed to every callback.

list_opts()

@type list_opts() :: keyword()

Options accepted by list/3 and list_by_kind/4:

  • :from -- starting sequence (inclusive). Default: 0.
  • :to -- ending sequence (inclusive). Default: no limit.
  • :limit -- max events returned. Default: 1000.
  • :order -- :asc (default) or :desc.

sequence()

@type sequence() :: non_neg_integer()

Monotonic sequence within a thread; starts at 0.

thread_id()

@type thread_id() :: binary()

Thread identifier; arbitrary binary.

Callbacks

append(config, thread_id, kind, payload, opts)

@callback append(
  config(),
  thread_id(),
  Raxol.Agent.ThreadEvent.kind(),
  payload :: term(),
  opts :: keyword()
) :: {:ok, Raxol.Agent.ThreadEvent.t()} | {:error, term()}

Append an event. The adapter assigns the next sequence for the given thread_id atomically and returns the materialized event.

opts:

  • :metadata -- arbitrary map; preserved verbatim.
  • :recorded_at -- override the timestamp (default DateTime.utc_now/0).

latest(config, thread_id)

@callback latest(config(), thread_id()) ::
  {:ok, Raxol.Agent.ThreadEvent.t()} | {:error, :not_found}

Return the highest-sequence event for thread_id.

list(config, thread_id, list_opts)

@callback list(config(), thread_id(), list_opts()) :: {:ok, [Raxol.Agent.ThreadEvent.t()]}

Return all events for thread_id matching opts.

list_by_kind(config, thread_id, kind, list_opts)

@callback list_by_kind(
  config(),
  thread_id(),
  Raxol.Agent.ThreadEvent.kind(),
  list_opts()
) :: {:ok, [Raxol.Agent.ThreadEvent.t()]}

Return events of kind for thread_id, sequence-ordered per opts.

truncate(config, thread_id, before)

@callback truncate(config(), thread_id(), before :: sequence()) :: :ok | {:error, term()}

Remove events with sequence < before for thread_id. The tail (events at or after before) is retained. Useful for retention policies.

Functions

append(adapter, thread_id, kind, payload, opts \\ [])

@spec append(
  {module(), config()} | nil,
  thread_id(),
  Raxol.Agent.ThreadEvent.kind(),
  term(),
  keyword()
) :: {:ok, Raxol.Agent.ThreadEvent.t() | :no_log} | {:error, term()}

Dispatch append to the adapter. Returns {:ok, event} or {:error, reason}. Returns {:ok, :no_log} when the adapter tuple is nil so callers can treat "no thread log" as "successfully no-op" without branching.

latest(arg1, thread_id)

@spec latest({module(), config()} | nil, thread_id()) ::
  {:ok, Raxol.Agent.ThreadEvent.t()} | {:error, :not_found}

Dispatch latest. Returns {:error, :not_found} for nil adapter.

list(adapter, thread_id, opts \\ [])

@spec list({module(), config()} | nil, thread_id(), list_opts()) ::
  {:ok, [Raxol.Agent.ThreadEvent.t()]}

Dispatch list to the adapter. Returns {:ok, []} for nil adapter.

list_by_kind(adapter, thread_id, kind, opts \\ [])

@spec list_by_kind(
  {module(), config()} | nil,
  thread_id(),
  Raxol.Agent.ThreadEvent.kind(),
  list_opts()
) :: {:ok, [Raxol.Agent.ThreadEvent.t()]}

Dispatch list_by_kind. Returns {:ok, []} for nil adapter.

normalize(module)

@spec normalize(module() | {module(), config()} | nil) :: {module(), config()} | nil

Normalize the thread-log opt into {module, config} form.

Accepts a bare module, a {module, config} tuple, or nil. Returns nil for nil input.

truncate(arg1, thread_id, before)

@spec truncate({module(), config()} | nil, thread_id(), sequence()) ::
  :ok | {:error, term()}

Dispatch truncate. No-op for nil adapter.