handler) {}
}
diff --git a/java/river/src/main/java/com/riverqueue/package-info.java b/java/river/src/main/java/com/riverqueue/package-info.java
index bfc4db07b..43930b474 100644
--- a/java/river/src/main/java/com/riverqueue/package-info.java
+++ b/java/river/src/main/java/com/riverqueue/package-info.java
@@ -1,8 +1,8 @@
/**
* Transactional River jobs on Postgres and SQLite. Define a {@link com.riverqueue.JobType}, insert
* through a {@link com.riverqueue.Client}, and register worker lambdas through {@link
- * com.riverqueue.Workers.Builder}. Only a running {@code Workers} instance owns resources; close it
- * or call {@code stop()} when the application stops.
+ * com.riverqueue.Workers}. Start the configured {@code Client} to obtain a {@code RunHandle}; close
+ * the handle or call {@code stop()} when the application stops.
*
* Connection-first overloads require an existing JDBC transaction with auto-commit disabled.
* They use savepoints and leave commit, rollback, and close to the caller. Other overloads open and
diff --git a/java/river/src/test/java/com/riverqueue/ClientLifecycleTest.java b/java/river/src/test/java/com/riverqueue/ClientLifecycleTest.java
new file mode 100644
index 000000000..6fda7cf5e
--- /dev/null
+++ b/java/river/src/test/java/com/riverqueue/ClientLifecycleTest.java
@@ -0,0 +1,261 @@
+package com.riverqueue;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.lang.reflect.Proxy;
+import java.nio.file.Path;
+import java.sql.SQLException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import javax.sql.DataSource;
+import org.junit.jupiter.api.Test;
+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.ValueSource;
+
+@SuppressWarnings("try") // Resources intentionally scope subscriptions and background activity.
+class ClientLifecycleTest {
+ private static final JobType TYPE = JobType.of("lifecycle_test", String.class);
+ @RegisterExtension final TestDatabase databases = new TestDatabase();
+ @TempDir Path directory;
+
+ private Client.Builder builder() {
+ return Client.builder(databases.open(directory.resolve("river.db")))
+ .leadership(false)
+ .pollInterval(Duration.ofMillis(10))
+ .pollOnly(true)
+ .queue("default", 1)
+ .stopTimeout(Duration.ofSeconds(5));
+ }
+
+ @Test
+ void clientCanInsertWithoutStarting() {
+ var database = databases.open(directory.resolve("river.db"));
+ var client = Client.builder(database).build();
+ var job = client.insert(TYPE, "insert only").job();
+ assertEquals(Job.State.AVAILABLE, client.get(job.id()).state());
+ assertThrows(IllegalArgumentException.class, client::start);
+ assertEquals(Job.State.AVAILABLE, new Client(database).get(job.id()).state());
+ }
+
+ @Test
+ void concurrentStartsAllowOnlyOneRun() throws Exception {
+ var client = builder().worker(TYPE, context -> {}).build();
+ var ready = new CountDownLatch(4);
+ var start = new CountDownLatch(1);
+ var runs = new LinkedBlockingQueue();
+ try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
+ var attempts = new ArrayList>();
+ for (int i = 0; i < 4; i++) {
+ attempts.add(
+ executor.submit(
+ () -> {
+ ready.countDown();
+ start.await();
+ try {
+ runs.add(client.start());
+ return true;
+ } catch (IllegalStateException expected) {
+ return false;
+ }
+ }));
+ }
+ try {
+ assertTrue(ready.await(5, TimeUnit.SECONDS));
+ start.countDown();
+ int started = 0;
+ for (var attempt : attempts) if (attempt.get(5, TimeUnit.SECONDS)) started++;
+ assertEquals(1, started);
+ assertEquals(1, runs.size());
+ } finally {
+ start.countDown();
+ // Wait for every contender before closing the winner: otherwise another could start.
+ for (var attempt : attempts) attempt.get(5, TimeUnit.SECONDS);
+ for (var run : runs) run.close();
+ }
+ }
+ }
+
+ @Test
+ void copiedConfigurationRecomputesDefaultRescueDelay() {
+ var builder = Client.builder(Database.connect("jdbc:sqlite::memory:"));
+ var defaultClient = builder.build();
+ var explicitClient = builder.rescueAfter(Duration.ofHours(1)).build();
+ assertDoesNotThrow(() -> defaultClient.toBuilder().jobTimeout(Duration.ofHours(2)).build());
+ assertEquals(
+ "rescueAfter must not be less than jobTimeout",
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> explicitClient.toBuilder().jobTimeout(Duration.ofHours(2)).build())
+ .getMessage());
+ }
+
+ @Test
+ void failedDatabaseStartupCanBeRetried() throws Exception {
+ var database = databases.open(directory.resolve("river.db"));
+ var fail = new AtomicBoolean(true);
+ var source =
+ (DataSource)
+ Proxy.newProxyInstance(
+ DataSource.class.getClassLoader(),
+ new Class[] {DataSource.class},
+ (proxy, method, args) -> {
+ if (!method.getName().equals("getConnection"))
+ throw new UnsupportedOperationException();
+ if (fail.compareAndSet(true, false))
+ throw new SQLException("temporary startup failure");
+ return database.connection();
+ });
+ var client =
+ Client.builder(new Database(source, database.dialect()).withSchema(database.schema()))
+ .leadership(false)
+ .pollOnly(true)
+ .queue("default", 1)
+ .workers(new Workers().add(TYPE, context -> {}))
+ .build();
+ assertEquals(
+ RiverException.Code.DATABASE, assertThrows(RiverException.class, client::start).code());
+ var events = new LinkedBlockingQueue();
+ try (var subscription = client.subscribe(events::add, Client.EventKind.JOB_COMPLETED);
+ var run = client.start()) {
+ var job = client.insert(TYPE, "retry").job();
+ var event = events.poll(5, TimeUnit.SECONDS);
+ assertNotNull(event);
+ assertEquals(job.id(), event.job().id());
+ }
+ }
+
+ @Test
+ void failedPilotStartupCanBeRetried() {
+ var fail = new AtomicBoolean(true);
+ var client =
+ builder()
+ .worker(TYPE, context -> {})
+ .build()
+ .withPilot(
+ new Pilot() {
+ @Override
+ public void starting(boolean leadership) {
+ if (fail.compareAndSet(true, false))
+ throw new IllegalStateException("temporary pilot failure");
+ }
+ });
+ assertEquals(
+ "temporary pilot failure",
+ assertThrows(IllegalStateException.class, client::start).getMessage());
+ try (var run = client.start()) {
+ assertSame(client, run.client());
+ }
+ }
+
+ @Test
+ void sameClientInsertWakesRuntimeWithoutPolling() throws Exception {
+ var fetched = new CompletableFuture();
+ var worked = new CompletableFuture();
+ var client =
+ builder()
+ .pollInterval(Duration.ofHours(1))
+ .worker(TYPE, context -> worked.complete(context.client()))
+ .build()
+ .withPilot(
+ new Pilot() {
+ @Override
+ public java.util.List claim(
+ java.sql.Connection connection,
+ Client.Driver driver,
+ Claim claim,
+ Client.Transaction> next)
+ throws Exception {
+ var rows = next.run(connection);
+ if (rows.isEmpty()) fetched.complete(null);
+ return rows;
+ }
+ });
+ try (var run = client.start()) {
+ fetched.get(5, TimeUnit.SECONDS);
+ client.insert(TYPE, "wake");
+ assertSame(client, worked.get(5, TimeUnit.SECONDS));
+ }
+ }
+
+ @Test
+ void subscriptionsSurviveRestartAndOldHandlesCannotStopNewRuns() throws Exception {
+ var client = builder().workers(new Workers().add(TYPE, context -> {})).build();
+ var events = new LinkedBlockingQueue();
+ var removed = new LinkedBlockingQueue();
+ try (var subscription = client.subscribe(events::add, Client.EventKind.JOB_COMPLETED)) {
+ // Already queued work may finish immediately after start; subscribe beforehand.
+ var firstJob = client.insert(TYPE, "first").job();
+ var first = client.start();
+ try (first) {
+ assertEquals(client.id(), first.id());
+ assertThrows(IllegalStateException.class, client::start);
+ var event = events.poll(5, TimeUnit.SECONDS);
+ assertNotNull(event);
+ assertEquals(firstJob.id(), event.job().id());
+ }
+ var closed = client.subscribe(removed::add);
+ closed.close();
+ closed.close();
+ try (var second = client.start()) {
+ assertNotSame(first, second);
+ assertEquals(first.id(), second.id());
+ first.stopAndCancel();
+ first.close();
+ assertThrows(IllegalStateException.class, client::start);
+ var secondJob = client.insert(TYPE, "second").job();
+ var event = events.poll(5, TimeUnit.SECONDS);
+ assertNotNull(event);
+ assertEquals(secondJob.id(), event.job().id());
+ assertTrue(removed.isEmpty());
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void timedOutStopPreventsOverlappingRuns(boolean cancel) throws Exception {
+ var entered = new CountDownLatch(1);
+ var release = new CountDownLatch(1);
+ var events = new LinkedBlockingQueue();
+ var client =
+ builder()
+ .stopTimeout(Duration.ofMillis(100))
+ .worker(
+ TYPE,
+ context -> {
+ entered.countDown();
+ release.await();
+ })
+ .build();
+ try (var subscription = client.subscribe(events::add)) {
+ var run = client.start();
+ try {
+ client.insert(TYPE, "blocked");
+ assertTrue(entered.await(5, TimeUnit.SECONDS));
+ assertThrows(
+ RiverException.class,
+ () -> {
+ if (cancel) run.stopAndCancel();
+ else run.stop();
+ });
+ assertThrows(IllegalStateException.class, client::start);
+ } finally {
+ release.countDown();
+ assertNotNull(events.poll(5, TimeUnit.SECONDS));
+ run.stop();
+ }
+ try (var restarted = client.start()) {
+ assertEquals(run.id(), restarted.id());
+ }
+ }
+ }
+}
diff --git a/java/river/src/test/java/com/riverqueue/ClientTest.java b/java/river/src/test/java/com/riverqueue/ClientTest.java
index dfb0d6e90..2bb20316f 100644
--- a/java/river/src/test/java/com/riverqueue/ClientTest.java
+++ b/java/river/src/test/java/com/riverqueue/ClientTest.java
@@ -295,13 +295,13 @@ void fetchOnlyKnownKindsLeavesPeerJobsAvailable() throws Exception {
var unknown = river.insert(peer, new Email("peer@example.com")).job();
var worked = new CountDownLatch(1);
try (var workers =
- river
- .workers()
+ river.toBuilder()
.queue("default", 4)
.leadership(false)
.pollOnly(true)
.fetchOnlyKnownKinds(true)
- .add(EMAIL, context -> worked.countDown())
+ .worker(EMAIL, context -> worked.countDown())
+ .build()
.start()) {
river.insert(EMAIL, new Email("java@example.com"));
assertTrue(worked.await(10, TimeUnit.SECONDS));
@@ -320,18 +320,18 @@ void removingQueueDrainsItsCommittedClaims() throws Exception {
var first = river.insert(EMAIL, new Email("first@example.com")).job();
var second = river.insert(EMAIL, new Email("second@example.com")).job();
try (var workers =
- river
- .workers()
+ river.toBuilder()
.queue("default", 1)
.leadership(false)
.pollOnly(true)
- .add(
+ .worker(
EMAIL,
context -> {
calls.incrementAndGet();
entered.countDown();
release.await();
})
+ .build()
.start()) {
assertTrue(entered.await(10, TimeUnit.SECONDS));
var removal =
@@ -364,16 +364,15 @@ void workerCompletionCommitsApplicationChangesWithJob() throws Exception {
return null;
});
var release = new CountDownLatch(1);
- var done = new CompletableFuture();
+ var done = new CompletableFuture();
var failed = new CompletableFuture();
try (var workers =
- river
- .workers()
+ river.toBuilder()
.queue("default", 1)
.leadership(false)
.pollInterval(Duration.ofMillis(10))
.errorHandler(failed::complete)
- .add(
+ .worker(
EMAIL,
context -> {
release.await();
@@ -391,11 +390,14 @@ void workerCompletionCommitsApplicationChangesWithJob() throws Exception {
return context.complete(c);
});
})
+ .build()
.start()) {
- workers.subscribe(
- event -> {
- if (event.kind() == Workers.EventKind.JOB_COMPLETED) done.complete(event);
- });
+ workers
+ .client()
+ .subscribe(
+ event -> {
+ if (event.kind() == Client.EventKind.JOB_COMPLETED) done.complete(event);
+ });
var inserted = river.insert(EMAIL, new Email("one@example.com"));
release.countDown();
var completed = done.get(10, TimeUnit.SECONDS);
@@ -530,12 +532,12 @@ void sqliteQueueNotificationFailureRollsBackQueueMutation() throws Exception {
new Client(
TestDatabase.sqlite(directory.resolve("river.db")), Clock.fixed(NOW, ZoneOffset.UTC));
try (var workers =
- river
- .workers()
+ river.toBuilder()
.queue("default", 1)
.leadership(false)
.pollOnly(true)
- .add(EMAIL, ignored -> {})
+ .worker(EMAIL, ignored -> {})
+ .build()
.start();
var connection = river.database().connection()) {
connection.setAutoCommit(false);
@@ -560,15 +562,14 @@ void sqliteQueueNotificationFailureRollsBackQueueMutation() throws Exception {
@SuppressWarnings("try") // Resources intentionally scope background activity.
void committedCompletionWinsOverSubsequentHandlerFailure() throws Exception {
var river = database();
- var event = new CompletableFuture();
+ var event = new CompletableFuture();
var failedEvents = new java.util.concurrent.atomic.AtomicInteger();
try (var workers =
- river
- .workers()
+ river.toBuilder()
.queue("default", 1)
.leadership(false)
.pollOnly(true)
- .add(
+ .worker(
EMAIL,
context -> {
Job completed = context.transaction(context::complete);
@@ -576,11 +577,15 @@ void committedCompletionWinsOverSubsequentHandlerFailure() throws Exception {
assertEquals(Job.State.COMPLETED, completed.state());
throw new IllegalStateException("failure after committed completion");
})
+ .build()
.start();
- var subscription = workers.subscribe(event::complete, Workers.EventKind.JOB_COMPLETED);
+ var subscription =
+ workers.client().subscribe(event::complete, Client.EventKind.JOB_COMPLETED);
var failures =
- workers.subscribe(
- ignored -> failedEvents.incrementAndGet(), Workers.EventKind.JOB_FAILED)) {
+ workers
+ .client()
+ .subscribe(
+ ignored -> failedEvents.incrementAndGet(), Client.EventKind.JOB_FAILED)) {
var inserted = river.insert(EMAIL, new Email("one@example.com"));
assertEquals(inserted.job().id(), event.get(10, TimeUnit.SECONDS).job().id());
}
@@ -592,10 +597,9 @@ void committedCompletionWinsOverSubsequentHandlerFailure() throws Exception {
void retryPolicyReceivesFailedAttemptSnapshot() throws Exception {
var river = database();
var observed = new CompletableFuture>();
- var failed = new CompletableFuture();
+ var failed = new CompletableFuture();
try (var workers =
- river
- .workers()
+ river.toBuilder()
.queue("default", 1)
.leadership(false)
.pollOnly(true)
@@ -604,13 +608,15 @@ void retryPolicyReceivesFailedAttemptSnapshot() throws Exception {
observed.complete(job);
return Duration.ofDays(1);
})
- .add(
+ .worker(
EMAIL,
ignored -> {
throw new IllegalStateException("retry me");
})
+ .build()
.start();
- var subscription = workers.subscribe(failed::complete, Workers.EventKind.JOB_FAILED)) {
+ var subscription =
+ workers.client().subscribe(failed::complete, Client.EventKind.JOB_FAILED)) {
var inserted =
river.insert(
EMAIL,
@@ -633,18 +639,18 @@ void jobKindLengthMatchesDatabaseConstraint() throws Exception {
var type = JobType.of("k".repeat(127), Email.class);
String oversized = "k".repeat(128);
assertThrows(IllegalArgumentException.class, () -> JobType.of(oversized, Email.class));
- var completed = new CompletableFuture();
+ var completed = new CompletableFuture();
var worked = new CompletableFuture>();
try (var workers =
- client
- .workers()
+ client.toBuilder()
.queue("default", 1)
.leadership(false)
.pollOnly(true)
- .add(type, context -> worked.complete(context.job()))
+ .worker(type, context -> worked.complete(context.job()))
+ .build()
.start();
var subscription =
- workers.subscribe(completed::complete, Workers.EventKind.JOB_COMPLETED)) {
+ workers.client().subscribe(completed::complete, Client.EventKind.JOB_COMPLETED)) {
var args = new Email("long-kind@example.com");
var inserted = client.insert(type, args).job();
assertEquals(type.kind(), inserted.kind());
@@ -696,19 +702,19 @@ public void afterAttempt(
var entered = new CompletableFuture>();
var release = new CountDownLatch(1);
var workers =
- river
- .workers()
+ river.toBuilder()
.queue("default", 1)
.leadership(false)
.pollOnly(true)
.stopTimeout(Duration.ofMillis(100))
.stuckThreshold(Duration.ofMinutes(1))
- .add(
+ .worker(
EMAIL,
context -> {
entered.complete(context);
release.await();
})
+ .build()
.start();
try {
river.insert(EMAIL, new Email("one@example.com"));
diff --git a/java/river/src/test/java/com/riverqueue/HookMiddlewareTest.java b/java/river/src/test/java/com/riverqueue/HookMiddlewareTest.java
index cb47eda4b..cd6abd7a0 100644
--- a/java/river/src/test/java/com/riverqueue/HookMiddlewareTest.java
+++ b/java/river/src/test/java/com/riverqueue/HookMiddlewareTest.java
@@ -202,9 +202,10 @@ void workBeginFailureSkipsHandlerAndEndHooks() throws Exception {
trace.add("after");
}
});
- var completed = new CompletableFuture();
- try (var workers = workers(river).add(TYPE, context -> trace.add("handler")).start();
- var subscription = workers.subscribe(completed::complete)) {
+ var completed = new CompletableFuture();
+ try (var workers =
+ workers(river).worker(TYPE, context -> trace.add("handler")).build().start();
+ var subscription = workers.client().subscribe(completed::complete)) {
river.insert(TYPE, "one", InsertOptions.builder().maxAttempts(1).build());
assertEquals(Job.State.DISCARDED, completed.get(5, TimeUnit.SECONDS).job().state());
}
@@ -270,10 +271,10 @@ void workEndHooksChainOutcomesInsideMiddleware(String outcome, String action) th
}
});
}
- var completed = new CompletableFuture();
+ var completed = new CompletableFuture();
try (var workers =
workers(river)
- .add(
+ .worker(
TYPE,
context -> {
trace.add("handler");
@@ -284,8 +285,9 @@ void workEndHooksChainOutcomesInsideMiddleware(String outcome, String action) th
default -> {}
}
})
+ .build()
.start();
- var subscription = workers.subscribe(completed::complete)) {
+ var subscription = workers.client().subscribe(completed::complete)) {
river.insert(TYPE, "one", InsertOptions.builder().maxAttempts(1).build());
var expected =
action.equals("clear") || (outcome.equals("success") && action.equals("preserve"))
@@ -302,9 +304,8 @@ void workEndHooksChainOutcomesInsideMiddleware(String outcome, String action) th
if (action.equals("clear")) assertNull(observed.get());
}
- private Workers.Builder workers(Client river) {
- return river
- .workers()
+ private Client.Builder workers(Client river) {
+ return river.toBuilder()
.queue("default", 1)
.leadership(false)
.pollOnly(true)
diff --git a/java/river/src/test/java/com/riverqueue/IdentifierTest.java b/java/river/src/test/java/com/riverqueue/IdentifierTest.java
index 1f6bbad26..8f6b4743a 100644
--- a/java/river/src/test/java/com/riverqueue/IdentifierTest.java
+++ b/java/river/src/test/java/com/riverqueue/IdentifierTest.java
@@ -10,8 +10,8 @@
class IdentifierTest {
private static final JobType TYPE = JobType.of("identifier_test", String.class);
- private final Workers.Builder workers =
- new Client(Database.connect("jdbc:sqlite::memory:")).workers();
+ private final Client.Builder workers =
+ new Client(Database.connect("jdbc:sqlite::memory:")).toBuilder();
@Test
void acceptsAnonymousPeriodicJobs() {
diff --git a/java/river/src/test/java/com/riverqueue/LeadershipTest.java b/java/river/src/test/java/com/riverqueue/LeadershipTest.java
index b66b7209e..95cc57bf3 100644
--- a/java/river/src/test/java/com/riverqueue/LeadershipTest.java
+++ b/java/river/src/test/java/com/riverqueue/LeadershipTest.java
@@ -75,8 +75,7 @@ public void start() {
}
});
try (var workers =
- client
- .workers()
+ client.toBuilder()
.queue("default", 1)
.pollOnly(true)
.serviceInterval(Duration.ofHours(1))
@@ -86,7 +85,8 @@ public void start() {
errors.add(error);
reported.join();
})
- .add(TYPE, context -> {})
+ .worker(TYPE, context -> {})
+ .build()
.start()) {
try {
assertNotNull(errors.poll(5, TimeUnit.SECONDS));
@@ -139,8 +139,7 @@ void leaseCoversTheRenewalInterval(long seconds) throws Exception {
var ready = new CountDownLatch(1);
var interval = Duration.ofSeconds(seconds);
try (var workers =
- client
- .workers()
+ client.toBuilder()
.queue("default", 1)
.serviceInterval(interval)
.stopTimeout(Duration.ofSeconds(5))
@@ -150,7 +149,8 @@ void leaseCoversTheRenewalInterval(long seconds) throws Exception {
if (event.equals("listen_ready")) ready.countDown();
if (event.equals("leadership_check")) checked.add(event);
})
- .add(TYPE, context -> {})
+ .worker(TYPE, context -> {})
+ .build()
.start()) {
assertTrue(ready.await(5, TimeUnit.SECONDS));
assertNotNull(checked.poll(5, TimeUnit.SECONDS));
@@ -198,14 +198,14 @@ public void start() {
}
});
try (var workers =
- client
- .workers()
+ client.toBuilder()
.queue("default", 1)
.pollOnly(true)
.serviceInterval(Duration.ofMillis(20))
.stopTimeout(Duration.ofSeconds(5))
.errorHandler(errors::add)
- .add(TYPE, context -> {})
+ .worker(TYPE, context -> {})
+ .build()
.start()) {
assertNotNull(starts.poll(5, TimeUnit.SECONDS));
var original = lease(client);
@@ -235,8 +235,7 @@ void resignationWakesPeersOnBothDatabases() throws Exception {
var secondCheck = new CountDownLatch(1);
var errors = new LinkedBlockingQueue();
var first =
- client
- .workers()
+ client.toBuilder()
.queue("default", 1)
.id("first")
.serviceInterval(Duration.ofHours(1))
@@ -247,7 +246,8 @@ void resignationWakesPeersOnBothDatabases() throws Exception {
if (event.equals("listen_ready")) ready.countDown();
if (event.equals("leadership_check")) firstCheck.countDown();
})
- .add(TYPE, context -> {})
+ .worker(TYPE, context -> {})
+ .build()
.start();
try (first) {
assertTrue(firstCheck.await(5, TimeUnit.SECONDS));
@@ -260,8 +260,7 @@ public void start() {
}
});
try (var second =
- follower
- .workers()
+ follower.toBuilder()
.queue("default", 1)
.id("second")
.serviceInterval(Duration.ofHours(1))
@@ -272,7 +271,8 @@ public void start() {
if (event.equals("listen_ready")) ready.countDown();
if (event.equals("leadership_check")) secondCheck.countDown();
})
- .add(TYPE, context -> {})
+ .worker(TYPE, context -> {})
+ .build()
.start()) {
assertTrue(ready.await(5, TimeUnit.SECONDS));
assertTrue(secondCheck.await(5, TimeUnit.SECONDS));
@@ -312,8 +312,7 @@ public boolean rescue(Client client, Rescue request) {
}
});
var first =
- blocked
- .workers()
+ blocked.toBuilder()
.queue("default", 1)
.id("old-leader")
.serviceInterval(Duration.ofHours(1))
@@ -323,7 +322,8 @@ public boolean rescue(Client client, Rescue request) {
event -> {
if (event.equals("listen_ready")) ready.countDown();
})
- .add(TYPE, context -> {})
+ .worker(TYPE, context -> {})
+ .build()
.start();
try {
assertTrue(ready.await(5, TimeUnit.SECONDS));
@@ -338,14 +338,15 @@ public void start() {
}
});
try (var second =
- peer.workers()
+ peer.toBuilder()
.queue("default", 1)
.id("new-leader")
.pollOnly(true)
.serviceInterval(Duration.ofMillis(20))
.stopTimeout(Duration.ofSeconds(5))
.errorHandler(errors::add)
- .add(TYPE, context -> {})
+ .worker(TYPE, context -> {})
+ .build()
.start()) {
base.requestResign();
assertTrue(elected.await(5, TimeUnit.SECONDS));
@@ -379,14 +380,14 @@ public void start() {
}
});
try (var workers =
- client
- .workers()
+ client.toBuilder()
.queue("default", 1)
.pollOnly(true)
.serviceInterval(Duration.ofMillis(20))
.stopTimeout(Duration.ofSeconds(5))
.errorHandler(errors::add)
- .add(TYPE, context -> {})
+ .worker(TYPE, context -> {})
+ .build()
.start()) {
assertTrue(started.await(5, TimeUnit.SECONDS));
assertTrue(workers.isLeader());
@@ -419,8 +420,7 @@ public void producer(
}
});
var workers =
- client
- .workers()
+ client.toBuilder()
.queue("default", 1)
.pollOnly(true)
.serviceInterval(Duration.ofHours(1))
@@ -429,7 +429,8 @@ public void producer(
event -> {
if (event.equals("leadership_check")) checked.countDown();
})
- .add(TYPE, context -> {})
+ .worker(TYPE, context -> {})
+ .build()
.start();
try {
assertTrue(checked.await(5, TimeUnit.SECONDS));
diff --git a/java/river/src/test/java/com/riverqueue/NotificationCleanupTest.java b/java/river/src/test/java/com/riverqueue/NotificationCleanupTest.java
index d7bc09a6a..4a7814d72 100644
--- a/java/river/src/test/java/com/riverqueue/NotificationCleanupTest.java
+++ b/java/river/src/test/java/com/riverqueue/NotificationCleanupTest.java
@@ -87,13 +87,13 @@ public boolean clean(Client river, Instant now) {
}
});
try (var workers =
- client
- .workers()
+ client.toBuilder()
.queue("default", 1)
- .add(JobType.of("unused", String.class), context -> {})
+ .worker(JobType.of("unused", String.class), context -> {})
.pollOnly(true)
.serviceInterval(Duration.ofHours(1))
.errorHandler(errors::add)
+ .build()
.start()) {
assertEquals(List.of(2L), cleaned.get(5, TimeUnit.SECONDS));
}
diff --git a/java/river/src/test/java/com/riverqueue/NotificationTest.java b/java/river/src/test/java/com/riverqueue/NotificationTest.java
index a4718fc9f..22866628e 100644
--- a/java/river/src/test/java/com/riverqueue/NotificationTest.java
+++ b/java/river/src/test/java/com/riverqueue/NotificationTest.java
@@ -70,11 +70,10 @@ public List claim(
var worked = new CompletableFuture>();
var retry = new CompletableFuture();
var ready = new CompletableFuture();
- var paused = new CompletableFuture();
+ var paused = new CompletableFuture();
var errors = new LinkedBlockingQueue();
try (var workers =
- client
- .workers()
+ client.toBuilder()
.queue("default", 1)
.leadership(false)
.pollInterval(Duration.ofHours(1))
@@ -89,9 +88,11 @@ public List claim(
event -> {
if (event.equals("listen_ready")) ready.complete(null);
})
- .add(type, context -> worked.complete(context.job()))
+ .worker(type, context -> worked.complete(context.job()))
+ .build()
.start();
- var subscription = workers.subscribe(paused::complete, Workers.EventKind.QUEUE_PAUSED)) {
+ var subscription =
+ workers.client().subscribe(paused::complete, Client.EventKind.QUEUE_PAUSED)) {
try {
var error = errors.poll(5, TimeUnit.SECONDS);
assertNotNull(error);
diff --git a/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java b/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java
index b0ef87333..6a2a0c363 100644
--- a/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java
+++ b/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java
@@ -61,14 +61,13 @@ public JsonNode decode(Job job) {
var completed = new CompletableFuture();
var errors = new LinkedBlockingQueue();
var workers =
- client
- .workers()
+ client.toBuilder()
.queue("default", 1)
.leadership(false)
.pollOnly(true)
.stopTimeout(Duration.ofSeconds(5))
.errorHandler(errors::add)
- .add(
+ .worker(
TYPE,
context -> {
parent.complete(context);
@@ -98,6 +97,7 @@ public JsonNode decode(Job job) {
context.completePeer(peer, new InterruptedException("peer interrupted"));
completed.complete(null);
})
+ .build()
.start();
try {
client.insert(TYPE, "parent");
@@ -143,7 +143,7 @@ void peerTimeoutInterruptsTheOwningWorker(String phase) throws Exception {
var entered = new CompletableFuture();
var interrupted = new CompletableFuture();
var release = new CountDownLatch(1);
- var events = new LinkedBlockingQueue();
+ var events = new LinkedBlockingQueue();
var errors = new LinkedBlockingQueue();
var database = databases.open(directory.resolve("river.db"));
var client =
@@ -169,8 +169,7 @@ public JsonNode decode(Job job) {
.insert(TYPE, "peer", InsertOptions.builder().queue("peers").maxAttempts(1).build())
.job();
var workers =
- client
- .workers()
+ client.toBuilder()
.queue("default", 1)
.leadership(false)
.pollOnly(true)
@@ -178,7 +177,7 @@ public JsonNode decode(Job job) {
.stuckThreshold(Duration.ofMillis(20))
.stopTimeout(Duration.ofSeconds(5))
.errorHandler(errors::add)
- .add(
+ .worker(
TYPE,
new Workers.Handler<>() {
@Override
@@ -221,15 +220,16 @@ public void work(WorkContext context) throws Exception {
}
}
})
+ .build()
.start();
- try (var subscription = workers.subscribe(events::add)) {
+ try (var subscription = workers.client().subscribe(events::add)) {
client.insert(TYPE, "parent", InsertOptions.builder().maxAttempts(1).build());
entered.get(5, TimeUnit.SECONDS);
interrupted.get(5, TimeUnit.SECONDS);
for (int i = 0; i < 2; i++) {
var event = events.poll(5, TimeUnit.SECONDS);
assertNotNull(event);
- assertEquals(Workers.EventKind.JOB_FAILED, event.kind());
+ assertEquals(Client.EventKind.JOB_FAILED, event.kind());
}
assertEquals(Job.State.DISCARDED, client.get(peerJob.id()).state());
} finally {
diff --git a/java/river/src/test/java/com/riverqueue/PluginTest.java b/java/river/src/test/java/com/riverqueue/PluginTest.java
index e5d4dce97..f83b02a7d 100644
--- a/java/river/src/test/java/com/riverqueue/PluginTest.java
+++ b/java/river/src/test/java/com/riverqueue/PluginTest.java
@@ -110,18 +110,18 @@ void pluginsActivateAllJobInterfacesBeforeExplicitRegistrations(boolean pluginsF
.withPlugins(first)
.withPlugins(second))
.withPilot(new Pilot() {});
- var completed = new CompletableFuture();
+ var completed = new CompletableFuture();
try (var workers =
- river
- .workers()
+ river.toBuilder()
.queue("default", 1)
.leadership(false)
.pollOnly(true)
.pollInterval(Duration.ofMillis(10))
.stopTimeout(Duration.ofSeconds(5))
- .add(TYPE, context -> trace.add("handler"))
+ .worker(TYPE, context -> trace.add("handler"))
+ .build()
.start();
- var subscription = workers.subscribe(completed::complete)) {
+ var subscription = workers.client().subscribe(completed::complete)) {
river.insert(TYPE, "one");
assertEquals(Job.State.COMPLETED, completed.get(5, TimeUnit.SECONDS).job().state());
}
@@ -162,13 +162,13 @@ void pluginsIncludePeriodicStartHooks() throws Exception {
.withHooks((HookPeriodicJobsStart) () -> started.add("explicit"))
.withPlugins(first, second);
try (var workers =
- river
- .workers()
+ river.toBuilder()
.queue("default", 1)
.pollOnly(true)
.serviceInterval(Duration.ofHours(1))
.stopTimeout(Duration.ofSeconds(5))
- .add(TYPE, context -> {})
+ .worker(TYPE, context -> {})
+ .build()
.start()) {
assertEquals("first", started.poll(5, TimeUnit.SECONDS));
assertEquals("second", started.poll(5, TimeUnit.SECONDS));
diff --git a/java/river/src/test/java/com/riverqueue/ProtocolStorageTest.java b/java/river/src/test/java/com/riverqueue/ProtocolStorageTest.java
index 47fdbc6fe..9cafbdd19 100644
--- a/java/river/src/test/java/com/riverqueue/ProtocolStorageTest.java
+++ b/java/river/src/test/java/com/riverqueue/ProtocolStorageTest.java
@@ -34,12 +34,11 @@ void periodicMetadata() throws Exception {
var claimed = new CompletableFuture>();
var errors = new LinkedBlockingQueue();
try (var workers =
- client
- .workers()
+ client.toBuilder()
.queue("default", 1)
.pollOnly(true)
.errorHandler(errors::add)
- .add(TYPE, context -> claimed.complete(context.job()))
+ .worker(TYPE, context -> claimed.complete(context.job()))
.periodic(
"conformance_periodic",
Schedule.every(Duration.ofDays(1)),
@@ -47,6 +46,7 @@ void periodicMetadata() throws Exception {
"periodic",
InsertOptions.builder().metadata(Map.of("source", "conformance")).build(),
true)
+ .build()
.start()) {
var job = claimed.get(5, TimeUnit.SECONDS);
var metadata = client.get(job.id()).metadata();
@@ -110,12 +110,11 @@ public boolean clean(Client client, Instant now) {
fixture.required("attempt_error"),
Json.tree(client.get(inserted.id()).errors().getFirst()));
- var completed = new CompletableFuture();
+ var completed = new CompletableFuture();
var writeFailure = new CompletableFuture();
var errors = new LinkedBlockingQueue();
try (var workers =
- client
- .workers()
+ client.toBuilder()
.queue("default", 1)
.pollOnly(true)
.errorHandler(errors::add)
@@ -136,10 +135,11 @@ public boolean clean(Client client, Instant now) {
}
return Duration.ofDays(1);
})
- .add(TYPE, context -> context.output(Map.of("ok", true)))
+ .worker(TYPE, context -> context.output(Map.of("ok", true)))
+ .build()
.start();
var subscription =
- workers.subscribe(completed::complete, Workers.EventKind.JOB_COMPLETED)) {
+ workers.client().subscribe(completed::complete, Client.EventKind.JOB_COMPLETED)) {
if (!postgres)
assertNull(
writeFailure.get(10, TimeUnit.SECONDS), "Rescue callbacks must not hold the writer");
diff --git a/java/river/src/test/java/com/riverqueue/ProtocolTest.java b/java/river/src/test/java/com/riverqueue/ProtocolTest.java
index 6d6f015d4..3496f78a0 100644
--- a/java/river/src/test/java/com/riverqueue/ProtocolTest.java
+++ b/java/river/src/test/java/com/riverqueue/ProtocolTest.java
@@ -494,7 +494,7 @@ Stream snoozeCounters() throws Exception {
() ->
assertEquals(
fixture.required("expected_snoozes").asLong(),
- Workers.nextSnoozeCount(fixture.required("metadata")))));
+ RunHandle.nextSnoozeCount(fixture.required("metadata")))));
return tests.stream();
}
diff --git a/java/river/src/test/java/com/riverqueue/RescueTest.java b/java/river/src/test/java/com/riverqueue/RescueTest.java
index 549d5448f..a167925c9 100644
--- a/java/river/src/test/java/com/riverqueue/RescueTest.java
+++ b/java/river/src/test/java/com/riverqueue/RescueTest.java
@@ -44,14 +44,13 @@ public boolean rescue(Client client, Rescue rescue) {
});
var job = client.insert(TYPE, mode).job();
try (var workers =
- client
- .workers()
+ client.toBuilder()
.queue("other", 1)
.pollOnly(true)
.jobTimeout(Duration.ofMinutes(5))
.stopTimeout(Duration.ofSeconds(5))
.errorHandler(errors::add)
- .add(
+ .worker(
TYPE,
new Workers.Handler<>() {
@Override
@@ -72,6 +71,7 @@ public void work(WorkContext context) {
fail("Wrong queue");
}
})
+ .build()
.start()) {
var rescue = request.get(5, TimeUnit.SECONDS);
var expected =
@@ -131,8 +131,7 @@ public boolean clean(Client client, Instant now) {
return null;
});
var workers =
- client
- .workers()
+ client.toBuilder()
.queue("other", 1)
.pollOnly(true)
.serviceInterval(Duration.ofHours(1))
@@ -146,7 +145,7 @@ public boolean clean(Client client, Instant now) {
}
return Duration.ofDays(1);
})
- .add(
+ .worker(
TYPE,
new Workers.Handler<>() {
@Override
@@ -163,6 +162,7 @@ public void work(WorkContext context) {
fail("Wrong queue");
}
})
+ .build()
.start();
try {
entered.get(5, TimeUnit.SECONDS);
@@ -246,15 +246,14 @@ public boolean clean(Client client, Instant now) {
});
}
try (var workers =
- client
- .workers()
+ client.toBuilder()
.queue("other", 1)
.pollOnly(true)
.jobTimeout(Duration.ofNanos(defaultDisabled ? -1 : 1))
.rescueAfter(Duration.ofHours(1))
.stopTimeout(Duration.ofSeconds(5))
.errorHandler(errors::add)
- .add(
+ .worker(
TYPE,
new Workers.Handler<>() {
@Override
@@ -275,6 +274,7 @@ public void work(WorkContext context) {
fail("Wrong queue");
}
})
+ .build()
.start()) {
maintained.get(5, TimeUnit.SECONDS);
for (var entry : jobs.entrySet()) {
@@ -352,8 +352,7 @@ public List claim(
});
try (var follower =
peer
- ? peerClient
- .workers()
+ ? peerClient.toBuilder()
.queue("default", 1)
.leadership(false)
.pollInterval(Duration.ofHours(1))
@@ -364,13 +363,13 @@ public List claim(
event -> {
if (event.equals("listen_ready")) listening.complete(null);
})
- .add(TYPE, context -> worked.complete(context.job()))
+ .worker(TYPE, context -> worked.complete(context.job()))
+ .build()
.start()
: null) {
if (peer) fetched.get(5, TimeUnit.SECONDS);
try (var workers =
- client
- .workers()
+ client.toBuilder()
.queue(peer ? "other" : "default", 1)
.pollOnly(true)
.pollInterval(Duration.ofHours(1))
@@ -378,7 +377,8 @@ public List claim(
.stopTimeout(Duration.ofSeconds(5))
.retryPolicy(job -> Duration.ofSeconds(1))
.errorHandler(errors::add)
- .add(TYPE, context -> worked.complete(context.job()))
+ .worker(TYPE, context -> worked.complete(context.job()))
+ .build()
.start()) {
maintained.get(5, TimeUnit.SECONDS);
var rescued = client.get(inserted.id());
diff --git a/java/river/src/test/java/com/riverqueue/ResumableTest.java b/java/river/src/test/java/com/riverqueue/ResumableTest.java
index 255b7953e..b20beae92 100644
--- a/java/river/src/test/java/com/riverqueue/ResumableTest.java
+++ b/java/river/src/test/java/com/riverqueue/ResumableTest.java
@@ -25,17 +25,16 @@ class ResumableTest {
@SuppressWarnings("try") // Resources intentionally scope background activity.
void cursorOnlyStepRemainsRecoverableAfterItIsOmitted() throws Exception {
var client = new Client(databases.open(directory.resolve("cursor-only.db")));
- var events = new LinkedBlockingQueue();
+ var events = new LinkedBlockingQueue();
var received = new LinkedBlockingQueue();
try (var workers =
- client
- .workers()
+ client.toBuilder()
.queue("default", 1)
.pollOnly(true)
.leadership(false)
.retryPolicy(job -> Duration.ofDays(1))
.stopTimeout(Duration.ofSeconds(5))
- .add(
+ .worker(
TYPE,
context -> {
if (context.job().attempt() > 1)
@@ -47,14 +46,15 @@ void cursorOnlyStepRemainsRecoverableAfterItIsOmitted() throws Exception {
});
context.step("later", () -> {});
})
+ .build()
.start();
- var subscription = workers.subscribe(events::add)) {
+ var subscription = workers.client().subscribe(events::add)) {
var metadata = Map.of(Protocol.METADATA_RESUMABLE_CURSOR, Map.of("loop", 0));
var inserted =
client.insert(TYPE, "work", InsertOptions.builder().metadata(metadata).build()).job();
var failed = events.poll(5, TimeUnit.SECONDS);
assertNotNull(failed);
- assertEquals(Workers.EventKind.JOB_FAILED, failed.kind());
+ assertEquals(Client.EventKind.JOB_FAILED, failed.kind());
var savedMetadata = client.get(inserted.id()).metadata();
assertEquals(
Json.tree(metadata).path(Protocol.METADATA_RESUMABLE_CURSOR),
@@ -63,7 +63,7 @@ void cursorOnlyStepRemainsRecoverableAfterItIsOmitted() throws Exception {
client.retry(inserted.id());
var completed = events.poll(5, TimeUnit.SECONDS);
assertNotNull(completed);
- assertEquals(Workers.EventKind.JOB_COMPLETED, completed.kind());
+ assertEquals(Client.EventKind.JOB_COMPLETED, completed.kind());
assertEquals(java.util.List.of(0), java.util.List.copyOf(received));
}
}
@@ -83,20 +83,19 @@ void cursorOnlyStepRemainsRecoverableAfterItIsOmitted() throws Exception {
void checkpointCommitsWithApplicationWrites(
boolean priorCheckpoint, boolean rollback, boolean withCursor) throws Exception {
var client = new Client(databases.open(directory.resolve("river.db")));
- var events = new LinkedBlockingQueue();
+ var events = new LinkedBlockingQueue();
var errors = new LinkedBlockingQueue();
var preparations = new AtomicInteger();
var steps = new LinkedBlockingQueue();
try (var workers =
- client
- .workers()
+ client.toBuilder()
.queue("default", 1)
.pollOnly(true)
.leadership(false)
.retryPolicy(job -> Duration.ofDays(1))
.stopTimeout(Duration.ofSeconds(5))
.errorHandler(errors::add)
- .add(
+ .worker(
TYPE,
context -> {
if (priorCheckpoint)
@@ -146,8 +145,9 @@ void checkpointCommitsWithApplicationWrites(
if (withCursor) context.stepWithCursor("process", action);
else context.step("process", () -> action.run(Json.object().path("missing")));
})
+ .build()
.start();
- var subscription = workers.subscribe(events::add)) {
+ var subscription = workers.client().subscribe(events::add)) {
var inserted =
client
.insert(
@@ -159,7 +159,7 @@ void checkpointCommitsWithApplicationWrites(
.job();
var failed = events.poll(5, TimeUnit.SECONDS);
assertNotNull(failed);
- assertEquals(Workers.EventKind.JOB_FAILED, failed.kind());
+ assertEquals(Client.EventKind.JOB_FAILED, failed.kind());
var saved = client.get(inserted.id());
assertEquals(
rollback ? "prepare" : "process",
@@ -174,7 +174,7 @@ void checkpointCommitsWithApplicationWrites(
client.retry(inserted.id());
var completed = events.poll(5, TimeUnit.SECONDS);
assertNotNull(completed);
- assertEquals(Workers.EventKind.JOB_COMPLETED, completed.kind());
+ assertEquals(Client.EventKind.JOB_COMPLETED, completed.kind());
assertEquals(1, preparations.get());
assertEquals(
rollback || withCursor ? java.util.List.of(1, 2) : java.util.List.of(1),
@@ -289,22 +289,21 @@ void checkpointRetainsOnlyCommittedProgress(boolean rollback, boolean finishStep
@SuppressWarnings("try") // Resources intentionally scope background activity.
void cursorDoesNotCompleteFailedSteps(String mode) throws Exception {
var client = new Client(databases.open(directory.resolve("river.db")));
- var events = new LinkedBlockingQueue();
+ var events = new LinkedBlockingQueue();
var errors = new LinkedBlockingQueue();
var prepares = new AtomicInteger();
var calls = new AtomicInteger();
boolean first = mode.startsWith("first");
boolean nullCursor = mode.contains("null");
try (var workers =
- client
- .workers()
+ client.toBuilder()
.queue("default", 1)
.pollOnly(true)
.leadership(false)
.retryPolicy(job -> Duration.ofDays(1))
.stopTimeout(Duration.ofSeconds(5))
.errorHandler(errors::add)
- .add(
+ .worker(
TYPE,
context -> {
if (!first)
@@ -325,12 +324,13 @@ void cursorDoesNotCompleteFailedSteps(String mode) throws Exception {
context.step("process", () -> action.run(Json.tree(null)));
else context.stepWithCursor("process", action);
})
+ .build()
.start();
- var subscription = workers.subscribe(events::add)) {
+ var subscription = workers.client().subscribe(events::add)) {
var job = client.insert(TYPE, "args").job();
var failed = events.poll(5, TimeUnit.SECONDS);
assertNotNull(failed);
- assertEquals(Workers.EventKind.JOB_FAILED, failed.kind());
+ assertEquals(Client.EventKind.JOB_FAILED, failed.kind());
var metadata = client.get(job.id()).metadata();
assertEquals(
first ? "" : "prepare", metadata.path(Protocol.METADATA_RESUMABLE_STEP).asString(""));
@@ -338,7 +338,7 @@ void cursorDoesNotCompleteFailedSteps(String mode) throws Exception {
client.retry(job.id());
var completed = events.poll(5, TimeUnit.SECONDS);
assertNotNull(completed);
- assertEquals(Workers.EventKind.JOB_COMPLETED, completed.kind());
+ assertEquals(Client.EventKind.JOB_COMPLETED, completed.kind());
assertEquals(2, calls.get());
assertEquals(first ? 0 : 1, prepares.get());
}
@@ -379,19 +379,18 @@ void cursorInFirstStepIsBufferedAndNestedStepsAreRejected() throws Exception {
@SuppressWarnings("try") // Resources intentionally scope background activity.
void cursorMapsReplaceCompletedStepsAndPreserveNullValues() throws Exception {
var client = new Client(databases.open(directory.resolve("river.db")));
- var events = new LinkedBlockingQueue();
+ var events = new LinkedBlockingQueue();
var errors = new LinkedBlockingQueue();
var firstCalls = new AtomicInteger();
try (var workers =
- client
- .workers()
+ client.toBuilder()
.queue("default", 1)
.pollOnly(true)
.leadership(false)
.retryPolicy(job -> Duration.ofDays(1))
.stopTimeout(Duration.ofSeconds(5))
.errorHandler(errors::add)
- .add(
+ .worker(
TYPE,
context -> {
context.stepWithCursor(
@@ -417,14 +416,15 @@ void cursorMapsReplaceCompletedStepsAndPreserveNullValues() throws Exception {
assertTrue(cursor.path("nested").path("value").isNull());
});
})
+ .build()
.start();
- var subscription = workers.subscribe(events::add)) {
+ var subscription = workers.client().subscribe(events::add)) {
var job = client.insert(TYPE, "args").job();
for (int attempt = 1; attempt <= 3; attempt++) {
var event = events.poll(5, TimeUnit.SECONDS);
assertNotNull(event);
assertEquals(
- attempt == 3 ? Workers.EventKind.JOB_COMPLETED : Workers.EventKind.JOB_FAILED,
+ attempt == 3 ? Client.EventKind.JOB_COMPLETED : Client.EventKind.JOB_FAILED,
event.kind());
if (attempt == 2) {
var cursors = client.get(job.id()).metadata().path(Protocol.METADATA_RESUMABLE_CURSOR);
diff --git a/java/river/src/test/java/com/riverqueue/RetryTest.java b/java/river/src/test/java/com/riverqueue/RetryTest.java
index 30f9bdd73..683b2a0eb 100644
--- a/java/river/src/test/java/com/riverqueue/RetryTest.java
+++ b/java/river/src/test/java/com/riverqueue/RetryTest.java
@@ -112,8 +112,7 @@ public List claim(
var worked = new CompletableFuture>();
var errors = new LinkedBlockingQueue();
try (var workers =
- client
- .workers()
+ client.toBuilder()
.queue("urgent", 1)
.leadership(false)
.pollOnly(mode.equals("local"))
@@ -125,16 +124,17 @@ public List claim(
if (event.equals("listen_ready")) listening.complete(null);
})
.errorHandler(errors::add)
- .add(
+ .worker(
TYPE,
context -> {
listening.get(5, TimeUnit.SECONDS);
if (context.args().equals("retry")) worked.complete(context.job());
})
+ .build()
.start()) {
// Completing the warmup drains startup wake-ups before testing retry's notification.
idle.get(5, TimeUnit.SECONDS);
- var actor = mode.equals("local") ? client : new Client(database);
+ var actor = mode.equals("local") ? workers.client() : new Client(database);
if (mode.equals("transaction")) {
try (var connection = database.connection()) {
connection.setAutoCommit(false);
diff --git a/java/river/src/test/java/com/riverqueue/RunHandleTest.java b/java/river/src/test/java/com/riverqueue/RunHandleTest.java
new file mode 100644
index 000000000..46efd1c4f
--- /dev/null
+++ b/java/river/src/test/java/com/riverqueue/RunHandleTest.java
@@ -0,0 +1,995 @@
+package com.riverqueue;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.Test;
+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 org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class RunHandleTest {
+ private static final JobType TYPE = JobType.of("runtime_test", String.class);
+ @TempDir Path directory;
+ @RegisterExtension final TestDatabase databases = new TestDatabase();
+
+ private Client client() {
+ var database = databases.open(directory.resolve("river.db"));
+ return new Client(database);
+ }
+
+ private Client.Builder workers(Client client) {
+ return client.toBuilder()
+ .queue("default", 1)
+ .leadership(false)
+ .pollOnly(true)
+ .stopTimeout(Duration.ofSeconds(1));
+ }
+
+ @Test
+ void builderGeneratesAnIdForEachClient() {
+ var builder = workers(client()).worker(TYPE, context -> {});
+ try (var first = builder.build().start();
+ var second = builder.build().start()) {
+ assertNotEquals(first.id(), second.id());
+ }
+ try (var explicit = builder.id("chosen-id").build().start()) {
+ assertEquals("chosen-id", explicit.id());
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "PT0.000000001S,true",
+ "PT1H,true",
+ "PT23H59M59.999999999S,true",
+ "PT24H,false",
+ "PT24H0.000000001S,false",
+ "PT48H,false"
+ })
+ void builderRequiresServiceIntervalsBelowQueueRetention(String value, boolean valid) {
+ var builder = new Client(Database.connect("jdbc:postgresql://127.0.0.1:1/unused")).toBuilder();
+ var interval = Duration.parse(value);
+ if (valid) assertSame(builder, builder.serviceInterval(interval));
+ else
+ assertEquals(
+ "serviceInterval must be shorter than the queue retention period of one day",
+ assertThrows(IllegalArgumentException.class, () -> builder.serviceInterval(interval))
+ .getMessage());
+ }
+
+ @ParameterizedTest
+ @CsvSource({"REMOTE,false", "REMOTE,true", "TIMEOUT,false", "TIMEOUT,true"})
+ @SuppressWarnings("try") // Resources intentionally scope background activity.
+ void cancellationRespectsTheHandlersOutcome(WorkContext.Cancellation cause, boolean interrupted)
+ throws Exception {
+ var client = client();
+ var entered = new CompletableFuture>();
+ var outcome = new CompletableFuture();
+ var reported = new LinkedBlockingQueue();
+ try (var workers =
+ workers(client)
+ .stopTimeout(Duration.ofSeconds(5))
+ .errorHandler(reported::add)
+ .worker(
+ TYPE,
+ context -> {
+ entered.complete(context);
+ context.awaitCancellation();
+ if (interrupted) context.checkCancelled();
+ })
+ .build()
+ .start();
+ var subscription = workers.client().subscribe(outcome::complete)) {
+ var inserted =
+ client.insert(TYPE, "cancel", InsertOptions.builder().maxAttempts(1).build()).job();
+ var context = entered.get(5, TimeUnit.SECONDS);
+ if (cause == WorkContext.Cancellation.REMOTE) client.cancel(inserted.id());
+ else context.requestCancellation(cause);
+ var event = outcome.get(5, TimeUnit.SECONDS);
+ var expected =
+ !interrupted
+ ? Job.State.COMPLETED
+ : cause == WorkContext.Cancellation.REMOTE
+ ? Job.State.CANCELLED
+ : Job.State.DISCARDED;
+ assertEquals(cause, context.cancellation());
+ assertEquals(expected, event.job().state());
+ var job = client.get(inserted.id());
+ assertEquals(expected, job.state());
+ assertEquals(interrupted ? 1 : 0, job.errors().size());
+ assertEquals(1, job.attempt());
+ assertNotNull(job.finalizedAt());
+ }
+ assertTrue(reported.isEmpty(), () -> "Unexpected worker errors: " + reported);
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = Job.State.class,
+ names = {"AVAILABLE", "PENDING", "RETRYABLE", "RUNNING", "SCHEDULED"})
+ @SuppressWarnings("try") // Resources intentionally scope background activity.
+ void completionEventsReflectPersistedState(Job.State state) throws Exception {
+ var client = client();
+ var database = client.database();
+ var changed = new CompletableFuture();
+ var events = new LinkedBlockingQueue();
+ var reported = new LinkedBlockingQueue();
+ try (var workers =
+ workers(client)
+ .stopTimeout(Duration.ofSeconds(5))
+ .errorHandler(reported::add)
+ .worker(
+ TYPE,
+ context -> {
+ context.transaction(
+ connection -> {
+ // A rescuer or another attempt changed this row before the old handler
+ // returned.
+ try (var statement =
+ Sql.prepare(
+ connection,
+ "UPDATE "
+ + database.prefix()
+ + "river_job SET state='"
+ + state.value()
+ + "', scheduled_at=?, attempt=? WHERE id=?",
+ database.timestamp(Instant.now().plusSeconds(3600)),
+ state == Job.State.RUNNING ? 2 : 1,
+ context.job().id())) {
+ assertEquals(1, statement.executeUpdate());
+ }
+ return null;
+ });
+ changed.complete(null);
+ })
+ .build()
+ .start();
+ var subscription = workers.client().subscribe(events::add)) {
+ var job = client.insert(TYPE, "changed").job();
+ changed.get(5, TimeUnit.SECONDS);
+ workers.stop();
+ assertEquals(state, client.get(job.id()).state());
+ var expected =
+ switch (state) {
+ case AVAILABLE, RETRYABLE -> Client.EventKind.JOB_FAILED;
+ case SCHEDULED -> Client.EventKind.JOB_SNOOZED;
+ default -> null;
+ };
+ if (expected == null) assertTrue(events.isEmpty());
+ else {
+ assertEquals(1, events.size());
+ assertEquals(expected, events.element().kind());
+ assertEquals(state, events.element().job().state());
+ }
+ }
+ assertTrue(reported.isEmpty(), () -> "Unexpected worker errors: " + reported);
+ }
+
+ @Test
+ @SuppressWarnings("try") // Resources intentionally scope background activity.
+ void completionRetriesSurviveAnInterruptedHandler() throws Exception {
+ var attempts = new AtomicInteger();
+ var client =
+ client()
+ .withPilot(
+ new Pilot() {
+ @Override
+ public void afterAttempt(
+ Connection connection,
+ Client.Driver driver,
+ Job job)
+ throws SQLException {
+ if (attempts.getAndIncrement() == 0)
+ throw new SQLException("transient completion failure");
+ }
+ });
+ var completed = new LinkedBlockingQueue();
+ var reported = new LinkedBlockingQueue();
+ try (var workers =
+ workers(client)
+ .errorHandler(reported::add)
+ .worker(TYPE, context -> Thread.currentThread().interrupt())
+ .build()
+ .start();
+ var subscription =
+ workers.client().subscribe(completed::add, Client.EventKind.JOB_COMPLETED)) {
+ for (int i = 0; i < 2; i++) {
+ var job = client.insert(TYPE, "job " + i).job();
+ var event = completed.poll(5, TimeUnit.SECONDS);
+ assertNotNull(
+ event, "An interrupted handler must not strand completion or its worker slot");
+ assertEquals(job.id(), event.job().id());
+ assertEquals(Job.State.COMPLETED, client.get(job.id()).state());
+ }
+ assertEquals(3, attempts.get());
+ assertEquals(1, reported.size());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"args", "kind", "row"})
+ @SuppressWarnings("try") // Resources intentionally scope background activity.
+ void completionRetriesSurviveInvalidJobs(String invalid) throws Exception {
+ var attempts = new AtomicInteger();
+ var client =
+ client()
+ .withPilot(
+ new Pilot() {
+ @Override
+ public void afterAttempt(
+ Connection connection,
+ Client.Driver driver,
+ Job job)
+ throws SQLException {
+ if (job.args().isObject() && attempts.getAndIncrement() == 0)
+ throw new SQLException("transient completion failure");
+ }
+
+ @Override
+ public java.util.List claim(
+ Connection connection,
+ Client.Driver driver,
+ Pilot.Claim claim,
+ Client.Transaction> next)
+ throws Exception {
+ var rows = next.run(connection);
+ return rows.stream()
+ .map(
+ row ->
+ invalid.equals("row") && row.job().args().isObject()
+ ? new Client.Decoded(
+ row.job(), new IllegalArgumentException("invalid row"))
+ : row)
+ .toList();
+ }
+ });
+ var type =
+ JobType.of(
+ invalid.equals("kind") ? "unknown_kind" : TYPE.kind(),
+ tools.jackson.databind.JsonNode.class);
+ var handled = new LinkedBlockingQueue();
+ var reported = new LinkedBlockingQueue();
+ var finalized = new LinkedBlockingQueue();
+ try (var workers =
+ workers(client)
+ .queue("default", 2)
+ .stopTimeout(Duration.ofSeconds(5))
+ .errorHandler(reported::add)
+ .worker(TYPE, context -> handled.add(context.job().id()))
+ .build()
+ .start();
+ var subscription = workers.client().subscribe(finalized::add)) {
+ var jobs =
+ client.insertMany(
+ java.util.List.of(
+ type.submission(Json.object(), InsertOptions.builder().maxAttempts(1).build()),
+ TYPE.submission("valid")));
+ var bad = jobs.getFirst().job();
+ var good = jobs.getLast().job();
+ assertNotNull(finalized.poll(5, TimeUnit.SECONDS), "The first claimed job must settle");
+ assertNotNull(finalized.poll(5, TimeUnit.SECONDS), "Every committed claim must settle");
+ workers.stop();
+ assertEquals(java.util.List.of(good.id()), java.util.List.copyOf(handled));
+ assertEquals(Job.State.DISCARDED, client.get(bad.id()).state());
+ assertEquals(Job.State.COMPLETED, client.get(good.id()).state());
+ assertEquals(1, client.get(bad.id()).errors().size());
+ assertEquals(2, attempts.get());
+ // A rolled-back completion batch may report the same failure for both claims.
+ assertFalse(reported.isEmpty());
+ assertTrue(
+ reported.stream()
+ .allMatch(error -> error.toString().contains("transient completion failure")));
+ }
+ }
+
+ @Test
+ @SuppressWarnings("try") // Resources intentionally scope background activity.
+ void completedHandlerIsNotCancelledWhileAcknowledging() throws Exception {
+ var acknowledging = new CountDownLatch(1);
+ var release = new CountDownLatch(1);
+ var context = new CompletableFuture>();
+ var client =
+ client()
+ .withPilot(
+ new Pilot() {
+ @Override
+ public void afterAttempt(
+ Connection connection,
+ Client.Driver driver,
+ Job job)
+ throws InterruptedException {
+ acknowledging.countDown();
+ release.await();
+ }
+ });
+ var completed = new CompletableFuture();
+ var workers = workers(client).worker(TYPE, context::complete).build().start();
+ try (var subscription = workers.client().subscribe(completed::complete)) {
+ var job = client.insert(TYPE, "acknowledge").job();
+ assertTrue(acknowledging.await(5, TimeUnit.SECONDS));
+ // Stop requests cancellation while the handler's successful outcome is awaiting commit.
+ assertThrows(RiverException.class, workers::stopAndCancel);
+ assertNull(context.get(5, TimeUnit.SECONDS).cancellation());
+ release.countDown();
+ assertEquals(Client.EventKind.JOB_COMPLETED, completed.get(5, TimeUnit.SECONDS).kind());
+ assertEquals(Job.State.COMPLETED, client.get(job.id()).state());
+ } finally {
+ release.countDown();
+ workers.stopAndCancel();
+ }
+ }
+
+ @Test
+ @SuppressWarnings("try") // Resources intentionally scope background activity.
+ void jobTimeoutDoesNotDependOnTheDispatcher() throws Exception {
+ var blocked = new CountDownLatch(1);
+ var release = new CountDownLatch(1);
+ var client =
+ client()
+ .withPilot(
+ new Pilot() {
+ @Override
+ public void producer(
+ Connection connection, Client.Driver driver, Producer producer)
+ throws InterruptedException {
+ if (!producer.active().isEmpty()) {
+ blocked.countDown();
+ release.await();
+ }
+ }
+ });
+ var cancelled = new CompletableFuture();
+ var failed = new CompletableFuture();
+ var workers =
+ workers(client)
+ .jobTimeout(Duration.ofNanos(1))
+ .serviceInterval(Duration.ofNanos(1))
+ .stuckThreshold(Duration.ofMinutes(1))
+ .worker(
+ TYPE,
+ context -> {
+ context.awaitCancellation();
+ cancelled.complete(context.cancellation());
+ release.await();
+ context.checkCancelled();
+ })
+ .build()
+ .start();
+ try (var subscription = workers.client().subscribe(failed::complete)) {
+ client.insert(TYPE, "timeout", InsertOptions.builder().maxAttempts(1).build());
+ assertTrue(blocked.await(5, TimeUnit.SECONDS));
+ assertEquals(WorkContext.Cancellation.TIMEOUT, cancelled.get(5, TimeUnit.SECONDS));
+ release.countDown();
+ assertEquals(Job.State.DISCARDED, failed.get(5, TimeUnit.SECONDS).job().state());
+ } finally {
+ release.countDown();
+ workers.stopAndCancel();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "job",
+ "poll",
+ "service",
+ "stop",
+ "stuck",
+ "rescue",
+ "maintenance",
+ "reindex",
+ "retention"
+ })
+ void oversizedDurationsFailDuringConfiguration(String option) {
+ var builder = new Client(Database.connect("jdbc:postgresql://127.0.0.1:1/unused")).toBuilder();
+ var oversized = Duration.ofSeconds(Long.MAX_VALUE);
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> {
+ switch (option) {
+ case "job" -> builder.jobTimeout(oversized);
+ case "poll" -> builder.pollInterval(oversized);
+ case "service" -> builder.serviceInterval(oversized);
+ case "stop" -> builder.stopTimeout(oversized);
+ case "stuck" -> builder.stuckThreshold(oversized);
+ case "rescue" -> builder.rescueAfter(oversized);
+ case "maintenance" -> builder.maintenanceInterval(oversized);
+ case "reindex" -> builder.reindex(java.util.List.of(), oversized);
+ case "retention" -> builder.retention(oversized, Duration.ZERO, Duration.ZERO);
+ default -> throw new AssertionError(option);
+ }
+ });
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"error", "shutdown", "snooze"})
+ @SuppressWarnings("try") // Resources intentionally scope background activity.
+ void firstStepCursorSurvivesIncompleteAttempt(String outcome) throws Exception {
+ var client = client();
+ var entered = new CountDownLatch(1);
+ var finished = new CompletableFuture();
+ try (var workers =
+ workers(client)
+ .worker(
+ TYPE,
+ context -> {
+ context.stepWithCursor(
+ "loop",
+ cursor -> {
+ context.cursor(7);
+ entered.countDown();
+ if (outcome.equals("snooze")) context.snooze(Duration.ofSeconds(60));
+ if (outcome.equals("shutdown")) {
+ context.awaitCancellation();
+ context.checkCancelled();
+ }
+ throw new Exception("retry");
+ });
+ })
+ .build()
+ .start();
+ var subscription = workers.client().subscribe(finished::complete)) {
+ var job = client.insert(TYPE, "cursor", InsertOptions.builder().maxAttempts(1).build()).job();
+ assertTrue(entered.await(5, TimeUnit.SECONDS));
+ if (outcome.equals("shutdown")) workers.stopAndCancel();
+ var event = finished.get(5, TimeUnit.SECONDS);
+ assertEquals(job.id(), event.job().id());
+ assertFalse(event.job().metadata().has(Protocol.METADATA_RESUMABLE_STEP));
+ assertEquals(
+ Json.tree(java.util.Map.of("loop", 7)),
+ event.job().metadata().path(Protocol.METADATA_RESUMABLE_CURSOR));
+ }
+ }
+
+ @Test
+ @SuppressWarnings("try") // Resources intentionally scope background activity.
+ void oversizedSnoozeFailsTheAttemptInsteadOfStrandingIt() throws Exception {
+ var client = client();
+ var failed = new CompletableFuture();
+ try (var workers =
+ workers(client)
+ .worker(TYPE, context -> context.snooze(Duration.ofSeconds(Long.MAX_VALUE)))
+ .build()
+ .start();
+ var subscription = workers.client().subscribe(failed::complete)) {
+ var job = client.insert(TYPE, "snooze", InsertOptions.builder().maxAttempts(1).build()).job();
+ var event = failed.get(5, TimeUnit.SECONDS);
+ assertEquals(job.id(), event.job().id());
+ assertEquals(Job.State.DISCARDED, event.job().state());
+ assertTrue(event.job().errors().getFirst().error().contains("Snooze duration"));
+ }
+ }
+
+ // 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)
+ .worker(
+ 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");
+ });
+ })
+ .build()
+ .start();
+ var subscription =
+ workers.client().subscribe(completed::complete, Client.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 =
+ new Client(Database.connect("jdbc:postgresql://127.0.0.1:1/unused"))
+ .toBuilder()
+ .queue("default", 1)
+ .worker(TYPE, context -> {})
+ .jobTimeout(Duration.ofHours(2))
+ .rescueAfter(Duration.ofHours(1));
+ assertEquals(
+ "rescueAfter must not be less than jobTimeout",
+ assertThrows(IllegalArgumentException.class, () -> builder.build().start()).getMessage());
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {-1, 0, 120})
+ @SuppressWarnings("try") // Resources intentionally scope background activity.
+ void rescueDefaultAccountsForTheJobTimeout(int timeoutMinutes) throws Exception {
+ var rescue = new CompletableFuture();
+ var client =
+ client()
+ .withPilot(
+ new Pilot() {
+ @Override
+ public boolean rescue(Client client, Rescue request) {
+ rescue.complete(request);
+ return true;
+ }
+ });
+ var builder = workers(client).leadership(true).worker(TYPE, context -> {});
+ if (timeoutMinutes != 0) builder.jobTimeout(Duration.ofMinutes(timeoutMinutes));
+ try (var workers = builder.build().start()) {
+ var request = rescue.get(5, TimeUnit.SECONDS);
+ assertEquals(
+ Duration.ofMinutes(60 + Math.max(0, timeoutMinutes)),
+ Duration.between(request.horizon(), request.now()));
+ }
+ }
+
+ @Test
+ @SuppressWarnings("try") // Resources intentionally scope background activity.
+ void goQueueNotificationsRespectTopicsAndSurviveMalformedPayloads() throws Exception {
+ var client = client();
+ var ready = new CountDownLatch(1);
+ var errors = new LinkedBlockingQueue();
+ var events = new LinkedBlockingQueue