From 139b11ef03a89538f3e68360c9b1b27a4f5d0261 Mon Sep 17 00:00:00 2001 From: Brandur Date: Thu, 24 Sep 2026 15:20:08 -0500 Subject: [PATCH] A few `river_notification` improvements for SQLite Here, a few minor improvements for `river_notification` inspired somewhat by looking at Active Cable's notifications handling: * Per-topic subscription cursors: Each topic gets its own cursor so that we avoid replaying old notifications in case a client unsubscribes from a topic and later resubscribes. * Notification fetches can read up to 256 notifications at a time (previously we'd only fetch them one at a time). * Cleanup now deletes 10,000 rows (same number as job cleaner + queue cleaner) per batch instead of trying to do all rows beyond a delete horizon. This protects against degenerate cases where some kind of enormous backlog has built up. I honestly thought we were doing the second two already, but apparently not so it's good that we revisited this. --- CHANGELOG.md | 1 + .../sqlite_notification_cleaner.go | 68 +++++-- .../sqlite_notification_cleaner_test.go | 152 ++++++++++++++++ riverdriver/river_driver_interface.go | 5 +- .../internal/dbsqlc/river_notification.sql.go | 17 +- .../river_database_sql_driver.go | 5 +- riverdriver/riverdrivertest/listener.go | 167 ++++++++++++++++++ riverdriver/riverdrivertest/notification.go | 80 +++++++-- .../internal/dbsqlc/river_notification.sql | 8 +- .../internal/dbsqlc/river_notification.sql.go | 17 +- riverdriver/riverpgxv5/river_pgx_v5_driver.go | 5 +- .../internal/dbsqlc/river_notification.sql | 15 +- .../internal/dbsqlc/river_notification.sql.go | 67 +++++-- .../riversqlite/river_sqlite_driver.go | 7 +- .../riversqlite/river_sqlite_listener.go | 110 +++++++----- 15 files changed, 626 insertions(+), 98 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4fd232894..81faa3722 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed SQLite job list pagination skipping or repeating jobs by formatting cursor timestamps consistently with stored timestamps. [PR #1374](https://github.com/riverqueue/river/pull/1374). - Improved PostgreSQL job listing performance when filtering by one finalized state (`completed`, `cancelled`, or `discarded`) and sorting by finalized time, including in River UI. [PR #1374](https://github.com/riverqueue/river/pull/1374). - Fixed `JobRescuer` overwriting jobs that complete, leave the running state, or are claimed again by another worker after being fetched for rescue, preserving their state, errors, metadata, and timestamps across PostgreSQL and SQLite drivers. Fixes [#1302](https://github.com/riverqueue/river/issues/1302). [PR #1373](https://github.com/riverqueue/river/pull/1373). +- Fixed SQLite notification listeners delivering notifications from before a subscription or from an unsubscribe gap. Notification reads now fetch subscribed topics in bounded batches, and cleanup deletes expired notifications in batches of 10,000 rows (reduced to 1,000 after repeated timeouts), with pauses between batches to reduce write lock contention. [PR #1381](https://github.com/riverqueue/river/pull/1381). ## [0.47.0] - 2026-09-01 diff --git a/internal/maintenance/sqlite_notification_cleaner.go b/internal/maintenance/sqlite_notification_cleaner.go index de9e9526d..f5b6fbb59 100644 --- a/internal/maintenance/sqlite_notification_cleaner.go +++ b/internal/maintenance/sqlite_notification_cleaner.go @@ -9,9 +9,12 @@ import ( "github.com/riverqueue/river/riverdriver" "github.com/riverqueue/river/rivershared/baseservice" + "github.com/riverqueue/river/rivershared/circuitbreaker" "github.com/riverqueue/river/rivershared/riversharedmaintenance" "github.com/riverqueue/river/rivershared/startstop" "github.com/riverqueue/river/rivershared/testsignal" + "github.com/riverqueue/river/rivershared/util/randutil" + "github.com/riverqueue/river/rivershared/util/serviceutil" "github.com/riverqueue/river/rivershared/util/testutil" "github.com/riverqueue/river/rivershared/util/timeoututil" "github.com/riverqueue/river/rivershared/util/timeutil" @@ -24,7 +27,7 @@ const ( // SQLiteNotificationCleanerTestSignals are internal signals used exclusively in tests. type SQLiteNotificationCleanerTestSignals struct { - DeletedBatch testsignal.TestSignal[struct{}] // notifies when runOnce finishes a pass + DeletedBatch testsignal.TestSignal[struct{}] // notifies when a delete batch finishes } func (ts *SQLiteNotificationCleanerTestSignals) Init(tb testutil.TestingTB) { @@ -32,6 +35,8 @@ func (ts *SQLiteNotificationCleanerTestSignals) Init(tb testutil.TestingTB) { } type SQLiteNotificationCleanerConfig struct { + riversharedmaintenance.BatchSizes + // Interval is the amount of time to wait between cleaner runs. Interval time.Duration @@ -47,6 +52,8 @@ type SQLiteNotificationCleanerConfig struct { } func (c *SQLiteNotificationCleanerConfig) mustValidate() *SQLiteNotificationCleanerConfig { + c.MustValidate() + if c.Interval <= 0 { panic("SQLiteNotificationCleanerConfig.Interval must be above zero") } @@ -72,18 +79,25 @@ type SQLiteNotificationCleaner struct { TestSignals SQLiteNotificationCleanerTestSignals exec riverdriver.Executor + + // After repeated timeouts, keep using smaller batches until restart. + reducedBatchSizeBreaker *circuitbreaker.CircuitBreaker } // NewSQLiteNotificationCleaner returns a SQLite notification cleaner. func NewSQLiteNotificationCleaner(archetype *baseservice.Archetype, config *SQLiteNotificationCleanerConfig, exec riverdriver.Executor) *SQLiteNotificationCleaner { + batchSizes := config.WithDefaults() + return baseservice.Init(archetype, &SQLiteNotificationCleaner{ Config: (&SQLiteNotificationCleanerConfig{ + BatchSizes: batchSizes, 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, + exec: exec, + reducedBatchSizeBreaker: riversharedmaintenance.ReducedBatchSizeBreaker(batchSizes), }) } @@ -129,24 +143,58 @@ func (s *SQLiteNotificationCleaner) Start(ctx context.Context) error { //nolint: return nil } +func (s *SQLiteNotificationCleaner) batchSize() int { + if s.reducedBatchSizeBreaker.Open() { + return s.Config.Reduced + } + return s.Config.Default +} + type sqliteNotificationCleanerRunOnceResult struct { NumNotificationsDeleted int } func (s *SQLiteNotificationCleaner) runOnce(ctx context.Context) (*sqliteNotificationCleanerRunOnceResult, error) { - return timeoututil.WithTimeoutV(ctx, s.Config.Timeout, s.Name+".runOnce", func(ctx context.Context) (*sqliteNotificationCleanerRunOnceResult, error) { - numDeleted, err := s.exec.NotificationDeleteBefore(ctx, &riverdriver.NotificationDeleteBeforeParams{ - CreatedAtHorizon: time.Now().Add(-s.Config.RetentionPeriod), - Schema: s.Config.Schema, + res := &sqliteNotificationCleanerRunOnceResult{} + // Keep a fixed horizon so new expirations don't extend a cleanup pass. + createdAtHorizon := time.Now().Add(-s.Config.RetentionPeriod) + + for { + if err := ctx.Err(); err != nil { + return nil, err + } + + numDeleted, err := timeoututil.WithTimeoutV(ctx, s.Config.Timeout, s.Name+".runOnce", func(ctx context.Context) (int, error) { + numDeleted, err := s.exec.NotificationDeleteBefore(ctx, &riverdriver.NotificationDeleteBeforeParams{ + CreatedAtHorizon: createdAtHorizon, + Max: s.batchSize(), + Schema: s.Config.Schema, + }) + if err != nil { + return 0, err + } + + s.reducedBatchSizeBreaker.ResetIfNotOpen() + + return numDeleted, nil }) if err != nil { + if errors.Is(err, context.DeadlineExceeded) { + s.reducedBatchSizeBreaker.Trip() + } + return nil, err } s.TestSignals.DeletedBatch.Signal(struct{}{}) + res.NumNotificationsDeleted += numDeleted - return &sqliteNotificationCleanerRunOnceResult{ - NumNotificationsDeleted: numDeleted, - }, nil - }) + if numDeleted < s.batchSize() { + return res, nil + } + + // Each delete commits independently. Yield SQLite's writer lock before + // the next batch so job inserts and updates can make progress. + serviceutil.CancellableSleep(ctx, randutil.DurationBetween(riversharedmaintenance.BatchBackoffMin, riversharedmaintenance.BatchBackoffMax)) + } } diff --git a/internal/maintenance/sqlite_notification_cleaner_test.go b/internal/maintenance/sqlite_notification_cleaner_test.go index 472388500..d05705d7b 100644 --- a/internal/maintenance/sqlite_notification_cleaner_test.go +++ b/internal/maintenance/sqlite_notification_cleaner_test.go @@ -2,6 +2,7 @@ package maintenance import ( "context" + "errors" "testing" "time" @@ -10,10 +11,21 @@ import ( "github.com/riverqueue/river/riverdbtest" "github.com/riverqueue/river/riverdriver" "github.com/riverqueue/river/riverdriver/riverpgxv5" + "github.com/riverqueue/river/rivershared/riversharedmaintenance" "github.com/riverqueue/river/rivershared/riversharedtest" "github.com/riverqueue/river/rivershared/startstoptest" ) +type sqliteNotificationCleanerExecutor struct { + riverdriver.Executor + + notificationDeleteBeforeFunc func(context.Context, *riverdriver.NotificationDeleteBeforeParams) (int, error) +} + +func (e *sqliteNotificationCleanerExecutor) NotificationDeleteBefore(ctx context.Context, params *riverdriver.NotificationDeleteBeforeParams) (int, error) { + return e.notificationDeleteBeforeFunc(ctx, params) +} + func TestSQLiteNotificationCleaner(t *testing.T) { t.Parallel() @@ -59,6 +71,36 @@ func TestSQLiteNotificationCleaner(t *testing.T) { return count } + t.Run("CancelsBetweenBatches", func(t *testing.T) { + t.Parallel() + + cleaner, bundle := setup(t) + cleaner.Config.Default = 2 + cleaner.TestSignals.Init(t) + + require.NoError(t, bundle.exec.Exec(ctx, ` + INSERT INTO river_notification (created_at, payload, topic) + SELECT $1, 'old_payload', 'topic' FROM generate_series(1, 5) + `, time.Now().Add(-2*time.Hour))) + + ctx, cancelFunc := context.WithCancel(ctx) + defer cancelFunc() + cleaner.exec = &sqliteNotificationCleanerExecutor{ + Executor: bundle.exec, + notificationDeleteBeforeFunc: func(ctx context.Context, params *riverdriver.NotificationDeleteBeforeParams) (int, error) { + numDeleted, err := bundle.exec.NotificationDeleteBefore(ctx, params) + cancelFunc() + return numDeleted, err + }, + } + + _, err := cleaner.runOnce(ctx) + require.ErrorIs(t, err, context.Canceled) + cleaner.TestSignals.DeletedBatch.WaitOrTimeout() + cleaner.TestSignals.DeletedBatch.RequireEmpty() + require.Equal(t, 3, notificationCount(t, bundle.exec)) + }) + t.Run("Defaults", func(t *testing.T) { t.Parallel() @@ -68,6 +110,8 @@ func TestSQLiteNotificationCleaner(t *testing.T) { nil, ) + require.Equal(t, riversharedmaintenance.BatchSizeDefault, cleaner.Config.Default) + require.Equal(t, riversharedmaintenance.BatchSizeReduced, cleaner.Config.Reduced) require.Equal(t, SQLiteNotificationCleanerIntervalDefault, cleaner.Config.Interval) require.Equal(t, SQLiteNotificationCleanerRetentionPeriodDefault, cleaner.Config.RetentionPeriod) }) @@ -94,6 +138,114 @@ func TestSQLiteNotificationCleaner(t *testing.T) { require.Equal(t, 1, notificationCount(t, bundle.exec)) }) + t.Run("DeletesMultipleBatches", func(t *testing.T) { + t.Parallel() + + cleaner, bundle := setup(t) + cleaner.Config.Default = 2 + cleaner.TestSignals.Init(t) + + require.NoError(t, bundle.exec.Exec(ctx, ` + INSERT INTO river_notification (created_at, payload, topic) + SELECT $1, 'old_payload', 'topic' FROM generate_series(1, 5) + `, time.Now().Add(-2*time.Hour))) + require.NoError(t, bundle.exec.Exec(ctx, ` + INSERT INTO river_notification (payload, topic) VALUES ('new_payload', 'topic') + `)) + + res, err := cleaner.runOnce(ctx) + require.NoError(t, err) + require.Equal(t, 5, res.NumNotificationsDeleted) + for range 3 { // Two full batches followed by a partial batch. + cleaner.TestSignals.DeletedBatch.WaitOrTimeout() + } + cleaner.TestSignals.DeletedBatch.RequireEmpty() + require.Equal(t, 1, notificationCount(t, bundle.exec)) + }) + + t.Run("ReducedBatchSizeBreakerIgnoresOtherErrors", func(t *testing.T) { + t.Parallel() + + cleaner, bundle := setup(t) + + for _, queryErr := range []error{context.Canceled, errors.New("notification delete failed")} { + cleaner.exec = &sqliteNotificationCleanerExecutor{ + Executor: bundle.exec, + notificationDeleteBeforeFunc: func(context.Context, *riverdriver.NotificationDeleteBeforeParams) (int, error) { + return 0, queryErr + }, + } + + for range cleaner.reducedBatchSizeBreaker.Limit() { + _, err := cleaner.runOnce(ctx) + require.ErrorIs(t, err, queryErr) + } + require.Equal(t, riversharedmaintenance.BatchSizeDefault, cleaner.batchSize()) + } + }) + + t.Run("ReducedBatchSizeBreakerResetsOnSuccess", func(t *testing.T) { + t.Parallel() + + cleaner, bundle := setup(t) + var queryErr error + cleaner.exec = &sqliteNotificationCleanerExecutor{ + Executor: bundle.exec, + notificationDeleteBeforeFunc: func(_ context.Context, params *riverdriver.NotificationDeleteBeforeParams) (int, error) { + require.Equal(t, riversharedmaintenance.BatchSizeDefault, params.Max) + return 0, queryErr + }, + } + + for range 2 { + queryErr = context.DeadlineExceeded + for range cleaner.reducedBatchSizeBreaker.Limit() - 1 { + _, err := cleaner.runOnce(ctx) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Equal(t, riversharedmaintenance.BatchSizeDefault, cleaner.batchSize()) + } + + queryErr = nil + _, err := cleaner.runOnce(ctx) + require.NoError(t, err) + require.Equal(t, riversharedmaintenance.BatchSizeDefault, cleaner.batchSize()) + } + }) + + t.Run("ReducedBatchSizeBreakerTrips", func(t *testing.T) { + t.Parallel() + + cleaner, bundle := setup(t) + expectedMax := riversharedmaintenance.BatchSizeDefault + queryErr := context.DeadlineExceeded + cleaner.exec = &sqliteNotificationCleanerExecutor{ + Executor: bundle.exec, + notificationDeleteBeforeFunc: func(_ context.Context, params *riverdriver.NotificationDeleteBeforeParams) (int, error) { + require.Equal(t, expectedMax, params.Max) + return 0, queryErr + }, + } + + for range cleaner.reducedBatchSizeBreaker.Limit() - 1 { + _, err := cleaner.runOnce(ctx) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Equal(t, riversharedmaintenance.BatchSizeDefault, cleaner.batchSize()) + } + + _, err := cleaner.runOnce(ctx) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Equal(t, riversharedmaintenance.BatchSizeReduced, cleaner.batchSize()) + + // Once tripped, successful deletes keep using the reduced batch size. + expectedMax = riversharedmaintenance.BatchSizeReduced + queryErr = nil + for range 2 { + _, err := cleaner.runOnce(ctx) + require.NoError(t, err) + require.Equal(t, riversharedmaintenance.BatchSizeReduced, cleaner.batchSize()) + } + }) + t.Run("StartStopStress", func(t *testing.T) { t.Parallel() diff --git a/riverdriver/river_driver_interface.go b/riverdriver/river_driver_interface.go index d4b87f315..49c3067b6 100644 --- a/riverdriver/river_driver_interface.go +++ b/riverdriver/river_driver_interface.go @@ -280,8 +280,8 @@ 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. + // NotificationDeleteBefore deletes up to Max notifications before a certain + // time horizon, oldest first. // // 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 @@ -845,6 +845,7 @@ type NotifyManyParams struct { type NotificationDeleteBeforeParams struct { CreatedAtHorizon time.Time + Max int Schema string } diff --git a/riverdriver/riverdatabasesql/internal/dbsqlc/river_notification.sql.go b/riverdriver/riverdatabasesql/internal/dbsqlc/river_notification.sql.go index 33577934f..0fa8a6bfd 100644 --- a/riverdriver/riverdatabasesql/internal/dbsqlc/river_notification.sql.go +++ b/riverdriver/riverdatabasesql/internal/dbsqlc/river_notification.sql.go @@ -12,11 +12,22 @@ import ( const notificationDeleteBefore = `-- name: NotificationDeleteBefore :execrows DELETE FROM /* TEMPLATE: schema */river_notification -WHERE created_at < $1::timestamptz +WHERE id IN ( + SELECT id + FROM /* TEMPLATE: schema */river_notification + WHERE created_at < $1::timestamptz + ORDER BY created_at, id + LIMIT $2::bigint +) ` -func (q *Queries) NotificationDeleteBefore(ctx context.Context, db DBTX, createdAtHorizon time.Time) (int64, error) { - result, err := db.ExecContext(ctx, notificationDeleteBefore, createdAtHorizon) +type NotificationDeleteBeforeParams struct { + CreatedAtHorizon time.Time + Max int64 +} + +func (q *Queries) NotificationDeleteBefore(ctx context.Context, db DBTX, arg *NotificationDeleteBeforeParams) (int64, error) { + result, err := db.ExecContext(ctx, notificationDeleteBefore, arg.CreatedAtHorizon, arg.Max) if err != nil { return 0, err } diff --git a/riverdriver/riverdatabasesql/river_database_sql_driver.go b/riverdriver/riverdatabasesql/river_database_sql_driver.go index 8677d0fd5..07024be99 100644 --- a/riverdriver/riverdatabasesql/river_database_sql_driver.go +++ b/riverdriver/riverdatabasesql/river_database_sql_driver.go @@ -921,7 +921,10 @@ func (e *Executor) MigrationInsertManyAssumingMain(ctx context.Context, params * } func (e *Executor) NotificationDeleteBefore(ctx context.Context, params *riverdriver.NotificationDeleteBeforeParams) (int, error) { - numDeleted, err := dbsqlc.New().NotificationDeleteBefore(schemaTemplateParam(ctx, params.Schema), e.dbtx, params.CreatedAtHorizon) + numDeleted, err := dbsqlc.New().NotificationDeleteBefore(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.NotificationDeleteBeforeParams{ + CreatedAtHorizon: params.CreatedAtHorizon, + Max: int64(params.Max), + }) return int(numDeleted), interpretError(err) } diff --git a/riverdriver/riverdrivertest/listener.go b/riverdriver/riverdrivertest/listener.go index 7a0960758..55d558470 100644 --- a/riverdriver/riverdrivertest/listener.go +++ b/riverdriver/riverdrivertest/listener.go @@ -2,6 +2,7 @@ package riverdrivertest import ( "context" + "fmt" "testing" "time" @@ -70,6 +71,67 @@ func exerciseListener[TTx any](ctx context.Context, t *testing.T, driverWithPool require.NoError(t, listener.Close(ctx)) }) + t.Run("Listen_DoesNotReplayBeforeSubscription", func(t *testing.T) { + t.Parallel() + + listener, bundle := setupListener(ctx, t, driverWithPool) + connectListener(ctx, t, listener) + + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"old"}, Schema: listener.Schema(), Topic: "topic"})) + require.NoError(t, listener.Listen(ctx, "topic")) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"new"}, Schema: listener.Schema(), Topic: "topic"})) + + require.Equal(t, &riverdriver.Notification{Payload: "new", Topic: "topic"}, waitForNotification(ctx, t, listener)) + requireNoNotification(ctx, t, listener) + }) + + t.Run("Listen_NewTopicPreservesOtherTopics", func(t *testing.T) { + t.Parallel() + + listener, bundle := setupListener(ctx, t, driverWithPool) + connectListener(ctx, t, listener) + + require.NoError(t, listener.Listen(ctx, "topic1")) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"pending"}, Schema: listener.Schema(), Topic: "topic1"})) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"old"}, Schema: listener.Schema(), Topic: "topic2"})) + require.NoError(t, listener.Listen(ctx, "topic2")) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"new"}, Schema: listener.Schema(), Topic: "topic2"})) + + require.Equal(t, &riverdriver.Notification{Payload: "pending", Topic: "topic1"}, waitForNotification(ctx, t, listener)) + require.Equal(t, &riverdriver.Notification{Payload: "new", Topic: "topic2"}, waitForNotification(ctx, t, listener)) + requireNoNotification(ctx, t, listener) + }) + + t.Run("Listen_RepeatedPreservesPendingNotifications", func(t *testing.T) { + t.Parallel() + + listener, bundle := setupListener(ctx, t, driverWithPool) + connectListener(ctx, t, listener) + + require.NoError(t, listener.Listen(ctx, "topic")) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"pending"}, Schema: listener.Schema(), Topic: "topic"})) + require.NoError(t, listener.Listen(ctx, "topic")) + + require.Equal(t, &riverdriver.Notification{Payload: "pending", Topic: "topic"}, waitForNotification(ctx, t, listener)) + requireNoNotification(ctx, t, listener) + }) + + t.Run("Listen_ResubscribeSkipsGap", func(t *testing.T) { + t.Parallel() + + listener, bundle := setupListener(ctx, t, driverWithPool) + connectListener(ctx, t, listener) + + require.NoError(t, listener.Listen(ctx, "topic")) + require.NoError(t, listener.Unlisten(ctx, "topic")) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"gap"}, Schema: listener.Schema(), Topic: "topic"})) + require.NoError(t, listener.Listen(ctx, "topic")) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"new"}, Schema: listener.Schema(), Topic: "topic"})) + + require.Equal(t, &riverdriver.Notification{Payload: "new", Topic: "topic"}, waitForNotification(ctx, t, listener)) + requireNoNotification(ctx, t, listener) + }) + t.Run("RoundTrip", func(t *testing.T) { t.Parallel() @@ -190,4 +252,109 @@ func exerciseListener[TTx any](ctx context.Context, t *testing.T, driverWithPool require.NoError(t, listener.Listen(ctx, "topic1")) require.NoError(t, listener.Unlisten(ctx, "topic1")) }) + + t.Run("WaitForNotification_MultipleBatches", func(t *testing.T) { + t.Parallel() + + listener, bundle := setupListener(ctx, t, driverWithPool) + connectListener(ctx, t, listener) + + require.NoError(t, listener.Listen(ctx, "topic")) + payloads := make([]string, 600) // Exceeds the SQLite listener's read batch size. + for i := range payloads { + payloads[i] = fmt.Sprintf("payload_%d", i) + } + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: payloads, Schema: listener.Schema(), Topic: "ignored"})) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: payloads, Schema: listener.Schema(), Topic: "topic"})) + + for _, payload := range payloads { + require.Equal(t, &riverdriver.Notification{Payload: payload, Topic: "topic"}, waitForNotification(ctx, t, listener)) + } + requireNoNotification(ctx, t, listener) + }) + + t.Run("WaitForNotification_ReconnectDiscardsBufferedNotifications", func(t *testing.T) { + t.Parallel() + + listener, bundle := setupListener(ctx, t, driverWithPool) + connectListener(ctx, t, listener) + + require.NoError(t, listener.Listen(ctx, "topic")) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"first", "buffered"}, Schema: listener.Schema(), Topic: "topic"})) + require.Equal(t, "first", waitForNotification(ctx, t, listener).Payload) + + require.NoError(t, listener.Close(ctx)) + require.NoError(t, listener.Connect(ctx)) + require.NoError(t, listener.Listen(ctx, "topic")) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"new"}, Schema: listener.Schema(), Topic: "topic"})) + + require.Equal(t, &riverdriver.Notification{Payload: "new", Topic: "topic"}, waitForNotification(ctx, t, listener)) + requireNoNotification(ctx, t, listener) + }) + + t.Run("WaitForNotification_ResubscribeAfterCleanup", func(t *testing.T) { + t.Parallel() + + listener, bundle := setupListener(ctx, t, driverWithPool) + if bundle.driver.DatabaseName() != riverdriver.DatabaseNameSQLite { + t.Skip("SQLite notifications can be buffered after their outbox rows are deleted") + } + connectListener(ctx, t, listener) + + require.NoError(t, listener.Listen(ctx, "topic")) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"first", "buffered"}, Schema: listener.Schema(), Topic: "topic"})) + require.Equal(t, "first", waitForNotification(ctx, t, listener).Payload) + + require.NoError(t, listener.Unlisten(ctx, "topic")) + require.NoError(t, bundle.exec.Exec(ctx, "DELETE FROM river_notification")) + require.NoError(t, listener.Listen(ctx, "topic")) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"new"}, Schema: listener.Schema(), Topic: "topic"})) + + require.Equal(t, &riverdriver.Notification{Payload: "new", Topic: "topic"}, waitForNotification(ctx, t, listener)) + requireNoNotification(ctx, t, listener) + }) + + t.Run("WaitForNotification_ResubscribeDiscardsBufferedNotifications", func(t *testing.T) { + t.Parallel() + + listener, bundle := setupListener(ctx, t, driverWithPool) + if bundle.driver.DatabaseName() != riverdriver.DatabaseNameSQLite { + t.Skip("SQLite rechecks subscriptions when delivering buffered rows; Postgres can retain already received notifications") + } + connectListener(ctx, t, listener) + + require.NoError(t, listener.Listen(ctx, "topic")) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"first", "buffered"}, Schema: listener.Schema(), Topic: "topic"})) + require.Equal(t, "first", waitForNotification(ctx, t, listener).Payload) + + require.NoError(t, listener.Unlisten(ctx, "topic")) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"gap"}, Schema: listener.Schema(), Topic: "topic"})) + require.NoError(t, listener.Listen(ctx, "topic")) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"new"}, Schema: listener.Schema(), Topic: "topic"})) + + require.Equal(t, &riverdriver.Notification{Payload: "new", Topic: "topic"}, waitForNotification(ctx, t, listener)) + requireNoNotification(ctx, t, listener) + }) + + t.Run("WaitForNotification_UnlistenDiscardsBufferedNotifications", func(t *testing.T) { + t.Parallel() + + listener, bundle := setupListener(ctx, t, driverWithPool) + if bundle.driver.DatabaseName() != riverdriver.DatabaseNameSQLite { + t.Skip("SQLite rechecks subscriptions when delivering buffered rows; Postgres can retain already received notifications") + } + connectListener(ctx, t, listener) + + require.NoError(t, listener.Listen(ctx, "topic1")) + require.NoError(t, listener.Listen(ctx, "topic2")) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"first"}, Schema: listener.Schema(), Topic: "topic1"})) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"buffered"}, Schema: listener.Schema(), Topic: "topic2"})) + require.Equal(t, "first", waitForNotification(ctx, t, listener).Payload) + + require.NoError(t, listener.Unlisten(ctx, "topic2")) + require.NoError(t, bundle.exec.NotifyMany(ctx, &riverdriver.NotifyManyParams{Payload: []string{"new"}, Schema: listener.Schema(), Topic: "topic1"})) + + require.Equal(t, &riverdriver.Notification{Payload: "new", Topic: "topic1"}, waitForNotification(ctx, t, listener)) + requireNoNotification(ctx, t, listener) + }) } diff --git a/riverdriver/riverdrivertest/notification.go b/riverdriver/riverdrivertest/notification.go index 311f77dd9..2f1025084 100644 --- a/riverdriver/riverdrivertest/notification.go +++ b/riverdriver/riverdrivertest/notification.go @@ -13,8 +13,13 @@ import ( func exerciseNotification[TTx any](ctx context.Context, t *testing.T, executorWithTx func(ctx context.Context, t *testing.T) (riverdriver.Executor, riverdriver.Driver[TTx])) { t.Helper() - t.Run("NotificationDeleteBefore", func(t *testing.T) { - t.Parallel() + type testBundle struct { + exec riverdriver.Executor + horizon time.Time + } + + setup := func(ctx context.Context, t *testing.T) *testBundle { + t.Helper() exec, driver := executorWithTx(ctx, t) @@ -23,42 +28,89 @@ func exerciseNotification[TTx any](ctx context.Context, t *testing.T, executorWi VALUES ($1, $2, $3), ($4, $5, $6), - ($7, $8, $9) + ($7, $8, $9), + ($10, $11, $12) ` if driver.DatabaseName() == riverdriver.DatabaseNameSQLite { insertQuery = ` INSERT INTO river_notification (created_at, payload, topic) VALUES + (?, ?, ?), (?, ?, ?), (?, ?, ?), (?, ?, ?) ` } - createdAt := func(t time.Time) any { return t } + createdAtFunc := func(t time.Time) any { return t } if driver.DatabaseName() == riverdriver.DatabaseNameSQLite { // Keep this in the same format that the SQLite driver uses for // CreatedAtHorizon so SQLite's text comparison stays chronological. - createdAt = func(t time.Time) any { - const sqliteFormat = "2006-01-02 15:04:05.999" + createdAtFunc = func(t time.Time) any { + const sqliteFormat = "2006-01-02 15:04:05.000" return t.UTC().Round(time.Millisecond).Format(sqliteFormat) } } - now := time.Now().UTC() + // Include a trailing fractional zero to exercise SQLite's fixed-width format. + now := time.Now().UTC().Truncate(time.Second).Add(120 * time.Millisecond) require.NoError(t, exec.Exec(ctx, insertQuery, - createdAt(now.Add(-2*time.Hour)), "old_payload_1", "topic", - createdAt(now.Add(-61*time.Minute)), "old_payload_2", "topic", - createdAt(now.Add(-30*time.Minute)), "new_payload", "topic", + createdAtFunc(now.Add(-61*time.Minute)), "old_payload", "topic", + createdAtFunc(now.Add(-2*time.Hour)), "oldest_payload", "topic", + createdAtFunc(now.Add(-time.Hour)), "horizon_payload", "topic", + createdAtFunc(now.Add(-30*time.Minute)), "new_payload", "topic", )) - numDeleted, err := exec.NotificationDeleteBefore(ctx, &riverdriver.NotificationDeleteBeforeParams{ - CreatedAtHorizon: now.Add(-time.Hour), + return &testBundle{ + exec: exec, + horizon: now.Add(-time.Hour), + } + } + + t.Run("NotificationDeleteBefore", func(t *testing.T) { + t.Parallel() + + bundle := setup(ctx, t) + + numDeleted, err := bundle.exec.NotificationDeleteBefore(ctx, &riverdriver.NotificationDeleteBeforeParams{ + CreatedAtHorizon: bundle.horizon, + Max: 10, }) require.NoError(t, err) require.Equal(t, 2, numDeleted) var count int - require.NoError(t, exec.QueryRow(ctx, "SELECT count(*) FROM river_notification").Scan(&count)) - require.Equal(t, 1, count) + require.NoError(t, bundle.exec.QueryRow(ctx, "SELECT count(*) FROM river_notification").Scan(&count)) + require.Equal(t, 2, count) + }) + + t.Run("NotificationDeleteBefore_Limited", func(t *testing.T) { + t.Parallel() + + bundle := setup(ctx, t) + params := &riverdriver.NotificationDeleteBeforeParams{ + CreatedAtHorizon: bundle.horizon, + Max: 1, + } + + numDeleted, err := bundle.exec.NotificationDeleteBefore(ctx, params) + require.NoError(t, err) + require.Equal(t, 1, numDeleted) + + // Delete by age, even when the oldest notification was inserted later. + var oldestRemaining string + require.NoError(t, bundle.exec.QueryRow(ctx, "SELECT payload FROM river_notification ORDER BY created_at LIMIT 1").Scan(&oldestRemaining)) + require.Equal(t, "old_payload", oldestRemaining) + + numDeleted, err = bundle.exec.NotificationDeleteBefore(ctx, params) + require.NoError(t, err) + require.Equal(t, 1, numDeleted) + + numDeleted, err = bundle.exec.NotificationDeleteBefore(ctx, params) + require.NoError(t, err) + require.Zero(t, numDeleted) + + var count int + require.NoError(t, bundle.exec.QueryRow(ctx, "SELECT count(*) FROM river_notification").Scan(&count)) + require.Equal(t, 2, count) // Includes the notification exactly at the horizon. }) } diff --git a/riverdriver/riverpgxv5/internal/dbsqlc/river_notification.sql b/riverdriver/riverpgxv5/internal/dbsqlc/river_notification.sql index 576d2444d..ac2611dd5 100644 --- a/riverdriver/riverpgxv5/internal/dbsqlc/river_notification.sql +++ b/riverdriver/riverpgxv5/internal/dbsqlc/river_notification.sql @@ -13,4 +13,10 @@ CREATE TABLE river_notification ( -- name: NotificationDeleteBefore :execrows DELETE FROM /* TEMPLATE: schema */river_notification -WHERE created_at < @created_at_horizon::timestamptz; +WHERE id IN ( + SELECT id + FROM /* TEMPLATE: schema */river_notification + WHERE created_at < @created_at_horizon::timestamptz + ORDER BY created_at, id + LIMIT @max::bigint +); diff --git a/riverdriver/riverpgxv5/internal/dbsqlc/river_notification.sql.go b/riverdriver/riverpgxv5/internal/dbsqlc/river_notification.sql.go index cb460451c..da9f16f0b 100644 --- a/riverdriver/riverpgxv5/internal/dbsqlc/river_notification.sql.go +++ b/riverdriver/riverpgxv5/internal/dbsqlc/river_notification.sql.go @@ -12,11 +12,22 @@ import ( const notificationDeleteBefore = `-- name: NotificationDeleteBefore :execrows DELETE FROM /* TEMPLATE: schema */river_notification -WHERE created_at < $1::timestamptz +WHERE id IN ( + SELECT id + FROM /* TEMPLATE: schema */river_notification + WHERE created_at < $1::timestamptz + ORDER BY created_at, id + LIMIT $2::bigint +) ` -func (q *Queries) NotificationDeleteBefore(ctx context.Context, db DBTX, createdAtHorizon time.Time) (int64, error) { - result, err := db.Exec(ctx, notificationDeleteBefore, createdAtHorizon) +type NotificationDeleteBeforeParams struct { + CreatedAtHorizon time.Time + Max int64 +} + +func (q *Queries) NotificationDeleteBefore(ctx context.Context, db DBTX, arg *NotificationDeleteBeforeParams) (int64, error) { + result, err := db.Exec(ctx, notificationDeleteBefore, arg.CreatedAtHorizon, arg.Max) if err != nil { return 0, err } diff --git a/riverdriver/riverpgxv5/river_pgx_v5_driver.go b/riverdriver/riverpgxv5/river_pgx_v5_driver.go index 62601230d..5167ff6c6 100644 --- a/riverdriver/riverpgxv5/river_pgx_v5_driver.go +++ b/riverdriver/riverpgxv5/river_pgx_v5_driver.go @@ -867,7 +867,10 @@ func (e *Executor) MigrationInsertManyAssumingMain(ctx context.Context, params * } func (e *Executor) NotificationDeleteBefore(ctx context.Context, params *riverdriver.NotificationDeleteBeforeParams) (int, error) { - numDeleted, err := dbsqlc.New().NotificationDeleteBefore(schemaTemplateParam(ctx, params.Schema), e.dbtx, params.CreatedAtHorizon) + numDeleted, err := dbsqlc.New().NotificationDeleteBefore(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.NotificationDeleteBeforeParams{ + CreatedAtHorizon: params.CreatedAtHorizon, + Max: int64(params.Max), + }) return int(numDeleted), interpretError(err) } diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_notification.sql b/riverdriver/riversqlite/internal/dbsqlc/river_notification.sql index ddaf3d707..f1e70ac31 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_notification.sql +++ b/riverdriver/riversqlite/internal/dbsqlc/river_notification.sql @@ -8,14 +8,21 @@ CREATE TABLE river_notification ( -- name: NotificationDeleteBefore :execrows DELETE FROM /* TEMPLATE: schema */river_notification -WHERE created_at < cast(@created_at_horizon AS text); +WHERE id IN ( + SELECT id + FROM /* TEMPLATE: schema */river_notification + WHERE created_at < cast(@created_at_horizon AS text) + ORDER BY created_at, id + LIMIT @max +); --- name: NotificationGetAfter :one -SELECT * +-- name: NotificationGetAfter :many +SELECT id, payload, topic FROM /* TEMPLATE: schema */river_notification WHERE id > @after + AND topic IN (SELECT value FROM json_each(cast(@topics AS blob))) ORDER BY id ASC -LIMIT 1; +LIMIT @max; -- name: NotificationGetLastID :one SELECT cast(coalesce(max(id), 0) AS integer) diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_notification.sql.go b/riverdriver/riversqlite/internal/dbsqlc/river_notification.sql.go index 13eb7deb3..bead11a8e 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_notification.sql.go +++ b/riverdriver/riversqlite/internal/dbsqlc/river_notification.sql.go @@ -11,35 +11,70 @@ import ( const notificationDeleteBefore = `-- name: NotificationDeleteBefore :execrows DELETE FROM /* TEMPLATE: schema */river_notification -WHERE created_at < cast(?1 AS text) +WHERE id IN ( + SELECT id + FROM /* TEMPLATE: schema */river_notification + WHERE created_at < cast(?1 AS text) + ORDER BY created_at, id + LIMIT ?2 +) ` -func (q *Queries) NotificationDeleteBefore(ctx context.Context, db DBTX, createdAtHorizon string) (int64, error) { - result, err := db.ExecContext(ctx, notificationDeleteBefore, createdAtHorizon) +type NotificationDeleteBeforeParams struct { + CreatedAtHorizon string + Max int64 +} + +func (q *Queries) NotificationDeleteBefore(ctx context.Context, db DBTX, arg *NotificationDeleteBeforeParams) (int64, error) { + result, err := db.ExecContext(ctx, notificationDeleteBefore, arg.CreatedAtHorizon, arg.Max) if err != nil { return 0, err } return result.RowsAffected() } -const notificationGetAfter = `-- name: NotificationGetAfter :one -SELECT id, created_at, payload, topic +const notificationGetAfter = `-- name: NotificationGetAfter :many +SELECT id, payload, topic FROM /* TEMPLATE: schema */river_notification WHERE id > ?1 + AND topic IN (SELECT value FROM json_each(cast(?2 AS blob))) ORDER BY id ASC -LIMIT 1 +LIMIT ?3 ` -func (q *Queries) NotificationGetAfter(ctx context.Context, db DBTX, after int64) (*RiverNotification, error) { - row := db.QueryRowContext(ctx, notificationGetAfter, after) - var i RiverNotification - err := row.Scan( - &i.ID, - &i.CreatedAt, - &i.Payload, - &i.Topic, - ) - return &i, err +type NotificationGetAfterParams struct { + After int64 + Topics []byte + Max int64 +} + +type NotificationGetAfterRow struct { + ID int64 + Payload string + Topic string +} + +func (q *Queries) NotificationGetAfter(ctx context.Context, db DBTX, arg *NotificationGetAfterParams) ([]*NotificationGetAfterRow, error) { + rows, err := db.QueryContext(ctx, notificationGetAfter, arg.After, arg.Topics, arg.Max) + if err != nil { + return nil, err + } + defer rows.Close() + var items []*NotificationGetAfterRow + for rows.Next() { + var i NotificationGetAfterRow + if err := rows.Scan(&i.ID, &i.Payload, &i.Topic); err != nil { + return nil, err + } + items = append(items, &i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil } const notificationGetLastID = `-- name: NotificationGetLastID :one diff --git a/riverdriver/riversqlite/river_sqlite_driver.go b/riverdriver/riversqlite/river_sqlite_driver.go index a6b0264d9..58bcf1631 100644 --- a/riverdriver/riversqlite/river_sqlite_driver.go +++ b/riverdriver/riversqlite/river_sqlite_driver.go @@ -89,7 +89,7 @@ func (d *Driver) GetListener(params *riverdriver.GetListenenerParams) riverdrive pollInterval: notificationPollIntervalDefault, replacer: &d.replacer, schema: params.Schema, - topics: make(map[string]struct{}), + topics: make(map[string]int64), } } @@ -1178,7 +1178,10 @@ func (e *Executor) NotificationDeleteBefore(ctx context.Context, params *riverdr numDeleted, err := dbsqlc.New().NotificationDeleteBefore( schemaTemplateParam(ctx, params.Schema), e.dbtx, - timeString(params.CreatedAtHorizon), + &dbsqlc.NotificationDeleteBeforeParams{ + CreatedAtHorizon: timeString(params.CreatedAtHorizon), + Max: int64(params.Max), + }, ) return int(numDeleted), interpretError(err) } diff --git a/riverdriver/riversqlite/river_sqlite_listener.go b/riverdriver/riversqlite/river_sqlite_listener.go index 444c21f9d..fe5f31870 100644 --- a/riverdriver/riversqlite/river_sqlite_listener.go +++ b/riverdriver/riversqlite/river_sqlite_listener.go @@ -3,6 +3,7 @@ package riversqlite import ( "context" "database/sql" + "encoding/json" "errors" "sync" "time" @@ -10,9 +11,11 @@ import ( "github.com/riverqueue/river/riverdriver" "github.com/riverqueue/river/riverdriver/riversqlite/internal/dbsqlc" "github.com/riverqueue/river/rivershared/sqlctemplate" + "github.com/riverqueue/river/rivershared/util/maputil" ) const ( + notificationBatchSize = 256 notificationPollIntervalDefault = 50 * time.Millisecond ) @@ -20,9 +23,9 @@ const ( // table. SQLite doesn't have a native LISTEN/NOTIFY equivalent, so NotifyMany // appends rows to river_notification and this listener polls for rows with IDs // greater than its remembered lastID. The lastID marker is initialized to the -// current max ID on connect so historical rows aren't replayed, and advances -// past every observed row so unlistened topics don't get delivered later if -// they're re-listened. +// current max ID on connect so historical rows aren't replayed. Each topic also +// records the max ID at subscription time so earlier rows aren't delivered, +// including buffered rows from a previous subscription to the same topic. type Listener struct { afterConnectExec string // should only ever be used in testing dbPool *sql.DB @@ -36,10 +39,11 @@ type Listener struct { lastID int64 mu sync.Mutex + pending []*dbsqlc.NotificationGetAfterRow pollInterval time.Duration replacer *sqlctemplate.Replacer schema string - topics map[string]struct{} + topics map[string]int64 } type notificationPayload struct { @@ -52,6 +56,8 @@ func (l *Listener) Close(context.Context) error { defer l.mu.Unlock() l.isConnected = false + l.pending = nil + l.topics = nil return nil } @@ -105,19 +111,32 @@ func (l *Listener) Connect(ctx context.Context) error { return nil } -func (l *Listener) Listen(_ context.Context, topic string) error { +func (l *Listener) Listen(ctx context.Context, topic string) error { l.mu.Lock() defer l.mu.Unlock() if !l.isConnected { return errors.New("listener is not connected") } + if _, ok := l.topics[topic]; ok { + return nil + } + + // Only advance this topic's starting ID. Updating the global cursor here + // could skip pending messages for topics that are already subscribed. + lastID, err := dbsqlc.New().NotificationGetLastID( + schemaTemplateParam(ctx, l.schema), notificationDBTX(l.dbPool, l.replacer), + ) + if err != nil { + return err + } if l.topics == nil { - l.topics = make(map[string]struct{}) + l.topics = make(map[string]int64) } - l.topics[topic] = struct{}{} + // Cleanup may have removed rows that are still buffered locally. + l.topics[topic] = max(l.lastID, lastID) return nil } @@ -215,52 +234,61 @@ func (l *Listener) waitForNextPoll(ctx context.Context) error { } func (l *Listener) waitForNotificationOnce(ctx context.Context) (*riverdriver.Notification, bool, error) { - var ( - after int64 - dbPool *sql.DB - replacer *sqlctemplate.Replacer - schema string - ) - l.mu.Lock() + defer l.mu.Unlock() + if !l.isConnected { - l.mu.Unlock() return nil, false, errors.New("listener is not connected") } - after = l.lastID - dbPool = l.dbPool - replacer = l.replacer - schema = l.schema - l.mu.Unlock() - - if dbPool == nil { + if l.dbPool == nil { return nil, false, errors.New("database pool is nil") } - notification, err := dbsqlc.New().NotificationGetAfter( - schemaTemplateParam(ctx, schema), - notificationDBTX(dbPool, replacer), - after, - ) - if err != nil { - return nil, false, err - } - - l.mu.Lock() - defer l.mu.Unlock() + if len(l.pending) == 0 { + if len(l.topics) == 0 { + return nil, false, sql.ErrNoRows + } + topics, err := json.Marshal(maputil.Keys(l.topics)) + if err != nil { + return nil, false, err + } - if notification.ID > l.lastID { - l.lastID = notification.ID + // Serialize the fetch with subscription changes and connection resets, + // but release the lock between calls and while waiting for the next poll. + notifications, err := dbsqlc.New().NotificationGetAfter( + schemaTemplateParam(ctx, l.schema), + notificationDBTX(l.dbPool, l.replacer), + &dbsqlc.NotificationGetAfterParams{ + After: l.lastID, + Max: notificationBatchSize, + Topics: topics, + }, + ) + if err != nil { + return nil, false, err + } + if len(notifications) == 0 { + return nil, false, sql.ErrNoRows + } + l.pending = notifications + l.lastID = notifications[len(notifications)-1].ID } - if _, ok := l.topics[notification.Topic]; !ok { - return nil, false, nil + for len(l.pending) > 0 { + notification := l.pending[0] + l.pending[0] = nil + l.pending = l.pending[1:] + + // Subscriptions can change between fetching a batch and delivering it. + if startID, ok := l.topics[notification.Topic]; ok && notification.ID > startID { + return &riverdriver.Notification{ + Payload: notification.Payload, + Topic: notification.Topic, + }, true, nil + } } - return &riverdriver.Notification{ - Payload: notification.Payload, - Topic: notification.Topic, - }, true, nil + return nil, false, nil } func notificationDBTX(dbPool *sql.DB, replacer *sqlctemplate.Replacer) templateReplaceWrapper {