[go] Marshal metadata values when they are set

CHANGELOG.md

@@ -7,6 +7,10 @@ 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

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
 }

metadata_test.go

@@ -2,6 +2,7 @@ package river
 
 import (
 	"context"
+	"encoding/json"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -12,6 +13,16 @@ import (
 func TestMetadataSet(t *testing.T) {
 	t.Parallel()
 
+	t.Run("RejectsNonJSONValue", func(t *testing.T) {
+		t.Parallel()
+
+		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) {
 		t.Parallel()
 
@@ -36,6 +47,25 @@ 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]int{"count": 1}
+		require.NoError(t, MetadataSet(ctx, "key", value))
+
+		value["count"] = 2
+
+		encoded, err := json.Marshal(metadataUpdates)
+		require.NoError(t, err)
+		require.JSONEq(t, `{"key":{"count":1}}`, string(encoded))
 	})
 }