diff --git a/CHANGELOG.md b/CHANGELOG.md index 73864e81b..1b66a6430 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,11 @@ 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). +- 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 ### Changed diff --git a/client.go b/client.go index e06af4b0c..e0abd8956 100644 --- a/client.go +++ b/client.go @@ -2153,6 +2153,17 @@ 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.config.willExecuteJobs() || c.driver.SupportsListener() { + 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 } @@ -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.config.willExecuteJobs() && !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, @@ -2943,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 472bac298..8ee93b6b2 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() @@ -7320,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) {