Arbiter

A PostgreSQL job queue for Haskell applications.

• Transactional job processing: jobs and database operations commit together
• At-least-once delivery with visibility timeouts and heartbeats
• Per-group ordering (partitioned FIFO)
• Concurrent worker pools with LISTEN/NOTIFY wakeups and polling fallback
• Dead-letter queues
• Opt-in archiving of completed jobs with per-job retention
• Job trees with fan-out/fan-in result collection
• Cron/periodic job scheduling
• Job deduplication via unique keys
• Cross-queue per-job rate limiting with operator-tunable token-bucket policies
• Cross-queue per-job concurrency limits: at most N jobs with the same key can be in flight
• Integrated OpenTelemetry traces, with metrics and logs from arbiter-otel
• Observability callbacks, structured logging
• REST API with SSE and an embedded admin UI
• File-based liveness probes for Kubernetes and systemd
• More than 1,000 integration tests

[!NOTE]

The API is subject to breaking changes. A Hackage release following PVP is tentative.

Installation

Install directly from GitHub:

Cabal: Add this source repository to cabal.project:

source-repository-package
  type: git
  location: https://github.com/velveteer/arbiter.git
  tag: <commit-sha>
  subdir:
    arbiter-core
    arbiter-worker
    arbiter-simple
    arbiter-migrations

Stack: Add this source repository to stack.yaml:

extra-deps:
  - git: https://github.com/velveteer/arbiter.git
    commit: <commit-sha>
    subdirs:
      - arbiter-core
      - arbiter-worker
      - arbiter-simple
      - arbiter-migrations

Replace arbiter-simple with the package for the selected backend: arbiter-orville or arbiter-hasql.

Quick Start

Payload and Result Types

Define payload types with ToJSON and FromJSON instances. A queue whose handlers produce a result needs the same instances on the result type.

data EmailPayload
  = SendWelcome Text Text
  | SendReceipt Text Int
  deriving stock (Eq, Show, Generic)
  deriving anyclass (ToJSON, FromJSON)

data ImagePayload
  = ResizeImage Text Int Int
  | GenerateThumbnail Text
  deriving stock (Eq, Show, Generic)
  deriving anyclass (ToJSON, FromJSON)

data Score = Score
  { sharpness :: Double
  , sizeBytes :: Int
  }
  deriving stock (Eq, Show, Generic)
  deriving anyclass (ToJSON, FromJSON)

Type-Level Registry

Map queue table names to payload types at the type level. Queue declares a queue whose handlers store no result. QueueWithResult adds the result type its handlers produce.

import Arbiter.Core.QueueRegistry (Queue, QueueSpec (..))

type AppRegistry =
  '[ Queue "email_queue" EmailPayload
   , QueueWithResult "image_queue" ImagePayload Score
   ]

The compiler checks the registry. Each payload type maps to one table. A duplicate table name or payload type is a type error.

Migrations

import Arbiter.Migrations qualified as Mig
import Data.Proxy (Proxy (..))
import System.Exit (die)

main :: IO ()
main = do
  result <- Mig.runMigrationsForRegistry (Proxy @AppRegistry) connStr "arbiter" Mig.defaultMigrationConfig
  case result of
    Mig.MigrationSuccess -> putStrLn "Migrations complete"
    Mig.MigrationError err -> die $ "Migration failed: " <> err

If the database user lacks CREATE privilege on the schema, create it manually first:

CREATE SCHEMA IF NOT EXISTS arbiter;
GRANT USAGE, CREATE ON SCHEMA arbiter TO your_app_user;

Migrations reconcile enableNotifications and enableEventStreaming. Run the migrations after a change to either option. The migration installs or removes the applicable triggers. Do not edit the migration history.

Multiple replicas can run migrations concurrently. migrationLockTimeout limits the wait for the migration lock. Arbiter returns MigrationError if the wait exceeds this limit. The default has no time limit. Use a direct database connection for migrations. A pooler in transaction mode cannot serialize migration sessions.

Inserting Jobs

import Arbiter.Core qualified as Arb
import Arbiter.Simple qualified as ArbS
import Data.Proxy (Proxy (..))

-- A producer needs no worker-pool config, the defaults are enough
env <- ArbS.createSimpleEnv (Proxy @AppRegistry) connStr "arbiter"

ArbS.runSimpleDb env $ do
  -- Ungrouped: processed concurrently by any available worker
  _ <- Arb.insertJob (Arb.defaultJob $ SendWelcome "alice@example.com" "Alice")

  -- Grouped: jobs with the same group key are processed one at a time
  _ <- Arb.insertJob (Arb.defaultGroupedJob "user-42" $ SendReceipt "alice@example.com" 1001)

insertJob returns Maybe (JobRead payload). It returns Nothing when a deduplication key causes Arbiter to skip the insert.

Configuring a Job

Start from defaultJob/defaultGroupedJob and apply setters:

job =
  Arb.defaultJob (SendWelcome "alice@example.com" "Alice")
    & Arb.setPriority 10
    & Arb.setMaxAttempts (Just 3)
    & Arb.setArchiveFor (Just Arb.dayRetention)

Processing Jobs

import Arbiter.Core qualified as Arb
import Arbiter.Simple qualified as ArbS
import Arbiter.Worker qualified as Worker
import Control.Monad (void)
import Control.Monad.IO.Class (liftIO)
import Data.Proxy (Proxy (..))
import Database.PostgreSQL.Simple qualified as PG

main :: IO ()
main = do
  -- One pool with five worker threads, using the arbiter-simple backend
  config <- Worker.transactionalWorkerConfig 5 processEmail
  let workers = [Worker.namedWorkerPool config]
  poolCfg <- Worker.poolConfigForWorkers workers
  env <- ArbS.createSimpleEnvWithConfig (Proxy @AppRegistry) connStr "arbiter" poolCfg
  ArbS.runSimpleDb env $ Worker.runWorkerPools workers

processEmail :: Arb.JobHandler (ArbS.SimpleDb AppRegistry IO) EmailPayload ()
processEmail conn job = do
  case Arb.payload job of
    SendWelcome recipient name -> do
      result <- liftIO $ sendEmail recipient ("Welcome, " <> name)
      case result of
        Left err -> Arb.throwRetryable err
        Right () -> pure ()

    SendReceipt recipient orderId -> do
      -- Transactional: this INSERT and the job ack commit together
      void $ liftIO $ PG.execute conn
        "INSERT INTO email_log (recipient, order_id) VALUES (?, ?)"
        (recipient, orderId)

transactionalWorkerConfig wraps each handler in a transaction and acks the job. If the handler returns, the job and the handler's writes commit together. If it throws, the transaction rolls back and the job is retried or moved to the DLQ.

manualWorkerConfig does not start a transaction. It supplies callbacks to ack, fail, cancel, or reprocess the job:

config <- Worker.manualWorkerConfig 5 processEmail

processEmail
    :: Arb.JobRead EmailPayload
    -> Worker.BatchCallbacks (ArbS.SimpleDb AppRegistry IO) EmailPayload ()
    -> ArbS.SimpleDb AppRegistry IO ()
processEmail job cbs = do
  liftIO $ deliverEmail (Arb.payload job)
  Worker.ack cbs job

Worker Configuration compares the two, and Batched Handlers covers taking several jobs per invocation.

Architecture

Arbiter does not use a broker or central coordinator. Each worker pool claims jobs directly from PostgreSQL. Add worker processes to increase capacity. There is no leader process.

The lifecycle under transactionalWorkerConfig:

1. Claim: The dispatcher claims visible jobs in per-group order, increments each attempt count, and hides each job for the visibility timeout. Admission is part of the same statement: a job whose rate-limit bucket is empty or whose concurrency pool is full is not claimed, and a claimed job has already spent its tokens and taken its slot. A heartbeat extends the timeout while the handler runs.
2. Run: The worker runs the handler inside a transaction. The handler's database work, its stored result, and the ack commit together.
3. Success: The job is acked and the transaction commits.
4. Failure: The transaction rolls back. A separate transaction retries the job with backoff or moves it to the dead-letter queue (DLQ).
5. Reclaim: If the visibility period ended and another worker claimed the job, the heartbeat or the ack throws and the worker abandons the job.

The claim operation applies the admission limits. The limits apply to worker pools in all processes and to clients that use the REST API. Claimants do not coordinate with each other.

Delivery is at least once. Arbiter can run a job again after a worker crash or an expired visibility timeout. Make non-transactional side effects idempotent.

With manualWorkerConfig and defaultBatchedWorkerConfig, step 2 does not start a transaction. The handler uses callbacks to complete, fail, cancel, or nack each job. The claim, heartbeat, and reclaim operations are unchanged.

Group Ordering

A group key permits one job or batch at a time in that group. Different groups can run concurrently.

• Same group key: Eligible jobs run in insertion order within each priority. A retrying job remains first until it succeeds or moves to the DLQ. A ready job can run before a delayed job. A group waits during a job's retry backoff or rate-limit delay.
• No group key: Any available worker can run the job concurrently.

Priority

Each job has an integer priority. Arbiter claims lower numbers first. The default is 0. Use a higher number for background work.

-- runs behind default-priority work
job = Arb.defaultJob payload & Arb.setPriority 10

For equal priorities, Arbiter uses insertion order.

Priority applies during a claim. It does not preempt work that is in flight. A new high-priority job waits for an available worker.

In a group, a retrying job remains first until it succeeds or moves to the DLQ. This rule has precedence over the priority of other jobs in that group.

A group is eligible when it has a ready job. Arbiter ranks the group by the lowest priority number in that group, including delayed jobs. A delayed high-priority job therefore increases the rank of its group.

Payload Kinds

A payload kind is an optional Text label stored with each job. Arbiter calls kindOf during insertion. Kind labels support job filters, queue statistics, metrics, traces, and the admin UI.

HasKind defines the label for one payload and the finite set of labels for its payload type:

The fallback instance for a payload type returns Nothing and an empty list. Declare a payload-specific instance to enable kind labels.

Constructor Labels

The generic implementation uses data-constructor names. Derive Generic and declare an empty instance:

data EmailPayload
  = SendWelcome UserId
  | SendReceipt OrderId
  deriving stock (Generic)

instance HasKind EmailPayload

For this instance, kindOf (SendReceipt 7) returns Just "SendReceipt" and kindsFor @EmailPayload returns ["SendWelcome", "SendReceipt"].

Custom Labels

Implement both members when constructor names are not suitable:

data EmailKind = Welcome | Receipt | PasswordReset
  deriving stock (Bounded, Enum, Show)

emailKindText :: EmailKind -> Text
emailKindText = T.toLower . T.pack . show

data EmailPayload = EmailPayload
  { emailKind :: EmailKind
  , emailTo :: Text
  }

instance HasKind EmailPayload where
  kindOf = Just . emailKindText . emailKind
  kindsFor = map emailKindText [minBound .. maxBound]

For this instance, kindOf (EmailPayload Receipt "a@b.c") returns Just "receipt". kindsFor @EmailPayload returns ["welcome", "receipt", "passwordreset"].

kindsFor must contain each non-Nothing value that kindOf can return. Arbiter excludes undeclared labels from kind metrics and the kindCounts statistics field.

Labels from a Nested Type

constructorKind and constructorKinds read the constructors of any Generic type. They do not require a HasKind instance on that type.

data Envelope = Envelope
  { envelopeTraceId :: Text
  , envelopePayload :: EmailPayload
  }

instance HasKind Envelope where
  kindOf = Just . constructorKind . envelopePayload
  kindsFor = constructorKinds @EmailPayload

This form supports wrapper payloads and external sum types without an orphan HasKind instance.

Label Use

The finite kindsFor set limits metric cardinality. Span attributes can include an undeclared label.

API details: Arbiter.Core.Job.Kind.

Deduplication

A dedup key decides what happens when a job's key is already queued:

-- IgnoreDuplicate: silently skip if key exists
job1 = Arb.defaultJob payload & Arb.setDedupKey (Just $ IgnoreDuplicate "order-123")

-- ReplaceDuplicate: replace the existing job and re-arm it for a fresh run
job2 = Arb.defaultJob payload & Arb.setDedupKey (Just $ ReplaceDuplicate "order-123")

Keys have queue scope. Jobs that have no key cannot cause a deduplication conflict.

ReplaceDuplicate copies each writable column from the new job: payload, priority, group key, attempt limit, admission keys, and retention. It clears the attempt count, last error, and active claim. The updated job is then ready for a new run.

Arbiter refuses replacement when the existing job is in flight, has a force-cancel flag, or has children in the queue or DLQ. For a refused replacement, insertJob returns Nothing and does not change the existing job. This return value states that the key existed. It does not state if replacement occurred.

See the Arbiter.Core.Job.Dedup haddocks for the key type.

Job Results

A handler can produce a result. With transactionalWorkerConfig, return the result. With a manual or batched configuration, pass it to ackWith or ackAllWith.

Arbiter stores a result only in these conditions:

• Job with a parent: Arbiter stores the result for the parent to read with Worker.childResults or Worker.mergedChildResults. It removes the result after the parent completes.
• Standalone root job: Arbiter stores the result in the job archive entry if archiving is enabled for that job. If archiving is disabled, Arbiter discards the result.

Arbiter uses ToJSON to store a result and FromJSON to read it. Records and sum types with these instances can be result types.

A parent can use Worker.childResults or Worker.mergedChildResults. Worker.mergedChildResults requires a Monoid result type and combines the child results. It replaces a result that it cannot decode with mempty. A change to the result format can therefore cause an incomplete rollup. Worker.childResults returns an Either for each child and lets the caller handle decode errors.

Use a Maybe result type to make storage conditional for each run. Nothing does not create an archive result or a result row for a parent.

data SyncReport = SyncReport
  { rowsChanged :: Int
  , notes :: [Text]
  }
  deriving stock (Eq, Show, Generic)
  deriving anyclass (ToJSON, FromJSON)

type SyncRegistry = '[ QueueWithResult "sync_queue" SyncPayload (Maybe SyncReport) ]

syncHandler :: Arb.JobHandler (ArbS.SimpleDb SyncRegistry IO) SyncPayload (Maybe SyncReport)
syncHandler _conn job = do
  report <- runSync (Arb.payload job)
  pure $ if rowsChanged report == 0 then Nothing else Just report

See the Arbiter.Core.JobResult haddocks for how a result is encoded.

Job Trees (Fan-out/Fan-in)

Children run in parallel. Parents run when all of their children are acked or DLQ'd.

import Arbiter.Core.JobTree (leaf, rollup, (<~~))
import Data.List.NonEmpty (NonEmpty ((:|)))

data PipelinePayload
  = ProcessChunk Text
  | AggregateSection Text
  | Aggregate
  deriving stock (Generic)
  deriving anyclass (ToJSON, FromJSON)

-- One entry for the whole tree: children and parents share the queue.
type PipelineRegistry = '[ QueueWithResult "pipeline_queue" PipelinePayload [Text] ]

myTree = Arb.defaultJob Aggregate <~~
  ( Arb.defaultJob (ProcessChunk "chunk-1")
      :| [ Arb.defaultJob (ProcessChunk "chunk-2")
         , Arb.defaultJob (ProcessChunk "chunk-3")
         ]
  )
Right _ <- Arb.insertJobTree myTree

Multi-level trees use rollup and leaf:

myTree = rollup (Arb.defaultJob Aggregate)
  ( rollup (Arb.defaultJob (AggregateSection "section-1"))
      ( leaf (Arb.defaultJob (ProcessChunk "leaf-1a"))
          :| [leaf (Arb.defaultJob (ProcessChunk "leaf-1b"))]
      )
      :| [ rollup (Arb.defaultJob (AggregateSection "section-2"))
             (leaf (Arb.defaultJob (ProcessChunk "leaf-2a")) :| [])
         ]
  )

A nested rollup does not automatically merge results into the next level. Each intermediate finalizer must return the merged value.

A parent reads its immediate child results with Worker.mergedChildResults. This function merges successful results and reports DLQ entries by the key used for retryFromDLQ. Arbiter removes intermediate results when it acks the parent.

handler :: Arb.JobHandler (ArbS.SimpleDb PipelineRegistry IO) PipelinePayload [Text]
handler _conn job =
  case Arb.payload job of
    ProcessChunk name -> pure ["processed: " <> name]
    AggregateSection name -> do
      (childResults, dlqFailures) <- Worker.mergedChildResults job
      if not (null dlqFailures)
        then Arb.throwPermanent $ name <> ": has failed children"
        else processSection childResults
    Aggregate -> do
      (childResults, _) <- Worker.mergedChildResults job
      sendToS3 childResults
      pure childResults

config <- Worker.transactionalWorkerConfig 4 handler

Tree-scoped cancellation:

• throwTreeCancel cancels the root and all descendants.
• throwBranchCancel deletes the current job's parent and all descendants of that parent. This includes the current job and its siblings.

Chunked Data Migration

To migrate a large table in parts, assign a set of row identifiers to each child job. The parent runs after all child jobs finish:

import Data.List.NonEmpty qualified as NE

data MigrationJob
  = MigrateChunk [Int64]
  | MigrationComplete
  deriving stock (Generic)
  deriving anyclass (ToJSON, FromJSON)

type MigrationRegistry =
  '[ QueueWithResult "migration_queue" MigrationJob (Sum Int) ]

rowIds <- findRowsToMigrate  -- SELECT id FROM orders WHERE needs_migration
case NE.nonEmpty (chunksOf 1000 rowIds) of  -- chunksOf is from the split package
  Nothing -> reportComplete 0
  Just chunks -> do
    let tree = Arb.defaultJob MigrationComplete
          <~~ fmap (Arb.defaultJob . MigrateChunk) chunks
    Right _ <- Arb.insertJobTree tree
    pure ()

handler conn job = case Arb.payload job of
  MigrateChunk ids -> do
    rowCount <- migrateRows conn ids
    pure (Sum rowCount)

  MigrationComplete -> do
    (Sum totalRows, _) <- Worker.mergedChildResults job
    reportComplete totalRows
    pure (Sum totalRows)

See the Arbiter.Core.JobTree haddocks for the tree builders.

Cron Jobs

import Arbiter.Worker.Cron qualified as Cron

  let Right healthCheck = Cron.cronJob
        "health-check"        -- unique name
        "*/5 * * * *"         -- every 5 minutes (UTC)
        Cron.SkipOverlap      -- skip tick if previous job is still pending/running
        (\_kind tick -> Arb.defaultJob (RunHealthCheck tick))

      -- with backfill: catch up on missed ticks after downtime or scheduler delays
      Right nightlyReport = Cron.cronJob
        "nightly-report"
        "0 3 * * *"           -- 03:00 UTC daily
        Cron.AllowOverlap $ \kind tick -> -- each tick produces its own job
          let jobPriority = case kind of
                Cron.Replay -> 10
                Cron.Live -> 0
           in Arb.defaultJob (GenerateReport tick) & Arb.setPriority jobPriority
      nightlyWithBackfill = nightlyReport {Cron.backfill = Cron.Backfill 86400}

      -- in a specific timezone (validated at construction)
      Right marketOpen = Cron.cronJobInTimezone
        "market-open"
        "America/New_York"    -- IANA tz name
        "30 9 * * 1-5"        -- 09:30 local, Mon-Fri (DST-aware)
        Cron.SkipOverlap
        (\_kind tick -> Arb.defaultJob (OpeningBell tick))

  config <- Worker.transactionalWorkerConfig 4 processScheduled
  let configWithCron =
        config {Worker.cronJobs = [healthCheck, nightlyWithBackfill, marketOpen]}

The builder receives a TickKind (Live for the current minute, Replay for any catch-up tick) and the tick time.

A schedule enqueues jobs on its configured pool. Its builder must return the payload type for that pool. Configure schedules for other queues on their respective pools.

Time zones. Expressions use UTC by default. Use cronJobInTimezone and an IANA name, such as America/New_York, for local time. A schedule of 30 2 * * * does not run on a spring transition day that has no 02:30. A schedule of 30 1 * * * runs one time on a fall transition day that has two occurrences of 01:30.

Backfill. BackfillPolicy replays missed minutes after downtime or a scheduler pause. The policy duration limits the replay period.

Runtime overrides. Use the REST API or admin UI to change a schedule's expression, overlap policy, time zone, and enabled state. These changes do not require a deployment. Set an override to null to use the value from code.

See the Arbiter.Worker.Cron haddocks for the schedule type.

Error Handling

Arb.throwRetryable "API timeout"       -- retry with backoff
Arb.throwPermanent "Invalid payload"   -- move to DLQ immediately
Arb.throwTreeCancel "Pipeline aborted" -- cancel entire tree
Arb.throwBranchCancel "Subtask failed" -- cancel current branch
Arb.throwNack                          -- reprocess later, not a failure (no attempt consumed)

The BatchCallbacks record gives a batched handler these dispositions for each job: failRetry, failPermanent, cancelBranch, cancelTree, and nack. One job's disposition does not change other completed jobs in the batch. A thrown exception applies to all jobs that the handler has not finalized.

Other exceptions are retryable. Arbiter retries a job until it reaches maxAttempts, and then moves it to the DLQ. A payload decode error is permanent because another attempt uses the same invalid payload. Arbiter moves such a job directly to the DLQ.

Exception Classification

Set the disposition where the application classifies the error:

processCharge conn job = do
  result <- liftIO $ chargeCard (Arb.payload job)
  case result of
    Left (RateLimited retryAfter) -> Arb.throwRetryable ("gateway busy: " <> retryAfter)
    Left (CardDeclined reason) -> Arb.throwPermanent ("declined: " <> reason)
    Left (BadRequest reason) -> Arb.throwPermanent reason
    Right receipt -> pure receipt

A retryable error uses one attempt. Arbiter retries the job after the backoff. A permanent error moves the job and its message directly to the DLQ.

throwNack does not record a failure, use an attempt, or call a failure hook. Arbiter processes the job again after the remaining visibility period. Use throwNack when a valid job has an unmet precondition:

processExport conn job = do
  ready <- liftIO $ upstreamReady (Arb.payload job)
  unless ready Arb.throwNack
  runExport conn job

For a tree, throwBranchCancel cancels the current child and its branch. throwTreeCancel cancels the complete tree. Cancellations do not call the failure hook.

Trace Errors

A failed job adds an error status and message to its consumer span. This applies to retries and permanent failures. A batch span can include successful jobs, and therefore does not get an error status. The spans for failed jobs contain the error.

A cancel or nack does not change the span status. See OpenTelemetry.

See the Arbiter.Core.Exceptions haddocks for each disposition.

Dead-Letter Queue

Arbiter moves a job to its queue's DLQ after its last attempt or after a handler calls throwPermanent. See Error Handling.

A DLQJob contains two identifiers. dlqPrimaryKey identifies the DLQ row. jobSnapshot contains the failed job, including its identifier, payload, attempt count, and last error. retryFromDLQ and deleteDLQJob accept the DLQ row identifier.

import Arbiter.Core.Job.DLQ qualified as DLQ

entries <- Arb.listDLQJobs @OrderPayload 50 0
traverse_
  (\e -> logFailure (DLQ.dlqPrimaryKey e) (Arb.lastError (DLQ.jobSnapshot e)))
  entries

Retry

retryFromDLQ takes a DLQ row id and returns the requeued job.

requeued <- Arb.retryFromDLQ @OrderPayload dlqId

A retry recovers the applicable DLQ tree in one statement. The specified row can be any member of the tree. Arbiter restores the root, all descendants in the DLQ, and their finalizers. A retry of one failed fan-out child also restores failed siblings.

A restored finalizer is suspended if it has children in the DLQ or main queue. A finalizer with no children is ready and uses its stored snapshot. Arbiter suspends a queued rollup parent when it restores children below that parent. Arbiter refuses to restore a child if its parent is no longer in the main queue. This rule prevents orphan jobs.

A retried job retains its job identifier, payload, priority, group key, parent link, attempt limit, retention, and admission keys. Arbiter clears its attempt count and error, and makes it immediately visible. The retained identifier preserves parent links in the restored tree.

The parentState in a DLQ rollup finalizer snapshot contains the collected child results. Arbiter records this state before it deletes the children.

[!IMPORTANT] A retry removes the deduplication key. After a job has entered the DLQ, a new insert can use the old IgnoreDuplicate key.

Deletion

deleteDLQJob removes one entry and deleteDLQJobsBatch removes several. Both are permanent.

The REST API and admin UI expose the same list, retry, and delete operations.

See the Arbiter.Core.Job.DLQ haddocks for the entry type.

Rate Limiting

Use an arbitrary key to limit the job rate. A policy applies to all queues in a registry. One policy can control a resource used by multiple queues.

Define a HasRateLimit instance for the payload. Its rateLimitFor function selects a policy for each job. The migration finds and initializes all policies that the selector can use. A separate policy list is not required.

import Arbiter.RateLimit

-- Application functions on the payload.
isTransactional :: EmailPayload -> Bool
recipientDomain :: EmailPayload -> Text

transactional, bulk :: Policy
transactional = tokenBucket "transactional" 100 1 -- 100/second, burst 100
bulk          = tokenBucket "bulk" 1000 3600      -- 1000/hour, burst 1000

instance HasRateLimit EmailPayload where
  rateLimitFor =
    chooseWhen isTransactional
      (limitBy transactional recipientDomain)
      (limitBy bulk recipientDomain)

tokenBucket prefix n period permits n jobs in each period and a burst of up to n jobs. To configure the burst independently, construct a Policy. Set policyMax for the burst. Set policyRefill and policyInterval for the sustained rate. Use rateLimitCost to assign a higher cost to a job. Use addRateLimitTokens to add tokens manually.

When a bucket denies a job, Arbiter makes the job invisible until sufficient tokens are available. Arbiter does not poll the denied job. The API and admin UI show the number of throttled jobs for each policy. An operator can also change a policy at run time.

A fixed window is a manual bucket: declare it with a refill of 0 and reset it at the boundary from a cron.

daily :: Policy
daily =
  Policy
    { policyPrefix = "daily"
    , policyMax = 1000
    , policyRefill = 0
    , policyInterval = 86400
    }

-- In an hourly/daily cron at the window boundary:
resetRateLimitBuckets (policyPrefixOf daily)

Bucket state is not durable by default. After a database crash or failover, each bucket resets to full. Each key can then use one maximum burst before the sustained rate applies. Use durable buckets for strict external quotas or manual buckets that represent credit. Durable bucket state persists across a restart, but can reduce throughput:

runMigrationsForRegistry (Proxy @AppRegistry) connStr "arbiter"
  defaultMigrationConfig { rateLimitDurability = Durable }

Durability is a property of the migrated schema. It is not a property of the registry type. The same registry can use an unlogged staging schema and a durable production schema.

[!IMPORTANT] A job uses tokens when Arbiter claims it. Retries and redeliveries use tokens again. Configure policies for the claim rate.

Arbiter limits a rateLimitCost to the bucket maximum. A job with a higher cost empties a full bucket and can run. A rate limit controls arrivals over time. Use a concurrency limit to control the number of simultaneous jobs.

HTTP 429 Responses

Select the response based on the scope of the limit.

One key is throttled. Empty the bucket for that key. Jobs with the same key then wait for a refill. Nack the current job. Read the key from the job to use the suffix that the claim operation used:

import Arbiter.RateLimit (addRateLimitTokens)
import Data.Foldable (traverse_)

sendEmail job cbs = do
  outcome <- liftIO $ postToVendor (Arb.payload job)
  case outcome of
    TooManyRequests retryAfter -> do
      -- Empty the bucket. Any amount at or above its burst works, tokens floor at zero.
      traverse_ (\key -> addRateLimitTokens key (-1000)) (Arb.jobRateLimitKey (Arb.payloadKeys job))
      void $ Arb.setVisibilityTimeout retryAfter job
      Worker.nack cbs job
    Sent -> Worker.ack cbs job

Nothing means that the selector did not assign a policy to the job. There is no bucket to empty. An empty bucket refills according to its policy.

[!IMPORTANT] This example uses a manual handler. With transactionalWorkerConfig, the bucket update and handler run in one transaction. A retry rolls back the bucket update. Manual and batched callbacks commit independently.

throwRetryable does not specify a delay. The pool calculates the delay from the attempt count, backoffStrategy, and jitter. To use a Retry-After value, set the job visibility period and nack the job.

The complete policy is too fast. Set a lower override and clear it when the vendor recovers. Both functions accept the declared policy. Import that policy to prevent a duplicate declaration:

import Arbiter.RateLimit (Policy (..), clearRateLimit, setRateLimit)

import MyApp.Queue.Policies (transactional)

-- half the declared refill, burst and interval unchanged
void $ setRateLimit transactional {policyRefill = policyRefill transactional / 2}

-- back to what the code declares
void $ clearRateLimit transactional

setRateLimit overrides burst, refill, and interval together. A record update on the declared policy changes one field and leaves the rest alone.

See the Arbiter.RateLimit haddocks for the selector DSL and the policy type.

Concurrency Limiting

Set a maximum number of concurrent jobs for each key. A HasConcurrency instance specifies a pool and a key suffix for each job. A pool consists of a prefix and a default limit. Keys apply to all queues in a registry.

import Arbiter.Concurrency (ConcurrencyPolicy, HasConcurrency (..), concurrencyBy, concurrencyPool)

-- Application function on the payload.
tenantOf :: SyncPayload -> Text

-- At most 2 sync jobs per tenant in flight at once.
syncPool :: ConcurrencyPolicy
syncPool = concurrencyPool "tenant-sync" 2

instance HasConcurrency SyncPayload where
  concurrencyFor = concurrencyBy syncPool tenantOf

Build the selector with noConcurrency, concurrencyBy, globalConcurrency, or concurrencyByCase. The pool limit applies separately to each key with that prefix. An operator can change the pool limit through the API or admin UI. The override applies until an operator clears it. Arbiter then uses the declared default. A value of 0 prevents claims for all keys in the pool.

Concurrency Limit 1 and Group Keys

Both options permit one in-flight job for each key. Their failure behavior is different:

Use a group key for a serial sequence, such as an event stream or state machine. Use concurrency 1 as a mutex, such as one synchronization per tenant. A job can use both features.

[!IMPORTANT] The limit counts claims. A job occupies a slot until Arbiter acks, retries, nacks, or reclaims it. An unacked job continues to occupy a slot after its handler times out.

Arbiter periodically removes inactive keys. It also reconstructs in-flight counts after a restart or failover.

External Limit Updates

A handler can update an override in response to an external capacity signal. For example, update the override when a vendor response reports a new limit.

import Arbiter.Concurrency (setConcurrencyLimit)

import MyApp.Queue.Policies (syncPool)

syncHandler :: Arb.JobHandler (ArbS.SimpleDb SyncRegistry IO) SyncPayload ()
syncHandler _conn job = do
  outcome <- liftIO $ runSync (Arb.payload job)
  case outcome of
    CapacityChanged seats ->
      void $ setConcurrencyLimit syncPool seats
    Ok -> pure ()

The next claim uses the new limit. This change does not require a deployment or restart. Code with a MonadArbiter instance can write the override.

clearConcurrencyLimit syncPool removes the override. Both functions accept a declared pool and use its prefix.

The handler in this example is transactional. The override and job ack commit in the same transaction. If the handler throws an exception, the transaction rolls back both changes.

See the Arbiter.Concurrency haddocks for the selector DSL and the pool type.

Archiving Completed Jobs

Completed jobs are deleted on ack by default. Set archiveFor to keep a copy in a per-queue archive for that many seconds after completion.

job1 = Arb.defaultJob payload & Arb.setArchiveFor (Just Arb.dayRetention)       -- 24h
job2 = Arb.defaultJob payload & Arb.setArchiveFor (Just $ Arb.dayRetention * 7) -- 1 week

Archiving is optional for each job. Arbiter automatically removes expired entries. An archive entry contains the result from its handler. Use the REST API or admin UI to list, re-enqueue, or delete archived jobs.

A re-enqueued job has no parent. It retains its payload and settings. Re-enqueueing one member of a completed tree creates one independent job. To recover a failed tree, retry it from the dead-letter queue.

See the Arbiter.Core.Job.Archive haddocks for the archive row and its queries.

Worker Configuration

A WorkerConfig defines a handler, thread count, timing values, and callbacks. Configuration constructors return this record. Update its fields before pool creation.

poolConfigForWorkers calculates the database pool size from a list of worker pools. Pass the same list to poolConfigForWorkers and runWorkerPools.

Multiple Queues

One process can run one pool for each queue. Create each configuration and name it with namedWorkerPool. Pass the same list to poolConfigForWorkers, runWorkerPools, and shutdownPools.

If one pool exits, Arbiter stops the other pools. After all pools stop, Arbiter throws the first recorded failure.

ARBITER_ENABLED_QUEUES is a comma-separated list of pool names. runWorkerPools starts the named pools. If the variable is not set, it starts all configured pools. This variable permits different deployments to use the same binary.

ARBITER_ENABLED_QUEUES=email_queue,image_queue

Arbiter checks the names against the configured pools at startup. An unknown name causes an exception.

Configuration Types

transactionalWorkerConfig runs the handler in a transaction. A normal return acks the job and stores the returned result. An exception rolls back the work before Arbiter retries the job or moves it to the DLQ. Arbiter finalizes the job on each path.

manualWorkerConfig and defaultBatchedWorkerConfig do not start a handler transaction. They supply finalization callbacks. Create a transaction for the writes that require one. For example, an HTTP request can run without a held database connection. The handler must ack, fail, or nack each job. Arbiter reprocesses jobs that the handler does not finalize before visibility expires.

Wrap a callback in withDbTransaction to commit the ack and application writes atomically. The callback's transaction becomes a savepoint. The success hook runs when Arbiter releases that savepoint. It can run before the outer transaction commits. If the outer transaction rolls back, Arbiter can process the job again after the hook has run.

Batching is independent of transaction mode. Use defaultBatchedWorkerConfig when per-job overhead is too high and each job requires a separate disposition in one claim.

Timings

visibilityTimeout is how long a claim holds a job, and jobHeartbeatInterval is how often the worker renews that hold. maxJobDuration caps how long a handler may run at all. See Leases and Deadlines.

See the WorkerConfig haddocks for all options.

Batched Handlers

defaultBatchedWorkerConfig configures a manual handler that receives up to batchSize jobs in each invocation. The handler can combine operations for these jobs. A batch of grouped jobs contains one group. A batch of ungrouped jobs contains jobs from the ready set. Use the supplied callbacks to finalize each job.

-- defaultBatchedWorkerConfig <workerCount> <batchSize> handler
config <- Worker.defaultBatchedWorkerConfig 10 5 batchHandler

batchHandler
  :: NonEmpty (Arb.JobRead ImagePayload)
  -> Worker.BatchCallbacks (ArbS.SimpleDb AppRegistry IO) ImagePayload Score
  -> ArbS.SimpleDb AppRegistry IO ()
batchHandler jobs cbs = do
  -- bulkProcess :: [Arb.JobRead ImagePayload] -> IO [(Arb.JobRead ImagePayload, Score)]
  scored <- liftIO $ bulkProcess (toList jobs)
  -- Bulk-ack the whole batch in one transaction.
  Worker.ackAllWith cbs scored

Each callback runs in a separate transaction. Wrap a callback in withDbTransaction to commit the ack and application writes in one transaction:

batchHandler jobs cbs =
  for_ jobs $ \job -> do
    score <- liftIO $ scoreImage (Arb.payload job)
    Arb.withDbTransaction $ do
      recordCharge (Arb.payload job)
      Worker.ackWith cbs job score

onJobSuccess does not commit in the transaction with these writes. It can run for a job that Arbiter later processes again. Put effects that must occur one time in the same transaction as the ack.

A disposition applies to one job. A failure, cancellation, or nack does not change completed jobs in the batch. Arbiter reprocesses an unfinalized job. ackWith and ackAllWith store the queue result. ack and ackAll do not store a result and work with all queues. The BatchCallbacks haddocks list all dispositions.

Observability Hooks

ObservabilityHooks contains callbacks for points in the job lifecycle. Start with defaultObservabilityHooks, update the required fields, and assign the record to the pool configuration. The default callbacks have no effect.

myHooks = Arb.defaultObservabilityHooks
  { Arb.onJobSuccess = \job startTime endTime ->
      liftIO $ recordHistogram "jobs.duration" (diffUTCTime endTime startTime)
  , Arb.onJobFailedAndMovedToDLQ = \err job ->
      liftIO $ sendAlert (Arb.primaryKey job) err
  , Arb.onJobHeartbeat = \job now startTime ->
      liftIO $ recordGauge "jobs.running_duration" (realToFrac $ diffUTCTime now startTime)
  }

config <- Worker.transactionalWorkerConfig 5 handler
let instrumented = config { Worker.observabilityHooks = myHooks }

A hook runs in the pool monad. It can read the database and write to a metrics client.

Hook Invocation

Each claimed job calls onJobClaimed. It then has one of these outcomes:

A cancellation calls onJobCancelled. A nack does not call a hook.

Each onJobFailure call is followed by onJobRetry or onJobFailedAndMovedToDLQ. Measure failure duration in onJobFailure. Count failures in one of the two outcome hooks to prevent duplicate counts.

If a failure update finds no row, Arbiter calls onJobUnavailable. Another worker owns the job and reports its outcome.

Each successful heartbeat extension for a running job calls onJobHeartbeat. Reclaimed or cancelled jobs do not call this hook.

Hook Composition

ObservabilityHooks is a Monoid. The <> operator runs the left callback before the right callback at each point. Arbiter runs the right callback if the left callback throws an exception. withHooks combines a record with the hooks in an existing configuration:

let instrumented = Worker.withHooks (myHooks <>) config

arbiter-otel uses this method to add instrumentation. Its metrics and application hooks can use one configuration. See OpenTelemetry.

Hook Restrictions

Arbiter discards a hook return value. It catches hook exceptions, logs them at Warning, and continues the worker.

onJobSuccess can run for a job that Arbiter processes again. Put effects that must occur one time in the same transaction as the ack. See Batched Handlers.

Reaper activity reports through onMaintenance on WorkerConfig, not through the hooks record.

See the ObservabilityHooks haddocks for each callback's arguments.

Leases and Deadlines

config { Worker.visibilityTimeout = 60 }     -- how long a claim holds a job (default)
config { Worker.jobHeartbeatInterval = 30 }  -- how often the worker renews that hold (default)
config { Worker.maxJobDuration = Just 300 }  -- longest a handler may run (default: Nothing)

A claim is a lease. It sets not_visible_until on the row. The worker owns the job until that time. A heartbeat thread renews the lease at each jobHeartbeatInterval while the handler runs. This renewal permits a slow job to use a short visibilityTimeout.

Three things end a handler before it returns:

A reclaim check requires a database response. The lease fence uses the locally stored deadline. It stops the handler if the worker cannot contact the database and the lease expires. The fence is always active and has no configuration.

maxJobDuration

config <- Worker.transactionalWorkerConfig 4 processReport
let reportConfig = config { Worker.maxJobDuration = Just 300 }

An exceeded limit is a retryable failure. Arbiter applies the configured backoff and moves the job to the DLQ at maxAttempts. last_error contains the exceeded duration. Set this limit for handlers that call external services or occupy concurrency pool slots.

[!IMPORTANT] If maxJobDuration is not set, an unresponsive handler retains its job while the process runs. The heartbeat continues to renew the lease. Another worker cannot reclaim the job.

Timing Constraints

jobHeartbeatInterval must be less than visibilityTimeout. The pool does not start if the values are invalid. After a database connection failure, the worker can continue for up to one visibilityTimeout after its last successful renewal. The minimum period is the difference between the two settings. The exact period depends on the point of failure in the heartbeat cycle.

Arbiter retries a failed extension. It shortens the interval between attempts as the lease expiration time approaches. The fence stops the handler if all attempts fail before expiration.

Reduce visibilityTimeout to stop handlers sooner after lost leases. This also causes earlier redelivery after a worker stops.

See the WorkerConfig haddocks for every timing field.

Graceful Shutdown

Install signal handlers after construction of the worker configurations. Pass the same pool list to shutdownPools and runWorkerPools:

import System.Posix.Signals qualified as Signals

emailConfig <- Worker.transactionalWorkerConfig 3 processEmail
imageConfig <- Worker.transactionalWorkerConfig 2 processImage

let workers = [Worker.namedWorkerPool emailConfig, Worker.namedWorkerPool imageConfig]
    shutdown = Signals.Catch $ Worker.shutdownPools workers
void $ Signals.installHandler Signals.sigTERM shutdown Nothing
void $ Signals.installHandler Signals.sigINT shutdown Nothing

poolCfg <- Worker.poolConfigForWorkers workers
env <- ArbS.createSimpleEnvWithConfig (Proxy @AppRegistry) connStr "arbiter" poolCfg
ArbS.runSimpleDb env $ Worker.runWorkerPools workers

The dispatcher stops new claims and waits for in-flight jobs. runWorkerPools returns when those jobs finish or when gracefulShutdownTimeout expires. Arbiter does not finalize a job that still runs at the timeout. It redelivers the job after its visibility period expires.

Backoff Strategies

config { Worker.backoffStrategy = exponentialBackoff 2.0 3600 }  -- base^attempts, cap 1h
config { Worker.backoffStrategy = linearBackoff 30 600 }         -- +30s/attempt, cap 10m
config { Worker.backoffStrategy = constantBackoff 60 }           -- always 60s
config { Worker.backoffStrategy = Custom (\n -> fromIntegral n * 15) }

config { Worker.jitter = FullJitter }   -- random(0, delay)
config { Worker.jitter = EqualJitter }  -- delay/2 + random(0, delay/2) (default)
config { Worker.jitter = NoJitter }

The claim operation increments the attempt count before the handler starts. Arbiter calculates the delay from the new count. The first failure is attempt

1. Therefore, exponentialBackoff 2.0 gives a two-second delay before the first retry.

Jitter changes the calculated delay. The default EqualJitter selects a value between one half and all of the calculated delay. The strategy value is the maximum delay.

A nack does not use the backoff strategy. The job remains unavailable for the rest of its lease. Set the job visibility timeout before the nack to control this period. See Error Handling.

See the Arbiter.Worker.BackoffStrategy haddocks for every strategy and jitter mode.

Wakeups (LISTEN/NOTIFY)

PostgreSQL LISTEN/NOTIFY sends immediate notifications for new jobs, pause and resume operations, force-cancel operations, and manual cron runs. The MonadArbiter instance supplies a shared listener hub to the workers. This applies to SimpleDb, HasqlDb, and custom monads.

If there is no listener, workers check for jobs at each pollInterval. Control operations use these intervals:

The environment does not open a listener connection until a worker pool starts. A producer process that does not start a pool does not open this connection. While workers run, the listener uses one pool connection.

On the provided backends, useDedicatedListener creates a separate listener connection:

env <- ArbS.useDedicatedListener connStr =<< ArbS.createSimpleEnv (Proxy @AppRegistry) connStr "arbiter"

disableListener disables the listener and uses polling mode:

env <- ArbS.disableListener <$> ArbS.createSimpleEnv (Proxy @AppRegistry) connStr "arbiter"

Logging

A pool writes structured JSON logs. logConfig sets the destination and level. defaultLogConfig writes Info and higher levels to stdout.

Available destinations are stdout, stderr, a fast-logger LoggerSet, and a custom callback. The callback receives the level, message, and structured context as [Pair]. Use it to send Arbiter logs to an application logging system. silentLogConfig disables logging.

Arbiter adds job context to handler logs and pool logs.

See the Arbiter.Worker.Logger haddocks for LogConfig and every destination.

Liveness Probes

The heartbeat loop updates livenessFile at each worker heartbeat. A probe can check this file without a database query. The default file is arbiter-worker-<workerId> in the system temporary directory.

Configure the probe to fail when the file is stale:

livenessProbe:
  exec:
    command: ["sh", "-c", "find ${TMPDIR:-/tmp}/arbiter-worker-* -mmin -5 | grep -q ."]
  initialDelaySeconds: 30
  periodSeconds: 60

find exits with status 0 when no file matches -mmin. Therefore, a command that uses only find passes when the heartbeat is stale. The grep -q . command makes the probe fail when no current heartbeat file exists. The pool removes the file after a normal shutdown.

The REST API provides two other checks. GET health is a readiness check that queries the database. GET health/live is a liveness check that does not query the database. See REST API and Admin UI.

Pausing Work

A pause stops new claims. It does not stop in-flight jobs. These jobs can complete while the queue is paused.

pauseChildren does not suspend in-flight jobs or jobs that have a delay or backoff. resumeChildren keeps a finalizer suspended if its children are still queued. Arbiter resumes the finalizer after its children finish.

LISTEN/NOTIFY sends a pause notification to running pools. Without a listener, each pool reads the pause state at its next worker heartbeat. This can take one workerHeartbeatInterval. See Wakeups.

The REST API and admin UI provide the same controls. An operator can pause work without a deployment.

REST API and Admin UI

The arbiter-servant and arbiter-servant-ui packages provide a REST API and admin dashboard. Use them as standalone WAI applications or integrate them into an existing Servant API.

import Arbiter.Servant qualified as Servant

config <- Servant.initArbiterServer (Proxy @AppRegistry) connStr "arbiter"
Servant.runArbiterAPI 8080 config

Embed as a sub-route in an existing Servant application:

import Arbiter.Servant qualified as Servant
import Arbiter.Servant.UI qualified as ServantUI

type MyAPI =
  "api" :> MyBusinessRoutes
    :<|> "arbiter" :> (Servant.ArbiterAPI AppRegistry :<|> ServantUI.AdminUI)

See the arbiter-servant-ui haddocks for the UI's route type.

POST jobs and POST jobs/batch enqueue jobs. Services in other languages can use these endpoints without an Arbiter library. The admin UI uses the other per-queue endpoints for operator functions.

Endpoints

Per-queue endpoints under /api/v1/:queue/:

Global endpoints under /api/v1/:

Consuming over HTTP

POST claim applies the worker-pool claim operation. It uses admission tokens, increments the attempt count, records a claimant, and makes each job invisible for the lease period. A paused queue does not return leases. The pause applies to HTTP consumers and worker pools.

POST /api/v1/email_queue/claim
{"maxJobs": 5, "leaseSeconds": 60}

maxJobs defaults to 1 and clamps to 1000. leaseSeconds defaults to 60 and clamps to 3600.

Each response job contains claimSeq and claimedBy. These fields identify the lease. Each finalization request must include them:

POST /api/v1/email_queue/jobs/41/ack
{"claimSeq": 7, "claimedBy": "0f5e...c31"}

On a queue declared with QueueWithResult, ack also takes the result:

POST /api/v1/email_queue/jobs/41/ack
{"claimSeq": 7, "claimedBy": "0f5e...c31", "result": ["delivered"]}

Arbiter stores the result in the parent rollup for a child job, or in the archive entry for an archived root job. This is the same behavior as ackWith. A body that does not match the queue result type returns 400. Omit result to store no result. Mounting this route requires FromJSON and ToJSON for the result type.

ack completes the job. nack restores the used attempt and keeps the job invisible for the remainder of its lease. extend sets a later lease expiration, as a worker heartbeat does. The request body requires seconds. Arbiter limits the value to 3600 and measures it from the request time. If the lease fields do not match the row, the endpoint returns 409. A previous lease holder cannot ack a reclaimed job.

These routes finalize leases created by POST claim. They return 409 for a lease held by a worker pool.

The server does not renew an HTTP lease automatically. The consumer must call extend. After an unextended lease expires, another consumer can claim the job. This behavior provides at-least-once delivery after a consumer failure.

[!IMPORTANT] Arbiter does not authenticate claim or finalization requests. Add authentication before exposing these routes. Use WAI middleware, a Servant authentication combinator, or an authenticating proxy.

Maintenance

POST maintenance performs one pass of schema maintenance. It processes stale workers, exhausted and cancelled jobs, rate-limit buckets, concurrency counts, archive retention, and group summaries.

Concurrent callers cannot run the same operation. The default has no minimum interval. Set maintenanceInterval in the server configuration to define the minimum interval for each operation. maintenanceTimeout limits one statement.

maintenanceSparseInterval defines a separate minimum interval for schema-wide operations. maintenanceBucketIdle specifies how long a rate-limit bucket must be inactive before removal.

The response gives the affected row count for each completed operation and the names of failed operations. Operations absent from both lists were skipped:

{"ops": {"sweep-stale-workers": 2, "purge-archives": 140}, "failed": []}

Worker pools run this maintenance in their reaper. Use the endpoint when a deployment does not run a worker pool.

OpenTelemetry

Arbiter includes spans and W3C trace-context propagation. These features do not require configuration. The arbiter-otel package adds metrics, gauges, and OTel log records over OTLP:

import Arbiter.Otel qualified as Otel

main :: IO ()
main = do
  env <- createSimpleEnv (Proxy @AppRegistry) connStr "arbiter"

  runSimpleDb env $
    Otel.runWorkerPools [namedWorkerPool emailCfg, namedWorkerPool imageCfg]

Use Otel.runWorkerPools in place of runWorkerPools. The arguments are the same. It installs the SDK, instruments the pools, and starts the gauges. Call it one time in each process.

Standard OTEL_* variables configure exporters, endpoints, and intervals. Set OTEL_SDK_DISABLED=true to disable the SDK. Arbiter sends logs to OTel and to the configured log destination. Each log contains the job trace, ID, queue, and attempt.

Use runWorkerPoolsWith and a bracket from Arbiter.Otel to manage the telemetry handle, install a separate SDK, or start pools by another method. The With functions also accept the base log configuration for the gauge loop.

Traces

Each enqueue records the current span. Each claim starts a process <queue> consumer span and links it to the enqueue span. The link works across processes and for jobs that a handler enqueues. A REST API enqueue joins the request trace when the server uses newOpenTelemetryWaiMiddleware (hs-opentelemetry-instrumentation-wai).

Both spans carry the job's payload kind as arbiter.kind. The producer span derives it from the payload. The consumer span reads the stored label.

Arbiter.Core.Trace has the helpers for annotating a job's span, opening child spans, and wrapping an enqueue made outside a handler.

Metrics

arbiter-otel reports job activity, queue depth, admission policies, reaper activity, Arbiter table health, and PostgreSQL health. The Arbiter.Otel.MetricNames module defines the name and unit of each instrument.

Admission metrics use the policy as the key. They do not use admission keys. On these metrics, policy_kind is the policy type: rate_limit or concurrency.

Queue depth and the job counters set kind only to a label from the payload's kindsFor set. A payload that declares no labels exports no kind. The number of series is therefore bounded by that set.

Grant pg_read_all_stats to collect PostgreSQL health data outside the Arbiter role. One replica scans during each interval. The other replicas export that reading. Across replicas, use max for queue depth and PostgreSQL health. Use sum for per-process counters and latencies.

Queue and Postgres gauges are scanned once per OTEL_METRIC_EXPORT_INTERVAL (default 60s).

Prometheus

Arbiter sends metrics over OTLP. Configure Prometheus to scrape an OTel collector. Arbiter does not support OTEL_METRICS_EXPORTER=prometheus. This setting disables metrics.

Local stack

arbiter-demo/run-local.sh runs this repository's demo against Grafana's LGTM stack at http://localhost:8000, with the dashboard at /dash. The live demo runs the same stack.

The dashboard and the alert rules assume metrics arrive over OTLP through a collector.

Backend Integration

The MonadArbiter typeclass separates the core from database libraries. Arbiter provides three adapters: arbiter-simple, arbiter-orville, and arbiter-hasql. Use the adapter for the database library in your application if you want to share connections.

Benchmarks

Jobs/sec with arbiter-hasql, prepared claims, PostgreSQL 18, GHC 9.14.1, Apple M5 Pro. Single-job mode / batched mode (batch size 10).

Pre-loaded queue (1M jobs, 4 pools × 10 workers):

scheduled + backoff: a fifth of jobs scheduled seconds out, a fifth failing once into backoff. dormant: half the backlog parked 30 days out.

Steady state (10 producers inserting continuously, 4 pools × 10 workers):

Group size and skew (300k jobs, 4 pools × 10 workers):

Admission gating (steady state, 10 producers, 256 keys):

One pool is 10 workers and one dispatcher.

arbiter-simple (postgresql-simple)

This backend uses postgresql-simple and resource-pool. Handlers receive a raw Connection. Nested transactions automatically use savepoints.

env <- ArbS.createSimpleEnv (Proxy @AppRegistry) connStr "arbiter"
ArbS.runSimpleDb env $ Arb.insertJob (Arb.defaultJob $ SendWelcome "alice@example.com" "Alice")

Share a transaction with external database work:

PG.withTransaction conn $ do
  PG.execute conn "INSERT INTO orders (id) VALUES (?)" (PG.Only orderId)
  ArbS.inTransaction @AppRegistry conn "arbiter" $
    Arb.insertJob (Arb.defaultJob (ProcessOrder orderId))

See the arbiter-simple haddocks for the env and pool constructors.

arbiter-orville (orville-postgresql)

This backend uses orville-postgresql. Orville manages its connections and transactions. Handlers do not receive a connection parameter. Define a custom monad with MonadOrville and MonadArbiter instances:

{-# LANGUAGE TypeFamilies #-}

instance MonadArbiter AppM where
  type RegistryOf AppM = AppRegistry
  type Handler AppM job result = job -> AppM result
  getSchema = asks appSchema
  -- ... executeQuery / executeStatement / withDbTransaction / runHandlerWithConnection

Handler specifies the handler type. Because Orville does not pass a connection, this type contains the job argument only. Use Arb.JobHandler AppM payload result in handler signatures. Writing a Backend covers the methods elided above.

Orville does not expose its pooled connections for LISTEN/NOTIFY. Create a DedicatedListen from Arbiter.Core.Listen with the Orville pool connection string. Store it in the reader environment and return it from getListener.

createOrvilleConnectionOptions accepts an Arbiter PoolConfig. Use poolConfigForWorkers to calculate the Orville pool size:

import Arbiter.Core.Listen (DedicatedListen, dedicatedListener, newDedicatedListen)

data AppEnv = AppEnv
  { appSchema  :: SchemaName
  , appOrville :: O.OrvilleState
  , appListen  :: DedicatedListen
  }

main :: IO ()
main = do
  poolCfg <- Worker.poolConfigForWorkers workers
  orvillePool <- O.createConnectionPool (createOrvilleConnectionOptions connStr poolCfg)
  listen <- newDedicatedListen connStr
  let env =
        AppEnv
          { appSchema = "arbiter"
          , appOrville = O.newOrvilleState O.defaultErrorDetailLevel orvillePool
          , appListen = listen
          }
  runAppM env $ Worker.runWorkerPools workers

instance MonadArbiter AppM where
  -- ... RegistryOf / Handler / getSchema and the query methods, as above
  getListener = asks (Just . dedicatedListener . appListen)

See the arbiter-orville haddocks for the connection options.

arbiter-hasql (hasql)

This backend uses hasql and resource-pool. Handlers receive a Hasql.Connection for typed queries in the worker transaction.

env <- ArbH.createHasqlEnv (Proxy @AppRegistry) connStr "arbiter"
ArbH.runHasqlDb env $ Arb.insertJob (Arb.defaultJob $ SendWelcome "alice@example.com" "Alice")

Share a transaction with external hasql work:

-- Session.script (hasql >= 1.10) or Session.sql (hasql < 1.10)
_ <- Hasql.use conn (Session.script "BEGIN")
ArbH.inTransaction @AppRegistry conn "arbiter" $
  Arb.insertJob (Arb.defaultJob (ProcessOrder orderId))
_ <- Hasql.use conn (Session.script "COMMIT")

See the arbiter-hasql haddocks for the env and pool constructors.

Writing a Backend

A backend is a MonadArbiter instance. Applications that use arbiter-orville must define this instance. Use the same interface to implement another adapter.

The class specifies a registry, handler type, and these database operations:

• RegistryOf specifies the queue registry for the monad.
• Handler specifies the handler type. It can include a connection argument if the database library supplies one.
• getSchema returns the Arbiter schema.
• executeQuery and executeStatement run a Query. The value contains its SQL text, parameters, and decoder.
• withDbTransaction starts a transaction or savepoint. A nested call must start a savepoint. This lets a finalization callback participate in the caller's transaction. See Worker Configuration.
• runHandlerWithConnection checks out a connection and runs a handler.
• getListener returns the shared LISTEN/NOTIFY listener. Return Nothing for polling mode. See Wakeups.

executeQueryPrepared is optional and uses executeQuery by default. Override it if the backend can prepare a statement one time for each connection and reuse the plan. The claim operation uses this method. See the performance data in Backend Integration.

See the MonadArbiter haddocks for each method's signature.