Begin SQLite read-then-write transactions immediate

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- The SQLite driver now begins the job scheduler's transaction, and its own transaction in `JobSchedule`, with `BEGIN IMMEDIATE`. Under WAL, a deferred transaction that reads before writing fails at once with `database is locked (SQLITE_BUSY)` when another connection commits in between, without waiting for `busy_timeout`, so job scheduling could fail on a pool of more than one connection. An immediate transaction takes the write lock up front and waits for it up to `busy_timeout` instead. River's other transactions write first and stay deferred. A pool of one connection remains the recommended configuration. [PR #1597](https://github.com/riverqueue/river/pull/1597).
+
 ## [0.51.2] - 2026-11-09
 
 ### Fixed

cmd/river/rivercli/command.go

@@ -195,7 +195,9 @@ func openSQLitePool(protocol, urlWithoutProtocol string) (*sql.DB, error) {
 		return nil, fmt.Errorf("error connecting to SQLite database: %w", err)
 	}
 
-	// See notes on this in `riversharedtest.DBPoolSQLite`.
+	// A single connection, because a database URL from the command line may
+	// not set `busy_timeout`, which concurrent connections need to wait for
+	// SQLite's write lock instead of failing with SQLITE_BUSY.
 	dbPool.SetMaxOpenConns(1)
 
 	return dbPool, nil

internal/maintenance/job_scheduler.go

@@ -171,7 +171,8 @@ func (s *JobScheduler) runOnce(ctx context.Context) (*schedulerRunOnceResult, er
 
 	for {
 		numScheduled, err := timeoututil.WithTimeoutV(ctx, riversharedmaintenance.TimeoutDefault, s.Name+".runOnce", func(ctx context.Context) (int, error) {
-			execTx, err := s.exec.Begin(ctx)
+			// Scheduling reads eligible jobs before updating them.
+			execTx, err := s.exec.BeginReadThenWrite(ctx)
 			if err != nil {
 				return 0, fmt.Errorf("error starting transaction: %w", err)
 			}

internal/maintenance/job_scheduler_test.go

@@ -3,6 +3,7 @@ package maintenance
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 	"os"
@@ -36,6 +37,10 @@ func (e *jobSchedulerExecutorMock) Begin(ctx context.Context) (riverdriver.Execu
 	return e, nil
 }
 
+func (e *jobSchedulerExecutorMock) BeginReadThenWrite(ctx context.Context) (riverdriver.ExecutorTx, error) {
+	return e, nil
+}
+
 func (e *jobSchedulerExecutorMock) JobSchedule(ctx context.Context, params *riverdriver.JobScheduleParams) ([]*riverdriver.JobScheduleResult, error) {
 	e.TestSignals.JobScheduleStarted.Signal(struct{}{})
 	return e.jobScheduleFunc(ctx, params)
@@ -55,6 +60,17 @@ func (ts *jobSchedulerExecutorMockTestSignals) Init(t *testing.T) {
 	ts.JobScheduleStarted.Init(t)
 }
 
+// jobSchedulerReadThenWriteExecutor fails Begin, so a test can require that the
+// job scheduler begins its transaction with BeginReadThenWrite, which SQLite
+// begins `BEGIN IMMEDIATE`.
+type jobSchedulerReadThenWriteExecutor struct {
+	riverdriver.Executor
+}
+
+func (e *jobSchedulerReadThenWriteExecutor) Begin(ctx context.Context) (riverdriver.ExecutorTx, error) {
+	return nil, errors.New("job scheduler should begin its transaction with BeginReadThenWrite")
+}
+
 func TestJobScheduler(t *testing.T) {
 	t.Parallel()
 
@@ -317,6 +333,20 @@ func TestJobScheduler(t *testing.T) {
 		riversharedtest.WaitOrTimeout(t, stopped)
 	})
 
+	t.Run("BeginsReadThenWrite", func(t *testing.T) {
+		t.Parallel()
+
+		tx := riverdbtest.TestTxPgx(ctx, t)
+		scheduler, bundle := setup(t, &testOpts{exec: &jobSchedulerReadThenWriteExecutor{riverpgxv5.New(nil).UnwrapExecutor(tx)}})
+
+		job := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: new(rivertype.JobStateScheduled), ScheduledAt: new(time.Now().Add(-1 * time.Hour))})
+
+		_, err := scheduler.runOnce(ctx)
+		require.NoError(t, err)
+
+		requireJobStateAvailable(t, scheduler, bundle.exec, job)
+	})
+
 	t.Run("CanRunMultipleTimes", func(t *testing.T) {
 		t.Parallel()
 

riverdriver/river_driver_interface.go

@@ -204,6 +204,14 @@ type Executor interface {
 	// subtransactions (like riverdriver/riverdatabasesql for database/sql).
 	Begin(ctx context.Context) (ExecutorTx, error)
 
+	// BeginReadThenWrite is like Begin, for a transaction that reads before it
+	// writes. SQLite begins it with `BEGIN IMMEDIATE` so that it takes the write
+	// lock before its first read and waits for busy_timeout instead of failing
+	// with SQLITE_BUSY when another connection commits after the read. Other
+	// drivers treat it the same as Begin. In a transaction, it begins a
+	// subtransaction like Begin.
+	BeginReadThenWrite(ctx context.Context) (ExecutorTx, error)
+
 	// ColumnExists checks whether a column for a particular table exists for
 	// the schema in the current search schema.
 	ColumnExists(ctx context.Context, params *ColumnExistsParams) (bool, error)

riverdriver/riverdatabasesql/river_database_sql_driver.go

@@ -173,6 +173,12 @@ func (e *Executor) Begin(ctx context.Context) (riverdriver.ExecutorTx, error) {
 	return &ExecutorTx{Executor: Executor{nil, templateReplaceWrapper{tx, &e.driver.replacer}, e.driver}, tx: tx}, nil
 }
 
+// BeginReadThenWrite is the same as Begin, because Postgres doesn't fail a
+// transaction for reading before it writes.
+func (e *Executor) BeginReadThenWrite(ctx context.Context) (riverdriver.ExecutorTx, error) {
+	return e.Begin(ctx)
+}
+
 func (e *Executor) ColumnExists(ctx context.Context, params *riverdriver.ColumnExistsParams) (bool, error) {
 	// Schema injection is a bit different on this one because we're querying a table with a schema name.
 	schema := "CURRENT_SCHEMA"
@@ -1202,6 +1208,12 @@ func (t *ExecutorTx) Begin(ctx context.Context) (riverdriver.ExecutorTx, error)
 	}).Begin(ctx)
 }
 
+// BeginReadThenWrite is the same as Begin, because Postgres doesn't fail a
+// transaction for reading before it writes.
+func (t *ExecutorTx) BeginReadThenWrite(ctx context.Context) (riverdriver.ExecutorTx, error) {
+	return t.Begin(ctx)
+}
+
 func (t *ExecutorTx) Commit(ctx context.Context) error {
 	// unfortunately, `database/sql` does not take a context ...
 	return t.tx.Commit()
@@ -1235,6 +1247,12 @@ func (t *ExecutorSubTx) Begin(ctx context.Context) (riverdriver.ExecutorTx, erro
 	return &ExecutorSubTx{Executor: Executor{nil, templateReplaceWrapper{t.tx, &t.driver.replacer}, t.driver}, savepointNum: nextSavepointNum, beginOnce: savepointutil.NewBeginOnlyOnce(t.beginOnce), tx: t.tx}, nil
 }
 
+// BeginReadThenWrite is the same as Begin, because Postgres doesn't fail a
+// transaction for reading before it writes.
+func (t *ExecutorSubTx) BeginReadThenWrite(ctx context.Context) (riverdriver.ExecutorTx, error) {
+	return t.Begin(ctx)
+}
+
 func (t *ExecutorSubTx) Commit(ctx context.Context) error {
 	defer t.beginOnce.Done()
 

riverdriver/riverdrivertest/client_maintenance_test.go

@@ -50,6 +50,14 @@ func (e *maintenanceRecoveryExecutor) Begin(ctx context.Context) (riverdriver.Ex
 	return &maintenanceRecoveryExecutorTx{ExecutorTx: tx, driver: e.driver}, nil
 }
 
+func (e *maintenanceRecoveryExecutor) BeginReadThenWrite(ctx context.Context) (riverdriver.ExecutorTx, error) {
+	tx, err := e.Executor.BeginReadThenWrite(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return &maintenanceRecoveryExecutorTx{ExecutorTx: tx, driver: e.driver}, nil
+}
+
 func (e *maintenanceRecoveryExecutor) LeaderResign(ctx context.Context, params *riverdriver.LeaderResignParams) (bool, error) {
 	if e.driver.resignAttempts.Add(1) <= e.driver.resignTimeouts {
 		<-ctx.Done()

riverdriver/riverdrivertest/executor_tx.go

@@ -32,6 +32,163 @@ func exerciseExecutorTx[TTx any](ctx context.Context, t *testing.T,
 		return exec
 	}
 
+	t.Run("BeginReadThenWrite", func(t *testing.T) {
+		t.Parallel()
+
+		// Runs fn from numWorkers goroutines numIterations times each and
+		// fails with every error returned.
+		runConcurrently := func(t *testing.T, numWorkers, numIterations int, fn func(worker int) error) {
+			t.Helper()
+
+			errs := make(chan error, numIterations*numWorkers)
+			var wg sync.WaitGroup
+			for worker := range numWorkers {
+				wg.Go(func() {
+					for range numIterations {
+						if err := fn(worker); err != nil {
+							errs <- err
+						}
+					}
+				})
+			}
+			wg.Wait()
+			close(errs)
+
+			var allErrs []error
+			for err := range errs {
+				allErrs = append(allErrs, err)
+			}
+			require.NoError(t, errors.Join(allErrs...), "%d of %d operations failed", len(allErrs), numIterations*numWorkers)
+		}
+
+		// Transactions on separate connections that each read and then write.
+		// In SQLite under WAL, a deferred transaction that has read can't be
+		// upgraded to a write transaction once another connection has
+		// committed, and fails with SQLITE_BUSY without consulting
+		// busy_timeout. Transactions begun immediate wait for each other.
+		t.Run("ConcurrentReadThenWrite", func(t *testing.T) {
+			t.Parallel()
+
+			driver, schema := driverWithSchema(ctx, t, nil)
+			exec := driver.GetExecutor()
+
+			job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Schema: schema})
+			insertParams := testfactory.Job_Build(t, &testfactory.JobOpts{Schema: schema})
+
+			// Transactions must be able to run on separate connections at
+			// once, or nothing here contends for SQLite's write lock.
+			tx1, err := exec.Begin(ctx)
+			require.NoError(t, err)
+			beginCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
+			tx2, err := exec.Begin(beginCtx)
+			require.NoError(t, err, "test pool should allow concurrent transactions")
+			require.NoError(t, tx2.Rollback(ctx))
+			cancel()
+			require.NoError(t, tx1.Rollback(ctx))
+
+			runConcurrently(t, 8, 20, func(int) error {
+				tx, err := exec.BeginReadThenWrite(ctx)
+				if err != nil {
+					return err
+				}
+				defer dbutil.RollbackWithoutCancel(ctx, tx)
+
+				if _, err := tx.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID, Schema: schema}); err != nil {
+					return err
+				}
+				if _, err := tx.JobInsertFull(ctx, insertParams); err != nil {
+					return err
+				}
+				return tx.Commit(ctx)
+			})
+		})
+
+		// The job scheduler's transaction reads eligible jobs and then updates
+		// them, while other connections insert. Half the schedulers begin a
+		// transaction the way the job scheduler does, and half call
+		// JobSchedule outside one, where it begins its own.
+		t.Run("ConcurrentJobSchedule", func(t *testing.T) {
+			t.Parallel()
+
+			driver, schema := driverWithSchema(ctx, t, nil)
+			exec := driver.GetExecutor()
+
+			insertParams := testfactory.Job_Build(t, &testfactory.JobOpts{
+				ScheduledAt: new(time.Now().Add(-1 * time.Minute)),
+				Schema:      schema,
+				State:       new(rivertype.JobStateScheduled),
+			})
+			scheduleParams := &riverdriver.JobScheduleParams{Max: 100, Now: new(time.Now()), Schema: schema}
+
+			runConcurrently(t, 8, 20, func(worker int) error {
+				switch worker % 4 {
+				case 0, 1:
+					_, err := exec.JobInsertFull(ctx, insertParams)
+					return err
+				case 2:
+					tx, err := exec.BeginReadThenWrite(ctx)
+					if err != nil {
+						return err
+					}
+					defer dbutil.RollbackWithoutCancel(ctx, tx)
+
+					if _, err := tx.JobSchedule(ctx, scheduleParams); err != nil {
+						return err
+					}
+					return tx.Commit(ctx)
+				default:
+					_, err := exec.JobSchedule(ctx, scheduleParams)
+					return err
+				}
+			})
+
+			// Everything inserted is scheduled by a last pass.
+			_, err := exec.JobSchedule(ctx, scheduleParams)
+			require.NoError(t, err)
+			numScheduled, err := exec.JobCountByState(ctx, &riverdriver.JobCountByStateParams{Schema: schema, State: rivertype.JobStateScheduled})
+			require.NoError(t, err)
+			require.Zero(t, numScheduled)
+		})
+
+		t.Run("InSubtransactionBeginsSubtransaction", func(t *testing.T) {
+			t.Parallel()
+
+			exec := setup(ctx, t)
+
+			tx, err := exec.BeginReadThenWrite(ctx)
+			require.NoError(t, err)
+			defer dbutil.RollbackWithoutCancel(ctx, tx)
+
+			outerJob := testfactory.Job(ctx, t, tx, &testfactory.JobOpts{})
+
+			subTx, err := tx.BeginReadThenWrite(ctx)
+			require.NoError(t, err)
+
+			innerJob := testfactory.Job(ctx, t, subTx, &testfactory.JobOpts{})
+			require.NoError(t, subTx.Rollback(ctx))
+
+			_, err = tx.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: innerJob.ID})
+			require.ErrorIs(t, err, rivertype.ErrNotFound)
+			_, err = tx.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: outerJob.ID})
+			require.NoError(t, err)
+		})
+
+		t.Run("InTransactionBeginsSubtransaction", func(t *testing.T) {
+			t.Parallel()
+
+			exec := setup(ctx, t)
+
+			tx, err := exec.BeginReadThenWrite(ctx)
+			require.NoError(t, err)
+
+			job := testfactory.Job(ctx, t, tx, &testfactory.JobOpts{})
+			require.NoError(t, tx.Rollback(ctx))
+
+			_, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID})
+			require.ErrorIs(t, err, rivertype.ErrNotFound)
+		})
+	})
+
 	t.Run("Begin", func(t *testing.T) {
 		t.Parallel()
 

riverdriver/riverpgxv5/river_pgx_v5_driver.go

@@ -141,6 +141,12 @@ func (e *Executor) Begin(ctx context.Context) (riverdriver.ExecutorTx, error) {
 	return &ExecutorTx{Executor: Executor{templateReplaceWrapper{tx, &e.driver.replacer}, e.driver}, tx: tx}, nil
 }
 
+// BeginReadThenWrite is the same as Begin, because Postgres doesn't fail a
+// transaction for reading before it writes.
+func (e *Executor) BeginReadThenWrite(ctx context.Context) (riverdriver.ExecutorTx, error) {
+	return e.Begin(ctx)
+}
+
 func (e *Executor) ColumnExists(ctx context.Context, params *riverdriver.ColumnExistsParams) (bool, error) {
 	// Schema injection is a bit different on this one because we're querying a table with a schema name.
 	schema := "CURRENT_SCHEMA"