diff --git a/java/API.md b/java/API.md index fc9ab8ec2..bd5a9fe23 100644 --- a/java/API.md +++ b/java/API.md @@ -16,10 +16,12 @@ Java exposes every Go helper. Stored values and protocol behavior still follow G | `JobCompleteTx` | `Client.complete(connection, job)`, `WorkContext.complete(connection)` | Completion keeps `Job` and commits with application changes. Only a running attempt can be completed. A committed completion remains authoritative if the handler subsequently throws. | | Context client, cancellation, cancel/snooze errors | `WorkContext.client`, cancellation methods, `cancel`, `discard`, `snooze` | Java has no Go context parameter convention. Control methods end the attempt; virtual-thread interruption supplements cooperative cancellation. | | Plugins, hooks, and middleware | `Plugin`, `HookInsertBegin`, `HookPeriodicJobsStart`, `HookWorkBegin`, `HookWorkEnd`, `JobInsertMiddleware`, `WorkerMiddleware` | `withPlugins` activates every hook and middleware interface on a value. `withHooks` and `withMiddleware` restrict registration to that side. Plugins run first, as in Go. Functional interfaces support lambdas and checked exceptions. | +| `ResumableSetStepTx`, `ResumableSetStepCursorTx` | `WorkContext.checkpoint(connection[, cursor])` | Checkpoints commit with application writes; failed transactions leave the last committed progress intact. | +| Worker timeout | `Workers.Handler.timeout(job)` | A default method preserves lambda handlers; per-job overrides also govern rescue decisions. | | Retry policy | `RetryPolicy` | A lambda receives the full job and returns `Duration`; the default uses the same error-count backoff and jitter. | | `Queues`, queue CRUD/control APIs | `Workers.addQueue/removeQueue`, `Client.queues()` | Runtime concurrency and persisted queue state have separate owners. Reads and mutations support caller transactions. | | `Subscribe`, `EventKind`, `Event` | `Workers.subscribe`, `EventKind`, `Event`, `Subscription` | Enum filters and an `AutoCloseable` subscription replace channels. Callbacks are synchronous; queue events include their queue. | -| Periodic jobs and schedules | `Workers.Builder.periodic`, `Schedule` | Java time types and a functional scheduling interface; cron implementation details are package-private. | +| Periodic jobs and schedules | `Workers.Builder.periodic`, `Workers.periodicJobs()`, `PeriodicJobBundle`, `PeriodicJob`, `Schedule` | Java time types and a functional scheduling interface; cron implementation details are package-private. | | Leadership notifications and maintenance config | `requestResign`, worker builder settings | Coordination stays in the database; operational durations use `Duration`. | | `rivermigrate` | `Migrator`, `Direction`, `Options`, executable CLI | Named options replace booleans and sentinel target versions. Down defaults to one migration; explicit target zero removes the line. | | Driver, pool, transaction integration | `Database`, JDBC `DataSource`/`Connection` | Use existing Java database pools. Database drivers remain optional library dependencies. | @@ -94,13 +96,12 @@ configuration code. These are implementation limits, not claims that Java conventions require a smaller feature set: -- OSS periodic definitions cannot be added or removed on a running runtime and - currently carry fixed arguments rather than a constructor invoked each time. +- OSS periodic definitions carry fixed arguments rather than a constructor + invoked each time. - Events do not yet include Go's per-job timing statistics. Java supplies callback delivery rather than a channel buffer and its associated subscription options. -- Resumable progress is buffered until the attempt ends; transactional checkpoint - helpers are not exposed. The Go logging middleware and worker-test harness have - no packaged Java equivalents. +- The Go logging middleware and worker-test harness have no packaged Java + equivalents. - Some maintenance configuration is coarser: reindexing accepts an interval, polling policy belongs to a runtime, and insert defaults belong to a job type. diff --git a/java/DIFFERENCES.md b/java/DIFFERENCES.md index 73f056f5a..b1cac6d3e 100644 --- a/java/DIFFERENCES.md +++ b/java/DIFFERENCES.md @@ -39,21 +39,34 @@ reproducing Go's nonterminating date loop. - SQL is kept in dialect-specific resource catalogs. The Go sqlc driver seam is not part of the Java API. +- Resumable cursor updates are saved even when the first step fails before any + step completes. Go currently drops cursor updates in that case. Updating a + cursor never marks a step completed; a failed plain step still runs on retry. +- Resumable steps cannot be nested in Java. Each attempt has one active step and + checkpoint scope, so rejecting nesting avoids ambiguous resume order and + checkpoint ownership. Go currently permits nested steps. +- Transactional checkpoints require the job to still be running with the same + `attempt` and `attempted_at`. Go updates by job ID alone. Java rejects a stale + attempt's checkpoint to prevent it from changing a replacement attempt. +- A transactional checkpoint with no cursors stores `river:resumable_cursor` as + JSON null, clearing any older cursor map. Go may omit that key; readers in both + languages accept null as an empty cursor map. - SQLite follows Go's JSONB storage and millisecond timestamp representation; Postgres uses microsecond timestamps. These are shared storage contracts, not configurable Java serialization choices. -No intentional differences in stored uniqueness hashes, job state values, reserved -metadata, migration versions, or notification payloads are allowed. Any remaining -conformance failure in these areas is a defect, not an API design choice. +Stored uniqueness hashes, job state values, migration versions, and notification +payloads must agree with Go. Reserved metadata uses the same shapes and encodings, +with the resumable persistence differences documented above. Other conformance +failures in these areas are defects, not API design choices. The following are current API limitations, not differences required by Java conventions. The [API review](API.md) maps the complete application surface. -OSS periodic registrations and polling policies belong to a worker configuration; -start a new runtime to change its periodic definitions, or use separate runtimes -for queues with different polling intervals. Fetch wakeups are coalesced without -a separate cooldown setting. Reindexing supports the daily UTC default or a +OSS periodic registrations are local to a worker runtime and can be changed +while it runs. Polling policies belong to a worker configuration; use separate +runtimes for queues with different polling intervals. Fetch wakeups are coalesced +without a separate cooldown setting. Reindexing supports the daily UTC default or a duration override, rather than an arbitrary cron expression. Job-type defaults replace process-wide insert defaults. @@ -69,6 +82,5 @@ that are not application APIs. Java has no bundled `riverlog` middleware or `rivertest.Worker` harness. Applications can write the shared log metadata format explicitly and test with JUnit against -an isolated database. Resumable steps buffer progress until the attempt ends; -Go's transactional resumable checkpoint helpers are not yet exposed. See the -[README examples](README.md#features) for these API limitations. +an isolated database. See the [README examples](README.md#features) for the +available APIs. diff --git a/java/README.md b/java/README.md index 8e6fe5874..3d7057805 100644 --- a/java/README.md +++ b/java/README.md @@ -282,6 +282,31 @@ interruptible I/O. `context.awaitCancellation()` waits for the signal. A normal return reports success even if cancellation or a timeout was requested. Call `context.checkCancelled()` when abandoning unfinished work. +### Worker timeouts + +`Workers.Builder.jobTimeout` sets the default attempt timeout. Override +`Handler.timeout(job)` for a particular worker; the callback receives typed +arguments and can choose a duration per job: + +```java +workerConfig.add(email, new Workers.Handler() { + @Override + public Duration timeout(Job job) { + return Duration.ofMinutes(5); + } + + @Override + public void work(WorkContext context) { + mailer.send(context.args().address(), context.args().subject()); + } +}); +``` + +Returning `null` or `Duration.ZERO` inherits the runtime default, as lambda +handlers do. A negative duration disables the timeout. The callback also runs when +the leader considers abandoned jobs for rescue, so keep it thread-safe and free of +side effects. Rescue respects longer worker deadlines and disabled timeouts. + ### Getting the client within workers [`context.client()`](https://riverqueue.com/docs/context-client) gives a handler @@ -493,10 +518,39 @@ var workerConfig = client.workers() ``` Periodic IDs use the same syntax as job kinds and must fit within 127 UTF-8 bytes. +Use `null` for an anonymous registration. `workers.periodicJobs().add(...)` +returns a handle for removing that registration. Builder registration returns no +handle, so an anonymous definition added through the builder can only be removed +with `workers.periodicJobs().clear()`. OSS schedules are kept in memory and can miss occurrences during downtime or -leader changes. Periodic definitions are fixed when the Java runtime starts; -restart it to change them. +leader changes. Add or remove definitions on a running runtime: + +```java +var periodicJobs = workers.periodicJobs(); // PeriodicJobBundle +var handle = periodicJobs.add(new PeriodicJob<>( + "hourly-status", Schedule.every(Duration.ofHours(1)), email, + new SendEmail("ops@example.com", "Hourly status"), InsertOptions.defaults(), true)); +periodicJobs.remove(handle); +// Or remove a named registration without its handle: +periodicJobs.removeById("hourly-status"); +``` + +Registrations belong to that runtime; configure them on every eligible leader. +Only the leader inserts jobs. `runOnStart` also applies when adding a definition +while leading. Removing a definition leaves previously inserted jobs in place, +and an insertion already in progress may finish. An ID can be reused after removal. +Periodic jobs require leader election to be enabled. + +Definitions without `runOnStart` are picked up on the next maintenance pass. +After stopping a runtime, its bundle still accepts additions and removals, but +those changes do not enqueue jobs or restart the runtime. + +The bundle also provides `addMany`, `removeMany`, `removeManyById`, and `clear`. +`addMany` validates the entire batch before registering any jobs and returns +handles in input order. Handles identify one registration, so removing an old +handle cannot remove a replacement that reused the same ID. The same +`PeriodicJob` definition can be passed to `Workers.Builder.periodic` before start. Interval schedules retain their planned cadence when maintenance runs late. Each job's `scheduled_at` defaults to its planned occurrence; an explicit @@ -532,9 +586,29 @@ workerConfig.add(email, context -> { ``` For loops, `context.stepWithCursor(name, cursor -> ...)` receives a saved JSON -cursor, and `context.cursor(value)` updates it. Java buffers step progress until -the attempt ends; it does not yet expose transactional checkpoint helpers. -A process crash can lose buffered progress, so steps must remain idempotent. +cursor, and `context.cursor(value)` updates it. Progress is normally buffered until +the attempt ends. A process crash can lose buffered progress, so steps must +remain idempotent. + +Use `context.checkpoint(connection)` inside a step to persist its progress in the +same transaction as application writes: + +```java +workerConfig.add(email, context -> { + context.step("record-delivery", () -> context.transaction(connection -> { + deliveries.record(connection, context.job().id()); + context.checkpoint(connection); + return null; + })); +}); +``` + +Call `context.checkpoint(connection, cursorValue)` inside `stepWithCursor` to +save cursor progress. A retry resumes that cursor step; a checkpoint without a +cursor skips a completed ordinary step. The connection must have auto-commit +disabled. Let transaction failures propagate out of the step so a rollback leaves +the last committed checkpoint authoritative. Neither overload commits the caller's +transaction. ### Scheduled jobs @@ -743,8 +817,9 @@ reindexing configured Postgres indexes. Leadership is enabled by default. `rescueAfter` must not be shorter than `jobTimeout`, so another client does not rescue an attempt that is still allowed to run. Its default is one hour, plus an -explicitly configured positive job timeout. Worker durations and snooze delays -must fit in signed 64-bit nanoseconds, up to about 292 years. +explicitly configured positive job timeout. Worker overrides are checked for each +rescue candidate even when they exceed `rescueAfter`. Worker durations and snooze +delays must fit in signed 64-bit nanoseconds, up to about 292 years. ```java var workerConfig = client.workers() diff --git a/java/conformance/feature-inventory.json b/java/conformance/feature-inventory.json index 09d0c276c..701cc1227 100644 --- a/java/conformance/feature-inventory.json +++ b/java/conformance/feature-inventory.json @@ -276,8 +276,7 @@ "id": "client.PeriodicJobs", "rationale": "Language-native API to add or remove periodic jobs at runtime; the enqueue behavior itself is classified under config.PeriodicJobs.", "source": "client.go:river.Client.PeriodicJobs", - "java_surface": "Client, Workers, Queues, Database", - "java_difference": "OSS periodic definitions belong to a runtime configuration; change them by starting a new worker runtime. Pro durable definitions can be registered on Pro." + "java_surface": "Workers.periodicJobs, PeriodicJobBundle" }, { "applicability": "not_applicable", diff --git a/java/river/src/main/java/com/riverqueue/Client.java b/java/river/src/main/java/com/riverqueue/Client.java index 19db1098d..b3a63512f 100644 --- a/java/river/src/main/java/com/riverqueue/Client.java +++ b/java/river/src/main/java/com/riverqueue/Client.java @@ -324,11 +324,13 @@ public Job complete(Connection connection, Job job, Object metadata) { Sql.query(database, "complete"), "completed", database.timestamp(clock.instant()), + null, + "{}", + Json.encode(updates), "completed", database.timestamp(clock.instant()), database.timestamp(job.scheduledAt()), job.attempt(), - Json.encode(updates), null, null, job.id(), diff --git a/java/river/src/main/java/com/riverqueue/PeriodicJob.java b/java/river/src/main/java/com/riverqueue/PeriodicJob.java new file mode 100644 index 000000000..f98dd1bbc --- /dev/null +++ b/java/river/src/main/java/com/riverqueue/PeriodicJob.java @@ -0,0 +1,50 @@ +package com.riverqueue; + +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.Objects; +import tools.jackson.databind.node.ObjectNode; + +/** + * A recurring job definition shared by builder configuration and runtime registration. The ID may + * be null for an anonymous registration; named IDs must be unique within a runtime. Arguments are + * reused for each occurrence. RunOnStart also applies when adding a job while leading. + */ +public record PeriodicJob( + String id, + Schedule schedule, + JobType type, + A args, + InsertOptions options, + boolean runOnStart) { + public PeriodicJob { + if (id != null + && (id.getBytes(StandardCharsets.UTF_8).length > 127 + || !Protocol.USER_SPECIFIED_ID_OR_KIND.matcher(id).matches())) + throw new IllegalArgumentException("Invalid periodic job ID"); + Objects.requireNonNull(schedule, "schedule"); + Objects.requireNonNull(type, "type"); + Objects.requireNonNull(args, "args"); + Objects.requireNonNull(options, "options"); + } + + void insert(Client river, Instant scheduledAt) { + var base = options.resolve(type.defaults()); + var metadata = (ObjectNode) base.metadata().deepCopy(); + metadata.put("periodic", true); + if (id != null) metadata.put(Protocol.METADATA_PERIODIC_JOB_ID, id); + var options = + new InsertOptions( + base.maxAttempts(), + metadata, + base.pending(), + base.priority(), + base.queue(), + base.scheduledAt(), + base.tags(), + base.unique()); + // A periodic occurrence supplies the default date without turning an available job into + // a scheduled one or overriding an explicitly configured date, matching Go's enqueuer. + river.transaction(c -> river.insert(c, type, args, options, scheduledAt)); + } +} diff --git a/java/river/src/main/java/com/riverqueue/PeriodicJobBundle.java b/java/river/src/main/java/com/riverqueue/PeriodicJobBundle.java new file mode 100644 index 000000000..f30be7c01 --- /dev/null +++ b/java/river/src/main/java/com/riverqueue/PeriodicJobBundle.java @@ -0,0 +1,143 @@ +package com.riverqueue; + +import java.time.Instant; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.function.BooleanSupplier; + +/** + * Thread-safe periodic registrations owned by a {@link Workers} instance. Only the leader enqueues + * occurrences, so update every eligible runtime to change schedules cluster-wide. Removal leaves + * previously inserted jobs alone, and an insertion already in progress may finish. Registrations + * can still be added and removed after stop, but no further jobs are enqueued. + */ +public final class PeriodicJobBundle { + private final boolean enabled; + private final Map entries = new LinkedHashMap<>(); + private final BooleanSupplier running; + private final Runnable wake; + + PeriodicJobBundle( + List> jobs, boolean enabled, BooleanSupplier running, Runnable wake) { + this.enabled = enabled; + this.running = running; + this.wake = wake; + addMany(jobs); + } + + /** Adds a definition, returning a handle that identifies this particular registration. */ + public Handle add(PeriodicJob job) { + return addMany(List.of(job)).getFirst(); + } + + /** Constructs a periodic definition and adds it to this bundle. */ + public Handle add( + String id, + Schedule schedule, + JobType type, + A args, + InsertOptions options, + boolean runOnStart) { + return add(new PeriodicJob<>(id, schedule, type, args, options, runOnStart)); + } + + /** + * Adds definitions atomically, returning handles in input order. Duplicate IDs leave the entire + * bundle unchanged. RunOnStart jobs wake the leader for immediate insertion. + */ + public List addMany(List> jobs) { + var additions = List.copyOf(jobs).stream().map(Entry::new).toList(); + if (additions.isEmpty()) return List.of(); + synchronized (entries) { + if (!enabled) throw new IllegalStateException("Periodic jobs require leader election"); + var ids = new HashSet(); + for (var entry : entries.values()) if (entry.job.id() != null) ids.add(entry.job.id()); + for (var entry : additions) + if (entry.job.id() != null && !ids.add(entry.job.id())) + throw new IllegalArgumentException("Duplicate periodic job ID: " + entry.job.id()); + for (var entry : additions) entries.put(entry.handle, entry); + } + if (running.getAsBoolean() && additions.stream().anyMatch(entry -> entry.job.runOnStart())) + wake.run(); + return additions.stream().map(entry -> entry.handle).toList(); + } + + /** Removes all definitions from this bundle. */ + public void clear() { + synchronized (entries) { + entries.clear(); + } + } + + boolean contains(Entry entry) { + synchronized (entries) { + return entries.get(entry.handle) == entry; + } + } + + /** + * Removes the registration identified by a handle. Returns false for a removed handle or one + * belonging to another bundle, even if a new registration reuses the same ID. + */ + public boolean remove(Handle handle) { + Objects.requireNonNull(handle, "handle"); + synchronized (entries) { + return entries.remove(handle) != null; + } + } + + /** + * Removes a named definition, returning whether it existed. Null never matches anonymous jobs. + */ + public boolean removeById(String id) { + if (id == null) return false; + synchronized (entries) { + return entries.values().removeIf(entry -> id.equals(entry.job.id())); + } + } + + /** Removes registrations by handle; unknown or already removed handles are ignored. */ + public void removeMany(List handles) { + var values = List.copyOf(handles); + synchronized (entries) { + values.forEach(entries::remove); + } + } + + /** Removes named definitions; unknown IDs are ignored. */ + public void removeManyById(List ids) { + var values = new HashSet<>(List.copyOf(ids)); + synchronized (entries) { + entries.values().removeIf(entry -> values.contains(entry.job.id())); + } + } + + List snapshot() { + synchronized (entries) { + return List.copyOf(entries.values()); + } + } + + static final class Entry { + final Handle handle = new Handle(); + final PeriodicJob job; + // Only the runtime's serial maintenance passes advance these fields. + Instant next; + Instant term; + + Entry(PeriodicJob job) { + this.job = job; + } + } + + /** + * An opaque identity for one registration, returned by {@link #add(PeriodicJob)} or {@link + * #addMany}. + */ + public static final class Handle { + private Handle() {} + } +} diff --git a/java/river/src/main/java/com/riverqueue/Pilot.java b/java/river/src/main/java/com/riverqueue/Pilot.java index 89596f37f..02495a46b 100644 --- a/java/river/src/main/java/com/riverqueue/Pilot.java +++ b/java/river/src/main/java/com/riverqueue/Pilot.java @@ -82,12 +82,18 @@ default boolean rescue(Client river, Rescue rescue) { return false; } + /** + * Rescue parameters. Evaluate timeout for each known job: negative disables rescue, zero permits + * it immediately, and positive durations protect attempts until their deadline. Cancellation and + * unknown-kind handling take precedence over the timeout. Evaluate the callback outside database + * transactions: it decodes arguments and invokes the worker's timeout method. + */ record Rescue( Instant now, Instant horizon, Set activeIds, Set kinds, - boolean timeoutDisabled, + java.util.function.Function, java.time.Duration> timeout, RetryPolicy retryPolicy) {} record Claim( diff --git a/java/river/src/main/java/com/riverqueue/WorkContext.java b/java/river/src/main/java/com/riverqueue/WorkContext.java index 3a7be3cac..97084e641 100644 --- a/java/river/src/main/java/com/riverqueue/WorkContext.java +++ b/java/river/src/main/java/com/riverqueue/WorkContext.java @@ -12,17 +12,24 @@ /** Per-attempt arguments, cancellation, output, and resumable checkpoints. */ public final class WorkContext { private final CancellationState cancellation; + private ObjectNode checkpointFallback = Json.object(); + private String checkpointStep; + private boolean checkpointed; private String completedStep; private String currentStep; private final ObjectNode cursors; + private boolean cursorsChanged; private Exception stepFailure; private final Job job; private final ObjectNode metadata = Json.object(); private final Client client; private Workers runtime; final java.util.Map peers; + private ObjectNode previousCursors; + private String previousStep; private final Set steps = new HashSet<>(); private boolean resumeMatched; + volatile Duration timeout; WorkContext(Client client, Job job) { this(client, job, true); @@ -51,7 +58,11 @@ public A args() { return job.args(); } - /** Internal companion-attempt seam used by River Pro's batch worker. */ + /** + * 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. + */ public java.util.List> claimPeers( JobType type, Client.Transaction> claim) { if (runtime == null) throw new IllegalStateException("No worker runtime"); @@ -73,6 +84,7 @@ WorkContext withDecodedJob(Job decoded) { // Notifications and supervision may still hold the provisional context during decoding. var context = new WorkContext<>(client, decoded, true, cancellation); context.runtime = runtime; + context.timeout = timeout; return context; } @@ -114,6 +126,32 @@ void checkRuntimeCancellation() { if (isCancelled()) throw new IllegalStateException("Batch attempt cancelled"); } + /** + * Persists the current resumable step in the caller's transaction. Call inside a step, after its + * application writes. Let transaction failures escape the step so retries use committed progress. + * The connection must have auto-commit disabled. + */ + public Job checkpoint(Connection transaction) { + return persistCheckpoint(transaction, cursors.deepCopy()); + } + + /** Persists the current resumable step and JSON cursor alongside application writes. */ + public Job checkpoint(Connection transaction, Object cursor) { + var updates = cursors.deepCopy(); + if (currentStep == null) + throw new IllegalStateException("Checkpoint requires an active resumable step"); + updates.set(currentStep, Json.tree(cursor)); + return persistCheckpoint(transaction, updates); + } + + synchronized ObjectNode checkpointFallback() { + return checkpointFallback.deepCopy(); + } + + String checkpointStep() { + return checkpointStep; + } + /** Completes this attempt atomically with application writes in the supplied transaction. */ public Job complete(Connection transaction) { return client.complete(transaction, job, metadataUpdates()); @@ -149,11 +187,10 @@ synchronized Exception finish(Exception failure) { if (stepFailure != null) failure = stepFailure; if (failure == null && !resumeMatched) failure = new IllegalArgumentException("Saved resumable step was not found"); - if (failure != null && completedStep != null) { - metadata(Protocol.METADATA_RESUMABLE_STEP, completedStep); - if (!cursors.isEmpty() || job.metadata().has(Protocol.METADATA_RESUMABLE_CURSOR)) - metadata(Protocol.METADATA_RESUMABLE_CURSOR, cursors.isEmpty() ? null : cursors); - } + if (failure != null && (completedStep != null || cursorsChanged)) + metadata.setAll(resumableMetadata(completedStep, cursors)); + if (failure == null || completedStep != null || cursorsChanged) + checkpointFallback = Json.object(); return failure; } @@ -161,6 +198,54 @@ public void output(Object value) { metadata(Protocol.METADATA_OUTPUT, value); } + private Job persistCheckpoint(Connection transaction, ObjectNode updates) { + if (currentStep == null) + throw new IllegalStateException("Checkpoint requires an active resumable step"); + if (job.state() != Job.State.RUNNING) throw new IllegalStateException("Job must be running"); + // Only this step's progress belongs to the caller's transaction. Completion restores the + // prior buffered progress if the checkpoint rolled back, without replacing a committed one. + checkpointStep = currentStep; + checkpointFallback = resumableMetadata(previousStep, previousCursors); + completedStep = null; + cursorsChanged = false; + var updated = + client.transaction( + transaction, + connection -> { + try (var statement = + Sql.prepare( + connection, + Sql.query(client.database(), "checkpoint"), + currentStep, + Json.encode(updates.isEmpty() ? null : updates), + job.id(), + job.attempt(), + client.database().timestamp(job.attemptedAt())); + var rows = statement.executeQuery()) { + if (!rows.next()) + throw new RiverException( + RiverException.Code.REJECTED, "Job attempt is no longer running"); + return client.read(rows).mapArgs(ignored -> job.args()); + } + }); + checkpointed = true; + cursors.removeAll(); + cursors.setAll(updates); + return updated; + } + + private ObjectNode resumableMetadata(String step, ObjectNode cursors) { + var updates = Json.object(); + if (step != null) updates.put(Protocol.METADATA_RESUMABLE_STEP, step); + if (!cursors.isEmpty() + || checkpointed + || job.metadata().has(Protocol.METADATA_RESUMABLE_CURSOR)) + updates.set( + Protocol.METADATA_RESUMABLE_CURSOR, + cursors.isEmpty() ? Json.tree(null) : cursors.deepCopy()); + return updates; + } + public Client client() { return client; } @@ -211,10 +296,12 @@ public void cursor(Object value) { if (currentStep == null) throw new IllegalStateException("Cursor requires an active resumable step"); cursors.set(currentStep, Json.tree(value)); + cursorsChanged = true; } private void executeStep(String name, boolean withCursor, CursorStep action) throws Exception { if (stepFailure != null) throw stepFailure; + if (currentStep != null) throw new IllegalStateException("Resumable steps cannot be nested"); if (name.isEmpty() || !steps.add(name)) { stepFailure = new IllegalArgumentException("duplicate resumable step or empty name: " + name); throw stepFailure; @@ -226,6 +313,8 @@ private void executeStep(String name, boolean withCursor, CursorStep action) thr } if (!resumeMatched || !withCursor || !cursors.has(name)) return; } + previousStep = completedStep; + previousCursors = cursors.deepCopy(); currentStep = name; try { action.run(cursors.path(name)); diff --git a/java/river/src/main/java/com/riverqueue/Workers.java b/java/river/src/main/java/com/riverqueue/Workers.java index 45dd72c4b..567d515e1 100644 --- a/java/river/src/main/java/com/riverqueue/Workers.java +++ b/java/river/src/main/java/com/riverqueue/Workers.java @@ -24,7 +24,6 @@ import java.util.function.Consumer; import org.postgresql.PGConnection; import tools.jackson.databind.JsonNode; -import tools.jackson.databind.node.ObjectNode; /** A bounded worker runtime using one virtual thread per active job. */ public final class Workers implements AutoCloseable { @@ -63,7 +62,8 @@ public final class Workers implements AutoCloseable { private long leaderEligibleAt = System.nanoTime(); private volatile Connection listener; private final Map queues = new ConcurrentHashMap<>(); - private final Map periodicNext = new ConcurrentHashMap<>(); + private final PeriodicJobBundle periodic; + private final AtomicBoolean periodicChanged = new AtomicBoolean(); private final java.util.Set pausedQueues = ConcurrentHashMap.newKeySet(); private final ReentrantLock queueLock = new ReentrantLock(); private final Condition queueChanged = queueLock.newCondition(); @@ -83,6 +83,15 @@ private Workers(Builder config) { new Protocol.Dispatcher( id, active, pendingCancellation, this::queueNotification, this::leadershipNotification); queues.putAll(config.queues); + periodic = + new PeriodicJobBundle( + config.periodic, + config.leadership, + running::get, + () -> { + if (periodicChanged.compareAndSet(false, true) && isLeader()) + leadershipWake.release(); + }); if (queues.isEmpty() || config.handlers.isEmpty()) throw new IllegalArgumentException("Workers require queues and handlers"); if (config.reindexInterval != null) nextReindex = Instant.now().plus(config.reindexInterval); @@ -128,6 +137,11 @@ boolean isLeader(long now) { return running.get() && term != null && term.active(now); } + /** Returns this runtime's live, thread-safe periodic job bundle. */ + public PeriodicJobBundle periodicJobs() { + return periodic; + } + public void removeQueue(String name) { queueLock.lock(); try { @@ -401,6 +415,7 @@ private void launch(Client.Decoded decoded, Registration registration) { row, context, () -> { + context.timeout = timeout(registration.handler, context.job()); workMiddleware( context, 0, @@ -506,7 +521,9 @@ List> claimPeers( // leave all claimed jobs owned by this attempt so its failure settles the entire batch. 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)); + 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); @@ -515,7 +532,12 @@ List> claimPeers( for (var decoded : rows) { if (decoded.failure() != null) throw decoded.failure(); var provisional = parent.peers.get(decoded.job().id()).context(); - var context = provisional.withDecodedJob(river.typed(decoded.job(), type)); + // 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); @@ -528,7 +550,9 @@ 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()); - complete(peer.row(), context, context.finish(failure)); + failure = context.finish(failure); + activeThreads.remove(context.job().id()); + complete(peer.row(), context, failure); parent.peers.remove(context.job().id(), peer); queueLock.lock(); try { @@ -606,11 +630,13 @@ private void complete(Job original, WorkContext context, Exception new Object[] { state.value(), river.database().timestamp(now), + context == null ? null : context.checkpointStep(), + context == null ? "{}" : Json.encode(context.checkpointFallback()), + Json.encode(updates), state.value(), river.database().timestamp(state.isFinalized() ? now : null), river.database().timestamp(scheduled), attempt, - Json.encode(updates), errors, errors, original.id(), @@ -961,7 +987,6 @@ private void maintain() throws Exception { } if (!running.get()) return; if (!elected.equals(initializedTerm)) { - periodicNext.clear(); river.pilot().leadershipStarted(); for (var hook : river.hooks()) if (hook instanceof HookPeriodicJobsStart periodicHook) periodicHook.start(); @@ -978,6 +1003,7 @@ private void maintain() throws Exception { report(error); } finally { maintenanceBusy.set(false); + if (periodicChanged.get()) leadershipWake.release(); } }); } catch (java.util.concurrent.RejectedExecutionException error) { @@ -991,32 +1017,35 @@ private void maintain() throws Exception { } private void services(Instant now, Instant term) throws SQLException { - for (var periodic : config.periodic) { + periodicChanged.set(false); + for (var state : periodic.snapshot()) { + var job = state.job; if (!ownsTerm(term)) return; + if (!periodic.contains(state)) continue; try { - Instant next = periodicNext.get(periodic.id); + Instant next = term.equals(state.term) ? state.next : null; if (next == null) next = - periodic.runOnStart + job.runOnStart() ? now - : periodic - .schedule + : job.schedule() .next(now.atZone(java.time.ZoneId.systemDefault())) .map(java.time.ZonedDateTime::toInstant) .orElse(Instant.MAX); if (!next.isAfter(now)) { Instant following = - periodic - .schedule + job.schedule() .next(next.atZone(java.time.ZoneId.systemDefault())) .map(java.time.ZonedDateTime::toInstant) .orElse(Instant.MAX); if (!ownsTerm(term)) return; - periodic.insert(river, next); + if (!periodic.contains(state)) continue; + job.insert(river, next); next = following; } if (!ownsTerm(term)) return; - periodicNext.put(periodic.id, next); + state.next = next; + state.term = term; } catch (Exception error) { // One failing schedule or insertion must not block other jobs or leader maintenance. report(error); @@ -1202,21 +1231,14 @@ private void rescue(Instant now, Instant term) { now.minus(config.rescueAfter), java.util.Set.copyOf(active.keySet()), java.util.Set.copyOf(config.handlers.keySet()), - config.jobTimeout.isNegative(), + this::rescueTimeout, config.retryPolicy))) return; long after = 0; while (ownsTerm(term)) { final long cursor = after; - var ready = new ArrayList(); var batch = river.transaction( c -> { - // Reserve the SQLite writer before reading: another commit would make this - // snapshot impossible to upgrade when persisting the rescue. - if (river.database().dialect() == Database.Dialect.SQLITE) - try (var statement = Sql.prepare(c, Sql.query(river.database(), "writer_lock"))) { - statement.executeUpdate(); - } var stuck = new ArrayList>(); try (var statement = Sql.prepare( @@ -1227,63 +1249,77 @@ private void rescue(Instant now, Instant term) { var rows = statement.executeQuery()) { while (rows.next()) stuck.add(river.read(rows)); } - for (var job : stuck) { - if (active.containsKey(job.id()) - || (config.jobTimeout.isNegative() - && config.handlers.containsKey(job.kind()))) continue; - boolean cancel = false; - if (job.metadata().path("cancel_attempted_at").isString()) - try { - cancel = - !Instant.parse(job.metadata().path("cancel_attempted_at").asString()) - .equals(Instant.parse("0001-01-01T00:00:00Z")); - } catch (RuntimeException ignored) { - } - Job.State state = - cancel - ? Job.State.CANCELLED - : !config.handlers.containsKey(job.kind()) - || job.attempt() >= job.maxAttempts() - ? Job.State.DISCARDED - : Job.State.RETRYABLE; - Instant scheduled = job.scheduledAt(); - if (!state.isFinalized()) { - Duration delay = retryDelay(job, now); - scheduled = now.plus(delay); - if (bypassScheduler(delay)) state = Job.State.AVAILABLE; - } - String error = - Json.encode( - new Job.AttemptError( - now, job.attempt(), "Stuck job rescued by JobRescuer", "")); - String metadata = - Json.encode( - Map.of( - Protocol.METADATA_RESCUE_COUNT, - job.metadata().path(Protocol.METADATA_RESCUE_COUNT).asLong(0) + 1)); - try (var statement = - Sql.prepare( - c, - Sql.query(river.database(), "rescue_update"), - state.value(), - river.database().timestamp(state.isFinalized() ? now : null), - river.database().timestamp(scheduled), - job.attempt(), - metadata, - error, - error, - job.id(), - job.attempt(), - river.database().timestamp(job.attemptedAt())); - var rows = statement.executeQuery()) { - if (rows.next() && state == Job.State.AVAILABLE) - ready.add( - new FetchWakeup( - Database.instant(rows.getString("scheduled_at")), job.queue())); - } - } return stuck; }); + var rescues = new ArrayList(); + for (var job : batch) { + if (active.containsKey(job.id())) continue; + boolean cancel = false; + if (job.metadata().path("cancel_attempted_at").isString()) + try { + cancel = + !Instant.parse(job.metadata().path("cancel_attempted_at").asString()) + .equals(Instant.parse("0001-01-01T00:00:00Z")); + } catch (RuntimeException ignored) { + } + if (!cancel && config.handlers.containsKey(job.kind())) { + var timeout = rescueTimeout(job); + if (timeout.isNegative() + || Duration.between(job.attemptedAt(), now).compareTo(timeout) < 0) continue; + } + Job.State state = + cancel + ? Job.State.CANCELLED + : !config.handlers.containsKey(job.kind()) || job.attempt() >= job.maxAttempts() + ? Job.State.DISCARDED + : Job.State.RETRYABLE; + Instant scheduled = job.scheduledAt(); + if (!state.isFinalized()) { + Duration delay = retryDelay(job, now); + scheduled = now.plus(delay); + if (bypassScheduler(delay)) state = Job.State.AVAILABLE; + } + String error = + Json.encode( + new Job.AttemptError(now, job.attempt(), "Stuck job rescued by JobRescuer", "")); + String metadata = + Json.encode( + Map.of( + Protocol.METADATA_RESCUE_COUNT, + job.metadata().path(Protocol.METADATA_RESCUE_COUNT).asLong(0) + 1)); + rescues.add(new Rescue(error, job, metadata, scheduled, state)); + } + if (!ownsTerm(term)) return; + var ready = new ArrayList(); + river.transaction( + c -> { + for (var rescue : rescues) { + var job = rescue.job(); + if (active.containsKey(job.id())) continue; + var state = rescue.state(); + try (var statement = + Sql.prepare( + c, + Sql.query(river.database(), "rescue_update"), + state.value(), + river.database().timestamp(state.isFinalized() ? now : null), + river.database().timestamp(rescue.scheduled()), + job.attempt(), + rescue.metadata(), + rescue.error(), + rescue.error(), + job.id(), + job.attempt(), + river.database().timestamp(job.attemptedAt())); + var rows = statement.executeQuery()) { + if (rows.next() && state == Job.State.AVAILABLE) + ready.add( + new FetchWakeup( + Database.instant(rows.getString("scheduled_at")), job.queue())); + } + } + return null; + }); // Only committed rescues may wake fetching; use the database's timestamp precision. fetchWakeups.addAll(ready); if (!ready.isEmpty()) wake.release(); @@ -1292,14 +1328,28 @@ private void rescue(Instant now, Instant term) { } } + private record Rescue( + String error, Job job, String metadata, Instant scheduled, Job.State state) {} + + private Duration rescueTimeout(Job job) { + try { + return timeout(job); + } catch (Throwable error) { + // Malformed arguments or timeout callbacks must not prevent abandoned jobs being rescued. + report(error); + return Duration.ZERO; + } + } + private void supervise() { while (running.get() || dispatcherStopped.getCount() != 0 || !active.isEmpty()) { for (var context : active.values()) { // Attempts retain their deadlines while fetching is blocked or gracefully stopping. - if (!config.jobTimeout.isNegative() - && Duration.between(context.job().attemptedAt(), Instant.now()) - .compareTo(config.jobTimeout) - >= 0) context.requestCancellation(WorkContext.Cancellation.TIMEOUT); + var timeout = context.timeout; + if (timeout != null + && !timeout.isNegative() + && Duration.between(context.job().attemptedAt(), Instant.now()).compareTo(timeout) >= 0) + context.requestCancellation(WorkContext.Cancellation.TIMEOUT); if (context.forceIfStuck(config.stuckThreshold)) { observe("stuck_job"); var thread = activeThreads.get(context.job().id()); @@ -1315,6 +1365,26 @@ private void supervise() { } } + private Duration timeout(Job job) { + var registration = config.handlers.get(job.kind()); + return registration == null + ? config.jobTimeout + : timeout(registration, job.mapArgs(ignored -> river.pilot().decode(job))); + } + + private Duration timeout(Registration registration, Job decoded) { + return timeout( + registration.handler, + decoded.mapArgs(args -> Json.decode(args, registration.type.argsType()))); + } + + private Duration timeout(Handler handler, Job job) { + var timeout = handler.timeout(job); + if (timeout == null || timeout.isZero()) return config.jobTimeout; + Builder.duration(timeout, "Worker timeout"); + return timeout; + } + private void observe(String event) { for (var observer : config.observers) try { @@ -1428,7 +1498,7 @@ public static final class Builder { private boolean leadership = true; private Duration maintenanceInterval = Duration.ofSeconds(5); private final List> observers = new ArrayList<>(); - private final List> periodic = new ArrayList<>(); + private final List> periodic = new ArrayList<>(); private Duration pollInterval = Duration.ofSeconds(1); private boolean pollOnly; private final Map queues = new java.util.LinkedHashMap<>(); @@ -1543,6 +1613,16 @@ public Builder observe(Consumer value) { return this; } + /** Registers a periodic definition before starting this runtime. */ + public Builder periodic(PeriodicJob job) { + Objects.requireNonNull(job, "job"); + if (job.id() != null + && periodic.stream().anyMatch(existing -> job.id().equals(existing.id()))) + throw new IllegalArgumentException("Duplicate periodic job ID: " + job.id()); + periodic.add(job); + return this; + } + public Builder periodic( String id, Schedule schedule, @@ -1550,21 +1630,7 @@ public Builder periodic( A args, InsertOptions options, boolean runOnStart) { - if (id == null - || id.getBytes(StandardCharsets.UTF_8).length > 127 - || !Protocol.USER_SPECIFIED_ID_OR_KIND.matcher(id).matches()) - throw new IllegalArgumentException("Invalid periodic job ID"); - if (periodic.stream().anyMatch(job -> job.id().equals(id))) - throw new IllegalArgumentException("Duplicate periodic job ID: " + id); - periodic.add( - new Periodic<>( - id, - Objects.requireNonNull(schedule), - Objects.requireNonNull(type), - Objects.requireNonNull(args), - Objects.requireNonNull(options), - runOnStart)); - return this; + return periodic(new PeriodicJob<>(id, schedule, type, args, options, runOnStart)); } public Builder pollInterval(Duration value) { @@ -1692,37 +1758,21 @@ private record Completion( Object[] params, java.util.concurrent.CompletableFuture> done) {} + /** A worker handler. Lambdas inherit the runtime's attempt timeout. */ @FunctionalInterface public interface Handler { + /** + * Overrides the timeout for this job. Null or zero inherits the runtime default; a negative + * duration disables timeouts. Positive values must fit in signed 64-bit nanoseconds. Called + * once before work and when evaluating an abandoned job for rescue, so implementations must be + * thread-safe and free of side effects. + */ + default Duration timeout(Job job) { + return null; + } + void work(WorkContext context) throws Exception; } private record Registration(JobType type, Handler handler) {} - - private record Periodic( - String id, - Schedule schedule, - JobType type, - A args, - InsertOptions options, - boolean runOnStart) { - void insert(Client river, Instant scheduledAt) { - var base = options.resolve(type.defaults()); - var metadata = (ObjectNode) base.metadata().deepCopy(); - metadata.put("periodic", true).put(Protocol.METADATA_PERIODIC_JOB_ID, id); - var options = - new InsertOptions( - base.maxAttempts(), - metadata, - base.pending(), - base.priority(), - base.queue(), - base.scheduledAt(), - base.tags(), - base.unique()); - // A periodic occurrence supplies the default date without turning an available job into - // a scheduled one or overriding an explicitly configured date, matching Go's enqueuer. - river.transaction(c -> river.insert(c, type, args, options, scheduledAt)); - } - } } diff --git a/java/river/src/main/resources/com/riverqueue/postgres.sql b/java/river/src/main/resources/com/riverqueue/postgres.sql index 9d2d4b10c..d5a92e3ca 100644 --- a/java/river/src/main/resources/com/riverqueue/postgres.sql +++ b/java/river/src/main/resources/com/riverqueue/postgres.sql @@ -151,14 +151,17 @@ WHERE id IN (SELECT id FROM {schema}river_job WHERE queue=? AND state='available AND NOT EXISTS(SELECT 1 FROM {schema}river_queue q WHERE q.name=river_job.queue AND paused_at IS NOT NULL) AND kind IN (SELECT jsonb_array_elements_text(?::jsonb)) ORDER BY priority,scheduled_at,id LIMIT ? FOR UPDATE SKIP LOCKED) RETURNING {columns} +-- name: checkpoint +UPDATE {schema}river_job SET metadata=metadata || jsonb_build_object('river:resumable_step',?::text,'river:resumable_cursor',?::jsonb) +WHERE id=? AND state='running' AND attempt=? AND attempted_at=? RETURNING {columns} -- name: complete -WITH choice AS (SELECT ?::text AS target_state, ?::timestamptz AS now) +WITH choice AS (SELECT ?::text AS target_state, ?::timestamptz AS now, ?::text AS checkpoint_step, ?::jsonb AS checkpoint_fallback, ?::jsonb AS metadata_updates) UPDATE {schema}river_job SET state=CASE WHEN state != 'running' THEN state WHEN (SELECT target_state FROM choice) IN ('available','retryable','scheduled') AND metadata->>'cancel_attempted_at' IS NOT NULL THEN 'cancelled' ELSE ?::{schema}river_job_state END, finalized_at=CASE WHEN state != 'running' THEN finalized_at WHEN (SELECT target_state FROM choice) IN ('available','retryable','scheduled') AND metadata->>'cancel_attempted_at' IS NOT NULL THEN (SELECT now FROM choice) ELSE ? END, scheduled_at=CASE WHEN state != 'running' THEN scheduled_at WHEN (SELECT target_state FROM choice) IN ('available','retryable','scheduled') AND metadata->>'cancel_attempted_at' IS NOT NULL THEN scheduled_at ELSE ? END, attempt=CASE WHEN state != 'running' THEN attempt WHEN (SELECT target_state FROM choice) IN ('available','retryable','scheduled') AND metadata->>'cancel_attempted_at' IS NOT NULL THEN attempt ELSE ? END, -metadata=coalesce(metadata || nullif(?::jsonb,'{}'::jsonb),metadata), +metadata=coalesce(metadata || nullif(CASE WHEN (SELECT checkpoint_step FROM choice) IS NOT NULL AND metadata->>'river:resumable_step' IS DISTINCT FROM (SELECT checkpoint_step FROM choice) THEN (SELECT checkpoint_fallback FROM choice) ELSE '{}'::jsonb END || (SELECT metadata_updates FROM choice),'{}'::jsonb),metadata), errors=CASE WHEN state != 'running' OR ?::text IS NULL THEN errors ELSE array_append(errors,?::jsonb) END WHERE id=? AND attempt=? AND attempted_at=? RETURNING {columns} @@ -170,7 +173,7 @@ DELETE FROM {schema}river_job WHERE id IN (SELECT id FROM {schema}river_job WHERE state IN ('cancelled','completed','discarded') AND finalized_at < ? AND queue NOT IN (SELECT jsonb_array_elements_text(?::jsonb)) AND (? OR queue IN (SELECT jsonb_array_elements_text(?::jsonb))) ORDER BY id LIMIT ?) -- name: rescue_select -SELECT {columns} FROM {schema}river_job WHERE state='running' AND attempted_at? ORDER BY id LIMIT 1000 FOR UPDATE SKIP LOCKED +SELECT {columns} FROM {schema}river_job WHERE state='running' AND attempted_at? ORDER BY id LIMIT 1000 -- name: database_version SELECT version(), coalesce(current_setting('yb_enable_listen_notify', true), 'off')::boolean diff --git a/java/river/src/main/resources/com/riverqueue/sqlite.sql b/java/river/src/main/resources/com/riverqueue/sqlite.sql index f67d4e132..f38cfd267 100644 --- a/java/river/src/main/resources/com/riverqueue/sqlite.sql +++ b/java/river/src/main/resources/com/riverqueue/sqlite.sql @@ -162,14 +162,29 @@ WHERE id IN (SELECT id FROM river_job WHERE queue=? AND state='available' AND sc AND NOT EXISTS(SELECT 1 FROM river_queue q WHERE q.name=river_job.queue AND paused_at IS NOT NULL) AND kind IN (SELECT value FROM json_each(?)) ORDER BY priority,scheduled_at,id LIMIT ?) RETURNING {columns} +-- name: checkpoint +UPDATE river_job SET metadata=jsonb_set(metadata,'$."river:resumable_step"',?,'$."river:resumable_cursor"',jsonb(?)) +WHERE id=? AND state='running' AND attempt=? AND attempted_at=? RETURNING {columns} -- name: complete -WITH choice AS (SELECT ? AS target_state, ? AS now) +WITH choice AS (SELECT ? AS target_state, ? AS now, ? AS checkpoint_step, jsonb(?) AS checkpoint_fallback, jsonb(?) AS metadata_updates) UPDATE river_job SET state=CASE WHEN state != 'running' THEN state WHEN (SELECT target_state FROM choice) IN ('available','retryable','scheduled') AND (CASE WHEN typeof(metadata)<>'text' OR json_valid(metadata) THEN json_extract(metadata,'$.cancel_attempted_at') END) IS NOT NULL THEN 'cancelled' ELSE ? END, finalized_at=CASE WHEN state != 'running' THEN finalized_at WHEN (SELECT target_state FROM choice) IN ('available','retryable','scheduled') AND (CASE WHEN typeof(metadata)<>'text' OR json_valid(metadata) THEN json_extract(metadata,'$.cancel_attempted_at') END) IS NOT NULL THEN (SELECT now FROM choice) ELSE ? END, scheduled_at=CASE WHEN state != 'running' THEN scheduled_at WHEN (SELECT target_state FROM choice) IN ('available','retryable','scheduled') AND (CASE WHEN typeof(metadata)<>'text' OR json_valid(metadata) THEN json_extract(metadata,'$.cancel_attempted_at') END) IS NOT NULL THEN scheduled_at ELSE ? END, attempt=CASE WHEN state != 'running' THEN attempt WHEN (SELECT target_state FROM choice) IN ('available','retryable','scheduled') AND (CASE WHEN typeof(metadata)<>'text' OR json_valid(metadata) THEN json_extract(metadata,'$.cancel_attempted_at') END) IS NOT NULL THEN attempt ELSE ? END, -metadata=CASE WHEN typeof(metadata)='text' AND NOT json_valid(metadata) THEN metadata ELSE jsonb_patch(metadata,jsonb(?)) END, +metadata=CASE WHEN typeof(metadata)='text' AND NOT json_valid(metadata) THEN metadata ELSE ( + -- MergePatch recursively merges objects and deletes nulls; cursor maps must be replaced whole. + SELECT CASE + WHEN json_type(metadata_updates,'$."river:resumable_cursor"') IS NOT NULL + THEN jsonb_set(merged,'$."river:resumable_cursor"',jsonb_extract(metadata_updates,'$."river:resumable_cursor"')) + WHEN use_fallback AND json_type(checkpoint_fallback,'$."river:resumable_cursor"') IS NOT NULL + THEN jsonb_set(merged,'$."river:resumable_cursor"',jsonb_extract(checkpoint_fallback,'$."river:resumable_cursor"')) + ELSE merged END + FROM ( + SELECT *, jsonb_patch(CASE WHEN use_fallback THEN jsonb_patch(metadata,checkpoint_fallback) ELSE metadata END,metadata_updates) AS merged + FROM (SELECT *, checkpoint_step IS NOT NULL AND json_extract(metadata,'$."river:resumable_step"') IS NOT checkpoint_step AS use_fallback FROM choice) + ) +) END, errors=CASE WHEN state != 'running' OR ? IS NULL THEN errors ELSE jsonb_insert(CASE WHEN typeof(errors)='text' AND NOT json_valid(errors) THEN jsonb_array(errors) WHEN coalesce(json_type(errors),'array') <> 'array' THEN jsonb_array(json(errors)) ELSE coalesce(errors,jsonb('[]')) END,'$[#]',jsonb(?)) END WHERE id=? AND attempt=? AND attempted_at=? RETURNING {columns} diff --git a/java/river/src/test/java/com/riverqueue/IdentifierTest.java b/java/river/src/test/java/com/riverqueue/IdentifierTest.java index 9bc2613a1..1f6bbad26 100644 --- a/java/river/src/test/java/com/riverqueue/IdentifierTest.java +++ b/java/river/src/test/java/com/riverqueue/IdentifierTest.java @@ -5,7 +5,7 @@ import java.time.Duration; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.NullAndEmptySource; +import org.junit.jupiter.params.provider.EmptySource; import org.junit.jupiter.params.provider.ValueSource; class IdentifierTest { @@ -13,6 +13,12 @@ class IdentifierTest { private final Workers.Builder workers = new Client(Database.connect("jdbc:sqlite::memory:")).workers(); + @Test + void acceptsAnonymousPeriodicJobs() { + assertDoesNotThrow(() -> periodic(null)); + assertDoesNotThrow(() -> periodic(null)); + } + @ParameterizedTest @ValueSource( strings = { @@ -68,7 +74,7 @@ void periodicIdLengthCountsUtf8Bytes() { } @ParameterizedTest - @NullAndEmptySource + @EmptySource @ValueSource( strings = { "a", @@ -87,8 +93,7 @@ void periodicIdLengthCountsUtf8Bytes() { "daily😀" }) void rejectsInvalidKindAndPeriodicIdSyntax(String value) { - if (value != null) - assertThrows(IllegalArgumentException.class, () -> JobType.of(value, String.class)); + assertThrows(IllegalArgumentException.class, () -> JobType.of(value, String.class)); assertThrows(IllegalArgumentException.class, () -> periodic(value)); } diff --git a/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java b/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java index 21992b839..f2d5c590d 100644 --- a/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java +++ b/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java @@ -7,12 +7,15 @@ import java.time.Instant; import java.util.ArrayList; 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.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.ValueSource; import tools.jackson.databind.JsonNode; class PeerCancellationTest { @@ -34,6 +37,7 @@ void cancellationDuringArgumentDecodingReachesTheTypedPeer( WorkContext.Cancellation cause, String when) throws Exception { var decoding = new CompletableFuture(); var decode = new CompletableFuture(); + var decodes = new AtomicInteger(); var database = databases.open(directory.resolve("river.db")); var client = new Client(database) @@ -42,6 +46,7 @@ void cancellationDuringArgumentDecodingReachesTheTypedPeer( @Override public JsonNode decode(Job job) { if (job.queue().equals("peers")) { + decodes.incrementAndGet(); decoding.complete(null); decode.join(); } @@ -102,6 +107,7 @@ public JsonNode decode(Job job) { if (when.equals("forced")) assertTrue(provisional.forceIfStuck(Duration.ZERO)); decode.complete(null); var context = typed.get(5, TimeUnit.SECONDS); + assertEquals(1, decodes.get(), "Peer arguments must only be decoded once"); if (when.equals("after_decode")) provisional.requestCancellation(cause); assertEquals(cause, context.cancellation()); assertTrue(context.awaitCancellation(Duration.ZERO)); @@ -129,4 +135,106 @@ else if (cause == WorkContext.Cancellation.SHUTDOWN) { } assertTrue(errors.isEmpty(), () -> "Unexpected worker errors: " + errors); } + + @ParameterizedTest + @ValueSource(strings = {"decode", "work"}) + void peerTimeoutInterruptsTheOwningWorker(String phase) throws Exception { + var entered = new CompletableFuture(); + var interrupted = new CompletableFuture(); + var release = new CountDownLatch(1); + var events = new LinkedBlockingQueue(); + var errors = new LinkedBlockingQueue(); + var database = databases.open(directory.resolve("river.db")); + var client = + new Client(database) + .withPilot( + new Pilot() { + @Override + public JsonNode decode(Job job) { + if (phase.equals("decode") && job.queue().equals("peers")) { + entered.complete(null); + try { + release.await(); + } catch (InterruptedException error) { + interrupted.complete(null); + throw new IllegalStateException("Peer decoding interrupted", error); + } + } + return job.args(); + } + }); + var peerJob = + client + .insert(TYPE, "peer", InsertOptions.builder().queue("peers").maxAttempts(1).build()) + .job(); + var workers = + client + .workers() + .queue("default", 1) + .leadership(false) + .pollOnly(true) + .jobTimeout(phase.equals("decode") ? Duration.ofNanos(1) : Duration.ofHours(1)) + .stuckThreshold(Duration.ofMillis(20)) + .stopTimeout(Duration.ofSeconds(5)) + .errorHandler(errors::add) + .add( + TYPE, + new Workers.Handler<>() { + @Override + public Duration timeout(Job job) { + return job.args().equals("parent") ? Duration.ofNanos(-1) : Duration.ofNanos(1); + } + + @Override + public void work(WorkContext context) throws Exception { + 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, + 1); + var rows = statement.executeQuery()) { + while (rows.next()) jobs.add(client.readPartial(rows)); + } + return jobs; + }); + assertEquals(1, peers.size()); + entered.complete(null); + try { + release.await(); + } catch (InterruptedException error) { + assertEquals( + WorkContext.Cancellation.TIMEOUT, peers.getFirst().cancellation()); + assertNull(context.cancellation(), "The parent's timeout is disabled"); + interrupted.complete(null); + throw error; + } + } + }) + .start(); + try (var subscription = workers.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(Job.State.DISCARDED, client.get(peerJob.id()).state()); + } finally { + release.countDown(); + workers.stopAndCancel(); + } + assertTrue(errors.isEmpty(), () -> "Unexpected worker errors: " + errors); + } } diff --git a/java/river/src/test/java/com/riverqueue/PeriodicJobBundleTest.java b/java/river/src/test/java/com/riverqueue/PeriodicJobBundleTest.java new file mode 100644 index 000000000..4fd432a08 --- /dev/null +++ b/java/river/src/test/java/com/riverqueue/PeriodicJobBundleTest.java @@ -0,0 +1,140 @@ +package com.riverqueue; + +import static org.junit.jupiter.api.Assertions.*; + +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; + +class PeriodicJobBundleTest { + private static final JobType TYPE = JobType.of("periodic_bundle_test", String.class); + + @Test + void addManyRejectsDuplicatesWithoutPartiallyRegisteringJobs() { + var wakes = new AtomicInteger(); + var bundle = new PeriodicJobBundle(List.of(), true, () -> true, wakes::incrementAndGet); + var existing = bundle.add(job("existing")); + assertThrows( + IllegalArgumentException.class, () -> bundle.addMany(List.of(job("new"), job("existing")))); + assertThrows( + IllegalArgumentException.class, () -> bundle.addMany(List.of(job("new"), job("new")))); + assertFalse(bundle.removeById("new")); + assertEquals(0, wakes.get()); + assertTrue(bundle.remove(existing)); + + var handles = bundle.addMany(List.of(job("first"), job("second"))); + assertEquals(0, wakes.get()); + assertTrue(bundle.removeById("first")); + assertFalse(bundle.remove(handles.getFirst()), "Handles must follow input order"); + assertTrue(bundle.remove(handles.get(1))); + } + + @Test + void additionsWakeOnlyRunOnStartJobsWhileRunning() { + var running = new java.util.concurrent.atomic.AtomicBoolean(true); + var wakes = new AtomicInteger(); + var bundle = new PeriodicJobBundle(List.of(), true, running::get, wakes::incrementAndGet); + bundle.add(job("scheduled")); + assertEquals(0, wakes.get()); + var immediate = + new PeriodicJob<>( + "immediate", + Schedule.every(Duration.ofHours(1)), + TYPE, + "args", + InsertOptions.defaults(), + true); + bundle.addMany(List.of(job("another"), immediate)); + assertEquals(1, wakes.get()); + running.set(false); + bundle.clear(); + var handle = bundle.add(immediate); + assertEquals(1, wakes.get()); + assertTrue(bundle.remove(handle)); + bundle.add(immediate); + assertTrue(bundle.removeById("immediate")); + } + + @Test + void anonymousRegistrationsHaveDistinctHandles() { + var bundle = bundle(); + var handles = bundle.addMany(List.of(job(null), job(null))); + assertNotSame(handles.getFirst(), handles.get(1)); + assertFalse(bundle.removeById(null)); + assertTrue(bundle.remove(handles.getFirst())); + assertTrue(bundle.remove(handles.get(1))); + } + + private static PeriodicJobBundle bundle() { + return new PeriodicJobBundle(List.of(), true, () -> true, () -> {}); + } + + @Test + void concurrentAddsDoNotRegisterDuplicateIds() throws Exception { + var bundle = bundle(); + var start = new CompletableFuture(); + var added = new AtomicInteger(); + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + var attempts = new java.util.ArrayList>(); + for (int i = 0; i < 10; i++) + attempts.add( + executor.submit( + () -> { + start.join(); + try { + bundle.add(job("same")); + added.incrementAndGet(); + } catch (IllegalArgumentException expected) { + assertEquals("Duplicate periodic job ID: same", expected.getMessage()); + } + })); + start.complete(null); + for (var attempt : attempts) attempt.get(5, java.util.concurrent.TimeUnit.SECONDS); + } + assertEquals(1, added.get()); + assertTrue(bundle.removeById("same")); + assertFalse(bundle.removeById("same")); + } + + @Test + void handlesCannotRemoveReplacementsOrJobsInAnotherBundle() { + var first = bundle(); + var second = bundle(); + var old = first.add(job("reused")); + assertTrue(first.removeById("reused")); + var replacement = first.add(job("reused")); + var other = second.add(job("reused")); + assertFalse(first.remove(old)); + assertFalse(first.remove(other)); + assertFalse(second.remove(replacement)); + assertTrue(first.remove(replacement)); + assertTrue(second.remove(other)); + } + + private static PeriodicJob job(String id) { + return new PeriodicJob<>( + id, Schedule.every(Duration.ofHours(1)), TYPE, "args", InsertOptions.defaults(), false); + } + + @Test + void removeManyAndClearReleaseIdsWithoutReusingHandles() { + var bundle = bundle(); + var handles = bundle.addMany(List.of(job("first"), job("second"), job("third"))); + bundle.removeMany(List.of(handles.getFirst(), handles.get(1), handles.getFirst())); + assertFalse(bundle.removeById("first")); + assertFalse(bundle.removeById("second")); + bundle.removeManyById(List.of("third", "missing")); + assertFalse(bundle.remove(handles.get(2))); + var cleared = bundle.add(job("first")); + bundle.add(job("second")); + bundle.clear(); + var replacement = bundle.add(job("first")); + assertFalse(bundle.remove(cleared)); + assertFalse(bundle.removeById("second")); + assertTrue(bundle.remove(replacement)); + bundle.clear(); + } +} diff --git a/java/river/src/test/java/com/riverqueue/ProtocolStorageTest.java b/java/river/src/test/java/com/riverqueue/ProtocolStorageTest.java index 75a7f23ad..081ccdba6 100644 --- a/java/river/src/test/java/com/riverqueue/ProtocolStorageTest.java +++ b/java/river/src/test/java/com/riverqueue/ProtocolStorageTest.java @@ -109,7 +109,7 @@ public boolean clean(Client client, Instant now) { Json.tree(client.get(inserted.id()).errors().getFirst())); var completed = new CompletableFuture(); - var writeConflict = new CompletableFuture(); + var writeFailure = new CompletableFuture(); var errors = new LinkedBlockingQueue(); try (var workers = client @@ -120,16 +120,16 @@ public boolean clean(Client client, Instant now) { .retryPolicy( job -> { if (!postgres) { - // Retry selection is between rescue's read and update. A second writer - // must already be excluded, or its commit would invalidate that snapshot. + // Retry selection runs after the read transaction closes, so another + // writer can commit without invalidating rescue's update transaction. try (var connection = database.connection(); var statement = connection.createStatement()) { statement.execute("PRAGMA busy_timeout=0"); statement.executeUpdate( "UPDATE river_job SET priority = 2 WHERE id = " + job.id()); - writeConflict.complete(null); + writeFailure.complete(null); } catch (java.sql.SQLException error) { - writeConflict.complete(error); + writeFailure.complete(error); } } return Duration.ofDays(1); @@ -138,14 +138,13 @@ public boolean clean(Client client, Instant now) { .start(); var subscription = workers.subscribe(completed::complete, Workers.EventKind.JOB_COMPLETED)) { - if (!postgres) { - var conflict = writeConflict.get(5, TimeUnit.SECONDS); - assertNotNull(conflict, "Rescue must reserve the writer before selecting jobs"); - assertEquals(5, conflict.getErrorCode(), "Expected SQLITE_BUSY for the competing writer"); - } + if (!postgres) + assertNull( + writeFailure.get(5, TimeUnit.SECONDS), "Rescue callbacks must not hold the writer"); maintained.get(5, TimeUnit.SECONDS); var rescued = client.get(inserted.id()); assertEquals(Job.State.RETRYABLE, rescued.state()); + if (!postgres) assertEquals(2, rescued.priority()); assertEquals(3, rescued.metadata().required(rescueKey).asInt()); assertEquals(2, rescued.errors().size()); assertEquals(fixture.required("attempt_error"), Json.tree(rescued.errors().getFirst())); diff --git a/java/river/src/test/java/com/riverqueue/RescueTest.java b/java/river/src/test/java/com/riverqueue/RescueTest.java index f85a5ecfb..26530af95 100644 --- a/java/river/src/test/java/com/riverqueue/RescueTest.java +++ b/java/river/src/test/java/com/riverqueue/RescueTest.java @@ -19,12 +19,278 @@ import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; +import tools.jackson.databind.JsonNode; class RescueTest { private static final JobType TYPE = JobType.of("rescue_test", String.class); @RegisterExtension final TestDatabase databases = new TestDatabase(); @TempDir Path directory; + @ParameterizedTest + @ValueSource(strings = {"default", "disabled", "error", "override", "throw", "zero"}) + void pilotReceivesWorkerTimeoutCallback(String mode) throws Exception { + var request = new CompletableFuture(); + var errors = new LinkedBlockingQueue(); + var client = + new Client(databases.open(directory.resolve("river.db"))) + .withPilot( + new Pilot() { + @Override + public boolean rescue(Client client, Rescue rescue) { + request.complete(rescue); + return true; + } + }); + var job = client.insert(TYPE, mode).job(); + try (var workers = + client + .workers() + .queue("other", 1) + .pollOnly(true) + .jobTimeout(Duration.ofMinutes(5)) + .stopTimeout(Duration.ofSeconds(5)) + .errorHandler(errors::add) + .add( + TYPE, + new Workers.Handler<>() { + @Override + public Duration timeout(Job job) { + assertEquals(mode, job.args()); + return switch (mode) { + case "disabled" -> Duration.ofNanos(-1); + case "error" -> throw new AssertionError("timeout error"); + case "override" -> Duration.ofMinutes(2); + case "throw" -> throw new IllegalStateException("bad timeout"); + case "zero" -> Duration.ZERO; + default -> null; + }; + } + + @Override + public void work(WorkContext context) { + fail("Wrong queue"); + } + }) + .start()) { + var rescue = request.get(5, TimeUnit.SECONDS); + var expected = + switch (mode) { + case "disabled" -> Duration.ofNanos(-1); + case "error", "throw" -> Duration.ZERO; + case "override" -> Duration.ofMinutes(2); + default -> Duration.ofMinutes(5); + }; + assertEquals(expected, rescue.timeout().apply(job)); + } + if (mode.equals("error")) assertInstanceOf(AssertionError.class, errors.remove()); + if (mode.equals("throw")) assertInstanceOf(IllegalStateException.class, errors.remove()); + assertTrue(errors.isEmpty(), () -> "Unexpected worker errors: " + errors); + } + + @ParameterizedTest + @ValueSource(strings = {"decode", "retry", "timeout"}) + void rescueCallbacksReleaseDatabaseLocksAndRejectStaleAttempts(String mode) throws Exception { + var entered = new CompletableFuture(); + var release = new CompletableFuture(); + var maintained = new CompletableFuture(); + var errors = new LinkedBlockingQueue(); + var database = databases.open(directory.resolve("river.db")); + var client = + new Client(database) + .withPilot( + new Pilot() { + @Override + public JsonNode decode(Job job) { + if (mode.equals("decode")) { + entered.complete(null); + release.join(); + } + return job.args(); + } + + @Override + public boolean clean(Client client, Instant now) { + maintained.complete(null); + return true; + } + }); + var job = client.insert(TYPE, "stuck").job(); + client.transaction( + connection -> { + try (var statement = + Sql.prepare( + connection, + "UPDATE " + + database.prefix() + + "river_job SET state='running',attempt=1,attempted_at=? WHERE id=?", + database.timestamp(Instant.EPOCH), + job.id())) { + statement.executeUpdate(); + } + return null; + }); + var workers = + client + .workers() + .queue("other", 1) + .pollOnly(true) + .serviceInterval(Duration.ofHours(1)) + .stopTimeout(Duration.ofSeconds(5)) + .errorHandler(errors::add) + .retryPolicy( + ignored -> { + if (mode.equals("retry")) { + entered.complete(null); + release.join(); + } + return Duration.ofDays(1); + }) + .add( + TYPE, + new Workers.Handler<>() { + @Override + public Duration timeout(Job job) { + if (mode.equals("timeout")) { + entered.complete(null); + release.join(); + } + return Duration.ofNanos(1); + } + + @Override + public void work(WorkContext context) { + fail("Wrong queue"); + } + }) + .start(); + try { + entered.get(5, TimeUnit.SECONDS); + var write = + CompletableFuture.runAsync( + () -> + client.transaction( + connection -> { + try (var statement = + Sql.prepare( + connection, + "UPDATE " + + database.prefix() + + "river_job SET attempt=2,attempted_at=? WHERE id=?", + database.timestamp(Instant.now()), + job.id())) { + statement.executeUpdate(); + } + return null; + })); + write.get(2, TimeUnit.SECONDS); + release.complete(null); + maintained.get(5, TimeUnit.SECONDS); + var updated = client.get(job.id()); + assertEquals(Job.State.RUNNING, updated.state()); + assertEquals(2, updated.attempt()); + assertTrue(updated.errors().isEmpty()); + } finally { + release.complete(null); + workers.stop(); + } + assertTrue(errors.isEmpty(), () -> "Unexpected worker errors: " + errors); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void rescueHonorsWorkerTimeouts(boolean defaultDisabled) throws Exception { + var maintained = new CompletableFuture(); + var errors = new LinkedBlockingQueue(); + var client = + new Client(databases.open(directory.resolve("river.db"))) + .withPilot( + new Pilot() { + @Override + public boolean clean(Client client, Instant now) { + maintained.complete(null); + return true; + } + }); + var jobs = new java.util.LinkedHashMap(); + for (var args : + List.of("cancelled", "default", "disabled", "error", "invalid", "long", "short", "zero")) { + var job = + client + .insert( + TYPE, + args, + InsertOptions.builder() + .maxAttempts(1) + .metadata( + args.equals("cancelled") + ? java.util.Map.of("cancel_attempted_at", Instant.now().toString()) + : java.util.Map.of()) + .build()) + .job(); + jobs.put(args, job.id()); + client.transaction( + connection -> { + try (var statement = + Sql.prepare( + connection, + "UPDATE " + + client.database().prefix() + + "river_job SET state='running',attempt=1,attempted_at=? WHERE id=?", + client.database().timestamp(Instant.now().minus(Duration.ofHours(2))), + job.id())) { + statement.executeUpdate(); + } + return null; + }); + } + try (var workers = + client + .workers() + .queue("other", 1) + .pollOnly(true) + .jobTimeout(Duration.ofNanos(defaultDisabled ? -1 : 1)) + .rescueAfter(Duration.ofHours(1)) + .stopTimeout(Duration.ofSeconds(5)) + .errorHandler(errors::add) + .add( + TYPE, + new Workers.Handler<>() { + @Override + public Duration timeout(Job job) { + return switch (job.args()) { + case "cancelled", "disabled" -> Duration.ofNanos(-1); + case "error" -> throw new AssertionError("timeout error"); + case "invalid" -> throw new IllegalStateException("bad timeout"); + case "zero" -> Duration.ZERO; + case "long" -> Duration.ofHours(3); + case "short" -> Duration.ofMinutes(1); + default -> null; + }; + } + + @Override + public void work(WorkContext context) { + fail("Wrong queue"); + } + }) + .start()) { + maintained.get(5, TimeUnit.SECONDS); + for (var entry : jobs.entrySet()) { + var expected = + switch (entry.getKey()) { + case "cancelled" -> Job.State.CANCELLED; + case "default", "zero" -> defaultDisabled ? Job.State.RUNNING : Job.State.DISCARDED; + case "disabled", "long" -> Job.State.RUNNING; + default -> Job.State.DISCARDED; + }; + assertEquals(expected, client.get(entry.getValue()).state(), entry.getKey()); + } + } + assertEquals("timeout error", errors.remove().getMessage()); + assertEquals("bad timeout", errors.remove().getMessage()); + assertTrue(errors.isEmpty(), () -> "Unexpected worker errors: " + errors); + } + @ParameterizedTest @ValueSource(strings = {"local", "peer", "peer_claim_retry", "peer_notification_retry"}) void shortRescuedRetriesWakeFetchingBeforeMaintenance(String mode) throws Exception { diff --git a/java/river/src/test/java/com/riverqueue/ResumableTest.java b/java/river/src/test/java/com/riverqueue/ResumableTest.java new file mode 100644 index 000000000..f8106a545 --- /dev/null +++ b/java/river/src/test/java/com/riverqueue/ResumableTest.java @@ -0,0 +1,390 @@ +package com.riverqueue; + +import static org.junit.jupiter.api.Assertions.*; + +import java.nio.file.Path; +import java.time.Duration; +import java.time.Instant; +import java.util.Map; +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.ValueSource; + +class ResumableTest { + private static final JobType TYPE = JobType.of("checkpoint_test", String.class); + @RegisterExtension final TestDatabase databases = new TestDatabase(); + @TempDir Path directory; + + @ParameterizedTest + @CsvSource({ + "false,false,false", + "false,false,true", + "false,true,false", + "false,true,true", + "true,false,false", + "true,false,true", + "true,true,false", + "true,true,true" + }) + 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 errors = new LinkedBlockingQueue(); + var preparations = new AtomicInteger(); + var steps = new LinkedBlockingQueue(); + try (var workers = + client + .workers() + .queue("default", 1) + .pollOnly(true) + .leadership(false) + .retryPolicy(job -> Duration.ofDays(1)) + .stopTimeout(Duration.ofSeconds(5)) + .errorHandler(errors::add) + .add( + TYPE, + context -> { + if (priorCheckpoint) + context.stepWithCursor( + "initial", + cursor -> { + assertEquals(1, context.job().attempt()); + context.transaction(connection -> context.checkpoint(connection, 1)); + }); + context.step( + "prepare", () -> assertEquals(1, preparations.incrementAndGet())); + WorkContext.CursorStep action = + cursor -> { + steps.add(context.job().attempt()); + if (context.job().attempt() > 1) { + if (withCursor && !rollback) + assertEquals(2, cursor.path("offset").asInt()); + else assertTrue(cursor.isMissingNode()); + return; + } + context.output("deferred"); + if (withCursor) context.cursor(Map.of("offset", 1)); + context.transaction( + connection -> { + client.insert( + connection, + TYPE, + "application write", + InsertOptions.builder().queue("other").build()); + var updated = + withCursor + ? context.checkpoint(connection, Map.of("offset", 2)) + : context.checkpoint(connection); + assertEquals(Job.State.RUNNING, updated.state()); + assertEquals( + "process", + updated + .metadata() + .path(Protocol.METADATA_RESUMABLE_STEP) + .asString()); + assertFalse(updated.metadata().has(Protocol.METADATA_OUTPUT)); + if (rollback) throw new IllegalStateException("rollback"); + return null; + }); + throw new IllegalStateException("retry after commit"); + }; + if (withCursor) context.stepWithCursor("process", action); + else context.step("process", () -> action.run(Json.object().path("missing"))); + }) + .start(); + var subscription = workers.subscribe(events::add)) { + var inserted = + client + .insert( + TYPE, + "work", + InsertOptions.builder() + .metadata(Map.of("unrelated", Map.of("keep", true))) + .build()) + .job(); + var failed = events.poll(5, TimeUnit.SECONDS); + assertNotNull(failed); + assertEquals(Workers.EventKind.JOB_FAILED, failed.kind()); + var saved = client.get(inserted.id()); + assertEquals( + rollback ? "prepare" : "process", + saved.metadata().path(Protocol.METADATA_RESUMABLE_STEP).asString()); + assertEquals("deferred", saved.metadata().path(Protocol.METADATA_OUTPUT).asString()); + assertTrue(saved.metadata().path("unrelated").path("keep").asBoolean()); + assertEquals(rollback ? 1 : 2, client.list(JobQuery.all()).jobs().size()); + if (rollback) { + var cursor = saved.metadata().path(Protocol.METADATA_RESUMABLE_CURSOR); + assertTrue(cursor.isMissingNode() || cursor.isNull()); + } + client.retry(inserted.id()); + var completed = events.poll(5, TimeUnit.SECONDS); + assertNotNull(completed); + assertEquals(Workers.EventKind.JOB_COMPLETED, completed.kind()); + assertEquals(1, preparations.get()); + assertEquals( + rollback || withCursor ? java.util.List.of(1, 2) : java.util.List.of(1), + java.util.List.copyOf(steps)); + } + assertTrue(errors.isEmpty(), () -> "Unexpected worker errors: " + errors); + } + + @Test + void checkpointRejectsInvalidTransactionsAndStaleAttempts() throws Exception { + var client = new Client(databases.open(directory.resolve("river.db"))); + var inserted = client.insert(TYPE, "work").job(); + client.transaction( + connection -> { + try (var statement = + Sql.prepare( + connection, + "UPDATE " + + client.database().prefix() + + "river_job SET state='running',attempt=1,attempted_at=? WHERE id=?", + client.database().timestamp(Instant.now()), + inserted.id())) { + statement.executeUpdate(); + } + return null; + }); + var job = client.get(inserted.id(), TYPE); + var context = new WorkContext<>(client, job); + try (var connection = client.database().connection()) { + assertThrows(IllegalStateException.class, () -> context.checkpoint(connection)); + context.step( + "process", + () -> { + assertThrows(IllegalArgumentException.class, () -> context.checkpoint(connection)); + connection.setAutoCommit(false); + client.complete(connection, job); + assertEquals( + RiverException.Code.REJECTED, + assertThrows(RiverException.class, () -> context.checkpoint(connection)).code()); + assertEquals(Job.State.COMPLETED, client.get(connection, job.id()).state()); + connection.rollback(); + try (var statement = + Sql.prepare( + connection, + "UPDATE " + client.database().prefix() + "river_job SET attempt=2 WHERE id=?", + job.id())) { + statement.executeUpdate(); + } + assertThrows(RiverException.class, () -> context.checkpoint(connection)); + connection.rollback(); + }); + } + assertFalse(client.get(job.id()).metadata().has(Protocol.METADATA_RESUMABLE_STEP)); + } + + @ParameterizedTest + @CsvSource({"false,false", "true,false", "false,true"}) + void checkpointRetainsOnlyCommittedProgress(boolean rollback, boolean finishStep) + throws Exception { + var client = new Client(databases.open(directory.resolve("river.db"))); + var inserted = client.insert(TYPE, "work").job(); + client.transaction( + connection -> { + try (var statement = + Sql.prepare( + connection, + "UPDATE " + + client.database().prefix() + + "river_job SET state='running',attempt=1,attempted_at=? WHERE id=?", + client.database().timestamp(Instant.now()), + inserted.id())) { + statement.executeUpdate(); + } + return null; + }); + var context = new WorkContext<>(client, client.get(inserted.id(), TYPE)); + var failure = new IllegalStateException("retry"); + WorkContext.Step work = + () -> + context.stepWithCursor( + "process", + cursor -> { + context.transaction(connection -> context.checkpoint(connection, 1)); + context.transaction( + connection -> { + context.checkpoint(connection, 2); + if (rollback) throw failure; + return null; + }); + if (!finishStep) throw failure; + }); + if (finishStep) work.run(); + else assertSame(failure, assertThrows(IllegalStateException.class, work::run)); + context.finish(failure); + if (finishStep) { + assertEquals( + "process", context.metadataUpdates().path(Protocol.METADATA_RESUMABLE_STEP).asString()); + assertTrue(context.metadataUpdates().path(Protocol.METADATA_RESUMABLE_CURSOR).isNull()); + } else assertTrue(context.metadataUpdates().isEmpty(), "Checkpoint must not be buffered again"); + assertEquals( + rollback ? 1 : 2, + client + .get(inserted.id()) + .metadata() + .path(Protocol.METADATA_RESUMABLE_CURSOR) + .path("process") + .asInt()); + } + + @ParameterizedTest + @ValueSource(strings = {"plain", "cursor", "null", "first", "first_null"}) + void cursorDoesNotCompleteFailedSteps(String mode) throws Exception { + var client = new Client(databases.open(directory.resolve("river.db"))); + 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() + .queue("default", 1) + .pollOnly(true) + .leadership(false) + .retryPolicy(job -> Duration.ofDays(1)) + .stopTimeout(Duration.ofSeconds(5)) + .errorHandler(errors::add) + .add( + TYPE, + context -> { + if (!first) + context.step("prepare", () -> assertEquals(1, prepares.incrementAndGet())); + WorkContext.CursorStep action = + cursor -> { + calls.incrementAndGet(); + if (context.job().attempt() == 1) { + context.cursor(nullCursor ? null : Map.of("offset", 1)); + throw new IllegalStateException("retry"); + } + if (!mode.equals("plain")) { + if (nullCursor) assertTrue(cursor.isNull()); + else assertEquals(1, cursor.path("offset").asInt()); + } + }; + if (mode.equals("plain")) + context.step("process", () -> action.run(Json.tree(null))); + else context.stepWithCursor("process", action); + }) + .start(); + var subscription = workers.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()); + var metadata = client.get(job.id()).metadata(); + assertEquals( + first ? "" : "prepare", metadata.path(Protocol.METADATA_RESUMABLE_STEP).asString("")); + assertTrue(metadata.path(Protocol.METADATA_RESUMABLE_CURSOR).has("process")); + client.retry(job.id()); + var completed = events.poll(5, TimeUnit.SECONDS); + assertNotNull(completed); + assertEquals(Workers.EventKind.JOB_COMPLETED, completed.kind()); + assertEquals(2, calls.get()); + assertEquals(first ? 0 : 1, prepares.get()); + } + assertTrue(errors.isEmpty(), () -> "Unexpected worker errors: " + errors); + } + + @Test + void cursorInFirstStepIsBufferedAndNestedStepsAreRejected() throws Exception { + var client = new Client(databases.open(directory.resolve("river.db"))); + var context = new WorkContext<>(client, client.insert(TYPE, "work").job()); + var failure = new IllegalStateException("retry"); + assertSame( + failure, + assertThrows( + IllegalStateException.class, + () -> + context.stepWithCursor( + "first", + cursor -> { + assertThrows( + IllegalStateException.class, () -> context.step("nested", () -> {})); + context.cursor(Map.of("offset", 3)); + throw failure; + }))); + assertSame(failure, context.finish(failure)); + assertFalse(context.metadataUpdates().has(Protocol.METADATA_RESUMABLE_STEP)); + assertEquals( + 3, + context + .metadataUpdates() + .path(Protocol.METADATA_RESUMABLE_CURSOR) + .path("first") + .path("offset") + .asInt()); + } + + @Test + void cursorMapsReplaceCompletedStepsAndPreserveNullValues() throws Exception { + var client = new Client(databases.open(directory.resolve("river.db"))); + var events = new LinkedBlockingQueue(); + var errors = new LinkedBlockingQueue(); + var firstCalls = new AtomicInteger(); + try (var workers = + client + .workers() + .queue("default", 1) + .pollOnly(true) + .leadership(false) + .retryPolicy(job -> Duration.ofDays(1)) + .stopTimeout(Duration.ofSeconds(5)) + .errorHandler(errors::add) + .add( + TYPE, + context -> { + context.stepWithCursor( + "first", + cursor -> { + firstCalls.incrementAndGet(); + if (context.job().attempt() == 1) { + context.cursor(1); + throw new IllegalStateException("retry first"); + } + assertEquals(1, cursor.asInt()); + }); + context.stepWithCursor( + "second", + cursor -> { + if (context.job().attempt() == 2) { + context.cursor( + Json.parse("{\"offset\":2,\"nested\":{\"value\":null}}")); + throw new IllegalStateException("retry second"); + } + assertEquals(2, cursor.path("offset").asInt()); + assertTrue(cursor.path("nested").has("value")); + assertTrue(cursor.path("nested").path("value").isNull()); + }); + }) + .start(); + var subscription = workers.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, + event.kind()); + if (attempt == 2) { + var cursors = client.get(job.id()).metadata().path(Protocol.METADATA_RESUMABLE_CURSOR); + assertFalse(cursors.has("first")); + assertTrue(cursors.has("second")); + } + if (attempt < 3) client.retry(job.id()); + } + assertEquals(2, firstCalls.get()); + } + assertTrue(errors.isEmpty(), () -> "Unexpected worker errors: " + errors); + } +} diff --git a/java/river/src/test/java/com/riverqueue/SchedulingTest.java b/java/river/src/test/java/com/riverqueue/SchedulingTest.java index e6f273eb4..0d3e565da 100644 --- a/java/river/src/test/java/com/riverqueue/SchedulingTest.java +++ b/java/river/src/test/java/com/riverqueue/SchedulingTest.java @@ -30,6 +30,202 @@ class SchedulingTest { @RegisterExtension final TestDatabase databases = new TestDatabase(); @TempDir Path directory; + @Test + void anonymousPeriodicJobsOmitTheIdMetadata() throws Exception { + var client = new Client(databases.open(directory.resolve("river.db"))); + var worked = new CompletableFuture>(); + var errors = new LinkedBlockingQueue(); + try (var workers = + client + .workers() + .queue("default", 1) + .pollOnly(true) + .stopTimeout(Duration.ofSeconds(5)) + .errorHandler(errors::add) + .add(TYPE, context -> worked.complete(context.job())) + .periodic( + new PeriodicJob<>( + null, + Schedule.every(Duration.ofDays(1)), + TYPE, + "args", + InsertOptions.defaults(), + true)) + .start()) { + var job = worked.get(5, TimeUnit.SECONDS); + assertTrue(job.metadata().path("periodic").asBoolean()); + assertFalse(job.metadata().has(Protocol.METADATA_PERIODIC_JOB_ID)); + } + assertTrue(errors.isEmpty(), () -> "Unexpected worker errors: " + errors); + } + + @Test + void dynamicPeriodicJobsRequireLeadershipAndValidIds() throws Exception { + var client = new Client(databases.open(directory.resolve("river.db"))); + try (var workers = + client + .workers() + .queue("default", 1) + .pollOnly(true) + .leadership(false) + .add(TYPE, context -> {}) + .start()) { + assertThrows( + IllegalStateException.class, + () -> + workers + .periodicJobs() + .add( + "dynamic", + Schedule.every(Duration.ofDays(1)), + TYPE, + "args", + InsertOptions.defaults(), + true)); + } + try (var workers = + client.workers().queue("default", 1).pollOnly(true).add(TYPE, context -> {}).start()) { + for (var id : java.util.List.of("", "invalid id", "x".repeat(128))) + assertThrows( + IllegalArgumentException.class, + () -> + workers + .periodicJobs() + .add( + id, + Schedule.every(Duration.ofDays(1)), + TYPE, + "args", + InsertOptions.defaults(), + true)); + } + } + + @Test + void dynamicPeriodicJobsWaitForLeadership() throws Exception { + var checks = new LinkedBlockingQueue(); + var errors = new LinkedBlockingQueue(); + var worked = new LinkedBlockingQueue(); + var client = new Client(databases.open(directory.resolve("river.db"))); + var builder = + client + .workers() + .queue("default", 1) + .pollOnly(true) + .serviceInterval(Duration.ofMillis(20)) + .stopTimeout(Duration.ofSeconds(5)) + .errorHandler(errors::add) + .add(TYPE, context -> worked.add(context.args())); + var first = builder.observe(checks::add).start(); + try { + assertEquals("leadership_check", checks.poll(5, TimeUnit.SECONDS)); + assertTrue(first.isLeader()); + try (var second = builder.start()) { + second + .periodicJobs() + .add( + "dynamic", + Schedule.every(Duration.ofDays(1)), + TYPE, + "args", + InsertOptions.defaults(), + true); + // The leader's builder snapshot and the follower's local registration are independent. + assertNull(worked.poll(100, TimeUnit.MILLISECONDS)); + assertFalse(second.isLeader()); + first.stop(); + assertEquals("args", worked.poll(5, TimeUnit.SECONDS)); + assertTrue(second.isLeader()); + } + } finally { + first.stop(); + } + assertTrue(errors.isEmpty(), () -> "Unexpected worker errors: " + errors); + } + + @Test + void dynamicPeriodicJobsWakeMaintenanceAndCanBeReplaced() throws Exception { + var entered = new CompletableFuture(); + var release = new CompletableFuture(); + var worked = new LinkedBlockingQueue(); + var errors = new LinkedBlockingQueue(); + var client = + new Client(databases.open(directory.resolve("river.db"))) + .withPilot( + new Pilot() { + @Override + public void maintain(Client client, Instant now) { + if (entered.complete(null)) release.join(); + } + }); + var workers = + client + .workers() + .queue("default", 1) + .pollOnly(true) + .serviceInterval(Duration.ofHours(1)) + .stopTimeout(Duration.ofSeconds(5)) + .errorHandler(errors::add) + .add(TYPE, context -> worked.add(context.args())) + .start(); + try { + entered.get(5, TimeUnit.SECONDS); + workers + .periodicJobs() + .add( + "dynamic", + Schedule.every(Duration.ofDays(1)), + TYPE, + "first", + InsertOptions.defaults(), + true); + assertThrows( + IllegalArgumentException.class, + () -> + workers + .periodicJobs() + .add( + "dynamic", + Schedule.every(Duration.ofDays(1)), + TYPE, + "duplicate", + InsertOptions.defaults(), + true)); + release.complete(null); + assertEquals("first", worked.poll(5, TimeUnit.SECONDS)); + assertTrue(workers.periodicJobs().removeById("dynamic")); + assertFalse(workers.periodicJobs().removeById("dynamic")); + workers + .periodicJobs() + .add( + "dynamic", + Schedule.every(Duration.ofDays(1)), + TYPE, + "replacement", + InsertOptions.defaults(), + true); + assertEquals("replacement", worked.poll(5, TimeUnit.SECONDS)); + assertTrue(workers.periodicJobs().removeById("dynamic")); + } finally { + release.complete(null); + workers.stop(); + } + var stopped = + workers + .periodicJobs() + .add( + "stopped", + Schedule.every(Duration.ofDays(1)), + TYPE, + "args", + InsertOptions.defaults(), + true); + assertTrue(workers.periodicJobs().remove(stopped)); + workers.periodicJobs().clear(); + assertTrue(worked.isEmpty()); + assertTrue(errors.isEmpty(), () -> "Unexpected worker errors: " + errors); + } + @ParameterizedTest @ValueSource(strings = {"initial schedule", "following schedule", "insertion"}) void failedPeriodicJobDoesNotBlockMaintenanceOrOtherJobs(String failurePoint) throws Exception { @@ -228,4 +424,63 @@ public Optional next(ZonedDateTime after) { } assertTrue(errors.isEmpty(), () -> "Unexpected worker errors: " + errors); } + + @ParameterizedTest + @ValueSource(strings = {"handle", "id", "handles", "ids", "clear"}) + void removedPeriodicJobIsNotInsertedAfterItsScheduleReturns(String removal) throws Exception { + var entered = new CompletableFuture(); + var release = new CompletableFuture(); + var maintained = new CompletableFuture(); + var errors = new LinkedBlockingQueue(); + var client = + new Client(databases.open(directory.resolve("river.db"))) + .withPilot( + new Pilot() { + @Override + public void maintain(Client client, Instant now) { + if (entered.isDone()) maintained.complete(null); + } + }); + var workers = + client + .workers() + .queue("default", 1) + .pollOnly(true) + .stopTimeout(Duration.ofSeconds(5)) + .errorHandler(errors::add) + .add(TYPE, context -> {}) + .start(); + try { + var handle = + workers + .periodicJobs() + .add( + "removed", + after -> { + entered.complete(null); + release.join(); + return Optional.of(after.plusDays(1)); + }, + TYPE, + "args", + InsertOptions.defaults(), + true); + entered.get(5, TimeUnit.SECONDS); + switch (removal) { + case "handle" -> assertTrue(workers.periodicJobs().remove(handle)); + case "id" -> assertTrue(workers.periodicJobs().removeById("removed")); + case "handles" -> workers.periodicJobs().removeMany(java.util.List.of(handle)); + case "ids" -> workers.periodicJobs().removeManyById(java.util.List.of("removed")); + case "clear" -> workers.periodicJobs().clear(); + default -> fail("Unknown removal mode"); + } + release.complete(null); + maintained.get(5, TimeUnit.SECONDS); + assertTrue(client.list(JobQuery.all()).jobs().isEmpty()); + } finally { + release.complete(null); + workers.stop(); + } + assertTrue(errors.isEmpty(), () -> "Unexpected worker errors: " + errors); + } } diff --git a/java/river/src/test/java/com/riverqueue/TimeoutTest.java b/java/river/src/test/java/com/riverqueue/TimeoutTest.java new file mode 100644 index 000000000..76ed0815f --- /dev/null +++ b/java/river/src/test/java/com/riverqueue/TimeoutTest.java @@ -0,0 +1,82 @@ +package com.riverqueue; + +import static org.junit.jupiter.api.Assertions.*; + +import java.nio.file.Path; +import java.time.Duration; +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.ValueSource; + +class TimeoutTest { + private static final JobType TYPE = JobType.of("timeout_test", String.class); + @RegisterExtension final TestDatabase databases = new TestDatabase(); + @TempDir Path directory; + + @ParameterizedTest + @ValueSource(strings = {"inherit", "override", "disabled", "longer", "zero", "overflow", "throw"}) + void handlerTimeoutControlsTheAttempt(String mode) throws Exception { + var client = new Client(databases.open(directory.resolve("river.db"))); + var calls = new AtomicInteger(); + var worked = new AtomicInteger(); + var result = new CompletableFuture(); + var errors = new LinkedBlockingQueue(); + boolean invalid = java.util.Set.of("overflow", "throw").contains(mode); + boolean completes = mode.equals("disabled") || mode.equals("longer"); + try (var workers = + client + .workers() + .queue("default", 1) + .pollOnly(true) + .leadership(false) + .jobTimeout(mode.equals("override") ? Duration.ofHours(1) : Duration.ofNanos(1)) + .stopTimeout(Duration.ofSeconds(5)) + .errorHandler(errors::add) + .add( + TYPE, + new Workers.Handler<>() { + @Override + public Duration timeout(Job job) { + assertEquals(mode, job.args()); + calls.incrementAndGet(); + return switch (job.args()) { + case "disabled" -> Duration.ofNanos(-1); + case "longer" -> Duration.ofHours(1); + case "override" -> Duration.ofNanos(1); + case "zero" -> Duration.ZERO; + case "overflow" -> Duration.ofSeconds(Long.MAX_VALUE); + case "throw" -> throw new IllegalStateException("bad timeout"); + default -> null; + }; + } + + @Override + public void work(WorkContext context) throws Exception { + worked.incrementAndGet(); + if (completes) + assertFalse(context.awaitCancellation(Duration.ofMillis(100))); + else { + assertTrue(context.awaitCancellation(Duration.ofSeconds(5))); + assertEquals(WorkContext.Cancellation.TIMEOUT, context.cancellation()); + context.checkCancelled(); + } + } + }) + .start(); + var subscription = workers.subscribe(result::complete)) { + client.insert(TYPE, mode, InsertOptions.builder().maxAttempts(1).build()); + var event = result.get(5, TimeUnit.SECONDS); + assertEquals( + completes ? Workers.EventKind.JOB_COMPLETED : Workers.EventKind.JOB_FAILED, event.kind()); + assertEquals(completes ? Job.State.COMPLETED : Job.State.DISCARDED, event.job().state()); + assertEquals(1, calls.get()); + assertEquals(invalid ? 0 : 1, worked.get()); + } + assertTrue(errors.isEmpty(), () -> "Unexpected worker errors: " + errors); + } +}