From 65b04e06a4cc242c7714de9e825c37d52b7574ed Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Sun, 11 Oct 2026 20:14:38 -0700 Subject: [PATCH 1/4] reject negative queue concurrency limits The queue update endpoint stores whatever `global_limit` and `local_limit` it receives in `river_queue.metadata.concurrency`. The form's `min="0"` attribute is the only guard, so a direct API call (or a crafted request) can persist a negative limit that River Pro has no meaningful interpretation for. Validate the concurrency override before opening a transaction and return a 400 via `apierror.NewBadRequestf` when either limit is negative, matching the endpoint's other bad request responses. Zero remains valid because it means "no limit". Rejecting the request up front keeps invalid configuration out of the database entirely rather than relying on every reader to cope with it. Add queue update tests covering a negative global limit and a negative local limit. --- CHANGELOG.md | 1 + handler_api_endpoint.go | 9 +++++++++ handler_api_endpoint_test.go | 34 ++++++++++++++++++++++++++++++++++ 3 files changed, 44 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 862bcd08..95bb197c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- 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 diff --git a/handler_api_endpoint.go b/handler_api_endpoint.go index 4ff966b8..303f954d 100644 --- a/handler_api_endpoint.go +++ b/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) diff --git a/handler_api_endpoint_test.go b/handler_api_endpoint_test.go index e5e90064..c6966acc 100644 --- a/handler_api_endpoint_test.go +++ b/handler_api_endpoint_test.go @@ -1034,6 +1034,40 @@ func TestAPIHandlerQueueUpdate(t *testing.T) { }, resp.Concurrency) }) + 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("SortsPartitionByArgs", func(t *testing.T) { t.Parallel() From 6347805db72d2c286797a5c281b1cb41068590a0 Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Sun, 11 Oct 2026 20:15:21 -0700 Subject: [PATCH 2/4] read empty queue concurrency override as none River Pro treats a missing `concurrency` key, `null`, and `{}` in `river_queue.metadata` as "no override", and any non-empty object as a full override. The queue serializer unmarshals the key into a `*ConcurrencyConfig`, which yields nil for a missing key or `null` but a non-nil zero config for `{}`. The queue detail page then shows the override toggle as enabled with 0/0 limits even though River Pro is using each client's own configuration. Extract the override through `queueConcurrencyFromMetadata`, which decodes the raw `concurrency` value into a map first and returns nil when it is absent, `null`, empty, or not an object. Only a non-empty object decodes into a `ConcurrencyConfig`, with omitted fields left at zero rather than merged, mirroring River Pro's semantics. The frontend already treats a `null` concurrency as "no override", so no UI change is needed. Add unit tests for empty metadata, a missing key, `null`, `{}`, malformed values, a full override, and a partial override, plus a queue get endpoint test showing `{"concurrency": {}}` is returned as no override. --- CHANGELOG.md | 1 + handler_api_endpoint.go | 40 +++++++++++++++------ handler_api_endpoint_test.go | 69 ++++++++++++++++++++++++++++++++++++ 3 files changed, 99 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 95bb197c..2dac1d52 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ 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. [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). diff --git a/handler_api_endpoint.go b/handler_api_endpoint.go index 303f954d..947b6ea8 100644 --- a/handler_api_endpoint.go +++ b/handler_api_endpoint.go @@ -1265,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 { diff --git a/handler_api_endpoint_test.go b/handler_api_endpoint_test.go index c6966acc..efb9f1e6 100644 --- a/handler_api_endpoint_test.go +++ b/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() @@ -1112,6 +1126,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() From db27d6ce3a068128711113476cdfb690bf8433ac Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Sun, 11 Oct 2026 20:16:26 -0700 Subject: [PATCH 3/4] omit producer concurrency when none is reported The producer list decodes `metadata.concurrency` into a non-pointer `ConcurrencyConfig` and always returns its address when the metadata parses. A producer whose metadata has no `concurrency` key, `null`, or `{}` is therefore reported with a zero config instead of none. The queue detail page uses producer configs to prefill the concurrency form and to decide whether clients agree, so these phantom zero configs show up as 0/0 limits and can make clients look inconsistent. Extract the config through `producerConcurrencyFromMetadata`, which returns nil unless the `concurrency` object contains at least one of `global_limit`, `local_limit`, or `partition`. Producers also keep limiter state like `running` in that object, so it is ignored when deciding whether a config is present. The API type and the frontend's `ConcurrencyConfig | null` producer type already allow a missing config, and the comparison logic handles `null`, so no frontend change is needed. Add unit tests for empty metadata, a missing key, `null`, `{}`, running state without config, malformed values, explicit zero limits, and a full config. --- CHANGELOG.md | 2 +- .../prohandler/pro_handler_api_endpoints.go | 47 ++++++++++---- .../pro_handler_api_endpoints_test.go | 64 +++++++++++++++++++ 3 files changed, 101 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2dac1d52..0a6a5fc8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,7 +13,7 @@ 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. [PR #704](https://github.com/riverqueue/riverui/pull/704). +- 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). diff --git a/riverproui/internal/prohandler/pro_handler_api_endpoints.go b/riverproui/internal/prohandler/pro_handler_api_endpoints.go index a5937412..99b95a22 100644 --- a/riverproui/internal/prohandler/pro_handler_api_endpoints.go +++ b/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 // diff --git a/riverproui/internal/prohandler/pro_handler_api_endpoints_test.go b/riverproui/internal/prohandler/pro_handler_api_endpoints_test.go index ad37fac3..d4450e33 100644 --- a/riverproui/internal/prohandler/pro_handler_api_endpoints_test.go +++ b/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() From 2cbda6b0c41dc2984050379a2ccb3a0babf68d28 Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Sun, 11 Oct 2026 20:16:50 -0700 Subject: [PATCH 4/4] test clearing and setting queue concurrency The queue update endpoint distinguishes an omitted `concurrency` field, an explicit `null`, and an object through `ExplicitNullable`, but only the object case has coverage, and none of the tests decode the request from JSON or check what lands in `river_queue.metadata`. A regression in how `null` decodes or in what the clear path writes would go unnoticed. Add queue update tests that decode real JSON request bodies. Clearing with `{"concurrency": null}` replaces an existing override with `{}` metadata and returns no override, which River Pro reads as "no override". Setting a full object stores it verbatim, including a zero `local_limit` and sorted `partition.by_args`, and returns the same config. --- handler_api_endpoint_test.go | 48 ++++++++++++++++++++++++++++++++++++ 1 file changed, 48 insertions(+) diff --git a/handler_api_endpoint_test.go b/handler_api_endpoint_test.go index efb9f1e6..74b4a28d 100644 --- a/handler_api_endpoint_test.go +++ b/handler_api_endpoint_test.go @@ -1048,6 +1048,28 @@ 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() @@ -1082,6 +1104,32 @@ func TestAPIHandlerQueueUpdate(t *testing.T) { 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()