Snodo.Extensions.Tasks.Store behaviour (snodo_tasks v0.2.0)

Application-owned persistence contract for the Tasks extension.

A request context crosses this boundary only through authorize/3. The resulting access value is opaque and deliberately scoped to one action. All mutations after that point are explicit, JSON-safe events guarded by an expected snapshot revision.

Creation atomically persists the initial Task and its serializable work descriptor. Runners then acquire renewable, opaque claims. A lease can authorize only worker lifecycle events for its exact task and claim generation; it is not a substitute for fresh request authorization on tasks/get, tasks/update, or tasks/cancel.

claim_next/3 is the recovery seam. Expired or adapter-invalidated claims may be reacquired at a higher generation, fencing the previous worker. Both exact and recovery claims must also enforce the snapshot's persisted retry_at against store-authoritative time. Reaping removes the entire aggregate using the Task's creation-based TTL.

Store references use {module, state} so implementations may be processes, database repositories, external job systems, or immutable test doubles.

Summary

Types

access()

action()

authority()

claim_next_result()

claim_result()

lookup_result()

ref()

transition_result()

worker_lease()

Callbacks

authorize(store, t, action)

claim(store, task_id, owner_id, lease_ms)

claim_next(store, owner_id, lease_ms)

create(store, t, t, access)

get(store, task_id, access)

reap(store)

release(store, worker_lease)

renew(store, worker_lease, lease_ms)

transition(store, task_id, expected_revision, t, authority)

worker_snapshot(store, task_id, worker_lease)

Functions

authorize(arg, context, requested)

Acquires opaque, action-bound request access without retaining the context.

claim(arg, task_id, owner_id, lease_ms)

Atomically claims one exact nonterminal task whose retry delay is due.

claim_next(arg, owner_id, lease_ms)

Atomically claims the next available, retry-due recoverable task.

create(arg, task, work, access)

get(arg, task_id, access)

reap(arg)

Deletes complete task aggregates whose creation-based TTL has elapsed.

release(arg, lease)

Releases a matching claim so another runner may recover it immediately.

renew(arg, lease, lease_ms)

Extends a live worker claim without changing task state or revision.

transition(arg, task_id, expected_revision, event, authority)

validate_ref!(ref)

worker_snapshot(arg, task_id, lease)

Reads the current aggregate through an exact, live worker lease.

Types

access()

@type access() :: term()

action()

@type action() ::
  {:create, String.t()}
  | {:get, String.t()}
  | {:update, String.t()}
  | {:cancel, String.t()}

authority()

@type authority() :: {:request, access()} | {:worker, worker_lease()}

claim_next_result()

@type claim_next_result() ::
  {:ok, Snodo.Extensions.Tasks.Snapshot.t(), worker_lease()}
  | :empty
  | {:error, term()}

claim_result()

@type claim_result() ::
  {:ok, Snodo.Extensions.Tasks.Snapshot.t(), worker_lease()}
  | {:deferred, pos_integer()}
  | :unavailable
  | :not_found
  | {:error, term()}

lookup_result()

@type lookup_result() ::
  {:ok, Snodo.Extensions.Tasks.Snapshot.t()} | :not_found | {:error, term()}

ref()

@type ref() :: {module(), term()}

transition_result()

@type transition_result() ::
  {:ok, Snodo.Extensions.Tasks.Transition.t()}
  | {:conflict, Snodo.Extensions.Tasks.Snapshot.t()}
  | :not_found
  | {:error, term()}

worker_lease()

@type worker_lease() :: term()

Callbacks

authorize(store, t, action)

@callback authorize(store :: term(), Snodo.Context.t(), action()) ::
  {:ok, access()} | {:error, term()}

claim(store, task_id, owner_id, lease_ms)

@callback claim(
  store :: term(),
  task_id :: String.t(),
  owner_id :: String.t(),
  lease_ms :: pos_integer()
) ::
  claim_result()

claim_next(store, owner_id, lease_ms)

@callback claim_next(store :: term(), owner_id :: String.t(), lease_ms :: pos_integer()) ::
  claim_next_result()

create(store, t, t, access)

@callback create(
  store :: term(),
  Snodo.Extensions.Tasks.Task.t(),
  Snodo.Extensions.Tasks.Work.t(),
  access()
) :: {:ok, Snodo.Extensions.Tasks.Snapshot.t()} | {:error, term()}

get(store, task_id, access)

@callback get(store :: term(), task_id :: String.t(), access()) :: lookup_result()

reap(store)

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

release(store, worker_lease)

@callback release(store :: term(), worker_lease()) :: :ok | {:error, term()}

renew(store, worker_lease, lease_ms)

@callback renew(store :: term(), worker_lease(), lease_ms :: pos_integer()) ::
  {:ok, worker_lease()} | {:error, term()}

transition(store, task_id, expected_revision, t, authority)

@callback transition(
  store :: term(),
  task_id :: String.t(),
  expected_revision :: non_neg_integer(),
  Snodo.Extensions.Tasks.Event.t(),
  authority()
) :: transition_result()

worker_snapshot(store, task_id, worker_lease)

@callback worker_snapshot(store :: term(), task_id :: String.t(), worker_lease()) ::
  lookup_result()

Functions

authorize(arg, context, requested)

@spec authorize(ref(), Snodo.Context.t(), action()) ::
  {:ok, access()} | {:error, term()}

Acquires opaque, action-bound request access without retaining the context.

claim(arg, task_id, owner_id, lease_ms)

@spec claim(ref(), String.t(), String.t(), pos_integer()) :: claim_result()

Atomically claims one exact nonterminal task whose retry delay is due.

claim_next(arg, owner_id, lease_ms)

@spec claim_next(ref(), String.t(), pos_integer()) :: claim_next_result()

Atomically claims the next available, retry-due recoverable task.

create(arg, task, work, access)

@spec create(
  ref(),
  Snodo.Extensions.Tasks.Task.t(),
  Snodo.Extensions.Tasks.Work.t(),
  access()
) ::
  {:ok, Snodo.Extensions.Tasks.Snapshot.t()} | {:error, term()}

get(arg, task_id, access)

@spec get(ref(), String.t(), access()) :: lookup_result()

reap(arg)

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

Deletes complete task aggregates whose creation-based TTL has elapsed.

release(arg, lease)

@spec release(ref(), worker_lease()) :: :ok | {:error, term()}

Releases a matching claim so another runner may recover it immediately.

renew(arg, lease, lease_ms)

@spec renew(ref(), worker_lease(), pos_integer()) ::
  {:ok, worker_lease()} | {:error, term()}

Extends a live worker claim without changing task state or revision.

transition(arg, task_id, expected_revision, event, authority)

@spec transition(
  ref(),
  String.t(),
  non_neg_integer(),
  Snodo.Extensions.Tasks.Event.t(),
  authority()
) :: transition_result()

validate_ref!(ref)

@spec validate_ref!(term()) :: ref()

worker_snapshot(arg, task_id, lease)

@spec worker_snapshot(ref(), String.t(), worker_lease()) :: lookup_result()

Reads the current aggregate through an exact, live worker lease.