diff --git a/CHANGELOG.md b/CHANGELOG.md index 862bcd08..0a6a5fc8 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/handler_api_endpoint.go b/handler_api_endpoint.go index 4ff966b8..947b6ea8 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) @@ -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 { diff --git a/handler_api_endpoint_test.go b/handler_api_endpoint_test.go index e5e90064..74b4a28d 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() @@ -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() 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()