diff --git a/consumers/README.md b/consumers/README.md index 7e0f62e..b994584 100644 --- a/consumers/README.md +++ b/consumers/README.md @@ -3,11 +3,157 @@ [![Haskell-CI](https://github.com/scrive/consumers/actions/workflows/haskell-ci.yml/badge.svg?branch=master)](https://github.com/scrive/consumers/actions/workflows/haskell-ci.yml) [![Hackage version](https://img.shields.io/hackage/v/consumers.svg?label=Hackage)](https://hackage.haskell.org/package/consumers) -Library for setting up concurrent consumers of data stored inside -PostgreSQL database in a simple, declarative manner. +A PostgreSQL-backed job queue for Haskell. Jobs are rows in tables you own, so +you can enqueue them in the same transaction as the business write that +triggered them: no separate broker, no dual-write problem. Dispatch is driven +by `LISTEN`/`NOTIFY` for low latency and `FOR UPDATE SKIP LOCKED` for +contention-free reservation. -See the `examples/` directory for a usage example. +## Features -If you want to add metrics, see the -[`consumers-metrics-prometheus`](https://hackage.haskell.org/package/consumers-metrics-prometheus) -package to seamlessly instrument your consumer. +- **Postgres is the queue.** Jobs live in your own table; enqueue is just an + `INSERT` in the same transaction as the rest of your write. No Redis, no + Kafka, no broker to operate. +- **Multiple independent queues.** Each `ConsumerConfig` points at its own jobs + table paired with its own registry table (conventionally `foo_jobs` and + `foo_consumers`). The registry schema also permits one table to track + workers for several queues, distinguished by the `name` column. +- **Low-latency dispatch.** Optional `LISTEN`/`NOTIFY` wakes the consumer the + instant a job is committed; a configurable polling interval is the fallback + (and handles delayed/retried jobs). +- **Non-blocking reservation.** Jobs are claimed with + `SELECT … FOR UPDATE SKIP LOCKED`, so workers never block each other. +- **Scheduled jobs.** Every job has a `run_at` timestamp. One-shot delays, + retries, and recurring jobs are all just different values of `run_at`; + recurrence is implemented by having `ccProcessJob` return `RerunAfter` or + `RerunAt`. +- **At-least-once semantics with a flexible retry hook.** `ccOnException` + receives the exception and the job and returns the next `Action` + (`MarkProcessed`, `RerunAfter`, `RerunAt`, or `Remove`), so you can + implement any backoff policy you like. +- **Dead-consumer reclamation.** Each consumer heartbeats its + `last_activity` every 30 seconds; any consumer idle for more than 60 seconds + is presumed dead and its reserved jobs are released back to the queue. +- **Graceful shutdown.** `runConsumer` returns a finalizer that waits for + in-flight jobs to finish and releases reservations. +- **Bounded concurrency.** `ccMaxRunningJobs` caps how many jobs a single + consumer process runs in parallel. +- **Structured logging** via [`log-base`](https://hackage.haskell.org/package/log-base); + per-job context is attached through `ccJobLogData`. +- **Optional Prometheus metrics** via the sibling + [`consumers-metrics-prometheus`](https://hackage.haskell.org/package/consumers-metrics-prometheus) + package. + +## Quick start + +A complete, runnable example lives in +[`example/Example.hs`](example/Example.hs). The shape of an integration is: + +1. Create a jobs table with the required columns (`id`, `run_at`, `finished_at`, + `reserved_by`, `attempts`) plus whatever payload columns you need, and a + paired registry table (`id`, `name`, `last_activity`). See the Haddock on + `ccJobsTable` and `ccConsumersTable` for the exact contract. +2. Enqueue a job by `INSERT`ing a row, typically in the same transaction as + the write that caused it. +3. Build a `ConsumerConfig` describing the tables, how to deserialize a job, + what to do with it (`ccProcessJob`), and what to do on failure + (`ccOnException`). +4. Call `runConsumer cfg connSource` to start the worker. It returns an + `m (m ())`: the outer action starts the daemons, the inner action waits for + in-flight jobs at shutdown. Wrap the pair with `finalize` to tie shutdown to + your main loop. + +## Architecture + +Each call to `runConsumer` spawns three daemon threads inside your process: + +- **Listener.** Waits on the `LISTEN`/`NOTIFY` channel (if configured) and/or + a `ccNotificationTimeout` timer, and pokes the Dispatcher whenever it's time + to check for due jobs. +- **Dispatcher.** Reserves up to `ccMaxRunningJobs - inFlight` due jobs in a + single `SELECT … FOR UPDATE SKIP LOCKED` (setting `reserved_by` and bumping + `attempts`), then forks one worker thread per reserved job. Reserving a + whole batch in one round-trip amortizes the query cost across all the jobs + in it. Each worker runs `ccProcessJob` in its own DB transaction; the + results are folded back into a batched `UPDATE`. +- **Monitor.** Updates this consumer's `last_activity` every 30 seconds and + scans the registry table for peers that have gone silent for more than + 60 seconds. Any jobs reserved by such a peer are released (with + `ccOnException` applied), and the dead consumer row is removed. + +``` + ┌──────────────────── Consumer process ────────────────────┐ + │ │ + NOTIFY ─────►│ Listener ──poke──► Dispatcher ──fork──► Worker pool │ + │ │ │ │ + │ ┌──heartbeat──┐ │ │ │ + │ │ ▼ ▼ ▼ │ + │ Monitor SELECT … FOR UPDATE ccProcessJob │ + │ │ SKIP LOCKED; UPDATE │ │ + └──────┼──────────────┼───────────────────────────┼────────┘ + │ │ │ + ▼ ▼ ▼ + ┌───────────────────────────────────────────────────────┐ + │ PostgreSQL │ + │ consumers table ◄────► jobs table │ + └───────────────────────────────────────────────────────┘ +``` + +## Job lifecycle + +A job's state is encoded implicitly in four columns: +`run_at`, `reserved_by`, `finished_at`, and `attempts`. There is no explicit +status enum; the combination of those columns is the state. + +| State | `run_at` | `reserved_by` | `finished_at` | +|----------------|-----------------|---------------|---------------| +| Queued | `> NOW()` | NULL | NULL | +| Ready | `≤ NOW()` | NULL | NULL | +| Reserved | `≤ NOW()` | some consumer | NULL | +| Completed | NULL | NULL | `NOT NULL` | +| Rescheduled | `> NOW()` | NULL | NULL | +| Stuck | `≤ NOW()` | dead consumer | NULL | +| Removed | (row deleted) | + +``` + INSERT + │ + ▼ + ┌──────────┐ run_at ≤ NOW() ┌──────────┐ + │ Queued │ ──────────────────► │ Ready │ ◄─┐ + └──────────┘ └────┬─────┘ │ + │ │ + Dispatcher reserves │ + reserved_by := me │ + attempts += 1 │ + │ │ + ▼ │ + ┌──────────┐ │ + │ Reserved │ │ + └────┬─────┘ │ + │ │ + ccProcessJob │ + │ │ + ┌────────────────────────────┼────────────────────────────┐ + │ │ │ + Ok MarkProcessed Ok/Failed RerunAfter Ok/Failed Remove + Ok Remove Ok/Failed RerunAt (or Ok MarkProcessed) + │ │ │ + ▼ ▼ ▼ + ┌───────────┐ ┌─────────────┐ ┌──────────┐ + │ Completed │ │ Rescheduled │ │ Removed │ + └───────────┘ └─────────────┘ └──────────┘ +``` + +If the consumer dies mid-processing the row sits in the Stuck state until the +Monitor on another consumer notices and reclaims it (`ccOnException` is +applied, `reserved_by` is cleared, and the job returns to Ready). + +## Observability + +`ccJobLogData` attaches a list of structured fields to every log line emitted +while a job is processed; set it to include the job ID and any other +correlation data. For Prometheus metrics, drop in +[`consumers-metrics-prometheus`](https://hackage.haskell.org/package/consumers-metrics-prometheus), +which wraps `runConsumer` and exposes histograms and gauges for running, +overdue, and processed jobs without any changes to your consumer code. diff --git a/consumers/src/Database/PostgreSQL/Consumers.hs b/consumers/src/Database/PostgreSQL/Consumers.hs index 0899905..f8d9ec2 100644 --- a/consumers/src/Database/PostgreSQL/Consumers.hs +++ b/consumers/src/Database/PostgreSQL/Consumers.hs @@ -1,3 +1,27 @@ +-- | A PostgreSQL-backed job queue. Start here. +-- +-- A consumer is a worker process that pulls rows out of a jobs table you own, +-- runs your handler on each one, and writes the result back as either a +-- completion, a reschedule, or a delete. Enqueueing a job is just an @INSERT@ +-- on the same table, so it composes with the rest of your transactions. +-- +-- The two things you actually need are: +-- +-- * 'ConsumerConfig' (from "Database.PostgreSQL.Consumers.Config"): describes +-- your jobs table, how to deserialize a job, and what to do with one. +-- * 'runConsumer': starts the consumer's daemon threads and returns a +-- finalizer you run at shutdown (typically via 'finalize'). +-- +-- Re-exported submodules: +-- +-- * "Database.PostgreSQL.Consumers.Config": 'ConsumerConfig', 'Action', +-- 'Result'. +-- * "Database.PostgreSQL.Consumers.Utils": supporting machinery, including +-- the 'finalize' bracket and the 'StopExecution' / 'ThrownFrom' exceptions +-- used by the consumer's internal threads. +-- +-- See the package README for an architectural overview and a job-lifecycle +-- diagram. module Database.PostgreSQL.Consumers ( runConsumer , runConsumerWithIdleSignal diff --git a/consumers/src/Database/PostgreSQL/Consumers/Components.hs b/consumers/src/Database/PostgreSQL/Consumers/Components.hs index bb7d957..c98800a 100644 --- a/consumers/src/Database/PostgreSQL/Consumers/Components.hs +++ b/consumers/src/Database/PostgreSQL/Consumers/Components.hs @@ -1,3 +1,25 @@ +-- | Consumer runtime: the three daemon threads that drive a running consumer. +-- +-- Each call to 'runConsumer' forks three long-running threads inside the +-- caller's process: +-- +-- * __Listener__: waits on the configured @LISTEN@/@NOTIFY@ channel (if any) +-- and on a 'ccNotificationTimeout' timer, and signals the dispatcher +-- whenever it's time to look for due jobs. +-- * __Monitor__: every 30 seconds, updates this consumer's @last_activity@ +-- in the consumers table as a heartbeat, then scans for peer consumers +-- whose heartbeat is more than 60 seconds stale. Any jobs reserved by such +-- a peer are released (with 'ccOnException' applied so retry policy still +-- runs) and the dead consumer row is deleted. +-- * __Dispatcher__: reserves due jobs with +-- @SELECT … FOR UPDATE SKIP LOCKED@, setting @reserved_by@ and incrementing +-- @attempts@, then forks one worker per reserved job up to +-- 'ccMaxRunningJobs'. Each worker runs 'ccProcessJob' in its own +-- transaction; the results are folded back into a single batched @UPDATE@ +-- per dispatch cycle. +-- +-- The @spawn*@ functions are exposed for testing and instrumentation; normal +-- users should call 'runConsumer'. module Database.PostgreSQL.Consumers.Components ( runConsumer , runConsumerWithIdleSignal diff --git a/consumers/src/Database/PostgreSQL/Consumers/Config.hs b/consumers/src/Database/PostgreSQL/Consumers/Config.hs index fde9636..944544d 100644 --- a/consumers/src/Database/PostgreSQL/Consumers/Config.hs +++ b/consumers/src/Database/PostgreSQL/Consumers/Config.hs @@ -1,3 +1,17 @@ +-- | Static configuration of a consumer. +-- +-- 'ConsumerConfig' is the contract between your application and the consumer +-- runtime: it names the jobs table and the consumer-registry table, says how +-- to deserialize a job, and provides the handler ('ccProcessJob') plus its +-- exception fallback ('ccOnException'). Each invocation of 'ccProcessJob' is +-- expected to run in its own database transaction and to be idempotent: a +-- crashed consumer's job will be reclaimed by the monitor on another consumer +-- and retried, so a partial side effect must be safe to re-apply. +-- +-- The handler returns a 'Result' wrapping an 'Action'. The 'Ok' / 'Failed' +-- distinction is purely a signal for logging and metrics; both can carry any +-- 'Action'. If 'ccProcessJob' throws, 'ccOnException' is called with the +-- exception and the job and returns the 'Action' directly. module Database.PostgreSQL.Consumers.Config ( Action (..) , Result (..) @@ -15,13 +29,23 @@ import Database.PostgreSQL.PQTypes.SQL.Raw -- | Action to take after a job was processed. data Action - = MarkProcessed - | RerunAfter Interval - | RerunAt UTCTime - | Remove + = -- | Clear @run_at@ and stamp @finished_at@ with the current time. The + -- row is kept for auditing and will never be processed again. + MarkProcessed + | -- | Set @run_at@ to @NOW() + interval@. The job becomes eligible again + -- after the given delay. + RerunAfter Interval + | -- | Set @run_at@ to the given absolute time. + RerunAt UTCTime + | -- | Delete the row from the jobs table. + Remove deriving (Eq, Ord, Show) --- | Result of processing a job. +-- | Result of processing a job. 'Ok' and 'Failed' both carry an 'Action' and +-- mutate the row in the same way; the distinction is a signal for logging and +-- metrics. Typically you'd use @'Failed' ('RerunAfter' n)@ to indicate that +-- the job was retried due to a recoverable error rather than completing +-- normally. data Result = Ok Action | Failed Action deriving (Eq, Ord, Show) @@ -56,6 +80,9 @@ data ConsumerConfig m idx job = forall row. FromRow row => ConsumerConfig -- * __attempts__ - represents number of job processing attempts made so -- far. Needs to be not nullable, of type INTEGER. Initial value of a fresh -- job should be 0, therefore it makes sense to make the column default to 0. + -- The counter is incremented when the dispatcher reserves the row, so the + -- value visible inside 'ccProcessJob' and 'ccOnException' already includes + -- the currently-running attempt (i.e. it is 1 on the first run). , ccConsumersTable :: !(RawSQL ()) -- ^ Name of a database table where registered consumers are stored. The table -- itself needs to have the following columns: @@ -68,7 +95,8 @@ data ConsumerConfig m idx job = forall row. FromRow row => ConsumerConfig -- with one table. Set to 'ccJobsTable'. -- -- * __last_activity__ - represents the last registered activity of the - -- consumer. It's updated periodically by all currently running consumers + -- consumer. Needs to be not nullable, of a type comparable with @now()@ + -- (TIMESTAMPTZ is recommended). It's updated periodically by all currently running consumers -- every 30 seconds to prove that they are indeed running. They also check for -- the registered consumers that didn't update their status for a minute. If -- any such consumers are found, they are presumed to be not working and all @@ -92,6 +120,10 @@ data ConsumerConfig m idx job = forall row. FromRow row => ConsumerConfig -- consumer will check for pending jobs either when notification is received -- or no notification is received for 'ccNotificationTimeout' microseconds -- since the last check. + -- + -- @NOTIFY@ only fires when the transaction that issued it commits, so + -- enqueueing a job and notifying in the same transaction will never wake a + -- consumer for a row it cannot yet see. , ccNotificationTimeout :: !Int -- ^ Timeout of checking for any pending jobs (@'run_at <= NOW()'@), in -- microseconds. The consumer checks the database for any pending jobs after @@ -105,15 +137,22 @@ data ConsumerConfig m idx job = forall row. FromRow row => ConsumerConfig -- retried, you can set it to -1, then listening will never timeout. Otherwise -- it needs to be a positive number. , ccMaxRunningJobs :: !Int - -- ^ Maximum amount of jobs that can be processed in parallel. + -- ^ Maximum amount of jobs that can be processed in parallel by this + -- consumer process. To scale beyond a single process, run multiple + -- consumers against the same jobs table: reservation uses + -- @SELECT … FOR UPDATE SKIP LOCKED@, so peers never block each other and + -- each job is handed to exactly one consumer. , ccProcessJob :: !(job -> m Result) -- ^ Function that processes a job. It's recommended to process each job in a -- separate DB transaction, otherwise you'll have to remember to commit your -- changes to the database manually. , ccOnException :: !(SomeException -> job -> m Action) - -- ^ Action taken if a job processing function throws an exception. For - -- robustness it's best to ensure that it doesn't throw. If it does, the - -- exception will be logged and the job in question postponed by a day. + -- ^ Action taken if a job processing function throws an exception. No + -- backoff schedule is imposed: inspect the job's @attempts@ column to + -- compute whatever retry policy you want (linear, exponential, capped, + -- give-up-after-N, etc.). For robustness it's best to ensure that this + -- handler itself doesn't throw. If it does, the exception will be logged + -- and the job in question postponed by a day. , ccJobLogData :: !(job -> [A.Pair]) -- ^ Data to attach to each log message while processing a job. } diff --git a/consumers/src/Database/PostgreSQL/Consumers/Consumer.hs b/consumers/src/Database/PostgreSQL/Consumers/Consumer.hs index a836db9..b608ca5 100644 --- a/consumers/src/Database/PostgreSQL/Consumers/Consumer.hs +++ b/consumers/src/Database/PostgreSQL/Consumers/Consumer.hs @@ -1,3 +1,15 @@ +-- | Consumer-registry bookkeeping. +-- +-- Every running consumer owns a row in the consumers table identified by a +-- 'ConsumerID'. That row serves two purposes: it is the value written into a +-- job's @reserved_by@ column when the consumer claims the job, and it carries +-- a @last_activity@ timestamp that the monitor thread updates as a heartbeat. +-- If a consumer's heartbeat goes stale (more than 60 seconds old) another +-- consumer's monitor will reclaim its reserved jobs and delete the row. +-- +-- 'runConsumer' takes care of calling 'registerConsumer' at startup and +-- 'unregisterConsumer' at shutdown; the functions are exposed for callers +-- that want to drive the lifecycle manually. module Database.PostgreSQL.Consumers.Consumer ( ConsumerID , registerConsumer @@ -46,7 +58,9 @@ registerConsumer ConsumerConfig {..} cs = runDBT cs defaultTransactionSettings $ ] fetchOne runIdentity --- | Unregister consumer with a given ID. +-- | Unregister a consumer. Releases any jobs still reserved by it (so they +-- become eligible for processing again) and removes the consumer row from +-- the registry. unregisterConsumer :: (MonadBase IO m, MonadMask m) => ConsumerConfig n idx job diff --git a/consumers/src/Database/PostgreSQL/Consumers/Utils.hs b/consumers/src/Database/PostgreSQL/Consumers/Utils.hs index bd7a830..c8ae1a1 100644 --- a/consumers/src/Database/PostgreSQL/Consumers/Utils.hs +++ b/consumers/src/Database/PostgreSQL/Consumers/Utils.hs @@ -1,3 +1,16 @@ +-- | Supporting machinery used by the consumer runtime. +-- +-- The pieces here are useful to consumer callers too: +-- +-- * 'finalize': bracket pattern that pairs an action returning a finalizer +-- (such as 'Database.PostgreSQL.Consumers.runConsumer') with a body, and +-- runs the finalizer when the body completes or throws. +-- * 'StopExecution' / 'ThrownFrom': the async-exception protocol used by the +-- consumer's daemon threads. 'StopExecution' is a graceful stop signal; +-- any other exception in a child thread is wrapped in 'ThrownFrom' and +-- re-thrown in the parent. +-- * 'forkP' / 'gforkP': fork variants that propagate child exceptions to +-- the parent thread using the protocol above. module Database.PostgreSQL.Consumers.Utils ( finalize , ThrownFrom (..)