Align queue concurrency override handling with River Pro

CHANGELOG.md

@@ -13,6 +13,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Show queues whose concurrency metadata is missing, `null`, or `{}` as having no concurrency override, matching how River Pro interprets them, and stop showing a zero concurrency config for producers that report none. [PR #704](https://github.com/riverqueue/riverui/pull/704).
+- Reject negative queue concurrency `global_limit` and `local_limit` values with a 400 error instead of storing them. [PR #704](https://github.com/riverqueue/riverui/pull/704).
 - Keep workflow diagram layouts compatible with Dagre's package exports after dependency upgrades. [PR #687](https://github.com/riverqueue/riverui/pull/687).
 
 ## [v0.21.0] - 2026-10-07

handler_api_endpoint.go

@@ -827,6 +827,15 @@ func (req *queueUpdateRequest) ExtractRaw(r *http.Request) error {
 }
 
 func (a *queueUpdateEndpoint[TTx]) Execute(ctx context.Context, req *queueUpdateRequest) (*RiverQueue, error) {
+	if concurrency := req.Concurrency.Value; concurrency != nil {
+		if concurrency.GlobalLimit < 0 {
+			return nil, apierror.NewBadRequestf("Invalid `global_limit` %d: must be 0 (no limit) or greater.", concurrency.GlobalLimit)
+		}
+		if concurrency.LocalLimit < 0 {
+			return nil, apierror.NewBadRequestf("Invalid `local_limit` %d: must be 0 (no limit) or greater.", concurrency.LocalLimit)
+		}
+	}
+
 	return dbutil.WithTxV(ctx, a.DB, func(ctx context.Context, execTx riverdriver.ExecutorTx) (*RiverQueue, error) {
 		tx := a.Driver.UnwrapTx(execTx)
 
@@ -1256,27 +1265,45 @@ type RiverQueue struct {
 }
 
 func riverQueueToSerializableQueue(internal rivertype.Queue, count *riverdriver.JobCountByQueueAndStateResult) *RiverQueue {
-	var concurrency *ConcurrencyConfig
-	if len(internal.Metadata) > 0 {
-		var metadata struct {
-			Concurrency *ConcurrencyConfig `json:"concurrency"`
-		}
-		if err := json.Unmarshal(internal.Metadata, &metadata); err == nil {
-			concurrency = metadata.Concurrency
-		}
-	}
-
 	return &RiverQueue{
 		CountAvailable: int(count.CountAvailable),
 		CountRunning:   int(count.CountRunning),
 		CreatedAt:      internal.CreatedAt,
-		Concurrency:    concurrency,
+		Concurrency:    queueConcurrencyFromMetadata(internal.Metadata),
 		Name:           internal.Name,
 		PausedAt:       internal.PausedAt,
 		UpdatedAt:      internal.UpdatedAt,
 	}
 }
 
+// queueConcurrencyFromMetadata extracts a queue's concurrency override from
+// its metadata. A missing key, `null`, and `{}` all mean "no override" to River
+// Pro, so they all return nil. Any other object is a full override where
+// omitted fields are zero values rather than being merged with client config.
+func queueConcurrencyFromMetadata(metadata []byte) *ConcurrencyConfig {
+	if len(metadata) == 0 {
+		return nil
+	}
+
+	var parsed struct {
+		Concurrency json.RawMessage `json:"concurrency"`
+	}
+	if err := json.Unmarshal(metadata, &parsed); err != nil {
+		return nil
+	}
+
+	var fields map[string]json.RawMessage
+	if err := json.Unmarshal(parsed.Concurrency, &fields); err != nil || len(fields) == 0 {
+		return nil
+	}
+
+	var concurrency ConcurrencyConfig
+	if err := json.Unmarshal(parsed.Concurrency, &concurrency); err != nil {
+		return nil
+	}
+	return &concurrency
+}
+
 func riverQueuesToSerializableQueues(internal []*rivertype.Queue, counts []*riverdriver.JobCountByQueueAndStateResult) *listResponse[RiverQueue] {
 	countsMap := make(map[string]*riverdriver.JobCountByQueueAndStateResult)
 	for _, count := range counts {

handler_api_endpoint_test.go

@@ -900,6 +900,20 @@ func TestAPIHandlerQueueGet(t *testing.T) {
 		require.Equal(t, queue.Name, resp.Name)
 	})
 
+	t.Run("EmptyConcurrencyObjectIsNoOverride", func(t *testing.T) {
+		t.Parallel()
+
+		endpoint, bundle := setupEndpoint(ctx, t, newQueueGetEndpoint)
+
+		queue := testfactory.Queue(ctx, t, bundle.exec, &testfactory.QueueOpts{
+			Metadata: []byte(`{"concurrency": {}}`),
+		})
+
+		resp, err := apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &queueGetRequest{Name: queue.Name})
+		require.NoError(t, err)
+		require.Nil(t, resp.Concurrency)
+	})
+
 	t.Run("NotFound", func(t *testing.T) {
 		t.Parallel()
 
@@ -1034,6 +1048,88 @@ func TestAPIHandlerQueueUpdate(t *testing.T) {
 		}, resp.Concurrency)
 	})
 
+	t.Run("ClearsOverrideWithNull", func(t *testing.T) {
+		t.Parallel()
+
+		endpoint, bundle := setupEndpoint(ctx, t, newQueueUpdateEndpoint)
+
+		queue := testfactory.Queue(ctx, t, bundle.exec, &testfactory.QueueOpts{
+			Metadata: []byte(`{"concurrency": {"global_limit": 10, "local_limit": 5}}`),
+		})
+
+		var req queueUpdateRequest
+		require.NoError(t, json.Unmarshal([]byte(`{"concurrency": null}`), &req))
+		req.Name = queue.Name
+
+		resp, err := apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &req)
+		require.NoError(t, err)
+		require.Nil(t, resp.Concurrency)
+
+		updatedQueue, err := bundle.client.QueueGetTx(ctx, bundle.tx, queue.Name)
+		require.NoError(t, err)
+		require.JSONEq(t, `{}`, string(updatedQueue.Metadata))
+	})
+
+	t.Run("NegativeGlobalLimit", func(t *testing.T) {
+		t.Parallel()
+
+		endpoint, bundle := setupEndpoint(ctx, t, newQueueUpdateEndpoint)
+
+		queue := testfactory.Queue(ctx, t, bundle.exec, nil)
+
+		_, err := apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &queueUpdateRequest{
+			Name: queue.Name,
+			Concurrency: apitype.ExplicitNullable[ConcurrencyConfig]{
+				Set:   true,
+				Value: &ConcurrencyConfig{GlobalLimit: -1, LocalLimit: 5},
+			},
+		})
+		uicommontest.RequireAPIError(t, apierror.NewBadRequestf("Invalid `global_limit` -1: must be 0 (no limit) or greater."), err)
+	})
+
+	t.Run("NegativeLocalLimit", func(t *testing.T) {
+		t.Parallel()
+
+		endpoint, bundle := setupEndpoint(ctx, t, newQueueUpdateEndpoint)
+
+		queue := testfactory.Queue(ctx, t, bundle.exec, nil)
+
+		_, err := apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &queueUpdateRequest{
+			Name: queue.Name,
+			Concurrency: apitype.ExplicitNullable[ConcurrencyConfig]{
+				Set:   true,
+				Value: &ConcurrencyConfig{GlobalLimit: 10, LocalLimit: -2},
+			},
+		})
+		uicommontest.RequireAPIError(t, apierror.NewBadRequestf("Invalid `local_limit` -2: must be 0 (no limit) or greater."), err)
+	})
+
+	t.Run("SetsFullOverride", func(t *testing.T) {
+		t.Parallel()
+
+		endpoint, bundle := setupEndpoint(ctx, t, newQueueUpdateEndpoint)
+
+		queue := testfactory.Queue(ctx, t, bundle.exec, nil)
+
+		var req queueUpdateRequest
+		require.NoError(t, json.Unmarshal([]byte(`{"concurrency": {"global_limit": 10, "local_limit": 0, "partition": {"by_args": ["b", "a"], "by_kind": true}}}`), &req))
+		req.Name = queue.Name
+
+		resp, err := apitest.InvokeHandler(ctx, endpoint.Execute, testMountOpts(t), &req)
+		require.NoError(t, err)
+		require.Equal(t, &ConcurrencyConfig{
+			GlobalLimit: 10,
+			Partition: PartitionConfig{
+				ByArgs: []string{"a", "b"},
+				ByKind: true,
+			},
+		}, resp.Concurrency)
+
+		updatedQueue, err := bundle.client.QueueGetTx(ctx, bundle.tx, queue.Name)
+		require.NoError(t, err)
+		require.JSONEq(t, `{"concurrency": {"global_limit": 10, "local_limit": 0, "partition": {"by_args": ["a", "b"], "by_kind": true}}}`, string(updatedQueue.Metadata))
+	})
+
 	t.Run("SortsPartitionByArgs", func(t *testing.T) {
 		t.Parallel()
 
@@ -1078,6 +1174,61 @@ func TestAPIHandlerQueueUpdate(t *testing.T) {
 	})
 }
 
+func TestQueueConcurrencyFromMetadata(t *testing.T) {
+	t.Parallel()
+
+	t.Run("EmptyMetadata", func(t *testing.T) {
+		t.Parallel()
+
+		require.Nil(t, queueConcurrencyFromMetadata(nil))
+		require.Nil(t, queueConcurrencyFromMetadata([]byte(`{}`)))
+	})
+
+	t.Run("EmptyObject", func(t *testing.T) {
+		t.Parallel()
+
+		require.Nil(t, queueConcurrencyFromMetadata([]byte(`{"concurrency": {}}`)))
+	})
+
+	t.Run("FullOverride", func(t *testing.T) {
+		t.Parallel()
+
+		require.Equal(t, &ConcurrencyConfig{
+			GlobalLimit: 10,
+			LocalLimit:  5,
+			Partition: PartitionConfig{
+				ByArgs: []string{"customer_id"},
+				ByKind: true,
+			},
+		}, queueConcurrencyFromMetadata([]byte(`{"concurrency": {"global_limit": 10, "local_limit": 5, "partition": {"by_args": ["customer_id"], "by_kind": true}}}`)))
+	})
+
+	t.Run("InvalidJSON", func(t *testing.T) {
+		t.Parallel()
+
+		require.Nil(t, queueConcurrencyFromMetadata([]byte(`{"concurrency": `)))
+		require.Nil(t, queueConcurrencyFromMetadata([]byte(`{"concurrency": "invalid"}`)))
+	})
+
+	t.Run("MissingKey", func(t *testing.T) {
+		t.Parallel()
+
+		require.Nil(t, queueConcurrencyFromMetadata([]byte(`{"other": true}`)))
+	})
+
+	t.Run("Null", func(t *testing.T) {
+		t.Parallel()
+
+		require.Nil(t, queueConcurrencyFromMetadata([]byte(`{"concurrency": null}`)))
+	})
+
+	t.Run("PartialOverrideZeroesOmittedFields", func(t *testing.T) {
+		t.Parallel()
+
+		require.Equal(t, &ConcurrencyConfig{LocalLimit: 3}, queueConcurrencyFromMetadata([]byte(`{"concurrency": {"local_limit": 3}}`)))
+	})
+}
+
 func TestStateAndCountGetEndpoint(t *testing.T) {
 	t.Parallel()
 

riverproui/internal/prohandler/pro_handler_api_endpoints.go

@@ -175,20 +175,10 @@ func (a *producerListEndpoint[TTx]) Execute(ctx context.Context, req *producerLi
 }
 
 func internalProducerToSerializableProducer(internal *riverprodriver.ProducerListByQueueResult) *uitype.RiverProducer {
-	var concurrency *uitype.ConcurrencyConfig
-	if len(internal.Producer.Metadata) > 0 {
-		var metadata struct {
-			Concurrency uitype.ConcurrencyConfig `json:"concurrency"`
-		}
-		if err := json.Unmarshal(internal.Producer.Metadata, &metadata); err == nil {
-			concurrency = &metadata.Concurrency
-		}
-	}
-
 	return &uitype.RiverProducer{
 		ID:          internal.Producer.ID,
 		ClientID:    internal.Producer.ClientID,
-		Concurrency: concurrency,
+		Concurrency: producerConcurrencyFromMetadata(internal.Producer.Metadata),
 		CreatedAt:   internal.Producer.CreatedAt,
 		MaxWorkers:  int(internal.Producer.MaxWorkers),
 		PausedAt:    internal.Producer.PausedAt,
@@ -198,6 +188,41 @@ func internalProducerToSerializableProducer(internal *riverprodriver.ProducerLis
 	}
 }
 
+// producerConcurrencyFromMetadata extracts the concurrency config a producer
+// reports in its metadata. A missing key, `null`, or an object without any
+// config fields means the producer has no concurrency config, so it returns
+// nil instead of a zero config. Producers also store limiter state like
+// `running` alongside the config, which doesn't count as config on its own.
+func producerConcurrencyFromMetadata(metadata []byte) *uitype.ConcurrencyConfig {
+	if len(metadata) == 0 {
+		return nil
+	}
+
+	var parsed struct {
+		Concurrency json.RawMessage `json:"concurrency"`
+	}
+	if err := json.Unmarshal(metadata, &parsed); err != nil {
+		return nil
+	}
+
+	var fields map[string]json.RawMessage
+	if err := json.Unmarshal(parsed.Concurrency, &fields); err != nil {
+		return nil
+	}
+	_, hasGlobalLimit := fields["global_limit"]
+	_, hasLocalLimit := fields["local_limit"]
+	_, hasPartition := fields["partition"]
+	if !hasGlobalLimit && !hasLocalLimit && !hasPartition {
+		return nil
+	}
+
+	var concurrency uitype.ConcurrencyConfig
+	if err := json.Unmarshal(parsed.Concurrency, &concurrency); err != nil {
+		return nil
+	}
+	return &concurrency
+}
+
 //
 // workflowCancelEndpoint
 //

riverproui/internal/prohandler/pro_handler_api_endpoints_test.go

@@ -31,6 +31,7 @@ import (
 	"riverqueue.com/riverui/internal/riverinternaltest/testfactory"
 	"riverqueue.com/riverui/internal/uicommontest"
 	"riverqueue.com/riverui/riverproui/internal/protestfactory"
+	"riverqueue.com/riverui/riverproui/internal/uitype"
 )
 
 type setupEndpointTestBundle struct {
@@ -128,6 +129,69 @@ func TestProAPIHandlerPeriodicJobList(t *testing.T) {
 	})
 }
 
+func TestProducerConcurrencyFromMetadata(t *testing.T) {
+	t.Parallel()
+
+	t.Run("EmptyMetadata", func(t *testing.T) {
+		t.Parallel()
+
+		require.Nil(t, producerConcurrencyFromMetadata(nil))
+		require.Nil(t, producerConcurrencyFromMetadata([]byte(`{}`)))
+	})
+
+	t.Run("EmptyObject", func(t *testing.T) {
+		t.Parallel()
+
+		require.Nil(t, producerConcurrencyFromMetadata([]byte(`{"concurrency": {}}`)))
+	})
+
+	t.Run("FullConfig", func(t *testing.T) {
+		t.Parallel()
+
+		require.Equal(t, &uitype.ConcurrencyConfig{
+			GlobalLimit: 10,
+			LocalLimit:  5,
+			Partition: uitype.PartitionConfig{
+				ByArgs: []string{"customer_id"},
+				ByKind: true,
+			},
+		}, producerConcurrencyFromMetadata([]byte(`{"concurrency": {"global_limit": 10, "local_limit": 5, "partition": {"by_args": ["customer_id"], "by_kind": true}, "running": {"a": 1}}}`)))
+	})
+
+	t.Run("InvalidJSON", func(t *testing.T) {
+		t.Parallel()
+
+		require.Nil(t, producerConcurrencyFromMetadata([]byte(`{"concurrency": `)))
+		require.Nil(t, producerConcurrencyFromMetadata([]byte(`{"concurrency": "invalid"}`)))
+	})
+
+	t.Run("MissingKey", func(t *testing.T) {
+		t.Parallel()
+
+		require.Nil(t, producerConcurrencyFromMetadata([]byte(`{"other": true}`)))
+	})
+
+	t.Run("Null", func(t *testing.T) {
+		t.Parallel()
+
+		require.Nil(t, producerConcurrencyFromMetadata([]byte(`{"concurrency": null}`)))
+	})
+
+	t.Run("RunningStateOnly", func(t *testing.T) {
+		t.Parallel()
+
+		require.Nil(t, producerConcurrencyFromMetadata([]byte(`{"concurrency": {"running": {"a": 1}}}`)))
+	})
+
+	t.Run("ZeroLimits", func(t *testing.T) {
+		t.Parallel()
+
+		// A producer that reports explicit zero limits has a config with no
+		// limits, which is distinct from having no config at all.
+		require.Equal(t, &uitype.ConcurrencyConfig{}, producerConcurrencyFromMetadata([]byte(`{"concurrency": {"global_limit": 0, "local_limit": 0}}`)))
+	})
+}
+
 func TestProAPIHandlerWorkflowCancel(t *testing.T) {
 	t.Parallel()