[Rust, JS] Match River Go's behavior in the Rust and JavaScript clients

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
 

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

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");

js/driver/pg/src/sql/jobs.ts

@@ -733,15 +733,15 @@ 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
             ? metadata
             : { ...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);
+}

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");

js/driver/prisma/src/driver.ts

@@ -317,15 +317,15 @@ 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
             ? metadata
             : { ...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);
+}

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({

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

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

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([

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) {

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

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) {

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({

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,
   });

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");

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);

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;