Snodo.Server.Executor (snodo v0.2.0)

Copy Markdown View Source

Optional, transport-neutral execution policy around the synchronous server core.

The executor knows nothing about JSON-RPC, protocol dialects, sessions, or transports. Callers submit a function that receives a cooperative Snodo.Cancellation token. The executor bounds concurrent work, optionally queues admitted work, enforces execution deadlines, and reports terminal outcomes back to the submitting process while both it and the executor remain alive. Work is cancelled without delivery when its reply owner terminates.

Direct callers can continue to invoke Snodo.Server.dispatch/3 without starting this or any other process.

Summary

Types

event()
execution_key()
execution_ref()
execution_timeout()
outcome()
server()

Functions

cancel(server, key, reason \\ nil)

Cancels queued or running work identified by its caller-supplied key.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

start_link(opts \\ [])

Starts an executor linked to the calling process.

stats(server)

Returns the current bounded-execution counters and limits.

submit(server, key, work, opts \\ [])

Submits one unit of work.

Types

event()

@type event() :: {:mcp_execution, pid(), execution_ref(), execution_key(), outcome()}

execution_key()

@type execution_key() :: term()

execution_ref()

@type execution_ref() :: reference()

execution_timeout()

@type execution_timeout() :: non_neg_integer() | :infinity

outcome()

@type outcome() ::
  {:completed, term()}
  | {:cancelled, term()}
  | {:timed_out, non_neg_integer()}
  | {:failed, term()}

server()

@type server() :: GenServer.server()

Functions

cancel(server, key, reason \\ nil)

@spec cancel(server(), execution_key(), term()) :: :ok | {:error, :not_found}

Cancels queued or running work identified by its caller-supplied key.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

start_link(opts \\ [])

@spec start_link(keyword()) :: GenServer.on_start()

Starts an executor linked to the calling process.

Options:

  • :name - a name to register the executor under.
  • :max_concurrency - the most work that runs at once. A positive integer, default 32.
  • :max_queue - the most admitted work that waits for a free slot. A non-negative integer, default 256. When both limits are reached, submit/4 returns {:error, :overloaded}.
  • :default_timeout - the deadline in milliseconds for work submitted without :timeout. A non-negative integer or :infinity, default 30,000.

An invalid option raises ArgumentError in the executor process, and the start fails.

stats(server)

@spec stats(server()) :: %{
  running: non_neg_integer(),
  queued: non_neg_integer(),
  max_concurrency: pos_integer(),
  max_queue: non_neg_integer()
}

Returns the current bounded-execution counters and limits.

submit(server, key, work, opts \\ [])

@spec submit(server(), execution_key(), (Snodo.Cancellation.t() -> term()), keyword()) ::
  {:ok, execution_ref()} | {:error, :duplicate_key | :overloaded}

Submits one unit of work.

key must be unique across running and queued executions. The supplied function runs in a supervised task and receives a cancellation token. On completion, the executor sends an event/0 message to :reply_to, which defaults to the submitting process. Reply-owner termination or executor shutdown ends that delivery guarantee and tears down the work instead.