diff --git a/AGENTS.md b/AGENTS.md index 9c8abaac5..4a22e7787 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -31,6 +31,15 @@ - **Terminology**: prefer "Postgres" over "PostgreSQL". - **Naming**: use idiomatic Go names; see `.golangci.yaml` for allowed short variable names. +## Cross-Language Parity + +This repo holds the Go reference implementation and its ports (`java/`, `js/`, `ruby/`, `rust/`). The ports aim for full compatibility with Go and with each other. + +- **One PR for all languages**: when adding a feature or fixing a bug in shared behavior, change Go and every affected port in the same PR. Splitting the work by language makes it hard to verify that the implementations still agree. +- **Check every port, even when the report names one**: a bug found in one port often exists in the others, and in the Go reference. Confirm which implementations are affected before scoping the fix. +- **Same tests everywhere**: give each affected implementation an equivalent test for the behavior, on every database it supports. +- **Document intended differences**: an API should be idiomatic for its language, but observable behavior should match Go. Record any difference that remains in the port's differences doc (for example `java/DIFFERENCES.md`). + ## Package Naming and Organization - Package names are lowercase, short, and representative; avoid `common`, `util`, or overly broad names. diff --git a/CHANGELOG.md b/CHANGELOG.md index 775e1c4e7..19a52fa38 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 +- Fixed SQLite metadata updates recursively merging nested objects and deleting keys set to JSON null. Updates now replace whole top-level values, matching Postgres and preventing completed resumable cursor steps from running again on later retries. [PR #1533](https://github.com/riverqueue/river/pull/1533). + - Fixed resumable transactional checkpoints being saved again after rollback or a failed write, and first-step cursors being lost on error, snooze, or cooperative shutdown. After a successful checkpoint write, the caller's transaction owns that progress; rollback may replay earlier unpersisted steps. The Rust, JavaScript, and Java runtimes now share the applicable behavior, including preserving progress on snooze. - Fixed `AddWorkerSafely` leaving a worker's primary kind and earlier aliases registered when a later kind alias conflicts with an existing registration. Failed registrations now leave the worker registry unchanged. [PR #1440](https://github.com/riverqueue/river/pull/1440). diff --git a/java/CHANGELOG.md b/java/CHANGELOG.md index d2a71b9cb..812c47f45 100644 --- a/java/CHANGELOG.md +++ b/java/CHANGELOG.md @@ -13,3 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `HookWorkEnd.workEnd` returns the attempt's failure: return the incoming failure to preserve it, a replacement to change it, or `null` to report success. Each hook receives the previous hook's result, so hooks migrated from `Extension.afterWork` must explicitly return the failure to preserve the outcome. A work-begin failure now skips both the handler and work-end hooks. [PR #1519](https://github.com/riverqueue/river/pull/1519). - Insertion middleware now receives the batch's submissions and returns its insertion results. It runs once per batch, including a one-element batch for single inserts, while insertion hooks run once per job. Hooks and middleware run inside the insertion's transaction. [PR #1519](https://github.com/riverqueue/river/pull/1519). - Renamed the internal River Pro storage interface and its registration method from `Plugin`/`Client.withPlugin` to `Pilot`/`Client.withPilot`. `Plugin` is now the common marker interface for application hooks and middleware. [PR #1519](https://github.com/riverqueue/river/pull/1519). + +### Fixed + +- Replace whole top-level metadata values on SQLite, including JSON null, matching Postgres. Completed resumable cursor steps no longer run again because of stale cursor entries. [PR #1533](https://github.com/riverqueue/river/pull/1533). diff --git a/java/conformance/src/main/java/com/riverqueue/conformance/RawStorage.java b/java/conformance/src/main/java/com/riverqueue/conformance/RawStorage.java index e58fdce26..317b5f883 100644 --- a/java/conformance/src/main/java/com/riverqueue/conformance/RawStorage.java +++ b/java/conformance/src/main/java/com/riverqueue/conformance/RawStorage.java @@ -159,7 +159,15 @@ INSERT INTO river_job (args,attempt,attempted_at,attempted_by,created_at,errors, + ",finalized_at=" + (sqlite ? "datetime('now','subsec')" : "now()") + ",metadata=" - + (sqlite ? "jsonb_patch(metadata,jsonb(?))" : "metadata||?::jsonb") + + (sqlite + ? """ + (WITH updates(doc) AS (SELECT jsonb(?)) + SELECT jsonb_group_object(key,jsonb(doc->fullkey)) FROM ( + SELECT key,fullkey,metadata AS doc FROM json_each(metadata) + WHERE key NOT IN (SELECT key FROM updates,json_each(doc)) + UNION ALL SELECT key,fullkey,doc FROM updates,json_each(doc))) + """ + : "metadata||?::jsonb") + ",errors=" + (params.path("state").asString().equals("completed") ? (sqlite diff --git a/java/river/src/main/resources/com/riverqueue/sqlite.sql b/java/river/src/main/resources/com/riverqueue/sqlite.sql index f38cfd267..75e2d53b0 100644 --- a/java/river/src/main/resources/com/riverqueue/sqlite.sql +++ b/java/river/src/main/resources/com/riverqueue/sqlite.sql @@ -173,17 +173,17 @@ finalized_at=CASE WHEN state != 'running' THEN finalized_at WHEN (SELECT target_ scheduled_at=CASE WHEN state != 'running' THEN scheduled_at WHEN (SELECT target_state FROM choice) IN ('available','retryable','scheduled') AND (CASE WHEN typeof(metadata)<>'text' OR json_valid(metadata) THEN json_extract(metadata,'$.cancel_attempted_at') END) IS NOT NULL THEN scheduled_at ELSE ? END, attempt=CASE WHEN state != 'running' THEN attempt WHEN (SELECT target_state FROM choice) IN ('available','retryable','scheduled') AND (CASE WHEN typeof(metadata)<>'text' OR json_valid(metadata) THEN json_extract(metadata,'$.cancel_attempted_at') END) IS NOT NULL THEN attempt ELSE ? END, metadata=CASE WHEN typeof(metadata)='text' AND NOT json_valid(metadata) THEN metadata ELSE ( - -- MergePatch recursively merges objects and deletes nulls; cursor maps must be replaced whole. - SELECT CASE - WHEN json_type(metadata_updates,'$."river:resumable_cursor"') IS NOT NULL - THEN jsonb_set(merged,'$."river:resumable_cursor"',jsonb_extract(metadata_updates,'$."river:resumable_cursor"')) - WHEN use_fallback AND json_type(checkpoint_fallback,'$."river:resumable_cursor"') IS NOT NULL - THEN jsonb_set(merged,'$."river:resumable_cursor"',jsonb_extract(checkpoint_fallback,'$."river:resumable_cursor"')) - ELSE merged END - FROM ( - SELECT *, jsonb_patch(CASE WHEN use_fallback THEN jsonb_patch(metadata,checkpoint_fallback) ELSE metadata END,metadata_updates) AS merged - FROM (SELECT *, checkpoint_step IS NOT NULL AND json_extract(metadata,'$."river:resumable_step"') IS NOT checkpoint_step AS use_fallback FROM choice) + -- Later sources replace whole top-level values, including null, like Postgres ||. + WITH sources(priority,doc) AS ( + SELECT 0,metadata + UNION ALL SELECT 1,checkpoint_fallback FROM choice + WHERE checkpoint_step IS NOT NULL AND json_extract(metadata,'$."river:resumable_step"') IS NOT checkpoint_step + UNION ALL SELECT 2,metadata_updates FROM choice + ), entries AS ( + SELECT priority,doc,key,fullkey FROM sources,json_each(doc) ) + SELECT jsonb_group_object(key,jsonb(doc->fullkey)) FROM entries AS entry + WHERE NOT EXISTS (SELECT 1 FROM entries AS later WHERE later.key=entry.key AND later.priority>entry.priority) ) END, errors=CASE WHEN state != 'running' OR ? IS NULL THEN errors ELSE jsonb_insert(CASE WHEN typeof(errors)='text' AND NOT json_valid(errors) THEN jsonb_array(errors) WHEN coalesce(json_type(errors),'array') <> 'array' THEN jsonb_array(json(errors)) ELSE coalesce(errors,jsonb('[]')) END,'$[#]',jsonb(?)) END WHERE id=? AND attempt=? AND attempted_at=? RETURNING {columns} @@ -224,7 +224,13 @@ state=CASE WHEN state IN ('completed','cancelled','discarded') THEN state ELSE ? finalized_at=CASE WHEN state IN ('completed','cancelled','discarded') THEN finalized_at ELSE ? END, scheduled_at=CASE WHEN state IN ('completed','cancelled','discarded') THEN scheduled_at ELSE ? END, attempt=CASE WHEN state IN ('completed','cancelled','discarded') THEN attempt ELSE ? END, -metadata=CASE WHEN typeof(metadata)='text' AND NOT json_valid(metadata) THEN metadata ELSE jsonb_patch(metadata,jsonb(?)) END, +-- Replace each top-level key like Postgres ||. jsonb_patch would merge nested objects and delete nulls. +metadata=CASE WHEN typeof(metadata)='text' AND NOT json_valid(metadata) THEN metadata ELSE (WITH updates(doc) AS (SELECT jsonb(?)) + SELECT jsonb_group_object(key,jsonb(doc->fullkey)) FROM ( + SELECT key,fullkey,metadata AS doc FROM json_each(metadata) + WHERE key NOT IN (SELECT key FROM updates,json_each(doc)) + UNION ALL SELECT key,fullkey,doc FROM updates,json_each(doc) + )) END, errors=CASE WHEN state IN ('completed','cancelled','discarded') OR ? IS NULL THEN errors ELSE jsonb_insert(CASE WHEN typeof(errors)='text' AND NOT json_valid(errors) THEN jsonb_array(errors) WHEN coalesce(json_type(errors),'array') <> 'array' THEN jsonb_array(json(errors)) ELSE coalesce(errors,jsonb('[]')) END,'$[#]',jsonb(?)) END WHERE state='running' AND id=? AND attempt=? AND attempted_at=? RETURNING {columns} -- name: periodic_scheduled_at diff --git a/java/river/src/test/java/com/riverqueue/ClientTest.java b/java/river/src/test/java/com/riverqueue/ClientTest.java index 882b4d3cb..aed0cd68b 100644 --- a/java/river/src/test/java/com/riverqueue/ClientTest.java +++ b/java/river/src/test/java/com/riverqueue/ClientTest.java @@ -48,6 +48,51 @@ void applicationTransactionOwnsCommitAndRollback() throws Exception { } } + @Test + void completionMetadataReplacesTopLevelKeys() { + var client = database(); + var database = client.database(); + var stored = + Json.parse( + """ + {"empty":{"old":true},"keep":{"x":1},"river:resumable_step":"a","river:resumable_cursor":{"c":1},"gone":5}"""); + var updates = + Json.parse( + """ + {"river:resumable_step":"c","river:resumable_cursor":{"d":7,"n":null},"gone":null}"""); + var updateObject = (tools.jackson.databind.node.ObjectNode) updates; + updateObject.put("", false); + updateObject.set("a\"b.c", Json.parse("[true,false,null]")); + updateObject.putNull("a\\b"); + updateObject.set("empty", Json.object()); + updateObject.put("large", 9_007_199_254_740_993L); + for (int index = 0; index < 1200; index++) + updateObject.set("key_" + index, Json.parse("{\"value\":" + index + ",\"nullable\":null}")); + var options = InsertOptions.builder().metadata(stored).build(); + var inserted = client.insert(EMAIL, new Email("merge@example.com"), options).job(); + var claimed = + client.transaction( + connection -> { + try (var statement = + Sql.prepare( + connection, + Sql.query(database, "claim"), + database.timestamp(NOW), + "worker", + "default", + database.timestamp(NOW), + 1); + var rows = statement.executeQuery()) { + assertTrue(rows.next()); + return client.read(rows); + } + }); + client.transaction(connection -> client.complete(connection, claimed, updates)); + var expected = ((tools.jackson.databind.node.ObjectNode) inserted.metadata()).deepCopy(); + updates.properties().forEach(entry -> expected.set(entry.getKey(), entry.getValue())); + assertEquals(expected, client.get(claimed.id()).metadata()); + } + @Test void crossKindUniqueConflictsPreserveTheIncumbent() { var client = database(); diff --git a/java/river/src/test/java/com/riverqueue/WorkersTest.java b/java/river/src/test/java/com/riverqueue/WorkersTest.java index c724b1ec9..b75176dd9 100644 --- a/java/river/src/test/java/com/riverqueue/WorkersTest.java +++ b/java/river/src/test/java/com/riverqueue/WorkersTest.java @@ -458,6 +458,49 @@ void oversizedSnoozeFailsTheAttemptInsteadOfStrandingIt() throws Exception { } } + // The subscription is intentionally held open until the worker completes. + @SuppressWarnings("try") + @Test + void resumableCursorUpdatesReplaceStoredCursors() throws Exception { + var client = client(); + var steps = new LinkedBlockingQueue(); + var completed = new CompletableFuture(); + try (var workers = + workers(client) + .retryPolicy(job -> Duration.ZERO) + .add( + TYPE, + context -> { + int attempt = context.job().attempt(); + context.step("a", () -> steps.add(attempt + ":a")); + context.stepWithCursor( + "c", + cursor -> { + steps.add(attempt + ":c"); + if (attempt > 1) return; + context.cursor(1); + throw new IllegalStateException("retry c"); + }); + context.stepWithCursor( + "d", + cursor -> { + steps.add(attempt + ":d:" + cursor.asInt(0)); + if (attempt > 2) return; + context.cursor(7); + throw new IllegalStateException("retry d"); + }); + }) + .start(); + var subscription = + workers.subscribe(completed::complete, Workers.EventKind.JOB_COMPLETED)) { + client.insert(TYPE, "resumable"); + assertEquals(3, completed.get(10, TimeUnit.SECONDS).job().attempt()); + } + // The third attempt resumes after "c"; a cursor left over from attempt one must not rerun it. + assertEquals( + java.util.List.of("1:a", "1:c", "2:c", "2:d:0", "3:d:7"), java.util.List.copyOf(steps)); + } + @Test void rescueCannotPreemptAConfiguredTimeout() { var builder = diff --git a/js/CHANGELOG.md b/js/CHANGELOG.md index 7c1896c13..64300a9ad 100644 --- a/js/CHANGELOG.md +++ b/js/CHANGELOG.md @@ -21,6 +21,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Replace whole top-level metadata values on SQLite, including JSON null, matching Postgres. Completed resumable cursor steps no longer run again because of stale cursor entries. [PR #1533](https://github.com/riverqueue/river/pull/1533). + - Preserve first-step resumable cursors on failure and step progress on snooze. Transactional checkpoints no longer leak into deferred updates after rollback or a failed write; rolling back a successful checkpoint write may replay earlier unpersisted steps. - Worker threads respect cancellation received while loading a handler module, before invoking its handler. diff --git a/js/driver/pg/src/driver.integration.test.ts b/js/driver/pg/src/driver.integration.test.ts index 25ecd0f93..9e1142f61 100644 --- a/js/driver/pg/src/driver.integration.test.ts +++ b/js/driver/pg/src/driver.integration.test.ts @@ -1308,6 +1308,82 @@ describe("PgDriver integration", () => { expect(raced[0]!.job!.metadata).not.toHaveProperty("stale_checkpoint"); }); + it("replaces top-level metadata keys on merge", async () => { + const kind = `${filePrefix}_merge`; + const queue = `${filePrefix}_merge`; + const worker = `${filePrefix}_merge_worker`; + const many = Object.fromEntries( + Array.from({ length: 1200 }, (_, index) => [ + `key_${index}`, + { value: index, nullable: null }, + ]) + ); + const stored = { + empty: { old: true }, + gone: 5, + keep: { x: 1 }, + "river:resumable_cursor": { c: 1 }, + "river:resumable_step": "a", + }; + const updates = { + ...many, + "": false, + 'a"b.c': [true, false, null], + "a\\b": null, + "&": true, + é: {}, + empty: {}, + gone: null, + "river:resumable_cursor": { d: 7, n: null }, + "river:resumable_step": "c", + }; + // The SQLite driver's tests expect this same result. + const merged = { + ...many, + "": false, + 'a"b.c': [true, false, null], + "a\\b": null, + "&": true, + é: {}, + empty: {}, + gone: null, + keep: { x: 1 }, + "river:resumable_cursor": { d: 7, n: null }, + "river:resumable_step": "c", + }; + const completed = await driver.jobInsert( + insertParams(kind, { metadata: stored, queue }) + ); + await driver.jobClaim({ + attemptedBy: worker, + kinds: [kind], + queues: [{ limit: 1, name: queue }], + }); + const updated = await driver.jobInsert( + insertParams(kind, { metadata: stored, queue }) + ); + + const [applied] = await driver.jobCompleteMany([ + { + attempt: 1, + attemptedBy: worker, + error: null, + finalizedAt: Temporal.Now.instant(), + id: completed.job.id, + kind: "complete", + metadata: updates, + output: null, + outputSet: false, + scheduledAt: null, + }, + ]); + expect(applied!.status).toBe("applied"); + expect(applied!.job!.metadata).toEqual(merged); + expect( + (await driver.jobUpdate(updated.job.id, { metadata: updates }))!.metadata + ).toEqual(merged); + }); + it("merges a rescued attempt's output without changing its state", async () => { const queue = `${filePrefix}_rescued_output`; const worker = `${filePrefix}_rescued_worker`; diff --git a/js/driver/sqlite/src/driver.test.ts b/js/driver/sqlite/src/driver.test.ts index 3506895e0..a9a41e008 100644 --- a/js/driver/sqlite/src/driver.test.ts +++ b/js/driver/sqlite/src/driver.test.ts @@ -1313,6 +1313,102 @@ describe("SqliteDriver", () => { ]); }); + test("replaces top-level metadata keys on merge, like Postgres", async () => { + const { driver } = await setup(); + const many = Object.fromEntries( + Array.from({ length: 1200 }, (_, index) => [ + `key_${index}`, + { value: index, nullable: null }, + ]) + ); + const stored = { + empty: { old: true }, + gone: 5, + keep: { x: 1 }, + "river:resumable_cursor": { c: 1 }, + "river:resumable_step": "a", + }; + const updates = { + ...many, + "": false, + 'a"b.c': [true, false, null], + "a\\b": null, + "&": true, + é: {}, + empty: {}, + gone: null, + "river:resumable_cursor": { d: 7, n: null }, + "river:resumable_step": "c", + }; + const merged = { + ...many, + "": false, + 'a"b.c': [true, false, null], + "a\\b": null, + "&": true, + é: {}, + empty: {}, + gone: null, + keep: { x: 1 }, + "river:resumable_cursor": { d: 7, n: null }, + "river:resumable_step": "c", + }; + const insert = async () => + ( + await driver.jobInsert({ + args: {}, + kind: "merge_job", + metadata: stored, + }) + ).job; + // Inserts add a unique nonce that every merge has to keep too. + const mergedFor = (job: SqliteJobRow) => ({ + ...merged, + "river:unique_nonce": job.metadata["river:unique_nonce"], + }); + const completion = (id: bigint) => ({ + attempt: 1, + attemptedBy: "merge-worker", + error: null, + finalizedAt: Temporal.Now.instant(), + id, + kind: "complete" as const, + metadata: updates, + output: null, + outputSet: false, + scheduledAt: null, + }); + + const completed = await insert(); + const stale = await insert(); + const updated = await insert(); + await driver.jobClaim({ + attemptedBy: "merge-worker", + kinds: ["merge_job"], + queues: [{ limit: 2, name: "default" }], + }); + // A second completion of the same attempt is stale but still merges. + await driver.jobCompleteMany([{ ...completion(stale.id), metadata: {} }]); + + const [applied, late] = await driver.jobCompleteMany([ + completion(completed.id), + completion(stale.id), + ]); + expect(applied!.status).toBe("applied"); + expect(applied!.job!.metadata).toEqual(mergedFor(completed)); + expect(late!.status).toBe("stale"); + expect(late!.job!.metadata).toEqual(mergedFor(stale)); + expect( + (await driver.jobUpdate(updated.id, { metadata: updates }))!.metadata + ).toEqual(mergedFor(updated)); + + // Key names are data, never JSON path syntax. + const odd = { "": 1, 'a"b': { new: 1 }, "a.b": [2], "a\\b": null }; + expect( + (await driver.jobUpdate(updated.id, { metadata: odd }))!.metadata + ).toEqual({ ...mergedFor(updated), ...odd }); + }); + test("polls the durable outbox and publishes cancellation IDs exactly", async () => { const { driver } = await setup(); const exactID = 9_007_199_254_740_999n; diff --git a/js/driver/sqlite/src/operations.ts b/js/driver/sqlite/src/operations.ts index 08744f713..4cb2e1443 100644 --- a/js/driver/sqlite/src/operations.ts +++ b/js/driver/sqlite/src/operations.ts @@ -87,6 +87,23 @@ const PREFILTER_METADATA_SQL = `(CASE WHEN ${invalidJsonTextSql("metadata")} THE /** SQL that is true when `metadata` can be read as JSON. */ const VALID_METADATA_SQL = `NOT ${invalidJsonTextSql("metadata")}`; +/** + * Replace whole top-level values, including JSON null, like Postgres's `||`. + * One JSON object is bound regardless of its size. Reading values with `->` + * preserves booleans and numeric tokens without a host-language round trip. + */ +const MERGE_METADATA_SQL = `(WITH updates(doc) AS (SELECT jsonb(?)) + SELECT CASE WHEN json_type(metadata) = 'object' THEN ( + SELECT jsonb_group_object(key, jsonb(doc -> fullkey)) + FROM ( + SELECT key, fullkey, metadata AS doc FROM json_each(metadata) + WHERE key NOT IN (SELECT key FROM updates, json_each(doc)) + UNION ALL + SELECT key, fullkey, doc FROM updates, json_each(doc) + ) + ) WHEN json_type(metadata) = 'array' THEN jsonb_insert(metadata, '$[#]', doc) + ELSE jsonb_array(json(metadata), doc) END FROM updates)`; + /** * Append the bound client ID to `attempted_by`, keeping the newest * `MAX_ATTEMPTED_BY - 1` (bound first) existing entries. Matches Go's @@ -286,7 +303,7 @@ export function completeJobs( END, metadata = CASE WHEN ? AND ${VALID_METADATA_SQL} - THEN jsonb_patch(json(metadata), json(?)) + THEN ${MERGE_METADATA_SQL} ELSE metadata END, scheduled_at = CASE @@ -332,7 +349,7 @@ export function completeJobs( database, `UPDATE river_job SET metadata = CASE WHEN ${VALID_METADATA_SQL} - THEN jsonb_patch(json(metadata), json(?)) ELSE metadata END + THEN ${MERGE_METADATA_SQL} ELSE metadata END WHERE id = ? AND state != 'running' AND attempt = ? @@ -550,8 +567,8 @@ function jsonContains(actual: JsonValue, expected: JsonValue): boolean { } /** - * Merge metadata and set output on a job with a JSON merge patch, like River - * for Go's SQLite `JobUpdate`. + * Merge metadata and set output on a job, replacing whole top-level values + * like the Postgres driver. */ export function updateJob( database: DatabaseSync, @@ -571,7 +588,7 @@ export function updateJob( database, `UPDATE river_job SET metadata = CASE - WHEN ? THEN jsonb_patch(json(metadata), json(?)) ELSE metadata END + WHEN ? THEN ${MERGE_METADATA_SQL} ELSE metadata END WHERE id = ? RETURNING ${JOB_COLUMNS}` ).get(hasMetadataPatch ? 1 : 0, encodeJson(metadataPatch, "metadata"), id); diff --git a/js/driver/sqlite/src/runtime.test.ts b/js/driver/sqlite/src/runtime.test.ts index 2e3e14ba1..afa7d3315 100644 --- a/js/driver/sqlite/src/runtime.test.ts +++ b/js/driver/sqlite/src/runtime.test.ts @@ -368,6 +368,64 @@ describe("SqliteDriver runtime", () => { } }); + test("drops a finished cursor step's cursor on later SQLite attempts", async () => { + const { driver } = await setup(); + const definition = defineJob({ kind: "sqlite_runtime_resumable_cursor" }); + const order: string[] = []; + const client = new Client(driver, { + clientId: "sqlite-runtime-resumable-cursor-test", + completionBatchSize: 1, + leaderElectionDisabled: true, + queues: { + default: { + fetchCooldown: { milliseconds: 1 }, + maxWorkers: 1, + pollInterval: { milliseconds: 5 }, + }, + }, + retryPolicy: (_job, now) => now.add({ hours: 1 }), + workers: new Workers().add(definition, async ({ job, resumable }) => { + await resumable.step("a", () => { + order.push("a"); + }); + await resumable.stepWithCursor("c", (cursor) => { + order.push(`c:${JSON.stringify(cursor)}`); + if (job.attempt === 1) { + resumable.setCursor(1); + throw new Error("retry inside c"); + } + }); + await resumable.stepWithCursor("d", (cursor) => { + order.push(`d:${JSON.stringify(cursor)}`); + if (job.attempt === 2) { + resumable.setCursor(7); + throw new Error("retry inside d"); + } + }); + }), + }); + const inserted = await client.insert(definition, {}); + const run = await client.start(); + try { + const first = await waitForJob(client, inserted.job.id, "retryable"); + expect(first.metadata["river:resumable_step"]).toBe("a"); + expect(first.metadata["river:resumable_cursor"]).toEqual({ c: 1 }); + await client.jobs.retry(inserted.job.id); + + const second = await waitForJob(client, inserted.job.id, "retryable"); + expect(second.metadata["river:resumable_step"]).toBe("c"); + await client.jobs.retry(inserted.job.id); + + await waitForJob(client, inserted.job.id, "completed"); + // The third attempt skips the finished `c` and resumes `d` at 7. + expect(order).toEqual(["a", "c:null", "c:1", "d:null", "d:7"]); + const done = await client.jobs.get(inserted.job.id); + expect(done?.metadata["river:resumable_cursor"]).toEqual({ d: 7 }); + } finally { + await run.stop(); + } + }); + test("fails jobs with invalid JSON columns without stalling the queue, like Go", async () => { const { database, driver } = await setup(); const job = defineJob({ kind: "sqlite_runtime_invalid_json" }); diff --git a/riverdriver/riverdrivertest/driver_client_test.go b/riverdriver/riverdrivertest/driver_client_test.go index 5dca04ec4..681905c1e 100644 --- a/riverdriver/riverdrivertest/driver_client_test.go +++ b/riverdriver/riverdrivertest/driver_client_test.go @@ -4,9 +4,12 @@ import ( "context" "database/sql" "encoding/json" + "errors" + "fmt" "maps" "math" "slices" + "sync" "testing" "time" @@ -1812,6 +1815,85 @@ func ExerciseClient[TTx any](ctx context.Context, t *testing.T, require.JSONEq(t, `{}`, string(fetchedQueue.Metadata)) }) + // A cursor step that completed on an earlier attempt isn't run again, even + // after a later cursor step stores its own cursor and fails. + t.Run("ResumableCursorStepNotRerunAfterCompletion", func(t *testing.T) { + t.Parallel() + + client, bundle := setup(t) + + type JobArgs struct { + testutil.JobArgsReflectKind[JobArgs] + } + + var ( + ranByAttempt = make(map[int][]string) + ranByAttemptMu sync.Mutex + ) + + river.AddWorker(bundle.config.Workers, river.WorkFunc(func(ctx context.Context, job *river.Job[JobArgs]) error { + ran := func(step string) { + ranByAttemptMu.Lock() + defer ranByAttemptMu.Unlock() + ranByAttempt[job.Attempt] = append(ranByAttempt[job.Attempt], step) + } + + river.ResumableStep(ctx, "a", nil, func(ctx context.Context) error { + ran("a") + return nil + }) + + river.ResumableStepCursor(ctx, "c", nil, func(ctx context.Context, cursor int) error { + ran(fmt.Sprintf("c:%d", cursor)) + if job.Attempt == 1 { + if err := river.ResumableSetCursor(ctx, 1); err != nil { + return err + } + return errors.New("step c failed") + } + return nil + }) + + river.ResumableStepCursor(ctx, "d", nil, func(ctx context.Context, cursor int) error { + ran(fmt.Sprintf("d:%d", cursor)) + if job.Attempt == 2 { + if err := river.ResumableSetCursor(ctx, 7); err != nil { + return err + } + return errors.New("step d failed") + } + return nil + }) + + return nil + })) + + subscribeChan := subscribe(t, client) + startClient(ctx, t, client) + + insertRes, err := client.Insert(ctx, &JobArgs{}, nil) + require.NoError(t, err) + + for range 2 { + event := riversharedtest.WaitOrTimeout(t, subscribeChan) + require.Equal(t, river.EventKindJobFailed, event.Kind) + + _, err := client.JobRetry(ctx, insertRes.Job.ID) + require.NoError(t, err) + } + + event := riversharedtest.WaitOrTimeout(t, subscribeChan) + require.Equal(t, river.EventKindJobCompleted, event.Kind) + + ranByAttemptMu.Lock() + defer ranByAttemptMu.Unlock() + require.Equal(t, map[int][]string{ + 1: {"a", "c:0"}, + 2: {"c:1", "d:0"}, + 3: {"d:7"}, + }, ranByAttempt) + }) + // SQLite only: a job with a JSON column that holds invalid JSON doesn't // stall its queue. Its attempt is failed like any undecodable job's (and // it's retried or discarded) with the invalid value left in place, and the diff --git a/riverdriver/riverdrivertest/job_update.go b/riverdriver/riverdrivertest/job_update.go index a4de886f8..150445d63 100644 --- a/riverdriver/riverdrivertest/job_update.go +++ b/riverdriver/riverdrivertest/job_update.go @@ -1282,6 +1282,42 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT require.ErrorContains(t, err, "error unmarshaling `tags`") }) + // A metadata merge replaces each top-level value outright instead of + // merging into it, and stores a JSON null instead of removing the key. + // A resumable job relies on this to drop the cursors of completed steps. + t.Run("MetadataUpdatesReplaceTopLevelValues", func(t *testing.T) { + t.Parallel() + + exec, _ := setup(ctx, t) + + now := time.Now().UTC() + + const ( + metadataBefore = `{"gone":5, "keep":{"x":1}, "river:resumable_cursor":{"c":1}, "river:resumable_step":"a"}` + metadataUpdates = `{"gone":null, "river:resumable_cursor":{"d":7, "n":null}, "river:resumable_step":"c"}` + metadataAfter = `{"gone":null, "keep":{"x":1}, "river:resumable_cursor":{"d":7, "n":null}, "river:resumable_step":"c"}` + ) + + runningJob := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{ + Metadata: []byte(metadataBefore), + State: new(rivertype.JobStateRunning), + }) + retryableJob := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{ + Metadata: []byte(metadataBefore), + ScheduledAt: new(now.Add(10 * time.Second)), + State: new(rivertype.JobStateRetryable), + }) + + jobsAfter, err := exec.JobSetStateIfRunningMany(ctx, setStateManyParams( + riverdriver.JobSetStateErrorRetryable(runningJob.ID, now, makeErrPayload(t, now), []byte(metadataUpdates)), + riverdriver.JobSetStateErrorRetryable(retryableJob.ID, now, makeErrPayload(t, now), []byte(metadataUpdates)), + )) + require.NoError(t, err) + require.Len(t, jobsAfter, 2) + require.JSONEq(t, metadataAfter, string(jobsAfter[0].Metadata)) + require.JSONEq(t, metadataAfter, string(jobsAfter[1].Metadata)) + }) + t.Run("UpdatesOnlyMetadataForAlreadyRetryableJobs", func(t *testing.T) { t.Parallel() @@ -1550,6 +1586,98 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT require.JSONEq(t, `{"key1":"val1","key2":"val2"}`, string(updatedJob.Metadata)) }) + t.Run("MetadataMergePreservesJSONValuesAndLiteralKeys", func(t *testing.T) { + t.Parallel() + + exec, _ := setup(ctx, t) + + job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{ + Metadata: []byte(`{"keep":123456789012345678901234567890,"\u0061":{"old":true},"a.b":{"old":true},"empty":{"old":true},"nullable":1}`), + }) + updatedJob, err := exec.JobUpdate(ctx, &riverdriver.JobUpdateParams{ + ID: job.ID, + MetadataDoMerge: true, + Metadata: []byte(`{"":false,"a":{"new":true},"a.b":{"new":true},"a\"b.c":[true,false,null],"a\\b":null,"&":true,"é":{},"empty":{},"nullable":null,"large":9007199254740993}`), + }) + require.NoError(t, err) + require.JSONEq(t, `{"keep":123456789012345678901234567890,"":false,"a":{"new":true},"a.b":{"new":true},"a\"b.c":[true,false,null],"a\\b":null,"&":true,"é":{},"empty":{},"nullable":null,"large":9007199254740993}`, string(updatedJob.Metadata)) + require.Equal(t, "123456789012345678901234567890", gjson.GetBytes(updatedJob.Metadata, "keep").Raw) + require.Equal(t, "9007199254740993", gjson.GetBytes(updatedJob.Metadata, "large").Raw) + require.True(t, gjson.GetBytes(updatedJob.Metadata, "a.new").Bool()) + }) + + t.Run("MetadataMergeReplacesTopLevelValues", func(t *testing.T) { + t.Parallel() + + exec, _ := setup(ctx, t) + + job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{ + Metadata: []byte(`{"gone":5, "keep":{"x":1}, "river:resumable_cursor":{"c":1}, "river:resumable_step":"a"}`), + }) + + updatedJob, err := exec.JobUpdate(ctx, &riverdriver.JobUpdateParams{ + ID: job.ID, + MetadataDoMerge: true, + Metadata: []byte(`{"gone":null, "river:resumable_cursor":{"d":7, "n":null}, "river:resumable_step":"c"}`), + }) + require.NoError(t, err) + require.JSONEq(t, + `{"gone":null, "keep":{"x":1}, "river:resumable_cursor":{"d":7, "n":null}, "river:resumable_step":"c"}`, + string(updatedJob.Metadata), + ) + }) + + t.Run("MetadataMergeWithManyKeys", func(t *testing.T) { + t.Parallel() + + exec, _ := setup(ctx, t) + + updates := make(map[string]any, 1200) + for i := range 1200 { + updates[fmt.Sprintf("key_%04d", i)] = map[string]any{"value": i, "nullable": nil} + } + encoded, err := json.Marshal(updates) + require.NoError(t, err) + job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{ + Metadata: []byte(`{"keep":true,"key_0000":{"old":true}}`), + }) + updatedJob, err := exec.JobUpdate(ctx, &riverdriver.JobUpdateParams{ + ID: job.ID, + MetadataDoMerge: true, + Metadata: encoded, + }) + require.NoError(t, err) + updates["keep"] = true + expected, err := json.Marshal(updates) + require.NoError(t, err) + require.JSONEq(t, string(expected), string(updatedJob.Metadata)) + }) + + t.Run("MetadataMergeWithNonObjectMetadata", func(t *testing.T) { + t.Parallel() + + for _, before := range []string{`[1,true,null]`, `true`, `42`, `"text"`, `null`} { + t.Run(before, func(t *testing.T) { + t.Parallel() + + exec, _ := setup(ctx, t) + + job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Metadata: []byte(before)}) + updatedJob, err := exec.JobUpdate(ctx, &riverdriver.JobUpdateParams{ + ID: job.ID, + MetadataDoMerge: true, + Metadata: []byte(`{"new":null}`), + }) + require.NoError(t, err) + expected := "[" + before + `,{"new":null}]` + if strings.HasPrefix(before, "[") { + expected = strings.TrimSuffix(before, "]") + `,{"new":null}]` + } + require.JSONEq(t, expected, string(updatedJob.Metadata)) + }) + } + }) + t.Run("NoArgs", func(t *testing.T) { t.Parallel() diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql index 6e651cbfe..8ef471786 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql @@ -696,10 +696,11 @@ RETURNING -- for JobSetStateIfRunning to use when falling back to non-running jobs. -- Metadata that isn't valid JSON is left in place, and columns are listed out -- to tolerate values that aren't valid JSON. See JobGetAvailable. +-- The shared metadata_merge.sql template applies top-level replacements. -- name: JobSetMetadataIfNotRunning :one UPDATE /* TEMPLATE: schema */river_job SET metadata = CASE WHEN typeof(metadata) <> 'text' OR json_valid(metadata) - THEN jsonb_patch(json(metadata), json(@metadata_updates)) + THEN /* TEMPLATE_BEGIN: metadata_merge_clause */ metadata /* TEMPLATE_END */ ELSE metadata END WHERE id = @id AND state != 'running' @@ -756,7 +757,7 @@ SET THEN cast(sqlc.narg('finalized_at') AS text) ELSE finalized_at END, metadata = CASE WHEN cast(@metadata_do_merge AS boolean) AND (typeof(metadata) <> 'text' OR json_valid(metadata)) - THEN jsonb_patch(json(metadata), json(@metadata_updates)) + THEN /* TEMPLATE_BEGIN: metadata_merge_clause */ metadata /* TEMPLATE_END */ ELSE metadata END, scheduled_at = CASE WHEN /* NOT should_cancel */(cast(@state AS text) <> 'available' AND @state <> 'retryable' AND @state <> 'scheduled' OR (CASE WHEN typeof(metadata) <> 'text' OR json_valid(metadata) THEN metadata -> 'cancel_attempted_at' END) IS NULL) AND cast(@scheduled_at_do_update AS boolean) THEN cast(@scheduled_at AS text) @@ -789,7 +790,7 @@ RETURNING -- name: JobUpdate :one UPDATE /* TEMPLATE: schema */river_job SET - metadata = CASE WHEN cast(@metadata_do_merge AS boolean) THEN jsonb_patch(json(metadata), json(@metadata)) ELSE metadata END + metadata = CASE WHEN cast(@metadata_do_merge AS boolean) THEN /* TEMPLATE_BEGIN: metadata_merge_clause */ metadata /* TEMPLATE_END */ ELSE metadata END WHERE id = @id RETURNING *; diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go index 0ff9958d2..8fc19993d 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go @@ -1801,9 +1801,9 @@ func (q *Queries) JobScheduleSetDiscarded(ctx context.Context, db DBTX, arg *Job const jobSetMetadataIfNotRunning = `-- name: JobSetMetadataIfNotRunning :one UPDATE /* TEMPLATE: schema */river_job SET metadata = CASE WHEN typeof(metadata) <> 'text' OR json_valid(metadata) - THEN jsonb_patch(json(metadata), json(?1)) + THEN /* TEMPLATE_BEGIN: metadata_merge_clause */ metadata /* TEMPLATE_END */ ELSE metadata END -WHERE id = ?2 +WHERE id = ?1 AND state != 'running' RETURNING id, @@ -1826,11 +1826,6 @@ RETURNING unique_states ` -type JobSetMetadataIfNotRunningParams struct { - MetadataUpdates interface{} - ID int64 -} - type JobSetMetadataIfNotRunningRow struct { ID int64 Column2 []byte @@ -1856,8 +1851,9 @@ type JobSetMetadataIfNotRunningRow struct { // for JobSetStateIfRunning to use when falling back to non-running jobs. // Metadata that isn't valid JSON is left in place, and columns are listed out // to tolerate values that aren't valid JSON. See JobGetAvailable. -func (q *Queries) JobSetMetadataIfNotRunning(ctx context.Context, db DBTX, arg *JobSetMetadataIfNotRunningParams) (*JobSetMetadataIfNotRunningRow, error) { - row := db.QueryRowContext(ctx, jobSetMetadataIfNotRunning, arg.MetadataUpdates, arg.ID) +// The shared metadata_merge.sql template applies top-level replacements. +func (q *Queries) JobSetMetadataIfNotRunning(ctx context.Context, db DBTX, id int64) (*JobSetMetadataIfNotRunningRow, error) { + row := db.QueryRowContext(ctx, jobSetMetadataIfNotRunning, id) var i JobSetMetadataIfNotRunningRow err := row.Scan( &i.ID, @@ -1907,15 +1903,15 @@ SET THEN cast(?8 AS text) ELSE finalized_at END, metadata = CASE WHEN cast(?9 AS boolean) AND (typeof(metadata) <> 'text' OR json_valid(metadata)) - THEN jsonb_patch(json(metadata), json(?10)) + THEN /* TEMPLATE_BEGIN: metadata_merge_clause */ metadata /* TEMPLATE_END */ ELSE metadata END, - scheduled_at = CASE WHEN /* NOT should_cancel */(cast(?1 AS text) <> 'available' AND ?1 <> 'retryable' AND ?1 <> 'scheduled' OR (CASE WHEN typeof(metadata) <> 'text' OR json_valid(metadata) THEN metadata -> 'cancel_attempted_at' END) IS NULL) AND cast(?11 AS boolean) - THEN cast(?12 AS text) + scheduled_at = CASE WHEN /* NOT should_cancel */(cast(?1 AS text) <> 'available' AND ?1 <> 'retryable' AND ?1 <> 'scheduled' OR (CASE WHEN typeof(metadata) <> 'text' OR json_valid(metadata) THEN metadata -> 'cancel_attempted_at' END) IS NULL) AND cast(?10 AS boolean) + THEN cast(?11 AS text) ELSE scheduled_at END, state = CASE WHEN /* should_cancel */((?1 = 'available' OR ?1 = 'retryable' OR ?1 = 'scheduled') AND (CASE WHEN typeof(metadata) <> 'text' OR json_valid(metadata) THEN metadata -> 'cancel_attempted_at' END) IS NOT NULL) THEN 'cancelled' ELSE ?1 END -WHERE id = ?13 +WHERE id = ?12 AND state = 'running' RETURNING id, @@ -1948,7 +1944,6 @@ type JobSetStateIfRunningParams struct { FinalizedAtDoUpdate bool FinalizedAt *string MetadataDoMerge bool - MetadataUpdates interface{} ScheduledAtDoUpdate bool ScheduledAt string ID int64 @@ -1994,7 +1989,6 @@ func (q *Queries) JobSetStateIfRunning(ctx context.Context, db DBTX, arg *JobSet arg.FinalizedAtDoUpdate, arg.FinalizedAt, arg.MetadataDoMerge, - arg.MetadataUpdates, arg.ScheduledAtDoUpdate, arg.ScheduledAt, arg.ID, @@ -2026,19 +2020,18 @@ func (q *Queries) JobSetStateIfRunning(ctx context.Context, db DBTX, arg *JobSet const jobUpdate = `-- name: JobUpdate :one UPDATE /* TEMPLATE: schema */river_job SET - metadata = CASE WHEN cast(?1 AS boolean) THEN jsonb_patch(json(metadata), json(?2)) ELSE metadata END -WHERE id = ?3 + metadata = CASE WHEN cast(?1 AS boolean) THEN /* TEMPLATE_BEGIN: metadata_merge_clause */ metadata /* TEMPLATE_END */ ELSE metadata END +WHERE id = ?2 RETURNING id, json(args), attempt, attempted_at, json(attempted_by), created_at, json(errors), finalized_at, kind, max_attempts, json(metadata), priority, queue, state, scheduled_at, json(tags), unique_key, unique_states ` type JobUpdateParams struct { MetadataDoMerge bool - Metadata interface{} ID int64 } func (q *Queries) JobUpdate(ctx context.Context, db DBTX, arg *JobUpdateParams) (*RiverJob, error) { - row := db.QueryRowContext(ctx, jobUpdate, arg.MetadataDoMerge, arg.Metadata, arg.ID) + row := db.QueryRowContext(ctx, jobUpdate, arg.MetadataDoMerge, arg.ID) var i RiverJob err := row.Scan( &i.ID, diff --git a/riverdriver/riversqlite/metadata_merge.sql b/riverdriver/riversqlite/metadata_merge.sql new file mode 100644 index 000000000..aebf7a4e9 --- /dev/null +++ b/riverdriver/riversqlite/metadata_merge.sql @@ -0,0 +1,24 @@ +-- Apply each supplied top-level value with jsonb_set, preserving JSON nulls. +-- Materialize the update list once; reevaluating it per step is costly on Turso. +(WITH RECURSIVE + original(doc) AS ( + SELECT CASE WHEN typeof(metadata) = 'text' AND NOT json_valid(metadata) + THEN '{}' ELSE metadata END + ), + updates AS MATERIALIZED ( + SELECT row_number() OVER () AS num, fullkey AS path, + @metadata_merge_updates -> fullkey AS value + FROM json_each(@metadata_merge_updates) + ), + merged(num, doc) AS ( + SELECT 0, doc FROM original + UNION ALL + SELECT updates.num, jsonb_set(merged.doc, updates.path, jsonb(updates.value)) + FROM merged JOIN updates ON updates.num = merged.num + 1 + ) + SELECT CASE json_type(original.doc) + WHEN 'object' THEN merged.doc + WHEN 'array' THEN jsonb(json_insert(json(original.doc), '$[#]', json(@metadata_merge_updates))) + ELSE jsonb_array(json(original.doc), jsonb(@metadata_merge_updates)) + END + FROM original, merged WHERE num = (SELECT count(*) FROM updates)) diff --git a/riverdriver/riversqlite/river_sqlite_driver.go b/riverdriver/riversqlite/river_sqlite_driver.go index 1fefd422a..7d1044bfc 100644 --- a/riverdriver/riversqlite/river_sqlite_driver.go +++ b/riverdriver/riversqlite/river_sqlite_driver.go @@ -974,17 +974,18 @@ func (e *Executor) JobSetStateIfRunningMany(ctx context.Context, params *riverdr setRes := make([]*rivertype.JobRow, 0, len(params.ID)) if err := dbutil.WithTx(ctx, e.executor(), func(ctx context.Context, execTx riverdriver.ExecutorTx) error { - ctx = schemaTemplateParam(ctx, params.Schema) dbtx := templateReplaceWrapper{dbtx: e.driver.UnwrapTx(execTx), replacer: &e.driver.replacer} // Should be a batch update, but that's currently impossible with SQLite/sqlc. https://github.com/sqlc-dev/sqlc/issues/3802 for i := range params.ID { + // Bind this job's metadata updates to the shared merge expression. + ctx := metadataMergeTemplateParam(schemaTemplateParam(ctx, params.Schema), params.MetadataUpdates[i]) + setStateParams := &dbsqlc.JobSetStateIfRunningParams{ - ID: params.ID[i], - Error: []byte("{}"), // even if not used, must be valid JSON because it's bed into the `json` function - MetadataUpdates: []byte("{}"), // even if not used, must be valid JSON because it's bed into the `json` function - Now: timeStringNullable(params.Now), - State: string(params.State[i]), + ID: params.ID[i], + Error: []byte("{}"), // even if not used, must be valid JSON because it's bed into the `json` function + Now: timeStringNullable(params.Now), + State: string(params.State[i]), } if params.Attempt[i] != nil { @@ -1001,7 +1002,6 @@ func (e *Executor) JobSetStateIfRunningMany(ctx context.Context, params *riverdr } if params.MetadataDoMerge[i] { setStateParams.MetadataDoMerge = true - setStateParams.MetadataUpdates = params.MetadataUpdates[i] } if params.ScheduledAt[i] != nil { setStateParams.ScheduledAtDoUpdate = true @@ -1012,10 +1012,7 @@ func (e *Executor) JobSetStateIfRunningMany(ctx context.Context, params *riverdr if err != nil { if errors.Is(err, sql.ErrNoRows) { var metadataRow *dbsqlc.JobSetMetadataIfNotRunningRow - metadataRow, err = dbsqlc.New().JobSetMetadataIfNotRunning(ctx, dbtx, &dbsqlc.JobSetMetadataIfNotRunningParams{ - ID: params.ID[i], - MetadataUpdates: sliceutil.FirstNonEmpty(params.MetadataUpdates[i], []byte("{}")), - }) + metadataRow, err = dbsqlc.New().JobSetMetadataIfNotRunning(ctx, dbtx, params.ID[i]) if err != nil { if errors.Is(err, sql.ErrNoRows) { continue @@ -1044,15 +1041,11 @@ func (e *Executor) JobSetStateIfRunningMany(ctx context.Context, params *riverdr } func (e *Executor) JobUpdate(ctx context.Context, params *riverdriver.JobUpdateParams) (*rivertype.JobRow, error) { - metadata := params.Metadata - if metadata == nil { - metadata = []byte("{}") - } + ctx = metadataMergeTemplateParam(schemaTemplateParam(ctx, params.Schema), params.Metadata) - job, err := dbsqlc.New().JobUpdate(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.JobUpdateParams{ + job, err := dbsqlc.New().JobUpdate(ctx, e.dbtx, &dbsqlc.JobUpdateParams{ ID: params.ID, MetadataDoMerge: params.MetadataDoMerge, - Metadata: metadata, }) if err != nil { return nil, interpretError(err) @@ -1884,6 +1877,21 @@ func migrationFromInternal(internal *dbsqlc.RiverMigration) *riverdriver.Migrati } } +// metadataMergeSQL is shared by job updates and both completion paths. sqlc +// cannot resolve this expression's CTEs, so use the existing template mechanism. +// The fixed SQL and parameter names keep the template replacer's cache enabled. +// +//go:embed metadata_merge.sql +var metadataMergeSQL string + +func metadataMergeTemplateParam(ctx context.Context, metadataUpdates []byte) context.Context { + return sqlctemplate.WithReplacements(ctx, map[string]sqlctemplate.Replacement{ + "metadata_merge_clause": {Stable: true, Value: metadataMergeSQL}, + }, map[string]any{ + "metadata_merge_updates": string(sliceutil.FirstNonEmpty(metadataUpdates, []byte("{}"))), + }) +} + func schemaTemplateParam(ctx context.Context, schema string) context.Context { if schema != "" { schema = dbutil.SafeIdentifier(schema) + "." diff --git a/ruby/CHANGELOG.md b/ruby/CHANGELOG.md index 100a996e6..de46723b2 100644 --- a/ruby/CHANGELOG.md +++ b/ruby/CHANGELOG.md @@ -20,6 +20,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Support large SQLite metadata updates without reaching the expression-depth limit, preserving top-level replacement and JSON null values. [PR #1533](https://github.com/riverqueue/river/pull/1533). + - A `River::Config` without an explicit `id:` generates its default client ID on first use in each process, so clients built before a fork (as with preloading Puma, Unicorn, or Pitchfork servers) no longer share one ID across forked workers. Explicit IDs are unchanged, and `Config#with` copies share the original's ID. [PR #1498](https://github.com/riverqueue/river/pull/1498). - Inserting a job stores its encoded args as given instead of decoding and re-encoding them in Ruby, which rewrote numbers like `1.10` as `1.1` and `1e3` as `1000.0` and rounded digits beyond `Float` precision. Postgres applies its usual `jsonb` normalization, and SQLite keeps the encoded bytes, matching River's Go drivers. Affects Active Record on Postgres and SQLite, and Sequel on SQLite. [PR #1503](https://github.com/riverqueue/river/pull/1503). - Wake workers when scheduled jobs become available or jobs are manually retried, including workers in other clients. [PR #1496](https://github.com/riverqueue/river/pull/1496). diff --git a/ruby/lib/driver/runtime.rb b/ruby/lib/driver/runtime.rb index f353230ca..7325aa657 100644 --- a/ruby/lib/driver/runtime.rb +++ b/ruby/lib/driver/runtime.rb @@ -670,12 +670,21 @@ def request_resign if runtime_postgres? "metadata || #{runtime_json(metadata)}" else - # Postgres's || replaces top-level values, including JSON null. - # JSON Merge Patch would recursively merge objects and delete nulls. - merged = metadata.reduce("metadata") do |expression, (key, value)| - path = runtime_quote("$.#{JSON.generate(key.to_s)}") - "jsonb_set(#{expression}, #{path}, jsonb(#{runtime_quote(JSON.generate(value))}))" - end + # Keep the SQL shape fixed regardless of the number of updated keys. + # -> preserves JSON booleans and large numeric tokens during aggregation. + merged = <<~SQL.strip + (WITH updates(doc) AS (SELECT #{runtime_json(metadata)}) + SELECT CASE WHEN json_type(metadata) = 'object' THEN ( + SELECT jsonb_group_object(key, jsonb(doc -> fullkey)) + FROM ( + SELECT key, fullkey, metadata AS doc FROM json_each(metadata) + WHERE key NOT IN (SELECT key FROM updates, json_each(doc)) + UNION ALL + SELECT key, fullkey, doc FROM updates, json_each(doc) + ) + ) WHEN json_type(metadata) = 'array' THEN jsonb_insert(metadata, '$[#]', doc) + ELSE jsonb_array(json(metadata), doc) END FROM updates) + SQL "CASE WHEN json_valid(metadata, 10) THEN #{merged} ELSE metadata END" end end diff --git a/ruby/spec/driver_shared_examples.rb b/ruby/spec/driver_shared_examples.rb index b6cda7868..399da71ab 100644 --- a/ruby/spec/driver_shared_examples.rb +++ b/ruby/spec/driver_shared_examples.rb @@ -134,10 +134,13 @@ class SimpleArgsWithInsertOpts < SimpleArgs it "merges metadata shallowly, preserving nulls and literal keys on both databases" do job = client.insert(SimpleArgs.new(job_num: 1), insert_opts: River::InsertOpts.new( - metadata: {"keep" => 1, "nested" => {"old" => true}, "nullable" => 2} + metadata: {"empty" => {"old" => true}, "keep" => 1, "nested" => {"old" => true}, "nullable" => 2} )).job updates = {'a"b' => "literal", "a.b" => [1, true], "a\\b" => false, "nested" => {"new" => true}, "nullable" => nil} + updates.merge!("" => false, "empty" => {}, "large" => 9_007_199_254_740_993) + 1_200.times { |index| updates["key_#{index}"] = {"value" => index, "nullable" => nil} } + driver.job_metadata_merge(job.id, updates) expect(client.job_get(job.id).metadata.to_h).to eq(job.metadata.to_h.merge(updates)) diff --git a/rust/CHANGELOG.md b/rust/CHANGELOG.md index 9909c7bd4..a83ef5c2d 100644 --- a/rust/CHANGELOG.md +++ b/rust/CHANGELOG.md @@ -20,6 +20,8 @@ Changes to River for Go are recorded in the [repository changelog](../CHANGELOG. ### Fixed +- Replace whole top-level metadata values on SQLite, including JSON null, matching Postgres. Completed resumable cursor steps no longer run again because of stale cursor entries. [PR #1533](https://github.com/riverqueue/river/pull/1533). + - Preserve first-step resumable cursors on failure and step progress on snooze. Transactional checkpoints no longer leak into deferred updates after rollback or a failed write; rolling back a successful checkpoint write may replay earlier unpersisted steps. ## [0.3.0] - 2026-10-07 diff --git a/rust/riverqueue/src/database/sqlite.rs b/rust/riverqueue/src/database/sqlite.rs index 4e8b26e21..2152e1341 100644 --- a/rust/riverqueue/src/database/sqlite.rs +++ b/rust/riverqueue/src/database/sqlite.rs @@ -95,6 +95,20 @@ const ERRORS_APPENDED: &str = concat!( ELSE jsonb(json_insert(json(coalesce(errors, jsonb('[]'))), '$[#]', json(?))) END" ); +/// Replace whole top-level values, including JSON null, like Postgres's `||`. +/// A single JSON object is bound; `->` preserves boolean and numeric tokens. +const METADATA_MERGED: &str = r#"(WITH updates(doc) AS (SELECT jsonb(?)) + SELECT CASE WHEN json_type(metadata) = 'object' THEN ( + SELECT jsonb_group_object(key, jsonb(doc -> fullkey)) + FROM ( + SELECT key, fullkey, metadata AS doc FROM json_each(metadata) + WHERE key NOT IN (SELECT key FROM updates, json_each(doc)) + UNION ALL + SELECT key, fullkey, doc FROM updates, json_each(doc) + ) + ) WHEN json_type(metadata) = 'array' THEN jsonb_insert(metadata, '$[#]', doc) + ELSE jsonb_array(json(metadata), doc) END FROM updates)"#; + const QUEUE_COLUMNS: &str = r#" created_at, json(metadata) AS metadata, @@ -1006,7 +1020,7 @@ pub(crate) async fn complete_decoded( END, -- Like River Go, metadata that isn't valid JSON is left in place. metadata = CASE - WHEN ? AND {metadata_valid} THEN jsonb_patch(json(metadata), json(?)) + WHEN ? AND {metadata_valid} THEN {METADATA_MERGED} ELSE metadata END, scheduled_at = CASE @@ -1058,7 +1072,7 @@ pub(crate) async fn merge_metadata_if_not_running( r#" UPDATE river_job SET metadata = CASE - WHEN {metadata_valid} THEN jsonb_patch(json(metadata), json(?)) + WHEN {metadata_valid} THEN {METADATA_MERGED} ELSE metadata END WHERE id = ? AND state != 'running' @@ -1082,7 +1096,7 @@ pub(crate) async fn update( let sql = format!( r#" UPDATE river_job - SET metadata = jsonb_patch(json(metadata), json(?)) + SET metadata = {METADATA_MERGED} WHERE id = ? RETURNING {JOB_COLUMNS} "# diff --git a/rust/riverqueue/tests/metadata_merge.rs b/rust/riverqueue/tests/metadata_merge.rs new file mode 100644 index 000000000..cc6c81e28 --- /dev/null +++ b/rust/riverqueue/tests/metadata_merge.rs @@ -0,0 +1,246 @@ +//! Metadata updates replace top-level keys on every database, as Postgres's +//! `||` does: nested objects aren't merged and a JSON null is stored as null. + +#![cfg(any(all(feature = "postgres", river_postgres_tests), feature = "sqlite"))] + +mod support; + +use riverqueue::{Client, InsertOpts, JobArgs, JobMetadata, JobRow, JobUpdateParams}; +use serde::{Deserialize, Serialize}; +use serde_json::{Map, Value, json}; + +#[derive(Clone, Debug, Deserialize, JobArgs, Serialize)] +#[river(kind = "metadata_merge")] +struct MergeArgs {} + +/// Updates a job's stored metadata through `client` and checks the merged +/// result, which must be the same on every database. +async fn assert_update_replaces_top_level_keys(client: &Client) { + let stored = json!({ + "empty": {"old": true}, + "gone": 5, + "keep": {"x": 1}, + "quoted \"key\".with.dots": {"old": true}, + "river:resumable_cursor": {"c": 1}, + "river:resumable_step": "a", + }); + let metadata: JobMetadata = stored.to_string().parse().unwrap(); + let inserted = client + .insert(MergeArgs {}) + .opts(InsertOpts::default().with_metadata(metadata)) + .await + .unwrap(); + + let updates = json!({ + "gone": null, + "new \"key\".with.dots": 1, + "quoted \"key\".with.dots": {"new": true}, + "river:resumable_cursor": {"d": 7, "n": null}, + "river:resumable_step": "c", + }); + let Value::Object(mut updates) = updates else { + unreachable!() + }; + updates.insert(String::new(), json!(false)); + updates.insert("a\\b".into(), Value::Null); + updates.insert("empty".into(), json!({})); + updates.insert("large".into(), json!(9_007_199_254_740_993_i64)); + for index in 0..1200 { + updates.insert( + format!("key_{index}"), + json!({"value": index, "nullable": null}), + ); + } + let expected_updates = updates.clone(); + let row = client + .jobs() + .update( + inserted.id(), + JobUpdateParams::default().metadata(Map::from_iter(updates)), + ) + .await + .unwrap(); + + let mut expected = json!({ + "gone": null, + "keep": {"x": 1}, + "new \"key\".with.dots": 1, + "quoted \"key\".with.dots": {"new": true}, + "river:resumable_cursor": {"d": 7, "n": null}, + "river:resumable_step": "c", + }); + expected.as_object_mut().unwrap().extend(expected_updates); + assert_eq!(stored_metadata(&row), expected); + let reread = client.jobs().get(inserted.id()).await.unwrap(); + assert_eq!(stored_metadata(&reread), expected); +} + +/// A row's metadata without the nonce a SQLite insert marks its row with. +fn stored_metadata(row: &JobRow) -> Value { + let mut metadata: Value = serde_json::from_str(row.metadata.as_raw().get()).unwrap(); + metadata + .as_object_mut() + .unwrap() + .remove("river:unique_nonce"); + metadata +} + +#[cfg(all(feature = "postgres", river_postgres_tests))] +#[tokio::test] +async fn postgres_update_replaces_top_level_keys() { + use riverqueue::database::PostgresDatabase; + + let schema = support::PostgresSchema::new("meta_merge").await; + let client = Client::builder( + PostgresDatabase::new(schema.pool.clone()).with_schema(schema.schema.clone()), + ) + .build() + .unwrap(); + + assert_update_replaces_top_level_keys(&client).await; + + schema.cleanup().await; +} + +#[cfg(feature = "sqlite")] +#[tokio::test] +async fn sqlite_update_replaces_top_level_keys() { + let (pool, path) = support::sqlite_file_pool(4).await; + let client = Client::builder(pool.clone()).build().unwrap(); + + assert_update_replaces_top_level_keys(&client).await; + + support::sqlite_cleanup(pool, path).await; +} + +#[cfg(feature = "sqlite")] +mod sqlite_resumable { + use std::{ + sync::{Arc, Mutex}, + time::Duration, + }; + + use riverqueue::{ + BoxError, Client, Job, JobArgs, JobState, QueueConfig, WorkContext, WorkError, WorkOutcome, + Worker, Workers, + }; + use serde::{Deserialize, Serialize}; + use serde_json::json; + + use super::support; + + #[derive(Clone, Debug, Deserialize, JobArgs, Serialize)] + #[river(kind = "metadata_merge_resumable")] + struct ResumableArgs {} + + /// One run of a step, as `(attempt, step, cursor)`. + type StepRun = (i32, &'static str, i64); + + /// Runs step `a`, then cursor steps `c` and `d`, recording each step run. + /// `c` fails on attempt 1 and `d` on attempt 2, each after setting a + /// cursor. + struct ResumableWorker { + runs: Arc>>, + } + + impl Worker for ResumableWorker { + type Error = BoxError; + + fn next_retry( + &self, + _job: &Job, + _error: &WorkError, + _now: chrono::DateTime, + ) -> Option { + Some(Duration::from_millis(10)) + } + + async fn work( + &self, + context: WorkContext, + job: Job, + ) -> Result { + let attempt = job.row.attempt; + context + .resumable_step("a", || async { + self.runs.lock().unwrap().push((attempt, "a", 0)); + Ok::<_, BoxError>(()) + }) + .await?; + for (step, failing_attempt, next_cursor) in [("c", 1, 1_i64), ("d", 2, 7)] { + let step_context = context.clone(); + context + .resumable_step_with_cursor(step, |cursor: i64| async move { + self.runs.lock().unwrap().push((attempt, step, cursor)); + if attempt == failing_attempt { + step_context.resumable_set_cursor(&next_cursor)?; + return Err("intentional resumable cursor failure".into()); + } + Ok::<_, BoxError>(()) + }) + .await?; + } + Ok(WorkOutcome::Complete) + } + } + + #[tokio::test] + async fn completed_cursor_step_is_not_rerun_after_a_later_step_fails() { + let (pool, path) = support::sqlite_file_pool(4).await; + let runs = Arc::new(Mutex::new(Vec::new())); + let mut workers = Workers::new(); + workers + .add(ResumableWorker { + runs: Arc::clone(&runs), + }) + .unwrap(); + let client = Client::builder(pool.clone()) + .workers(workers) + .queue( + "default", + QueueConfig::new(1) + .with_fetch_cooldown(Duration::from_millis(1)) + .with_fetch_poll_interval(Duration::from_millis(10)), + ) + .build() + .unwrap(); + let inserted = client.insert(ResumableArgs {}).await.unwrap(); + let mut run = client.start().unwrap(); + run.wait_ready().await.unwrap(); + + let row = tokio::time::timeout(Duration::from_secs(10), async { + loop { + let row = client.jobs().get(inserted.id()).await.unwrap(); + if row.state == JobState::Completed { + break row; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + run.stop().await.unwrap(); + + assert_eq!(row.attempt, 3); + // Attempt 3 skips `c`, which completed on attempt 2, and resumes `d` + // from the cursor it set. + assert_eq!( + *runs.lock().unwrap(), + [ + (1, "a", 0), + (1, "c", 0), + (2, "c", 1), + (2, "d", 0), + (3, "d", 7), + ] + ); + assert_eq!( + row.metadata + .get::("river:resumable_cursor") + .unwrap(), + Some(json!({"d": 7})) + ); + + support::sqlite_cleanup(pool, path).await; + } +}