[java] Isolate batch peer decoding failures

java/CHANGELOG.md

@@ -11,6 +11,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Expose `Client.QUEUE_DEFAULT` (`"default"`) for queue configuration and operations, matching the default queue constant in other River clients.
 
+### Fixed
+
+- Isolate batch peer decoding failures so valid peers continue working, preserving each failed peer's own error through handler failures and completion retries. [PR #1586](https://github.com/riverqueue/river/pull/1586).
+
 ## [0.3.0] - 2026-11-02
 
 ### Added

java/river/src/main/java/com/riverqueue/RunHandle.java

@@ -530,7 +530,7 @@ private void workMiddleware(WorkContext<?> context, int index, WorkContext.Step
           .work(context, () -> workMiddleware(context, index + 1, next));
   }
 
-  <A> List<WorkContext<A>> claimPeers(
+  <A> WorkContext.PeerClaims<A> claimPeers(
       WorkContext<?> parent, JobType<A> type, Client.Transaction<List<Client.Decoded>> operation) {
     parent.checkRuntimeCancellation();
     var rows =
@@ -540,40 +540,50 @@ <A> List<WorkContext<A>> claimPeers(
               river.pilot().afterPeerClaim(connection, river.driver(), claimed);
               return claimed;
             });
-    // Register every committed claim before decoding any arguments. A decoding failure must still
-    // leave all claimed jobs owned by this attempt so its failure settles the entire batch.
+    // Register every committed claim before decoding so failures cannot orphan running peers.
     for (var decoded : rows) {
       var context = new WorkContext<>(river, decoded.job(), false);
       context.timeout = config.jobTimeout;
-      parent.peers.put(decoded.job().id(), new WorkContext.Peer(decoded.job(), context));
+      parent.peers.put(decoded.job().id(), new WorkContext.Peer(decoded.job(), context, null));
       activeThreads.put(decoded.job().id(), Thread.currentThread());
       active.put(decoded.job().id(), context);
       // Publish before checking the parent so a concurrent cancellation cannot miss both.
       context.attach(this, parent);
     }
     var contexts = new ArrayList<WorkContext<A>>();
     for (var decoded : rows) {
-      if (decoded.failure() != null) throw decoded.failure();
       var provisional = parent.peers.get(decoded.job().id()).context();
-      // Decode the plugin's wire representation once, including when evaluating the timeout.
-      var job = decoded.job().mapArgs(ignored -> river.pilot().decode(decoded.job()));
-      var context =
-          provisional.withDecodedJob(job.mapArgs(args -> Json.decode(args, type.argsType())));
-      var registration = config.handlers.get(job.kind());
-      context.timeout = registration == null ? config.jobTimeout : timeout(registration, job);
-      parent.peers.put(decoded.job().id(), new WorkContext.Peer(decoded.job(), context));
-      active.put(decoded.job().id(), context);
-      contexts.add(context);
+      try {
+        if (decoded.failure() != null) throw decoded.failure();
+        // Decode the pilot's wire representation once, including when evaluating the timeout.
+        var job = decoded.job().mapArgs(ignored -> river.pilot().decode(decoded.job()));
+        var context =
+            provisional.withDecodedJob(job.mapArgs(args -> Json.decode(args, type.argsType())));
+        var registration = config.handlers.get(job.kind());
+        context.timeout = registration == null ? config.jobTimeout : timeout(registration, job);
+        parent.peers.put(decoded.job().id(), new WorkContext.Peer(decoded.job(), context, null));
+        active.put(decoded.job().id(), context);
+        contexts.add(context);
+      } catch (Exception failure) {
+        // Cancellation still interrupts the shared batch thread rather than becoming a decode
+        // error.
+        if (provisional.isCancelled() || Thread.currentThread().isInterrupted()) throw failure;
+        parent.peers.put(
+            decoded.job().id(),
+            new WorkContext.Peer(decoded.job(), provisional, provisional.finish(failure)));
+        activeThreads.remove(decoded.job().id());
+      }
     }
-    return List.copyOf(contexts);
+    return new WorkContext.PeerClaims<>(rows.size(), contexts);
   }
 
   void completePeer(WorkContext<?> parent, WorkContext<?> context, Exception failure) {
     var peer = parent.peers.get(context.job().id());
     if (peer == null || peer.context() != context)
       throw new IllegalArgumentException("Peer is not owned by this batch attempt");
     if (parent.cancellation() != null) context.requestCancellation(parent.cancellation());
-    failure = context.finish(failure);
+    // Preserve decode errors across handler failures and retries of durable completion.
+    failure = context.finish(peer.failure() == null ? failure : peer.failure());
     activeThreads.remove(context.job().id());
     complete(peer.row(), context, failure);
     parent.peers.remove(context.job().id(), peer);

java/river/src/main/java/com/riverqueue/WorkContext.java

@@ -66,9 +66,10 @@ public A args() {
   /**
    * Internal companion-attempt seam used by River Pro's batch worker. Peers use the runtime timeout
    * during decoding, then their worker override. A peer that ignores cancellation beyond the stuck
-   * threshold interrupts the thread claiming and working the batch.
+   * threshold interrupts the thread claiming and working the batch. Decoding failures are omitted
+   * from the returned peers and fail only their own attempts when the parent finishes.
    */
-  public <T> java.util.List<WorkContext<T>> claimPeers(
+  public <T> PeerClaims<T> claimPeers(
       JobType<T> type, Client.Transaction<java.util.List<Client.Decoded>> claim) {
     if (runtime == null) throw new IllegalStateException("No worker runtime");
     return runtime.claimPeers(this, type, claim);
@@ -93,7 +94,14 @@ <T> WorkContext<T> withDecodedJob(Job<T> decoded) {
     return context;
   }
 
-  record Peer(Job<JsonNode> row, WorkContext<?> context) {}
+  record Peer(Job<JsonNode> row, WorkContext<?> context, Exception failure) {}
+
+  /** Internal claim result; count includes undecodable peers retained for completion. */
+  public record PeerClaims<T>(int count, java.util.List<WorkContext<T>> peers) {
+    public PeerClaims {
+      peers = java.util.List.copyOf(peers);
+    }
+  }
 
   /**
    * Waits for the cancellation signal. Call {@link #checkCancelled()} afterward if leaving work

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

@@ -91,7 +91,7 @@ public JsonNode decode(Job<JsonNode> job) {
                             }
                             return jobs;
                           });
-                  var peer = peers.getFirst();
+                  var peer = peers.peers().getFirst();
                   typed.complete(peer);
                   finish.join();
                   context.completePeer(peer, new InterruptedException("peer interrupted"));
@@ -207,13 +207,14 @@ public void work(WorkContext<String> context) throws Exception {
                               }
                               return jobs;
                             });
-                    assertEquals(1, peers.size());
+                    assertEquals(1, peers.count());
                     entered.complete(null);
                     try {
                       release.await();
                     } catch (InterruptedException error) {
                       assertEquals(
-                          WorkContext.Cancellation.TIMEOUT, peers.getFirst().cancellation());
+                          WorkContext.Cancellation.TIMEOUT,
+                          peers.peers().getFirst().cancellation());
                       assertNull(context.cancellation(), "The parent's timeout is disabled");
                       interrupted.complete(null);
                       throw error;

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

@@ -0,0 +1,155 @@
+package com.riverqueue;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import tools.jackson.databind.JsonNode;
+
+class PeerDecodingTest {
+  private static final JobType<String> TYPE = JobType.of("peer_decoding", String.class);
+  @RegisterExtension final TestDatabase databases = new TestDatabase();
+  @TempDir Path directory;
+
+  @ParameterizedTest
+  @CsvSource({
+    "arguments,false,false",
+    "arguments,true,false",
+    "arguments,false,true",
+    "metadata,false,false",
+    "pilot,false,false",
+    "row,false,false"
+  })
+  @SuppressWarnings("try") // Resources scope subscriptions and the background runtime.
+  void decodingFailuresRetainTheirOwnOutcomes(
+      String failureKind, boolean handlerFails, boolean retryCompletion) throws Exception {
+    var database = databases.open(directory.resolve("river.db"));
+    var raw = new Client(database);
+    var options = InsertOptions.builder().queue("peers").maxAttempts(1).build();
+    var first = raw.insert(TYPE, "first", options).job();
+    var bad =
+        raw.insert(
+                JobType.of(TYPE.kind(), JsonNode.class),
+                failureKind.equals("arguments") ? Json.object() : Json.tree("bad"),
+                InsertOptions.builder()
+                    .queue("peers")
+                    .maxAttempts(2)
+                    .metadata(
+                        failureKind.equals("metadata")
+                            ? Map.of("river:resumable_cursor", List.of(1))
+                            : Map.of())
+                    .build())
+            .job();
+    var last = raw.insert(TYPE, "last", options).job();
+    var completions = new AtomicInteger();
+    var client =
+        raw.withPilot(
+            new Pilot() {
+              @Override
+              public void afterAttempt(
+                  Connection connection, Client.Driver driver, Job<JsonNode> job) {
+                if (job.id() == bad.id() && retryCompletion && completions.getAndIncrement() == 0)
+                  throw new IllegalStateException("completion failed once");
+              }
+
+              @Override
+              public JsonNode decode(Job<JsonNode> job) {
+                if (job.id() == bad.id() && failureKind.equals("pilot"))
+                  throw new IllegalArgumentException("invalid peer payload");
+                return job.args();
+              }
+            });
+    var claimed = new CompletableFuture<WorkContext.PeerClaims<String>>();
+    var errors = new LinkedBlockingQueue<Throwable>();
+    var events = new LinkedBlockingQueue<Client.Event>();
+    try (var workers =
+            client.toBuilder()
+                .queue("default", 1)
+                .leadership(false)
+                .pollOnly(true)
+                .retryPolicy(job -> Duration.ofDays(1))
+                .errorHandler(errors::add)
+                .worker(
+                    TYPE,
+                    context -> {
+                      var peers =
+                          context.claimPeers(
+                              TYPE,
+                              connection -> {
+                                var jobs = new ArrayList<Client.Decoded>();
+                                var now = database.timestamp(Instant.now());
+                                try (var statement =
+                                        Sql.prepare(
+                                            connection,
+                                            Sql.query(database, "claim"),
+                                            now,
+                                            "batch",
+                                            "peers",
+                                            now,
+                                            3);
+                                    var rows = statement.executeQuery()) {
+                                  while (rows.next()) {
+                                    var decoded = client.readPartial(rows);
+                                    if (decoded.job().id() == bad.id() && failureKind.equals("row"))
+                                      decoded =
+                                          new Client.Decoded(
+                                              decoded.job(),
+                                              new IllegalArgumentException("invalid peer row"));
+                                    jobs.add(decoded);
+                                  }
+                                }
+                                return jobs;
+                              });
+                      claimed.complete(peers);
+                      if (handlerFails) throw new IllegalStateException("handler failed");
+                      for (var peer : peers.peers()) context.completePeer(peer, null);
+                    })
+                .build()
+                .start();
+        var subscription = workers.client().subscribe(events::add)) {
+      var leader =
+          client.insert(TYPE, "leader", InsertOptions.builder().maxAttempts(1).build()).job();
+      var peers = claimed.get(5, TimeUnit.SECONDS);
+      assertEquals(3, peers.count());
+      assertEquals(
+          List.of(first.id(), last.id()), peers.peers().stream().map(p -> p.job().id()).toList());
+      for (int i = 0; i < 4; i++) assertNotNull(events.poll(5, TimeUnit.SECONDS));
+      for (long id : List.of(leader.id(), first.id(), last.id())) {
+        var job = client.jobs().get(id);
+        assertEquals(handlerFails ? Job.State.DISCARDED : Job.State.COMPLETED, job.state());
+        assertEquals(handlerFails ? 1 : 0, job.errors().size());
+        if (handlerFails) assertEquals("handler failed", job.errors().getFirst().error());
+      }
+      var failed = client.jobs().get(bad.id());
+      assertEquals(Job.State.RETRYABLE, failed.state());
+      assertEquals(1, failed.attempt());
+      assertEquals(1, failed.errors().size());
+      String message = failed.errors().getFirst().error();
+      assertTrue(
+          message.contains(
+              switch (failureKind) {
+                case "arguments" -> "String";
+                case "metadata" -> "Invalid resumable cursor metadata";
+                case "pilot" -> "invalid peer payload";
+                case "row" -> "invalid peer row";
+                default -> throw new AssertionError(failureKind);
+              }),
+          message);
+      assertEquals(retryCompletion ? 1 : 0, errors.size(), errors.toString());
+      if (retryCompletion) assertEquals(2, completions.get());
+    }
+  }
+}