Preliminary "foot in the door" SQLite support

Here, add an implementation for SQLite. The new driver passes the entire
driver test suite (including some newly added tests), and checks out
against a new end-to-end suite of client tests targeting it. I think at
this point we can consider it to be largely functional, although with
minimal real-world vetting, but at least alpha quality that we can start
putting in front of users.

The main reason this turned out to be quite an effort is that SQLite,
although nominally supporting a lot of the same syntax Postgres does,
just has a lot of unexpected limitations and quirks that all need to be
tracked down separately and a workaround added. For example:

* SQLite doesn't really have data types. Everything is either an
  integer, real, text, or blob, and anything more sophisticated just has
  to be shoehorned into one of these types. For example, there's no
  boolean. Despite the fanfare recently, there isn't even a jsonb. Jsonb
  gets pushed into a blob.

* The most annoying missing data type is a timestamp. Date/times are
  stored as either Unix integers or strings, and SQLite provides a
  number of obnoxious, error-prone, built-ins to work with them.

* No arrays.

* No modification operations in CTEs. (Although counting my lucky stars
  that CTEs are supported at least.)

* No listen/notify.

Aside from that though, we had to get some testing infrastructure in
place, and our testing has traditionally been quite affixed to Postgres,
which would've made a lot of changes necessary. However, the schema
additions in #848 do help to make this quite a bit easier with the
driver-based `TestSchema` and `TestTx` functions, and because they raise
schema automatically, it doesn't complicate test setup instructions in
any way by adding extra steps.

A few caveats:

* I've found that the combination of SQLite's reduced capabilities +
  sqlc's bugginess [1] make bit batch insert/update operations basically
  impossible (I tried every workaround I could possibly thin of), so
  these are done in loops of individual operations instead. I *think*
  this is okay for now though. For one, sqlc will hopefully get these
  limitations fixed eventually, and for two, SQLite databases will often
  be running locally, meaning the round trip cost per operation is much
  lower than what we'd see in a hosted Postgres somewhere.

* It's not totally clear that having a SQLite implementation will be
  "worth it enough" in the long run given that it will add some
  difficulty to non-trivial future database-related operations. My hope
  is that as it's in a prerelease state, we can gauge how bad it is to
  keep up to date. If it feels like way more effort than it's worth, we
  can still axe it before it ever becomes a functional driver.

[1] sqlc-dev/sqlc#3802 (comment)

Author: brandur

.gitignore

@@ -2,3 +2,4 @@
 /river
 /internal/cmd/riverbench/riverbench
 /internal/cmd/testdbman/testdbman
+/sqlite/

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Preliminary River driver for SQLite (`riverdriver/riversqlite`). This driver seems to produce good results as judged by the test suite, but so far as minimal real world vetting. Try it and let us know how it works out. [PR #870](https://github.com/riverqueue/river/pull/870).
+
 ## [0.22.0] - 2025-05-10
 
 ### Added
@@ -19,6 +23,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Resuming an already unpaused queue is now fully an no-op, and won't touch the row's `updated_at` like it (unintentionally) did before. [PR #870](https://github.com/riverqueue/river/pull/870).
 - The `riverdatabasesql` now fully supports raw connections through [`lib/pq`](https://github.com/lib/pq) rather than just `database/sql` through Pgx. We don't recommend the use of `lib/pq` as it's an unmaintained project, but this change should help with compatibility for older projects. [PR #883](https://github.com/riverqueue/river/pull/883).
 
 ## [0.21.0] - 2025-05-02

Makefile

@@ -28,6 +28,7 @@ generate/migrations: ## Sync changes of pgxv5 migrations to database/sql
 generate/sqlc: ## Generate sqlc
 	cd riverdriver/riverdatabasesql/internal/dbsqlc && sqlc generate
 	cd riverdriver/riverpgxv5/internal/dbsqlc && sqlc generate
+	cd riverdriver/riversqlite/internal/dbsqlc && sqlc generate
 
 # Looks at comments using ## on targets and uses them to produce a help output.
 .PHONY: help

client.go

@@ -339,6 +339,15 @@ type Config struct {
 	// instances of rivertype.WorkerMiddleware).
 	WorkerMiddleware []rivertype.WorkerMiddleware
 
+	// queuePollInterval is the amount of time between periodic checks for queue
+	// setting changes. This is only used in poll-only mode (when no notifier is
+	// provided).
+	//
+	// This is internal for the time being as it hasn't had any major demand to
+	// be exposed, but it's needed to make sure that our poll-only tests can
+	// finish in a timely manner.
+	queuePollInterval time.Duration
+
 	// Scheduler run interval. Shared between the scheduler and producer/job
 	// executors, but not currently exposed for configuration.
 	schedulerInterval time.Duration
@@ -400,6 +409,7 @@ func (c *Config) WithDefaults() *Config {
 		TestOnly:                    c.TestOnly,
 		WorkerMiddleware:            c.WorkerMiddleware,
 		Workers:                     c.Workers,
+		queuePollInterval:           c.queuePollInterval,
 		schedulerInterval:           cmp.Or(c.schedulerInterval, maintenance.JobSchedulerIntervalDefault),
 	}
 }
@@ -715,7 +725,7 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client
 	// we're actually going to be working jobs (as opposed to just enqueueing
 	// them):
 	if config.willExecuteJobs() {
-		if !driver.HasPool() {
+		if !driver.PoolIsSet() {
 			return nil, errMissingDatabasePoolWithQueues
 		}
 
@@ -1510,7 +1520,7 @@ var errNoDriverDBPool = errors.New("driver must have non-nil database pool to us
 //		// handle error
 //	}
 func (c *Client[TTx]) Insert(ctx context.Context, args JobArgs, opts *InsertOpts) (*rivertype.JobInsertResult, error) {
-	if !c.driver.HasPool() {
+	if !c.driver.PoolIsSet() {
 		return nil, errNoDriverDBPool
 	}
 
@@ -1587,7 +1597,7 @@ type InsertManyParams struct {
 //		// handle error
 //	}
 func (c *Client[TTx]) InsertMany(ctx context.Context, params []InsertManyParams) ([]*rivertype.JobInsertResult, error) {
-	if !c.driver.HasPool() {
+	if !c.driver.PoolIsSet() {
 		return nil, errNoDriverDBPool
 	}
 
@@ -1765,7 +1775,7 @@ func (c *Client[TTx]) insertManyParams(params []InsertManyParams) ([]*rivertype.
 // Unlike with `InsertMany`, unique conflicts cannot be handled gracefully. If a
 // unique constraint is violated, the operation will fail and no jobs will be inserted.
 func (c *Client[TTx]) InsertManyFast(ctx context.Context, params []InsertManyParams) (int, error) {
-	if !c.driver.HasPool() {
+	if !c.driver.PoolIsSet() {
 		return 0, errNoDriverDBPool
 	}
 
@@ -1865,20 +1875,23 @@ func (c *Client[TTx]) maybeNotifyInsertForQueues(ctx context.Context, tx riverdr
 		return nil
 	}
 
-	err := tx.NotifyMany(ctx, &riverdriver.NotifyManyParams{
-		Payload: payloads,
-		Schema:  c.config.Schema,
-		Topic:   string(notifier.NotificationTopicInsert),
-	})
-	if err != nil {
-		c.baseService.Logger.ErrorContext(
-			ctx,
-			c.baseService.Name+": Failed to send job insert notification",
-			slog.String("queues", strings.Join(queuesDeduped, ",")),
-			slog.String("err", err.Error()),
-		)
-		return err
+	if c.driver.SupportsListenNotify() {
+		err := tx.NotifyMany(ctx, &riverdriver.NotifyManyParams{
+			Payload: payloads,
+			Schema:  c.config.Schema,
+			Topic:   string(notifier.NotificationTopicInsert),
+		})
+		if err != nil {
+			c.baseService.Logger.ErrorContext(
+				ctx,
+				c.baseService.Name+": Failed to send job insert notification",
+				slog.String("queues", strings.Join(queuesDeduped, ",")),
+				slog.String("err", err.Error()),
+			)
+			return err
+		}
 	}
+
 	return nil
 }
 
@@ -1896,19 +1909,22 @@ func (c *Client[TTx]) notifyQueuePauseOrResume(ctx context.Context, tx riverdriv
 		return err
 	}
 
-	err = tx.NotifyMany(ctx, &riverdriver.NotifyManyParams{
-		Payload: []string{string(payload)},
-		Schema:  c.config.Schema,
-		Topic:   string(notifier.NotificationTopicControl),
-	})
-	if err != nil {
-		c.baseService.Logger.ErrorContext(
-			ctx,
-			c.baseService.Name+": Failed to send queue state change notification",
-			slog.String("err", err.Error()),
-		)
-		return err
+	if c.driver.SupportsListenNotify() {
+		err = tx.NotifyMany(ctx, &riverdriver.NotifyManyParams{
+			Payload: []string{string(payload)},
+			Schema:  c.config.Schema,
+			Topic:   string(notifier.NotificationTopicControl),
+		})
+		if err != nil {
+			c.baseService.Logger.ErrorContext(
+				ctx,
+				c.baseService.Name+": Failed to send queue state change notification",
+				slog.String("err", err.Error()),
+			)
+			return err
+		}
 	}
+
 	return nil
 }
 
@@ -1943,6 +1959,7 @@ func (c *Client[TTx]) addProducer(queueName string, queueConfig QueueConfig) *pr
 		Notifier:                     c.notifier,
 		Queue:                        queueName,
 		QueueEventCallback:           c.subscriptionManager.distributeQueueEvent,
+		QueuePollInterval:            c.config.queuePollInterval,
 		RetryPolicy:                  c.config.RetryPolicy,
 		SchedulerInterval:            c.config.schedulerInterval,
 		Schema:                       c.config.Schema,
@@ -1988,7 +2005,7 @@ type JobListResult struct {
 //		// handle error
 //	}
 func (c *Client[TTx]) JobList(ctx context.Context, params *JobListParams) (*JobListResult, error) {
-	if !c.driver.HasPool() {
+	if !c.driver.PoolIsSet() {
 		return nil, errNoDriverDBPool
 	}
 
@@ -1997,12 +2014,16 @@ func (c *Client[TTx]) JobList(ctx context.Context, params *JobListParams) (*JobL
 	}
 	params.schema = c.config.Schema
 
+	if c.driver.DatabaseName() == "sqlite" && params.metadataFragment != "" {
+		return nil, errors.New("JobListResult.Metadata is not supported on SQLite")
+	}
+
 	dbParams, err := params.toDBParams()
 	if err != nil {
 		return nil, err
 	}
 
-	jobs, err := dblist.JobList(ctx, c.driver.GetExecutor(), dbParams)
+	jobs, err := dblist.JobList(ctx, c.driver.GetExecutor(), dbParams, c.driver.SQLFragmentColumnIn)
 	if err != nil {
 		return nil, err
 	}
@@ -2033,7 +2054,7 @@ func (c *Client[TTx]) JobListTx(ctx context.Context, tx TTx, params *JobListPara
 		return nil, err
 	}
 
-	jobs, err := dblist.JobList(ctx, c.driver.UnwrapExecutor(tx), dbParams)
+	jobs, err := dblist.JobList(ctx, c.driver.UnwrapExecutor(tx), dbParams, c.driver.SQLFragmentColumnIn)
 	if err != nil {
 		return nil, err
 	}
@@ -2325,12 +2346,14 @@ func (c *Client[TTx]) queueUpdate(ctx context.Context, executorTx riverdriver.Ex
 			return nil, err
 		}
 
-		if err := executorTx.NotifyMany(ctx, &riverdriver.NotifyManyParams{
-			Payload: []string{string(payload)},
-			Schema:  c.config.Schema,
-			Topic:   string(notifier.NotificationTopicControl),
-		}); err != nil {
-			return nil, err
+		if c.driver.SupportsListenNotify() {
+			if err := executorTx.NotifyMany(ctx, &riverdriver.NotifyManyParams{
+				Payload: []string{string(payload)},
+				Schema:  c.config.Schema,
+				Topic:   string(notifier.NotificationTopicControl),
+			}); err != nil {
+				return nil, err
+			}
 		}
 	}
 

client_test.go

@@ -134,6 +134,7 @@ func newTestConfig(t *testing.T, schema string) *Config {
 		},
 		TestOnly:          true, // disables staggered start in maintenance services
 		Workers:           workers,
+		queuePollInterval: 50 * time.Millisecond,
 		schedulerInterval: riverinternaltest.SchedulerShortInterval,
 	}
 }
@@ -1242,7 +1243,7 @@ func Test_Client(t *testing.T) {
 		require.NoError(t, err)
 
 		event := riversharedtest.WaitOrTimeout(t, subscribeChan)
-		require.Equal(t, (withKindAliasesArgs{}).KindAliases(), []string{event.Job.Kind})
+		require.Equal(t, []string{event.Job.Kind}, (withKindAliasesArgs{}).KindAliases())
 	})
 
 	t.Run("StartStopStress", func(t *testing.T) {

cmd/river/go.sum

@@ -24,14 +24,18 @@ github.com/lmittmann/tint v1.0.7 h1:D/0OqWZ0YOGZ6AyC+5Y2kD8PBEzBk6rFHVSfOqCkF9Y=
 github.com/lmittmann/tint v1.0.7/go.mod h1:HIS3gSy7qNwGCj+5oRjAutErFBl4BzdQP6cJZ0NfMwE=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/riverqueue/river/riverdriver v0.20.2 h1:FDmWALB6DvYBBw479euIBg1KClxPmDpWjmZbhScxSBw=
-github.com/riverqueue/river/riverdriver v0.20.2/go.mod h1:vYSv6ZTEFWT0JVuGCwZDxJdc2U7ZMkwJQ+nPsa7/2mM=
-github.com/riverqueue/river/riverdriver/riverdatabasesql v0.20.2 h1:llBsU1hpKyIIzZroeVjM7uavmq3W+kXuSvkUCQ/3pg4=
-github.com/riverqueue/river/riverdriver/riverdatabasesql v0.20.2/go.mod h1:qPJ5qkfAqAYRKXxU1TNFsVwMd9dLIXEFDLrrGz6GAWM=
-github.com/riverqueue/river/riverdriver/riverpgxv5 v0.20.2 h1:O8e1vobbKhUmgbki0mLOvCptixMtBiMjJgkGPa4VFAY=
-github.com/riverqueue/river/riverdriver/riverpgxv5 v0.20.2/go.mod h1:zn3Lf6qzkq9kEOzYRe/fEgYl9c/eRTCdwBHtclxILEU=
-github.com/riverqueue/river/rivertype v0.20.2 h1:unmiQP7CWS6IDbDrp9cESNscPoMstxb6Luoz9kfNzOc=
-github.com/riverqueue/river/rivertype v0.20.2/go.mod h1:lmdl3vLNDfchDWbYdW2uAocIuwIN+ZaXqAukdSCFqWs=
+github.com/riverqueue/river v0.22.0 h1:PO4Ula2RqViQqNs6xjze7yFV6Zq4T3Ffv092+f4S8xQ=
+github.com/riverqueue/river v0.22.0/go.mod h1:IRoWoK4RGCiPuVJUV4EWcCl9d/TMQYkk0EEYV/Wgq+U=
+github.com/riverqueue/river/riverdriver v0.22.0 h1:i7OSFkUi6x4UKvttdFOIg7NYLYaBOFLJZvkZ0+JWS/8=
+github.com/riverqueue/river/riverdriver v0.22.0/go.mod h1:oNdjJCeAJhN/UiZGLNL+guNqWaxMFuSD4lr5x/v/was=
+github.com/riverqueue/river/riverdriver/riverdatabasesql v0.22.0 h1:+no3gToOK9SmWg0pDPKfOGSCsrxqqaFdD8K1NQndRbY=
+github.com/riverqueue/river/riverdriver/riverdatabasesql v0.22.0/go.mod h1:mygiHa1dnlKRjxT1//wIvfT2fMTbfXKm37NcsxoyBoQ=
+github.com/riverqueue/river/riverdriver/riverpgxv5 v0.22.0 h1:2TWbVL73gipJ2/4JNCQbifaNj+BCC/Zxpp30o1D8RTg=
+github.com/riverqueue/river/riverdriver/riverpgxv5 v0.22.0/go.mod h1:TZY/BG8w/nDxkraAEvvgyVupIz0b4+PQVUW0kIiy1fc=
+github.com/riverqueue/river/rivershared v0.22.0 h1:hLPHr98d6OEfmUJ4KpIXgoy2tbQ14htWILcRBHJF11U=
+github.com/riverqueue/river/rivershared v0.22.0/go.mod h1:BK+hvhECfdDLWNDH3xiGI95m2YoPfVtECZLT+my8XM8=
+github.com/riverqueue/river/rivertype v0.22.0 h1:rSRhbd5uV/BaFTPxReCxuYTAzx+/riBZJlZdREADvO4=
+github.com/riverqueue/river/rivertype v0.22.0/go.mod h1:lmdl3vLNDfchDWbYdW2uAocIuwIN+ZaXqAukdSCFqWs=
 github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
 github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=

docs/development.md

@@ -51,6 +51,7 @@ To run programs locally outside of tests, create and raise a development databas
     git tag riverdriver/$VERSION -m "release riverdriver/$VERSION"
     git tag riverdriver/riverpgxv5/$VERSION -m "release riverdriver/riverpgxv5/$VERSION"
     git tag riverdriver/riverdatabasesql/$VERSION -m "release riverdriver/riverdatabasesql/$VERSION"
+    git tag riverdriver/riversqlite/$VERSION -m "release riverdriver/riversqlite/$VERSION"
     git tag rivershared/$VERSION -m "release rivershared/$VERSION"
     git tag rivertype/$VERSION -m "release rivertype/$VERSION"
     git tag $VERSION