Add Java transactional checkpoints, dynamic periodic jobs, and worker timeouts

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<A>` 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.
 

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.

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<SendEmail>() {
+  @Override
+  public Duration timeout(Job<SendEmail> job) {
+    return Duration.ofMinutes(5);
+  }
+
+  @Override
+  public void work(WorkContext<SendEmail> 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()

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",

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

@@ -324,11 +324,13 @@ public <A> Job<A> complete(Connection connection, Job<A> 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(),

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<A>(
+    String id,
+    Schedule schedule,
+    JobType<A> 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));
+  }
+}