[java] Move worker lifecycle to Client and RunHandle

java/API.md

@@ -6,8 +6,8 @@ Java exposes every Go helper. Stored values and protocol behavior still follow G
 
 | Go surface | Java surface | Reason for the shape |
 | --- | --- | --- |
-| `NewClient`, `Config`, `Start`, `Stop`, `StopAndCancel` | `Client`, `Workers.Builder`, `start`, `stop`, `stopAndCancel` | A client can insert without starting anything. A running `Workers` owns threads and implements `AutoCloseable`. |
-| `JobArgs`, `Job[T]`, worker interfaces and registration | `JobType<A>`, `Job<A>`, `add(type, lambda)` | Records need no River interface; explicit kinds stay stable when Java classes are renamed. |
+| `NewClient`, `Config`, `Start`, `Stop`, `StopAndCancel` | `Client.builder`, `Client.start`, `RunHandle.stop/stopAndCancel` | A client can insert without starting anything. A `RunHandle` owns one execution and implements `AutoCloseable`. |
+| `JobArgs`, `Job[T]`, worker interfaces and registration | `JobType<A>`, `Job<A>`, `Workers.add(type, lambda)`, `Client.Builder.worker` | Records need no River interface; explicit kinds stay stable when Java classes are renamed. |
 | `Insert`, `InsertTx`, `InsertMany`, `InsertManyTx` | `insert`, `insertMany`, connection-first overloads | Arguments are typed; results contain persisted `Job<JsonNode>` rows, including cross-kind duplicates. Mixed batches use `JobType.submission`. |
 | `InsertOpts`, `UniqueOpts`, struct tags | `InsertOptions`, `Unique`, `JobType.uniqueBy` | Named builders and fluent options avoid positional flags. Explicit JSON paths replace struct tags. |
 | `JobGet`, `JobList`, cursor/order/filter params | `get`, `JobQuery`, `JobQuery.Page` | Raw reads use `JsonNode`; `get(id, type)` checks the kind and decodes a known argument type. Pagination remains explicit. |
@@ -19,10 +19,10 @@ Java exposes every Go helper. Stored values and protocol behavior still follow G
 | `ResumableSetStepTx`, `ResumableSetStepCursorTx` | `WorkContext.checkpoint(connection[, cursor])` | Checkpoints commit with application writes; failed transactions leave the last committed progress intact. |
 | Worker timeout | `Workers.Handler.timeout(job)` | A default method preserves lambda handlers; per-job overrides also govern rescue decisions. |
 | Retry policy | `RetryPolicy` | A lambda receives the full job and returns `Duration`; the default uses the same error-count backoff and jitter. |
-| `Queues`, queue CRUD/control APIs | `Workers.addQueue/removeQueue`, `Client.queues()` | Runtime concurrency and persisted queue state have separate owners. Reads and mutations support caller transactions. |
-| `Subscribe`, `EventKind`, `Event` | `Workers.subscribe`, `EventKind`, `Event`, `Subscription` | Enum filters and an `AutoCloseable` subscription replace channels. Callbacks are synchronous; queue events include their queue. |
-| Periodic jobs and schedules | `Workers.Builder.periodic`, `Workers.periodicJobs()`, `PeriodicJobBundle`, `PeriodicJob`, `Schedule` | Java time types and a functional scheduling interface; cron implementation details are package-private. |
-| Leadership notifications and maintenance config | `requestResign`, worker builder settings | Coordination stays in the database; operational durations use `Duration`. |
+| `Queues`, queue CRUD/control APIs | `RunHandle.addQueue/removeQueue`, `Client.queues()` | Runtime concurrency and persisted queue state have separate owners. Reads and mutations support caller transactions. |
+| `Subscribe`, `EventKind`, `Event` | `Client.subscribe`, `EventKind`, `Event`, `Subscription` | Enum filters and an `AutoCloseable` subscription replace channels. Callbacks are synchronous; queue events include their queue. |
+| Periodic jobs and schedules | `Client.Builder.periodic`, `RunHandle.periodicJobs()`, `PeriodicJobBundle`, `PeriodicJob`, `Schedule` | Java time types and a functional scheduling interface; cron implementation details are package-private. |
+| Leadership notifications and maintenance config | `requestResign`, client builder settings | Coordination stays in the database; operational durations use `Duration`. |
 | `rivermigrate` | `Migrator`, `Direction`, `Options`, executable CLI | Named options replace booleans and sentinel target versions. Down defaults to one migration; explicit target zero removes the line. |
 | Driver, pool, transaction integration | `Database`, JDBC `DataSource`/`Connection` | Use existing Java database pools. Database drivers remain optional library dependencies. |
 | Errors and row types | `RiverException.Code`, `Job.State`, nested result records | Exceptions carry failures; enums model closed sets; records keep related values together. Missing rows throw `NOT_FOUND`. |
@@ -37,7 +37,7 @@ Java exposes every Go helper. Stored values and protocol behavior still follow G
 - Added the common `Plugin` marker and `withPlugins` registration to install all
   of an implementation's hooks and middleware together. Renamed the internal Pro
   storage interface and registration from `Plugin`/`withPlugin` to `Pilot`/`withPilot`.
-- Removed no-op `Client.close()`. Close `Workers` and the application-owned pool.
+- Removed no-op `Client.close()`. Close the `RunHandle` returned by `Client.start()` and the application-owned pool.
 - Added typed retrieval and typed transactional completion, including running-state validation.
 - Unified bulk insertion around typed homogeneous lists and mixed-kind submissions,
   with both owned and caller-owned transactions.
@@ -79,8 +79,9 @@ configuration code.
 - `JobType.uniqueBy("account_id", "address")` selects top-level JSON fields without
   nested lists. The existing component-list overload handles nested paths and
   retains literal dots in field names.
-- Reusing a worker builder generates a separate runtime ID on each `start()`;
-  an explicit `id` remains the caller's responsibility to keep unique.
+- Reusing a client builder generates a separate client ID on each `build()`.
+  Restarting one client retains its ID; an explicit `id` remains the caller's
+  responsibility to keep unique.
 - Observer and subscriber failures are reported without losing a worker slot,
   changing the attempt outcome, or preventing delivery to the remaining callbacks.
   Failures in the error handler fall back to the system logger. These callbacks

java/CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- **Breaking:** Move runtime configuration to `Client.builder(database)` and startup to `Client.start()`, which returns an `AutoCloseable` `RunHandle` for `stop()`, `stopAndCancel()`, and runtime queue/periodic controls. `Workers` is now a handler registry supplied through `.workers(registry)`; `.worker(type, handler)` supports inline registration. Move subscriptions and event types to `Client`, allowing subscriptions before startup. A client allows one active run and can restart after shutdown completes. [PR #1555](https://github.com/riverqueue/river/pull/1555).
+
 ## [0.1.0] - 2026-10-26
 
 ### Changed

java/DIFFERENCES.md

@@ -19,9 +19,14 @@
 - 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.
+- `Client.start()` returns an `AutoCloseable` `RunHandle`; shutdown is on that
+  handle, as in JavaScript and Rust, instead of on `Client` as in Go and Ruby.
+  `Workers` only registers handlers. Each client permits one active run and can
+  restart after shutdown completes; an old handle cannot stop a later run.
 - Runtime events use enum kinds and are process-local callbacks. They are not a durable delivery
   mechanism. A callback should finish quickly, and callback failures are sent to
-  the runtime error handler.
+  the runtime error handler. Subscribe through `Client` before startup to avoid
+  missing early events; subscriptions survive restarts until closed.
 - Retry policies return relative `Duration` values. Exceptions and null, negative,
   or overflowing retry delays are reported and fall back to the default backoff.
 - Explicit retries that update a job send a standard insert notification in the
@@ -63,9 +68,9 @@ replace process-wide insert defaults.
 
 Insertion is independent of worker registration so a client can enqueue jobs
 owned by another language. New job kinds always use Go's validated kind syntax;
-the legacy bypass is omitted. `Workers.stop()` (also called by `close()`) and
-`Workers.stopAndCancel()` wait for attempts to finish instead of exposing a
-Go-style stopped channel. `Workers.Builder.stopTimeout` configures their waits.
+the legacy bypass is omitted. `RunHandle.stop()` (also called by `close()`) and
+`RunHandle.stopAndCancel()` wait for attempts to finish instead of exposing a
+Go-style stopped channel. `Client.Builder.stopTimeout` configures their waits.
 
 The [feature inventory](conformance/feature-inventory.json) accounts for all 304
 entries in the pinned upstream inventory, including internal driver details