When a background job follows a PostgreSQL write, an awkward question arises: what happens if the data has been committed but the job has not reached the queue? River closes this failure window by committing the business record and the job in one transaction.

The example uses Go, River, GORM, and Gin. The API creates a product, while a separate worker reserves inventory and completes publication. Inventory belongs to a warehouse service: an external system reached over HTTP that lives outside our database, takes 5 to 10 seconds to respond, and knows nothing about our transactions. Throughout the article, “warehouse service” refers to this system. We will use the example to examine atomic enqueueing, transaction duration, and repeated execution.

The complete example is available in the riverqueue-guide repository.

The article assumes familiarity with context.Context, database/sql, and ordinary PostgreSQL transactions. Running the example locally requires Docker and Docker Compose.

The problem River solves

Two independent writes

Consider a common HTTP flow:

  1. The API inserts a product into the products table.
  2. After the commit, it sends a publication job to a separate broker.
  3. A worker receives the job and reserves inventory in the warehouse service.

The process can fail between the first and second steps. The product already exists, but the job is lost, so publication never completes. Reversing the order is unsafe too: the broker may accept the job before the product transaction rolls back.

This is the classic problem of two independent writes, also known as a dual write.

Why enqueueing inside a transaction is not always safe

One tempting approach is to enqueue the job before COMMIT:

err := db.Transaction(func(tx *gorm.DB) error {
	product, err := createProduct(tx, input)
	if err != nil {
		return err
	}

	return queue.Enqueue(PublishProductArgs{ProductID: product.ID})
})

if err != nil {
	return err
}

The business operation and the job appear to share a transaction. But when queue uses a separate broker, that broker knows nothing about *gorm.DB or the PostgreSQL transaction. It may accept the job and hand it to a worker immediately after Enqueue, before the application reaches COMMIT.

The events can then occur in this order:

  1. The API inserts the product but has not committed the transaction yet.
  2. The broker accepts the job and starts a worker immediately.
  3. The worker runs a SELECT for the product through another connection.
  4. PostgreSQL does not expose the uncommitted row, so the worker receives not found.
  5. The API commits afterward, but the job may already have failed or been cancelled as impossible to complete.

PostgreSQL does not allow dirty reads. Under the default Read Committed isolation level, a query only sees data committed before that query began. Even when Read Uncommitted is requested, PostgreSQL treats it as Read Committed. A worker cannot peek at a row from an unfinished transaction, so the design must not depend on it doing so.

An automatic retry may hide the race: a later attempt starts after the commit and finds the product. Correctness should not depend on worker timing or on how the worker classifies not found. If the business transaction rolls back, the job remains in the external broker anyway.

River stores jobs in PostgreSQL, so the products row and the river_job row can be written through one *sql.Tx. A rollback leaves neither the product nor the job. After a successful commit, the job is already in the database and survives an API shutdown. River calls this transactional enqueueing.

Before the commit, other transactions cannot see the river_job row either. River therefore cannot claim the job before the related business data becomes visible.

POST /api/v1/products                    worker
        │                                  │
        │  one short transaction            │
        ├─ INSERT INTO products            │
        └─ INSERT INTO river_job ─────────►│
                                           │
                                  Reserve(product_id)
                                  5-10 seconds, no transaction
                                           │
                                  UPDATE products

Running and verifying the example

Docker Compose starts PostgreSQL, migrations, the API, the worker, and River UI:

docker compose up -d --build
curl http://localhost:8100/healthz

The health check should return:

{"status":"ok"}

Create a product:

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

The API returns 201 Created. Immediately after creation, in_stock and published_at are null: the slow work is already queued, but the HTTP request does not wait for it. A shortened response looks like this:

{
  "id": 1,
  "name": "Mechanical keyboard",
  "price_cents": 7990,
  "in_stock": null,
  "published_at": null
}

After 5 to 10 seconds, another request shows the reserved inventory and publication time:

curl -s http://localhost:8100/api/v1/products/1
docker compose logs worker | grep river

The queue is also available in River UI, where you can inspect the job arguments, state, and attempt history.

After a successful run, the job appears under Completed. Its details show the final state, arguments, and number of attempts.

Application architecture

Why not do everything in the HTTP handler

A background job is useful here because of transaction duration, not because Go cannot handle requests concurrently. Gin runs on net/http, and one slow request does not block the whole server.

The problem starts when an external call runs inside a business transaction. While the warehouse service takes 5 to 10 seconds to respond, the transaction holds a connection from the pool. The example permits ten open connections. Ten concurrent publications can occupy the entire pool, leaving even fast reads waiting for a connection.

The responsibilities are split as follows:

  • the HTTP handler quickly stores the product and the job;
  • the worker limits concurrent calls to the warehouse service;
  • the slow external call does not hold a transaction open;
  • River retries the job after a temporary failure.

Moving code into a worker does not make it reliable by itself. The worker still needs timeouts, must return errors to River, and must tolerate repeated execution. River covers these requirements in Writing reliable workers.

How to read the example

The repository contains several directories, but the main path is short:

HTTP handler
    │
    ▼
ProductUseCase.Create
    │
    ├── productRepo.Create
    └── PublishProductEnqueuer.PublishProduct

River worker
    │
    ▼
ProductUseCase.Publish
    │
    ├── warehouse.Reserve
    └── productRepo.MarkPublished

If this is your first time in the repository, read the files in this order:

  1. internal/domain/product/usecases/product/product_usecase.go: the complete business flow.
  2. internal/infrastructure/river/product/jobs/publish_product.go: the job argument contract.
  3. internal/infrastructure/river/product/enqueuer/publish_product.go: the bridge between the GORM transaction and River.
  4. internal/infrastructure/river/product/worker/publish_product.go: error classification.
  5. pkg/transaction/transaction.go: passing the current GORM transaction through context.
  6. internal/app/container.go and internal/app/worker.go: dependency wiring and process startup.

This order exposes the transaction and the job path first, before the HTTP DTOs, configuration, and Docker infrastructure.

Job arguments and Kind

Every River job type has an argument struct and a Kind() method:

type PublishProductArgs struct {
	ProductID int64 `json:"product_id"`
}

func (PublishProductArgs) Kind() string { return "publish_product" }

Only the product ID enters the queue, not the entire model. Arguments are serialized to JSON and may remain in river_job for a long time while the product changes. The worker therefore reloads current data from PostgreSQL. A small payload is also easier to store and version.

The value returned by Kind() is part of the persisted contract. Renaming the Go type is harmless, but changing the string "publish_product" is not: jobs with the old kind may already be waiting and need a deliberate migration path.

One PostgreSQL transaction for business data and River

The use case opens a GORM transaction, creates the product, and uses the same context to call the job enqueuer:

func (u *ProductUseCase) Create(
	ctx context.Context,
	input product.ProductCreateInput,
) (models.Product, error) {
	var created models.Product

	err := u.tm.InTransaction(ctx, func(ctx context.Context) error {
		item, err := u.productRepo.Create(ctx, input)
		if err != nil {
			return fmt.Errorf("create product: %w", err)
		}

		if err := u.jobEnqueuer.PublishProduct(ctx, item.ID); err != nil {
			return fmt.Errorf("enqueue publish job: %w", err)
		}

		created = item
		return nil
	})
	if err != nil {
		return models.Product{}, err
	}

	return created, nil
}

The use case does not import River. It only knows the local jobEnqueuer interface, leaving the queue implementation in the infrastructure layer.

The transaction manager stores *gorm.DB in a derived context. The repository retrieves it through ExtractDB, while the River adapter uses ExtractTx. The adapter then unwraps *sql.Tx and passes it to InsertTx:

func (e *PublishProductEnqueuer) PublishProduct(
	ctx context.Context,
	productID int64,
) error {
	gormTx, err := transaction.ExtractTx(ctx)
	if err != nil {
		return err
	}

	sqlTx, ok := gormTx.Statement.ConnPool.(*sql.Tx)
	if !ok {
		return errors.New("enqueuer: failed to extract sql transaction from gorm")
	}

	_, err = e.client.InsertTx(
		ctx,
		sqlTx,
		jobs.PublishProductArgs{ProductID: productID},
		nil,
	)
	return err
}

PostgreSQL provides the transaction. GORM remains an ORM and exposes the underlying *sql.Tx to the application. In this example, River and GORM use the same database/sql transaction. The River guide Using River with GORM documents this integration.

Only two final states are possible:

COMMIT   → product exists, job exists
ROLLBACK → no product, no job

Separate clients for the API and worker

The API creates an insert-only client with no registered workers:

riverEnqueueClient, err := river.NewClient(
	riverdatabasesql.New(sqlDB),
	&river.Config{},
)

It only inserts jobs. The API can be scaled independently without increasing queue concurrency.

A separate process registers the worker and sets its concurrency:

workers := river.NewWorkers()
river.AddWorker(
	workers,
	worker.NewPublishProductWorker(container.ProductUsecase),
)

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,
	},
)

MaxWorkers limits load on the warehouse service and PostgreSQL. SoftStopTimeout gives active jobs time to finish after SIGTERM; River then cancels their contexts. Every external call in the worker must therefore stop on ctx.Done() rather than continue in the background.

Running two processes requires a second composition root: configuration, the connection pool, migrations, and graceful shutdown are assembled twice. In return, queue concurrency no longer depends on the number of API replicas. If the worker client lived inside the API, five instances configured with ten workers each would make fifty concurrent calls to the warehouse service even though its capacity has no relation to HTTP traffic.

Worker reliability

Idempotency under repeated execution

River provides at-least-once execution: the same job may run more than once. A worker can update an external system and then fail before River records success.

In the example, Publish first checks PublishedAt, then calls the warehouse service and performs a conditional UPDATE:

func (u *ProductUseCase) Publish(ctx context.Context, productID int64) error {
	item, err := u.productRepo.Get(ctx, productID)
	if err != nil {
		return err
	}

	if item.PublishedAt != nil {
		return nil
	}

	inStock, err := u.warehouse.Reserve(ctx, productID)
	if err != nil {
		return err
	}

	_, err = u.productRepo.MarkPublished(ctx, productID, inStock)
	return err
}

The repository only updates a row that has not been published:

result := db.Model(&models.Product{}).
	Where("id = ? AND published_at IS NULL", id).
	Updates(map[string]any{
		"in_stock":     inStock,
		"published_at": time.Now(),
	})

The SQL condition protects the database state from a race, but it does not prevent a repeated call to the warehouse service. If the process fails after Reserve, the next run calls it again.

A production warehouse service should therefore accept an idempotency key, such as productID or a dedicated reservationID, and return the previous result on a retry. Unique jobs prevent duplicate enqueueing, but they do not change repeated-execution semantics.

Which errors to retry and which to cancel

River decides what happens next from the value returned by Work:

  • nil: the job completed;
  • an ordinary error: the failure is temporary, so River schedules another attempt;
  • river.JobCancel(err): the failure is permanent and retries cannot help;
  • river.JobSnooze(duration): postpone the job without counting the result as an error.

The worker remains a thin adapter:

func (w *PublishProductWorker) Work(
	ctx context.Context,
	job *river.Job[jobs.PublishProductArgs],
) error {
	err := w.productPublisher.Publish(ctx, job.Args.ProductID)
	if err == nil {
		return nil
	}

	if errors.Is(err, product.ErrProductNotFound) {
		return river.JobCancel(fmt.Errorf(
			"publish product %d: %w",
			job.Args.ProductID,
			err,
		))
	}

	return fmt.Errorf("publish product %d: %w", job.Args.ProductID, err)
}

Do not log an error and return nil: River will mark the work complete and the retry will be lost. Logs help diagnose failures, but the returned value controls the job lifecycle.

Migrations and diagnostics

River adds its own PostgreSQL tables, including river_job, river_queue, and river_migration. In the example, their migrations live next to the products migration and run through golang-migrate in the same step.

How to obtain the River migrations

River ships SQL migrations inside the riverdatabasesql driver module. The project extracts them with scripts/fetch_river_migrations.sh:

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

A local Go toolchain works too:

go mod download
./scripts/fetch_river_migrations.sh

The script creates matching *.up.sql and *.down.sql files in db/migrations. Read them and review the regular diff before committing, just as you would for any other schema change.

The script:

  1. It uses go list -m to locate riverdatabasesql in the Go module cache.
  2. It takes migrations from the exact River version pinned in go.mod.
  3. It copies the up and down files into the shared db/migrations directory.
  4. It gives every file a unique numeric version in the golang-migrate format.

Manual copying can drift from the library: the code may be upgraded while the SQL remains on an older version. The script records where the SQL came from and which version it belongs to, keeps the River schema beside the business schema, and lets both run with one command:

migrate -path db/migrations -database "$DATABASE_URL" up

The alternative is river migrate-up. That works, but the River schema and the application schema are then managed by different commands. This example uses one migration pipeline so the order remains visible in the repository and reproducible in local development, CI, and deployment.

The script is intended primarily for the initial import. When upgrading River, do not delete or rename migrations already applied in production. Keep their versions and add only new River migrations. Otherwise, golang-migrate will treat old SQL under a new number as a migration that has not run yet.

Observing the queue

River UI is enough for routine diagnostics. SQL queries against river_job help in CI and incident analysis:

SELECT state, count(*) AS jobs, max(attempt) AS max_attempt
FROM river_job
GROUP BY state
ORDER BY state;

Treat direct UPDATE statements against queue rows as an experiment, not an operating procedure. For retries, cancellation, and queue management, use River UI or the library’s public API. They enforce valid state transitions, while an ad hoc SQL script knows nothing about those rules.

Tests and limitations

What the tests cover

The test using rivertest.RequireInsertedTx finds the job through the same *sql.Tx that created the product:

job := rivertest.RequireInsertedTx[*riverdatabasesql.Driver](
	ctx,
	t,
	sqlTx,
	jobs.PublishProductArgs{},
	nil,
)

assert.Equal(t, created.ID, job.Args.ProductID)

Other tests exercise real Gin handlers, the GORM repository, and PostgreSQL. A local stub with no delay replaces the warehouse service so the suite finishes in milliseconds.

docker compose run --rm tests

A separate test should simulate an enqueue failure after a successful INSERT products and verify that GORM rolls back both writes. Cancelling the context before the transaction begins only tests an early return, not this path.

What remains before production

The example is intentionally small. A production version still needs decisions about:

  • handling a conditional UPDATE that affects no rows;
  • jobs that exhaust all attempts;
  • worker counts, timeouts, and pool size based on measurements;
  • closing database connections correctly during process shutdown;
  • validating the migration strategy and chosen River driver under load.

When to choose River

River fits when a job follows directly from a database change: send a notification after creating an order, rebuild an index after updating a document, or prepare a file after saving a report. In each case, the job must not be lost and must only appear after the write succeeds.

A separate queue, including one backed by Redis, remains reasonable when the platform already has one or the architecture requires an independent broker.

An atomic link to the SQL transaction is unnecessary when:

  • enqueueing is not accompanied by a business-data change;
  • a schedule starts the job, as with cleanup, cache warming, or a periodic report;
  • all required data was committed before the job was enqueued;
  • the arguments are self-contained and the worker does not need to read a newly created row;
  • there is no invariant that the database write and job must either both happen or both not happen.

Even without a shared transaction, the job still needs durable storage, retries, and idempotency. Those guarantees simply do not need to be atomic with a particular PostgreSQL row.

When a business record and a job must appear together or not at all, enforce that condition with one transaction.