From 8b56bf8ba29d94dbee138c9dd546ab05abeeada2 Mon Sep 17 00:00:00 2001 From: Brandur Date: Fri, 6 Nov 2026 08:47:58 -0600 Subject: [PATCH 1/2] [java] Isolate batch peer decoding failures --- java/CHANGELOG.md | 4 + .../main/java/com/riverqueue/RunHandle.java | 42 +++-- .../main/java/com/riverqueue/WorkContext.java | 14 +- .../com/riverqueue/PeerCancellationTest.java | 7 +- .../java/com/riverqueue/PeerDecodingTest.java | 155 ++++++++++++++++++ 5 files changed, 200 insertions(+), 22 deletions(-) create mode 100644 java/river/src/test/java/com/riverqueue/PeerDecodingTest.java diff --git a/java/CHANGELOG.md b/java/CHANGELOG.md index 2151bca98..4389cafbc 100644 --- a/java/CHANGELOG.md +++ b/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 #XXX](https://github.com/riverqueue/river/pull/XXX). + ## [0.3.0] - 2026-11-02 ### Added diff --git a/java/river/src/main/java/com/riverqueue/RunHandle.java b/java/river/src/main/java/com/riverqueue/RunHandle.java index 83fb38d8f..832631639 100644 --- a/java/river/src/main/java/com/riverqueue/RunHandle.java +++ b/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)); } - List> claimPeers( + WorkContext.PeerClaims claimPeers( WorkContext parent, JobType type, Client.Transaction> operation) { parent.checkRuntimeCancellation(); var rows = @@ -540,12 +540,11 @@ List> 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. @@ -553,19 +552,29 @@ List> claimPeers( } var contexts = new ArrayList>(); 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) { @@ -573,7 +582,8 @@ void completePeer(WorkContext parent, WorkContext context, Exception failu 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); diff --git a/java/river/src/main/java/com/riverqueue/WorkContext.java b/java/river/src/main/java/com/riverqueue/WorkContext.java index edf0d0706..e6e254e6e 100644 --- a/java/river/src/main/java/com/riverqueue/WorkContext.java +++ b/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 java.util.List> claimPeers( + public PeerClaims claimPeers( JobType type, Client.Transaction> claim) { if (runtime == null) throw new IllegalStateException("No worker runtime"); return runtime.claimPeers(this, type, claim); @@ -93,7 +94,14 @@ WorkContext withDecodedJob(Job decoded) { return context; } - record Peer(Job row, WorkContext context) {} + record Peer(Job row, WorkContext context, Exception failure) {} + + /** Internal claim result; count includes undecodable peers retained for completion. */ + public record PeerClaims(int count, java.util.List> peers) { + public PeerClaims { + peers = java.util.List.copyOf(peers); + } + } /** * Waits for the cancellation signal. Call {@link #checkCancelled()} afterward if leaving work diff --git a/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java b/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java index 66d05fa2a..5e845a651 100644 --- a/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java +++ b/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java @@ -91,7 +91,7 @@ public JsonNode decode(Job 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 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; diff --git a/java/river/src/test/java/com/riverqueue/PeerDecodingTest.java b/java/river/src/test/java/com/riverqueue/PeerDecodingTest.java new file mode 100644 index 000000000..f9cf4ed5c --- /dev/null +++ b/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 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 job) { + if (job.id() == bad.id() && retryCompletion && completions.getAndIncrement() == 0) + throw new IllegalStateException("completion failed once"); + } + + @Override + public JsonNode decode(Job job) { + if (job.id() == bad.id() && failureKind.equals("pilot")) + throw new IllegalArgumentException("invalid peer payload"); + return job.args(); + } + }); + var claimed = new CompletableFuture>(); + var errors = new LinkedBlockingQueue(); + var events = new LinkedBlockingQueue(); + 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(); + 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()); + } + } +} From 25d21d7233ce95cf1ca056a0523e8d51a5708b50 Mon Sep 17 00:00:00 2001 From: Brandur Date: Fri, 6 Nov 2026 08:49:19 -0600 Subject: [PATCH 2/2] [java] Link peer decoding fix to PR #1586 --- java/CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/java/CHANGELOG.md b/java/CHANGELOG.md index 4389cafbc..c61efaabf 100644 --- a/java/CHANGELOG.md +++ b/java/CHANGELOG.md @@ -13,7 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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 #XXX](https://github.com/riverqueue/river/pull/XXX). +- 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