diff --git a/CHANGELOG.md b/CHANGELOG.md index eedce02fd..dd132fb0e 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/cmd/river/rivercli/command.go b/cmd/river/rivercli/command.go index fcdcdf7a1..8169f2d95 100644 --- a/cmd/river/rivercli/command.go +++ b/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 diff --git a/internal/maintenance/job_scheduler.go b/internal/maintenance/job_scheduler.go index 09ce2d220..0310e6d32 100644 --- a/internal/maintenance/job_scheduler.go +++ b/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) } diff --git a/internal/maintenance/job_scheduler_test.go b/internal/maintenance/job_scheduler_test.go index 5e33cfcd7..b10417a83 100644 --- a/internal/maintenance/job_scheduler_test.go +++ b/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() diff --git a/riverdriver/river_driver_interface.go b/riverdriver/river_driver_interface.go index f76adc29b..56005ec44 100644 --- a/riverdriver/river_driver_interface.go +++ b/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) diff --git a/riverdriver/riverdatabasesql/river_database_sql_driver.go b/riverdriver/riverdatabasesql/river_database_sql_driver.go index 0671e852f..d28002857 100644 --- a/riverdriver/riverdatabasesql/river_database_sql_driver.go +++ b/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() diff --git a/riverdriver/riverdrivertest/client_maintenance_test.go b/riverdriver/riverdrivertest/client_maintenance_test.go index 028c47258..7d8076e80 100644 --- a/riverdriver/riverdrivertest/client_maintenance_test.go +++ b/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() diff --git a/riverdriver/riverdrivertest/executor_tx.go b/riverdriver/riverdrivertest/executor_tx.go index 65e1372db..885dcce59 100644 --- a/riverdriver/riverdrivertest/executor_tx.go +++ b/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() diff --git a/riverdriver/riverpgxv5/river_pgx_v5_driver.go b/riverdriver/riverpgxv5/river_pgx_v5_driver.go index 71d56fef6..a1b43db02 100644 --- a/riverdriver/riverpgxv5/river_pgx_v5_driver.go +++ b/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" diff --git a/riverdriver/riversqlite/river_sqlite_driver.go b/riverdriver/riversqlite/river_sqlite_driver.go index 7d1044bfc..0238bee9e 100644 --- a/riverdriver/riversqlite/river_sqlite_driver.go +++ b/riverdriver/riversqlite/river_sqlite_driver.go @@ -5,11 +5,20 @@ // thoroughly in the test suite, but has minimal real world use as of yet. // // River makes extensive use of internal operations that might run in parallel, -// which doesn't naturally play well with SQLite, which only allows one -// operation at a time, returning errors like "database is locked (5) -// (SQLITE_BUSY)" in case another tries to access it. A good workaround to avoid -// errors is to set the maximum pool size to one connection like -// `dbPool.SetMaxOpenConns(1)`. +// while SQLite allows only one writer at a time. With WAL and `busy_timeout` +// set, a transaction that writes first waits for the write lock, but one that +// reads and then writes fails at once with "database is locked (5) +// (SQLITE_BUSY)" if another connection commits in between. The driver begins +// River's transactions that read before writing (like the job scheduler's) +// with `BEGIN IMMEDIATE` so they wait too, and leaves the rest deferred. +// Operations can still fail if the lock is held for longer than +// `busy_timeout`, and since only one connection can write at a time, the +// fastest and most reliable configuration remains a pool of one connection +// like `dbPool.SetMaxOpenConns(1)`. Transactions that an application begins +// and passes to River are deferred by default in database/sql. If they read +// before writing on a pool of more than one connection, begin them immediate +// too (modernc.org/sqlite and mattn/go-sqlite3 take `_txlock=immediate` in the +// DSN). // // A known deficiency in this driver compared to Postgres is that due to // limitations in sqlc, it performs operations like completion and `InsertMany` @@ -192,6 +201,17 @@ type Executor struct { } func (e *Executor) Begin(ctx context.Context) (riverdriver.ExecutorTx, error) { + return e.begin(ctx, false) +} + +// BeginReadThenWrite begins a transaction with `BEGIN IMMEDIATE`, which takes +// SQLite's write lock up front. See beginImmediate. In a transaction, it begins +// a subtransaction like Begin, and the outer transaction keeps its own kind. +func (e *Executor) BeginReadThenWrite(ctx context.Context) (riverdriver.ExecutorTx, error) { + return e.begin(ctx, true) +} + +func (e *Executor) begin(ctx context.Context, immediate bool) (riverdriver.ExecutorTx, error) { if e.execTx != nil { return e.execTx.Begin(ctx) } @@ -201,7 +221,12 @@ func (e *Executor) Begin(ctx context.Context) (riverdriver.ExecutorTx, error) { return nil, err } - tx, err := conn.BeginTx(ctx, nil) + var tx *sql.Tx + if immediate { + tx, err = beginImmediate(ctx, conn) + } else { + tx, err = conn.BeginTx(ctx, nil) + } if err != nil { // Turso can execute BEGIN, then report context cancellation without // rolling back. Discard the connection while we still own it so an @@ -877,7 +902,11 @@ func (e *Executor) JobSchedule(ctx context.Context, params *riverdriver.JobSched // version (i.e. an advisory lock should maybe be taken around when checking // rows to avoid non-repeatable read anomalies), but since only one writer // at a time is possible for SQLite, this should be okay. - return dbutil.WithTxV(ctx, e.executor(), func(ctx context.Context, execTx riverdriver.ExecutorTx) ([]*riverdriver.JobScheduleResult, error) { + // + // It reads eligible jobs before updating them, so outside a transaction it + // begins one immediate (see beginImmediate). Inside one, the caller should + // have begun it with BeginReadThenWrite, as the job scheduler does. + return dbutil.WithTxV(ctx, e.readThenWriteExecutor(), func(ctx context.Context, execTx riverdriver.ExecutorTx) ([]*riverdriver.JobScheduleResult, error) { ctx = schemaTemplateParam(ctx, params.Schema) dbtx := templateReplaceWrapper{dbtx: e.driver.UnwrapTx(execTx), replacer: &e.driver.replacer} @@ -1489,6 +1518,22 @@ func (e *Executor) executor() riverdriver.Executor { return e } +// readThenWriteExecutor is like executor, but outside a transaction its Begin +// is BeginReadThenWrite, for transaction helpers wrapping an operation that +// reads before it writes. +func (e *Executor) readThenWriteExecutor() riverdriver.Executor { + if e.execTx != nil { + return e.execTx + } + return readThenWriteExecutor{e} +} + +type readThenWriteExecutor struct{ *Executor } + +func (e readThenWriteExecutor) Begin(ctx context.Context) (riverdriver.ExecutorTx, error) { + return e.BeginReadThenWrite(ctx) +} + type ExecutorTx struct { Executor @@ -1582,6 +1627,54 @@ func (t *ExecutorSubTx) Rollback(ctx context.Context) error { return nil } +// beginImmediate begins a transaction on conn that holds SQLite's write lock +// from the start. +// +// database/sql starts a deferred transaction (a plain `BEGIN`), which takes no +// lock until its first statement. Under WAL, a deferred transaction that has +// read can't be upgraded to a write transaction if another connection has +// committed since its read began, and SQLite returns SQLITE_BUSY +// (SQLITE_BUSY_SNAPSHOT) immediately without consulting busy_timeout, because +// waiting couldn't make the snapshot current. +// `BEGIN IMMEDIATE` takes the write lock up front, where busy_timeout does +// apply, so concurrent transactions on a pool of more than one connection wait +// for each other instead of failing. +// +// Only transactions that read before they write need it. One whose first +// statement writes takes the write lock at that statement, before it has read +// anything, and busy_timeout applies there too. Holding the write lock across +// reads keeps other connections from writing for longer, so River's other +// transactions stay deferred. +// +// database/sql has no way to choose the kind of `BEGIN` (drivers like +// modernc.org/sqlite and mattn/go-sqlite3 only accept it as a `_txlock` DSN +// parameter, which Turso doesn't support), so the deferred transaction is +// replaced on the same connection. A deferred `BEGIN` holds nothing yet, so +// ending it is free, and the SQLite drivers commit and roll back by executing +// `COMMIT` and `ROLLBACK` on the connection, which apply to the immediate +// transaction in its place. +func beginImmediate(ctx context.Context, conn *sql.Conn) (*sql.Tx, error) { + tx, err := conn.BeginTx(ctx, nil) + if err != nil { + return nil, err + } + + if _, err := tx.ExecContext(ctx, "ROLLBACK"); err != nil { + _ = tx.Rollback() + return nil, err + } + + if _, err := tx.ExecContext(ctx, "BEGIN IMMEDIATE"); err != nil { + // The connection is out of a transaction at this point, so + // database/sql's own rollback is expected to fail. The caller discards + // the connection either way. + _ = tx.Rollback() + return nil, err + } + + return tx, nil +} + func interpretError(err error) error { if errors.Is(err, sql.ErrNoRows) { return rivertype.ErrNotFound diff --git a/riverdriver/riversqlite/river_sqlite_driver_test.go b/riverdriver/riversqlite/river_sqlite_driver_test.go index 9873917fc..50fe23be7 100644 --- a/riverdriver/riversqlite/river_sqlite_driver_test.go +++ b/riverdriver/riversqlite/river_sqlite_driver_test.go @@ -28,6 +28,63 @@ func TestDurationAsString(t *testing.T) { func TestExecutorBegin(t *testing.T) { t.Parallel() + t.Run("FailedBeginImmediateDiscardsConnection", func(t *testing.T) { + t.Parallel() + + // beginImmediate replaces database/sql's deferred BEGIN by executing + // ROLLBACK and BEGIN IMMEDIATE. If either fails, it must still end + // the *sql.Tx, or closing the connection waits on it forever. + for _, failingQuery := range []string{"ROLLBACK", "BEGIN IMMEDIATE"} { + t.Run(failingQuery, func(t *testing.T) { + t.Parallel() + + execErr := errors.New("exec failed: " + failingQuery) + + var opened, closed int + dbPool := sql.OpenDB(&beginTestConnector{ + connectFunc: func() (driver.Conn, error) { + opened++ + return &beginImmediateTestConn{ + beginTestConn: &beginTestConn{ + beginFunc: func() (driver.Tx, error) { return beginTestTx{}, nil }, + closeFunc: func() { closed++ }, + }, + execFunc: func(query string) error { + if query == failingQuery { + return execErr + } + return nil + }, + }, nil + }, + }) + t.Cleanup(func() { require.NoError(t, dbPool.Close()) }) + dbPool.SetMaxOpenConns(1) + exec := New(dbPool).GetExecutor() + + type beginResult struct { + err error + tx riverdriver.ExecutorTx + } + resultChan := make(chan beginResult, 1) + go func() { + tx, err := exec.BeginReadThenWrite(context.Background()) + resultChan <- beginResult{err: err, tx: tx} + }() + + select { + case result := <-resultChan: + require.ErrorIs(t, result.err, execErr) + require.Nil(t, result.tx) + case <-time.After(5 * time.Second): + require.FailNow(t, "BeginReadThenWrite hung discarding the connection") + } + require.Equal(t, 1, opened) + require.Equal(t, 1, closed) + }) + } + }) + t.Run("FailedBeginDiscardsConnection", func(t *testing.T) { t.Parallel() @@ -68,6 +125,19 @@ func TestExecutorBegin(t *testing.T) { }) } +type beginImmediateTestConn struct { + *beginTestConn + + execFunc func(query string) error +} + +func (c *beginImmediateTestConn) ExecContext(_ context.Context, query string, _ []driver.NamedValue) (driver.Result, error) { + if err := c.execFunc(query); err != nil { + return nil, err + } + return driver.RowsAffected(0), nil +} + type beginTestConn struct { driver.Conn @@ -94,6 +164,11 @@ func (c *beginTestConnector) Connect(context.Context) (driver.Conn, error) { return c.connectFunc() } +type beginTestTx struct{} + +func (beginTestTx) Commit() error { return nil } +func (beginTestTx) Rollback() error { return nil } + func TestInterpretError(t *testing.T) { t.Parallel() diff --git a/rivershared/riversharedtest/riversharedtest.go b/rivershared/riversharedtest/riversharedtest.go index a02143ade..ea1a9e931 100644 --- a/rivershared/riversharedtest/riversharedtest.go +++ b/rivershared/riversharedtest/riversharedtest.go @@ -179,30 +179,20 @@ func dbPoolSQLite(ctx context.Context, tb testing.TB, schema, driverName string) // Turso ignores `_pragma` so we don't bother passing it. Turso has better // defaults and WAL is enabled out of the box. if driverName != "turso" { - databaseURLBuilder.WriteString("?_pragma=journal_mode(WAL)") + databaseURLBuilder.WriteString("?_pragma=journal_mode(WAL)&_pragma=busy_timeout(30000)") } dbPool, err := sql.Open(driverName, databaseURLBuilder.String()) require.NoError(tb, err) tb.Cleanup(func() { require.NoError(tb, dbPool.Close()) }) - // River does enough concurrent work that given multiple active SQLite - // connections, it'll immediately start erroring with "database is locked - // (5) (SQLITE_BUSY)" because SQLite can only handle one operation at a time - // and explicitly errors if another is in flight. To prevent this problem, - // we constrain the maximum pool size to 1 so it limits concurrent access - // for us. - // - // I've seen some broad recommendations that it might be better to - // always set maximum connections for a single SQLite database to 1 - // anyway. See for example: - // - // https://news.ycombinator.com/item?id=30369095 - // - // An alternative approach is to increase `PRAGMA busy_timeout`, but I've - // found that we still run into `SQLITE_BUSY` errors with that at higher - // iteration counts like `-run TestClientWithDriverRiverSQLite -count 100`. - dbPool.SetMaxOpenConns(1) + // More than one connection, so tests exercise concurrent transactions. + // With `busy_timeout`, concurrent writers wait for SQLite's write lock + // instead of failing with "database is locked (5) (SQLITE_BUSY)", and + // riversqlite begins its transactions that read before writing with + // `BEGIN IMMEDIATE` so they wait too. Turso ignores `_pragma`, but defaults + // to a busy timeout of its own. + dbPool.SetMaxOpenConns(4) return dbPool }