Cyclium.WorkClaims behaviour (Cyclium v0.1.5)

Copy Markdown

Behaviour for lease-based distributed work claiming.

When configured, ensures at-most-once execution of episodes across clustered nodes. Each episode's dedupe_key is claimed before execution — only the node holding an active lease runs the work.

Configuration

# Use the built-in Ecto-based implementation:
config :cyclium, work_claims: Cyclium.WorkClaims.EctoClaims

# Or a SQL Server-optimized adapter in consuming apps:
config :cyclium, work_claims: MyApp.WorkClaims.SqlServer

# Or omit entirely — no claiming, fully backwards compatible

Gate functions

All integration points use gate_* functions which return {:ok, :passthrough} when no implementation is configured. This makes work claims opt-in with zero impact on existing deployments.

Summary

Types

acquire_result()
owner_result()

Callbacks

acquire(dedupe_key, owner_node, opts)
complete(dedupe_key, owner_node)
fail(dedupe_key, owner_node, error_detail)
reclaim_expired(limit)
renew(dedupe_key, owner_node, lease_seconds)

Functions

gate_acquire(dedupe_key, owner_node, opts \\ [])

Attempt to acquire a lease on dedupe_key. Returns {:ok, :passthrough} if work claims are not configured.

gate_complete(dedupe_key, owner_node)

Mark work as completed. No-op if unconfigured.

gate_fail(dedupe_key, owner_node, error_detail \\ %{})

Mark work as failed. No-op if unconfigured.

gate_reclaim_expired(limit \\ 100)

Reclaim up to limit expired claims. Returns empty list if unconfigured.

gate_renew(dedupe_key, owner_node, lease_seconds)

Renew the lease for dedupe_key. No-op if unconfigured.

Types

acquire_result()

@type acquire_result() :: {:ok, Cyclium.Schemas.WorkClaim.t()} | {:error, :busy}

owner_result()

@type owner_result() :: :ok | {:error, :not_owner}

Callbacks

acquire(dedupe_key, owner_node, opts)

@callback acquire(dedupe_key :: String.t(), owner_node :: String.t(), opts :: keyword()) ::
  acquire_result()

complete(dedupe_key, owner_node)

@callback complete(dedupe_key :: String.t(), owner_node :: String.t()) :: owner_result()

fail(dedupe_key, owner_node, error_detail)

@callback fail(dedupe_key :: String.t(), owner_node :: String.t(), error_detail :: map()) ::
  owner_result()

reclaim_expired(limit)

@callback reclaim_expired(limit :: pos_integer()) ::
  {:ok, [Cyclium.Schemas.WorkClaim.t()]}

renew(dedupe_key, owner_node, lease_seconds)

@callback renew(
  dedupe_key :: String.t(),
  owner_node :: String.t(),
  lease_seconds :: pos_integer()
) :: owner_result()

Functions

gate_acquire(dedupe_key, owner_node, opts \\ [])

Attempt to acquire a lease on dedupe_key. Returns {:ok, :passthrough} if work claims are not configured.

gate_complete(dedupe_key, owner_node)

Mark work as completed. No-op if unconfigured.

gate_fail(dedupe_key, owner_node, error_detail \\ %{})

Mark work as failed. No-op if unconfigured.

gate_reclaim_expired(limit \\ 100)

Reclaim up to limit expired claims. Returns empty list if unconfigured.

gate_renew(dedupe_key, owner_node, lease_seconds)

Renew the lease for dedupe_key. No-op if unconfigured.