preserve internal queue metadata

Rate-limit rollups need durable queue-scoped storage without exposing runtime
state through River's public metadata APIs or allowing ordinary queue updates
to erase it.

Reserve top-level `river:*` metadata keys, preserve them atomically across user
metadata replacement, and filter them from public reads, notifications, and
producer polling. Wrap non-object user metadata only when runtime keys must
coexist with it so the public representation remains unchanged.

Retain queues while a rollup has unexpired debt, lock cleanup candidates before
deletion, and use one database-relative timestamp for each cleanup pass. Pass
River's existing producer stale horizon through the pilot boundary so Pro can
apply the same liveness definition to concurrency and retained rate debt.

Expose shared transaction advisory locks through the driver seam so unlimited
fetches can inspect configuration concurrently while producer lifecycle changes
and limited fetches coordinate through the exclusive counterpart.

Author: bgentry

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 (
@@ -895,9 +896,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 +2303,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
@@ -2890,7 +2892,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,
 	}
 

client_pilot_test.go

@@ -29,6 +29,7 @@ type pilotSpy struct {
 	jobInsertManyCalls            atomic.Int64
 	jobRetryCalls                 atomic.Int64
 	pilotInitCalls                atomic.Int64
+	pilotInitParams               *riverpilot.PilotInitParams
 
 	testSignals pilotSpyTestSignals
 }
@@ -106,6 +107,7 @@ 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)
 }
@@ -325,6 +327,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()

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"}`),
 		})

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)

internal/maintenance/queue_cleaner_test.go

@@ -3,6 +3,7 @@ package maintenance
 import (
 	"context"
 	"fmt"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -19,6 +20,36 @@ import (
 	"github.com/riverqueue/river/rivertype"
 )
 
+type queueCleanerExecutorMock struct {
+	riverdriver.Executor
+
+	queueDeleteExpiredFunc func(ctx context.Context, params *riverdriver.QueueDeleteExpiredParams) ([]string, error)
+}
+
+func (m *queueCleanerExecutorMock) QueueDeleteExpired(ctx context.Context, params *riverdriver.QueueDeleteExpiredParams) ([]string, error) {
+	return m.queueDeleteExpiredFunc(ctx, params)
+}
+
+type queueCleanerTimeGenerator struct {
+	now      time.Time
+	nowCalls atomic.Int32
+}
+
+func (g *queueCleanerTimeGenerator) Now() time.Time {
+	g.nowCalls.Add(1)
+	return g.now
+}
+
+func (g *queueCleanerTimeGenerator) NowOrNil() *time.Time {
+	now := g.Now()
+	return &now
+}
+
+func (g *queueCleanerTimeGenerator) StubNow(now time.Time) time.Time {
+	g.now = now
+	return now
+}
+
 func TestQueueCleaner(t *testing.T) {
 	t.Parallel()
 
@@ -71,6 +102,34 @@ func TestQueueCleaner(t *testing.T) {
 		startstoptest.Stress(ctx, t, cleaner)
 	})
 
+	t.Run("UsesOneNowForExpiration", func(t *testing.T) {
+		t.Parallel()
+
+		cleaner, bundle := setup(t)
+		now := time.Date(2026, time.July, 10, 12, 0, 0, 0, time.UTC)
+		timeGenerator := &queueCleanerTimeGenerator{now: now}
+
+		var capturedParams *riverdriver.QueueDeleteExpiredParams
+		cleaner.Time = timeGenerator
+		cleaner.exec = &queueCleanerExecutorMock{
+			Executor: bundle.exec,
+			queueDeleteExpiredFunc: func(ctx context.Context, params *riverdriver.QueueDeleteExpiredParams) ([]string, error) {
+				paramsCopy := *params
+				capturedParams = &paramsCopy
+				return nil, nil
+			},
+		}
+
+		res, err := cleaner.runOnce(ctx)
+		require.NoError(t, err)
+		require.Empty(t, res.QueuesDeleted)
+		require.Equal(t, int32(1), timeGenerator.nowCalls.Load())
+		require.NotNil(t, capturedParams)
+		require.NotNil(t, capturedParams.Now)
+		require.Equal(t, now, *capturedParams.Now)
+		require.Equal(t, now.Add(-cleaner.Config.RetentionPeriod), capturedParams.UpdatedAtHorizon)
+	})
+
 	t.Run("DeletesExpiredQueues", func(t *testing.T) {
 		t.Parallel()
 

internal/rivercommon/river_common.go

@@ -1,8 +1,10 @@
 package rivercommon
 
 import (
+	"encoding/json"
 	"errors"
 	"regexp"
+	"strings"
 	"time"
 )
 
@@ -12,12 +14,48 @@ import (
 const (
 	// AllQueuesString is a special string that can be used to indicate all
 	// queues in some operations, particularly pause and resume.
-	AllQueuesString    = "*"
-	MaxAttemptsDefault = 25
-	PriorityDefault    = 1
-	QueueDefault       = "default"
+	AllQueuesString              = "*"
+	MaxAttemptsDefault           = 25
+	PriorityDefault              = 1
+	QueueDefault                 = "default"
+	QueueMetadataKeyUserMetadata = "river:user_metadata"
 )
 
+// QueueMetadataForUserWrite removes reserved keys from user-supplied queue
+// metadata while preserving every nonreserved value.
+func QueueMetadataForUserWrite(metadata []byte) []byte {
+	var object map[string]json.RawMessage
+	if err := json.Unmarshal(metadata, &object); err != nil {
+		return metadata
+	}
+
+	for key := range object {
+		if strings.HasPrefix(key, "river:") {
+			delete(object, key)
+		}
+	}
+
+	metadataWithoutReserved, err := json.Marshal(object)
+	if err != nil {
+		return metadata
+	}
+	return metadataWithoutReserved
+}
+
+// QueueMetadataWithoutReserved removes River's reserved top-level metadata
+// keys before queue metadata crosses a public API boundary. A wrapped
+// non-object user value is restored to its original representation.
+func QueueMetadataWithoutReserved(metadata []byte) []byte {
+	var object map[string]json.RawMessage
+	if err := json.Unmarshal(metadata, &object); err != nil {
+		return metadata
+	}
+	if userMetadata, ok := object[QueueMetadataKeyUserMetadata]; ok {
+		return userMetadata
+	}
+	return QueueMetadataForUserWrite(metadata)
+}
+
 // HotOperationTimeout attempts to standardize timeouts for some "hot"
 // operations like locking available jobs or completing finished jobs. It's
 // somewhat questionable whether it makes sense to share timing on these

internal/rivercommon/river_common_test.go

@@ -25,3 +25,15 @@ func TestJobKindRE(t *testing.T) {
 	require.NotRegexp(t, UserSpecifiedIDOrKindRE, "with,comma")
 	require.NotRegexp(t, UserSpecifiedIDOrKindRE, ":no_leading_special_characters")
 }
+
+func TestQueueMetadataWithoutReserved(t *testing.T) {
+	t.Parallel()
+
+	require.JSONEq(t,
+		`{"user":"value"}`,
+		string(QueueMetadataWithoutReserved([]byte(`{"river:internal":{"state":1},"user":"value"}`))),
+	)
+	require.Equal(t, []byte(`not json`), QueueMetadataWithoutReserved([]byte(`not json`)))
+	require.JSONEq(t, `[1,2]`, string(QueueMetadataWithoutReserved([]byte(`{"river:user_metadata":[1,2],"river:internal":{}}`))))
+	require.JSONEq(t, `{"user":"value"}`, string(QueueMetadataForUserWrite([]byte(`{"river:internal":{},"user":"value"}`))))
+}

producer_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/riverqueue/river/rivershared/riversharedtest"
 	"github.com/riverqueue/river/rivershared/startstoptest"
 	"github.com/riverqueue/river/rivershared/testfactory"
+	"github.com/riverqueue/river/rivershared/util/dbutil"
 	"github.com/riverqueue/river/rivershared/util/ptrutil"
 	"github.com/riverqueue/river/rivershared/util/randutil"
 	"github.com/riverqueue/river/rivershared/util/testutil"
@@ -745,6 +746,22 @@ func testProducer(t *testing.T, makeProducer func(ctx context.Context, t *testin
 		producer.config.QueuePollInterval = 50 * time.Millisecond
 
 		startProducer(t, ctx, ctx, producer)
+		if producer.config.Notifier == nil {
+			producer.testSignals.PolledQueueConfig.WaitOrTimeout()
+
+			schemaAndTable := "river_queue"
+			if producer.config.Schema != "" {
+				schemaAndTable = dbutil.SafeIdentifier(producer.config.Schema) + ".river_queue"
+			}
+			err := bundle.exec.QueryRow(ctx,
+				`UPDATE `+schemaAndTable+` SET metadata = metadata || '{"river:rate_limit_rollup":{"version":1}}' WHERE name = $1 RETURNING 1`,
+				producer.config.Queue,
+			).Scan(new(int))
+			require.NoError(t, err)
+
+			producer.testSignals.PolledQueueConfig.WaitOrTimeout()
+			producer.testSignals.MetadataChanged.RequireEmpty()
+		}
 
 		updateMetadata := func(newMetadata []byte) {
 			t.Helper()

riverdriver/river_driver_interface.go

@@ -274,6 +274,7 @@ type Executor interface {
 
 	NotifyMany(ctx context.Context, params *NotifyManyParams) error
 	PGAdvisoryXactLock(ctx context.Context, key int64) (*struct{}, error)
+	PGAdvisoryXactLockShared(ctx context.Context, key int64) (*struct{}, error)
 
 	QueueCreateOrSetUpdatedAt(ctx context.Context, params *QueueCreateOrSetUpdatedAtParams) (*rivertype.Queue, error)
 	QueueDeleteExpired(ctx context.Context, params *QueueDeleteExpiredParams) ([]string, error)
@@ -814,6 +815,7 @@ type QueueCreateOrSetUpdatedAtParams struct {
 
 type QueueDeleteExpiredParams struct {
 	Max              int
+	Now              *time.Time
 	Schema           string
 	UpdatedAtHorizon time.Time
 }