Skip to content
BranchesTags

Repository files navigation

River Guide

A minimal, self-contained example of using River — a Postgres-backed background job queue for Go — with GORM, Gin, and Docker Compose.

The scenario: a REST API creates a product, and a background job completes its publication by reserving stock at an external warehouse — an emulated slow service (internal/infrastructure/warehouse/client.go) that takes 5–10 seconds per reservation. The interesting part is how the job is enqueued: in the same database transaction as the product row.

POST /api/v1/products                  worker process
        │                                     │
   GORM transaction (milliseconds)            │
   ├── INSERT INTO products                   │
   └── INSERT INTO river_job ──────► river picks it up
        (atomic!)                             │
                                 1. warehouse Reserve()   ← 5–10 s, NO transaction held
                                 2. UPDATE products
                                    SET in_stock = ..., published_at = now()
                                    WHERE published_at IS NULL

If the transaction rolls back, the job never exists. If the process dies right after commit, the job is still in Postgres and will be worked. No Redis, no dual-write bugs.

If a job runs out of retries, a periodic job finds the stuck publications and enqueues one new job for each product.

Why async at all? Doing the 5–10 s reservation in the HTTP handler would hold a database/sql pool connection for the whole request; ten concurrent POSTs exhaust the pool (MaxOpenConns: 10) and even instant GETs queue behind them. With River the handler returns in ~20 ms, and the slow work runs in the worker — outside any transaction, bounded by the queue's MaxWorkers, retried automatically if the warehouse fails.

Services

Service Port Purpose
api localhost:8100 REST API (Gin + GORM), enqueues River jobs transactionally
worker – River worker process, works the default queue
river (web UI) localhost:8083 River UI: inspect, retry, cancel jobs
postgres localhost:5433 PostgreSQL 17: products + River's river_job
migrate – One-shot golang-migrate runner
tests / lint – Test suite and golangci-lint (run on demand)

Hot reload via air: edit any .go file and the service rebuilds.

Quick start

docker compose up -d --build

Wait ~30 seconds for the first build, then verify:

curl http://localhost:8100/healthz
# {"status":"ok"}

Try it with curl

1. Create a product — the response comes back in ~20 ms

curl -s -X POST http://localhost:8100/api/v1/products \
  -H "Content-Type: application/json" \
  -d '{"name":"Mechanical keyboard","description":"Tactile switches, hot-swap","price_cents":7990}'

Response (201 Created) — note in_stock and published_at are still null: the slow warehouse reservation is now a queued job the API doesn't wait for:

{"id":1,"name":"Mechanical keyboard","description":"Tactile switches, hot-swap","price_cents":7990,"in_stock":null,"published_at":null,"created_at":"2026-08-31T10:43:52.525446Z"}

2. Watch the worker do the slow work (5–10 seconds)

# right after the POST: the job is running, product not published yet
curl -s http://localhost:8100/api/v1/products/1

# the running job is visible in worker logs...
docker compose logs worker
2026/08/31 10:43:53 [river] job=1 kind=publish_product attempt=1 product_id=1: started

...and in the queue (state: running):

./scripts/river_jobs.sh list

3. After ~10 seconds: reservation recorded

curl -s http://localhost:8100/api/v1/products/1
{"id":1,"name":"Mechanical keyboard","description":"Tactile switches, hot-swap","price_cents":7990,"in_stock":11,"published_at":"2026-08-31T10:44:00.575761Z","created_at":"2026-08-31T10:43:52.525446Z"}
docker compose logs worker | grep river
2026/08/31 10:43:53 [river] job=1 kind=publish_product attempt=1 product_id=1: started
2026/08/31 10:44:00 [river] job=1 kind=publish_product attempt=1 product_id=1: completed

Compare the timings: the POST answered in ~20 ms while the work it started ran 5–10 seconds in the worker. The API's DB pool was occupied only for the brief enqueue transaction.

Periodic jobs: the stuck-publications sweeper

The publish job retries, but retries run out. River gives a job 25 attempts by default; if the warehouse is down for two hours, the job ends up discarded and the product keeps published_at = null forever. Nothing brings it back — that is what the sweeper is for.

The trigger for this work is not a user action but time, so there is no business row to tie a job to and no shared transaction to preserve. What matters instead is the shape of the schedule:

schedule (River scheduler, leader only)
    │  every STUCK_PUBLICATIONS_SCAN_INTERVAL
    ▼
dispatch_stuck_publications   ← periodic job: only finds work and enqueues it
    │  select a batch of ids + InsertManyTx, one transaction
    ▼
river_job × batch_limit       ← ordinary jobs: one per product
    │
    ▼
publish_product worker        ← retries, error isolation, timeouts, per product

A periodic job does not "run on a schedule": on every tick River calls a constructor and inserts whatever args it returns (Periodic jobs). Three properties of that scheduler shape the code:

  • Only the leader schedules. Worker replicas elect a leader through River's own tables, and only the leader inserts — ten replicas still produce one series of ticks. An insert-only client (the api process) runs no scheduler at all.
  • The scheduler keeps its state in memory. After a restart or a leader change it starts from scratch, and missed ticks are never made up. RunOnStart: true is the recommended hedge, and it is what makes the first tick visible right after docker compose up.
  • The constructor must not block. Its job is to return args, not to query the database.

The tempting shortcut — one periodic job that selects a batch and publishes it in a loop — puts a single attempt counter on a hundred products: a failure on the 71st loses the progress of the first 70, and the retry starts the batch from the beginning. Hence the split. The periodic job only dispatches; every product gets its own job, its own attempt counter and its own timeout.

The pieces

Three decisions in there are worth spelling out.

MaxAttempts: 1 on the dispatcher. The usual advice is not to hide problems behind a single attempt, but the dispatcher already has retries: the next tick. A failed tick goes straight to discarded (and to whatever ErrorHandler you wire up), and a minute later the scheduler inserts a fresh one. Keeping internal retries as well would mean two independent retry loops over the same batch. This only works because the dispatcher is idempotent and stateless between ticks — a periodic job with an asymmetric effect (sending mail, spending a quota) must not be configured this way.

UniqueOpts{ByPeriod: interval} on the dispatcher, UniqueOpts{ByArgs: true} on publish_product. ByPeriod means a tick is skipped if the previous dispatcher has not finished yet. ByArgs means a product whose publish job is already queued or running is not queued twice — routine now that a stuck product can land in a batch while its original job is still retrying.

The grace period lives in SQL. created_at <= NOW() - interval '10 minutes', not just published_at IS NULL. Ten minutes is roughly how long the original job needs to exhaust its retries; without the condition the dispatcher would compete with work that is still in flight. Unique jobs would smooth over most of those collisions anyway, but a condition in the query is the honest version.

Configuration

Variable Default Purpose
STUCK_PUBLICATIONS_SCAN_INTERVAL 60 (seconds) How often the dispatcher runs. River refuses intervals under a second and recommends a minute or more.
STUCK_PUBLICATIONS_BATCH_LIMIT 100 How many products one tick may enqueue. Caps the burst on a warehouse that is only just recovering.

Both live in .env: how aggressively to sweep is an operational parameter, tuned against metrics (how much backlog piles up, how much load the warehouse tolerates), not a constant in the code.

Watch it work

Start the stack first — make stuck talks to a running PostgreSQL:

docker compose up -d --build

Read the worker log through a filter. In dev the worker runs under air, which prints dozens of watching ... lines on every rebuild, and the River lines drown in them:

docker compose logs -f worker | grep river

The first tick fires immediately, without waiting for the interval, thanks to RunOnStart. Nothing is stuck yet, so both counters are zero:

2026/09/06 11:22:50 [river] job=49 kind=dispatch_stuck_publications attempt=1: dispatched enqueued=0 skipped_duplicates=0

Now create something for the sweeper to find. make stuck inserts a product older than the grace period with no publish job behind it — exactly the state a discarded job leaves behind:

make stuck

Within one scan interval the dispatcher picks up the batch and enqueues one job per product. This is the two-level split in the log: one schedule line, then independent jobs each with their own job id and attempt.

2026/09/06 11:23:10 [river] job=50 kind=dispatch_stuck_publications attempt=1: dispatched enqueued=3 skipped_duplicates=0
2026/09/06 11:23:11 [river] job=53 kind=publish_product attempt=1 product_id=11: started
2026/09/06 11:23:11 [river] job=51 kind=publish_product attempt=1 product_id=9: started
2026/09/06 11:23:11 [river] job=52 kind=publish_product attempt=1 product_id=10: started

To see deduplication, make the publication outlast the interval: raise WAREHOUSE_MIN_DELAY / WAREHOUSE_MAX_DELAY and lower STUCK_PUBLICATIONS_SCAN_INTERVAL, then restart the worker. The next tick finds the same products still unpublished and tries to enqueue them again:

2026/09/06 11:23:30 [river] job=54 kind=dispatch_stuck_publications attempt=1: dispatched enqueued=0 skipped_duplicates=3
2026/09/06 11:23:50 [river] job=58 kind=dispatch_stuck_publications attempt=1: dispatched enqueued=0 skipped_duplicates=3

enqueued=0 with skipped_duplicates=3 is UniqueOpts{ByArgs: true} doing its job: the schedule keeps ticking and the queue does not grow. Without it every tick would add three more jobs for the same three products.

The same story from the queue's side:

./scripts/river_jobs.sh list

Or open the River UI: the dispatcher's rows are the history of ticks, the publish_product rows are the individual attempts.

In production a steadily growing skipped_duplicates is the signal to watch: the original publish jobs are not finishing in time and the grace period is too short.

Two things that will bite you

Missed ticks are never made up, and that is not a River bug. The scheduler is started by the leader, keeps its state in memory, and begins from zero after a restart. Deploy at 03:00 with an hourly interval and the 03:30 tick simply does not happen. For sweeping that is fine — the cleanup just shifts. If a single missed run is an incident, OSS periodic jobs are the wrong tool: look at River Pro's durable schedules or an external scheduler that calls Insert.

ByArgs uniqueness includes completed jobs. River's default ByState set covers completed, so a product that was published successfully cannot be enqueued again until the job cleaner removes the old row (24 hours by default). It never blocks the sweeper — a published product is not stuck and is never selected — but it will surprise you if you reset published_at by hand to replay the flow. Delete the old river_job row too, or use a fresh product. And unique jobs are not idempotency: they keep the queue clean, they do not guarantee a single execution. The conditional UPDATE ... WHERE published_at IS NULL stays where it is.

Guide history

The repository is built commit by commit, each one matching an article:

Commit Date Article
create jobs via river, part 1 2026-08-31 21:18 +0500 River в Go: фоновые задачи в одной транзакции с PostgreSQL
periodic jobs, part 2 2026-09-05 18:20 +0500 River: периодические задачи без cron 
git log --oneline

River migrations

River needs its own schema in Postgres (river_job, river_queue, river_leader, river_migration). This guide keeps ALL schema — River tables and business tables — in db/migrations, applied by one golang-migrate step (migrate compose service).

The River migration files were fetched from the riverdatabasesql driver of the exact River version pinned in go.mod with:

docker compose run --rm --entrypoint "" migrate ./scripts/fetch_river_migrations.sh

The script (scripts/fetch_river_migrations.sh) reads the SQL straight from the Go module cache — so the schema always matches your River version — and renames the files to golang-migrate style. Use it again whenever you upgrade River and need the new migrations. (The river CLI installed in the image can also apply its schema directly with river migrate-up --database-url "$DATABASE_URL", but then River schema and business schema would live in two pipelines.)

Worker configuration notes

The worker client is configured in internal/app/worker.go:

client, err := river.NewClient(
    riverdatabasesql.New(container.DB),
    &river.Config{
        Workers: workers,
        Queues: map[string]river.QueueConfig{
            river.QueueDefault: {MaxWorkers: 10},
        },
        SoftStopTimeout: 30 * time.Second,
    },
)
  • SoftStopTimeout — how long in-flight jobs may keep running after SIGTERM before River cancels their contexts. On shutdown the worker first stops fetching new jobs, then waits up to this timeout for running jobs to finish; anything still running past the deadline gets its context cancelled (and will be retried later, since it never reported success). Together with signal.NotifyContext + <-client.Stopped() this is the whole graceful-shutdown recipe — see Graceful shutdown. Tune it above your longest expected job.
  • MaxWorkers — per-queue concurrency limit. It caps how many reservations hit the warehouse (and how many DB connections the worker may need) at once; size it against the Postgres connection pool (Multiple queues).
  • Job timeout — River cancels jobs after 1 minute by default (Writing reliable workers). The 5–10 s reservation fits comfortably; for longer jobs override Timeout() on the worker.

The slow warehouse call is deliberately made outside any transaction: a 5–10 s transaction would hold a pool connection the whole time. The DB is touched only for the short final UPDATE — the pattern recommended in Writing reliable workers.

Inspecting River jobs

Everything River knows lives in Postgres tables (river_job, river_queue, river_migration). The scripts/river_jobs.sh helper wraps the most useful queries:

./scripts/river_jobs.sh stats                 # counts by state
./scripts/river_jobs.sh list                  # last 20 jobs
./scripts/river_jobs.sh kind publish_product  # by kind
./scripts/river_jobs.sh show 1                # full row: args, errors, metadata
./scripts/river_jobs.sh failed                # retryable + discarded
./scripts/river_jobs.sh errors                # last error of each job
./scripts/river_jobs.sh retry 1               # make a retryable job available now

Or open the River UI to see the same jobs graphically.

Tests

Tests run the real stack — Gin router → usecase → GORM repo → PostgreSQL inside a rollback-per-test transaction — plus rivertest assertions that a job was really inserted into the same transaction:

docker compose run --rm tests

The sweeper is covered the same way: the select is a table test (a fresh product is not picked up, an 11-minute-old one is, a published one is not, and an overflowing batch returns the oldest first), and the dispatcher asserts the batch of inserted jobs with rivertest.RequireManyInsertedTx — including the second tick, where ByArgs turns the same products into skipped_duplicates.

Lint

docker compose run --rm lint

Project layout

cmd/
  api/            REST API entry point
  worker/         River worker entry point
config/           env-based configuration
internal/
  app/            container wiring, api/worker bootstrap
  domain/product/ models, repo, usecase (framework-free business logic)
  infrastructure/
    db/postgres/    GORM connection
    river/product/  jobs (args), enqueuer (InsertTx/InsertManyTx), workers
  transport/http/ gin handlers + DTOs
  test_helpers/   rollback-per-test transactions
db/
  migrations/     River schema (001–007) + products
  init/           test database bootstrap
pkg/
  transaction/    GORM transaction propagation via context
scripts/
  river_jobs.sh             psql-based job inspection
  fetch_river_migrations.sh fetch River's SQL migrations into db/migrations

Key files, in reading order:

Further reading

For the full walkthrough in Russian — architecture decisions, River vs asynq comparison, and why each piece is built the way it is — read ARTICLE.md.

About

No description, website, or topics provided.

Resources

Readme
Activity

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

Packages

Contributors

Languages