Replace top-level metadata keys on SQLite merges

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.

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).

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).

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

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

java/river/src/test/java/com/riverqueue/ClientTest.java

@@ -48,6 +48,51 @@
     }
   }
 
+  @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();
@@ -248,7 +293,7 @@
     var peer = JobType.of("peer_only", Email.class);
     var unknown = river.insert(peer, new Email("peer@example.com")).job();
     var worked = new CountDownLatch(1);
     try (var workers =
         river
             .workers()
             .queue("default", 4)
@@ -482,7 +527,7 @@
     var river =
         new Client(
             TestDatabase.sqlite(directory.resolve("river.db")), Clock.fixed(NOW, ZoneOffset.UTC));
     try (var workers =
             river
                 .workers()
                 .queue("default", 1)
@@ -529,8 +574,8 @@
                       throw new IllegalStateException("failure after committed completion");
                     })
                 .start();
         var subscription = workers.subscribe(event::complete, Workers.EventKind.JOB_COMPLETED);
         var failures =
             workers.subscribe(
                 ignored -> failedEvents.incrementAndGet(), Workers.EventKind.JOB_FAILED)) {
       var inserted = river.insert(EMAIL, new Email("one@example.com"));
@@ -561,7 +606,7 @@
                       throw new IllegalStateException("retry me");
                     })
                 .start();
         var subscription = workers.subscribe(failed::complete, Workers.EventKind.JOB_FAILED)) {
       var inserted =
           river.insert(
               EMAIL,
@@ -593,7 +638,7 @@
                 .pollOnly(true)
                 .add(type, context -> worked.complete(context.job()))
                 .start();
         var subscription =
             workers.subscribe(completed::complete, Workers.EventKind.JOB_COMPLETED)) {
       var args = new Email("long-kind@example.com");
       var inserted = client.insert(type, args).job();

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<String>();
+    var completed = new CompletableFuture<Workers.Event>();
+    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 =

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.

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,
+      "<key>&": 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,
+      "<key>&": 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`;

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,
+      "<key>&": 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,
+      "<key>&": 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;