test fixes

Author: timgit

test/monitorBackoffTest.ts

@@ -181,143 +181,148 @@ helper.describePostgresOnly('monitor backoff', function () {
     expect(new Date(forced.capturedOn).getTime()).toBe(new Date(cached.capturedOn).getTime())
   })
 
-  it('measures the pin on the server, not the latency of the call', async function () {
-    const boss = await startBoss()
+  it('does not run when vacuum monitoring is turned off', async function () {
+    const boss = await startBoss({ __test__monitor_stats_seconds: 90, monitorVacuum: false })
 
-    await boss.send('backoff', {})
+    await boss.supervise()
 
-    // One connection, and something else already holding it. This is the shape a busy app hands the
-    // supervisor: the same pool serves every fetch and complete, so the wait for a connection lands
-    // inside any stopwatch the client wraps around the call — while pinning nothing, because the
-    // transaction has not begun yet. Deferring on that number would blame the job table for pool
-    // contention, and deferring monitoring would do nothing to relieve it.
-    const pool = new pg.Pool({ connectionString: helper.getConnectionString(), max: 1 })
+    expect(await backoffUntil(ctx.schema)).toBeNull()
+    expect(await warnings(ctx.schema)).toHaveLength(0)
+  })
 
-    try {
-      const hog = pool.query('SELECT pg_sleep(2)')
+  it('stamps the capture time when the counts land, not when the pass claimed them', async function () {
+    const boss = await startBoss({ __test__delay_maint_ms: 0 })
 
-      await new Promise(resolve => setTimeout(resolve, 50))
+    await boss.send('backoff', {})
 
-      const started = Date.now()
-      const result = await pool.query(plans.cacheQueueStats(ctx.schema, 'job', ['backoff']))
-      const callSeconds = (Date.now() - started) / 1000
+    const before = new Date()
+    await boss.supervise()
 
-      const rows = (Array.isArray(result) ? result : [result]).flatMap(r => r?.rows ?? [])
-      const pinSeconds = rows.reduce((max, row) => Math.max(max, Number(row.pinSeconds) || 0), 0)
+    const stamped = await monitorOn(ctx.schema, 'backoff')
 
-      expect(callSeconds).toBeGreaterThan(1.5)
-      expect(pinSeconds).toBeLessThan(0.5)
+    helper.assertTruthy(stamped)
 
-      await hog
-    } finally {
-      await pool.end()
-    }
+    // cacheQueueStats re-stamps monitor_on in the same statement that writes the counts. Without
+    // that the interval is measured start-to-start, and stops delaying anything at all the moment
+    // one aggregate outruns it.
+    expect(new Date(stamped).getTime()).toBeGreaterThanOrEqual(before.getTime())
   })
 
-  // The same key cacheQueueStats/refreshQueueStats derive, computed in SQL so the test cannot drift
-  // from the implementation's hashing.
-  const LOCK_KEY = (schema: string) =>
-    `('x' || encode(sha224((current_database() || '.pgboss.${schema}queue-stats')::bytea), 'hex'))::bit(64)::bigint`
-
-  async function holdStatsLock (schema: string) {
-    const client = new pg.Client(helper.getConnectionString())
-    await client.connect()
-    await client.query('BEGIN')
-    await client.query(`SELECT pg_advisory_xact_lock(${LOCK_KEY(schema)})`)
-
-    return async () => {
-      await client.query('ROLLBACK')
-      await client.end()
+  // These three need a second, independent connection - one to hold the queue-stats lock or the
+  // pool while another instance tries to proceed. PGlite has no server and CockroachDB has no
+  // advisory locks, so the scenario cannot be staged on either.
+  helper.describeMultiConnectionOnly('under contention', function () {
+    it('measures the pin on the server, not the latency of the call', async function () {
+      const boss = await startBoss()
+
+      await boss.send('backoff', {})
+
+      // One connection, and something else already holding it. This is the shape a busy app hands the
+      // supervisor: the same pool serves every fetch and complete, so the wait for a connection lands
+      // inside any stopwatch the client wraps around the call — while pinning nothing, because the
+      // transaction has not begun yet. Deferring on that number would blame the job table for pool
+      // contention, and deferring monitoring would do nothing to relieve it.
+      const pool = new pg.Pool({ connectionString: helper.getConnectionString(), max: 1 })
+
+      try {
+        const hog = pool.query('SELECT pg_sleep(2)')
+
+        await new Promise(resolve => setTimeout(resolve, 50))
+
+        const started = Date.now()
+        const result = await pool.query(plans.cacheQueueStats(ctx.schema, 'job', ['backoff']))
+        const callSeconds = (Date.now() - started) / 1000
+
+        const rows = (Array.isArray(result) ? result : [result]).flatMap(r => r?.rows ?? [])
+        const pinSeconds = rows.reduce((max, row) => Math.max(max, Number(row.pinSeconds) || 0), 0)
+
+        expect(callSeconds).toBeGreaterThan(1.5)
+        expect(pinSeconds).toBeLessThan(0.5)
+
+        await hog
+      } finally {
+        await pool.end()
+      }
+    })
+
+    // The same key cacheQueueStats/refreshQueueStats derive, computed in SQL so the test cannot drift
+    // from the implementation's hashing.
+    const LOCK_KEY = (schema: string) =>
+      `('x' || encode(sha224((current_database() || '.pgboss.${schema}queue-stats')::bytea), 'hex'))::bit(64)::bigint`
+
+    async function holdStatsLock (schema: string) {
+      const client = new pg.Client(helper.getConnectionString())
+      await client.connect()
+      await client.query('BEGIN')
+      await client.query(`SELECT pg_advisory_xact_lock(${LOCK_KEY(schema)})`)
+
+      return async () => {
+        await client.query('ROLLBACK')
+        await client.end()
+      }
     }
-  }
-
-  it('skips the aggregate instead of queueing behind another instance', async function () {
-    const boss = await startBoss({ monitorIntervalSeconds: 1 })
-
-    await boss.send('backoff', {})
-    await boss.supervise()
-
-    const [before] = await boss.getQueueStats('backoff')
-
-    await boss.send('backoff', {})
-
-    const release = await holdStatsLock(ctx.schema)
 
-    try {
-      await new Promise(resolve => setTimeout(resolve, 1500))
+    it('skips the aggregate instead of queueing behind another instance', async function () {
+      const boss = await startBoss({ monitorIntervalSeconds: 1 })
 
-      // Blocking here would hold this backend's snapshot for the whole of the other instance's
-      // scan, which is the overlap the backoff exists to prevent. The pass gives up instead.
-      const started = Date.now()
+      await boss.send('backoff', {})
       await boss.supervise()
-      const seconds = (Date.now() - started) / 1000
 
-      expect(seconds).toBeLessThan(1)
+      const [before] = await boss.getQueueStats('backoff')
 
-      const [after] = await boss.getQueueStats('backoff')
+      await boss.send('backoff', {})
 
-      // And it gives up without writing counts: an empty stats CTE still LEFT JOINs, so an
-      // unguarded UPDATE would COALESCE every one of them to zero.
-      expect(after.totalCount).toBe(before.totalCount)
+      const release = await holdStatsLock(ctx.schema)
 
-      // capturedOn does move, because trySetQueueMonitorTime stamped monitor_on when it claimed the
-      // interval, a statement earlier and before the lock was attempted. The queue therefore sits
-      // out one interval rather than retrying immediately — the counts are at most one
-      // monitorIntervalSeconds behind, against a staleness budget measured in hours.
-      expect(new Date(after.capturedOn).getTime()).toBeGreaterThan(new Date(before.capturedOn).getTime())
-    } finally {
-      await release()
-    }
-  })
-
-  it('serves the cache when a forced read loses the same race', async function () {
-    const boss = await startBoss()
-
-    await boss.send('backoff', {})
-    await boss.supervise()
-
-    const [cached] = await boss.getQueueStats('backoff')
+      try {
+        await new Promise(resolve => setTimeout(resolve, 1500))
 
-    await boss.send('backoff', {})
+        // Blocking here would hold this backend's snapshot for the whole of the other instance's
+        // scan, which is the overlap the backoff exists to prevent. The pass gives up instead.
+        const started = Date.now()
+        await boss.supervise()
+        const seconds = (Date.now() - started) / 1000
 
-    const release = await holdStatsLock(ctx.schema)
+        expect(seconds).toBeLessThan(1)
 
-    try {
-      // refreshQueueStats had no gate at all before this: every instance holding a stale cache ran
-      // its own whole-table scan, and two different queue names did not even contend.
-      const [forced] = await boss.getQueueStats('backoff', { force: true })
+        const [after] = await boss.getQueueStats('backoff')
 
-      expect(forced.totalCount).toBe(cached.totalCount)
-      expect(new Date(forced.capturedOn).getTime()).toBe(new Date(cached.capturedOn).getTime())
-    } finally {
-      await release()
-    }
-  })
+        // And it gives up without writing counts: an empty stats CTE still LEFT JOINs, so an
+        // unguarded UPDATE would COALESCE every one of them to zero.
+        expect(after.totalCount).toBe(before.totalCount)
 
-  it('does not run when vacuum monitoring is turned off', async function () {
-    const boss = await startBoss({ __test__monitor_stats_seconds: 90, monitorVacuum: false })
+        // capturedOn does move, because trySetQueueMonitorTime stamped monitor_on when it claimed the
+        // interval, a statement earlier and before the lock was attempted. The queue therefore sits
+        // out one interval rather than retrying immediately — the counts are at most one
+        // monitorIntervalSeconds behind, against a staleness budget measured in hours.
+        expect(new Date(after.capturedOn).getTime()).toBeGreaterThan(new Date(before.capturedOn).getTime())
+      } finally {
+        await release()
+      }
+    })
 
-    await boss.supervise()
+    it('serves the cache when a forced read loses the same race', async function () {
+      const boss = await startBoss()
 
-    expect(await backoffUntil(ctx.schema)).toBeNull()
-    expect(await warnings(ctx.schema)).toHaveLength(0)
-  })
+      await boss.send('backoff', {})
+      await boss.supervise()
 
-  it('stamps the capture time when the counts land, not when the pass claimed them', async function () {
-    const boss = await startBoss({ __test__delay_maint_ms: 0 })
+      const [cached] = await boss.getQueueStats('backoff')
 
-    await boss.send('backoff', {})
+      await boss.send('backoff', {})
 
-    const before = new Date()
-    await boss.supervise()
+      const release = await holdStatsLock(ctx.schema)
 
-    const stamped = await monitorOn(ctx.schema, 'backoff')
-
-    helper.assertTruthy(stamped)
+      try {
+        // refreshQueueStats had no gate at all before this: every instance holding a stale cache ran
+        // its own whole-table scan, and two different queue names did not even contend.
+        const [forced] = await boss.getQueueStats('backoff', { force: true })
 
-    // cacheQueueStats re-stamps monitor_on in the same statement that writes the counts. Without
-    // that the interval is measured start-to-start, and stops delaying anything at all the moment
-    // one aggregate outruns it.
-    expect(new Date(stamped).getTime()).toBeGreaterThanOrEqual(before.getTime())
+        expect(forced.totalCount).toBe(cached.totalCount)
+        expect(new Date(forced.capturedOn).getTime()).toBe(new Date(cached.capturedOn).getTime())
+      } finally {
+        await release()
+      }
+    })
   })
 })

test/monitorVacuumTest.ts

@@ -88,7 +88,12 @@ async function startBoss (options: object = {}) {
   return boss
 }
 
-helper.describePostgresOnly('vacuum monitoring', function () {
+// describeMultiConnectionOnly, not describePostgresOnly: nearly every case here needs a second
+// independent connection to hold the horizon while pg-boss observes it, and PGlite has no server to
+// open one against (getConnectionString returns a placeholder that fails DNS). The subsystem is not
+// meaningfully exercisable there anyway - it reads real autovacuum activity out of
+// pg_stat_user_tables against a genuine MVCC horizon.
+helper.describeMultiConnectionOnly('vacuum monitoring', function () {
   it('stays quiet while the table is under its own autovacuum budget', async function () {
     const boss = await startBoss()