diff --git a/java/.gitignore b/java/.gitignore index b9752d435..f7807e706 100644 --- a/java/.gitignore +++ b/java/.gitignore @@ -1,5 +1,8 @@ target/ .idea/ *.iml +.classpath +.project +.settings/ .conformance/ __pycache__/ diff --git a/java/API.md b/java/API.md index 92a6ae2ce..3980a8ba2 100644 --- a/java/API.md +++ b/java/API.md @@ -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`, `Job`, `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` 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` 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` 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. | @@ -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. diff --git a/java/CHANGELOG.md b/java/CHANGELOG.md index 963e4c7c4..b466f8fa6 100644 --- a/java/CHANGELOG.md +++ b/java/CHANGELOG.md @@ -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 diff --git a/java/DIFFERENCES.md b/java/DIFFERENCES.md index 616285f59..123daf41c 100644 --- a/java/DIFFERENCES.md +++ b/java/DIFFERENCES.md @@ -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. diff --git a/java/README.md b/java/README.md index f0c3aa511..fb24542f4 100644 --- a/java/README.md +++ b/java/README.md @@ -228,11 +228,17 @@ var mixed = client.insertMany(List.of( ## Reading typed jobs -`client.get(id)` returns `Job` 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` because an ID does not identify a Java argument class. Supply a job type when it is known: ```java -Job job = client.get(jobId, email); +Job job = client.jobs().get(jobId, email); System.out.println(job.args().address()); ``` @@ -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())); } ``` @@ -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`; @@ -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(); @@ -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 diff --git a/java/conformance/feature-inventory.json b/java/conformance/feature-inventory.json index 4894575d1..2536cd5d9 100644 --- a/java/conformance/feature-inventory.json +++ b/java/conformance/feature-inventory.json @@ -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", @@ -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", @@ -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", @@ -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", @@ -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", @@ -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", @@ -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", diff --git a/java/conformance/src/main/java/com/riverqueue/conformance/Adapter.java b/java/conformance/src/main/java/com/riverqueue/conformance/Adapter.java index c0df8f421..aa4b2bcd5 100644 --- a/java/conformance/src/main/java/com/riverqueue/conformance/Adapter.java +++ b/java/conformance/src/main/java/com/riverqueue/conformance/Adapter.java @@ -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; @@ -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( @@ -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())); @@ -328,6 +328,7 @@ c, query(params), params.path("all").asBoolean(false))) "jobs", Json.tree( river + .jobs() .deleteMany( transaction(params), query(params), @@ -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); diff --git a/java/conformance/src/main/java/com/riverqueue/conformance/RawStorage.java b/java/conformance/src/main/java/com/riverqueue/conformance/RawStorage.java index 317b5f883..5381e739f 100644 --- a/java/conformance/src/main/java/com/riverqueue/conformance/RawStorage.java +++ b/java/conformance/src/main/java/com/riverqueue/conformance/RawStorage.java @@ -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(); @@ -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 = @@ -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(); diff --git a/java/conformance/src/main/java/com/riverqueue/conformance/RuntimeAdapter.java b/java/conformance/src/main/java/com/riverqueue/conformance/RuntimeAdapter.java index 2d18408b3..41f7515b4 100644 --- a/java/conformance/src/main/java/com/riverqueue/conformance/RuntimeAdapter.java +++ b/java/conformance/src/main/java/com/riverqueue/conformance/RuntimeAdapter.java @@ -72,7 +72,7 @@ JsonNode handle(Client river, String method, JsonNode params) throws Exception { case "wait" -> waitFor(river, params); case "work" -> { if (workers != null) throw new IllegalArgumentException("Client already running"); - var job = river.get(params.path("id").asLong()); + var job = river.jobs().get(params.path("id").asLong()); var start = Json.object() .put("client_id", params.path("client_id").asString("java-conformance-worker")) @@ -260,7 +260,7 @@ else if (params.has("job_timeout_ms")) private JsonNode waitFor(Client river, JsonNode params) throws Exception { long deadline = System.nanoTime() + Duration.ofSeconds(20).toNanos(); while (System.nanoTime() < deadline) { - var job = river.get(params.path("id").asLong()); + var job = river.jobs().get(params.path("id").asLong()); boolean match = params.has("states") ? false : job.state().isFinalized(); for (var state : params.path("states")) if (job.state().value().equals(state.asString())) match = true; diff --git a/java/river/src/main/java/com/riverqueue/Client.java b/java/river/src/main/java/com/riverqueue/Client.java index 752f5a9d8..f0fd4b309 100644 --- a/java/river/src/main/java/com/riverqueue/Client.java +++ b/java/river/src/main/java/com/riverqueue/Client.java @@ -116,6 +116,11 @@ public String id() { return id; } + /** Returns job operations. Insert jobs with this client's insert methods. */ + public Jobs jobs() { + return new Jobs(this, clock); + } + public Queues queues() { return new Queues(this); } @@ -635,54 +640,6 @@ List workerMiddleware() { return workerMiddleware; } - public Job cancel(long id) { - return transaction(c -> cancel(c, id)); - } - - public Job cancel(Connection connection, long id) { - return atomic( - connection, - c -> { - var before = locked(c, id); - if (before.state().isFinalized()) return before; - try (var statement = - Sql.prepare( - c, - Sql.query(database, "cancel"), - database.timestamp(clock.instant()), - Json.encode(clock.instant()), - id); - var rows = statement.executeQuery()) { - if (!rows.next()) return get(c, id); - var job = read(rows); - pilot.afterStateChange(c, driver(), job); - notify(c, Protocol.cancel(id, job.queue())); - return job; - } - }); - } - - public Job delete(long id) { - return transaction(c -> delete(c, id)); - } - - public Job delete(Connection connection, long id) { - return atomic( - connection, - c -> { - var job = locked(c, id); - if (job.state() == Job.State.RUNNING) - throw new RiverException(RiverException.Code.REJECTED, "Cannot delete a running job"); - try (var statement = Sql.prepare(c, Sql.query(database, "delete"), id); - var rows = statement.executeQuery()) { - rows.next(); - var deleted = read(rows); - pilot.afterDelete(c, driver(), deleted); - return deleted; - } - }); - } - private int deleteFinalized( Connection connection, Instant before, @@ -705,122 +662,6 @@ private int deleteFinalized( } } - /** Completes a running attempt in the caller's transaction, preserving its argument type. */ - public Job complete(Connection connection, Job job) { - return complete(connection, job, java.util.Map.of()); - } - - /** Completes a running attempt and merges metadata in the caller's transaction. */ - public Job complete(Connection connection, Job job, Object metadata) { - Objects.requireNonNull(job, "job"); - if (job.state() != Job.State.RUNNING) throw new IllegalArgumentException("Job must be running"); - var updates = metadata == null ? Json.object() : Json.tree(metadata); - if (!updates.isObject()) throw new IllegalArgumentException("Metadata must be a JSON object"); - return atomic( - connection, - c -> { - try (var statement = - Sql.prepare( - c, - Sql.query(database, "complete"), - "completed", - database.timestamp(clock.instant()), - null, - "{}", - Json.encode(updates), - "completed", - database.timestamp(clock.instant()), - database.timestamp(job.scheduledAt()), - job.attempt(), - null, - null, - job.id(), - job.attempt(), - database.timestamp(job.attemptedAt())); - var rows = statement.executeQuery()) { - if (!rows.next()) return get(c, job.id()).mapArgs(ignored -> job.args()); - var completed = read(rows); - pilot.afterStateChange(c, driver(), completed); - return completed.mapArgs(ignored -> job.args()); - } - }); - } - - /** - * Deletes a page of matching non-running jobs in query order. Running jobs do not count toward - * the limit. Concurrent changes that stop selected jobs matching the filters or cursor can reduce - * the returned batch size. Unfiltered deletion requires {@code all=true}. - */ - public List> deleteMany(JobQuery query, boolean all) { - return transaction(connection -> deleteMany(connection, query, all)); - } - - /** Deletes matching non-running jobs atomically within the caller's transaction. */ - public List> deleteMany(Connection connection, JobQuery query, boolean all) { - if (!all && !query.hasFilters()) - throw new IllegalArgumentException("Bulk deletion requires filters or explicit all=true"); - return atomic( - connection, - c -> { - // Reserve SQLite's writer before listing so another commit cannot invalidate the - // snapshot. - if (database.dialect() == Database.Dialect.SQLITE) - try (var statement = Sql.prepare(c, Sql.query(database, "writer_lock"))) { - statement.executeUpdate(); - } - var result = new ArrayList>(); - var jobs = list(c, query, true).jobs(); - var params = new ArrayList(); - String sql = query.deleteSql(database, params); - try (var statement = Sql.prepare(c, sql, params.toArray())) { - for (var job : jobs) { - statement.setLong(params.size() + 1, job.id()); - try (var rows = statement.executeQuery()) { - // Recheck every filter atomically: a peer may have updated this job since listing. - if (!rows.next()) continue; - var deleted = read(rows); - pilot.afterDelete(c, driver(), deleted); - result.add(deleted); - } - } - } - return List.copyOf(result); - }); - } - - public Job get(long id) { - return transaction(c -> get(c, id)); - } - - /** Loads and decodes a job, rejecting a kind that does not match the supplied type. */ - public Job get(long id, JobType type) { - return transaction(connection -> get(connection, id, type)); - } - - /** Loads and decodes a job using the caller's connection. */ - public Job get(Connection connection, long id, JobType type) { - var job = get(connection, id); - if (!job.kind().equals(type.kind())) - throw new IllegalArgumentException( - "Expected job kind " + type.kind() + ", got " + job.kind()); - return typed(job, type); - } - - public Job get(Connection connection, long id) { - return atomic( - connection, - transaction -> { - try (var statement = Sql.prepare(transaction, Sql.query(database, "get"), id); - var rows = statement.executeQuery()) { - if (!rows.next()) - throw new RiverException(RiverException.Code.NOT_FOUND, "Job " + id + " not found"); - return read(rows); - } catch (SQLException e) { - throw databaseError("Get job", e); - } - }); - } - public Job.InsertResult insert(JobType type, A args) { return insert(type, args, InsertOptions.defaults()); } @@ -1077,45 +918,6 @@ private Job.InsertResult insertOne( } } - public JobQuery.Page list(JobQuery query) { - return transaction(c -> list(c, query)); - } - - public JobQuery.Page list(Connection connection, JobQuery query) { - return list(connection, query, false); - } - - private JobQuery.Page list(Connection connection, JobQuery query, boolean forDelete) { - return atomic( - connection, - transaction -> { - var params = new ArrayList(); - String sql = query.sql(database, params, forDelete); - try (var statement = Sql.prepare(transaction, sql, params.toArray()); - var rows = statement.executeQuery()) { - var jobs = new ArrayList>(); - while (rows.next()) jobs.add(read(rows)); - return new JobQuery.Page( - jobs.isEmpty() ? null : query.cursor(jobs.getLast()), List.copyOf(jobs)); - } catch (SQLException e) { - throw databaseError("List jobs", e); - } - }); - } - - private Job locked(Connection connection, long id) throws SQLException { - if (database.dialect() == Database.Dialect.SQLITE) - try (var statement = Sql.prepare(connection, Sql.query(database, "writer_lock"))) { - statement.executeUpdate(); - } - try (var statement = Sql.prepare(connection, Sql.query(database, "lock_get"), id); - var rows = statement.executeQuery()) { - if (!rows.next()) - throw new RiverException(RiverException.Code.NOT_FOUND, "Job " + id + " not found"); - return read(rows); - } - } - /** Requests that the current leader resign after this transaction commits. */ public void requestResign(Connection connection) { atomic( @@ -1153,50 +955,6 @@ void notifyInsert(Connection connection, String queue) throws SQLException { notify(connection, Protocol.insert(queue)); } - public Job output(long id, Object output) { - return transaction(connection -> output(connection, id, output)); - } - - public Job output(Connection connection, long id, Object output) { - return atomic( - connection, - c -> { - try (var statement = - Sql.prepare(c, Sql.query(database, "output"), Json.encode(output), id); - var rows = statement.executeQuery()) { - if (!rows.next()) - throw new RiverException(RiverException.Code.NOT_FOUND, "Job " + id + " not found"); - return read(rows); - } - }); - } - - public Job retry(long id) { - return transaction(c -> retry(c, id)); - } - - public Job retry(Connection connection, long id) { - return atomic( - connection, - c -> { - var before = locked(c, id); - try (var statement = - Sql.prepare( - c, - Sql.query(database, "retry"), - database.timestamp(clock.instant()), - id, - database.timestamp(clock.instant())); - var rows = statement.executeQuery()) { - boolean updated = rows.next(); - var job = updated ? read(rows) : before; - pilot.afterStateChange(c, driver(), job); - if (updated) notifyInsert(c, job.queue()); - return job; - } - }); - } - private void reset() { transaction( c -> { diff --git a/java/river/src/main/java/com/riverqueue/JobUpdateOptions.java b/java/river/src/main/java/com/riverqueue/JobUpdateOptions.java new file mode 100644 index 000000000..7350f6250 --- /dev/null +++ b/java/river/src/main/java/com/riverqueue/JobUpdateOptions.java @@ -0,0 +1,73 @@ +package com.riverqueue; + +import tools.jackson.databind.JsonNode; +import tools.jackson.databind.node.ObjectNode; + +/** + * Immutable updates to job metadata and recorded output. Metadata is merged at the top level; + * supplied values replace whole values, including nested objects and JSON null. Explicit output + * takes precedence over the metadata's "output" key and must encode to at most 32 MiB of UTF-8 + * JSON; larger values throw {@link IllegalArgumentException} during construction. + * + * @param metadata metadata keys to merge; null means no metadata updates + * @param output output to set; a null reference leaves it unchanged, a JSON null node stores null + */ +public record JobUpdateOptions(JsonNode metadata, JsonNode output) { + public JobUpdateOptions { + metadata = metadata == null ? Json.object() : metadata.deepCopy(); + if (!metadata.isObject()) throw new IllegalArgumentException("metadata must be a JSON object"); + output = output == null ? null : output.deepCopy(); + if (output != null) RecordedOutput.checkSize(output); + } + + /** Returns a builder for named updates. */ + public static Builder builder() { + return new Builder(); + } + + /** Returns options that leave the job unchanged. */ + public static JobUpdateOptions defaults() { + return builder().build(); + } + + /** Returns a copy of the metadata updates. */ + @Override + public JsonNode metadata() { + return metadata.deepCopy(); + } + + ObjectNode metadataUpdates() { + var updates = (ObjectNode) metadata.deepCopy(); + if (output != null) updates.set(Protocol.METADATA_OUTPUT, output.deepCopy()); + return updates; + } + + /** Returns a copy of the output, or null when output was not supplied. */ + @Override + public JsonNode output() { + return output == null ? null : output.deepCopy(); + } + + /** Fluent construction distinguishes omitted output from {@code output(null)}. */ + public static final class Builder { + private JsonNode metadata; + private JsonNode output; + + /** Creates immutable options from the supplied updates. */ + public JobUpdateOptions build() { + return new JobUpdateOptions(metadata, output); + } + + /** Merges the supplied JSON object into existing metadata at the top level. */ + public Builder metadata(Object value) { + metadata = Json.tree(value).deepCopy(); + return this; + } + + /** Replaces recorded output; passing null stores JSON null. */ + public Builder output(Object value) { + output = Json.tree(value).deepCopy(); + return this; + } + } +} diff --git a/java/river/src/main/java/com/riverqueue/Jobs.java b/java/river/src/main/java/com/riverqueue/Jobs.java new file mode 100644 index 000000000..506af84ff --- /dev/null +++ b/java/river/src/main/java/com/riverqueue/Jobs.java @@ -0,0 +1,294 @@ +package com.riverqueue; + +import java.sql.Connection; +import java.sql.SQLException; +import java.time.Clock; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import tools.jackson.databind.JsonNode; + +/** + * Job operations obtained from {@link Client#jobs()}. Connection-first overloads use the caller's + * transaction and never commit or close it; other overloads own a transaction. Insert jobs through + * {@link Client#insert(JobType, Object)} or {@link Client#insertMany(List)}. + */ +public final class Jobs { + private final Clock clock; + private final Client river; + + Jobs(Client river, Clock clock) { + this.clock = clock; + this.river = river; + } + + /** Cancels a job, signalling a running attempt to stop. */ + public Job cancel(long id) { + return river.transaction(c -> cancel(c, id)); + } + + /** Cancels a job, signalling a running attempt to stop. Uses the caller's transaction. */ + public Job cancel(Connection connection, long id) { + return river.transaction( + connection, + c -> { + var before = locked(c, id); + if (before.state().isFinalized()) return before; + try (var statement = + Sql.prepare( + c, + Sql.query(river.database(), "cancel"), + river.database().timestamp(clock.instant()), + Json.encode(clock.instant()), + id); + var rows = statement.executeQuery()) { + if (!rows.next()) return get(c, id); + var job = river.read(rows); + river.pilot().afterStateChange(c, river.driver(), job); + river.notify(c, Protocol.cancel(id, job.queue())); + return job; + } + }); + } + + /** Completes a running attempt in the caller's transaction, preserving its argument type. */ + public Job complete(Connection connection, Job job) { + return complete(connection, job, java.util.Map.of()); + } + + /** Completes a running attempt and merges metadata in the caller's transaction. */ + public Job complete(Connection connection, Job job, Object metadata) { + Objects.requireNonNull(job, "job"); + if (job.state() != Job.State.RUNNING) throw new IllegalArgumentException("Job must be running"); + var updates = metadata == null ? Json.object() : Json.tree(metadata); + if (!updates.isObject()) throw new IllegalArgumentException("Metadata must be a JSON object"); + return river.transaction( + connection, + c -> { + try (var statement = + Sql.prepare( + c, + Sql.query(river.database(), "complete"), + "completed", + river.database().timestamp(clock.instant()), + null, + "{}", + Json.encode(updates), + "completed", + river.database().timestamp(clock.instant()), + river.database().timestamp(job.scheduledAt()), + job.attempt(), + null, + null, + job.id(), + job.attempt(), + river.database().timestamp(job.attemptedAt())); + var rows = statement.executeQuery()) { + if (!rows.next()) return get(c, job.id()).mapArgs(ignored -> job.args()); + var completed = river.read(rows); + river.pilot().afterStateChange(c, river.driver(), completed); + return completed.mapArgs(ignored -> job.args()); + } + }); + } + + /** Deletes a job, rejecting running jobs. */ + public Job delete(long id) { + return river.transaction(c -> delete(c, id)); + } + + /** Deletes a job, rejecting running jobs. Uses the caller's transaction. */ + public Job delete(Connection connection, long id) { + return river.transaction( + connection, + c -> { + var job = locked(c, id); + if (job.state() == Job.State.RUNNING) + throw new RiverException(RiverException.Code.REJECTED, "Cannot delete a running job"); + try (var statement = Sql.prepare(c, Sql.query(river.database(), "delete"), id); + var rows = statement.executeQuery()) { + rows.next(); + var deleted = river.read(rows); + river.pilot().afterDelete(c, river.driver(), deleted); + return deleted; + } + }); + } + + /** + * Deletes a page of matching non-running jobs in query order. Running jobs do not count toward + * the limit. Concurrent changes that stop selected jobs matching the filters or cursor can reduce + * the returned batch size. Unfiltered deletion requires {@code all=true}. + */ + public List> deleteMany(JobQuery query, boolean all) { + return river.transaction(connection -> deleteMany(connection, query, all)); + } + + /** Deletes matching non-running jobs atomically within the caller's transaction. */ + public List> deleteMany(Connection connection, JobQuery query, boolean all) { + if (!all && !query.hasFilters()) + throw new IllegalArgumentException("Bulk deletion requires filters or explicit all=true"); + return river.transaction( + connection, + c -> { + // Reserve SQLite's writer before listing so another commit cannot invalidate the + // snapshot. + if (river.database().dialect() == Database.Dialect.SQLITE) + try (var statement = Sql.prepare(c, Sql.query(river.database(), "writer_lock"))) { + statement.executeUpdate(); + } + var result = new ArrayList>(); + var jobs = list(c, query, true).jobs(); + var params = new ArrayList(); + String sql = query.deleteSql(river.database(), params); + try (var statement = Sql.prepare(c, sql, params.toArray())) { + for (var job : jobs) { + statement.setLong(params.size() + 1, job.id()); + try (var rows = statement.executeQuery()) { + // Recheck every filter atomically: a peer may have updated this job since listing. + if (!rows.next()) continue; + var deleted = river.read(rows); + river.pilot().afterDelete(c, river.driver(), deleted); + result.add(deleted); + } + } + } + return List.copyOf(result); + }); + } + + /** Loads a job with its arguments as JSON. */ + public Job get(long id) { + return river.transaction(c -> get(c, id)); + } + + /** Loads and decodes a job, rejecting a kind that does not match the supplied type. */ + public Job get(long id, JobType type) { + return river.transaction(connection -> get(connection, id, type)); + } + + /** Loads and decodes a job using the caller's connection. */ + public Job get(Connection connection, long id, JobType type) { + var job = get(connection, id); + if (!job.kind().equals(type.kind())) + throw new IllegalArgumentException( + "Expected job kind " + type.kind() + ", got " + job.kind()); + return river.typed(job, type); + } + + /** Loads a job with its arguments as JSON. Uses the caller's transaction. */ + public Job get(Connection connection, long id) { + return river.transaction( + connection, + transaction -> { + try (var statement = Sql.prepare(transaction, Sql.query(river.database(), "get"), id); + var rows = statement.executeQuery()) { + if (!rows.next()) + throw new RiverException(RiverException.Code.NOT_FOUND, "Job " + id + " not found"); + return river.read(rows); + } catch (SQLException e) { + throw Client.databaseError("Get job", e); + } + }); + } + + /** Lists a page of jobs matching the query. */ + public JobQuery.Page list(JobQuery query) { + return river.transaction(c -> list(c, query)); + } + + /** Lists a page of jobs matching the query. Uses the caller's transaction. */ + public JobQuery.Page list(Connection connection, JobQuery query) { + return list(connection, query, false); + } + + private JobQuery.Page list(Connection connection, JobQuery query, boolean forDelete) { + return river.transaction( + connection, + transaction -> { + var params = new ArrayList(); + String sql = query.sql(river.database(), params, forDelete); + try (var statement = Sql.prepare(transaction, sql, params.toArray()); + var rows = statement.executeQuery()) { + var jobs = new ArrayList>(); + while (rows.next()) jobs.add(river.read(rows)); + return new JobQuery.Page( + jobs.isEmpty() ? null : query.cursor(jobs.getLast()), List.copyOf(jobs)); + } catch (SQLException e) { + throw Client.databaseError("List jobs", e); + } + }); + } + + private Job locked(Connection connection, long id) throws SQLException { + if (river.database().dialect() == Database.Dialect.SQLITE) + try (var statement = Sql.prepare(connection, Sql.query(river.database(), "writer_lock"))) { + statement.executeUpdate(); + } + try (var statement = Sql.prepare(connection, Sql.query(river.database(), "lock_get"), id); + var rows = statement.executeQuery()) { + if (!rows.next()) + throw new RiverException(RiverException.Code.NOT_FOUND, "Job " + id + " not found"); + return river.read(rows); + } + } + + /** Makes a job available for another attempt. */ + public Job retry(long id) { + return river.transaction(c -> retry(c, id)); + } + + /** Makes a job available for another attempt. Uses the caller's transaction. */ + public Job retry(Connection connection, long id) { + return river.transaction( + connection, + c -> { + var before = locked(c, id); + try (var statement = + Sql.prepare( + c, + Sql.query(river.database(), "retry"), + river.database().timestamp(clock.instant()), + id, + river.database().timestamp(clock.instant())); + var rows = statement.executeQuery()) { + boolean updated = rows.next(); + var job = updated ? river.read(rows) : before; + river.pilot().afterStateChange(c, river.driver(), job); + if (updated) river.notifyInsert(c, job.queue()); + return job; + } + }); + } + + /** + * Merges job metadata and sets recorded output immediately, returning the updated job. Unlike + * {@link WorkContext#output(Object)}, this writes before the attempt finishes. Other metadata + * keys are preserved. Throws {@link RiverException} with code {@code NOT_FOUND} for a missing + * job. + */ + public Job update(long id, JobUpdateOptions options) { + return river.transaction(connection -> update(connection, id, options)); + } + + /** + * Applies job updates in a caller-owned transaction using a savepoint. The connection must have + * auto-commit disabled; the caller owns commit or rollback. Empty options leave the job + * unchanged. Throws {@link RiverException} with code {@code NOT_FOUND} for a missing job. + */ + public Job update(Connection connection, long id, JobUpdateOptions options) { + var updates = Objects.requireNonNull(options, "options").metadataUpdates(); + if (updates.isEmpty()) return get(connection, id); + return river.transaction( + connection, + c -> { + try (var statement = + Sql.prepare(c, Sql.query(river.database(), "update"), Json.encode(updates), id); + var rows = statement.executeQuery()) { + if (!rows.next()) + throw new RiverException(RiverException.Code.NOT_FOUND, "Job " + id + " not found"); + return river.read(rows); + } + }); + } +} diff --git a/java/river/src/main/java/com/riverqueue/RecordedOutput.java b/java/river/src/main/java/com/riverqueue/RecordedOutput.java new file mode 100644 index 000000000..4cad67ceb --- /dev/null +++ b/java/river/src/main/java/com/riverqueue/RecordedOutput.java @@ -0,0 +1,17 @@ +package com.riverqueue; + +import java.nio.charset.StandardCharsets; +import tools.jackson.databind.JsonNode; + +/** Size validation shared by immediate and buffered job output. */ +final class RecordedOutput { + private static final int MAX_BYTES = 32 * 1024 * 1024; + + private RecordedOutput() {} + + static void checkSize(JsonNode value) { + int size = Json.encode(value).getBytes(StandardCharsets.UTF_8).length; + if (size > MAX_BYTES) + throw new IllegalArgumentException("output is too large: " + size + " bytes (max 32 MiB)"); + } +} diff --git a/java/river/src/main/java/com/riverqueue/RunHandle.java b/java/river/src/main/java/com/riverqueue/RunHandle.java index 8346562da..83fb38d8f 100644 --- a/java/river/src/main/java/com/riverqueue/RunHandle.java +++ b/java/river/src/main/java/com/riverqueue/RunHandle.java @@ -723,7 +723,7 @@ private void completeBatches() { if (result.next()) rows.add(river.readPartial(result).job()); else { try { - rows.add(river.get(c, completion.original.id())); + rows.add(river.jobs().get(c, completion.original.id())); } catch (RiverException error) { if (error.code() != RiverException.Code.NOT_FOUND) throw error; rows.add(null); diff --git a/java/river/src/main/java/com/riverqueue/WorkContext.java b/java/river/src/main/java/com/riverqueue/WorkContext.java index 03c4bc95b..edf0d0706 100644 --- a/java/river/src/main/java/com/riverqueue/WorkContext.java +++ b/java/river/src/main/java/com/riverqueue/WorkContext.java @@ -169,7 +169,7 @@ String checkpointStep() { /** Completes this attempt atomically with application writes in the supplied transaction. */ public Job complete(Connection transaction) { - return client.complete(transaction, job, metadataUpdates()); + return client.jobs().complete(transaction, job, metadataUpdates()); } public void discard(String reason) { @@ -218,8 +218,14 @@ synchronized Exception finish(Exception failure) { return failure; } - public void output(Object value) { - metadata(Protocol.METADATA_OUTPUT, value); + /** + * Buffers output until the attempt finishes. Throws {@link IllegalArgumentException} if its + * encoded UTF-8 JSON exceeds 32 MiB, preserving any previously buffered output. + */ + public synchronized void output(Object value) { + var output = Json.tree(value).deepCopy(); + RecordedOutput.checkSize(output); + metadata.set(Protocol.METADATA_OUTPUT, output); } private Job persistCheckpoint(Connection transaction, ObjectNode updates) { diff --git a/java/river/src/main/resources/com/riverqueue/postgres.sql b/java/river/src/main/resources/com/riverqueue/postgres.sql index 517578057..5dc77a328 100644 --- a/java/river/src/main/resources/com/riverqueue/postgres.sql +++ b/java/river/src/main/resources/com/riverqueue/postgres.sql @@ -83,8 +83,8 @@ UPDATE {schema}river_job SET state=CASE WHEN state='running' THEN state ELSE 'ca finalized_at=CASE WHEN state='running' THEN finalized_at ELSE ? END, metadata=jsonb_set(metadata,'{cancel_attempted_at}',?::jsonb,true) WHERE id=? AND state NOT IN ('cancelled','completed','discarded') AND finalized_at IS NULL RETURNING {columns} --- name: output -UPDATE {schema}river_job SET metadata=jsonb_set(metadata,'{output}',?::jsonb,true) WHERE id=? RETURNING {columns} +-- name: update +UPDATE {schema}river_job SET metadata=metadata||?::jsonb WHERE id=? RETURNING {columns} -- name: filter_metadata metadata @> ?::jsonb -- name: filter_tags_all diff --git a/java/river/src/main/resources/com/riverqueue/sqlite.sql b/java/river/src/main/resources/com/riverqueue/sqlite.sql index dc140d684..c499bf7fe 100644 --- a/java/river/src/main/resources/com/riverqueue/sqlite.sql +++ b/java/river/src/main/resources/com/riverqueue/sqlite.sql @@ -86,8 +86,17 @@ UPDATE river_job SET state=CASE WHEN state='running' THEN state ELSE 'cancelled' finalized_at=CASE WHEN state='running' THEN finalized_at ELSE ? END, metadata=jsonb_set(metadata,'$.cancel_attempted_at',jsonb(?)) WHERE id=? AND state NOT IN ('cancelled','completed','discarded') AND finalized_at IS NULL RETURNING {columns} --- name: output -UPDATE river_job SET metadata=jsonb_set(metadata,'$.output',jsonb(?)) WHERE id=? RETURNING {columns} +-- name: update +WITH updates(doc) AS (SELECT jsonb(?)) +UPDATE river_job SET metadata=CASE WHEN json_type(metadata)='object' THEN ( + -- Replace whole top-level values, preserving nulls and exact numeric tokens. + SELECT jsonb_group_object(key,jsonb(doc->fullkey)) FROM ( + SELECT key,fullkey,metadata AS doc FROM json_each(metadata) + WHERE key NOT IN (SELECT key FROM updates,json_each(doc)) + UNION ALL SELECT key,fullkey,doc FROM updates,json_each(doc) + ) +) ELSE metadata END +WHERE id=? RETURNING {columns} -- name: filter_metadata_container ({type}='{kind}' AND {children}) -- name: filter_metadata_element diff --git a/java/river/src/test/java/com/riverqueue/BulkDeleteTest.java b/java/river/src/test/java/com/riverqueue/BulkDeleteTest.java index f3300790a..791f335da 100644 --- a/java/river/src/test/java/com/riverqueue/BulkDeleteTest.java +++ b/java/river/src/test/java/com/riverqueue/BulkDeleteTest.java @@ -78,9 +78,10 @@ void acceptsEachFilter(String filter) { } assertEquals( List.of(matching.id()), - client.deleteMany(query.build(), false).stream().map(Job::id).toList()); + client.jobs().deleteMany(query.build(), false).stream().map(Job::id).toList()); assertEquals( - List.of(other.id()), client.list(JobQuery.all()).jobs().stream().map(Job::id).toList()); + List.of(other.id()), + client.jobs().list(JobQuery.all()).jobs().stream().map(Job::id).toList()); } @Test @@ -90,10 +91,14 @@ void callerRollbackRestoresDeletedJobs() throws Exception { try (var connection = client.database().connection()) { connection.setAutoCommit(false); assertEquals( - 1, client.deleteMany(connection, JobQuery.builder().ids(job.id()).build(), false).size()); - assertTrue(client.list(connection, JobQuery.all()).jobs().isEmpty()); + 1, + client + .jobs() + .deleteMany(connection, JobQuery.builder().ids(job.id()).build(), false) + .size()); + assertTrue(client.jobs().list(connection, JobQuery.all()).jobs().isEmpty()); connection.rollback(); - assertEquals(job.id(), client.get(job.id()).id()); + assertEquals(job.id(), client.jobs().get(job.id()).id()); } } @@ -116,13 +121,15 @@ public void afterDelete( assertThrows( IllegalStateException.class, () -> - client.deleteMany(connection, JobQuery.builder().kinds(TYPE.kind()).build(), false)); + client + .jobs() + .deleteMany(connection, JobQuery.builder().kinds(TYPE.kind()).build(), false)); assertEquals( List.of(first.id(), second.id()), - client.list(connection, JobQuery.all()).jobs().stream().map(Job::id).toList()); + client.jobs().list(connection, JobQuery.all()).jobs().stream().map(Job::id).toList()); connection.commit(); } - assertEquals(2, base.list(JobQuery.all()).jobs().size()); + assertEquals(2, base.jobs().list(JobQuery.all()).jobs().size()); } @ParameterizedTest @@ -145,7 +152,7 @@ public void afterDelete( if (job.id() != first.id()) return; // The batch has listed all three rows. Commit a peer's change before it reaches the // second. - if (change.equals("delete")) base.delete(raced.id()); + if (change.equals("delete")) base.jobs().delete(raced.id()); else base.transaction( peer -> { @@ -165,12 +172,13 @@ public void afterDelete( }); assertEquals( List.of(first.id(), last.id()), - client.deleteMany(JobQuery.builder().kinds(TYPE.kind()).build(), false).stream() + client.jobs().deleteMany(JobQuery.builder().kinds(TYPE.kind()).build(), false).stream() .map(Job::id) .toList()); assertEquals(List.of(first.id(), last.id()), deleted); - if (change.equals("claim")) assertEquals(Job.State.RUNNING, base.get(raced.id()).state()); - else assertTrue(base.list(JobQuery.all()).jobs().isEmpty()); + if (change.equals("claim")) + assertEquals(Job.State.RUNNING, base.jobs().get(raced.id()).state()); + else assertTrue(base.jobs().list(JobQuery.all()).jobs().isEmpty()); } @ParameterizedTest @@ -203,9 +211,9 @@ void preservesJobsThatNoLongerMatchFilters(String change) { "boundary", InsertOptions.builder().scheduledAt(Instant.parse("2026-01-01T00:00:00Z")).build()) .job(); - var first = base.cancel(base.insert(TYPE, "first", options).job().id()); - var raced = base.cancel(base.insert(TYPE, "raced", options).job().id()); - var last = base.cancel(base.insert(TYPE, "last", options).job().id()); + var first = base.jobs().cancel(base.insert(TYPE, "first", options).job().id()); + var raced = base.jobs().cancel(base.insert(TYPE, "raced", options).job().id()); + var last = base.jobs().cancel(base.insert(TYPE, "last", options).job().id()); var query = JobQuery.builder() .kinds(TYPE.kind()) @@ -227,7 +235,7 @@ public void afterDelete( deleted.add(job.id()); if (job.id() != first.id()) return; // Commit a peer's update after listing, before deletion reaches the second row. - if (change.equals("states")) base.retry(raced.id()); + if (change.equals("states")) base.jobs().retry(raced.id()); else base.transaction( peer -> { @@ -259,35 +267,38 @@ public void afterDelete( }); assertEquals( List.of(first.id(), last.id()), - client.deleteMany(query.after(query.cursor(boundary)), false).stream() + client.jobs().deleteMany(query.after(query.cursor(boundary)), false).stream() .map(Job::id) .toList()); assertEquals(List.of(first.id(), last.id()), deleted); assertEquals( change.equals("states") ? Job.State.AVAILABLE : Job.State.CANCELLED, - base.get(raced.id()).state()); + base.jobs().get(raced.id()).state()); } @Test void rejectsUnfilteredDeletionUnlessExplicit() { var client = client(); var job = client.insert(TYPE, "kept").job(); - assertThrows(IllegalArgumentException.class, () -> client.deleteMany(JobQuery.all(), false)); + assertThrows( + IllegalArgumentException.class, () -> client.jobs().deleteMany(JobQuery.all(), false)); assertThrows( IllegalArgumentException.class, () -> - client.deleteMany( - JobQuery.builder() - .metadata(Map.of()) - .tagsAll() - .tagsAny() - .priorities() - .limit(1) - .descending(true) - .build(), - false)); - assertEquals(job.id(), client.get(job.id()).id()); - assertEquals(1, client.deleteMany(JobQuery.all(), true).size()); + client + .jobs() + .deleteMany( + JobQuery.builder() + .metadata(Map.of()) + .tagsAll() + .tagsAny() + .priorities() + .limit(1) + .descending(true) + .build(), + false)); + assertEquals(job.id(), client.jobs().get(job.id()).id()); + assertEquals(1, client.jobs().deleteMany(JobQuery.all(), true).size()); } @Test @@ -311,12 +322,15 @@ void skipsRunningJobsAndPreservesSingleDeleteRejection() { }); assertEquals( RiverException.Code.REJECTED, - assertThrows(RiverException.class, () -> client.delete(running.id())).code()); + assertThrows(RiverException.class, () -> client.jobs().delete(running.id())).code()); assertEquals( List.of(available.id()), - client.deleteMany(JobQuery.builder().kinds(TYPE.kind()).limit(1).build(), false).stream() + client + .jobs() + .deleteMany(JobQuery.builder().kinds(TYPE.kind()).limit(1).build(), false) + .stream() .map(Job::id) .toList()); - assertEquals(Job.State.RUNNING, client.get(running.id()).state()); + assertEquals(Job.State.RUNNING, client.jobs().get(running.id()).state()); } } diff --git a/java/river/src/test/java/com/riverqueue/ClientLifecycleTest.java b/java/river/src/test/java/com/riverqueue/ClientLifecycleTest.java index 6fda7cf5e..338f0d178 100644 --- a/java/river/src/test/java/com/riverqueue/ClientLifecycleTest.java +++ b/java/river/src/test/java/com/riverqueue/ClientLifecycleTest.java @@ -41,9 +41,9 @@ void clientCanInsertWithoutStarting() { var database = databases.open(directory.resolve("river.db")); var client = Client.builder(database).build(); var job = client.insert(TYPE, "insert only").job(); - assertEquals(Job.State.AVAILABLE, client.get(job.id()).state()); + assertEquals(Job.State.AVAILABLE, client.jobs().get(job.id()).state()); assertThrows(IllegalArgumentException.class, client::start); - assertEquals(Job.State.AVAILABLE, new Client(database).get(job.id()).state()); + assertEquals(Job.State.AVAILABLE, new Client(database).jobs().get(job.id()).state()); } @Test diff --git a/java/river/src/test/java/com/riverqueue/ClientTest.java b/java/river/src/test/java/com/riverqueue/ClientTest.java index 2bb20316f..dd0ce3daf 100644 --- a/java/river/src/test/java/com/riverqueue/ClientTest.java +++ b/java/river/src/test/java/com/riverqueue/ClientTest.java @@ -36,13 +36,13 @@ void applicationTransactionOwnsCommitAndRollback() throws Exception { try (var connection = river.database().connection()) { connection.setAutoCommit(false); long rolledBack = river.insert(connection, EMAIL, new Email("one@example.com")).job().id(); - assertEquals(rolledBack, river.get(connection, rolledBack).id()); - assertThrows(RiverException.class, () -> river.get(rolledBack)); + assertEquals(rolledBack, river.jobs().get(connection, rolledBack).id()); + assertThrows(RiverException.class, () -> river.jobs().get(rolledBack)); connection.rollback(); - assertThrows(RiverException.class, () -> river.get(rolledBack)); + assertThrows(RiverException.class, () -> river.jobs().get(rolledBack)); long committed = river.insert(connection, EMAIL, new Email("two@example.com")).job().id(); connection.commit(); - assertEquals(committed, river.get(committed).id()); + assertEquals(committed, river.jobs().get(committed).id()); assertFalse(connection.isClosed()); assertFalse(connection.getAutoCommit()); } @@ -87,10 +87,10 @@ void completionMetadataReplacesTopLevelKeys() { return client.read(rows); } }); - client.transaction(connection -> client.complete(connection, claimed, updates)); + client.transaction(connection -> client.jobs().complete(connection, claimed, updates)); var expected = ((tools.jackson.databind.node.ObjectNode) inserted.metadata()).deepCopy(); updates.properties().forEach(entry -> expected.set(entry.getKey(), entry.getValue())); - assertEquals(expected, client.get(claimed.id()).metadata()); + assertEquals(expected, client.jobs().get(claimed.id()).metadata()); } @Test @@ -122,8 +122,8 @@ void crossKindUniqueConflictsPreserveTheIncumbent() { assertFalse(original.uniqueSkippedAsDuplicate()); assertTrue(duplicate.uniqueSkippedAsDuplicate()); assertEquals(original.job(), duplicate.job()); - assertEquals(original.job(), client.get(original.job().id())); - assertEquals(1, client.list(JobQuery.all()).jobs().size()); + assertEquals(original.job(), client.jobs().get(original.job().id())); + assertEquals(1, client.jobs().list(JobQuery.all()).jobs().size()); } @ParameterizedTest @@ -148,7 +148,7 @@ record Requested(List value) {} assertEquals(original.id(), duplicate.job().id()); assertEquals(original.kind(), duplicate.job().kind()); assertEquals(Json.tree(new Existing("original")), Json.tree(duplicate.job().args())); - assertEquals(1, client.list(JobQuery.all()).jobs().size()); + assertEquals(1, client.jobs().list(JobQuery.all()).jobs().size()); } @ParameterizedTest @@ -215,10 +215,10 @@ void delayedCompletionPreservesARescheduledJob(Job.State state) { } return null; }); - var before = client.get(inserted.id()); - var completed = client.transaction(connection -> client.complete(connection, claimed)); + var before = client.jobs().get(inserted.id()); + var completed = client.transaction(connection -> client.jobs().complete(connection, claimed)); assertEquals(before, completed); - assertEquals(before, client.get(inserted.id())); + assertEquals(before, client.jobs().get(inserted.id())); } @Test @@ -239,7 +239,7 @@ void failedBatchRollsBackWithoutAbortingApplicationTransaction() throws Exceptio connection.commit(); assertEquals( List.of(kept.job().id()), - river.list(JobQuery.all()).jobs().stream().map(Job::id).toList()); + river.jobs().list(JobQuery.all()).jobs().stream().map(Job::id).toList()); } } @@ -306,8 +306,8 @@ void fetchOnlyKnownKindsLeavesPeerJobsAvailable() throws Exception { river.insert(EMAIL, new Email("java@example.com")); assertTrue(worked.await(10, TimeUnit.SECONDS)); } - assertEquals(Job.State.AVAILABLE, river.get(unknown.id()).state()); - assertEquals(0, river.get(unknown.id()).attempt()); + assertEquals(Job.State.AVAILABLE, river.jobs().get(unknown.id()).state()); + assertEquals(0, river.jobs().get(unknown.id()).attempt()); } @Test @@ -347,8 +347,8 @@ void removingQueueDrainsItsCommittedClaims() throws Exception { } removal.get(10, TimeUnit.SECONDS); assertEquals(1, calls.get()); - assertEquals(Job.State.COMPLETED, river.get(first.id()).state()); - assertEquals(Job.State.AVAILABLE, river.get(second.id()).state()); + assertEquals(Job.State.COMPLETED, river.jobs().get(first.id()).state()); + assertEquals(Job.State.AVAILABLE, river.jobs().get(second.id()).state()); } } @@ -402,9 +402,9 @@ void workerCompletionCommitsApplicationChangesWithJob() throws Exception { release.countDown(); var completed = done.get(10, TimeUnit.SECONDS); assertEquals(inserted.job().id(), completed.job().id()); - assertEquals(Job.State.COMPLETED, river.get(inserted.job().id()).state()); + assertEquals(Job.State.COMPLETED, river.jobs().get(inserted.job().id()).state()); assertEquals( - "delivered", river.get(inserted.job().id()).metadata().path("output").asString()); + "delivered", river.jobs().get(inserted.job().id()).metadata().path("output").asString()); assertEquals( Integer.valueOf(1), river.transaction( @@ -432,11 +432,12 @@ void defaultListingUsesStableIdOrder() { InsertOptions.builder().scheduledAt(NOW.plusSeconds(3600)).build()) .job(); var second = river.insert(EMAIL, new Email("now@example.com")).job(); - var page = river.list(JobQuery.all().limit(1)); + var page = river.jobs().list(JobQuery.all().limit(1)); assertEquals(first.id(), page.jobs().getFirst().id()); assertEquals( second.id(), river + .jobs() .list(JobQuery.builder().after(page.cursor()).limit(1).build()) .jobs() .getFirst() @@ -444,6 +445,7 @@ void defaultListingUsesStableIdOrder() { assertEquals( second.id(), river + .jobs() .list(JobQuery.builder().order(JobQuery.Order.TIME).limit(1).build()) .jobs() .getFirst() @@ -460,15 +462,15 @@ void typedGetAndMixedBatchRetainArgumentKinds() throws Exception { EMAIL.submission(new Email("one@example.com")), report.submission( new Report(42), InsertOptions.builder().queue("reports").build()))); - Job email = river.get(results.get(0).job().id(), EMAIL); - Job typedReport = river.get(results.get(1).job().id(), report); + Job email = river.jobs().get(results.get(0).job().id(), EMAIL); + Job typedReport = river.jobs().get(results.get(1).job().id(), report); assertEquals(new Email("one@example.com"), email.args()); assertEquals(new Report(42), typedReport.args()); assertEquals("reports", typedReport.queue()); - assertThrows(IllegalArgumentException.class, () -> river.get(email.id(), report)); + assertThrows(IllegalArgumentException.class, () -> river.jobs().get(email.id(), report)); try (var connection = river.database().connection()) { connection.setAutoCommit(false); - assertThrows(IllegalArgumentException.class, () -> river.complete(connection, email)); + assertThrows(IllegalArgumentException.class, () -> river.jobs().complete(connection, email)); var inserted = river.insertMany( connection, @@ -477,7 +479,7 @@ void typedGetAndMixedBatchRetainArgumentKinds() throws Exception { InsertOptions.builder().queue("mail").build()); assertEquals("mail", inserted.getFirst().job().queue()); connection.rollback(); - assertThrows(RiverException.class, () -> river.get(inserted.getFirst().job().id())); + assertThrows(RiverException.class, () -> river.jobs().get(inserted.getFirst().job().id())); } } @@ -521,7 +523,8 @@ public void insertBegin( var kept = base.insert(connection, EMAIL, new Email("kept@example.com")).job(); connection.commit(); assertEquals( - List.of(kept.id()), base.list(JobQuery.all()).jobs().stream().map(Job::id).toList()); + List.of(kept.id()), + base.jobs().list(JobQuery.all()).jobs().stream().map(Job::id).toList()); } } @@ -554,7 +557,7 @@ void sqliteQueueNotificationFailureRollsBackQueueMutation() throws Exception { assertFalse(river.queues().get(connection, "default").metadata().has("changed")); var kept = river.insert(connection, EMAIL, new Email("kept@example.com")).job(); connection.commit(); - assertEquals(kept.id(), river.get(kept.id()).id()); + assertEquals(kept.id(), river.jobs().get(kept.id()).id()); } } @@ -654,7 +657,7 @@ void jobKindLengthMatchesDatabaseConstraint() throws Exception { var args = new Email("long-kind@example.com"); var inserted = client.insert(type, args).job(); assertEquals(type.kind(), inserted.kind()); - assertEquals(args, client.get(inserted.id(), type).args()); + assertEquals(args, client.jobs().get(inserted.id(), type).args()); assertEquals(type.kind(), worked.get(10, TimeUnit.SECONDS).kind()); assertEquals(args, worked.get().args()); assertEquals(inserted.id(), completed.get(10, TimeUnit.SECONDS).job().id()); @@ -743,7 +746,7 @@ public void afterAttempt( } } } - assertEquals(Job.State.COMPLETED, river.list(JobQuery.all()).jobs().getFirst().state()); + assertEquals(Job.State.COMPLETED, river.jobs().list(JobQuery.all()).jobs().getFirst().state()); } @Test diff --git a/java/river/src/test/java/com/riverqueue/HookMiddlewareTest.java b/java/river/src/test/java/com/riverqueue/HookMiddlewareTest.java index cd6abd7a0..6ad4b53ed 100644 --- a/java/river/src/test/java/com/riverqueue/HookMiddlewareTest.java +++ b/java/river/src/test/java/com/riverqueue/HookMiddlewareTest.java @@ -165,7 +165,9 @@ void insertMiddlewareFailureRollsBackWrites(String operation, boolean callerTran } assertEquals( callerTransaction ? List.of("keep") : List.of(), - base.list(JobQuery.all()).jobs().stream().map(job -> job.args().asString()).toList()); + base.jobs().list(JobQuery.all()).jobs().stream() + .map(job -> job.args().asString()) + .toList()); } @Test diff --git a/java/river/src/test/java/com/riverqueue/JobQueryTest.java b/java/river/src/test/java/com/riverqueue/JobQueryTest.java index ec1a0927b..9bd3ef997 100644 --- a/java/river/src/test/java/com/riverqueue/JobQueryTest.java +++ b/java/river/src/test/java/com/riverqueue/JobQueryTest.java @@ -51,9 +51,9 @@ void metadataContainmentPreservesJsonSemantics(String stored, String fragment, b .job(); var query = JobQuery.builder().metadata(Json.parse(fragment)).build(); var expected = matches ? List.of(inserted.id()) : List.of(); - assertEquals(expected, client.list(query).jobs().stream().map(Job::id).toList()); - assertEquals(expected, client.deleteMany(query, false).stream().map(Job::id).toList()); - assertEquals(matches ? 0 : 1, client.list(JobQuery.all()).jobs().size()); + assertEquals(expected, client.jobs().list(query).jobs().stream().map(Job::id).toList()); + assertEquals(expected, client.jobs().deleteMany(query, false).stream().map(Job::id).toList()); + assertEquals(matches ? 0 : 1, client.jobs().list(JobQuery.all()).jobs().size()); } @ParameterizedTest @@ -90,9 +90,10 @@ void metadataFiltersRequireEmptyContainers(String container, boolean nested) { filter.set( "tenant", nested ? Json.object().set("settings", Json.parse(empty)) : Json.parse(empty)); var query = JobQuery.builder().metadata(filter).build(); - assertEquals(expected, client.list(query).jobs().stream().map(Job::id).toList()); - assertEquals(expected, client.deleteMany(query, false).stream().map(Job::id).toList()); - assertEquals(remaining, client.list(JobQuery.all()).jobs().stream().map(Job::id).toList()); + assertEquals(expected, client.jobs().list(query).jobs().stream().map(Job::id).toList()); + assertEquals(expected, client.jobs().deleteMany(query, false).stream().map(Job::id).toList()); + assertEquals( + remaining, client.jobs().list(JobQuery.all()).jobs().stream().map(Job::id).toList()); } @Test diff --git a/java/river/src/test/java/com/riverqueue/JobUpdateOptionsTest.java b/java/river/src/test/java/com/riverqueue/JobUpdateOptionsTest.java new file mode 100644 index 000000000..469770afc --- /dev/null +++ b/java/river/src/test/java/com/riverqueue/JobUpdateOptionsTest.java @@ -0,0 +1,57 @@ +package com.riverqueue; + +import static org.junit.jupiter.api.Assertions.*; + +import java.util.Map; +import org.junit.jupiter.api.Test; +import tools.jackson.databind.node.ObjectNode; + +class JobUpdateOptionsTest { + @Test + void buildersSnapshotInputsAndOptionsCopyTheirAccessors() { + var metadata = Json.object().set("nested", Json.object().put("saved", true)); + var output = Json.object().put("saved", true); + var builder = JobUpdateOptions.builder().metadata(metadata).output(output); + ((ObjectNode) metadata.path("nested")).put("saved", false); + output.put("saved", false); + var options = builder.build(); + builder.metadata(Map.of("later", true)).output("later"); + ((ObjectNode) options.metadata().path("nested")).put("saved", false); + ((ObjectNode) options.output()).put("saved", false); + + assertEquals(Json.parse("{\"nested\":{\"saved\":true}}"), options.metadata()); + assertEquals(Json.parse("{\"saved\":true}"), options.output()); + } + + @Test + void constructorCopiesInputs() { + var metadata = Json.object().put("saved", true); + var output = Json.object().put("saved", true); + var options = new JobUpdateOptions(metadata, output); + metadata.put("saved", false); + output.put("saved", false); + + assertEquals(Json.parse("{\"saved\":true}"), options.metadata()); + assertEquals(Json.parse("{\"saved\":true}"), options.output()); + } + + @Test + void defaultsDistinguishOmittedOutputFromJsonNull() { + assertEquals(Json.object(), JobUpdateOptions.defaults().metadata()); + assertNull(JobUpdateOptions.defaults().output()); + assertTrue(JobUpdateOptions.builder().output(null).build().output().isNull()); + assertEquals(Json.parse("false"), JobUpdateOptions.builder().output(false).build().output()); + } + + @Test + void metadataRequiresAnObject() { + for (var metadata : Json.parse("[null,[],true,42,\"text\"]")) { + assertThrows(IllegalArgumentException.class, () -> new JobUpdateOptions(metadata, null)); + assertThrows( + IllegalArgumentException.class, + () -> JobUpdateOptions.builder().metadata(metadata).build()); + } + assertThrows( + IllegalArgumentException.class, () -> JobUpdateOptions.builder().metadata(null).build()); + } +} diff --git a/java/river/src/test/java/com/riverqueue/JobUpdateTest.java b/java/river/src/test/java/com/riverqueue/JobUpdateTest.java new file mode 100644 index 000000000..1ec924b81 --- /dev/null +++ b/java/river/src/test/java/com/riverqueue/JobUpdateTest.java @@ -0,0 +1,230 @@ +package com.riverqueue; + +import static org.junit.jupiter.api.Assertions.*; + +import java.nio.file.Path; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.stream.Stream; +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.MethodSource; +import tools.jackson.databind.JsonNode; +import tools.jackson.databind.node.ObjectNode; + +class JobUpdateTest { + private static final JobType TYPE = JobType.of("update_test", String.class); + @RegisterExtension final TestDatabase databases = new TestDatabase(); + @TempDir Path directory; + + @Test + void applicationOwnsCommitAndRollback() throws Exception { + var client = client(); + var original = client.insert(TYPE, "original").job(); + var updates = JobUpdateOptions.builder().output("saved").build(); + try (var connection = client.database().connection()) { + connection.setAutoCommit(false); + var updated = client.jobs().update(connection, original.id(), updates); + assertEquals("saved", updated.metadata().path("output").asString()); + assertEquals(updated, client.jobs().get(connection, original.id())); + assertEquals(original, client.jobs().get(original.id())); + connection.rollback(); + assertEquals(original, client.jobs().get(original.id())); + + client.jobs().update(connection, original.id(), updates); + connection.commit(); + assertEquals(updated, client.jobs().get(original.id())); + assertFalse(connection.isClosed()); + assertFalse(connection.getAutoCommit()); + } + } + + private Client client() { + return new Client(databases.open(directory.resolve("river.db"))); + } + + @Test + void connectionRequiresAnExplicitTransaction() throws Exception { + var client = client(); + var original = client.insert(TYPE, "original").job(); + try (var connection = client.database().connection()) { + assertThrows( + IllegalArgumentException.class, + () -> + client + .jobs() + .update( + connection, original.id(), JobUpdateOptions.builder().output("new").build())); + assertThrows( + IllegalArgumentException.class, + () -> client.jobs().update(connection, original.id(), JobUpdateOptions.defaults())); + } + assertEquals(original, client.jobs().get(original.id())); + } + + @Test + void emptyOptionsAndOmittedOutputLeaveValuesUnchanged() { + var client = client(); + var original = + client + .insert( + TYPE, + "original", + InsertOptions.builder().metadata(Map.of("output", "saved")).build()) + .job(); + assertEquals(original, client.jobs().update(original.id(), JobUpdateOptions.defaults())); + assertEquals( + original, + client.jobs().update(original.id(), JobUpdateOptions.builder().metadata(Map.of()).build())); + assertEquals( + "saved", + client + .jobs() + .update(original.id(), JobUpdateOptions.builder().metadata(Map.of("new", true)).build()) + .metadata() + .path("output") + .asString()); + } + + @Test + void explicitOutputOverridesMetadataWithoutChangingOtherJobFields() { + var client = client(); + var original = + client + .insert( + TYPE, + "original", + InsertOptions.builder().metadata(Map.of("keep", true, "output", "old")).build()) + .job(); + var updates = + JobUpdateOptions.builder() + .metadata(Map.of("output", "metadata value", "new", true)) + .output(Map.of("explicit", true)) + .build(); + var updated = client.jobs().update(original.id(), updates); + var expected = (ObjectNode) original.metadata().deepCopy(); + expected.put("new", true); + expected.set("output", Json.tree(Map.of("explicit", true))); + + assertEquals(expected, updated.metadata()); + assertEquals(updated, client.jobs().get(original.id())); + // Job metadata is a detached snapshot; compare all remaining job fields too. + ((ObjectNode) updated.metadata()).removeAll(); + ((ObjectNode) original.metadata()).removeAll(); + assertEquals(original, updated); + assertEquals("metadata value", updates.metadata().path("output").asString()); + } + + @Test + void mergesWholeMetadataValuesAndPreservesExactJson() { + var client = client(); + var stored = + Json.parse( + "{\"keep\":{\"x\":1},\"nested\":{\"old\":true},\"river:resumable_step\":\"normalize\",\"large\":123456789012345678901234567890,\"fraction\":1.234567890123456789}"); + var original = + client.insert(TYPE, "original", InsertOptions.builder().metadata(stored).build()).job(); + client + .jobs() + .update(original.id(), JobUpdateOptions.builder().metadata(Map.of("later", true)).build()); + var metadata = Json.object(); + metadata.set("nested", Json.object()); + metadata.putNull("nullable"); + metadata.set("a\"b.c", Json.parse("[true,false,null]")); + metadata.put("", false); + metadata.putNull("a\\b"); + for (int i = 0; i < 1200; i++) metadata.put("key_" + i, i); + var expected = (ObjectNode) original.metadata().deepCopy(); + expected.put("later", true); + metadata.properties().forEach(entry -> expected.set(entry.getKey(), entry.getValue())); + + var updated = + client.jobs().update(original.id(), JobUpdateOptions.builder().metadata(metadata).build()); + + assertEquals(expected, updated.metadata()); + assertEquals(updated, client.jobs().get(original.id())); + } + + @Test + void missingJobsReportNotFoundWithoutPoisoningTheTransaction() throws Exception { + var client = client(); + for (var updates : + new JobUpdateOptions[] { + JobUpdateOptions.defaults(), JobUpdateOptions.builder().output("new").build() + }) { + assertEquals( + RiverException.Code.NOT_FOUND, + assertThrows(RiverException.class, () -> client.jobs().update(-1, updates)).code()); + client.transaction( + connection -> { + assertEquals( + RiverException.Code.NOT_FOUND, + assertThrows( + RiverException.class, () -> client.jobs().update(connection, -1, updates)) + .code()); + return client.insert(connection, TYPE, "usable"); + }); + } + } + + @ParameterizedTest + @MethodSource("outputValues") + void outputReplacesThePreviousValueAndPreservesOtherMetadata(JsonNode output) { + var client = client(); + var original = + client + .insert( + TYPE, + "original", + InsertOptions.builder() + .metadata(Map.of("output", Map.of("old", true), "keep", true)) + .build()) + .job(); + var updated = + client.jobs().update(original.id(), JobUpdateOptions.builder().output(output).build()); + var expected = (ObjectNode) original.metadata().deepCopy(); + expected.set("output", output); + + assertEquals(expected, updated.metadata()); + assertEquals(updated, client.jobs().get(original.id())); + } + + static Stream outputValues() { + return Json.parse( + "[null,false,true,42,9007199254740993,1.25,\"null\",\"a 'quoted' string\",{},[],[true,null]]") + .valueStream(); + } + + @Test + void updateIsVisibleBeforeTheWorkerFinishesAndSurvivesCompletion() throws Exception { + var database = databases.open(directory.resolve("river.db")); + var observed = new CompletableFuture>(); + var completed = new CompletableFuture(); + var client = + Client.builder(database) + .queue("default", 1) + .worker( + TYPE, + context -> { + context + .client() + .jobs() + .update( + context.job().id(), JobUpdateOptions.builder().output("saved").build()); + observed.complete(context.client().jobs().get(context.job().id())); + }) + .build(); + client.subscribe(completed::complete, Client.EventKind.JOB_COMPLETED); + var inserted = client.insert(TYPE, "original").job(); + try (var run = client.start()) { + assertSame(client, run.client()); + var running = observed.get(10, TimeUnit.SECONDS); + assertEquals(Job.State.RUNNING, running.state()); + assertEquals("saved", running.metadata().path("output").asString()); + assertEquals(inserted.id(), completed.get(10, TimeUnit.SECONDS).job().id()); + assertEquals("saved", client.jobs().get(inserted.id()).metadata().path("output").asString()); + } + } +} diff --git a/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java b/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java index 6a2a0c363..66d05fa2a 100644 --- a/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java +++ b/java/river/src/test/java/com/riverqueue/PeerCancellationTest.java @@ -116,7 +116,7 @@ public JsonNode decode(Job job) { assertFalse( provisional.forceIfStuck(Duration.ZERO), "Finished attempts must stop being supervised"); workers.stop(); - var job = client.get(inserted.id()); + var job = client.jobs().get(inserted.id()); if (when.equals("forced")) { assertEquals(1, job.attempt()); assertEquals(1, job.errors().size()); @@ -231,7 +231,7 @@ public void work(WorkContext context) throws Exception { assertNotNull(event); assertEquals(Client.EventKind.JOB_FAILED, event.kind()); } - assertEquals(Job.State.DISCARDED, client.get(peerJob.id()).state()); + assertEquals(Job.State.DISCARDED, client.jobs().get(peerJob.id()).state()); } finally { release.countDown(); workers.stopAndCancel(); diff --git a/java/river/src/test/java/com/riverqueue/ProtocolStorageTest.java b/java/river/src/test/java/com/riverqueue/ProtocolStorageTest.java index 9cafbdd19..169928955 100644 --- a/java/river/src/test/java/com/riverqueue/ProtocolStorageTest.java +++ b/java/river/src/test/java/com/riverqueue/ProtocolStorageTest.java @@ -49,7 +49,7 @@ void periodicMetadata() throws Exception { .build() .start()) { var job = claimed.get(5, TimeUnit.SECONDS); - var metadata = client.get(job.id()).metadata(); + var metadata = client.jobs().get(job.id()).metadata(); assertTrue(metadata.required("periodic").asBoolean()); assertEquals( "conformance_periodic", @@ -108,7 +108,7 @@ public boolean clean(Client client, Instant now) { }); assertEquals( fixture.required("attempt_error"), - Json.tree(client.get(inserted.id()).errors().getFirst())); + Json.tree(client.jobs().get(inserted.id()).errors().getFirst())); var completed = new CompletableFuture(); var writeFailure = new CompletableFuture(); @@ -144,16 +144,16 @@ public boolean clean(Client client, Instant now) { assertNull( writeFailure.get(10, TimeUnit.SECONDS), "Rescue callbacks must not hold the writer"); maintained.get(5, TimeUnit.SECONDS); - var rescued = client.get(inserted.id()); + var rescued = client.jobs().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())); - client.retry(inserted.id()); + client.jobs().retry(inserted.id()); assertEquals(inserted.id(), completed.get(5, TimeUnit.SECONDS).job().id()); - var job = client.get(inserted.id()); + var job = client.jobs().get(inserted.id()); assertEquals(Job.State.COMPLETED, job.state()); assertEquals( Json.tree(Map.of("ok", true)), @@ -199,7 +199,7 @@ void storedStatesAndUniqueBits() throws Exception { } return null; }); - var job = client.get(inserted.id()); + var job = client.jobs().get(inserted.id()); assertEquals(state, job.state().value()); assertEquals(List.of(job.state()), job.uniqueStates()); assertEquals(bit, job.uniqueStates().getFirst().bit()); diff --git a/java/river/src/test/java/com/riverqueue/RecordedOutputTest.java b/java/river/src/test/java/com/riverqueue/RecordedOutputTest.java new file mode 100644 index 000000000..06cfd02ec --- /dev/null +++ b/java/river/src/test/java/com/riverqueue/RecordedOutputTest.java @@ -0,0 +1,111 @@ +package com.riverqueue; + +import static org.junit.jupiter.api.Assertions.*; + +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +class RecordedOutputTest { + private static final int MAX_BYTES = 32 * 1024 * 1024; + private static final JobType TYPE = JobType.of("output_test", String.class); + @RegisterExtension final TestDatabase databases = new TestDatabase(); + @TempDir Path directory; + + @ParameterizedTest + @ValueSource(strings = {"x", "é", "\"", "<"}) + void jobUpdatesEnforceEncodedOutputLimit(String character) { + var client = new Client(databases.open(directory.resolve("river.db"))); + var original = client.insert(TYPE, "work").job(); + var belowLimit = outputOfSize(character, MAX_BYTES - 1); + assertDoesNotThrow(() -> JobUpdateOptions.builder().output(belowLimit).build()); + var atLimit = outputOfSize(character, MAX_BYTES); + var updated = + client + .jobs() + .update( + original.id(), + JobUpdateOptions.builder().metadata(Map.of("kept", true)).output(atLimit).build()); + assertEquals(atLimit, updated.metadata().path("output").asString()); + + var oversized = outputOfSize(character, MAX_BYTES + 1); + var error = + assertThrows( + IllegalArgumentException.class, + () -> + client + .jobs() + .update( + original.id(), + JobUpdateOptions.builder() + .metadata(Map.of("kept", false)) + .output(oversized) + .build())); + assertEquals("output is too large: 33554433 bytes (max 32 MiB)", error.getMessage()); + assertThrows( + IllegalArgumentException.class, () -> new JobUpdateOptions(null, Json.tree(oversized))); + assertEquals(updated, client.jobs().get(original.id())); + } + + private static String outputOfSize(String character, int bytes) { + int encodedCharacterSize = Json.encode(character).getBytes(StandardCharsets.UTF_8).length - 2; + // Account for the JSON string's quotes as well as escaped and multibyte characters. + return character.repeat((bytes - 2) / encodedCharacterSize) + + "x".repeat((bytes - 2) % encodedCharacterSize); + } + + @ParameterizedTest + @ValueSource(strings = {"x", "é", "\"", "<"}) + void workerOutputEnforcesEncodedOutputLimitAndPreservesPreviousOutput(String character) + throws Exception { + var completed = new CompletableFuture(); + var atLimit = outputOfSize(character, MAX_BYTES); + var client = + Client.builder(databases.open(directory.resolve("river.db"))) + .queue("default", 1) + .worker( + TYPE, + context -> { + context.metadata("kept", true); + context.output(outputOfSize(character, MAX_BYTES - 1)); + context.output(atLimit); + var error = + assertThrows( + IllegalArgumentException.class, + () -> context.output(outputOfSize(character, MAX_BYTES + 1))); + assertEquals( + "output is too large: 33554433 bytes (max 32 MiB)", error.getMessage()); + }) + .build(); + client.subscribe( + completed::complete, Client.EventKind.JOB_COMPLETED, Client.EventKind.JOB_FAILED); + var inserted = client.insert(TYPE, "work").job(); + try (var run = client.start()) { + assertSame(client, run.client()); + var event = completed.get(30, TimeUnit.SECONDS); + assertEquals(Client.EventKind.JOB_COMPLETED, event.kind()); + var saved = client.jobs().get(inserted.id()); + assertEquals(atLimit, saved.metadata().path("output").asString()); + assertTrue(saved.metadata().path("kept").asBoolean()); + } + } + + @Test + void workerOutputSnapshotsMutableValuesAndAcceptsJsonNull() { + var client = new Client(databases.open(directory.resolve("river.db"))); + var context = new WorkContext<>(client, client.insert(TYPE, "work").job()); + var output = Json.object().put("value", "saved"); + context.output(output); + output.put("value", "changed"); + assertEquals("saved", context.metadataUpdates().path("output").path("value").asString()); + context.output(null); + assertTrue(context.metadataUpdates().path("output").isNull()); + } +} diff --git a/java/river/src/test/java/com/riverqueue/RescueTest.java b/java/river/src/test/java/com/riverqueue/RescueTest.java index a167925c9..d87dac40e 100644 --- a/java/river/src/test/java/com/riverqueue/RescueTest.java +++ b/java/river/src/test/java/com/riverqueue/RescueTest.java @@ -186,7 +186,7 @@ public void work(WorkContext context) { write.get(2, TimeUnit.SECONDS); release.complete(null); maintained.get(5, TimeUnit.SECONDS); - var updated = client.get(job.id()); + var updated = client.jobs().get(job.id()); assertEquals(Job.State.RUNNING, updated.state()); assertEquals(2, updated.attempt()); assertTrue(updated.errors().isEmpty()); @@ -285,7 +285,7 @@ public void work(WorkContext context) { case "disabled", "long" -> Job.State.RUNNING; default -> Job.State.DISCARDED; }; - assertEquals(expected, client.get(entry.getValue()).state(), entry.getKey()); + assertEquals(expected, client.jobs().get(entry.getValue()).state(), entry.getKey()); } } assertEquals("timeout error", errors.remove().getMessage()); @@ -381,7 +381,7 @@ public List claim( .build() .start()) { maintained.get(5, TimeUnit.SECONDS); - var rescued = client.get(inserted.id()); + var rescued = client.jobs().get(inserted.id()); assertEquals(Job.State.AVAILABLE, rescued.state()); var attempt = worked.get(5, TimeUnit.SECONDS); assertEquals(inserted.id(), attempt.id()); diff --git a/java/river/src/test/java/com/riverqueue/ResumableTest.java b/java/river/src/test/java/com/riverqueue/ResumableTest.java index b20beae92..3473b6691 100644 --- a/java/river/src/test/java/com/riverqueue/ResumableTest.java +++ b/java/river/src/test/java/com/riverqueue/ResumableTest.java @@ -55,12 +55,12 @@ void cursorOnlyStepRemainsRecoverableAfterItIsOmitted() throws Exception { var failed = events.poll(5, TimeUnit.SECONDS); assertNotNull(failed); assertEquals(Client.EventKind.JOB_FAILED, failed.kind()); - var savedMetadata = client.get(inserted.id()).metadata(); + var savedMetadata = client.jobs().get(inserted.id()).metadata(); assertEquals( Json.tree(metadata).path(Protocol.METADATA_RESUMABLE_CURSOR), savedMetadata.path(Protocol.METADATA_RESUMABLE_CURSOR)); assertFalse(savedMetadata.has(Protocol.METADATA_RESUMABLE_STEP)); - client.retry(inserted.id()); + client.jobs().retry(inserted.id()); var completed = events.poll(5, TimeUnit.SECONDS); assertNotNull(completed); assertEquals(Client.EventKind.JOB_COMPLETED, completed.kind()); @@ -160,18 +160,18 @@ void checkpointCommitsWithApplicationWrites( var failed = events.poll(5, TimeUnit.SECONDS); assertNotNull(failed); assertEquals(Client.EventKind.JOB_FAILED, failed.kind()); - var saved = client.get(inserted.id()); + var saved = client.jobs().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()); + assertEquals(rollback ? 1 : 2, client.jobs().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()); + client.jobs().retry(inserted.id()); var completed = events.poll(5, TimeUnit.SECONDS); assertNotNull(completed); assertEquals(Client.EventKind.JOB_COMPLETED, completed.kind()); @@ -201,7 +201,7 @@ void checkpointRejectsInvalidTransactionsAndStaleAttempts() throws Exception { } return null; }); - var job = client.get(inserted.id(), TYPE); + var job = client.jobs().get(inserted.id(), TYPE); var context = new WorkContext<>(client, job); try (var connection = client.database().connection()) { assertThrows(IllegalStateException.class, () -> context.checkpoint(connection)); @@ -210,11 +210,11 @@ void checkpointRejectsInvalidTransactionsAndStaleAttempts() throws Exception { () -> { assertThrows(IllegalArgumentException.class, () -> context.checkpoint(connection)); connection.setAutoCommit(false); - client.complete(connection, job); + client.jobs().complete(connection, job); assertEquals( RiverException.Code.REJECTED, assertThrows(RiverException.class, () -> context.checkpoint(connection)).code()); - assertEquals(Job.State.COMPLETED, client.get(connection, job.id()).state()); + assertEquals(Job.State.COMPLETED, client.jobs().get(connection, job.id()).state()); connection.rollback(); try (var statement = Sql.prepare( @@ -227,7 +227,7 @@ void checkpointRejectsInvalidTransactionsAndStaleAttempts() throws Exception { connection.rollback(); }); } - assertFalse(client.get(job.id()).metadata().has(Protocol.METADATA_RESUMABLE_STEP)); + assertFalse(client.jobs().get(job.id()).metadata().has(Protocol.METADATA_RESUMABLE_STEP)); } @ParameterizedTest @@ -250,7 +250,7 @@ void checkpointRetainsOnlyCommittedProgress(boolean rollback, boolean finishStep } return null; }); - var context = new WorkContext<>(client, client.get(inserted.id(), TYPE)); + var context = new WorkContext<>(client, client.jobs().get(inserted.id(), TYPE)); var failure = new IllegalStateException("retry"); WorkContext.Step work = () -> @@ -277,6 +277,7 @@ void checkpointRetainsOnlyCommittedProgress(boolean rollback, boolean finishStep assertEquals( rollback ? 1 : 2, client + .jobs() .get(inserted.id()) .metadata() .path(Protocol.METADATA_RESUMABLE_CURSOR) @@ -331,11 +332,11 @@ void cursorDoesNotCompleteFailedSteps(String mode) throws Exception { var failed = events.poll(5, TimeUnit.SECONDS); assertNotNull(failed); assertEquals(Client.EventKind.JOB_FAILED, failed.kind()); - var metadata = client.get(job.id()).metadata(); + var metadata = client.jobs().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()); + client.jobs().retry(job.id()); var completed = events.poll(5, TimeUnit.SECONDS); assertNotNull(completed); assertEquals(Client.EventKind.JOB_COMPLETED, completed.kind()); @@ -427,11 +428,12 @@ void cursorMapsReplaceCompletedStepsAndPreserveNullValues() throws Exception { attempt == 3 ? Client.EventKind.JOB_COMPLETED : Client.EventKind.JOB_FAILED, event.kind()); if (attempt == 2) { - var cursors = client.get(job.id()).metadata().path(Protocol.METADATA_RESUMABLE_CURSOR); + var cursors = + client.jobs().get(job.id()).metadata().path(Protocol.METADATA_RESUMABLE_CURSOR); assertFalse(cursors.has("first")); assertTrue(cursors.has("second")); } - if (attempt < 3) client.retry(job.id()); + if (attempt < 3) client.jobs().retry(job.id()); } assertEquals(2, firstCalls.get()); } diff --git a/java/river/src/test/java/com/riverqueue/RetryTest.java b/java/river/src/test/java/com/riverqueue/RetryTest.java index fb76a4d2c..a325a8608 100644 --- a/java/river/src/test/java/com/riverqueue/RetryTest.java +++ b/java/river/src/test/java/com/riverqueue/RetryTest.java @@ -52,10 +52,10 @@ void retryNotifiesOnlyWhenTheRowChanges(Job.State state) { } return null; }); - var before = client.get(inserted.id()); + var before = client.jobs().get(inserted.id()); var notifications = new AtomicInteger(); client.onInsertCommit(notifications::incrementAndGet); - var retried = client.retry(inserted.id()); + var retried = client.jobs().retry(inserted.id()); if (state == Job.State.AVAILABLE || state == Job.State.RUNNING) { assertEquals(before, retried); assertEquals(0, notifications.get()); @@ -139,15 +139,15 @@ public List claim( if (mode.equals("transaction")) { try (var connection = database.connection()) { connection.setAutoCommit(false); - assertEquals(Job.State.AVAILABLE, actor.retry(connection, inserted.id()).state()); - assertEquals(Job.State.SCHEDULED, client.get(inserted.id()).state()); + assertEquals(Job.State.AVAILABLE, actor.jobs().retry(connection, inserted.id()).state()); + assertEquals(Job.State.SCHEDULED, client.jobs().get(inserted.id()).state()); assertFalse(worked.isDone()); connection.rollback(); - assertEquals(Job.State.SCHEDULED, client.get(inserted.id()).state()); - actor.retry(connection, inserted.id()); + assertEquals(Job.State.SCHEDULED, client.jobs().get(inserted.id()).state()); + actor.jobs().retry(connection, inserted.id()); connection.commit(); } - } else actor.retry(inserted.id()); + } else actor.jobs().retry(inserted.id()); assertEquals(inserted.id(), worked.get(5, TimeUnit.SECONDS).id()); } assertTrue(errors.isEmpty(), () -> "Unexpected worker errors: " + errors); @@ -163,9 +163,9 @@ void retryClearsAnEarlierCancellationRequest() { .insert( TYPE, "retry", InsertOptions.builder().metadata(Map.of("preserved", true)).build()) .job(); - assertTrue(client.cancel(inserted.id()).metadata().has("cancel_attempted_at")); + assertTrue(client.jobs().cancel(inserted.id()).metadata().has("cancel_attempted_at")); - var retried = client.retry(inserted.id()); + var retried = client.jobs().retry(inserted.id()); assertEquals(Job.State.AVAILABLE, retried.state()); assertFalse(retried.metadata().has("cancel_attempted_at")); assertTrue(retried.metadata().path("preserved").asBoolean()); @@ -193,14 +193,14 @@ void rolledBackRetriesPreserveEarlierCommittedNotifications(boolean earlierInser client.transaction( connection, nested -> { - client.retry(nested, inserted.id()); + client.jobs().retry(nested, inserted.id()); throw new IllegalStateException("roll back retry"); })); - assertEquals(inserted, client.get(connection, inserted.id())); + assertEquals(inserted, client.jobs().get(connection, inserted.id())); assertEquals(0, notifications.get(), "Callbacks must wait for the outer commit"); return null; }); - assertEquals(inserted, client.get(inserted.id())); + assertEquals(inserted, client.jobs().get(inserted.id())); assertEquals(earlierInsert ? 1 : 0, notifications.get()); } @@ -224,12 +224,17 @@ void sqliteNotificationFailureRollsBackTheRetry() { "CREATE TRIGGER reject_insert BEFORE INSERT ON river_notification " + "WHEN NEW.topic = 'river_insert' BEGIN SELECT RAISE(ABORT, 'reject insert'); END"); } - assertThrows(RiverException.class, () -> client.retry(connection, inserted.id())); - assertEquals(inserted, client.get(connection, inserted.id())); - client.output(connection, inserted.id(), "outer transaction remains usable"); + assertThrows(RiverException.class, () -> client.jobs().retry(connection, inserted.id())); + assertEquals(inserted, client.jobs().get(connection, inserted.id())); + client + .jobs() + .update( + connection, + inserted.id(), + JobUpdateOptions.builder().output("outer transaction remains usable").build()); return null; }); - var job = client.get(inserted.id()); + var job = client.jobs().get(inserted.id()); assertEquals(Job.State.SCHEDULED, job.state()); assertEquals( "outer transaction remains usable", diff --git a/java/river/src/test/java/com/riverqueue/RunHandleTest.java b/java/river/src/test/java/com/riverqueue/RunHandleTest.java index 46efd1c4f..48e8836b6 100644 --- a/java/river/src/test/java/com/riverqueue/RunHandleTest.java +++ b/java/river/src/test/java/com/riverqueue/RunHandleTest.java @@ -96,7 +96,7 @@ void cancellationRespectsTheHandlersOutcome(WorkContext.Cancellation cause, bool var inserted = client.insert(TYPE, "cancel", InsertOptions.builder().maxAttempts(1).build()).job(); var context = entered.get(5, TimeUnit.SECONDS); - if (cause == WorkContext.Cancellation.REMOTE) client.cancel(inserted.id()); + if (cause == WorkContext.Cancellation.REMOTE) client.jobs().cancel(inserted.id()); else context.requestCancellation(cause); var event = outcome.get(5, TimeUnit.SECONDS); var expected = @@ -107,7 +107,7 @@ void cancellationRespectsTheHandlersOutcome(WorkContext.Cancellation cause, bool : Job.State.DISCARDED; assertEquals(cause, context.cancellation()); assertEquals(expected, event.job().state()); - var job = client.get(inserted.id()); + var job = client.jobs().get(inserted.id()); assertEquals(expected, job.state()); assertEquals(interrupted ? 1 : 0, job.errors().size()); assertEquals(1, job.attempt()); @@ -161,7 +161,7 @@ void completionEventsReflectPersistedState(Job.State state) throws Exception { var job = client.insert(TYPE, "changed").job(); changed.get(5, TimeUnit.SECONDS); workers.stop(); - assertEquals(state, client.get(job.id()).state()); + assertEquals(state, client.jobs().get(job.id()).state()); var expected = switch (state) { case AVAILABLE, RETRYABLE -> Client.EventKind.JOB_FAILED; @@ -212,7 +212,7 @@ public void afterAttempt( assertNotNull( event, "An interrupted handler must not strand completion or its worker slot"); assertEquals(job.id(), event.job().id()); - assertEquals(Job.State.COMPLETED, client.get(job.id()).state()); + assertEquals(Job.State.COMPLETED, client.jobs().get(job.id()).state()); } assertEquals(3, attempts.get()); assertEquals(1, reported.size()); @@ -283,9 +283,9 @@ public java.util.List claim( assertNotNull(finalized.poll(5, TimeUnit.SECONDS), "Every committed claim must settle"); workers.stop(); assertEquals(java.util.List.of(good.id()), java.util.List.copyOf(handled)); - assertEquals(Job.State.DISCARDED, client.get(bad.id()).state()); - assertEquals(Job.State.COMPLETED, client.get(good.id()).state()); - assertEquals(1, client.get(bad.id()).errors().size()); + assertEquals(Job.State.DISCARDED, client.jobs().get(bad.id()).state()); + assertEquals(Job.State.COMPLETED, client.jobs().get(good.id()).state()); + assertEquals(1, client.jobs().get(bad.id()).errors().size()); assertEquals(2, attempts.get()); // A rolled-back completion batch may report the same failure for both claims. assertFalse(reported.isEmpty()); @@ -325,7 +325,7 @@ public void afterAttempt( assertNull(context.get(5, TimeUnit.SECONDS).cancellation()); release.countDown(); assertEquals(Client.EventKind.JOB_COMPLETED, completed.get(5, TimeUnit.SECONDS).kind()); - assertEquals(Job.State.COMPLETED, client.get(job.id()).state()); + assertEquals(Job.State.COMPLETED, client.jobs().get(job.id()).state()); } finally { release.countDown(); workers.stopAndCancel(); @@ -745,7 +745,7 @@ void callbackFailuresDoNotLoseWorkerSlots(String callback) throws Exception { var event = completed.poll(5, TimeUnit.SECONDS); assertNotNull(event, "Callback failure must not strand the attempt or its worker slot"); assertEquals(inserted.job().id(), event.job().id()); - assertEquals(Job.State.COMPLETED, client.get(event.job().id()).state()); + assertEquals(Job.State.COMPLETED, client.jobs().get(event.job().id()).state()); } } assertEquals(2, handled.get()); @@ -881,7 +881,7 @@ void shutdownRespectsTheHandlersOutcome(boolean interrupted) throws Exception { assertEquals( interrupted ? Client.EventKind.JOB_INTERRUPTED : Client.EventKind.JOB_COMPLETED, event.kind()); - var job = client.get(inserted.id()); + var job = client.jobs().get(inserted.id()); assertEquals(interrupted ? Job.State.AVAILABLE : Job.State.COMPLETED, job.state()); assertEquals(interrupted ? 0 : 1, job.attempt()); assertTrue(job.errors().isEmpty()); @@ -951,7 +951,7 @@ public java.util.List claim( assertEquals( cancel ? Client.EventKind.JOB_INTERRUPTED : Client.EventKind.JOB_COMPLETED, event.kind()); workers.stop(); - var job = client.get(inserted.id()); + var job = client.jobs().get(inserted.id()); assertEquals(cancel ? Job.State.AVAILABLE : Job.State.COMPLETED, job.state()); assertEquals(cancel ? 0 : 1, job.attempt()); assertTrue(job.errors().isEmpty()); diff --git a/java/river/src/test/java/com/riverqueue/SchedulingTest.java b/java/river/src/test/java/com/riverqueue/SchedulingTest.java index 969c0eadd..73355dfeb 100644 --- a/java/river/src/test/java/com/riverqueue/SchedulingTest.java +++ b/java/river/src/test/java/com/riverqueue/SchedulingTest.java @@ -330,7 +330,7 @@ public void maintain(Client client, Instant now) { .build() .start()) { maintained.get(5, TimeUnit.SECONDS); - assertEquals(1, client.list(JobQuery.all()).jobs().size()); + assertEquals(1, client.jobs().list(JobQuery.all()).jobs().size()); assertEquals("schedule failed", errors.remove().getMessage()); assertTrue(errors.isEmpty()); } @@ -485,7 +485,7 @@ public void maintain(Client client, Instant now) { } release.complete(null); maintained.get(5, TimeUnit.SECONDS); - assertTrue(client.list(JobQuery.all()).jobs().isEmpty()); + assertTrue(client.jobs().list(JobQuery.all()).jobs().isEmpty()); } finally { release.complete(null); workers.stop(); diff --git a/java/river/src/test/java/com/riverqueue/WorkersTest.java b/java/river/src/test/java/com/riverqueue/WorkersTest.java index a213e3cb3..c28497465 100644 --- a/java/river/src/test/java/com/riverqueue/WorkersTest.java +++ b/java/river/src/test/java/com/riverqueue/WorkersTest.java @@ -55,7 +55,7 @@ void registryAndBuilderAreSnapshotted() throws Exception { assertNotNull(event); assertEquals(firstJob.id(), event.job().id()); } - assertEquals(Job.State.AVAILABLE, client.get(secondJob.id()).state()); - assertEquals(Job.State.AVAILABLE, client.get(thirdJob.id()).state()); + assertEquals(Job.State.AVAILABLE, client.jobs().get(secondJob.id()).state()); + assertEquals(Job.State.AVAILABLE, client.jobs().get(thirdJob.id()).state()); } } diff --git a/ruby/CHANGELOG.md b/ruby/CHANGELOG.md index 5ea54fb7f..7964a27e9 100644 --- a/ruby/CHANGELOG.md +++ b/ruby/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- Enforce the 32 MiB encoded UTF-8 JSON limit for `Job#output=`, preserving previous output when validation fails. [PR #1569](https://github.com/riverqueue/river/pull/1569). + ## [0.15.0] - 2026-10-30 ### Added diff --git a/ruby/lib/worker.rb b/ruby/lib/worker.rb index a96b28005..f4efc1e91 100644 --- a/ruby/lib/worker.rb +++ b/ruby/lib/worker.rb @@ -139,8 +139,13 @@ def method_missing(name, ...) end # Stores a worker result under the conventional +"output"+ metadata key. + # Raises ArgumentError if its encoded UTF-8 JSON exceeds 32 MiB, leaving any + # previously recorded output unchanged. def output=(value) - update_metadata("output" => value) + encoded = JSON.generate(value) + raise ArgumentError, "output is too large: #{encoded.bytesize} bytes (max 32 MiB)" if encoded.bytesize > 32 * 1024 * 1024 + + @metadata_updates["output"] = JSON.parse(encoded) end def respond_to_missing?(name, include_private = false) diff --git a/ruby/spec/client_driver_shared_examples.rb b/ruby/spec/client_driver_shared_examples.rb index 436e39936..8ff132070 100644 --- a/ruby/spec/client_driver_shared_examples.rb +++ b/ruby/spec/client_driver_shared_examples.rb @@ -2,6 +2,7 @@ require "timeout" require "riverqueue/testing" +require_relative "recorded_output_shared_examples" require_relative "runtime_draining_shared_examples" require_relative "runtime_finishing_shared_examples" require_relative "transactional_completion_shared_examples" @@ -247,6 +248,7 @@ RSpec.shared_examples "client driver end to end" do it_behaves_like "cancellation while draining" it_behaves_like "externally claimed job finalization" + it_behaves_like "recorded output size limit" it_behaves_like "transactional job completion" it "keeps an omitted cursor-only step recoverable" do received = [] diff --git a/ruby/spec/recorded_output_shared_examples.rb b/ruby/spec/recorded_output_shared_examples.rb new file mode 100644 index 000000000..f2ca4f8fe --- /dev/null +++ b/ruby/spec/recorded_output_shared_examples.rb @@ -0,0 +1,32 @@ +# frozen_string_literal: true + +require "riverqueue/testing" + +RSpec.shared_examples "recorded output size limit" do + ["x", "é", '"'].each do |character| + it "limits recorded output by encoded UTF-8 bytes for #{character.inspect}" do + limit = 32 * 1024 * 1024 + encoded_character_size = JSON.generate(character).bytesize - 2 + output_of_size = ->(bytes) do + character * ((bytes - 2) / encoded_character_size) + "x" * ((bytes - 2) % encoded_character_size) + end + at_limit = output_of_size.call(limit) + worker = Object.new + check_output = ->(job) do + job.update_metadata("kept" => true) + job.output = output_of_size.call(limit - 1) + job.output = at_limit + expect { job.output = output_of_size.call(limit + 1) } + .to raise_error(ArgumentError, "output is too large: 33554433 bytes (max 32 MiB)") + end + worker.define_singleton_method(:work) { |job| check_output.call(job) } + client = River::Client.new(@driver, config: River::Config.new(workers: River::Workers.new.add(:output, worker))) + row = client.insert(River::JobArgsHash.new(:output, {})).job + + result = River::Testing.perform_job(client, row.id) + + expect(result).to have_attributes(error: nil, outcome: :completed) + expect(client.job_get(row.id).metadata).to include("kept" => true, "output" => at_limit) + end + end +end diff --git a/ruby/spec/upstream_runtime_spec.rb b/ruby/spec/upstream_runtime_spec.rb index 222bc17fb..7d0e5db39 100644 --- a/ruby/spec/upstream_runtime_spec.rb +++ b/ruby/spec/upstream_runtime_spec.rb @@ -5,6 +5,7 @@ require "riverqueue/testing" require_relative "support/river_sqlite_schema_fixture" require_relative "row_decoding_shared_examples" +require_relative "recorded_output_shared_examples" require_relative "runtime_draining_shared_examples" require_relative "runtime_finishing_shared_examples" require_relative "transactional_completion_shared_examples" @@ -13,6 +14,7 @@ it_behaves_like "SQLite corrupt job runtime" it_behaves_like "cancellation while draining" it_behaves_like "externally claimed job finalization" + it_behaves_like "recorded output size limit" it_behaves_like "transactional job completion" before do