WIP: Improve top-level documentation

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.

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

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

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.
   }

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