Anubis.Server.Session.Scheduler (anubis_mcp v1.14.0)

Copy Markdown

Request scheduling engine for a session.

Owns the single in-flight request slot, the FIFO request queue, callbacks deferred while a request is in flight, and cancellation of in-flight or queued requests. At most one request executes at a time per session: while one runs, competing requests queue and competing casts/infos are deferred, then drained in order once the in-flight request completes.

State fields (in_flight, request_queue, deferred_callbacks, pending_requests) live in the Session's state map; this module transforms them and performs the side effects (task spawn, monitors, caller replies, logging, telemetry) at the edges. Anubis.Server.Session delegates to this module and supplies callbacks to prepare frames and to apply drained deferred items back into its own dispatch paths.

Summary

Types

apply_deferred_fn()
callbacks()
deferred_callback()
frame_fn()
in_flight()
queued_request()
state()

Functions

cancel(notification, state, callbacks)

Handles a notifications/cancelled notification, cancelling either the in-flight request or any queued entries matching the request ID.

defer(state, item)

Defers a cast or info message to be applied once the in-flight request completes.

enqueue_or_dispatch(request, ctx, from, state, callbacks)

Dispatches the request immediately when no request is in flight, otherwise appends it to the request queue.

handle_completion(ref, callback_result, state, callbacks)

Handles completion of the in-flight request task: decodes the handler result, replies to the caller, drains deferred callbacks, and dispatches the next queued request.

handle_down(reason, state, callbacks)

Handles a crash of the in-flight request task, replying to the caller with an internal error before finalizing.

reply_pending(map, error)

Replies to the in-flight and queued callers with the given error during session termination, terminating the in-flight worker task.

Types

apply_deferred_fn()

@type apply_deferred_fn() :: (deferred_callback(), state() ->
                          {:noreply, state()}
                          | {:noreply, state(), term()}
                          | {:stop, term(), state()})

callbacks()

@type callbacks() :: %{frame: frame_fn(), apply_deferred: apply_deferred_fn()}

deferred_callback()

@type deferred_callback() :: {:cast | :info, term()}

frame_fn()

@type frame_fn() :: (state(), map() | nil -> Anubis.Server.Frame.t())

in_flight()

@type in_flight() :: %{
  ref: reference(),
  pid: pid(),
  request_id: String.t(),
  from: GenServer.from(),
  started_at: integer(),
  method: String.t()
}

queued_request()

@type queued_request() :: {map(), map(), GenServer.from()}

state()

@type state() :: map()

Functions

cancel(notification, state, callbacks)

@spec cancel(map(), state(), callbacks()) ::
  {:noreply, state()} | {:stop, term(), state()}

Handles a notifications/cancelled notification, cancelling either the in-flight request or any queued entries matching the request ID.

defer(state, item)

@spec defer(state(), deferred_callback()) :: state()

Defers a cast or info message to be applied once the in-flight request completes.

enqueue_or_dispatch(request, ctx, from, state, callbacks)

@spec enqueue_or_dispatch(map(), map(), GenServer.from(), state(), callbacks()) ::
  {:noreply, state()}

Dispatches the request immediately when no request is in flight, otherwise appends it to the request queue.

handle_completion(ref, callback_result, state, callbacks)

@spec handle_completion(reference(), term(), state(), callbacks()) ::
  {:noreply, state()} | {:stop, term(), state()}

Handles completion of the in-flight request task: decodes the handler result, replies to the caller, drains deferred callbacks, and dispatches the next queued request.

handle_down(reason, state, callbacks)

@spec handle_down(term(), state(), callbacks()) ::
  {:noreply, state()} | {:stop, term(), state()}

Handles a crash of the in-flight request task, replying to the caller with an internal error before finalizing.

reply_pending(map, error)

@spec reply_pending(state(), Anubis.MCP.Error.t()) :: :ok

Replies to the in-flight and queued callers with the given error during session termination, terminating the in-flight worker task.