From dd2c0354a75c8ca2502fdf6faf3118d9c5057405 Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Fri, 25 Sep 2026 10:29:07 -0500 Subject: [PATCH 1/2] escape JSON paths in unique field names Fields marked `river:"unique"` are read and written as JSON paths, so a name like `user.id` can omit its value and deduplicate distinct jobs. Unnamed JSON tags also omit their fields instead of using the Go field name. Escape each path component, including leading colons, and fall back to the Go field name for unnamed tags. Keep the original field ordering so unaffected jobs retain their existing unique keys. Cover escaped components, literal and nested path collisions, and unnamed tags with explicit encoding fixtures and cross-driver insertion cases. --- CHANGELOG.md | 1 + internal/dbunique/db_unique_test.go | 82 +++++++++++ .../riverdrivertest/driver_client_test.go | 77 +++++++++++ rivershared/structtag/struct_tag.go | 127 +++++++++++++----- rivershared/structtag/struct_tag_test.go | 38 ++++++ 5 files changed, 292 insertions(+), 33 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4fd232894..f3c5ba5cd 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 +- Fixed `UniqueOpts.ByArgs` skipping distinct jobs when `river:"unique"` fields have JSON names containing path syntax (like `user.id` or `alice@example.com`), or an unnamed JSON tag like `json:",omitempty"`. Field names are now addressed literally, and unnamed tags use the Go field name. Unique keys change for affected jobs, so old and new clients may each insert a job with the same args during a rolling upgrade. [PR #1387](https://github.com/riverqueue/river/pull/1387). - Fixed SQLite job list pagination skipping or repeating jobs by formatting cursor timestamps consistently with stored timestamps. [PR #1374](https://github.com/riverqueue/river/pull/1374). - Improved PostgreSQL job listing performance when filtering by one finalized state (`completed`, `cancelled`, or `discarded`) and sorting by finalized time, including in River UI. [PR #1374](https://github.com/riverqueue/river/pull/1374). - Fixed `JobRescuer` overwriting jobs that complete, leave the running state, or are claimed again by another worker after being fetched for rescue, preserving their state, errors, metadata, and timestamps across PostgreSQL and SQLite drivers. Fixes [#1302](https://github.com/riverqueue/river/issues/1302). [PR #1373](https://github.com/riverqueue/river/pull/1373). diff --git a/internal/dbunique/db_unique_test.go b/internal/dbunique/db_unique_test.go index 8952df91a..673f3e711 100644 --- a/internal/dbunique/db_unique_test.go +++ b/internal/dbunique/db_unique_test.go @@ -101,6 +101,88 @@ func TestUniqueKey(t *testing.T) { uniqueOpts: UniqueOpts{ByArgs: true}, expectedJSON: `&kind=worker_1&args={"Recipient":"john@example.com","Subject":"Another Test Email"}`, }, + { + name: "ByArgsUniqueWithUnnamedJSONTagUsesFieldName", + argsFunc: func() rivertype.JobArgs { + //nolint:tagliatelle // non-snake keys are intentional + type EmailJobArgs struct { + JobArgsStaticKind + + Recipient string `json:",omitempty" river:"unique"` + Subject string `json:"subject" river:"unique"` + TemplateID int + } + return EmailJobArgs{ + JobArgsStaticKind: JobArgsStaticKind{kind: "worker_1"}, + Recipient: "john@example.com", + Subject: "Another Test Email", + TemplateID: 102, + } + }, + uniqueOpts: UniqueOpts{ByArgs: true}, + expectedJSON: `&kind=worker_1&args={"Recipient":"john@example.com","subject":"Another Test Email"}`, + }, + { + name: "ByArgsWithCompatibleJSONNames", + argsFunc: func() rivertype.JobArgs { + //nolint:tagliatelle // Exercise names whose escaped sort order differs. + type Nested struct { + Amount int `json:"a&b" river:"unique"` + Upper string `json:"Y" river:"unique"` + } + //nolint:tagliatelle // Exercise valid names whose escaping must not change hashes. + type Args struct { + JobArgsStaticKind + + Dollar string `json:"$x" river:"unique"` + Nested Nested `json:"nested"` + Unicode string `json:"é<" river:"unique"` + Upper string `json:"Y" river:"unique"` + } + return Args{ + JobArgsStaticKind: JobArgsStaticKind{kind: "worker_1"}, + Dollar: "dollar", + Nested: Nested{Amount: 1, Upper: "nested"}, + Unicode: "unicode", + Upper: "upper", + } + }, + uniqueOpts: UniqueOpts{ByArgs: true}, + // Preserve the old byte ordering even though escaped `\$x` sorts after `Y`. + expectedJSON: `&kind=worker_1&args={"$x":"dollar","Y":"upper","nested":{"Y":"nested","a&b":1},"é\u003c":"unicode"}`, + }, + { + name: "ByArgsUniqueWithPathSyntaxInJSONTags", + argsFunc: func() rivertype.JobArgs { + type Nested struct { + Value string `json:"inner@key" river:"unique"` + } + //nolint:tagliatelle // non-snake keys are intentional + type PathSyntaxJobArgs struct { + JobArgsStaticKind + + Bang string `json:"!bang" river:"unique"` + Colon string `json:":x" river:"unique"` + Email string `json:"alice@example.com" river:"unique"` + Literal string `json:"outer.key.inner@key" river:"unique"` + Nested Nested `json:"outer.key"` + UserID string `json:"user.id" river:"unique"` + X string `json:"x" river:"unique"` + } + return PathSyntaxJobArgs{ + JobArgsStaticKind: JobArgsStaticKind{kind: "worker_1"}, + Bang: "bang", + Colon: "colon", + Email: "email", + Literal: "literal", + Nested: Nested{Value: "nested"}, + UserID: "u1", + X: "x", + } + }, + uniqueOpts: UniqueOpts{ByArgs: true}, + expectedJSON: `&kind=worker_1&args={"!bang":"bang",":x":"colon","alice@example.com":"email","outer.key":{"inner@key":"nested"},"outer.key.inner@key":"literal","user.id":"u1","x":"x"}`, + }, { name: "ByArgsWithPointerToStruct", argsFunc: func() rivertype.JobArgs { diff --git a/riverdriver/riverdrivertest/driver_client_test.go b/riverdriver/riverdrivertest/driver_client_test.go index f1d720ebb..5acbb4f6e 100644 --- a/riverdriver/riverdrivertest/driver_client_test.go +++ b/riverdriver/riverdrivertest/driver_client_test.go @@ -418,6 +418,83 @@ func ExerciseClient[TTx any](ctx context.Context, t *testing.T, require.Equal(t, rivertype.JobStateCancelled, event.Job.State) }) + // Unique fields whose JSON keys contain gjson/sjson path syntax, or whose + // `json` tag has no name, are part of the unique key. + t.Run("InsertUniqueByArgsUniqueFieldsWithPathSyntaxKeys", func(t *testing.T) { + t.Parallel() + + client, bundle := setup(t) + + type User struct { + ID string `json:"id" river:"unique"` + } + + //nolint:tagliatelle // non-snake keys are intentional + type JobArgs struct { + testutil.JobArgsReflectKind[JobArgs] + + Bang string `json:"!bang" river:"unique"` + Colon string `json:":colon" river:"unique"` + Email string `json:"alice@example.com" river:"unique"` + Other string `json:"other"` + Recipient string `json:",omitempty" river:"unique"` + User User `json:"user"` + UserID string `json:"user.id" river:"unique"` + Wildcard string `json:"wild*?" river:"unique"` + } + + river.AddWorker(bundle.config.Workers, river.WorkFunc(func(ctx context.Context, job *river.Job[JobArgs]) error { + return nil + })) + + insert := func(t *testing.T, args *JobArgs) *rivertype.JobInsertResult { + t.Helper() + + insertRes, err := client.Insert(ctx, args, &river.InsertOpts{ + UniqueOpts: river.UniqueOpts{ByArgs: true}, + }) + require.NoError(t, err) + return insertRes + } + + baseArgs := JobArgs{ + Bang: "bang", + Colon: "colon", + Email: "email", + Other: "other", + Recipient: "recipient", + User: User{ID: "nested"}, + UserID: "u1", + Wildcard: "wildcard", + } + + insertRes0 := insert(t, &baseArgs) + require.False(t, insertRes0.UniqueSkippedAsDuplicate) + + // A change to a field that isn't unique is still a duplicate. + args := baseArgs + args.Other = "changed" + insertRes1 := insert(t, &args) + require.True(t, insertRes1.UniqueSkippedAsDuplicate) + require.Equal(t, insertRes0.Job.ID, insertRes1.Job.ID) + + for _, modify := range []func(args *JobArgs){ + func(args *JobArgs) { args.Bang = "changed" }, + func(args *JobArgs) { args.Colon = "changed" }, + func(args *JobArgs) { args.Email = "changed" }, + func(args *JobArgs) { args.Recipient = "changed" }, + func(args *JobArgs) { args.User.ID = "changed" }, + func(args *JobArgs) { args.UserID = "u2" }, + func(args *JobArgs) { args.Wildcard = "changed" }, + } { + args := baseArgs + modify(&args) + + insertRes := insert(t, &args) + require.False(t, insertRes.UniqueSkippedAsDuplicate, "args: %+v", args) + } + }) + t.Run("InsertUniqueByPeriod", func(t *testing.T) { t.Parallel() diff --git a/rivershared/structtag/struct_tag.go b/rivershared/structtag/struct_tag.go index 66ce8a371..dfac05d54 100644 --- a/rivershared/structtag/struct_tag.go +++ b/rivershared/structtag/struct_tag.go @@ -1,9 +1,10 @@ package structtag import ( + "cmp" "fmt" "reflect" - "sort" + "slices" "strings" "sync" @@ -12,7 +13,9 @@ import ( "github.com/riverqueue/river/rivertype" ) -// ExtractValues extracts the raw JSON values of the specified keys from the JSON-encoded args. +// ExtractValues extracts the raw JSON values of the specified keys from the +// JSON-encoded args. Keys are gjson paths like those returned by +// SortedFieldsWithTag. func ExtractValues(encodedArgs []byte, uniqueKeys []string) []string { // Use GetManyBytes to retrieve multiple values at once results := gjson.GetManyBytes(encodedArgs, uniqueKeys...) @@ -32,6 +35,30 @@ func ExtractValues(encodedArgs []byte, uniqueKeys []string) []string { return uniqueValues } +// fieldPath is the location of a tagged field within a struct's JSON encoding. +type fieldPath struct { + // path is a gjson/sjson path to the field, with each component escaped so + // that JSON keys containing path syntax are treated literally. + path string + + // sortKey is the unescaped components joined with `.`. Paths are sorted by + // it so that fields keep the order they had before components were + // escaped, which keeps unique keys stable for ordinary field names. + sortKey string +} + +func newFieldPath(components []string) fieldPath { + escaped := make([]string, len(components)) + for i, component := range components { + escaped[i] = escapePathComponent(component) + } + + return fieldPath{ + path: strings.Join(escaped, "."), + sortKey: strings.Join(components, "."), + } +} + type uniqueFieldCacheKey struct { typ reflect.Type tagValue string @@ -46,6 +73,11 @@ var ( // SortedFieldsWithTag retrieves unique fields with caching to avoid // extracting fields from the same struct type repeatedly. +// +// Fields are returned as gjson/sjson paths suitable for use with ExtractValues +// and sjson. Each path component is escaped so a JSON key containing path +// syntax like `.`, `@`, or `*` addresses that key literally. Paths are sorted +// by their unescaped, dot-joined JSON keys. func SortedFieldsWithTag(args rivertype.JobArgs, tagValue string) ([]string, error) { var ( typ = reflect.TypeOf(args) @@ -61,7 +93,7 @@ func SortedFieldsWithTag(args rivertype.JobArgs, tagValue string) ([]string, err cacheMutex.RUnlock() // Not in cache; retrieve using reflection - fields, err := sortedFieldsWithTagUncached(reflect.TypeOf(args), tagValue, nil, make(map[reflect.Type]struct{})) + fields, err := sortedFieldsWithTagUncached(typ, tagValue) if err != nil { return nil, err } @@ -74,14 +106,37 @@ func SortedFieldsWithTag(args rivertype.JobArgs, tagValue string) ([]string, err return fields, nil } -// sortedFieldsWithTagUncached uses reflection to retrieve the JSON keys of fields +// sortedFieldsWithTagUncached uses reflection to retrieve the escaped JSON +// paths of fields marked with `river:""`, sorted by their unescaped +// JSON keys. +func sortedFieldsWithTagUncached(typ reflect.Type, tagValue string) ([]string, error) { + fieldPaths, err := fieldPathsWithTag(typ, tagValue, nil, make(map[reflect.Type]struct{})) + if err != nil { + return nil, err + } + + // Sort by unescaped keys for consistent ordering that matches the order + // used before path components were escaped. Break ties (possible when a + // key containing `.` collides with a nested path) by escaped path. + slices.SortFunc(fieldPaths, func(a, b fieldPath) int { + return cmp.Or(strings.Compare(a.sortKey, b.sortKey), strings.Compare(a.path, b.path)) + }) + + var fields []string + for _, fieldPath := range fieldPaths { + fields = append(fields, fieldPath.path) + } + return fields, nil +} + +// fieldPathsWithTag uses reflection to retrieve the JSON paths of fields // marked with `river:""` among potentially other comma-separated -// values. The return values are the JSON keys using the same logic as the -// `json` struct tag. +// values. Path components are the JSON keys using the same logic as the `json` +// struct tag. Results are unsorted. // // typesSeen should be a map passed through to make sure that recursive types // don't cause a stack overflow. -func sortedFieldsWithTagUncached(typ reflect.Type, tagValue string, path []string, typesSeen map[reflect.Type]struct{}) ([]string, error) { +func fieldPathsWithTag(typ reflect.Type, tagValue string, path []string, typesSeen map[reflect.Type]struct{}) ([]fieldPath, error) { // Handle pointer to struct if typ.Kind() == reflect.Pointer { typ = typ.Elem() @@ -100,7 +155,7 @@ func sortedFieldsWithTagUncached(typ reflect.Type, tagValue string, path []strin } typesSeen[typ] = struct{}{} - var uniqueFields []string + var uniqueFields []fieldPath // Iterate over all fields for field := range typ.Fields() { @@ -108,19 +163,8 @@ func sortedFieldsWithTagUncached(typ reflect.Type, tagValue string, path []strin continue } - var uniqueName string - { - // Get the corresponding JSON key - jsonTag := field.Tag.Get("json") - - if jsonTag == "" { - // If no JSON tag, use the field name as-is - uniqueName = field.Name - } else { - // Handle cases like `json:"recipient,omitempty"` - uniqueName = parseJSONTag(jsonTag) - } - } + // Get the corresponding JSON key + uniqueName := parseJSONTag(field.Name, field.Tag.Get("json")) // Check for `river:"unique"` tag, possibly among other comma-separated values var hasUniqueTag bool @@ -142,7 +186,7 @@ func sortedFieldsWithTagUncached(typ reflect.Type, tagValue string, path []strin fullPath = append(path, uniqueName) //nolint:gocritic } - uniqueSubFields, err := sortedFieldsWithTagUncached(field.Type, tagValue, fullPath, typesSeen) + uniqueSubFields, err := fieldPathsWithTag(field.Type, tagValue, fullPath, typesSeen) if err != nil { return nil, err } @@ -154,31 +198,48 @@ func sortedFieldsWithTagUncached(typ reflect.Type, tagValue string, path []strin // JSON serialization as a unique value. This may not be the // greatest idea practically, but keeping it in place for // backwards compatibility. - uniqueFields = append(uniqueFields, strings.Join(append(path, uniqueName), ".")) + uniqueFields = append(uniqueFields, newFieldPath(append(path, uniqueName))) } continue } if hasUniqueTag { - uniqueFields = append(uniqueFields, strings.Join(append(path, uniqueName), ".")) + uniqueFields = append(uniqueFields, newFieldPath(append(path, uniqueName))) } } - // Sort the uniqueFields alphabetically for consistent ordering - sort.Strings(uniqueFields) - return uniqueFields, nil } -// parseJSONTag extracts the JSON key from the struct tag. -// It handles tags with options, e.g., `json:"recipient,omitempty"`. -func parseJSONTag(tag string) string { +// escapePathComponent escapes a JSON object key for use as a single component +// of a gjson or sjson path so that it addresses the key literally. +// +// gjson.Escape handles path separators, wildcards, modifiers, and queries. +// Also escape a leading colon so sjson doesn't strip it as an object-key +// directive. Both libraries unescape a backslash before any byte. +func escapePathComponent(key string) string { + escaped := gjson.Escape(key) + if strings.HasPrefix(escaped, ":") { + return `\` + escaped + } + return escaped +} + +// parseJSONTag returns the JSON key for a field with the given name and `json` +// struct tag. It handles tags with options, e.g., `json:"recipient,omitempty"`, +// and like encoding/json falls back to the field name when the tag has no name +// (e.g. `json:",omitempty"`). +// +// Preserve the historical handling of invalid JSON tag names; only the empty +// name falls back to the field name here. +func parseJSONTag(fieldName, tag string) string { // Tags can be like "recipient,omitempty", so split by comma - if before, _, ok := strings.Cut(tag, ","); ok { - return before + name, _, _ := strings.Cut(tag, ",") + if name == "" { + return fieldName } - return tag + return name } func typeStructOrPointerToStruct(typ reflect.Type) bool { diff --git a/rivershared/structtag/struct_tag_test.go b/rivershared/structtag/struct_tag_test.go index 3a49c5745..822a90a5d 100644 --- a/rivershared/structtag/struct_tag_test.go +++ b/rivershared/structtag/struct_tag_test.go @@ -9,6 +9,38 @@ import ( "github.com/riverqueue/river/rivertype" ) +func TestEscapePathComponent(t *testing.T) { + t.Parallel() + + for _, tt := range []struct { + key string + expected string + }{ + {key: "", expected: ""}, + {key: "user_id", expected: "user_id"}, + {key: "user-id", expected: "user-id"}, + {key: "UserID", expected: "UserID"}, + {key: "123", expected: "123"}, + {key: "-1", expected: "-1"}, + {key: "a b", expected: "a b"}, + {key: "éλ", expected: "éλ"}, + {key: "a:b", expected: "a:b"}, + {key: ":x", expected: `\:x`}, + {key: "::x", expected: `\::x`}, + {key: "user.id", expected: `user\.id`}, + {key: "alice@example.com", expected: `alice\@example\.com`}, + {key: "!x", expected: `\!x`}, + {key: "[x]", expected: `\[x\]`}, + {key: "{x}", expected: `\{x\}`}, + {key: "a*b?c#d|e", expected: `a\*b\?c\#d\|e`}, + {key: `a\b`, expected: `a\\b`}, + {key: ":a.b", expected: `\:a\.b`}, + {key: "$x", expected: `\$x`}, + } { + require.Equal(t, tt.expected, escapePathComponent(tt.key), "key: %q", tt.key) + } +} + func TestExtractValues(t *testing.T) { t.Parallel() @@ -78,6 +110,12 @@ func TestExtractValues(t *testing.T) { uniqueKeys: []string{"a", "b"}, expectedValues: []string{"undefined", "undefined"}, }, + { + name: "EscapedPathComponents", + encodedArgs: []byte(`{"user.id":"u1","alice@example.com":2,":x":3,"x":4,"user":{"id":"nested"}}`), + uniqueKeys: []string{`user\.id`, `alice\@example\.com`, `\:x`, "x", "user.id"}, + expectedValues: []string{`"u1"`, "2", "3", "4", `"nested"`}, + }, } for _, tt := range testCases { From a8b2624df10b83f870a9e247db4cb6ea22884b54 Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Fri, 25 Sep 2026 10:29:10 -0500 Subject: [PATCH 2/2] preserve literal keys in all-args uniqueness The all-args unique key builder interprets object keys as JSON paths. Keys containing path syntax can lose their values or collide with other keys, while an empty key causes an insertion error. Walk the object directly and rebuild it in key order with raw values. Preserve the previous key encoding and first-value handling for repeated keys so unaffected hashes remain stable. Continue rejecting scalar and nonempty array args instead of silently hashing them as empty objects. Cover literal keys and duplicate detection across drivers. Pin the key encoding with explicit compatibility fixtures, including Unicode, control characters, and raw nested values, and verify rejection of non-object args. --- CHANGELOG.md | 2 +- internal/dbunique/db_unique.go | 89 ++++++++++++++++--- internal/dbunique/db_unique_test.go | 63 +++++++++++++ .../riverdrivertest/driver_client_test.go | 58 ++++++++++++ 4 files changed, 197 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f3c5ba5cd..b5e5548a5 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 -- Fixed `UniqueOpts.ByArgs` skipping distinct jobs when `river:"unique"` fields have JSON names containing path syntax (like `user.id` or `alice@example.com`), or an unnamed JSON tag like `json:",omitempty"`. Field names are now addressed literally, and unnamed tags use the Go field name. Unique keys change for affected jobs, so old and new clients may each insert a job with the same args during a rolling upgrade. [PR #1387](https://github.com/riverqueue/river/pull/1387). +- Fixed `UniqueOpts.ByArgs` skipping distinct jobs or failing inserts when JSON keys contain path syntax (like `user.id`), are empty, or come from unnamed tags like `json:",omitempty"`. Unaffected unique keys remain unchanged; affected jobs may be inserted again after upgrading or by old and new clients during a rolling upgrade. [PR #1387](https://github.com/riverqueue/river/pull/1387). - Fixed SQLite job list pagination skipping or repeating jobs by formatting cursor timestamps consistently with stored timestamps. [PR #1374](https://github.com/riverqueue/river/pull/1374). - Improved PostgreSQL job listing performance when filtering by one finalized state (`completed`, `cancelled`, or `discarded`) and sorting by finalized time, including in River UI. [PR #1374](https://github.com/riverqueue/river/pull/1374). - Fixed `JobRescuer` overwriting jobs that complete, leave the running state, or are claimed again by another worker after being fetched for rescue, preserving their state, errors, metadata, and timestamps across PostgreSQL and SQLite drivers. Fixes [#1302](https://github.com/riverqueue/river/issues/1302). [PR #1373](https://github.com/riverqueue/river/pull/1373). diff --git a/internal/dbunique/db_unique.go b/internal/dbunique/db_unique.go index cf6037316..8cab7aaad 100644 --- a/internal/dbunique/db_unique.go +++ b/internal/dbunique/db_unique.go @@ -2,6 +2,8 @@ package dbunique import ( "crypto/sha256" + "encoding/json" + "errors" "slices" "strings" "time" @@ -12,7 +14,6 @@ import ( "github.com/riverqueue/river/rivershared/structtag" "github.com/riverqueue/river/rivershared/uniquestates" "github.com/riverqueue/river/rivershared/util/ptrutil" - "github.com/riverqueue/river/rivershared/util/sliceutil" "github.com/riverqueue/river/rivertype" ) @@ -93,20 +94,10 @@ func buildUniqueKeyString(timeGen rivertype.TimeGenerator, uniqueOpts *UniqueOpt encodedArgsForUnique = sortedJSONWithOnlyUniqueValues } else { // Use all keys from EncodedArgs sorted alphabetically - keys := sliceutil.Map(gjson.GetBytes(params.EncodedArgs, "@keys").Array(), func(v gjson.Result) string { return v.String() }) - slices.Sort(keys) - - sortedJSON := make([]byte, 0, len(params.EncodedArgs)) - sortedJSON = append(sortedJSON, "{}"...) - sjsonOpts := &sjson.Options{ReplaceInPlace: true} - for _, key := range keys { - sortedJSON, err = sjson.SetRawBytesOptions(sortedJSON, key, []byte(gjson.GetBytes(params.EncodedArgs, key).Raw), sjsonOpts) - if err != nil { - // Should not happen unless key was invalid - return "", err - } + encodedArgsForUnique, err = appendSortedObject(make([]byte, 0, len(params.EncodedArgs)), params.EncodedArgs) + if err != nil { + return "", err } - encodedArgsForUnique = sortedJSON } sb.WriteString("&args=") @@ -129,3 +120,73 @@ func buildUniqueKeyString(timeGen rivertype.TimeGenerator, uniqueOpts *UniqueOpt return sb.String(), nil } + +// appendJSONKey appends key to buf as a JSON string, encoded the same way sjson +// encodes object keys: verbatim between quotes, unless the key contains a byte +// below 0x20 or above 0x7f, `"`, or `\`, in which case it's encoded with +// encoding/json (which also escapes `<`, `>`, `&`, U+2028, and U+2029, and +// replaces invalid UTF-8 with U+FFFD). Matching sjson keeps unique keys +// identical to those of earlier versions, which built unique args with sjson. +func appendJSONKey(buf []byte, key string) []byte { + for i := range len(key) { + if key[i] < ' ' || key[i] > 0x7f || key[i] == '"' || key[i] == '\\' { + encodedKey, _ := json.Marshal(key) //nolint:errchkjson // marshaling a string can't fail + return append(buf, encodedKey...) + } + } + + buf = append(buf, '"') + buf = append(buf, key...) + return append(buf, '"') +} + +// appendSortedObject appends to buf a compact JSON object containing each +// top-level key of encodedObject along with its raw value, sorted by key. If +// a key appears more than once, its first value is used. As in the previous +// path-based implementation, empty input and an empty array produce `{}`; +// other non-object input is rejected. +// +// Keys are walked directly rather than addressed as gjson/sjson paths so that +// keys containing path syntax (like `.`, `@`, or a leading `:`) and empty +// keys are included literally, and distinct values of such keys produce +// distinct output. For all other keys, output is byte-identical to that +// produced by setting each key onto `{}` with sjson. +func appendSortedObject(buf, encodedObject []byte) ([]byte, error) { + type keyValue struct { + key string + rawValue string + } + + var ( + keyValues []keyValue + keysSeen = make(map[string]struct{}) + err error + ) + gjson.ParseBytes(encodedObject).ForEach(func(key, value gjson.Result) bool { + if key.Type != gjson.String { + err = errors.New("unique args must encode a JSON object") + return false + } + if _, ok := keysSeen[key.Str]; !ok { + keysSeen[key.Str] = struct{}{} + keyValues = append(keyValues, keyValue{key: key.Str, rawValue: value.Raw}) + } + return true + }) + if err != nil { + return nil, err + } + + slices.SortFunc(keyValues, func(a, b keyValue) int { return strings.Compare(a.key, b.key) }) + + buf = append(buf, '{') + for i, keyValue := range keyValues { + if i > 0 { + buf = append(buf, ',') + } + buf = appendJSONKey(buf, keyValue.key) + buf = append(buf, ':') + buf = append(buf, keyValue.rawValue...) + } + return append(buf, '}'), nil +} diff --git a/internal/dbunique/db_unique_test.go b/internal/dbunique/db_unique_test.go index 673f3e711..bb3f545cf 100644 --- a/internal/dbunique/db_unique_test.go +++ b/internal/dbunique/db_unique_test.go @@ -22,6 +22,60 @@ func (a JobArgsStaticKind) Kind() string { return a.kind } +func TestAppendSortedObject(t *testing.T) { + t.Parallel() + + t.Run("Encodes", func(t *testing.T) { + t.Parallel() + + // Compare exact bytes: equivalent JSON with different key escaping or + // whitespace would change the hashes of jobs inserted before upgrading. + for _, tt := range []struct { + name string + encodedObject string + expected string + }{ + {name: "ASCIIHTML", encodedObject: `{"a\u003cb\u0026c\u003e":1}`, expected: `{"a":1}`}, + {name: "Colon", encodedObject: `{"x":2,":x":1}`, expected: `{":x":1,"x":2}`}, + {name: "ControlCharacters", encodedObject: `{"line\nbreak":1,"\u0000":2}`, expected: `{"\u0000":2,"line\nbreak":1}`}, + {name: "DEL", encodedObject: `{"\u007f":1}`, expected: "{\"\x7f\":1}"}, + {name: "DuplicateKeys", encodedObject: `{"a":1,"\u0061":2}`, expected: `{"a":1}`}, + {name: "EmptyArray", encodedObject: `[]`, expected: `{}`}, + {name: "EmptyInput", encodedObject: ``, expected: `{}`}, + {name: "EmptyKey", encodedObject: `{"a":2,"":1}`, expected: `{"":1,"a":2}`}, + {name: "EmptyObject", encodedObject: `{}`, expected: `{}`}, + {name: "EscapedKey", encodedObject: `{"quote\"slash\\":1}`, expected: `{"quote\"slash\\":1}`}, + {name: "LiteralAndNestedKeys", encodedObject: `{"file.name":"a","file":{"name":"b"}}`, expected: `{"file":{"name":"b"},"file.name":"a"}`}, + {name: "NumericKeys", encodedObject: `{"10":2,"0":1}`, expected: `{"0":1,"10":2}`}, + {name: "RawValues", encodedObject: `{"b":null,"a":{"y":1, "x":[1, 2]}}`, expected: `{"a":{"y":1, "x":[1, 2]},"b":null}`}, + {name: "UnicodeHTML", encodedObject: `{"\u00e9\u003c":1}`, expected: `{"é\u003c":1}`}, + {name: "UnicodeSeparators", encodedObject: `{"\u2029":2,"\u2028":1}`, expected: `{"\u2028":1,"\u2029":2}`}, + } { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + encoded, err := appendSortedObject(nil, []byte(tt.encodedObject)) + require.NoError(t, err) + require.Equal(t, tt.expected, string(encoded)) + }) + } + }) + + t.Run("RejectsNonObjectArgs", func(t *testing.T) { + t.Parallel() + + for _, encodedArgs := range []string{`null`, `[1,2]`, `"str"`, `true`, `123`} { + args := JobArgsStaticKind{kind: "kind"} + _, err := UniqueKey(&riversharedtest.TimeStub{}, &UniqueOpts{ByArgs: true}, &rivertype.JobInsertParams{ + Args: args, + EncodedArgs: []byte(encodedArgs), + Kind: args.Kind(), + }) + require.EqualError(t, err, "unique args must encode a JSON object", "encoded args: %s", encodedArgs) + } + }) +} + func TestUniqueKey(t *testing.T) { t.Parallel() @@ -426,6 +480,15 @@ func TestUniqueKey(t *testing.T) { // args JSON should be sorted alphabetically: expectedJSON: `&kind=worker_3&args={"count":10,"description":"A generic job without unique fields."}`, }, + { + name: "ByArgsWithNoUniqueFieldsAndLiteralKeys", + argsFunc: func() rivertype.JobArgs { return JobArgsStaticKind{kind: "worker_3"} }, + modifyInsertParamsFunc: func(params *rivertype.JobInsertParams) { + params.EncodedArgs = []byte(`{"x":"x","file.name":"file","alice@example.com":"email",":x":"colon","[x":"bracket","{x":"brace","":"empty"}`) + }, + uniqueOpts: UniqueOpts{ByArgs: true}, + expectedJSON: `&kind=worker_3&args={"":"empty",":x":"colon","[x":"bracket","alice@example.com":"email","file.name":"file","x":"x","{x":"brace"}`, + }, { name: "ByArgsWithEmptyEncodedArgs", argsFunc: func() rivertype.JobArgs { diff --git a/riverdriver/riverdrivertest/driver_client_test.go b/riverdriver/riverdrivertest/driver_client_test.go index 5acbb4f6e..87a0cf410 100644 --- a/riverdriver/riverdrivertest/driver_client_test.go +++ b/riverdriver/riverdrivertest/driver_client_test.go @@ -3,6 +3,8 @@ package riverdrivertest import ( "context" "database/sql" + "encoding/json" + "maps" "math" "slices" "testing" @@ -230,6 +232,16 @@ func TestClientWithDriverRiverTurso(t *testing.T) { ) } +// customJSONArgs are job args encoded as an arbitrary JSON object, like args +// with a custom MarshalJSON implementation. +type customJSONArgs struct { + values map[string]string +} + +func (customJSONArgs) Kind() string { return "customJSON" } + +func (a customJSONArgs) MarshalJSON() ([]byte, error) { return json.Marshal(a.values) } + type noOpArgs struct { Name string `json:"name"` } @@ -418,6 +430,52 @@ func ExerciseClient[TTx any](ctx context.Context, t *testing.T, require.Equal(t, rivertype.JobStateCancelled, event.Job.State) }) + // Keys containing gjson/sjson path syntax (and the empty key) are distinct + // keys when unique by all args, so args differing in their values aren't + // duplicates. + t.Run("InsertUniqueByArgsAllArgsWithPathSyntaxKeys", func(t *testing.T) { + t.Parallel() + + client, bundle := setup(t) + + river.AddWorker(bundle.config.Workers, river.WorkFunc(func(ctx context.Context, job *river.Job[customJSONArgs]) error { + return nil + })) + + insert := func(t *testing.T, values map[string]string) *rivertype.JobInsertResult { + t.Helper() + + insertRes, err := client.Insert(ctx, customJSONArgs{values: values}, &river.InsertOpts{ + UniqueOpts: river.UniqueOpts{ByArgs: true}, + }) + require.NoError(t, err) + return insertRes + } + + var ( + keys = []string{"", "!x", ":x", "[x", "alice@example.com", "file.name", "x*?", "x#", "x|", `x\y`, "{x"} + baseValues = map[string]string{"x": "x"} + ) + for _, key := range keys { + baseValues[key] = "value" + } + + insertRes0 := insert(t, baseValues) + require.False(t, insertRes0.UniqueSkippedAsDuplicate) + + insertRes1 := insert(t, maps.Clone(baseValues)) + require.True(t, insertRes1.UniqueSkippedAsDuplicate) + require.Equal(t, insertRes0.Job.ID, insertRes1.Job.ID) + + for _, key := range keys { + values := maps.Clone(baseValues) + values[key] = "other" + + insertRes := insert(t, values) + require.False(t, insertRes.UniqueSkippedAsDuplicate, "key: %q", key) + } + }) + // Unique fields whose JSON keys contain gjson/sjson path syntax, or whose // `json` tag has no name, are part of the unique key. t.Run("InsertUniqueByArgsUniqueFieldsWithPathSyntaxKeys", func(t *testing.T) {