preserve reserved queue metadata

client.go

@@ -48,12 +48,13 @@ const (
 	FetchPollIntervalDefault = 1 * time.Second
 	FetchPollIntervalMin     = 1 * time.Millisecond
 
-	JobStuckThresholdDefault = 10 * time.Second
-	JobTimeoutDefault        = 1 * time.Minute
-	MaxAttemptsDefault       = rivercommon.MaxAttemptsDefault
-	PriorityDefault          = rivercommon.PriorityDefault
-	QueueDefault             = rivercommon.QueueDefault
-	QueueNumWorkersMax       = 10_000
+	JobStuckThresholdDefault            = 10 * time.Second
+	JobTimeoutDefault                   = 1 * time.Minute
+	MaxAttemptsDefault                  = rivercommon.MaxAttemptsDefault
+	PriorityDefault                     = rivercommon.PriorityDefault
+	producerStaleRetentionPeriodDefault = 5 * time.Minute
+	QueueDefault                        = rivercommon.QueueDefault
+	QueueNumWorkersMax                  = 10_000
 )
 
 var (
@@ -823,11 +824,12 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client
 	}
 
 	client.queues = &QueueBundle{
-		clientFetchCooldown:     config.FetchCooldown,
-		clientFetchPollInterval: config.FetchPollInterval,
-		clientWillExecuteJobs:   config.willExecuteJobs(),
-		producerAdd:             client.producerAdd,
-		producerRemove:          client.producerRemove,
+		clientFetchCooldown:      config.FetchCooldown,
+		clientFetchPollInterval:  config.FetchPollInterval,
+		clientWillExecuteJobs:    config.willExecuteJobs(),
+		producerAdd:              client.producerAdd,
+		producerRemove:           client.producerRemove,
+		producerRemoveOnAddError: client.producerRemoveOnAddError,
 	}
 
 	baseservice.Init(archetype, &client.baseService)
@@ -895,9 +897,10 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client
 		client.pilot = &riverpilot.StandardPilot{}
 	}
 	client.pilot.PilotInit(archetype, (&riverpilot.PilotInitParams{
-		Insert:               client.insertMany,
-		NotifyNonTxJobInsert: client.notifyProducerWithoutListenerJobFetch,
-		WorkerMetadata:       workerMetadata,
+		Insert:                       client.insertMany,
+		NotifyNonTxJobInsert:         client.notifyProducerWithoutListenerJobFetch,
+		ProducerStaleRetentionPeriod: producerStaleRetentionPeriodDefault,
+		WorkerMetadata:               workerMetadata,
 	}).Validate())
 	pluginPilot, _ := client.pilot.(pilotPlugin)
 
@@ -2301,7 +2304,7 @@ func (c *Client[TTx]) producerAdd(queueName string, queueConfig QueueConfig) (*p
 		RetryPolicy:                  c.config.RetryPolicy,
 		SchedulerInterval:            c.config.schedulerInterval,
 		Schema:                       c.config.Schema,
-		StaleProducerRetentionPeriod: 5 * time.Minute,
+		StaleProducerRetentionPeriod: producerStaleRetentionPeriodDefault,
 		Workers:                      c.config.Workers,
 	})
 	c.producersByQueueName[queueName] = producer
@@ -2316,6 +2319,11 @@ func (c *Client[TTx]) producerRemove(ctx context.Context, queueName string) erro
 	if !ok {
 		return &QueueNotFoundError{Name: queueName}
 	}
+	if producer.startupCleanupPending.Load() {
+		if err := producer.abortStartup(ctx); err != nil {
+			return err
+		}
+	}
 
 	shouldStop, stopped, finalizeStop := producer.StopInit()
 	if shouldStop {
@@ -2333,6 +2341,15 @@ func (c *Client[TTx]) producerRemove(ctx context.Context, queueName string) erro
 	return nil
 }
 
+func (c *Client[TTx]) producerRemoveOnAddError(queueName string, producer *producer) {
+	c.producersMu.Lock()
+	defer c.producersMu.Unlock()
+
+	if c.producersByQueueName[queueName] == producer && !producer.startupCleanupPending.Load() {
+		delete(c.producersByQueueName, queueName)
+	}
+}
+
 var nameRegex = regexp.MustCompile(`^(?:[a-z0-9])+(?:[_|\-]?[a-z0-9]+)*$`)
 
 func validateQueueName(queueName string) error {
@@ -2890,7 +2907,7 @@ func (c *Client[TTx]) queueUpdate(ctx context.Context, executorTx riverdriver.Ex
 
 	controlEvent := &controlEventPayload{
 		Action:   controlActionMetadataChanged,
-		Metadata: params.Metadata,
+		Metadata: queue.Metadata,
 		Queue:    queue.Name,
 	}
 
@@ -2923,9 +2940,10 @@ type QueueBundle struct {
 
 	clientWillExecuteJobs bool
 
-	fetchCtx       context.Context                                                    //nolint:containedctx
-	producerAdd    func(queueName string, queueConfig QueueConfig) (*producer, error) // add producer to associated client
-	producerRemove func(ctx context.Context, queueName string) error                  // remove producer from associated client
+	fetchCtx                 context.Context                                                    //nolint:containedctx
+	producerAdd              func(queueName string, queueConfig QueueConfig) (*producer, error) // add producer to associated client
+	producerRemove           func(ctx context.Context, queueName string) error                  // remove producer from associated client
+	producerRemoveOnAddError func(queueName string, producer *producer)
 
 	// Mutex that's acquired when client is starting and stopping and when a
 	// queue is being added so that we can be sure that a client is fully
@@ -2958,6 +2976,7 @@ func (b *QueueBundle) Add(queueName string, queueConfig QueueConfig) error {
 	// Start the queue if the client is already started.
 	if b.fetchCtx != nil && b.fetchCtx.Err() == nil {
 		if err := producer.StartWorkContext(b.fetchCtx, b.workCtx); err != nil {
+			b.producerRemoveOnAddError(queueName, producer)
 			return err
 		}
 	}

client_pilot_test.go

@@ -2,6 +2,7 @@ package river
 
 import (
 	"context"
+	"errors"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -29,6 +30,10 @@ type pilotSpy struct {
 	jobInsertManyCalls            atomic.Int64
 	jobRetryCalls                 atomic.Int64
 	pilotInitCalls                atomic.Int64
+	pilotInitParams               *riverpilot.PilotInitParams
+	producerInitErr               error
+	producerInitID                int64
+	producerShutdownErr           error
 
 	testSignals pilotSpyTestSignals
 }
@@ -106,12 +111,16 @@ func (p *pilotSpy) PeriodicJobUpsertMany(ctx context.Context, exec riverdriver.E
 
 func (p *pilotSpy) PilotInit(archetype *baseservice.Archetype, params *riverpilot.PilotInitParams) {
 	p.pilotInitCalls.Add(1)
+	p.pilotInitParams = params
 	p.testSignals.PilotInit.Signal(struct{}{})
 	p.StandardPilot.PilotInit(archetype, params)
 }
 
 func (p *pilotSpy) ProducerInit(ctx context.Context, exec riverdriver.Executor, params *riverpilot.ProducerInitParams) (int64, riverpilot.ProducerState, error) {
 	p.testSignals.ProducerInit.Signal(struct{}{})
+	if p.producerInitErr != nil {
+		return p.producerInitID, nil, p.producerInitErr
+	}
 	return p.StandardPilot.ProducerInit(ctx, exec, params)
 }
 
@@ -122,6 +131,9 @@ func (p *pilotSpy) ProducerKeepAlive(ctx context.Context, exec riverdriver.Execu
 
 func (p *pilotSpy) ProducerShutdown(ctx context.Context, exec riverdriver.Executor, params *riverpilot.ProducerShutdownParams) error {
 	p.testSignals.ProducerShutdown.Signal(struct{}{})
+	if p.producerShutdownErr != nil {
+		return p.producerShutdownErr
+	}
 	return p.StandardPilot.ProducerShutdown(ctx, exec, params)
 }
 
@@ -325,6 +337,7 @@ func Test_Client_PilotUsage(t *testing.T) {
 		require.NotZero(t, insertRes.Job.ID)
 
 		pilot.testSignals.JobGetAvailable.WaitOrTimeout()
+		require.Equal(t, 5*time.Minute, pilot.pilotInitParams.ProducerStaleRetentionPeriod)
 		pilot.testSignals.JobSetStateIfRunningMany.WaitOrTimeout()
 		pilot.testSignals.ProducerInit.WaitOrTimeout()
 		pilot.testSignals.ProducerKeepAlive.WaitOrTimeout()
@@ -334,4 +347,52 @@ func Test_Client_PilotUsage(t *testing.T) {
 
 		pilot.testSignals.ProducerShutdown.WaitOrTimeout()
 	})
+
+	t.Run("QueueAddCleansUpFailedProducerStartup", func(t *testing.T) {
+		t.Parallel()
+
+		client, pilot := setup(t, nil)
+		pilot.testSignals.Init(t)
+		startClient(ctx, t, client)
+
+		startupErr := errors.New("producer startup failed")
+		pilot.producerInitErr = startupErr
+		pilot.producerInitID = 123
+
+		err := client.Queues().Add("dynamic_queue", QueueConfig{MaxWorkers: 1})
+		require.ErrorIs(t, err, startupErr)
+		pilot.testSignals.ProducerShutdown.WaitOrTimeout()
+
+		pilot.producerInitErr = nil
+		require.NoError(t, client.Queues().Add("dynamic_queue", QueueConfig{MaxWorkers: 1}))
+	})
+
+	t.Run("QueueAddRetainsProducerUntilCleanupSucceeds", func(t *testing.T) {
+		t.Parallel()
+
+		client, pilot := setup(t, nil)
+		pilot.testSignals.Init(t)
+		startClient(ctx, t, client)
+
+		startupErr := errors.New("producer startup failed")
+		cleanupErr := errors.New("producer cleanup failed")
+		pilot.producerInitErr = startupErr
+		pilot.producerInitID = 123
+		pilot.producerShutdownErr = cleanupErr
+
+		err := client.Queues().Add("dynamic_queue", QueueConfig{MaxWorkers: 1})
+		var queueAddCleanupErr *QueueAddCleanupError
+		require.ErrorAs(t, err, &queueAddCleanupErr)
+		require.ErrorIs(t, err, startupErr)
+		require.ErrorIs(t, err, cleanupErr)
+
+		err = client.Queues().Add("dynamic_queue", QueueConfig{MaxWorkers: 1})
+		require.ErrorIs(t, err, &QueueAlreadyAddedError{})
+
+		pilot.producerShutdownErr = nil
+		require.NoError(t, client.Queues().Remove(ctx, "dynamic_queue"))
+
+		pilot.producerInitErr = nil
+		require.NoError(t, client.Queues().Add("dynamic_queue", QueueConfig{MaxWorkers: 1}))
+	})
 }

client_test.go

@@ -6348,6 +6348,32 @@ func Test_Client_QueueGet(t *testing.T) {
 		require.Nil(t, queueRes.PausedAt)
 	})
 
+	t.Run("HidesReservedMetadata", func(t *testing.T) {
+		t.Parallel()
+
+		client, bundle := setup(t)
+		exec := client.driver.GetExecutor()
+
+		queue := testfactory.Queue(ctx, t, exec, &testfactory.QueueOpts{
+			Metadata: []byte(`{"foo":"bar"}`),
+			Schema:   bundle.schema,
+		})
+		err := exec.QueryRow(ctx,
+			`UPDATE `+dbutil.SafeIdentifier(bundle.schema)+`.river_queue SET metadata = metadata || '{"river:rate_limit_rollup":{"version":1}}' WHERE name = $1 RETURNING 1`,
+			queue.Name,
+		).Scan(new(int))
+		require.NoError(t, err)
+
+		queueRes, err := client.QueueGet(ctx, queue.Name)
+		require.NoError(t, err)
+		require.JSONEq(t, `{"foo":"bar"}`, string(queueRes.Metadata))
+
+		listRes, err := client.QueueList(ctx, NewQueueListParams())
+		require.NoError(t, err)
+		require.Len(t, listRes.Queues, 1)
+		require.JSONEq(t, `{"foo":"bar"}`, string(listRes.Queues[0].Metadata))
+	})
+
 	t.Run("ReturnsErrNotFoundIfQueueDoesNotExist", func(t *testing.T) {
 		t.Parallel()
 
@@ -6606,9 +6632,14 @@ func Test_Client_QueueUpdate(t *testing.T) {
 
 		queue := testfactory.Queue(ctx, t, client.driver.GetExecutor(), &testfactory.QueueOpts{Schema: bundle.schema})
 		require.Equal(t, []byte(`{}`), queue.Metadata)
+		err = client.driver.GetExecutor().QueryRow(ctx,
+			`UPDATE `+dbutil.SafeIdentifier(bundle.schema)+`.river_queue SET metadata = metadata || '{"river:rate_limit_rollup":{"version":1}}' WHERE name = $1 RETURNING 1`,
+			queue.Name,
+		).Scan(new(int))
+		require.NoError(t, err)
 
 		queue, err = client.QueueUpdate(ctx, queue.Name, &QueueUpdateParams{
-			Metadata: []byte(`{"foo":"bar"}`),
+			Metadata: []byte(`{"foo":"bar","river:attempted_override":true}`),
 		})
 		require.NoError(t, err)
 		require.JSONEq(t, `{"foo":"bar"}`, string(queue.Metadata))
@@ -6621,6 +6652,18 @@ func Test_Client_QueueUpdate(t *testing.T) {
 			Queue:    queue.Name,
 		}, notif.payload)
 
+		var (
+			attemptedOverrideIsNull bool
+			rollup                  []byte
+		)
+		err = client.driver.GetExecutor().QueryRow(ctx,
+			`SELECT metadata -> 'river:rate_limit_rollup', metadata -> 'river:attempted_override' IS NULL FROM `+dbutil.SafeIdentifier(bundle.schema)+`.river_queue WHERE name = $1`,
+			queue.Name,
+		).Scan(&rollup, &attemptedOverrideIsNull)
+		require.NoError(t, err)
+		require.JSONEq(t, `{"version":1}`, string(rollup))
+		require.True(t, attemptedOverrideIsNull)
+
 		queue, err = client.QueueUpdate(ctx, queue.Name, &QueueUpdateParams{
 			Metadata: []byte(`{"foo":"baz"}`),
 		})

error.go

@@ -59,6 +59,24 @@ func (e *QueueAlreadyAddedError) Is(target error) bool {
 	return ok
 }
 
+// QueueAddCleanupError is returned when adding a queue fails after its
+// producer was initialized and River cannot clean up that producer. The queue
+// remains registered so that Remove can retry cleanup without losing track of
+// it.
+type QueueAddCleanupError struct {
+	AddErr     error
+	CleanupErr error
+	Name       string
+}
+
+func (e *QueueAddCleanupError) Error() string {
+	return fmt.Sprintf("queue %q add failed: %v; cleanup failed: %v", e.Name, e.AddErr, e.CleanupErr)
+}
+
+func (e *QueueAddCleanupError) Unwrap() []error {
+	return []error{e.AddErr, e.CleanupErr}
+}
+
 // QueueNotFoundError is returned when attempting to remove a queue that does
 // not exist on the Client.
 type QueueNotFoundError struct {

internal/maintenance/queue_cleaner.go

@@ -161,10 +161,12 @@ func (s *QueueCleaner) runOnce(ctx context.Context) (*queueCleanerRunOnceResult,
 			ctx, cancelFunc := context.WithTimeout(ctx, riversharedmaintenance.TimeoutDefault)
 			defer cancelFunc()
 
+			now := s.Time.Now()
 			queuesDeleted, err := s.exec.QueueDeleteExpired(ctx, &riverdriver.QueueDeleteExpiredParams{
 				Max:              s.batchSize(),
+				Now:              &now,
 				Schema:           s.Config.Schema,
-				UpdatedAtHorizon: time.Now().Add(-s.Config.RetentionPeriod),
+				UpdatedAtHorizon: now.Add(-s.Config.RetentionPeriod),
 			})
 			if err != nil {
 				return nil, fmt.Errorf("error deleting expired queues: %w", err)