Arbiter

A PostgreSQL job queue for Haskell applications.

• Transactional job processing: jobs and database operations commit together
• At-least-once delivery with leases 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

Installation

Add the repository to your project. Replace arbiter-simple with arbiter-orville or arbiter-hasql in the lists below to use that backend. Orville also needs arbiter-libpq for the listener, and Hasql also needs pqi-ffi.

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-libpq
    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-libpq
      - arbiter-migrations

Quick Start

Payload and Result Types

Payload and result types need ToJSON and FromJSON:

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

Queue has no result. QueueWithResult names the result type.

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

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

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;

Rerun the migrations after changing enableNotifications, enableEventStreaming, or rateLimitDurability. Do not edit the migration history.

Replicas can migrate concurrently. migrationLockTimeout bounds the wait for the migration lock, unbounded by default. A transaction-mode pooler 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
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 Nothing when a dedup key skips the insert.

Configuring a Job

Configure defaultJob or defaultGroupedJob with 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 of five worker threads, each handler in a transaction
  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 commits the handler's writes and the ack together. An exception rolls back, then retries or moves to the DLQ.

manualWorkerConfig runs no transaction and passes callbacks:

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

Configuration details: Worker Configuration, Batched Handlers.

Architecture

Each worker pool claims jobs from PostgreSQL. There is no broker and no leader. Worker capacity scales with process count.

The lifecycle under transactionalWorkerConfig:

1. Claim: The dispatcher claims visible jobs in per-group order, increments each attempt count, and hides each job for its lease. The same statement applies admission: a job with an empty rate-limit bucket is parked until its bucket refills, and a job whose concurrency key is at its limit is skipped. A heartbeat extends the lease while the handler runs.
2. Run: The handler runs in a transaction. Its database work, its stored result, and the ack commit together.
3. Success: The transaction commits.
4. Failure: The transaction rolls back. A second transaction retries the job with backoff or moves it to the dead-letter queue (DLQ).
5. Reclaim: Another worker claimed the job after its lease expired. The heartbeat or the ack throws and the worker drops the job.

Admission limits apply to every worker pool in every process and to REST API clients.

Delivery is at least once. Non-transactional side effects must be idempotent.

With manualWorkerConfig and batchedWorkerConfig, step 2 has no transaction and the handler finalizes each job through callbacks.

Group Ordering

A group key admits one job or batch at a time. Different groups run concurrently.

• Same group key: Insertion order within priority. A retrying job stays first until it succeeds or moves to the DLQ. A ready job runs before a delayed job. The group waits through a job's backoff or rate-limit delay.
• No group key: Any free worker, concurrently.

Priority

Lower numbers claim first. The default is 0.

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

A group is eligible when it has a ready job.

Payload Kinds

A payload kind is an optional Text label stored with each job at insert. Label Use lists where it appears.

HasKind has two members:

With no instance for the payload, kindOf returns Nothing and kindsFor is [].

Constructor Labels

The default methods use constructor names. Derive Generic and declare an empty instance:

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

instance HasKind EmailPayload

kindOf (SendReceipt 7) is Just "SendReceipt". kindsFor @EmailPayload is ["SendWelcome", "SendReceipt"].

Custom Labels

Implement both members for other labels:

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]

kindOf (EmailPayload Receipt "a@b.c") is Just "receipt". kindsFor @EmailPayload is ["welcome", "receipt", "passwordreset"].

kindsFor must list every label kindOf can return. Kind metrics and kindCounts skip undeclared labels.

Labels from a Nested Type

constructorKind and constructorKinds read the constructors of any Generic type:

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

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

Label Use

API details: Arbiter.Core.Job.Kind.

Deduplication

A dedup key sets what happens when the key is already queued:

-- skip the insert
job1 = Arb.defaultJob payload & Arb.setDedupKey (Just $ IgnoreDuplicate "order-123")

-- replace the queued job
job2 = Arb.defaultJob payload & Arb.setDedupKey (Just $ ReplaceDuplicate "order-123")

Keys are scoped to one queue. A job in the DLQ or the archive is not a duplicate.

ReplaceDuplicate copies payload, priority, group key, attempt limit, admission keys, and retention from the new job. It clears the attempt count, last error, and claim.

A replacement is refused when the queued job is in flight, has a force-cancel flag, has a different parent, or has children in the queue or DLQ. The queued job is unchanged.

Key type: Arbiter.Core.Job.Dedup Haddocks.

Job Results

transactionalWorkerConfig stores a handler's return value. Manual and batched handlers pass results to ackWith or ackAllWith.

Result types need ToJSON and FromJSON.

A parent reads its children with one of:

Nothing in a Maybe result stores nothing:

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

Result encoding: Arbiter.Core.JobResult Haddocks.

Job Trees (Fan-out/Fan-in)

Children run in parallel. A parent runs when every child is acked or in the DLQ.

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")) :| [])
         ]
  )

Each intermediate finalizer returns the value for the level above.

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

Spawning Children at Runtime

spawn inserts children under the running job and suspends it, in one transaction. The job wakes when every child has left the main queue.

batchHandler jobs cbs = for_ jobs $ \job -> do
  (results, _) <- Worker.mergedChildResults job
  chunks <- discoverChunks (Arb.payload job) results
  case NE.nonEmpty chunks of
    Nothing -> do
      sendToS3 results
      Worker.ack cbs job
    Just cs -> Worker.spawn cbs job (fmap (Arb.defaultJob . ProcessChunk) cs)

With transactionalWorkerConfig, call Arb.spawnChildren in the handler. The worker's ack suspends the job in the same transaction. Every round stores the handler's return value. Return mempty from a round that spawns.

• Hooks. A spawn round fires onJobSuccess.
• Attempts. A spawn refunds the attempt its claim spent. Rounds are unbounded.
• Results. Each spawn deletes the previous round's results.
• Crash before commit. The job is reprocessed and spawns again. Give the children dedup keys when a repeat insert is wrong.
• Child in the DLQ. It wakes the finalizer. Its error is in that round's mergedChildResults. The next spawn detaches it. A DLQ retry restores it as a root.
• Dedup conflict on a child. The handler aborts and commits nothing. The abort names the spawning job. Batch siblings are nacked.
• Callback after a spawn. Ignored, with a warning.

Chunked Data Migration

Each child migrates one set of row ids:

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)

Tree builders: Arbiter.Core.JobTree Haddocks.

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 a catch-up tick) and the tick time.

A schedule enqueues one job per tick on its pool's queue. Tree jobs spawn children at runtime.

Time zones. Expressions are UTC. cronJobInTimezone takes an IANA name. 30 2 * * * skips a spring transition day with no 02:30. 30 1 * * * runs once on a fall transition day with two 01:30s. A * minute or hour field (*/5 * * * *, 0 * * * *) runs through both 01:00 hours.

Backfill. Backfill n replays ticks missed in the last n seconds after downtime or a scheduler pause. If a tick fails to insert, the scheduler logs the failure and continues with later ticks. A failed tick is skipped once a later tick fires. For a Backfill schedule, a failed newest tick is retried on the next pass.

Runtime overrides. The REST API and admin UI set a schedule's expression, overlap policy, time zone, and enabled state. An override of null restores the value from code.

Schedule type: Arbiter.Worker.Cron Haddocks.

Error Handling

Arb.throwRetryable "API timeout"             -- retry with backoff
Arb.throwRetryableAfter 30 "Rate limited"    -- retry in 30 seconds
Arb.throwPermanent "Invalid payload"         -- move to the DLQ now
Arb.throwTreeCancel "Pipeline aborted"       -- delete the tree
Arb.throwBranchCancel "Subtask failed"       -- delete the branch
Arb.throwNack                                -- reprocess after the lease, no attempt spent
Arb.throwNackAfter 60                        -- reprocess in 60 seconds, no attempt spent

Delays are in seconds and round up to whole seconds. The pool's backoff and jitter do not apply to them. A grouped job parked by either delay holds its group until it runs, as a backoff does.

A batched handler uses failRetry, failRetryAfter, failPermanent, cancelBranch, cancelTree, nack, and nackAfter on BatchCallbacks per job. A thrown exception applies to every job the handler has not finalized.

Exception Classification

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

throwNackAfter for an unmet precondition, checked again in a minute:

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

Trace Errors

A failed job sets an error status and message on its consumer span. A batch span has no error status. A cancel or nack leaves the span status unchanged. Trace details: OpenTelemetry.

Exception dispositions: Arbiter.Core.Exceptions Haddocks.

Dead-Letter Queue

A job moves to its queue's DLQ after its last attempt or after throwPermanent. Failure modes: Error Handling.

dlqPrimaryKey is the DLQ row id, which retryFromDLQ and deleteDLQJob take. jobSnapshot is the failed job.

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

requeued <- Arb.retryFromDLQ @OrderPayload dlqId

One retry restores the whole tree that contains the row, in one statement:

• The root, every descendant in the DLQ, and their finalizers return to the main queue.
• A retry of one fan-out child also restores its failed siblings.
• A restored finalizer is suspended while it has children in the DLQ or main queue. With no children it is ready and uses its stored snapshot.
• A queued rollup parent is suspended when children are restored below it.
• A child whose parent has left the main queue is not restored.

A retried job keeps its id, payload, priority, group key, parent link, attempt limit, retention, and admission keys. Its attempt count and error are cleared. It is visible at once.

parentState in a rollup finalizer snapshot holds the child results collected before the children were deleted.

[!IMPORTANT] A retry drops the deduplication key. After a job enters the DLQ, a new insert can reuse its IgnoreDuplicate key.

Bulk Retry

retryFromDLQBatch retries several entries. retryDLQFiltered retries every entry the filters match when it starts, oldest first. With no filters it retries the whole DLQ. Each tree retries in its own transaction. A failure part way leaves the trees already retried in the main queue.

-- Every entry whose last error mentions the outage
count <- Arb.retryDLQFiltered @OrderPayload [Arb.FilterErrorText "connection refused"]

Both return the number of entries whose retry restored a tree. An entry restored with an earlier entry's tree, or one whose parent is gone, is not counted.

Deletion

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

Retention

Entries stay until they are retried or deleted. A queue's DLQ retention purges old ones:

Arb.setQueueDLQRetention "orders" (Just (30 * 86400)) -- 30 days
Arb.setQueueDLQRetention "orders" Nothing             -- keep entries

The retention lives on the queue's arbiter_queues row. Every worker pool and REST server sharing the schema applies the same value. The reaper deletes entries that failed longer ago than the retention, up to 1000 per queue per pass. A purge resumes a parent left without children, as deleteDLQJob does. getQueue returns the current retention as dlqRetention.

The REST API and admin UI expose list, retry, bulk retry, and delete.

Entry type: Arbiter.Core.Job.DLQ Haddocks.

Rate Limiting

A policy limits the claim rate per key across every queue in the registry. HasRateLimit selects a policy and key for each job. The migration creates every policy the selector can return.

import Arbiter.RateLimit

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

transactional, bulk :: RateLimitPolicy
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 admits n jobs per period seconds with a burst of n. For a different burst, build a RateLimitPolicy:

rateLimitCost sets a job's token cost, 0 or more. The default is 1. addRateLimitTokens adjusts a bucket's tokens.

A denied job is invisible until its bucket has enough tokens. The API and admin UI show the throttled count per policy and accept policy changes at run time.

A fixed window is a bucket with refill 0, reset from a cron:

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

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

Durability

Buckets are unlogged by default. After a crash or failover every bucket is full. Durable buckets survive a restart and cost throughput:

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

Durability is set per migrated schema.

[!IMPORTANT] Tokens are spent at claim. Retries and redeliveries spend tokens again.

A cost above the effective max is clamped to it. The effective max is the operator override when set, else policyMax.

HTTP 429 Responses

One key is throttled. Empty its bucket, set the lease to Retry-After, and nack:

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 the burst works. Tokens stop 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

[!IMPORTANT] With transactionalWorkerConfig the bucket update rolls back with the retry. Manual and batched callbacks commit on their own.

Policy-wide throttling. Override the policy. Clear the override after recovery:

import Arbiter.RateLimit (RateLimitPolicy (..), 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

Selector DSL and policy type: Arbiter.RateLimit Haddocks.

Concurrency Limiting

A policy is a prefix and a default limit. It caps jobs in flight per key across every queue in the registry. HasConcurrency selects a policy and key for each job.

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

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

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

instance HasConcurrency SyncPayload where
  concurrencyFor = concurrencyBy syncPolicy tenantOf

Selectors: noConcurrency, concurrencyBy, globalConcurrency, chooseWhen, concurrencyByCase. An operator can override the limit from the API or admin UI. Clearing the override restores the declared default. Limit 0 admits nothing.

Concurrency Limit 1 and Group Keys

Both admit one in-flight job per key.

A job can use both.

[!IMPORTANT] A job holds a slot from claim until ack, retry, nack, or reclaim. A lease expiry does not release it.

The reaper prunes idle keys and rebuilds in-flight counts after a restart or failover.

External Limit Updates

Set the override from a handler when a vendor reports new capacity:

import Arbiter.Concurrency (ConcurrencyPolicy (..), setConcurrencyLimit)

import MyApp.Queue.Policies (syncPolicy)

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 syncPolicy {cpLimit = seats}
    Ok -> pure ()

clearConcurrencyLimit syncPolicy removes the override. In a transactional handler the override commits with the ack.

Selector DSL and policy type: Arbiter.Concurrency Haddocks.

Archiving Completed Jobs

An acked job is deleted. archiveFor keeps a copy in the queue's archive for that many seconds:

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

The archive entry holds the handler's result. The reaper deletes expired entries. The REST API and admin UI list, re-enqueue, and delete archived jobs.

A re-enqueued job keeps its payload and settings and has no parent. Retry a failed tree from the dead-letter queue.

Archive row and queries: Arbiter.Core.Job.Archive Haddocks.

Worker Configuration

WorkerConfig holds the handler, thread count, timings, and callbacks. Set fields before the pool starts.

poolConfigForWorkers sizes the database pool for a list of worker pools. Pass the same list to poolConfigForWorkers, runWorkerPools, and shutdownWorkerPools.

Multiple Queues

One pool per queue, named with namedWorkerPool:

let workers = [Worker.namedWorkerPool emailConfig, Worker.namedWorkerPool imageConfig]

When one pool exits, the others stop. runWorkerPools then throws the first failure.

ARBITER_ENABLED_QUEUES selects pools by name. Unset, every pool starts. An unknown name throws at startup.

ARBITER_ENABLED_QUEUES=email_queue,image_queue

Configuration Types

A manual or batched handler must ack, fail, or nack each job:

A nack consumes no attempt. The job is redelivered after its lease expires.

withDbTransaction around a callback commits the ack with the application writes. Inside an outer transaction the callback is a savepoint. onJobSuccess fires at the savepoint release, before the outer commit.

Timings

Timing details: Leases and Deadlines. Field reference: WorkerConfig Haddocks.

Batched Handlers

batchedWorkerConfig workers batchSize handler passes up to batchSize jobs per call. A grouped batch holds one group. An ungrouped batch holds ready jobs.

-- batchedWorkerConfig <workerCount> <batchSize> handler
config <- Worker.batchedWorkerConfig 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 commits on its own. withDbTransaction commits a callback with application writes:

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 fires after the callback's transaction or savepoint completes. Inside an outer transaction, this is before the outer commit. Put once-only database effects in the ack's transaction rather than in the success hook.

If an outer transaction rolls back

A settlement callback can return successfully and still have its database changes rolled back by a later failure:

Arb.withDbTransaction $ do
  Worker.ack cbs job   -- succeeds and releases its savepoint
  updateSomethingElse  -- if this throws, the outer transaction rolls back

In this case, let the exception escape the handler. Catching it to log and then rethrow is fine. Do not catch it and continue processing or settling those jobs. PostgreSQL restores the job, but the worker still considers it settled and no longer heartbeats it. Another worker can reclaim it after its lease expires.

This restriction concerns an outer rollback after a callback returned successfully, not every exception thrown by a callback. It also applies to transactions opened directly through the driver. Success hooks already fired at savepoint release are not undone by the rollback and may fire on redelivery. They are not proof that the outer transaction committed.

hoistBatchCallbacks calls the callbacks from another monad through a monad morphism. It must keep the worker's connection and schema so each callback joins the handler's transaction. arbiter-orville provides orvilleBatchedHandler, which builds that morphism from the worker's own env.

Callback signatures: BatchCallbacks Haddocks.

Observability Hooks

ObservabilityHooks holds one callback per lifecycle point. Start from defaultObservabilityHooks and set fields:

myHooks = Arb.defaultObservabilityHooks
  { Arb.onJobSuccess = \job startTime endTime ->
      liftIO $ recordHistogram "jobs.duration" (diffUTCTime endTime startTime)
  , Arb.onJobFailedAndMovedToDLQ = \job err ->
      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 }

Hooks run in the pool monad.

Hook Invocation

Every claimed job fires onJobClaimed, then one row of:

onJobHeartbeat fires on each lease extension of a running job.

Hook Composition

ObservabilityHooks is a Monoid. <> runs the left callback, then the right. The right callback runs even when the left one throws. withHooks combines a record with a configuration's hooks:

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

Hook Restrictions

A hook's return value is ignored. If a hook throws, the worker logs the exception at Warning and carries on.

onJobSuccess reports a completed ack operation, not a durable outer transaction commit. In a batched handler's outer transaction, it fires when the callback's savepoint is released. If the outer transaction rolls back, the job is redelivered and the hook may fire again. See Batched Handlers.

Reaper activity reports through onMaintenance on WorkerConfig.

Hooks in Another Monad

hoistObservabilityHooks runs hooks written in one monad from a worker in another, through a monad morphism such as lift:

let pool = config {Worker.observabilityHooks = Arb.hoistObservabilityHooks lift appHooks}

arbiter-orville provides orvilleHooks for this.

Callback arguments: ObservabilityHooks Haddocks.

Leases and Deadlines

config { Worker.visibilityTimeout = 60 }     -- the lease length (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 on the row's not_visible_until. One guard per worker pool renews due jobs in a shared statement at the jobHeartbeatInterval cadence. The guard continues fencing leases if an extension stalls.

An extension that remains stuck past its timeout and settlement grace releases the guard's renewal slot so later batches can be renewed. Its late response cannot update a replacement attempt's timers or clear its slot. The stuck driver call retains its checked-out connection until it exits, so recovery still requires another available pool connection. The guard abandons one extension at a time. If the replacement also stalls, renewal waits for the abandoned call to exit, so a wedged backend holds at most two of the pool's connections.

Early handler termination:

The lease fence uses a local deadline and requires no database response. It is always active.

A batch whose lease has already expired when it reaches the guard is rejected before its handler runs. Each of its jobs fires onJobClaimed, then onJobUnavailable. Keep worker and database clocks synchronized. An expired timestamp is not treated as evidence of clock skew.

The guard registers before onJobClaimed runs. Claim hooks receive heartbeat protection, and their runtime counts toward maxJobDuration. A blocked claim hook is interrupted by the duration deadline or by lease loss, just like a handler.

maxJobDuration

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

last_error records the exceeded duration.

[!IMPORTANT] Without maxJobDuration, a hung handler holds its job and renews its lease for the life of the process.

Timing Constraints

jobHeartbeatInterval must be less than visibilityTimeout. The pool refuses to start otherwise.

After the database goes away, a handler continues for at most one visibilityTimeout after its last renewal. It continues for at least the difference between the two settings. Failed extensions retry until the lease expires.

Timing fields: WorkerConfig Haddocks.

Graceful Shutdown

Install signal handlers after the worker configurations exist. Pass the same pool list to shutdownWorkerPools 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.shutdownWorkerPools 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 claiming and waits for in-flight jobs. runWorkerPools returns when they finish or after gracefulShutdownTimeout. A job still running at the timeout is left unfinalized and redelivered after its lease expires.

Backoff Strategies

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

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

The first failure is attempt 1. With the default EqualJitter, exponentialBackoff 2.0 waits 1 to 2 seconds before the first retry.

A nack skips the backoff. The job stays invisible for the rest of its lease. To change that, call setVisibilityTimeout before the nack.

Strategies and jitter modes: Arbiter.Worker.BackoffStrategy Haddocks.

Wakeups (LISTEN/NOTIFY)

LISTEN/NOTIFY wakes workers at once for new jobs, pause and resume, force-cancel, and manual cron runs. SimpleDb, HasqlDb, and custom MonadArbiter instances supply the listener.

Without a listener, workers poll every pollInterval. Control paths fall back to:

The listener opens with the first worker pool and holds one pool connection. A producer-only process opens none.

useDedicatedListener opens a separate listener connection:

import Arbiter.Hasql qualified as ArbH
import Pqi.Ffi qualified as Ffi

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

let connect = ArbH.toHasqlConnect Ffi.adapter connStr
env <- ArbH.useDedicatedListener connect =<< ArbH.createHasqlEnv (Proxy @AppRegistry) connect "arbiter"

disableListener switches to polling:

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 above to stdout. silentLogConfig writes nothing.

Handler and pool logs carry the job context.

LogConfig and destinations: Arbiter.Worker.Logger Haddocks.

Liveness Probes

The heartbeat loop touches livenessFile at each worker heartbeat. The default is arbiter-worker-<workerId> in the system temp directory. A probe checks the file's age:

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

grep -q . fails the probe when no fresh file exists. find alone exits 0. A normal shutdown removes the file.

The REST API adds GET health, a readiness check that queries the database, and GET health/live, a liveness check that does not. See REST API and Admin UI.

Pausing Work

A pause stops new claims. In-flight jobs run to completion.

pauseChildren skips in-flight jobs and jobs in a delay or backoff. resumeChildren leaves a finalizer suspended while its children are queued.

LISTEN/NOTIFY delivers a pause at once. Without a listener, each pool reads the pause state at its next worker heartbeat. Details: Wakeups.

The REST API and admin UI expose the same controls.

REST API and Admin UI

arbiter-servant and arbiter-servant-ui provide the REST API and admin dashboard, as standalone WAI applications or as routes in an existing Servant API.

import Arbiter.Servant qualified as Servant
import Arbiter.Simple (createSimpleEnv, runSimpleDb)

env <- createSimpleEnv (Proxy @AppRegistry) connStr "arbiter"
config <- Servant.createArbiterServer (runSimpleDb env)
Servant.runArbiterAPI 8080 config

The server runs over any backend. Pass runHasqlDb env for a hasql application. The SSE stream uses the env's listener.

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)

Server and route types: arbiter-servant Haddocks. UI route type: arbiter-servant-ui Haddocks.

POST jobs and POST jobs/batch enqueue from any language.

OpenAPI

arbiter-servant-openapi serves an OpenAPI 3 description of the API at /openapi.json:

import Arbiter.Servant (ArbiterAPI, arbiterServer, createArbiterServer)
import Arbiter.Servant.OpenApi (OpenApiAPI, openApiServer)
import Arbiter.Servant.UI (AdminUI, adminUIServer)

type MyApp = ArbiterAPI AppRegistry :<|> OpenApiAPI :<|> AdminUI

config <- createArbiterServer (runSimpleDb env)
run 8080 $
  serve (Proxy @MyApp) (arbiterServer config :<|> openApiServer @AppRegistry "1.0.0" :<|> adminUIServer)

[!NOTE] Mount OpenApiAPI before AdminUI. AdminUI is a Raw route that matches every path.

openApiSpec @AppRegistry "1.0.0" returns the same document as a Value. Module: arbiter-servant-openapi Haddocks.

Endpoints

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

Global endpoints under /api/:

Operating jobs

dlq/:id/retry and archive/:id/reenqueue take an optional replacement payload:

POST /api/queues/email_queue/dlq/12/retry
Content-Type: application/json

{"payload": {"tag": "SendWelcome", "contents": "alice@example.com"}}

jobs/:id/reschedule sets when a job becomes visible:

POST /api/queues/email_queue/jobs/41/reschedule
Content-Type: application/json

{"runAt": "2026-10-01T09:00:00Z"}

An in-flight, suspended, cancelled or exhausted job returns 409.

Consuming over HTTP

POST claim is the worker-pool claim: it spends admission tokens, increments the attempt count, records a claimant, and hides each job for the lease. A paused queue returns no leases.

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

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

claimSeq and claimedBy in each returned job identify the lease. Every finalization request carries them:

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

On a QueueWithResult queue, ack takes the result:

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

The result is stored as ackWith stores it. A body that does not match the result type returns 400. Omit result to store none. The route needs FromJSON and ToJSON on the result type.

A mismatched lease, a lease that a registered worker holds, a suspended job, or a force-cancelled job returns 409.

The server does not renew an HTTP lease. After it expires another consumer can claim the job.

[!IMPORTANT] These routes have no authentication. Protect them with WAI middleware, a Servant authentication combinator, or an authenticating proxy.

Maintenance

POST maintenance runs one pass over stale workers, exhausted and cancelled jobs, rate-limit buckets, concurrency counts, archive and DLQ retention, and group summaries. Each operation runs in one caller at a time.

The response lists the affected row count per completed operation and the names of failed operations. Skipped operations appear in neither:

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

Worker pools run the same pass in their reaper. A pass covers every queue the migrations created in the schema, including queues outside the caller's registry. The endpoint supports deployments without worker pools.

OpenTelemetry

Spans and W3C trace-context propagation are built in. arbiter-otel adds metrics, gauges, and OTel log records over OTLP:

import Arbiter.Otel qualified as Otel

main :: IO ()
main = do
  let workers = [namedWorkerPool emailCfg, namedWorkerPool imageCfg]
  poolCfg <- poolConfigForWorkers workers
  env <- createSimpleEnvWithConfig (Proxy @AppRegistry) connStr "arbiter" poolCfg

  runSimpleDb env $ Otel.runWorkerPools workers

Otel.runWorkerPools replaces runWorkerPools with the same arguments. It installs the SDK, instruments the pools, and starts the gauges. Call it once per process.

Standard OTEL_* variables configure the SDK. OTEL_SDK_DISABLED=true turns it off. Logs go to OTel and to the configured log destination, with the job trace, id, queue, and attempt.

To own the telemetry handle, open it with a bracket such as withTelemetry, or with withExternalTelemetry for providers your application owns. Pass it and the gauge loop's base log config to runWorkerPoolsWith. For another pool runner, apply instrumentPools to the pools and wrap the runner in withGauges.

Traces

An enqueue records the current span. A claim starts a process <queue> consumer span linked to it, across processes and for jobs a handler enqueues. A REST API enqueue joins the request trace under newOpenTelemetryWaiMiddleware (hs-opentelemetry-instrumentation-wai).

Both spans carry the payload kind as arbiter.kind.

Arbiter.Core.Trace annotates a job's span, opens child spans, and wraps an enqueue made outside a handler.

Metrics

arbiter-otel reports job activity, queue depth, admission policies, reaper activity, Arbiter table health, and PostgreSQL health. Arbiter.Otel.MetricNames lists each instrument with its type, unit, and attributes.

Admission metrics are keyed by policy. The admitted counter and the keys and limit gauges also carry policy_kind, rate_limit or concurrency. The in-flight, busiest-key, and tokens gauges carry policy only.

kind on arbiter.queue.depth_by_kind and the job counters is one of the payload's kindsFor labels, or absent.

PostgreSQL health outside the Arbiter role needs pg_read_all_stats. One replica scans per interval and the rest export that reading.

Queue and PostgreSQL gauges scan once per OTEL_METRIC_EXPORT_INTERVAL (default 60s).

Prometheus

Metrics leave over OTLP. Scrape an OTel collector. OTEL_METRICS_EXPORTER=prometheus turns metrics off.

Local Stack

arbiter-demo/run-local.sh runs the demo against Grafana's LGTM stack at http://localhost:8000, dashboard at /dash. The live demo runs the same stack. The dashboard and alert rules expect metrics over OTLP through a collector.

Backend Integration

MonadArbiter separates the core from the database library. Three adapters ship: arbiter-simple, arbiter-orville, and arbiter-hasql. Each shares connections with its database library.

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.

* The workers drain all 500k ready jobs before the 10-second trial ends. The real rate is higher.

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)

postgresql-simple and resource-pool. Handlers receive a Connection. Nested transactions are 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))

Environment and pool constructors: arbiter-simple Haddocks.

arbiter-orville (orville-postgresql)

orville-postgresql. Orville owns its connections and transactions. OrvilleDb runs over the application's own MonadOrville monad, so Arbiter queries use its connection and join its open transaction:

O.withTransaction $ do
  O.insertEntity ordersTable order
  ArbO.runOrvilleDb @AppRegistry (ArbO.OrvilleEnv {ArbO.schema = "arbiter", ArbO.listener = Nothing}) $
    Arb.insertJob (Arb.defaultJob (ProcessOrder orderId))

Single-job handlers run in the application monad, as JobRead payload -> AppM result. A handler that inserts jobs wraps those calls in runOrvilleDb to share its transaction.

Batched and manual handlers, and observability hooks, run in the worker's monad. Arbiter.Orville.Worker adapts ones written in AppM. orvilleBatchedHandler runs the callbacks in the worker's own OrvilleEnv, on the handler's connection and inside its transaction:

import Arbiter.Orville.Worker (orvilleBatchedHandler, orvilleHooks)

processEmail :: Arb.JobRead EmailPayload -> Worker.BatchCallbacks AppM EmailPayload () -> AppM ()

config <- Worker.manualWorkerConfig 5 (orvilleBatchedHandler processEmail)
let emailPool = config {Worker.observabilityHooks = orvilleHooks appHooks}

For LISTEN/NOTIFY, build a Listener with newLibPQListener from arbiter-libpq and the pool's connection string, and put it in the OrvilleEnv. Nothing polls.

Size the Orville pool with poolConfigForWorkers:

import Arbiter.LibPQ (newLibPQListener)
import Arbiter.Orville qualified as ArbO

main :: IO ()
main = do
  poolCfg <- Worker.poolConfigForWorkers workers
  orvillePool <- O.createConnectionPool (ArbO.toOrvilleConnectionOptions connStr poolCfg)
  listen <- newLibPQListener connStr
  let orvilleState = O.newOrvilleState O.defaultErrorDetailLevel orvillePool
      arbiterEnv = ArbO.OrvilleEnv {ArbO.schema = "arbiter", ArbO.listener = Just listen}
  runAppM orvilleState . ArbO.runOrvilleDb arbiterEnv $
    Worker.runWorkerPools workers

AppM needs MonadOrville and MonadUnliftIO. An application that wants AppM itself to be the MonadArbiter instance can write one from the primitives in Arbiter.Orville.MonadArbiter. Writing a Backend lists the members.

Connection options: arbiter-orville Haddocks.

arbiter-hasql (hasql)

hasql and resource-pool. Handlers receive a Hasql.Connection in the worker transaction.

import Pqi.Ffi qualified as Ffi

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

toHasqlConnect takes the transport adapter and the connection string. pqi-ffi wraps libpq. pqi-native is pure Haskell.

For hasql 1.x, turn off the hasql2 flag in cabal.project. Then toHasqlConnect takes only the connection string:

package arbiter-hasql
  flags: -hasql2

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")

With a custom pool, the environment reserves one connection for LISTEN/NOTIFY. Any adapter can open the pool:

import Data.Pool (defaultPoolConfig, newPool)
import Pqi.Ffi qualified as Ffi

let acquire = Hasql.acquire Ffi.adapter (ArbH.hasqlSettings connStr) >>= either (fail . show) pure
pool <- newPool (defaultPoolConfig acquire Hasql.release 60 10)
env <- ArbH.createHasqlEnvWithPool (Proxy @AppRegistry) pool "arbiter"

Environment and pool constructors: arbiter-hasql Haddocks.

Writing a Backend

A backend is a MonadArbiter instance. An arbiter-orville application can define one on its own monad in place of OrvilleDb.

Method signatures: MonadArbiter Haddocks.