Add pseudo listen/notify mechanism for SQLite

A fairly long-standing issue with SQLite compared to Postgres is the
lack of a listen/notify system because SQLite doesn't offer such a
mechanism. This has forced SQLite to operate in poll-only mode which
means there's a longer delay before it sees jobs that it can work.

Honker [1] hit Hacker News a few weeks/months back. It implements a
listen/notify-like mechanism, and when reading about how it's
implemented, I realized it wouldn't be all that crazy to do something
similar for River. Basically, it's poll based, and relies on polling a
single table fairly actively. That's something that's not great in
general, but we already do that for poll-only, except that it's multiple
tables instead of just one (i.e. `river_job` + `river_leader`).

Here, I propose that we implement listen/notify using a similar strategy
by adding a `river_notification` in SQLite. Like listen/notify, this
table has a listen/notify, along with a timestamp that we can use to
track event creation and prune older rows. We add a SQLite `Listener`
similar to the one for Postgres, but which polls `river_notification`,
remembering the last ID that it so it can make sure to only fetch new
notifications in each River process.

A new maintenance service that runs only in SQLite takes care of pruning
events in the table older than 5 minutes.

A nice property of `river_notification` is that like listen/notify, it
all works transactionally as well. New rows are hidden until the
transaction that inserted them commits, keeping them invisible from the
listener. After commit, the listener sees them and responds. We track a
`lastID` in the listener for more efficient reads, which seems like it
might be a problem due to out-of-order IDs, but for better or worse this
is not possible in SQLite due its single-writer model. IDs always commit
in order.

You might think that this would create a table that's overly high volume
because 1,000 job insertions would create 1,000 notifications, but
luckily we've been deduplicating notifications through the use of a
notify limiter (i.e. `insertNotifyLimiter.ShouldTrigger(queue)`) for a
long time now. Leadership notifications are also low volume, so this
table should never have to be very big at any given time, which is nice.

Requires a database migration, but that might dovetail fairly nicely
with #1224 (convert SQLite json to jsonb), which also does.

I've experimented with a few versions of this. This one also adds a
`river_notification` to Postgres, though doesn't use it. The rationale
here is (1) that the Postgres `river_notification` makes it easier to
test the client as if it was SQLite but without having to bring a SQLite
driver into the top-level project as a dependency, and (2) it may be a
useful alternative to poll-only even in Postgres down the line, where
something like a bouncer makes listen/notify difficult to use.

[1] https://github.com/russellromney/honker

Author: brandur

client.go

@@ -992,6 +992,13 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client
 			client.testSignals.queueCleaner = &queueCleaner.TestSignals
 		}
 
+		if driver.DatabaseName() == riverdriver.DatabaseNameSQLite {
+			sqliteNotificationCleaner := maintenance.NewSQLiteNotificationCleaner(archetype, &maintenance.SQLiteNotificationCleanerConfig{
+				Schema: config.Schema,
+			}, driver.GetExecutor())
+			maintenanceServices = append(maintenanceServices, sqliteNotificationCleaner)
+		}
+
 		{
 			var scheduleFunc func(time.Time) time.Time
 			if config.ReindexerSchedule != nil {
@@ -2378,8 +2385,6 @@ type JobListResult struct {
 	LastCursor *JobListCursor
 }
 
-const databaseNameSQLite = "sqlite"
-
 var errJobListParamsMetadataNotSupportedSQLite = errors.New("JobListParams.Metadata is not supported on SQLite")
 
 // JobList returns a paginated list of jobs matching the provided filters. The
@@ -2401,7 +2406,7 @@ func (c *Client[TTx]) JobList(ctx context.Context, params *JobListParams) (*JobL
 	}
 	params.schema = c.config.Schema
 
-	if c.driver.DatabaseName() == databaseNameSQLite && params.metadataCalled {
+	if c.driver.DatabaseName() == riverdriver.DatabaseNameSQLite && params.metadataCalled {
 		return nil, errJobListParamsMetadataNotSupportedSQLite
 	}
 
@@ -2442,7 +2447,7 @@ func (c *Client[TTx]) JobListTx(ctx context.Context, tx TTx, params *JobListPara
 	}
 	params.schema = c.config.Schema
 
-	if c.driver.DatabaseName() == databaseNameSQLite && params.metadataCalled {
+	if c.driver.DatabaseName() == riverdriver.DatabaseNameSQLite && params.metadataCalled {
 		return nil, errJobListParamsMetadataNotSupportedSQLite
 	}
 

internal/maintenance/sqlite_notification_cleaner.go

@@ -0,0 +1,152 @@
+package maintenance
+
+import (
+	"cmp"
+	"context"
+	"errors"
+	"log/slog"
+	"time"
+
+	"github.com/riverqueue/river/riverdriver"
+	"github.com/riverqueue/river/rivershared/baseservice"
+	"github.com/riverqueue/river/rivershared/riversharedmaintenance"
+	"github.com/riverqueue/river/rivershared/startstop"
+	"github.com/riverqueue/river/rivershared/testsignal"
+	"github.com/riverqueue/river/rivershared/util/testutil"
+	"github.com/riverqueue/river/rivershared/util/timeutil"
+)
+
+const (
+	SQLiteNotificationCleanerIntervalDefault        = time.Minute
+	SQLiteNotificationCleanerRetentionPeriodDefault = 5 * time.Minute
+)
+
+// SQLiteNotificationCleanerTestSignals are internal signals used exclusively in tests.
+type SQLiteNotificationCleanerTestSignals struct {
+	DeletedBatch testsignal.TestSignal[struct{}] // notifies when runOnce finishes a pass
+}
+
+func (ts *SQLiteNotificationCleanerTestSignals) Init(tb testutil.TestingTB) {
+	ts.DeletedBatch.Init(tb)
+}
+
+type SQLiteNotificationCleanerConfig struct {
+	// Interval is the amount of time to wait between cleaner runs.
+	Interval time.Duration
+
+	// RetentionPeriod is the amount of time to keep notification rows around
+	// before they're removed.
+	RetentionPeriod time.Duration
+
+	// Schema where River tables are located. Empty string omits schema.
+	Schema string
+
+	// Timeout is the timeout for each delete query.
+	Timeout time.Duration
+}
+
+func (c *SQLiteNotificationCleanerConfig) mustValidate() *SQLiteNotificationCleanerConfig {
+	if c.Interval <= 0 {
+		panic("SQLiteNotificationCleanerConfig.Interval must be above zero")
+	}
+	if c.RetentionPeriod <= 0 {
+		panic("SQLiteNotificationCleanerConfig.RetentionPeriod must be above zero")
+	}
+	if c.Timeout <= 0 {
+		panic("SQLiteNotificationCleanerConfig.Timeout must be above zero")
+	}
+
+	return c
+}
+
+// SQLiteNotificationCleaner periodically removes old rows from SQLite's
+// notification outbox. It is only needed for the SQLite driver's emulated
+// listen/notify support.
+type SQLiteNotificationCleaner struct {
+	riversharedmaintenance.QueueMaintainerServiceBase
+	startstop.BaseStartStop
+
+	// exported for test purposes
+	Config      *SQLiteNotificationCleanerConfig
+	TestSignals SQLiteNotificationCleanerTestSignals
+
+	exec riverdriver.Executor
+}
+
+// NewSQLiteNotificationCleaner returns a SQLite notification cleaner.
+func NewSQLiteNotificationCleaner(archetype *baseservice.Archetype, config *SQLiteNotificationCleanerConfig, exec riverdriver.Executor) *SQLiteNotificationCleaner {
+	return baseservice.Init(archetype, &SQLiteNotificationCleaner{
+		Config: (&SQLiteNotificationCleanerConfig{
+			Interval:        cmp.Or(config.Interval, SQLiteNotificationCleanerIntervalDefault),
+			RetentionPeriod: cmp.Or(config.RetentionPeriod, SQLiteNotificationCleanerRetentionPeriodDefault),
+			Schema:          config.Schema,
+			Timeout:         cmp.Or(config.Timeout, riversharedmaintenance.TimeoutDefault),
+		}).mustValidate(),
+		exec: exec,
+	})
+}
+
+func (s *SQLiteNotificationCleaner) Start(ctx context.Context) error { //nolint:dupl
+	ctx, shouldStart, started, stopped := s.StartInit(ctx)
+	if !shouldStart {
+		return nil
+	}
+
+	s.StaggerStart(ctx)
+
+	go func() {
+		started()
+		defer stopped() // this defer should come first so it's last out
+
+		s.Logger.DebugContext(ctx, s.Name+riversharedmaintenance.LogPrefixRunLoopStarted)
+		defer s.Logger.DebugContext(ctx, s.Name+riversharedmaintenance.LogPrefixRunLoopStopped)
+
+		ticker := timeutil.NewTickerWithInitialTick(ctx, s.Config.Interval)
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
+
+			res, err := s.runOnce(ctx)
+			if err != nil {
+				if !errors.Is(err, context.Canceled) {
+					s.Logger.ErrorContext(ctx, s.Name+": Error cleaning SQLite notifications", slog.String("error", err.Error()))
+				}
+				continue
+			}
+
+			if res.NumNotificationsDeleted > 0 {
+				s.Logger.InfoContext(ctx, s.Name+riversharedmaintenance.LogPrefixRanSuccessfully,
+					slog.Int("num_notifications_deleted", res.NumNotificationsDeleted),
+				)
+			}
+		}
+	}()
+
+	return nil
+}
+
+type sqliteNotificationCleanerRunOnceResult struct {
+	NumNotificationsDeleted int
+}
+
+func (s *SQLiteNotificationCleaner) runOnce(ctx context.Context) (*sqliteNotificationCleanerRunOnceResult, error) {
+	ctx, cancelFunc := context.WithTimeout(ctx, s.Config.Timeout)
+	defer cancelFunc()
+
+	numDeleted, err := s.exec.NotificationDeleteBefore(ctx, &riverdriver.NotificationDeleteBeforeParams{
+		CreatedAtHorizon: time.Now().Add(-s.Config.RetentionPeriod),
+		Schema:           s.Config.Schema,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	s.TestSignals.DeletedBatch.Signal(struct{}{})
+
+	return &sqliteNotificationCleanerRunOnceResult{
+		NumNotificationsDeleted: numDeleted,
+	}, nil
+}

internal/maintenance/sqlite_notification_cleaner_test.go

@@ -0,0 +1,105 @@
+package maintenance
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/riverqueue/river/riverdbtest"
+	"github.com/riverqueue/river/riverdriver"
+	"github.com/riverqueue/river/riverdriver/riverpgxv5"
+	"github.com/riverqueue/river/rivershared/riversharedtest"
+	"github.com/riverqueue/river/rivershared/startstoptest"
+)
+
+func TestSQLiteNotificationCleaner(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+
+	type testBundle struct {
+		exec   riverdriver.Executor
+		schema string
+	}
+
+	setup := func(t *testing.T) (*SQLiteNotificationCleaner, *testBundle) {
+		t.Helper()
+
+		driver := riverpgxv5.New(riversharedtest.DBPool(ctx, t))
+		tx, schema := riverdbtest.TestTxPgxDriver(ctx, t, driver, nil)
+
+		bundle := &testBundle{
+			exec:   driver.UnwrapExecutor(tx),
+			schema: schema,
+		}
+
+		cleaner := NewSQLiteNotificationCleaner(
+			riversharedtest.BaseServiceArchetype(t),
+			&SQLiteNotificationCleanerConfig{
+				Interval:        time.Hour,
+				RetentionPeriod: time.Hour,
+				Schema:          bundle.schema,
+				Timeout:         time.Second,
+			},
+			bundle.exec,
+		)
+		cleaner.StaggerStartupDisable(true)
+		t.Cleanup(cleaner.Stop)
+
+		return cleaner, bundle
+	}
+
+	notificationCount := func(t *testing.T, exec riverdriver.Executor) int {
+		t.Helper()
+
+		var count int
+		require.NoError(t, exec.QueryRow(ctx, "SELECT count(*) FROM river_notification").Scan(&count))
+		return count
+	}
+
+	t.Run("Defaults", func(t *testing.T) {
+		t.Parallel()
+
+		cleaner := NewSQLiteNotificationCleaner(
+			riversharedtest.BaseServiceArchetype(t),
+			&SQLiteNotificationCleanerConfig{},
+			nil,
+		)
+
+		require.Equal(t, SQLiteNotificationCleanerIntervalDefault, cleaner.Config.Interval)
+		require.Equal(t, SQLiteNotificationCleanerRetentionPeriodDefault, cleaner.Config.RetentionPeriod)
+	})
+
+	t.Run("DeletesExpiredNotifications", func(t *testing.T) {
+		t.Parallel()
+
+		cleaner, bundle := setup(t)
+		cleaner.TestSignals.Init(t)
+
+		now := time.Now()
+		require.NoError(t, bundle.exec.Exec(ctx, `
+			INSERT INTO river_notification (created_at, payload, topic)
+			VALUES
+				($1, 'old_payload_1', 'topic'),
+				($2, 'old_payload_2', 'topic'),
+				($3, 'new_payload', 'topic')
+		`, now.Add(-2*time.Hour), now.Add(-61*time.Minute), now.Add(-30*time.Minute)))
+
+		res, err := cleaner.runOnce(ctx)
+		require.NoError(t, err)
+		require.Equal(t, 2, res.NumNotificationsDeleted)
+		cleaner.TestSignals.DeletedBatch.WaitOrTimeout()
+		require.Equal(t, 1, notificationCount(t, bundle.exec))
+	})
+
+	t.Run("StartStopStress", func(t *testing.T) {
+		t.Parallel()
+
+		cleaner, _ := setup(t)
+		cleaner.Logger = riversharedtest.LoggerWarn(t) // loop started/stop log is very noisy; suppress
+
+		startstoptest.Stress(ctx, t, cleaner)
+	})
+}

riverdriver/river_driver_interface.go

@@ -24,6 +24,11 @@ import (
 
 const AllQueuesString = "*"
 
+const (
+	DatabaseNamePostgres = "postgres"
+	DatabaseNameSQLite   = "sqlite"
+)
+
 const MigrationLineMain = "main"
 
 var (
@@ -138,12 +143,14 @@ type Driver[TTx any] interface {
 	// API is not stable. DO NOT USE.
 	SupportsListener() bool
 
-	// SupportsListenNotify indicates whether the underlying database supports
-	// listen/notify. This differs from SupportsListener in that even if a
-	// driver doesn't a support a listener but the database supports the
-	// underlying listen/notify mechanism, it will still broadcast in case there
-	// are other clients/drivers on the database that do support a listener. If
-	// listen/notify can't be supported at all, no broadcast attempt is made.
+	// SupportsListenNotify indicates whether the driver can broadcast
+	// notifications that a listener can receive, either through a native
+	// database mechanism like Postgres LISTEN/NOTIFY or a driver-specific
+	// emulation. This differs from SupportsListener in that even if a driver
+	// doesn't support a listener but the database supports the underlying
+	// notification mechanism, it will still broadcast in case there are other
+	// clients/drivers on the database that do support a listener. If
+	// notifications can't be supported at all, no broadcast attempt is made.
 	//
 	// API is not stable. DO NOT USE.
 	SupportsListenNotify() bool
@@ -256,6 +263,14 @@ type Executor interface {
 	// the `line` column was added to the migrations table.
 	MigrationInsertManyAssumingMain(ctx context.Context, params *MigrationInsertManyAssumingMainParams) ([]*Migration, error)
 
+	// NotificationDeleteBefore deletes notifications before a certain time
+	// horizon.
+	//
+	// A "notification" in this context refers to a row in `river_notification`
+	// which is a special table implemented in some databases (e.g. SQLite) that
+	// simulates Postgres' listen/notify when not available.
+	NotificationDeleteBefore(ctx context.Context, params *NotificationDeleteBeforeParams) (int, error)
+
 	NotifyMany(ctx context.Context, params *NotifyManyParams) error
 	PGAdvisoryXactLock(ctx context.Context, key int64) (*struct{}, error)
 
@@ -775,6 +790,11 @@ type NotifyManyParams struct {
 	Schema  string
 }
 
+type NotificationDeleteBeforeParams struct {
+	CreatedAtHorizon time.Time
+	Schema           string
+}
+
 type ProducerKeepAliveParams struct {
 	ID                    int64
 	QueueName             string
@@ -883,8 +903,10 @@ func MigrationLineMainTruncateTables(version int) []string {
 		return []string{"river_job", "river_leader"}
 	case 4:
 		return []string{"river_job", "river_leader", "river_queue"}
-	case 0, 5, 6:
+	case 5, 6:
 		return []string{"river_job", "river_leader", "river_queue", "river_client", "river_client_queue"}
+	case 0, 7:
+		return []string{"river_job", "river_leader", "river_queue", "river_client", "river_client_queue", "river_notification"}
 	}
 
 	panic(fmt.Sprintf("unrecognized migration version: %d", version))