From cc3d1366ace8289be1ddea85a2708c028ffeb8f3 Mon Sep 17 00:00:00 2001 From: Yaomeng1749 <176285760+Yaomeng1749@users.noreply.github.com> Date: Wed, 28 Oct 2026 16:10:17 -0700 Subject: [PATCH 1/3] [go] Fix `InsertManyFast` panic on drivers without listener support `InsertManyFast` inserts with `JobInsertFastManyNoReturning`, so its insert results are all nil, but it passed them to `notifyProducerWithoutListenerJobFetch`, which reads `Job.Queue` from each one. On a client with producers (one configured with `Queues`) and a driver that doesn't support a listener, `InsertManyFast` panicked after its transaction had already committed. Notify using the queues from insert params instead. A new `notifyProducerWithoutListenerJobFetchForQueues` takes queue names, and `notifyProducerWithoutListenerJobFetch` keeps its signature because it's handed to the pilot as `NotifyNonTxJobInsert`. To get at insert params, `insertManyFast` now takes `[]*rivertype.JobInsertParams` like `insertMany` does, and its callers build them first. Signed-off-by: Yaomeng1749 <176285760+Yaomeng1749@users.noreply.github.com> --- CHANGELOG.md | 4 ++++ client.go | 48 ++++++++++++++++++++++++++++++++-------------- client_test.go | 52 ++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 90 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 73864e81b..b86d9d751 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 + +- Fixed `Client.InsertManyFast` panicking after its transaction committed on a client configured with queues when the driver doesn't support a listener, such as a `riverdatabasesql` driver created with `New` instead of `NewWithPgxListener`. It now wakes producers for the inserted jobs' queues right away, like `Client.InsertMany`. [PR #1558](https://github.com/riverqueue/river/pull/1558). + ## [0.50.0] - 2026-10-26 ### Changed diff --git a/client.go b/client.go index e06af4b0c..b1c17308a 100644 --- a/client.go +++ b/client.go @@ -2157,6 +2157,17 @@ func (c *Client[TTx]) notifyProducerWithoutListenerJobFetch(_ context.Context, r return } + c.notifyProducerWithoutListenerJobFetchForQueues(sliceutil.Map(res, func(insertRes *rivertype.JobInsertResult) string { return insertRes.Job.Queue })) +} + +// Same as notifyProducerWithoutListenerJobFetch, but takes the queues that jobs +// were inserted into instead of insert results. Used by InsertManyFast, whose +// insert doesn't return job rows. +func (c *Client[TTx]) notifyProducerWithoutListenerJobFetchForQueues(queues []string) { + if c.driver.SupportsListener() { + return + } + c.producersMu.RLock() defer c.producersMu.RUnlock() @@ -2166,8 +2177,8 @@ func (c *Client[TTx]) notifyProducerWithoutListenerJobFetch(_ context.Context, r // Special case for when we were handling exactly one job, which is a very // common case. Acts as a minor optimization by avoiding the map allocation. - if len(res) == 1 { - if producer, ok := c.producersByQueueName[res[0].Job.Queue]; ok { + if len(queues) == 1 { + if producer, ok := c.producersByQueueName[queues[0]]; ok { producer.TriggerJobFetch() } @@ -2176,13 +2187,13 @@ func (c *Client[TTx]) notifyProducerWithoutListenerJobFetch(_ context.Context, r queuesTriggered := make(map[string]struct{}) - for _, insertRes := range res { - if _, ok := queuesTriggered[insertRes.Job.Queue]; ok { + for _, queue := range queues { + if _, ok := queuesTriggered[queue]; ok { continue } - queuesTriggered[insertRes.Job.Queue] = struct{}{} + queuesTriggered[queue] = struct{}{} - if producer, ok := c.producersByQueueName[insertRes.Job.Queue]; ok { + if producer, ok := c.producersByQueueName[queue]; ok { producer.TriggerJobFetch() } } @@ -2211,15 +2222,24 @@ func (c *Client[TTx]) InsertManyFast(ctx context.Context, params []InsertManyPar return 0, errNoDriverDBPool } + insertParams, err := c.insertManyParams(params) + if err != nil { + return 0, err + } + // Wrap in a transaction in case we need to notify about inserts. res, err := dbutil.WithTxV(ctx, c.driver.GetExecutor(), func(ctx context.Context, execTx riverdriver.ExecutorTx) ([]*rivertype.JobInsertResult, error) { - return c.insertManyFast(ctx, execTx, params) + return c.insertManyFast(ctx, execTx, insertParams) }) if err != nil { return 0, err } - c.notifyProducerWithoutListenerJobFetch(ctx, res) + // Fast inserts don't return job rows, so notify for the queues in insert + // params instead. + if !c.driver.SupportsListener() { + c.notifyProducerWithoutListenerJobFetchForQueues(sliceutil.Map(insertParams, func(insertParam *rivertype.JobInsertParams) string { return insertParam.Queue })) + } return len(res), nil } @@ -2248,19 +2268,19 @@ func (c *Client[TTx]) InsertManyFast(ctx context.Context, params []InsertManyPar // a unique constraint is violated, the operation will fail and no jobs will be // inserted. func (c *Client[TTx]) InsertManyFastTx(ctx context.Context, tx TTx, params []InsertManyParams) (int, error) { - res, err := c.insertManyFast(ctx, c.driver.UnwrapExecutor(tx), params) + insertParams, err := c.insertManyParams(params) if err != nil { return 0, err } - return len(res), nil -} -func (c *Client[TTx]) insertManyFast(ctx context.Context, execTx riverdriver.ExecutorTx, params []InsertManyParams) ([]*rivertype.JobInsertResult, error) { - insertParams, err := c.insertManyParams(params) + res, err := c.insertManyFast(ctx, c.driver.UnwrapExecutor(tx), insertParams) if err != nil { - return nil, err + return 0, err } + return len(res), nil +} +func (c *Client[TTx]) insertManyFast(ctx context.Context, execTx riverdriver.ExecutorTx, insertParams []*rivertype.JobInsertParams) ([]*rivertype.JobInsertResult, error) { return c.insertManyShared(ctx, execTx, insertParams, func(ctx context.Context, insertParams []*riverdriver.JobInsertFastParams) ([]*rivertype.JobInsertResult, error) { count, err := execTx.JobInsertFastManyNoReturning(ctx, &riverdriver.JobInsertFastManyParams{ Jobs: insertParams, diff --git a/client_test.go b/client_test.go index 472bac298..107531cf2 100644 --- a/client_test.go +++ b/client_test.go @@ -4340,6 +4340,58 @@ func Test_Client_InsertManyFast(t *testing.T) { require.NoError(t, client.Stop(ctx)) }) + t.Run("ProducerFetchLimiterCalled", func(t *testing.T) { + t.Parallel() + + var ( + dbPool = riversharedtest.DBPoolClone(ctx, t) + driver = NewDriverWithoutListenNotify(dbPool) + schema = riverdbtest.TestSchema(ctx, t, driver, nil) + config = newTestConfig(t, schema) + ) + + config.Queues["another_queue"] = QueueConfig{MaxWorkers: 1} + + client, err := NewClient(driver, config) + require.NoError(t, err) + client.producersByQueueName[QueueDefault].testSignals.Init(t) + client.producersByQueueName["another_queue"].testSignals.Init(t) + + startClient(ctx, t, client) + + count, err := client.InsertManyFast(ctx, []InsertManyParams{ + {Args: noOpArgs{}}, + {Args: noOpArgs{}}, + {Args: noOpArgs{}, InsertOpts: &InsertOpts{Queue: "another_queue"}}, + }) + require.NoError(t, err) + require.Equal(t, 3, count) + + // Each queue that received jobs is triggered once. + client.producersByQueueName[QueueDefault].testSignals.JobFetchTriggered.WaitOrTimeout() + client.producersByQueueName[QueueDefault].testSignals.JobFetchTriggered.RequireEmpty() + client.producersByQueueName["another_queue"].testSignals.JobFetchTriggered.WaitOrTimeout() + client.producersByQueueName["another_queue"].testSignals.JobFetchTriggered.RequireEmpty() + }) + + // Not called for drivers that support a listener. + t.Run("ProducerFetchLimiterNotCalled", func(t *testing.T) { + t.Parallel() + + client, _ := setup(t) + client.producersByQueueName[QueueDefault].testSignals.Init(t) + + startClient(ctx, t, client) + + _, err := client.InsertManyFast(ctx, []InsertManyParams{ + {Args: noOpArgs{}}, + {Args: noOpArgs{}}, + }) + require.NoError(t, err) + + client.producersByQueueName[QueueDefault].testSignals.JobFetchTriggered.RequireEmpty() + }) + t.Run("DoesNotTriggerInsertNotificationForNonAvailableJob", func(t *testing.T) { t.Parallel() From 6123852fecee6c92c01faa979dff6fa6d082560c Mon Sep 17 00:00:00 2001 From: Yaomeng1749 <176285760+Yaomeng1749@users.noreply.github.com> Date: Wed, 28 Oct 2026 16:10:32 -0700 Subject: [PATCH 2/3] [go] Don't send a nil queue control event from `QueueUpdate` `queueUpdate` returns a nil control event when metadata isn't being changed, but `QueueUpdate` passed it on to the producer anyway when the driver doesn't support a listener. The producer's fetch loop then dereferenced it, and with nothing recovering in that goroutine, the program exited. Only notify the producer when there's a control event. Signed-off-by: Yaomeng1749 <176285760+Yaomeng1749@users.noreply.github.com> --- CHANGELOG.md | 1 + client.go | 6 +++++- client_test.go | 37 +++++++++++++++++++++++++++++++++++++ 3 files changed, 43 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b86d9d751..1b66a6430 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - Fixed `Client.InsertManyFast` panicking after its transaction committed on a client configured with queues when the driver doesn't support a listener, such as a `riverdatabasesql` driver created with `New` instead of `NewWithPgxListener`. It now wakes producers for the inserted jobs' queues right away, like `Client.InsertMany`. [PR #1558](https://github.com/riverqueue/river/pull/1558). +- Fixed `Client.QueueUpdate` without metadata crashing the program when the client works that queue and the driver doesn't support a listener. [PR #1558](https://github.com/riverqueue/river/pull/1558). ## [0.50.0] - 2026-10-26 diff --git a/client.go b/client.go index b1c17308a..48399c26a 100644 --- a/client.go +++ b/client.go @@ -2963,7 +2963,11 @@ func (c *Client[TTx]) QueueUpdate(ctx context.Context, name string, params *Queu return nil, err } - c.notifyProducerWithoutListenerQueueControlEvent(name, controlEvent) + // No control event is returned when metadata isn't being changed, in which + // case there's nothing for the producer to act on. + if controlEvent != nil { + c.notifyProducerWithoutListenerQueueControlEvent(name, controlEvent) + } return queue, nil } diff --git a/client_test.go b/client_test.go index 107531cf2..8ee93b6b2 100644 --- a/client_test.go +++ b/client_test.go @@ -7372,6 +7372,43 @@ func Test_Client_QueueUpdate(t *testing.T) { client.producersByQueueName[QueueDefault].testSignals.QueueControlEventTriggered.RequireEmpty() }) + + t.Run("ProducerControlEventNotSentWithoutMetadata", func(t *testing.T) { + t.Parallel() + + var ( + dbPool = riversharedtest.DBPoolClone(ctx, t) + driver = NewDriverWithoutListenNotify(dbPool) + schema = riverdbtest.TestSchema(ctx, t, driver, nil) + config = newTestConfig(t, schema) + ) + + client, err := NewClient(driver, config) + require.NoError(t, err) + client.producersByQueueName[QueueDefault].testSignals.Init(t) + + startClient(ctx, t, client) + + // Metadata isn't changed, so there's no control event to send. + _, err = client.QueueUpdate(ctx, QueueDefault, &QueueUpdateParams{ + Metadata: nil, + }) + require.NoError(t, err) + + client.producersByQueueName[QueueDefault].testSignals.QueueControlEventTriggered.RequireEmpty() + + // The producer is still running and handles a later metadata change. + _, err = client.QueueUpdate(ctx, QueueDefault, &QueueUpdateParams{ + Metadata: []byte(`{"foo":"baz"}`), + }) + require.NoError(t, err) + + controlEvent := client.producersByQueueName[QueueDefault].testSignals.QueueControlEventTriggered.WaitOrTimeout() + require.NotNil(t, controlEvent) + require.Equal(t, controlActionMetadataChanged, controlEvent.Action) + + client.producersByQueueName[QueueDefault].testSignals.MetadataChanged.WaitOrTimeout() + }) } func Test_Client_QueueUpdateTx(t *testing.T) { From 36a2f7febfee82175cc7e7c9073e72a792f4b9aa Mon Sep 17 00:00:00 2001 From: Brandur Date: Thu, 29 Oct 2026 10:48:07 -0500 Subject: [PATCH 3/3] Small optimization to avoid allocating queue slice A small optimization on top of #1558 so that in case of an insertion-only client, skip unnecessary queue slice allocation. --- client.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/client.go b/client.go index 48399c26a..e0abd8956 100644 --- a/client.go +++ b/client.go @@ -2153,7 +2153,7 @@ func (c *Client[TTx]) insertManyParams(params []InsertManyParams) ([]*rivertype. // transaction, the producer wouldn't yet be able to access the new jobs that // triggered the notification because they're not committed yet. func (c *Client[TTx]) notifyProducerWithoutListenerJobFetch(_ context.Context, res []*rivertype.JobInsertResult) { - if c.driver.SupportsListener() { + if !c.config.willExecuteJobs() || c.driver.SupportsListener() { return } @@ -2237,7 +2237,7 @@ func (c *Client[TTx]) InsertManyFast(ctx context.Context, params []InsertManyPar // Fast inserts don't return job rows, so notify for the queues in insert // params instead. - if !c.driver.SupportsListener() { + if c.config.willExecuteJobs() && !c.driver.SupportsListener() { c.notifyProducerWithoutListenerJobFetchForQueues(sliceutil.Map(insertParams, func(insertParam *rivertype.JobInsertParams) string { return insertParam.Queue })) }