[go] Fix InsertManyFast and QueueUpdate panics on drivers without listener support

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

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
 }

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) {