From 6f21e7d14f8976e0327b56191e746120ff02c429 Mon Sep 17 00:00:00 2001 From: Abdullah Alqahtani <61029571+anqorithm@users.noreply.github.com> Date: Wed, 4 Nov 2026 02:05:35 +0100 Subject: [PATCH 1/2] [go] Marshal metadata values when they are set --- CHANGELOG.md | 1 + client_test.go | 35 +++++++++++++++++++++++++++++++++++ metadata.go | 11 ++++++++++- metadata_test.go | 43 ++++++++++++++++++++++++++++++++++++++++++- 4 files changed, 88 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index eedce02fd..dcc010a60 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 - Accept `sqlite:path` and `sqlite::memory:` in the CLI alongside the existing `sqlite://` forms, including offline migration SQL export. [PR #1588](https://github.com/riverqueue/river/pull/1588). - Reject empty SQLite paths in commands that connect to a database; bare prefixes remain valid for offline migration SQL export. [PR #1588](https://github.com/riverqueue/river/pull/1588). +- Fixed `MetadataSet` accepting values that can't be encoded as JSON, which prevented the worker's result from being recorded. Values are now encoded when set, so invalid values return an error and later mutations don't change buffered metadata. [PR #1589](https://github.com/riverqueue/river/pull/1589). - Fixed `JobListParams.Where` writing into the first `NamedArgs` it's given when merging multiple sets of named arguments, which could leak arguments into a map the caller reused for another query. It also no longer panics when the first set is `nil`. [PR #1575](https://github.com/riverqueue/river/pull/1575). ## [0.51.1] - 2026-11-05 diff --git a/client_test.go b/client_test.go index 271d04461..18101b628 100644 --- a/client_test.go +++ b/client_test.go @@ -1715,6 +1715,41 @@ func Test_Client_Common(t *testing.T) { require.True(t, workEndHookCalled) }) + t.Run("WithWorkerRejectingInvalidMetadata", func(t *testing.T) { + t.Parallel() + + _, bundle := setup(t) + + type JobArgs struct { + testutil.JobArgsReflectKind[JobArgs] + } + + AddWorkerArgs(bundle.config.Workers, JobArgs{}, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error { + if err := MetadataSet(ctx, "valid", "kept"); err != nil { + return err + } + return MetadataSet(ctx, "invalid", make(chan int)) + })) + + client, err := NewClient(riverpgxv5.New(bundle.dbPool), bundle.config) + require.NoError(t, err) + + subscribeChan := subscribe(t, client) + startClient(ctx, t, client) + + insertRes, err := client.Insert(ctx, JobArgs{}, &InsertOpts{MaxAttempts: 1}) + require.NoError(t, err) + + event := riversharedtest.WaitOrTimeout(t, subscribeChan) + require.Equal(t, EventKindJobFailed, event.Kind) + require.Equal(t, insertRes.Job.ID, event.Job.ID) + require.Equal(t, rivertype.JobStateDiscarded, event.Job.State) + require.NotNil(t, event.Job.FinalizedAt) + require.Len(t, event.Job.Errors, 1) + require.Contains(t, event.Job.Errors[0].Error, "json: unsupported type: chan int") + require.JSONEq(t, `{"valid":"kept"}`, string(event.Job.Metadata)) + }) + t.Run("WithWorkerSettingMetadata", func(t *testing.T) { t.Parallel() diff --git a/metadata.go b/metadata.go index eafba14d0..6d24f4601 100644 --- a/metadata.go +++ b/metadata.go @@ -2,6 +2,7 @@ package river import ( "context" + "encoding/json" "errors" "strings" @@ -20,6 +21,9 @@ var errMetadataNotSettable = errors.New("MetadataSet must be called within a wor // when the current work attempt finishes, whether the attempt succeeds or // errors. Values must be JSON marshalable because metadata is stored in a // jsonb column, and setting a key replaces any existing value at that key. +// Values are marshaled when MetadataSet is called, so later changes to them +// don't affect the stored metadata. If marshaling fails, MetadataSet returns +// the error without changing any metadata. // // Keys prefixed with `river:` are reserved for internal use and may not be set // by user code. @@ -33,6 +37,11 @@ func MetadataSet(ctx context.Context, key string, value any) error { return errMetadataNotSettable } - metadataUpdates[key] = value + encoded, err := json.Marshal(value) + if err != nil { + return err + } + + metadataUpdates[key] = json.RawMessage(encoded) return nil } diff --git a/metadata_test.go b/metadata_test.go index 94298915f..6a99defd4 100644 --- a/metadata_test.go +++ b/metadata_test.go @@ -2,6 +2,8 @@ package river import ( "context" + "encoding/json" + "math" "testing" "github.com/stretchr/testify/require" @@ -12,6 +14,29 @@ import ( func TestMetadataSet(t *testing.T) { t.Parallel() + t.Run("RejectsNonJSONValues", func(t *testing.T) { + t.Parallel() + + for _, tt := range []struct { + name string + value any + }{ + {name: "Channel", value: make(chan int)}, + {name: "InvalidRawMessage", value: json.RawMessage(`{"unfinished":`)}, + {name: "NaN", value: math.NaN()}, + } { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + metadataUpdates := map[string]any{"existing": "kept", "key": "previous"} + ctx := context.WithValue(context.Background(), jobexecutor.ContextKeyMetadataUpdates, metadataUpdates) + + require.Error(t, MetadataSet(ctx, "key", tt.value)) + require.Equal(t, map[string]any{"existing": "kept", "key": "previous"}, metadataUpdates) + }) + } + }) + t.Run("RejectsReservedPrefix", func(t *testing.T) { t.Parallel() @@ -36,6 +61,22 @@ func TestMetadataSet(t *testing.T) { err := MetadataSet(ctx, "key", "value") require.NoError(t, err) - require.Equal(t, "value", metadataUpdates["key"]) + encoded, err := json.Marshal(metadataUpdates) + require.NoError(t, err) + require.JSONEq(t, `{"key":"value"}`, string(encoded)) + }) + + t.Run("SnapshotsValue", func(t *testing.T) { + t.Parallel() + + metadataUpdates := map[string]any{} + ctx := context.WithValue(context.Background(), jobexecutor.ContextKeyMetadataUpdates, metadataUpdates) + value := map[string]any{"count": 1} + + require.NoError(t, MetadataSet(ctx, "key", value)) + value["count"] = math.NaN() + encoded, err := json.Marshal(metadataUpdates) + require.NoError(t, err) + require.JSONEq(t, `{"key":{"count":1}}`, string(encoded)) }) } From b6cc61fd286417453ea9a9d85911c1208269384d Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Tue, 10 Nov 2026 07:43:56 -0800 Subject: [PATCH 2/2] [go] trim MetadataSet tests and fix changelog section Reduce the invalid value test to a single case, since every case takes the same `json.Marshal` error path, and drop the Postgres-backed client test, which exercises the executor's existing error handling rather than the new validation. Mutate the snapshot test's value to another number instead of `NaN` so it fails on a plain mismatch. Move the changelog entry from the already released 0.51.2 section to `Unreleased`. --- CHANGELOG.md | 5 ++++- client_test.go | 35 ----------------------------------- metadata_test.go | 33 +++++++++++---------------------- 3 files changed, 15 insertions(+), 58 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index dcc010a60..82610fc51 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,13 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- Fixed `MetadataSet` accepting values that can't be encoded as JSON, which prevented the worker's result from being recorded. Values are now encoded when set, so invalid values return an error and later mutations don't change buffered metadata. [PR #1589](https://github.com/riverqueue/river/pull/1589). + ## [0.51.2] - 2026-11-09 ### Fixed - Accept `sqlite:path` and `sqlite::memory:` in the CLI alongside the existing `sqlite://` forms, including offline migration SQL export. [PR #1588](https://github.com/riverqueue/river/pull/1588). - Reject empty SQLite paths in commands that connect to a database; bare prefixes remain valid for offline migration SQL export. [PR #1588](https://github.com/riverqueue/river/pull/1588). -- Fixed `MetadataSet` accepting values that can't be encoded as JSON, which prevented the worker's result from being recorded. Values are now encoded when set, so invalid values return an error and later mutations don't change buffered metadata. [PR #1589](https://github.com/riverqueue/river/pull/1589). - Fixed `JobListParams.Where` writing into the first `NamedArgs` it's given when merging multiple sets of named arguments, which could leak arguments into a map the caller reused for another query. It also no longer panics when the first set is `nil`. [PR #1575](https://github.com/riverqueue/river/pull/1575). ## [0.51.1] - 2026-11-05 diff --git a/client_test.go b/client_test.go index 18101b628..271d04461 100644 --- a/client_test.go +++ b/client_test.go @@ -1715,41 +1715,6 @@ func Test_Client_Common(t *testing.T) { require.True(t, workEndHookCalled) }) - t.Run("WithWorkerRejectingInvalidMetadata", func(t *testing.T) { - t.Parallel() - - _, bundle := setup(t) - - type JobArgs struct { - testutil.JobArgsReflectKind[JobArgs] - } - - AddWorkerArgs(bundle.config.Workers, JobArgs{}, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error { - if err := MetadataSet(ctx, "valid", "kept"); err != nil { - return err - } - return MetadataSet(ctx, "invalid", make(chan int)) - })) - - client, err := NewClient(riverpgxv5.New(bundle.dbPool), bundle.config) - require.NoError(t, err) - - subscribeChan := subscribe(t, client) - startClient(ctx, t, client) - - insertRes, err := client.Insert(ctx, JobArgs{}, &InsertOpts{MaxAttempts: 1}) - require.NoError(t, err) - - event := riversharedtest.WaitOrTimeout(t, subscribeChan) - require.Equal(t, EventKindJobFailed, event.Kind) - require.Equal(t, insertRes.Job.ID, event.Job.ID) - require.Equal(t, rivertype.JobStateDiscarded, event.Job.State) - require.NotNil(t, event.Job.FinalizedAt) - require.Len(t, event.Job.Errors, 1) - require.Contains(t, event.Job.Errors[0].Error, "json: unsupported type: chan int") - require.JSONEq(t, `{"valid":"kept"}`, string(event.Job.Metadata)) - }) - t.Run("WithWorkerSettingMetadata", func(t *testing.T) { t.Parallel() diff --git a/metadata_test.go b/metadata_test.go index 6a99defd4..5112b29ec 100644 --- a/metadata_test.go +++ b/metadata_test.go @@ -3,7 +3,6 @@ package river import ( "context" "encoding/json" - "math" "testing" "github.com/stretchr/testify/require" @@ -14,27 +13,14 @@ import ( func TestMetadataSet(t *testing.T) { t.Parallel() - t.Run("RejectsNonJSONValues", func(t *testing.T) { + t.Run("RejectsNonJSONValue", func(t *testing.T) { t.Parallel() - for _, tt := range []struct { - name string - value any - }{ - {name: "Channel", value: make(chan int)}, - {name: "InvalidRawMessage", value: json.RawMessage(`{"unfinished":`)}, - {name: "NaN", value: math.NaN()}, - } { - t.Run(tt.name, func(t *testing.T) { - t.Parallel() - - metadataUpdates := map[string]any{"existing": "kept", "key": "previous"} - ctx := context.WithValue(context.Background(), jobexecutor.ContextKeyMetadataUpdates, metadataUpdates) - - require.Error(t, MetadataSet(ctx, "key", tt.value)) - require.Equal(t, map[string]any{"existing": "kept", "key": "previous"}, metadataUpdates) - }) - } + metadataUpdates := map[string]any{} + ctx := context.WithValue(context.Background(), jobexecutor.ContextKeyMetadataUpdates, metadataUpdates) + + require.EqualError(t, MetadataSet(ctx, "key", make(chan int)), "json: unsupported type: chan int") + require.Empty(t, metadataUpdates) }) t.Run("RejectsReservedPrefix", func(t *testing.T) { @@ -61,6 +47,7 @@ func TestMetadataSet(t *testing.T) { err := MetadataSet(ctx, "key", "value") require.NoError(t, err) + encoded, err := json.Marshal(metadataUpdates) require.NoError(t, err) require.JSONEq(t, `{"key":"value"}`, string(encoded)) @@ -71,10 +58,12 @@ func TestMetadataSet(t *testing.T) { metadataUpdates := map[string]any{} ctx := context.WithValue(context.Background(), jobexecutor.ContextKeyMetadataUpdates, metadataUpdates) - value := map[string]any{"count": 1} + value := map[string]int{"count": 1} require.NoError(t, MetadataSet(ctx, "key", value)) - value["count"] = math.NaN() + + value["count"] = 2 + encoded, err := json.Marshal(metadataUpdates) require.NoError(t, err) require.JSONEq(t, `{"key":{"count":1}}`, string(encoded))