Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions java/.gitignore
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
target/
.idea/
*.iml
.classpath
.project
.settings/
.conformance/
__pycache__/
13 changes: 7 additions & 6 deletions java/API.md
Original file line number Diff line number Diff line change
Expand Up @@ -10,10 +10,10 @@ Java exposes every Go helper. Stored values and protocol behavior still follow G
| `JobArgs`, `Job[T]`, worker interfaces and registration | `JobType<A>`, `Job<A>`, `Workers.add(type, lambda)`, `Client.Builder.worker` | Records need no River interface; explicit kinds stay stable when Java classes are renamed. |
| `Insert`, `InsertTx`, `InsertMany`, `InsertManyTx` | `insert`, `insertMany`, connection-first overloads | Arguments are typed; results contain persisted `Job<JsonNode>` rows, including cross-kind duplicates. Mixed batches use `JobType.submission`. |
| `InsertOpts`, `UniqueOpts`, struct tags | `InsertOptions`, `Unique`, `JobType.uniqueBy` | Named builders and fluent options avoid positional flags. Explicit JSON paths replace struct tags. |
| `JobGet`, `JobList`, cursor/order/filter params | `get`, `JobQuery`, `JobQuery.Page` | Raw reads use `JsonNode`; `get(id, type)` checks the kind and decodes a known argument type. Pagination remains explicit. |
| `JobCancel`, `JobDelete`, `JobDeleteMany`, `JobRetry` and transaction variants | `cancel`, `delete`, `deleteMany`, `retry`, connection-first overloads | Short operation names are unambiguous on `Client`; transaction ownership stays visible in the argument list. |
| `JobUpdate` output parameter, `RecordOutput` | `Client.output`, `WorkContext.output` | Names describe the supported operation; attempt output is buffered until completion. |
| `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. |
| `JobGet`, `JobList`, cursor/order/filter params | `client.jobs().get/list`, `JobQuery`, `JobQuery.Page` | Raw reads use `JsonNode`; `jobs().get(id, type)` checks the kind and decodes a known argument type. Pagination remains explicit. |
| `JobCancel`, `JobDelete`, `JobDeleteMany`, `JobRetry` and transaction variants | `client.jobs().cancel/delete/deleteMany/retry`, connection-first overloads | Job operations live on `Jobs`, alongside `Client.queues()`; transaction ownership stays visible in the argument list. |
| `JobUpdate`, `JobUpdateTx`, `RecordOutput` | `client.jobs().update`, `JobUpdateOptions`, `WorkContext.output` | Updates merge metadata and set output immediately; connection-first overloads support transactions. Attempt output remains buffered until completion. |
| `JobCompleteTx` | `client.jobs().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. |
Expand Down Expand Up @@ -57,8 +57,9 @@ Java exposes every Go helper. Stored values and protocol behavior still follow G
`Pilot`, `Client.Driver`, protocol JSON helpers, and explicitly marked companion
methods are not stable application extension APIs.

Options copy collections and metadata on construction; `InsertOptions.metadata()`
and `JobQuery.metadata()` also return copies. Job/result JSON values are snapshots,
Options copy collections and metadata on construction; `InsertOptions.metadata()`,
`JobQuery.metadata()`, and `JobUpdateOptions` JSON accessors also return copies.
Job/result JSON values are snapshots,
not database-backed objects. Builders are mutable and should be confined to
configuration code.

Expand Down
6 changes: 6 additions & 0 deletions java/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

## [Unreleased]

### Added

- **Breaking:** Added `client.jobs()` for job reads, cancellation, completion, deletion, retries, and updates, removing the corresponding flat `Client` methods. Insertions remain on `Client`. [PR #1569](https://github.com/riverqueue/river/pull/1569).
- **Breaking:** Replaced `Client.output` with `client.jobs().update` and immutable `JobUpdateOptions` for immediate output writes and top-level metadata merges, with a connection-first overload for caller-owned transactions. [PR #1569](https://github.com/riverqueue/river/pull/1569).

### Fixed

- Enforce the 32 MiB encoded UTF-8 JSON limit for explicit output in `JobUpdateOptions` and `WorkContext.output`, preserving previous output when validation fails. [PR #1569](https://github.com/riverqueue/river/pull/1569).
- Fixed a cancelled job that was retried being cancelled again as soon as its new attempt started. Retrying a job now clears its earlier attempt's cancellation request, like River for Go. [PR #1564](https://github.com/riverqueue/river/pull/1564).

## [0.2.0] - 2026-10-26
Expand Down
6 changes: 6 additions & 0 deletions java/DIFFERENCES.md
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,12 @@
stuck threshold. The JVM has no safe way to kill an arbitrary uncooperative
thread. Such a handler retains its slot until it returns; handlers should use
interruptible I/O or inspect their `WorkContext` cancellation signal.
- Job operations live on `client.jobs()` as in Rust and JavaScript, instead of using
Go's `Job` method prefix. `insert` and `insertMany` remain on `Client`.
- `client.jobs().update` takes immutable `JobUpdateOptions` and merges top-level metadata
as in TypeScript and Rust, in addition to setting output. Empty options leave
the job unchanged; there is no implicit copying of buffered worker output as
with Go's `JobUpdate`. Builder `output(null)` explicitly stores JSON null.
- Transaction variants use connection-first overloads instead of `Tx` suffixes.
Storage failures are unchecked `RiverException` values with a stable code and
original cause; JDBC callbacks may throw checked exceptions.
Expand Down
53 changes: 44 additions & 9 deletions java/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -228,11 +228,17 @@ var mixed = client.insertMany(List.of(

## Reading typed jobs

`client.get(id)` returns `Job<JsonNode>` because an ID does not identify a Java
Job operations live on `client.jobs()`, alongside queue operations on
`client.queues()`. Use `jobs().get`, `list`, `cancel`, `delete`, `deleteMany`,
`retry`, `complete`, and `update`. Insertions remain `client.insert` and
`client.insertMany`. Connection-first overloads use an existing transaction;
`jobs().complete` always requires one.

`client.jobs().get(id)` returns `Job<JsonNode>` because an ID does not identify a Java
argument class. Supply a job type when it is known:

```java
Job<SendEmail> job = client.get(jobId, email);
Job<SendEmail> job = client.jobs().get(jobId, email);
System.out.println(job.args().address());
```

Expand All @@ -248,9 +254,9 @@ unchanged when continuing from a page's cursor:

```java
var query = JobQuery.builder().kinds("send_email").limit(50).build();
var first = client.list(query);
var first = client.jobs().list(query);
if (!first.jobs().isEmpty()) {
var next = client.list(query.after(first.cursor()));
var next = client.jobs().list(query.after(first.cursor()));
}
```

Expand All @@ -262,9 +268,9 @@ selected state, or `SCHEDULED_AT` or `FINALIZED_AT` for an explicit timestamp.
A thrown exception triggers [retries](https://riverqueue.com/docs/job-retries)
until `maxAttempts` is exhausted, when the job becomes discarded. The default
policy uses River's quartic backoff with jitter. Configure a different policy
on the client builder, or explicitly retry an existing job with `client.retry(id)`.
on the client builder, or explicitly retry an existing job with `client.jobs().retry(id)`.
An explicit retry that updates a job notifies its queue after commit. Use
`client.retry(connection, id)` to make the retry part of an existing transaction.
`client.jobs().retry(connection, id)` to make the retry part of an existing transaction.
A retry policy receives the full job snapshot before the current failure is
appended to `errors`; use `job.errors().size() + 1` to include that failure.
It runs only when another attempt is possible. Return a nonnegative `Duration`;
Expand Down Expand Up @@ -306,12 +312,12 @@ var clientBuilder = Client.builder(database).queue("default", 10);
### Cancelling jobs

[Cancel](https://riverqueue.com/docs/cancelling-jobs) an enqueued or running job
with `client.cancel(id)`. Running attempts receive a cooperative cancellation
with `client.jobs().cancel(id)`. Running attempts receive a cooperative cancellation
signal. A handler can cancel itself permanently with `context.cancel(reason)`;
this ends the attempt without further retries.

```java
client.cancel(jobId);
client.jobs().cancel(jobId);

clientBuilder.worker(email, context -> {
context.checkCancelled();
Expand Down Expand Up @@ -610,9 +616,38 @@ clientBuilder.worker(email, context -> {
context.output(Map.of("delivered", true, "address", context.args().address()));
});

var output = client.get(jobId).metadata().path("output");
var output = client.jobs().get(jobId).metadata().path("output");
```

To persist output immediately, use `client.jobs().update` with `JobUpdateOptions`:

```java
var updates = JobUpdateOptions.builder()
.metadata(Map.of("provider_request_id", requestId))
.output(Map.of("normalized_text", normalized))
.build();
var updated = client.jobs().update(jobId, updates);

// Or include the update in an existing JDBC transaction:
client.jobs().update(connection, jobId, updates);
```

Metadata merges at the top level, preserving unrelated keys. Supplied nested
objects replace whole values; JSON null is preserved. Explicit `output` overrides
any `"output"` in the metadata updates. Omit it to leave output unchanged, or call
`.output(null)` to store JSON null. `JobUpdateOptions.defaults()` changes nothing.
Missing jobs throw `RiverException` with code `NOT_FOUND`.

Explicit output in `JobUpdateOptions` and `context.output` is limited to 32 MiB of
encoded UTF-8 JSON, including JSON escaping. Larger values throw
`IllegalArgumentException` without changing previously recorded output.

The connection overload requires auto-commit disabled and leaves commit/rollback
to the caller. Save step results before marking a resumable step complete.
`context.output` remains buffered until the attempt finishes and may overwrite
an immediate output update. Read fresh output with `client.jobs().get(jobId)`; existing
job objects are snapshots.

### Resumable jobs

[Named steps](https://riverqueue.com/docs/resumable-jobs) let retries skip work
Expand Down
14 changes: 7 additions & 7 deletions java/conformance/feature-inventory.json
Original file line number Diff line number Diff line change
Expand Up @@ -2298,7 +2298,7 @@
"bulk_delete_safety"
],
"source": "delete_many_params.go:river.JobDeleteManyParams.First",
"java_surface": "JobQuery and Client.deleteMany"
"java_surface": "JobQuery and Jobs.deleteMany"
},
{
"applicability": "api_equivalent",
Expand All @@ -2310,7 +2310,7 @@
"bulk_delete_safety"
],
"source": "delete_many_params.go:river.JobDeleteManyParams.IDs",
"java_surface": "JobQuery and Client.deleteMany"
"java_surface": "JobQuery and Jobs.deleteMany"
},
{
"applicability": "api_equivalent",
Expand All @@ -2322,7 +2322,7 @@
"bulk_delete_safety"
],
"source": "delete_many_params.go:river.JobDeleteManyParams.Kinds",
"java_surface": "JobQuery and Client.deleteMany"
"java_surface": "JobQuery and Jobs.deleteMany"
},
{
"applicability": "api_equivalent",
Expand All @@ -2331,7 +2331,7 @@
"id": "job_delete_many_params.Priorities",
"rationale": "Portable bulk-delete filter by priority; not yet accepted by the adapter's delete_many method.",
"source": "delete_many_params.go:river.JobDeleteManyParams.Priorities",
"java_surface": "JobQuery and Client.deleteMany"
"java_surface": "JobQuery and Jobs.deleteMany"
},
{
"applicability": "api_equivalent",
Expand All @@ -2343,7 +2343,7 @@
"bulk_delete_safety"
],
"source": "delete_many_params.go:river.JobDeleteManyParams.Queues",
"java_surface": "JobQuery and Client.deleteMany"
"java_surface": "JobQuery and Jobs.deleteMany"
},
{
"applicability": "api_equivalent",
Expand All @@ -2355,7 +2355,7 @@
"bulk_delete_safety"
],
"source": "delete_many_params.go:river.JobDeleteManyParams.States",
"java_surface": "JobQuery and Client.deleteMany"
"java_surface": "JobQuery and Jobs.deleteMany"
},
{
"applicability": "api_equivalent",
Expand All @@ -2367,7 +2367,7 @@
"bulk_delete_safety"
],
"source": "delete_many_params.go:river.JobDeleteManyParams.UnsafeAll",
"java_surface": "JobQuery and Client.deleteMany"
"java_surface": "JobQuery and Jobs.deleteMany"
},
{
"applicability": "api_equivalent",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import com.riverqueue.Job;
import com.riverqueue.JobQuery;
import com.riverqueue.JobType;
import com.riverqueue.JobUpdateOptions;
import com.riverqueue.Json;
import com.riverqueue.Migrator;
import com.riverqueue.RetryPolicy;
Expand Down Expand Up @@ -269,29 +270,26 @@ private JsonNode handle(String method, JsonNode params, String rawParams) throws
river = runtime.instrument(new Client(database));
yield Json.object();
}
case "get" -> normalized(scoped(params).get(params.path("id").asLong()));
case "tx_get" -> normalized(river.get(transaction(params), params.path("id").asLong()));
case "list" -> page(river.list(query(params)));
case "tx_list" -> page(river.list(transaction(params), query(params)));
case "cancel" -> normalized(river.cancel(params.path("id").asLong()));
case "tx_cancel" -> normalized(river.cancel(transaction(params), params.path("id").asLong()));
case "delete" -> normalized(river.delete(params.path("id").asLong()));
case "tx_delete" -> normalized(river.delete(transaction(params), params.path("id").asLong()));
case "retry" -> normalized(river.retry(params.path("id").asLong()));
case "tx_retry" -> normalized(river.retry(transaction(params), params.path("id").asLong()));
case "update" ->
normalized(
river.transaction(
c ->
params.has("output")
? river.output(c, params.path("id").asLong(), params.path("output"))
: river.get(c, params.path("id").asLong())));
case "get" -> normalized(scoped(params).jobs().get(params.path("id").asLong()));
case "tx_get" ->
normalized(river.jobs().get(transaction(params), params.path("id").asLong()));
case "list" -> page(river.jobs().list(query(params)));
case "tx_list" -> page(river.jobs().list(transaction(params), query(params)));
case "cancel" -> normalized(river.jobs().cancel(params.path("id").asLong()));
case "tx_cancel" ->
normalized(river.jobs().cancel(transaction(params), params.path("id").asLong()));
case "delete" -> normalized(river.jobs().delete(params.path("id").asLong()));
case "tx_delete" ->
normalized(river.jobs().delete(transaction(params), params.path("id").asLong()));
case "retry" -> normalized(river.jobs().retry(params.path("id").asLong()));
case "tx_retry" ->
normalized(river.jobs().retry(transaction(params), params.path("id").asLong()));
case "update" -> normalized(river.jobs().update(params.path("id").asLong(), updates(params)));
case "tx_update" ->
normalized(
params.has("output")
? river.output(
transaction(params), params.path("id").asLong(), params.path("output"))
: river.get(transaction(params), params.path("id").asLong()));
river
.jobs()
.update(transaction(params), params.path("id").asLong(), updates(params)));
case "delete_many" ->
Json.object()
.set(
Expand All @@ -300,8 +298,10 @@ private JsonNode handle(String method, JsonNode params, String rawParams) throws
river
.transaction(
c ->
river.deleteMany(
c, query(params), params.path("all").asBoolean(false)))
river
.jobs()
.deleteMany(
c, query(params), params.path("all").asBoolean(false)))
.stream()
.map(Adapter::normalized)
.toList()));
Expand All @@ -328,6 +328,7 @@ c, query(params), params.path("all").asBoolean(false)))
"jobs",
Json.tree(
river
.jobs()
.deleteMany(
transaction(params),
query(params),
Expand Down Expand Up @@ -594,6 +595,13 @@ private static Unique unique(JsonNode value, boolean nanos) {
value.path("exclude_kind").asBoolean(false));
}

private static JobUpdateOptions updates(JsonNode params) {
var builder = JobUpdateOptions.builder();
if (params.has("metadata")) builder.metadata(params.path("metadata"));
if (params.has("output")) builder.output(params.path("output"));
return builder.build();
}

private void validate(JsonNode schema, JsonNode value) {
if (schema.has("$ref")) {
validate(contract.at(schema.path("$ref").asString().substring(1)), value);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ INSERT INTO river_job (args,attempt,attempted_at,attempted_by,created_at,errors,
id = rows.getLong(1);
}
}
return Adapter.normalized(new Client(database).get(id));
return Adapter.normalized(new Client(database).jobs().get(id));
}
if (method.equals("leader")) {
try (var statement = connection.createStatement();
Expand Down Expand Up @@ -148,7 +148,7 @@ INSERT INTO river_job (args,attempt,attempted_at,attempted_by,created_at,errors,
id = rows.getLong(1);
}
}
return Adapter.normalized(new Client(database).get(id));
return Adapter.normalized(new Client(database).jobs().get(id));
}
if (method.equals("raw_finalize")) {
String error =
Expand Down Expand Up @@ -185,7 +185,7 @@ WHERE key NOT IN (SELECT key FROM updates,json_each(doc))
statement.setLong(4, params.path("id").asLong());
statement.executeUpdate();
}
return Adapter.normalized(new Client(database).get(params.path("id").asLong()));
return Adapter.normalized(new Client(database).jobs().get(params.path("id").asLong()));
}
if (method.equals("raw_replace_json_text")) {
String column = params.path("column").asString();
Expand Down
Loading
Loading