Snodo.Extensions.Tasks.Runner (snodo_tasks v0.3.2)

Copy Markdown View Source

Independently supervised execution, recovery, and mid-task input coordination.

Jobs are protected by renewable store claims. A claim generation fences an expired worker from committing after another runner has recovered the same descriptor. Recovery is intentionally at-least-once: applications should use Snodo.Extensions.Tasks.Work.idempotency_key for external side effects.

Executors may explicitly return {:retry, error, status_message}. The descriptor's finite retry policy is applied by the durable transition, the claim is released during backoff, and store-authoritative claim eligibility remains the source of truth after local timers or process restarts.

Request contexts and transport handles never enter runner state. An optional application-owned Snodo.Extensions.Tasks.WorkExecutor resolves persisted work on both the initial claim and every recovery claim.

Configure :instrumentation with an Snodo.Instrumentation sink to observe job starts/stops and every store transition attempted by the runner. Events include task identity and lifecycle classifications, never work input, authorization values, results, errors, or input responses.

Limits

  • :max_jobs - the most workers this runner runs at once, default 32. create_task/5 refuses a new task at the limit with {:error, {:capacity_exceeded, :max_jobs}} before storing it, and start_task/4 returns the same error and leaves the stored task unclaimed. A due retry waits for a free slot, and recovery claims no more work than the free slots. The count is this runner's own: its workers are processes in this BEAM, and none survive a restart.
  • :max_runtime_ms - how long one worker may run, default 3,600,000 (one hour), or :infinity. The time counts from the claim that started the worker and includes time spent waiting for input. A worker that runs longer is stopped and its task fails.
  • :reap_interval_ms - how often the runner calls Store.reap/1, default 60,000. nil turns periodic reaping off.

The store limits on stored and active tasks are options of each store.

Summary

Functions

Parks the calling worker or replays persisted input after recovery.

Commits cancellation, then cooperatively signals and stops local work.

Returns a specification to start this module under a supervisor.

Stores a new task with Store.create/4, then claims and starts its work.

Runs creation-based TTL cleanup and stops any matching local jobs.

Claims and starts work after its descriptor-bearing snapshot is visible.

Accepts persisted outstanding responses and wakes matching local workers.

Types

outcome()

@type outcome() ::
  {:completed, map()}
  | {:failed, map(), String.t() | nil}
  | {:retry, map(), String.t() | nil}

server()

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

Functions

await_input(runner, task_id, key, request, execution_token)

@spec await_input(server(), String.t(), String.t(), map(), Snodo.Cancellation.t()) ::
  {:ok, map()} | {:error, term()}

Parks the calling worker or replays persisted input after recovery.

cancel(runner, task_id, access)

@spec cancel(server(), String.t(), Snodo.Extensions.Tasks.Store.access()) ::
  :ok | :not_found | {:error, term()}

Commits cancellation, then cooperatively signals and stops local work.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

create_task(runner, task, work, access, fallback_work)

Stores a new task with Store.create/4, then claims and starts its work.

At :max_jobs nothing is stored and the call returns {:error, {:capacity_exceeded, :max_jobs}}. A store refusal, including {:capacity_exceeded, limit} from a store limit, is returned unchanged.

reap(runner)

@spec reap(server()) :: {:ok, [String.t()]} | {:error, term()}

Runs creation-based TTL cleanup and stops any matching local jobs.

start_link(opts)

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

start_task(runner, snapshot, fallback_work, opts \\ [])

@spec start_task(
  server(),
  Snodo.Extensions.Tasks.Snapshot.t(),
  (Snodo.Cancellation.t() -> outcome()),
  keyword()
) :: :ok | {:error, term()}

Claims and starts work after its descriptor-bearing snapshot is visible.

At :max_jobs the task is not claimed and the call returns {:error, {:capacity_exceeded, :max_jobs}}.

update(runner, task_id, responses, access)

@spec update(server(), String.t(), map(), Snodo.Extensions.Tasks.Store.access()) ::
  :ok | :not_found | {:error, term()}

Accepts persisted outstanding responses and wakes matching local workers.