diff --git a/js/CHANGELOG.md b/js/CHANGELOG.md
index a94f7c92d..9d8e2d797 100644
--- a/js/CHANGELOG.md
+++ b/js/CHANGELOG.md
@@ -11,9 +11,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Support Node.js 24 across all packages. River automatically installs a `Temporal` polyfill when the runtime has no native implementation, preserving native `Temporal` on Node.js 26.
 
+### Changed
+
+- Like River for Go, `insertMany` rejects an empty batch with a `ValidationError` instead of returning an empty result. [PR #1504](https://github.com/riverqueue/river/pull/1504).
+- Like River for Go, the client no longer caps `maxAttempts` at 32767: SQLite stores wider attempt counts, and the Postgres and Prisma drivers clamp `maxAttempts` to their 16-bit column on insert instead of failing. [PR #1504](https://github.com/riverqueue/river/pull/1504).
+- Like River for Go's `ResumableStep`, a failed resumable step records the step's own error as the attempt's error instead of wrapping it in a `LifecycleError`. [PR #1504](https://github.com/riverqueue/river/pull/1504).
+
 ### Fixed
 
 - Worker threads respect cancellation received while loading a handler module, before invoking its handler.
+- Like River for Go, a leader whose renewal finds its term gone, because it expired or another process replaced it, gives up leadership at once instead of running maintenance until its local deadline passes. [PR #1504](https://github.com/riverqueue/river/pull/1504).
+- `updateQueue` raising a full queue's `maxWorkers` starts jobs at once instead of waiting for a running job to finish. [PR #1504](https://github.com/riverqueue/river/pull/1504).
 
 ## [0.2.0] - 2026-10-07
 
diff --git a/js/docs/README.md b/js/docs/README.md
index d392f3c7e..a18ab0553 100644
--- a/js/docs/README.md
+++ b/js/docs/README.md
@@ -204,7 +204,7 @@ match.
 ### Batches
 
 `insertMany` inserts a batch atomically and returns results in input order;
-an empty batch returns immediately. Items may use different definitions, and
+an empty batch is rejected. Items may use different definitions, and
 each item's `args` is checked against its own definition:
 
 <!-- ts-setup
diff --git a/js/driver/pg/src/driver.integration.test.ts b/js/driver/pg/src/driver.integration.test.ts
index 194549a1f..25ecd0f93 100644
--- a/js/driver/pg/src/driver.integration.test.ts
+++ b/js/driver/pg/src/driver.integration.test.ts
@@ -145,6 +145,17 @@ describe("PgDriver integration", () => {
     }
   });
 
+  it("clamps a wide maxAttempts to smallint like Go", async () => {
+    const inserted = await driver.jobInsertMany([
+      insertParams(`${filePrefix}_clamped`, { maxAttempts: 40_000 }),
+    ]);
+
+    expect(inserted[0]!.job.maxAttempts).toBe(32_767);
+    expect((await driver.jobGet(inserted[0]!.job.id))?.maxAttempts).toBe(
+      32_767
+    );
+  });
+
   it("truncates sub-microsecond timestamps like Go instead of rounding", async () => {
     // Postgres would round `.0000019` up to `.000002`; Go's pgx truncates.
     const scheduledAt = Temporal.Instant.from("2026-08-30T12:00:00.0000019Z");
diff --git a/js/driver/pg/src/sql/jobs.ts b/js/driver/pg/src/sql/jobs.ts
index 7d0c9aa86..5727ebe6a 100644
--- a/js/driver/pg/src/sql/jobs.ts
+++ b/js/driver/pg/src/sql/jobs.ts
@@ -733,7 +733,7 @@ export async function jobInsertMany(
     [
       params.map(({ encodedArgs }) => encodedArgs),
       params.map(({ kind }) => kind),
-      params.map(({ maxAttempts }) => maxAttempts),
+      params.map(({ maxAttempts }) => clampSmallint(maxAttempts)),
       params.map(({ metadata }, index) =>
         JSON.stringify(
           nonces === null
@@ -741,7 +741,7 @@ export async function jobInsertMany(
             : { ...metadata, [UNIQUE_INSERT_NONCE_KEY]: nonces[index] }
         )
       ),
-      params.map(({ priority }) => priority),
+      params.map(({ priority }) => clampSmallint(priority)),
       params.map(({ queue }) => queue),
       params.map(({ scheduledAt }) => instantParameter(scheduledAt)),
       params.map(({ state }) => state),
@@ -1011,3 +1011,12 @@ function validateCompletionCommands(
     }
   }
 }
+
+/**
+ * Clamps an integer to Postgres's 16-bit `smallint` columns. Like River
+ * for Go's Postgres drivers, an insert stores a wider `max_attempts` or
+ * `priority` as 32767 instead of failing; SQLite stores the full value.
+ */
+function clampSmallint(value: number): number {
+  return Math.min(value, 32_767);
+}
diff --git a/js/driver/prisma/src/driver.integration.test.ts b/js/driver/prisma/src/driver.integration.test.ts
index 05bc0e699..52d1eeba5 100644
--- a/js/driver/prisma/src/driver.integration.test.ts
+++ b/js/driver/prisma/src/driver.integration.test.ts
@@ -120,6 +120,16 @@ describe("PrismaDriver integration", () => {
     expect(result.job.state).toBe("scheduled");
   });
 
+  it("clamps a wide maxAttempts to Postgres's smallint like Go", async () => {
+    const result = await client.insert(
+      job("clamped"),
+      {},
+      { maxAttempts: 40_000 }
+    );
+
+    expect(result.job.maxAttempts).toBe(32_767);
+  });
+
   it("preserves order for heterogeneous batches", async () => {
     const a = job("batch_a");
     const b = job("batch_b");
diff --git a/js/driver/prisma/src/driver.ts b/js/driver/prisma/src/driver.ts
index d2fdd9dc7..c0c66f664 100644
--- a/js/driver/prisma/src/driver.ts
+++ b/js/driver/prisma/src/driver.ts
@@ -317,7 +317,7 @@ export class PrismaInserter implements InsertDriver<PrismaClientLike> {
       sql,
       params.map(({ encodedArgs }) => encodedArgs),
       params.map(({ kind }) => kind),
-      params.map(({ maxAttempts }) => maxAttempts),
+      params.map(({ maxAttempts }) => clampSmallint(maxAttempts)),
       params.map(({ metadata }, index) =>
         JSON.stringify(
           nonces === null
@@ -325,7 +325,7 @@ export class PrismaInserter implements InsertDriver<PrismaClientLike> {
             : { ...metadata, [UNIQUE_INSERT_NONCE_KEY]: nonces[index] }
         )
       ),
-      params.map(({ priority }) => priority),
+      params.map(({ priority }) => clampSmallint(priority)),
       params.map(({ queue }) => queue),
       params.map(({ scheduledAt }) =>
         scheduledAt === undefined ? null : postgresTimestamp(scheduledAt)
@@ -543,3 +543,12 @@ function validateSchema(value: string): void {
     );
   }
 }
+
+/**
+ * Clamps an integer to Postgres's 16-bit `smallint` columns. Like River
+ * for Go's Postgres drivers, an insert stores a wider `max_attempts` or
+ * `priority` as 32767 instead of failing; SQLite stores the full value.
+ */
+function clampSmallint(value: number): number {
+  return Math.min(value, 32_767);
+}
diff --git a/js/driver/sqlite/src/driver.test.ts b/js/driver/sqlite/src/driver.test.ts
index 101ae1f8e..3506895e0 100644
--- a/js/driver/sqlite/src/driver.test.ts
+++ b/js/driver/sqlite/src/driver.test.ts
@@ -122,6 +122,25 @@ describe("SqliteDriver", () => {
     ).toEqual({ count: 0 });
   });
 
+  test("stores attempt counts wider than 16 bits, like Go", async () => {
+    const { driver } = await setup();
+    const job = defineJob()({ kind: "wide_attempts" });
+
+    const [{ job: inserted }] = await new Client(driver).insertMany([
+      { args: {}, job, options: { maxAttempts: 40_000 } },
+    ]);
+    expect(inserted.maxAttempts).toBe(40_000);
+    expect((await driver.jobGet(inserted.id))?.maxAttempts).toBe(40_000);
+
+    const { job: full } = await driver.jobInsert({
+      args: {},
+      attempt: 40_000,
+      kind: "wide_attempts",
+      maxAttempts: 40_001,
+    });
+    expect([full.attempt, full.maxAttempts]).toEqual([40_000, 40_001]);
+  });
+
   test("round-trips exact JSON numbers from JavaScript and other engines", async () => {
     const { database, driver } = await setup();
     const inserted = await driver.jobInsert({
diff --git a/js/driver/sqlite/src/driver.ts b/js/driver/sqlite/src/driver.ts
index b562ad04d..477a0028b 100644
--- a/js/driver/sqlite/src/driver.ts
+++ b/js/driver/sqlite/src/driver.ts
@@ -1818,7 +1818,12 @@ export class SqliteRuntime implements RuntimeDriver<
 
     const errors = (params.errors ?? []).map((error) => ({
       at: error.at.toString(),
-      attempt: validateSmallInteger(error.attempt, "error.attempt", 0, 32_767),
+      attempt: validateSmallInteger(
+        error.attempt,
+        "error.attempt",
+        0,
+        Number.MAX_SAFE_INTEGER
+      ),
       error: error.error,
       trace: error.trace,
     }));
@@ -1828,7 +1833,14 @@ export class SqliteRuntime implements RuntimeDriver<
         params.encodedArgs === undefined
           ? encodeJson(params.args, "args")
           : encodeEncodedJson(params.encodedArgs, "encodedArgs"),
-      attempt: validateSmallInteger(params.attempt ?? 0, "attempt", 0, 32_767),
+      // Like River for Go, store attempt counts SQLite's 64-bit integers
+      // hold, wider than Postgres's 16-bit columns.
+      attempt: validateSmallInteger(
+        params.attempt ?? 0,
+        "attempt",
+        0,
+        Number.MAX_SAFE_INTEGER
+      ),
       attemptedAt: sqliteTimestampOrNull(params.attemptedAt),
       // Go writes NULL for absent client IDs and errors, not an empty array.
       attemptedBy:
@@ -1844,7 +1856,7 @@ export class SqliteRuntime implements RuntimeDriver<
         params.maxAttempts ?? 25,
         "maxAttempts",
         1,
-        32_767
+        Number.MAX_SAFE_INTEGER
       ),
       metadata: encodeJson(metadata, "metadata"),
       priority: validateSmallInteger(params.priority ?? 1, "priority", 1, 4),
diff --git a/js/etc/riverqueue.api.md b/js/etc/riverqueue.api.md
index 92e2cf195..a74fa5458 100644
--- a/js/etc/riverqueue.api.md
+++ b/js/etc/riverqueue.api.md
@@ -869,7 +869,7 @@ export interface InsertClient<Transaction = unknown> {
   ): Promise<InsertResult<JobDefinitionInput<Definition>>>;
   /**
    * Insert a heterogeneous batch atomically, preserving input order in the
-   * result tuple. An empty batch resolves to `[]` without a database call.
+   * result tuple. An empty batch is rejected with a {@link ValidationError}.
    */
   insertMany<const Items extends readonly InsertManyItem[]>(
     items: Items & CheckedInsertManyItems<Items>,
diff --git a/js/src/client.test.ts b/js/src/client.test.ts
index ecf8d9557..702325946 100644
--- a/js/src/client.test.ts
+++ b/js/src/client.test.ts
@@ -279,7 +279,7 @@ describe("Client", () => {
 
     await expect(
       clientWithDefaults.insert(definition, { strings: [] }, { maxAttempts: 0 })
-    ).rejects.toThrow("maxAttempts must be a safe integer between 1 and 32767");
+    ).rejects.toThrow("maxAttempts must be a safe integer at least 1");
     await expect(
       clientWithDefaults.insert(definition, { strings: [] }, { queue: "" })
     ).rejects.toThrow("queue name must not be empty");
@@ -407,6 +407,12 @@ describe("Client", () => {
     expect(driver.insertedParams).toHaveLength(0);
   });
 
+  it("rejects an empty batch, like Go", async () => {
+    await expect(client.insertMany([])).rejects.toThrow(
+      new ValidationError("no jobs to insert")
+    );
+  });
+
   it("preserves each definition input across heterogeneous batches", async () => {
     const countJob = defineJob<{ count: number }>()({ kind: "count" });
     const results = await client.insertMany([
diff --git a/js/src/client.ts b/js/src/client.ts
index f74c2cdce..b0faf5c3d 100644
--- a/js/src/client.ts
+++ b/js/src/client.ts
@@ -222,7 +222,7 @@ export interface InsertClient<Transaction = unknown> {
 
   /**
    * Insert a heterogeneous batch atomically, preserving input order in the
-   * result tuple. An empty batch resolves to `[]` without a database call.
+   * result tuple. An empty batch is rejected with a {@link ValidationError}.
    */
   insertMany<const Items extends readonly InsertManyItem[]>(
     items: Items & CheckedInsertManyItems<Items>,
@@ -655,7 +655,7 @@ export class RiverClient<Transaction = unknown> implements Client<Transaction> {
     items: Items & CheckedInsertManyItems<Items>,
     options: TransactionOptions<Transaction> = {}
   ): Promise<InsertManyResults<Items>> {
-    if (items.length === 0) return [] as unknown as InsertManyResults<Items>;
+    if (items.length === 0) throw new ValidationError("no jobs to insert");
     return this.#operationScope(options.tx, async (tx) => {
       const allParams: JobInsertParams[] = [];
       for (const item of items) {
diff --git a/js/src/conformance.test.ts b/js/src/conformance.test.ts
index 2ee61a886..51c226f61 100644
--- a/js/src/conformance.test.ts
+++ b/js/src/conformance.test.ts
@@ -249,7 +249,7 @@ describe("Go protocol fixtures", () => {
         resumable.setCursor({ offset: 3 });
         throw new Error("retry");
       })
-    ).rejects.toThrow("failed");
+    ).rejects.toThrow("retry");
     expect(resumable.finish(true).metadata).toEqual({
       [keys.resumable_cursor]: { process: { offset: 3 } },
       [keys.resumable_step]: "process",
diff --git a/js/src/insert-options.ts b/js/src/insert-options.ts
index 930c547df..3b07bd44d 100644
--- a/js/src/insert-options.ts
+++ b/js/src/insert-options.ts
@@ -119,7 +119,9 @@ export function normalizeInsertOptions(
   const copy: NormalizedInsertOptions = {};
 
   if (options.maxAttempts !== undefined) {
-    requireInteger("maxAttempts", options.maxAttempts, 1, 32_767);
+    // Like River for Go, no 16-bit ceiling: SQLite stores the full value,
+    // and Postgres drivers clamp it to their `smallint` column.
+    requireInteger("maxAttempts", options.maxAttempts, 1);
     copy.maxAttempts = options.maxAttempts;
   }
   if (options.metadata !== undefined) {
diff --git a/js/src/resumable.test.ts b/js/src/resumable.test.ts
index e8ad51c62..efbc21c83 100644
--- a/js/src/resumable.test.ts
+++ b/js/src/resumable.test.ts
@@ -48,7 +48,7 @@ describe("Resumable", () => {
     });
   });
 
-  it("retains progress and the original cause when a step error is caught", async () => {
+  it("retains progress and the step's own error when it is caught", async () => {
     const resumable = new Resumable(client(), job({}));
     const cause = new Error("service unavailable");
     await resumable.step("first", () => undefined);
@@ -58,7 +58,7 @@ describe("Resumable", () => {
       })
       .catch(() => undefined);
     const finished = resumable.finish(false);
-    expect(finished.error?.cause).toBe(cause);
+    expect(finished.error).toBe(cause);
     expect(finished.metadata).toEqual({ "river:resumable_step": "first" });
   });
 
@@ -100,7 +100,7 @@ describe("Resumable", () => {
       resumable.step("after", () => {
         throw new Error("retry");
       })
-    ).rejects.toThrow('resumable step "after" failed');
+    ).rejects.toThrow(new Error("retry"));
 
     expect(visited).toEqual(["process"]);
     expect(resumable.finish(true).metadata).toEqual({
diff --git a/js/src/resumable.ts b/js/src/resumable.ts
index 2a9391356..df0bebae1 100644
--- a/js/src/resumable.ts
+++ b/js/src/resumable.ts
@@ -79,7 +79,7 @@ export class Resumable {
       this.#completedStep = name;
     } catch (cause: unknown) {
       this.#failure = stepError(name, cause);
-      throw this.#failure;
+      throw cause;
     } finally {
       this.#stepName = previousStepName;
     }
@@ -98,7 +98,7 @@ export class Resumable {
       this.#cursors.delete(name);
     } catch (cause: unknown) {
       this.#failure = stepError(name, cause);
-      throw this.#failure;
+      throw cause;
     } finally {
       this.#stepName = previousStepName;
     }
@@ -204,7 +204,12 @@ export function finishResumable(
   return resumable.finish(workerFailed);
 }
 
+/**
+ * A failed step's error, which fails the attempt. Like River for Go, it's the
+ * step's own error, so the job records the step's message.
+ */
 function stepError(name: string, cause: unknown): Error {
+  if (cause instanceof Error) return cause;
   return new LifecycleError(`resumable step ${JSON.stringify(name)} failed`, {
     cause,
   });
diff --git a/js/src/runtime/pilot-operations.ts b/js/src/runtime/pilot-operations.ts
index 8cb886bb5..d338671ad 100644
--- a/js/src/runtime/pilot-operations.ts
+++ b/js/src/runtime/pilot-operations.ts
@@ -928,8 +928,8 @@ function validatePreparedRow(
   if (typeof row.queue !== "string" || row.queue.length === 0) {
     fail("queue", "a non-empty string");
   }
-  if (!isIntegerBetween(row.maxAttempts, 1, 32_767)) {
-    fail("maxAttempts", "an integer from 1 to 32767");
+  if (!isIntegerBetween(row.maxAttempts, 1, Number.MAX_SAFE_INTEGER)) {
+    fail("maxAttempts", "a positive safe integer");
   }
   if (!isIntegerBetween(row.priority, 1, 4)) {
     fail("priority", "an integer from 1 to 4");
diff --git a/js/src/runtime/queue-producer.ts b/js/src/runtime/queue-producer.ts
index 4bf0f84af..e96436b90 100644
--- a/js/src/runtime/queue-producer.ts
+++ b/js/src/runtime/queue-producer.ts
@@ -458,7 +458,7 @@ export class QueueProducer {
         // Like River for Go's producer after a full fetch, claim again as
         // soon as any worker slot frees, subject to the fetch cooldown.
         if (databaseLikelyHasMore && capacity <= 0 && active.size > 0) {
-          await Promise.race(active);
+          await this.#waitForSlot(runtime, active);
           continue;
         }
         if (!runtime.paused && capacity > 0) {
@@ -587,7 +587,7 @@ export class QueueProducer {
 
         databaseLikelyHasMore = false;
         if (active.size >= runtime.config.maxWorkers) {
-          await Promise.race(active);
+          await this.#waitForSlot(runtime, active);
         } else {
           await this.#waitForQueue(runtime, signal);
         }
@@ -793,6 +793,26 @@ export class QueueProducer {
     });
   }
 
+  /**
+   * Wait for a worker slot to free, or for an explicit wake-up, such as a
+   * reconfiguration that raised the queue's capacity or a stop.
+   */
+  async #waitForSlot(
+    runtime: QueueRuntime,
+    active: ReadonlySet<Promise<unknown>>
+  ): Promise<void> {
+    const woken = Promise.withResolvers<undefined>();
+    const wake = () => {
+      woken.resolve(undefined);
+    };
+    runtime.wake = wake;
+    try {
+      await Promise.race([...active, woken.promise]);
+    } finally {
+      if (runtime.wake === wake) runtime.wake = null;
+    }
+  }
+
   /** Wait for the queue control poll interval or an explicit wake-up. */
   #waitForQueueControl(signal: AbortSignal): Promise<void> {
     if (signal.aborted) return Promise.reject(signal.reason);
diff --git a/js/src/services.ts b/js/src/services.ts
index 664608d88..ff798b4a4 100644
--- a/js/src/services.ts
+++ b/js/src/services.ts
@@ -440,9 +440,11 @@ export class RuntimeServices {
       );
       this.#lastError = null;
       if (leader === null || leader === undefined) {
-        if (this.#leader !== null && !this.#hasTrustedLeadership()) {
-          await this.#loseLeadership(false);
-        }
+        // Like River for Go's elector, a held term that can't be renewed
+        // (it expired, or another process replaced it, even under this
+        // client's ID) ends at once, without resigning a term this client
+        // no longer holds.
+        if (this.#leader !== null) await this.#loseLeadership(false);
         return true;
       }
       const expired = this.#expiredTerm;
diff --git a/rust/CHANGELOG.md b/rust/CHANGELOG.md
index 37491941a..161da85e0 100644
--- a/rust/CHANGELOG.md
+++ b/rust/CHANGELOG.md
@@ -10,6 +10,14 @@ Changes to River for Go are recorded in the [repository changelog](../CHANGELOG.
 
 ## [Unreleased]
 
+### Added
+
+- A worker cancels its job with a reason by returning `JobCancelError`, like River for Go's `JobCancel(err)`. Anywhere in the worker error's source chain, it cancels the job whatever attempts it has left, skips the error handler, and records `JobCancelError: <reason>` on the attempt as Go does. `WorkOutcome::Cancel` still cancels without a reason. [PR #1504](https://github.com/riverqueue/river/pull/1504).
+
+### Changed
+
+- **Breaking:** Attempt counts are now `i32` instead of `i16`, like River for Go's `int`: `JobRow::attempt` and `max_attempts`, `AttemptError::attempt` and `AttemptError::new`, `InsertOpts::max_attempts` and `with_max_attempts`, `InsertParams::max_attempts`, `ClientBuilder::default_max_attempts`, `MAX_ATTEMPTS_DEFAULT`, and `riverqueue-test`'s builders. On SQLite, which stores native integers, a job with more than 32,767 maximum attempts inserts and reads back unchanged, as in Go. On Postgres, whose columns are 16-bit, an insert clamps `max_attempts` to 32,767 like Go's drivers instead of failing. [PR #1504](https://github.com/riverqueue/river/pull/1504).
+
 ## [0.3.0] - 2026-10-07
 
 ### Changed
diff --git a/rust/riverqueue-macros/src/lib.rs b/rust/riverqueue-macros/src/lib.rs
index f09ec0bab..a16bfe1d3 100644
--- a/rust/riverqueue-macros/src/lib.rs
+++ b/rust/riverqueue-macros/src/lib.rs
@@ -318,11 +318,11 @@ fn expand_job_args(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStream>
         ));
     }
     if let Some(max_attempts) = &attributes.max_attempts
-        && !(1..=i16::MAX as u64).contains(&max_attempts.base10_parse::<u64>()?)
+        && !(1..=i32::MAX as u64).contains(&max_attempts.base10_parse::<u64>()?)
     {
         return Err(syn::Error::new_spanned(
             max_attempts,
-            "max_attempts must be between 1 and 32767",
+            "max_attempts must be between 1 and 2147483647",
         ));
     }
     if let Some(priority) = &attributes.priority
diff --git a/rust/riverqueue-test/src/lib.rs b/rust/riverqueue-test/src/lib.rs
index 836786796..e86199131 100644
--- a/rust/riverqueue-test/src/lib.rs
+++ b/rust/riverqueue-test/src/lib.rs
@@ -23,7 +23,7 @@ pub use require::{
 #[derive(Debug)]
 pub struct TestJobBuilder<A: JobArgs> {
     args: A,
-    attempt: i16,
+    attempt: i32,
     id: i64,
     metadata: Map<String, serde_json::Value>,
     state: JobState,
@@ -44,7 +44,7 @@ impl<A: JobArgs> TestJobBuilder<A> {
 
     /// Sets the current attempt.
     #[must_use]
-    pub const fn attempt(mut self, attempt: i16) -> Self {
+    pub const fn attempt(mut self, attempt: i32) -> Self {
         self.attempt = attempt;
         self
     }
diff --git a/rust/riverqueue-test/src/require.rs b/rust/riverqueue-test/src/require.rs
index fa43ebdfc..a67fd5ecc 100644
--- a/rust/riverqueue-test/src/require.rs
+++ b/rust/riverqueue-test/src/require.rs
@@ -24,7 +24,7 @@ const LIST_LIMIT: u32 = 10_000;
 #[derive(Clone, Debug, Default)]
 #[non_exhaustive]
 pub struct RequireInsertedOpts {
-    max_attempts: Option<i16>,
+    max_attempts: Option<i32>,
     priority: Option<i16>,
     queue: Option<String>,
     scheduled_at: Option<DateTime<Utc>>,
@@ -41,7 +41,7 @@ impl RequireInsertedOpts {
 
     /// Returns the expected maximum number of attempts.
     #[must_use]
-    pub const fn max_attempts(&self) -> Option<i16> {
+    pub const fn max_attempts(&self) -> Option<i32> {
         self.max_attempts
     }
 
@@ -77,7 +77,7 @@ impl RequireInsertedOpts {
 
     /// Expects this maximum number of attempts.
     #[must_use]
-    pub const fn with_max_attempts(mut self, max_attempts: i16) -> Self {
+    pub const fn with_max_attempts(mut self, max_attempts: i32) -> Self {
         self.max_attempts = Some(max_attempts);
         self
     }
diff --git a/rust/riverqueue/src/__private.rs b/rust/riverqueue/src/__private.rs
index cb6a3aac7..50d18e6ac 100644
--- a/rust/riverqueue/src/__private.rs
+++ b/rust/riverqueue/src/__private.rs
@@ -1288,7 +1288,7 @@ pub struct PreparedInsertParams {
     /// Stable job kind.
     pub kind: String,
     /// Maximum attempts, including the first.
-    pub max_attempts: i16,
+    pub max_attempts: i32,
     /// Arbitrary job metadata.
     pub metadata: crate::JobMetadata,
     /// Priority from one through four.
@@ -1465,7 +1465,7 @@ impl RawInsertResult {
 #[derive(Debug)]
 pub struct JobRowParts {
     pub id: i64,
-    pub attempt: i16,
+    pub attempt: i32,
     pub attempted_at: Option<DateTime<Utc>>,
     pub attempted_by: Vec<String>,
     pub created_at: DateTime<Utc>,
@@ -1473,7 +1473,7 @@ pub struct JobRowParts {
     pub errors: Vec<AttemptError>,
     pub finalized_at: Option<DateTime<Utc>>,
     pub kind: String,
-    pub max_attempts: i16,
+    pub max_attempts: i32,
     pub metadata: crate::JobMetadata,
     pub priority: i16,
     pub queue: String,
diff --git a/rust/riverqueue/src/client/builder.rs b/rust/riverqueue/src/client/builder.rs
index 8a4472503..e26b1f668 100644
--- a/rust/riverqueue/src/client/builder.rs
+++ b/rust/riverqueue/src/client/builder.rs
@@ -343,7 +343,7 @@ pub struct ClientBuilder {
     pub(super) allow_legacy_job_kinds: bool,
     pub(super) allow_unregistered_job_kinds: bool,
     pub(super) database: Database,
-    pub(super) default_max_attempts: i16,
+    pub(super) default_max_attempts: i32,
     pub(super) error_handler: Option<Arc<dyn crate::extension::DynErrorHandler>>,
     pub(super) fetch_cooldown: Duration,
     pub(super) fetch_only_known_kinds: bool,
@@ -403,7 +403,7 @@ impl ClientBuilder {
     /// [`MAX_ATTEMPTS_DEFAULT`](crate::MAX_ATTEMPTS_DEFAULT) (25), and must be
     /// at least one.
     #[must_use]
-    pub fn default_max_attempts(mut self, maximum: i16) -> Self {
+    pub fn default_max_attempts(mut self, maximum: i32) -> Self {
         self.default_max_attempts = maximum;
         self
     }
diff --git a/rust/riverqueue/src/client/completer.rs b/rust/riverqueue/src/client/completer.rs
index 14e581cc4..5bb3ec2e2 100644
--- a/rust/riverqueue/src/client/completer.rs
+++ b/rust/riverqueue/src/client/completer.rs
@@ -60,7 +60,7 @@ pub(super) const HOT_OPERATION_TIMEOUT: Duration = Duration::from_secs(10);
 pub(super) struct CompletionUpdate {
     /// Replacement attempt, sent only for snoozes and shutdown interrupts,
     /// which return an attempt that should not count.
-    pub(super) attempt: Option<i16>,
+    pub(super) attempt: Option<i32>,
     pub(super) cancellation: CancellationToken,
     pub(super) error: Option<AttemptError>,
     pub(super) event_kind: JobEventKind,
diff --git a/rust/riverqueue/src/client/executor.rs b/rust/riverqueue/src/client/executor.rs
index 07520ee97..efe2be2ba 100644
--- a/rust/riverqueue/src/client/executor.rs
+++ b/rust/riverqueue/src/client/executor.rs
@@ -16,9 +16,9 @@ use crate::client::{ClientInner, PeerLedger, peers};
 use crate::error::{Chain, panic_message};
 use crate::extension::{WorkEndpoint, WorkNext};
 use crate::{
-    AttemptError, BoxError, Client, Error, ErrorHandlerDecision, JobEventKind, JobMetadata, JobRow,
-    JobState, PanicError, WorkCancelled, WorkContext, WorkError, WorkOutcome, WorkResult,
-    WorkerTimeout,
+    AttemptError, BoxError, Client, Error, ErrorHandlerDecision, JobCancelError, JobEventKind,
+    JobMetadata, JobRow, JobState, PanicError, WorkCancelled, WorkContext, WorkError, WorkOutcome,
+    WorkResult, WorkerTimeout,
 };
 
 /// Runs one claimed job's attempt and persists its result.
@@ -426,9 +426,15 @@ pub(super) fn worker_join_result(
 ) -> WorkerResult {
     match result {
         Ok(Ok(outcome)) => Ok(outcome),
+        // Like Go's `JobCancel`, a cancellation the worker returns cancels
+        // the job and records its reason.
         Ok(Err(worker_error)) => Err(WorkerFailure {
             error: Chain(&worker_error).to_string(),
-            kind: WorkerFailureKind::Error,
+            kind: if JobCancelError::is_in_chain(worker_error.get_ref()) {
+                WorkerFailureKind::Cancelled
+            } else {
+                WorkerFailureKind::Error
+            },
             source: Some(worker_error),
             trace: String::new(),
         }),
diff --git a/rust/riverqueue/src/client/insert.rs b/rust/riverqueue/src/client/insert.rs
index faf8b8bbb..81c136fb5 100644
--- a/rust/riverqueue/src/client/insert.rs
+++ b/rust/riverqueue/src/client/insert.rs
@@ -876,7 +876,9 @@ impl Client {
                     .bind(Json(&encoded_args))
                     .bind(created_at)
                     .bind(&kind)
-                    .bind(opts.max_attempts)
+                    // Like River Go's Postgres drivers, a value wider than
+                    // the `smallint` column is clamped rather than failing.
+                    .bind(opts.max_attempts.min(i32::from(i16::MAX)))
                     .bind(Json(&*metadata))
                     .bind(opts.priority)
                     .bind(&opts.queue)
diff --git a/rust/riverqueue/src/client/mod.rs b/rust/riverqueue/src/client/mod.rs
index 2b987349d..68b517853 100644
--- a/rust/riverqueue/src/client/mod.rs
+++ b/rust/riverqueue/src/client/mod.rs
@@ -38,7 +38,7 @@ pub use self::queues::{
 };
 #[cfg(feature = "sqlite")]
 pub(crate) use self::record::FieldErrors;
-pub(crate) use self::record::{DecodedJob, UndecodableJob, saturating_i16, tolerant_row};
+pub(crate) use self::record::{DecodedJob, UndecodableJob, tolerant_row};
 #[cfg(feature = "postgres")]
 pub(crate) use self::record::{JobRecord, decode_job_row, job_projection};
 pub use self::resign::ResignRequest;
@@ -105,7 +105,7 @@ pub(crate) struct ClientInner {
     allow_unregistered_job_kinds: bool,
     completion_sender: Mutex<Option<mpsc::WeakSender<CompletionUpdate>>>,
     pub(crate) database: Database,
-    default_max_attempts: i16,
+    default_max_attempts: i32,
     error_handler: Option<Arc<dyn crate::extension::DynErrorHandler>>,
     pub(crate) events: broadcast::Sender<Event>,
     /// Minimum delay between fetches for queues without their own, and the
@@ -429,7 +429,7 @@ impl Client {
 }
 
 impl Client {
-    pub(crate) fn default_max_attempts(&self) -> i16 {
+    pub(crate) fn default_max_attempts(&self) -> i32 {
         self.inner.default_max_attempts
     }
 
diff --git a/rust/riverqueue/src/client/record.rs b/rust/riverqueue/src/client/record.rs
index e523c2a23..5835b6384 100644
--- a/rust/riverqueue/src/client/record.rs
+++ b/rust/riverqueue/src/client/record.rs
@@ -109,16 +109,6 @@ impl FieldErrors {
     }
 }
 
-/// Converts a persisted integer to `i16`, saturating at the type bounds.
-///
-/// River Go stores `attempt`, `max_attempts`, and `priority` as native
-/// integers on SQLite. Values beyond `i16` are only reachable through
-/// `max_attempts` in practice; saturating keeps such a job workable with
-/// identical retry decisions until its 32,767th attempt.
-pub(crate) fn saturating_i16(value: i64) -> i16 {
-    i16::try_from(value).unwrap_or(if value < 0 { i16::MIN } else { i16::MAX })
-}
-
 /// A Postgres job row. Columns the database constrains decode strictly, while
 /// those that can hold values River can't represent are kept as their decode
 /// results.
@@ -228,7 +218,7 @@ impl JobRecord {
                 .transpose(),
         );
         errors.finish(JobRow {
-            attempt: self.attempt,
+            attempt: self.attempt.into(),
             attempted_at: self.attempted_at,
             attempted_by: attempted_by.unwrap_or_default(),
             created_at: self.created_at,
@@ -237,7 +227,7 @@ impl JobRecord {
             finalized_at: self.finalized_at,
             id: self.id,
             kind: self.kind,
-            max_attempts: self.max_attempts,
+            max_attempts: self.max_attempts.into(),
             metadata,
             priority: self.priority,
             queue: self.queue,
@@ -273,15 +263,3 @@ pub(crate) fn job_projection(alias: &str) -> String {
          {alias}.tags::text[] AS tags, {alias}.unique_key, {alias}.unique_states::text AS unique_states"
     )
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn integers_saturate_to_i16() {
-        assert_eq!(saturating_i16(25), 25);
-        assert_eq!(saturating_i16(40_000), i16::MAX);
-        assert_eq!(saturating_i16(-40_000), i16::MIN);
-    }
-}
diff --git a/rust/riverqueue/src/client/tests.rs b/rust/riverqueue/src/client/tests.rs
index 3c769ffe0..d52f83f76 100644
--- a/rust/riverqueue/src/client/tests.rs
+++ b/rust/riverqueue/src/client/tests.rs
@@ -12,7 +12,7 @@ use super::completer::{persisted_completion_event_kind, with_completion_retries}
 use super::executor::scheduled_after;
 use super::notifier::dispatch_notification;
 use super::*;
-use crate::{AttemptError, JobEventKind, JobRow, JobState, WorkError, WorkResult};
+use crate::{AttemptError, JobCancelError, JobEventKind, JobRow, JobState, WorkError, WorkResult};
 #[cfg(feature = "sqlite")]
 use crate::{InsertOpts, InsertParams, WorkContext, WorkOutcome};
 #[cfg(feature = "sqlite")]
@@ -284,7 +284,7 @@ fn retry_row(error_count: usize) -> JobRow {
         .unwrap()
         .with_timezone(&Utc);
     JobRow {
-        attempt: i16::try_from(error_count).unwrap_or(i16::MAX),
+        attempt: i32::try_from(error_count).unwrap_or(i32::MAX),
         attempted_at: Some(now),
         attempted_by: vec!["test".to_owned()],
         created_at: now,
@@ -358,6 +358,41 @@ async fn worker_failures_record_the_error_chain_and_panic_value() {
     assert_eq!(panic.to_string(), "worker panicked: boom");
 }
 
+#[test]
+fn worker_job_cancel_errors_cancel_with_their_reason_like_go() {
+    // Like Go's `JobCancel(err)`, the attempt records the reason, and the
+    // failure cancels the job instead of being retried.
+    let failure = super::executor::worker_join_result(Ok(Err(WorkError::new(
+        JobCancelError::new("card expired"),
+    ))))
+    .unwrap_err();
+    assert_eq!(failure.error, "JobCancelError: card expired");
+    assert!(matches!(
+        failure.kind,
+        super::executor::WorkerFailureKind::Cancelled
+    ));
+
+    // It's recognized anywhere in the error's source chain, including
+    // inside an `io::Error`.
+    let failure = super::executor::worker_join_result(Ok(Err(WorkError::new(
+        std::io::Error::other(JobCancelError::new("card expired")),
+    ))))
+    .unwrap_err();
+    assert!(matches!(
+        failure.kind,
+        super::executor::WorkerFailureKind::Cancelled
+    ));
+
+    let failure = super::executor::worker_join_result(Ok(Err(WorkError::new(
+        std::io::Error::other("card declined"),
+    ))))
+    .unwrap_err();
+    assert!(matches!(
+        failure.kind,
+        super::executor::WorkerFailureKind::Error
+    ));
+}
+
 #[cfg(feature = "sqlite")]
 #[tokio::test]
 async fn extension_client_finds_only_the_installed_pilot() {
diff --git a/rust/riverqueue/src/database/sqlite.rs b/rust/riverqueue/src/database/sqlite.rs
index c6d67739b..4e8b26e21 100644
--- a/rust/riverqueue/src/database/sqlite.rs
+++ b/rust/riverqueue/src/database/sqlite.rs
@@ -27,7 +27,7 @@ use sqlx::sqlite::SqliteRow;
 
 use crate::{
     AttemptError, JobMetadata, JobRow, JobState, METADATA_KEY_UNIQUE_NONCE, Queue,
-    client::{DecodedJob, FieldErrors, UndecodableJob, go_time_json, saturating_i16, tolerant_row},
+    client::{DecodedJob, FieldErrors, UndecodableJob, go_time_json, tolerant_row},
     query::{JobListKeyset, JobListSqlPart},
 };
 
@@ -139,7 +139,7 @@ impl From<BackendError> for sqlx::Error {
 pub(crate) struct InsertJob<'a> {
     pub attempted_at: Option<DateTime<Utc>>,
     pub attempted_by: &'a [String],
-    pub attempt: i16,
+    pub attempt: i32,
     /// When unset, SQLite's `datetime('now', 'subsec')`, like Go's driver.
     pub created_at: Option<DateTime<Utc>>,
     pub encoded_args: &'a serde_json::value::RawValue,
@@ -147,7 +147,7 @@ pub(crate) struct InsertJob<'a> {
     pub finalized_at: Option<DateTime<Utc>>,
     pub id: Option<i64>,
     pub kind: &'a str,
-    pub max_attempts: i16,
+    pub max_attempts: i32,
     pub metadata: &'a JobMetadata,
     pub priority: i16,
     pub queue: &'a str,
@@ -196,7 +196,7 @@ pub(crate) struct ListJobs<'a> {
 
 #[derive(Clone, Debug)]
 pub(crate) struct CompleteJob<'a> {
-    pub attempt: Option<i16>,
+    pub attempt: Option<i32>,
     pub error: Option<&'a AttemptError>,
     pub finalized_at: Option<DateTime<Utc>>,
     pub id: i64,
@@ -249,7 +249,8 @@ pub(crate) struct NotificationInput<'a> {
 }
 
 // River Go stores `attempt`, `max_attempts`, and `priority` as native
-// integers, so they decode as `i64` and saturate into `JobRow`'s fields.
+// integers, so they decode as `i64`. A value `JobRow`'s field can't hold
+// makes the row undecodable, like other columns River can't represent.
 // JSON columns are read as bytes, since a value changed out of band might not
 // even be UTF-8.
 #[derive(Clone, Debug, FromRow)]
@@ -338,9 +339,12 @@ impl JobRecord {
             "unique_states",
             self.unique_states.map(decode_unique_states).transpose(),
         );
+        let attempt = errors.field("attempt", i32::try_from(self.attempt));
+        let max_attempts = errors.field("max_attempts", i32::try_from(self.max_attempts));
+        let priority = errors.field("priority", i16::try_from(self.priority));
         errors.finish(JobRow {
             id: self.id,
-            attempt: saturating_i16(self.attempt),
+            attempt,
             attempted_at: self.attempted_at,
             attempted_by,
             created_at: self.created_at,
@@ -348,9 +352,9 @@ impl JobRecord {
             errors: attempt_errors,
             finalized_at: self.finalized_at,
             kind: self.kind,
-            max_attempts: saturating_i16(self.max_attempts),
+            max_attempts,
             metadata,
-            priority: saturating_i16(self.priority),
+            priority,
             queue: self.queue,
             scheduled_at: self.scheduled_at,
             state,
@@ -2529,7 +2533,7 @@ mod tests {
         insert(
             connection,
             &InsertJob {
-                attempt: i16::from(running),
+                attempt: i32::from(running),
                 attempted_at: running.then_some(now - TimeDelta::hours(2)),
                 attempted_by: &[],
                 created_at: Some(now),
diff --git a/rust/riverqueue/src/extension.rs b/rust/riverqueue/src/extension.rs
index bee8a944b..7b62a3353 100644
--- a/rust/riverqueue/src/extension.rs
+++ b/rust/riverqueue/src/extension.rs
@@ -113,22 +113,65 @@ impl WorkCancelled {
     /// because `io::Error` does not expose its payload as a source.
     #[must_use]
     pub fn is_in_chain(error: &(dyn std::error::Error + 'static)) -> bool {
-        let mut current = Some(error);
-        while let Some(error) = current {
-            if error.is::<Self>() {
-                return true;
-            }
-            if let Some(payload) = error
-                .downcast_ref::<std::io::Error>()
-                .and_then(std::io::Error::get_ref)
-                && Self::is_in_chain(payload)
-            {
-                return true;
-            }
-            current = error.source();
+        in_chain::<Self>(error)
+    }
+}
+
+/// Error a worker returns to cancel its job with a reason.
+///
+/// The job is cancelled immediately, whatever attempts it has left, and the
+/// attempt's recorded error is `JobCancelError: ` followed by the reason. The
+/// error handler isn't called. Unlike [`WorkOutcome::Cancel`], which records a
+/// fixed message, it keeps why the worker cancelled the job. It's recognized
+/// anywhere in the worker error's source chain.
+///
+/// ```
+/// use riverqueue::{JobCancelError, WorkOutcome};
+///
+/// fn charge(card_expired: bool) -> Result<WorkOutcome, JobCancelError> {
+///     if card_expired {
+///         return Err(JobCancelError::new("card expired"));
+///     }
+///     Ok(WorkOutcome::Complete)
+/// }
+/// ```
+#[derive(Debug, thiserror::Error)]
+#[error("JobCancelError")]
+pub struct JobCancelError(#[source] BoxError);
+
+impl JobCancelError {
+    /// Cancels the job, recording `reason` as its attempt's error.
+    pub fn new(reason: impl Into<BoxError>) -> Self {
+        Self(reason.into())
+    }
+
+    /// Whether `error` or any error in its source chain is a
+    /// [`JobCancelError`].
+    #[must_use]
+    pub fn is_in_chain(error: &(dyn std::error::Error + 'static)) -> bool {
+        in_chain::<Self>(error)
+    }
+}
+
+/// Whether `error`'s source chain holds a `T`, looking inside errors wrapped
+/// by `std::io::Error::other` too, since `io::Error` doesn't expose its
+/// payload as a source.
+fn in_chain<T: std::error::Error + 'static>(error: &(dyn std::error::Error + 'static)) -> bool {
+    let mut current = Some(error);
+    while let Some(error) = current {
+        if error.is::<T>() {
+            return true;
+        }
+        if let Some(payload) = error
+            .downcast_ref::<std::io::Error>()
+            .and_then(std::io::Error::get_ref)
+            && in_chain::<T>(payload)
+        {
+            return true;
         }
-        false
+        current = error.source();
     }
+    false
 }
 
 /// Name of an internal runtime metric emitted to hooks.
diff --git a/rust/riverqueue/src/job.rs b/rust/riverqueue/src/job.rs
index a099fcf17..0a879a9ce 100644
--- a/rust/riverqueue/src/job.rs
+++ b/rust/riverqueue/src/job.rs
@@ -116,7 +116,7 @@ pub struct AttemptError {
     #[serde(serialize_with = "crate::encoding::go_time::serialize")]
     pub at: DateTime<Utc>,
     /// Attempt number on which the error occurred.
-    pub attempt: i16,
+    pub attempt: i32,
     /// Stringified worker error or panic value.
     pub error: String,
     /// Backtrace for a panic, otherwise empty.
@@ -126,7 +126,7 @@ pub struct AttemptError {
 impl AttemptError {
     /// Creates a persisted attempt error without a panic trace.
     #[must_use]
-    pub fn new(at: DateTime<Utc>, attempt: i16, error: impl Into<String>) -> Self {
+    pub fn new(at: DateTime<Utc>, attempt: i32, error: impl Into<String>) -> Self {
         Self {
             at,
             attempt,
@@ -154,7 +154,7 @@ pub struct InsertOpts {
     /// Options for an exact-version extension, which River carries to the
     /// extension's insert hook without persisting them.
     pub(crate) extension_options: Map<String, Value>,
-    max_attempts: Option<i16>,
+    max_attempts: Option<i32>,
     metadata: Option<JobMetadata>,
     pending: Option<bool>,
     priority: Option<i16>,
@@ -181,7 +181,7 @@ pub enum ScheduleOverride {
 impl InsertOpts {
     /// Returns the configured maximum attempts override.
     #[must_use]
-    pub const fn max_attempts(&self) -> Option<i16> {
+    pub const fn max_attempts(&self) -> Option<i32> {
         self.max_attempts
     }
 
@@ -229,7 +229,7 @@ impl InsertOpts {
 
     /// Overrides the maximum number of attempts, including the first.
     #[must_use]
-    pub const fn with_max_attempts(mut self, maximum: i16) -> Self {
+    pub const fn with_max_attempts(mut self, maximum: i32) -> Self {
         self.max_attempts = Some(maximum);
         self
     }
@@ -334,7 +334,7 @@ impl InsertOpts {
     }
 
     pub(crate) fn resolve(
-        client_max_attempts: i16,
+        client_max_attempts: i32,
         job_defaults: Self,
         call_overrides: Self,
     ) -> InsertParams {
@@ -367,7 +367,7 @@ pub struct InsertParams {
     /// Options for an exact-version extension; see [`InsertOpts`].
     pub(crate) extension_options: Map<String, Value>,
     /// Maximum number of attempts, including the first.
-    pub max_attempts: i16,
+    pub max_attempts: i32,
     /// Arbitrary JSON object metadata.
     pub metadata: JobMetadata,
     /// Insert in the pending state.
@@ -495,7 +495,7 @@ pub struct JobRow {
     /// Database-generated ID.
     pub id: i64,
     /// Current attempt number.
-    pub attempt: i16,
+    pub attempt: i32,
     /// Last attempt time.
     pub attempted_at: Option<DateTime<Utc>>,
     /// IDs of clients that attempted the job.
@@ -511,7 +511,7 @@ pub struct JobRow {
     /// Stable job kind.
     pub kind: String,
     /// Maximum attempts.
-    pub max_attempts: i16,
+    pub max_attempts: i32,
     /// Arbitrary and River-reserved metadata.
     pub metadata: JobMetadata,
     /// Priority from one through four.
diff --git a/rust/riverqueue/src/job/attempt_error.rs b/rust/riverqueue/src/job/attempt_error.rs
index e47a0fd94..5354cdd30 100644
--- a/rust/riverqueue/src/job/attempt_error.rs
+++ b/rust/riverqueue/src/job/attempt_error.rs
@@ -17,7 +17,6 @@ use serde::{
 use serde_json::value::RawValue;
 
 use super::AttemptError;
-use crate::client::saturating_i16;
 
 impl<'de> Deserialize<'de> for AttemptError {
     /// Decodes an attempt error like Go's `encoding/json`: fields match
@@ -206,15 +205,20 @@ fn strict_time(raw: &str) -> Result<DateTime<Utc>, String> {
         .ok_or_else(|| format!("attempt error time {raw} isn't an RFC 3339 timestamp"))
 }
 
+/// Converts a JSON attempt number to `i32`, saturating at its bounds.
+fn saturating_i32(value: i64) -> i32 {
+    i32::try_from(value).unwrap_or(if value < 0 { i32::MIN } else { i32::MAX })
+}
+
 /// Decodes `attempt` like Go's `encoding/json` decodes an `int`: an integer
-/// literal or `null`. `attempt` is narrower than Go's `int`, so values beyond
-/// `i16` saturate like other persisted attempt counts.
-fn strict_attempt(raw: &str) -> Result<i16, String> {
+/// literal or `null`. `attempt` is narrower than Go's 64-bit `int`, so values
+/// beyond `i32` saturate.
+fn strict_attempt(raw: &str) -> Result<i32, String> {
     if raw == "null" {
         return Ok(0);
     }
     raw.parse::<i64>()
-        .map(saturating_i16)
+        .map(saturating_i32)
         .map_err(|_| format!("attempt error attempt {raw} isn't an integer"))
 }
 
@@ -236,14 +240,13 @@ const MAX_EXACT_FLOAT_INTEGER: f64 = 9_007_199_254_740_992.0;
 
 /// Decodes `attempt` like Go: integers, and numbers or numeric strings with an
 /// integral value no larger in magnitude than 2^53. `attempt` is narrower
-/// than Go's `int`, so values beyond `i16` saturate like other persisted
-/// attempt counts. Unlike Go, a string in hexadecimal floating point notation
+/// than Go's 64-bit `int`, so values beyond `i32` saturate. Unlike Go, a string in hexadecimal floating point notation
 /// (such as `"0x1p4"`) isn't recognized and decodes as zero.
 #[allow(
     clippy::float_cmp,
     reason = "an exact comparison checks for an integral value"
 )]
-fn lenient_attempt(raw: &str) -> i16 {
+fn lenient_attempt(raw: &str) -> i32 {
     let number: Cow<'_, str> = if raw.starts_with('"') {
         match serde_json::from_str::<String>(raw) {
             Ok(text) => Cow::Owned(text.trim().to_owned()),
@@ -256,7 +259,7 @@ fn lenient_attempt(raw: &str) -> i16 {
     };
 
     if let Ok(integer) = number.parse::<i64>() {
-        return saturating_i16(integer);
+        return saturating_i32(integer);
     }
     match number.parse::<f64>() {
         Ok(float) if float == float.trunc() && float.abs() <= MAX_EXACT_FLOAT_INTEGER =>
@@ -265,7 +268,7 @@ fn lenient_attempt(raw: &str) -> i16 {
                 clippy::cast_possible_truncation,
                 reason = "the float is integral and within the exact integer range"
             )]
-            saturating_i16(float as i64)
+            saturating_i32(float as i64)
         }
         _ => 0,
     }
@@ -422,7 +425,7 @@ mod tests {
             .and_utc()
     }
 
-    fn attempt_error(at: DateTime<Utc>, attempt: i16, error: &str, trace: &str) -> AttemptError {
+    fn attempt_error(at: DateTime<Utc>, attempt: i32, error: &str, trace: &str) -> AttemptError {
         AttemptError::new(at, attempt, error).with_trace(trace)
     }
 
@@ -584,8 +587,8 @@ mod tests {
             ),
             (
                 "AttemptSaturates",
-                r#"{"attempt":40000}"#,
-                attempt_error(zero, i16::MAX, "", ""),
+                r#"{"attempt":4000000000}"#,
+                attempt_error(zero, i32::MAX, "", ""),
             ),
             (
                 "AttemptBool",
@@ -763,8 +766,8 @@ mod tests {
             ("Null", "null", attempt_error(zero, 0, "", "")),
             (
                 "AttemptSaturates",
-                r#"{"attempt":-40000}"#,
-                attempt_error(zero, i16::MIN, "", ""),
+                r#"{"attempt":-4000000000}"#,
+                attempt_error(zero, i32::MIN, "", ""),
             ),
         ] {
             assert_eq!(
diff --git a/rust/riverqueue/src/lib.rs b/rust/riverqueue/src/lib.rs
index ede0cab2d..058cecae7 100644
--- a/rust/riverqueue/src/lib.rs
+++ b/rust/riverqueue/src/lib.rs
@@ -63,8 +63,8 @@ pub use event::{
 };
 pub use extension::{
     DefaultRetryPolicy, ErrorHandler, ErrorHandlerDecision, Extensions, Hook, InsertContext,
-    InsertMiddleware, InsertNext, InsertedJob, Metric, MetricName, PanicError, Plugin, RetryPolicy,
-    WorkCancelled, WorkError, WorkMiddleware, WorkNext, WorkResult,
+    InsertMiddleware, InsertNext, InsertedJob, JobCancelError, Metric, MetricName, PanicError,
+    Plugin, RetryPolicy, WorkCancelled, WorkError, WorkMiddleware, WorkNext, WorkResult,
 };
 pub use job::{
     AttemptError, InsertBatch, InsertBatchResult, InsertOpts, InsertParams, InsertResult, Job,
@@ -107,7 +107,7 @@ pub use tokio_util;
 pub use worker::{WorkContext, WorkOutcome, Worker, WorkerTimeout, Workers};
 
 /// Default maximum number of attempts for a job.
-pub const MAX_ATTEMPTS_DEFAULT: i16 = 25;
+pub const MAX_ATTEMPTS_DEFAULT: i32 = 25;
 
 /// Default minimum delay between queue fetches.
 pub const FETCH_COOLDOWN_DEFAULT: std::time::Duration = std::time::Duration::from_millis(100);
diff --git a/rust/riverqueue/tests/postgres.rs b/rust/riverqueue/tests/postgres.rs
index a84234085..17b3c4054 100644
--- a/rust/riverqueue/tests/postgres.rs
+++ b/rust/riverqueue/tests/postgres.rs
@@ -582,6 +582,31 @@ async fn client_discards_a_failing_job_after_max_attempts() {
     database.cleanup().await;
 }
 
+// Postgres's `max_attempts` column is 16 bits, so like River Go's drivers a
+// larger value is clamped to 32,767 on insert, while SQLite stores it.
+#[tokio::test]
+async fn max_attempts_beyond_the_column_are_clamped() {
+    let database = support::PostgresSchema::current("rs_wide_max_attempts").await;
+    let client = Client::builder(database.pool.clone()).build().unwrap();
+
+    let inserted = client
+        .insert(EchoArgs {
+            message: "wide".to_owned(),
+        })
+        .opts(InsertOpts::default().with_max_attempts(40_000))
+        .await
+        .unwrap();
+    assert_eq!(inserted.job.row.max_attempts, i32::from(i16::MAX));
+    let listed = client
+        .jobs()
+        .list(JobListParams::default().ids([inserted.job.row.id]))
+        .await
+        .unwrap();
+    assert_eq!(listed.jobs[0].max_attempts, i32::from(i16::MAX));
+
+    database.cleanup().await;
+}
+
 #[tokio::test]
 async fn client_discards_a_job_of_an_unregistered_kind() {
     let database = support::PostgresSchema::current("rs_unknown_kind").await;
diff --git a/rust/riverqueue/tests/protocol_fixtures.rs b/rust/riverqueue/tests/protocol_fixtures.rs
index d58679a47..6b87106f0 100644
--- a/rust/riverqueue/tests/protocol_fixtures.rs
+++ b/rust/riverqueue/tests/protocol_fixtures.rs
@@ -104,7 +104,7 @@ fn retry_row(id: i64, now: DateTime<Utc>, previous_errors: usize) -> JobRow {
         riverqueue::encoding::encode_args(&serde_json::json!({})).unwrap(),
         now,
     );
-    row.attempt = i16::try_from(previous_errors + 1).unwrap();
+    row.attempt = i32::try_from(previous_errors + 1).unwrap();
     row.attempted_at = Some(now);
     row.attempted_by = vec!["fixture".to_owned()];
     row.errors = vec![AttemptError::new(now, 1, "previous failure"); previous_errors];
diff --git a/rust/riverqueue/tests/resilience_sqlite.rs b/rust/riverqueue/tests/resilience_sqlite.rs
index 3c72146e6..af72530f9 100644
--- a/rust/riverqueue/tests/resilience_sqlite.rs
+++ b/rust/riverqueue/tests/resilience_sqlite.rs
@@ -278,7 +278,7 @@ async fn claimed_rows_decode_individually_and_accept_go_integer_ranges() {
         .unwrap();
 
     // River Go stores native integers on SQLite, so `max_attempts` can exceed
-    // Rust's `i16`. Such a job must still be worked.
+    // Postgres's 16-bit columns. Such a job must still be worked.
     let wide = client.insert(ResilienceArgs {}).await.unwrap();
     sqlx::query("UPDATE river_job SET max_attempts = 40000 WHERE id = ?")
         .bind(wide.job.row.id)
@@ -348,7 +348,7 @@ async fn claimed_rows_decode_individually_and_accept_go_integer_ranges() {
     );
     assert!(errors.is_none());
     let wide = client.jobs().get(wide.job.row.id).await.unwrap();
-    assert_eq!(wide.max_attempts, i16::MAX);
+    assert_eq!(wide.max_attempts, 40_000);
 
     let odd_errors = client.jobs().get(odd_errors.job.row.id).await.unwrap();
     let zero_time = "0001-01-01T00:00:00Z".parse().unwrap();
diff --git a/rust/riverqueue/tests/sqlite_storage.rs b/rust/riverqueue/tests/sqlite_storage.rs
index 9fa610fbd..d10ddf58b 100644
--- a/rust/riverqueue/tests/sqlite_storage.rs
+++ b/rust/riverqueue/tests/sqlite_storage.rs
@@ -90,6 +90,28 @@ async fn control_and_resign_notifications_carry_their_fields() {
     pool.close().await;
 }
 
+/// River Go stores `max_attempts` as a native integer on SQLite, so a value
+/// beyond Postgres's 16-bit column inserts and lists unchanged.
+#[tokio::test]
+async fn max_attempts_beyond_sixteen_bits_insert_and_list_unchanged() {
+    let (client, pool) = setup().await;
+
+    let inserted = client
+        .insert(EmptyBatchArgs { value: 1 })
+        .opts(InsertOpts::default().with_max_attempts(40_000))
+        .await
+        .unwrap();
+    assert_eq!(inserted.job.row.max_attempts, 40_000);
+    let listed = client
+        .jobs()
+        .list(JobListParams::default().ids([inserted.job.row.id]))
+        .await
+        .unwrap();
+    assert_eq!(listed.jobs[0].max_attempts, 40_000);
+
+    pool.close().await;
+}
+
 #[tokio::test]
 async fn empty_batches_are_rejected_before_database_work() {
     let (client, pool) = setup().await;