fixes

Author: timgit

src/boss.ts

@@ -206,7 +206,10 @@ class Boss extends EventEmitter implements types.EventsMixin {
 
       if (this.#stopping) return
 
-      const warnings = rowsCacheStats.filter(i => i.queuedCount > (i.warningQueueSize || WARNINGS.LARGE_QUEUE.size))
+      // Coerce with Number(): CockroachDB returns these integer columns as strings, so a bare `>`
+      // would compare lexicographically ("100" > "9" === false) and silently miss the backlog. On
+      // standard Postgres these are already numbers, so Number() is a no-op.
+      const warnings = rowsCacheStats.filter(i => Number(i.queuedCount) > (Number(i.warningQueueSize) || WARNINGS.LARGE_QUEUE.size))
 
       for (const warning of warnings) {
         await emitAndPersistWarning(this.#warningContext,

src/manager.ts

@@ -1073,21 +1073,29 @@ class Manager extends EventEmitter implements types.EventsMixin {
     let count = 0
 
     for (const job of jobs) {
-      const canRetry = job.retry_count < job.retry_limit
+      // CockroachDB returns INT8 columns as strings. These rows come straight from a SELECT *, so
+      // unlike fetch/getJobById they are never normalized. Coerce the fields used in arithmetic and
+      // comparison below — otherwise `retry_count < retry_limit` is a lexicographic string compare
+      // ("9" < "10" === false, wrongly failing a retriable job) and `retry_count + 1` concatenates.
+      const retryCount = Number(job.retry_count)
+      const retryLimit = Number(job.retry_limit)
+      const retryDelay = Number(job.retry_delay)
+      const retryDelayMax = job.retry_delay_max != null ? Number(job.retry_delay_max) : null
+
+      const canRetry = retryCount < retryLimit
       let retried = false
 
       if (canRetry) {
         // Calculate start_after for retry
         let startAfter = job.start_after
         if (!job.retry_backoff) {
-          startAfter = new Date(Date.now() + job.retry_delay * 1000)
+          startAfter = new Date(Date.now() + retryDelay * 1000)
         } else {
-          const retryCount = job.retry_count + 1
-          const exp = Math.min(16, retryCount)
-          const delay = job.retry_delay * (Math.pow(2, exp) / 2 + Math.pow(2, exp) / 2 * Math.random())
+          const exp = Math.min(16, retryCount + 1)
+          const delay = retryDelay * (Math.pow(2, exp) / 2 + Math.pow(2, exp) / 2 * Math.random())
           // Match the canonical failJobs() SQL: LEAST(retry_delay_max, delay) caps the backoff,
           // treating NULL as "no cap" and 0 as a real cap. (`?:` would wrongly treat 0 as no cap.)
-          const cappedDelay = job.retry_delay_max != null ? Math.min(job.retry_delay_max, delay) : delay
+          const cappedDelay = retryDelayMax != null ? Math.min(retryDelayMax, delay) : delay
           startAfter = new Date(Date.now() + cappedDelay * 1000)
         }
 
@@ -1333,7 +1341,17 @@ class Manager extends EventEmitter implements types.EventsMixin {
 
     const { rows } = await this.db.executeSql(query.text, query.values)
 
-    return Object.assign(queue, rows.at(0) ||
+    const stats = rows.at(0)
+
+    // CockroachDB returns integer columns as strings; normalize the stats counts. (The queue fields
+    // merged in below come from getQueueCache -> getQueues, which already normalizes them.)
+    if (stats && this.config.backend === 'cockroachdb') {
+      for (const field of NUMERIC_QUEUE_FIELDS) {
+        if (stats[field] !== undefined && stats[field] !== null) stats[field] = Number(stats[field])
+      }
+    }
+
+    return Object.assign(queue, stats ||
             {
               deferredCount: 0,
               queuedCount: 0,

test/distributedDatabaseTest.ts

@@ -198,6 +198,53 @@ helper.describePglite('distributed database mode', { timeout: 20000 }, function
     expect(retried.state).toBe('retry')
   })
 
+  it('should retry, not fail, when the backend returns integer columns as strings', async function () {
+    // Regression: reinsertFailedJobs read the raw SELECT * rows from the distributed fail path. On
+    // CockroachDB, INT8 columns come back as strings, so `retry_count < retry_limit` was a
+    // lexicographic compare — "9" < "10" is false — which permanently failed a job that still had
+    // retries left. The DISTRIBUTED=true Postgres run can't catch this (node-pg returns numbers
+    // there), so we simulate CockroachDB's string typing with a wrapper over a real connection.
+    ctx.boss = await helper.start({ ...ctx.bossConfig, __test__distributed: true })
+
+    const jobId = await ctx.boss.send(ctx.schema, { test: 'stringints' }, { retryLimit: 10 })
+    helper.assertTruthy(jobId)
+
+    const [job] = await ctx.boss.fetch(ctx.schema)
+    expect(job.id).toBe(jobId)
+
+    const _db = await helper.getDb()
+    try {
+      // Put the job one retry short of its limit. Numerically 9 < 10, so it must still retry; only a
+      // string comparison ("9" < "10" === false) would wrongly fail it.
+      await _db.executeSql(`UPDATE ${ctx.schema}.job SET retry_count = 9 WHERE id = $1`, [jobId])
+
+      // Wrap the connection so SELECT * rows return integer columns as strings, exactly as
+      // CockroachDB's driver returns INT8. The fail path's select is the only SELECT * it issues.
+      const integerColumns = ['priority', 'retry_limit', 'retry_count', 'retry_delay', 'retry_delay_max', 'group_tier', 'expire_seconds', 'deletion_seconds', 'pending_dependencies']
+      const cockroachLike = {
+        executeSql: async (text: string, values?: unknown[]) => {
+          const result = await _db.executeSql(text, values)
+          if (/^\s*SELECT \* FROM/i.test(text)) {
+            for (const row of result.rows) {
+              for (const col of integerColumns) {
+                if (row[col] !== null && row[col] !== undefined) row[col] = String(row[col])
+              }
+            }
+          }
+          return result
+        }
+      }
+
+      await ctx.boss.fail(ctx.schema, jobId, null, { db: cockroachLike })
+
+      const retried = await ctx.boss.getJobById(ctx.schema, jobId)
+      helper.assertTruthy(retried)
+      expect(retried.state).toBe('retry')
+    } finally {
+      await _db.close()
+    }
+  })
+
   it('should compose completeDistributed inside a caller transaction and roll back with it', async function () {
     ctx.boss = await helper.start({ ...ctx.bossConfig, __test__distributed: true })
 
@@ -273,11 +320,29 @@ helper.describePglite('distributed database mode', { timeout: 20000 }, function
     expect(completed.state).toBe('completed')
   })
 
-  it('should work with noTablePartitioning mode', async function () {
-    // This test covers the noPartitioning path in plans.ts (lines 244, 260)
+  it('should return numeric stats counts under the cockroachdb backend', async function () {
+    // getQueueStats has its own backend === 'cockroachdb' coercion: the stats counts come from a raw
+    // stats query, not the normalized getQueues path, so they would otherwise be returned as strings
+    // on CockroachDB. Selecting the cockroachdb backend on Postgres runs that coercion loop and lets
+    // us assert the public counts come back as numbers.
+    ctx.boss = await helper.start({ ...ctx.bossConfig, backend: 'cockroachdb' })
+
+    const jobId = await ctx.boss.send(ctx.schema, { test: 'stats' })
+    helper.assertTruthy(jobId)
+
+    const stats = await ctx.boss.getQueueStats(ctx.schema)
+    expect(typeof stats.queuedCount).toBe('number')
+    expect(stats.totalCount).toBe(1)
+  })
+
+  it('should construct schema with the yugabytedb backend (no partitioning, no advisory locks)', async function () {
+    // Exercises the noTablePartitioning + noAdvisoryLocks construction path on plain Postgres by
+    // selecting the yugabytedb backend, whose only flags are those two (both PostgreSQL-compatible,
+    // they just remove features). The compatibility flags are derived from `backend` and are not
+    // settable directly — resolveBackend() overwrites them — so the backend is the only way in.
     ctx.boss = await helper.start({
       ...ctx.bossConfig,
-      noTablePartitioning: true
+      backend: 'yugabytedb'
     })
 
     // Basic send/fetch to verify everything works

test/pgliteAdapterTest.ts

@@ -6,7 +6,7 @@ function createFakePglite (): PGliteLike & { calls: Array<{ method: 'query' | 'e
   const calls: Array<{ method: 'query' | 'exec', text: string, params?: unknown[] }> = []
   return {
     calls,
-    async query (text: string, params?: unknown[]) {
+    async query (text: string, params?: unknown[]): Promise<{ rows: any[] }> {
       calls.push({ method: 'query', text, params })
       return { rows: [{ id: '1' }] }
     },