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.

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

Author: brandur

client.go

@@ -992,6 +992,14 @@ 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{
+				DatabaseName: driver.DatabaseName(),
+				Schema:       config.Schema,
+			}, driver.GetExecutor())
+			maintenanceServices = append(maintenanceServices, sqliteNotificationCleaner)
+		}
+
 		{
 			var scheduleFunc func(time.Time) time.Time
 			if config.ReindexerSchedule != nil {
@@ -2378,8 +2386,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 +2407,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 +2448,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,172 @@
+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 {
+	// DatabaseName is the name returned by the active driver. The service only
+	// starts when this is SQLite.
+	DatabaseName string
+
+	// 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 notificationDeleter
+}
+
+func NewSQLiteNotificationCleaner(archetype *baseservice.Archetype, config *SQLiteNotificationCleanerConfig, exec riverdriver.Executor) *SQLiteNotificationCleaner {
+	return newSQLiteNotificationCleaner(archetype, config, exec)
+}
+
+type notificationDeleter interface {
+	NotificationDeleteBefore(ctx context.Context, params *riverdriver.NotificationDeleteBeforeParams) (int, error)
+}
+
+func newSQLiteNotificationCleaner(archetype *baseservice.Archetype, config *SQLiteNotificationCleanerConfig, exec notificationDeleter) *SQLiteNotificationCleaner {
+	return baseservice.Init(archetype, &SQLiteNotificationCleaner{
+		Config: (&SQLiteNotificationCleanerConfig{
+			DatabaseName:    config.DatabaseName,
+			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 {
+	ctx, shouldStart, started, stopped := s.StartInit(ctx)
+	if !shouldStart {
+		return nil
+	}
+
+	// This service only applies to SQLite's notification outbox. If it's
+	// accidentally wired into another driver, leave it inert.
+	if s.Config.DatabaseName != riverdriver.DatabaseNameSQLite {
+		started()
+		stopped()
+		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,131 @@
+package maintenance
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/riverqueue/river/riverdriver"
+	"github.com/riverqueue/river/rivershared/riversharedtest"
+	"github.com/riverqueue/river/rivershared/startstoptest"
+)
+
+// Notably, these tests are a little unusual in that unlike other maintenance
+// services, they depend on a fake and don't execute real database operations.
+// We did this because we'd otherwise have to add a SQLite dependency to the
+// top-level project because it's only under the SQLite driver that we create
+// `river_notification`. The maintenance service is straightforward the database
+// operation exercised in driver tests, so this should be good enough to vet it.
+func TestSQLiteNotificationCleaner(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+
+	setup := func(t *testing.T) (*SQLiteNotificationCleaner, *sqliteNotificationCleanerExecutorMock) {
+		t.Helper()
+
+		exec := &sqliteNotificationCleanerExecutorMock{}
+
+		cleaner := newSQLiteNotificationCleaner(
+			riversharedtest.BaseServiceArchetype(t),
+			&SQLiteNotificationCleanerConfig{
+				DatabaseName:    riverdriver.DatabaseNameSQLite,
+				Interval:        time.Hour,
+				RetentionPeriod: time.Hour,
+				Timeout:         time.Second,
+			},
+			exec,
+		)
+		cleaner.StaggerStartupDisable(true)
+		t.Cleanup(cleaner.Stop)
+
+		return cleaner, exec
+	}
+
+	t.Run("Defaults", func(t *testing.T) {
+		t.Parallel()
+
+		cleaner := NewSQLiteNotificationCleaner(
+			riversharedtest.BaseServiceArchetype(t),
+			&SQLiteNotificationCleanerConfig{DatabaseName: riverdriver.DatabaseNameSQLite},
+			nil,
+		)
+
+		require.Equal(t, riverdriver.DatabaseNameSQLite, cleaner.Config.DatabaseName)
+		require.Equal(t, SQLiteNotificationCleanerIntervalDefault, cleaner.Config.Interval)
+		require.Equal(t, SQLiteNotificationCleanerRetentionPeriodDefault, cleaner.Config.RetentionPeriod)
+	})
+
+	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)
+	})
+
+	t.Run("DoesNotStartForNonSQLite", func(t *testing.T) {
+		t.Parallel()
+
+		cleaner := NewSQLiteNotificationCleaner(
+			riversharedtest.BaseServiceArchetype(t),
+			&SQLiteNotificationCleanerConfig{
+				DatabaseName: riverdriver.DatabaseNamePostgres,
+			},
+			nil,
+		)
+
+		started := cleaner.Started()
+		stopped := cleaner.Stopped()
+		require.NoError(t, cleaner.Start(ctx))
+		riversharedtest.WaitOrTimeout(t, started)
+		riversharedtest.WaitOrTimeout(t, stopped)
+		cleaner.Stop()
+	})
+
+	t.Run("RunOnce", func(t *testing.T) {
+		t.Parallel()
+
+		cleaner, exec := setup(t)
+		cleaner.TestSignals.Init(t)
+
+		res, err := cleaner.runOnce(ctx)
+		require.NoError(t, err)
+		require.Equal(t, sqliteNotificationCleanerExecutorMockNumDeleted, res.NumNotificationsDeleted)
+		cleaner.TestSignals.DeletedBatch.WaitOrTimeout()
+
+		params := exec.notificationDeleteBeforeParams()
+		require.Equal(t, cleaner.Config.Schema, params.Schema)
+		require.WithinDuration(t, time.Now().Add(-cleaner.Config.RetentionPeriod), params.CreatedAtHorizon, 5*time.Second)
+	})
+}
+
+const sqliteNotificationCleanerExecutorMockNumDeleted = 7
+
+type sqliteNotificationCleanerExecutorMock struct {
+	mu                            sync.Mutex
+	notificationDeleteBeforeParam *riverdriver.NotificationDeleteBeforeParams
+}
+
+func (m *sqliteNotificationCleanerExecutorMock) NotificationDeleteBefore(ctx context.Context, params *riverdriver.NotificationDeleteBeforeParams) (int, error) {
+	_ = ctx
+
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	paramsCopy := *params
+	m.notificationDeleteBeforeParam = &paramsCopy
+
+	return sqliteNotificationCleanerExecutorMockNumDeleted, nil
+}
+
+func (m *sqliteNotificationCleanerExecutorMock) notificationDeleteBeforeParams() *riverdriver.NotificationDeleteBeforeParams {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	return m.notificationDeleteBeforeParam
+}

metadata_test.go

@@ -4,8 +4,9 @@ import (
 	"context"
 	"testing"
 
-	"github.com/riverqueue/river/internal/jobexecutor"
 	"github.com/stretchr/testify/require"
+
+	"github.com/riverqueue/river/internal/jobexecutor"
 )
 
 func TestMetadataSet(t *testing.T) {