fix(connector): the Discord gateway sends queued posts, so its replies don't wait for the API to wake (#34131)

The always-on Discord gateway worker now runs the same background sweeps as the API (queued posts, held starts, cleanup), so Discord replies go out at once instead of waiting for the scaled-to-zero API to wake. Posts are claimed with leases, so running the sweeps in both services never sends one twice.

cloud/apps/connector/src/discord-gateway-worker.ts

@@ -54,6 +54,8 @@ export async function runDiscordGatewayWorker(input: {
     if (!leadership) {
       return 0
     }
+    // The API scales to zero, so the always-on leader runs the core's sweeps too; posts are leased.
+    const stopBackground = runtime.core.startBackground()
     try {
       const fatal = Promise.withResolvers<string>()
       const ended = Promise.race([
@@ -86,6 +88,7 @@ export async function runDiscordGatewayWorker(input: {
       }
       return reason === 'stopping' ? 0 : 1
     } finally {
+      stopBackground()
       await leadership.release()
     }
   } finally {

cloud/apps/connector/src/slack-posting.test.ts

@@ -1,6 +1,6 @@
 import { CONNECTOR_PATHS, ConnectorPollResponseSchema } from '@orca-cloud/connector-contract'
 import { afterEach, describe, expect, it, vi } from 'vitest'
-import { cleanupConnectorHarnesses } from './connector-server-harness.test-fixture.js'
+import { cleanupConnectorHarnesses, pauseBefore } from './connector-server-harness.test-fixture.js'
 import {
   BOT,
   CHANNEL,
@@ -279,6 +279,30 @@ describe('Slack thread status and posts', () => {
     expect((await states())[`reply:${harness.machine.hostId}:r2`]).toBe('uncertain')
   })
 
+  it('a second worker, as another role runs, claims nothing while the first holds the lease', async () => {
+    const harness = await startedThread()
+    const api = new ConnectorPosts(harness.database, harness.now, 'slack')
+    const gateway = new ConnectorPosts(harness.database, harness.now, 'slack')
+    expect(await api.claim()).not.toEqual([])
+    expect(await gateway.claim()).toEqual([])
+  })
+
+  // SQLite runs one transaction at a time, so only Postgres can overlap two claims.
+  it.skipIf(!process.env.ORCA_CONNECTOR_TEST_DATABASE_URL)(
+    'a claim that overlaps another one takes nothing',
+    async () => {
+      const harness = await startedThread()
+      const paused = pauseBefore(harness.database, 'UPDATE connector_posts SET state')
+      const api = new ConnectorPosts(paused.database, harness.now, 'slack')
+      const gateway = new ConnectorPosts(harness.database, harness.now, 'slack')
+      const first = api.claim()
+      await paused.atStatement
+      expect(await gateway.claim()).toEqual([])
+      paused.release()
+      expect(await first).not.toEqual([])
+    }
+  )
+
   it('a sender whose lease ran out changes nothing', async () => {
     const harness = await startedThread()
     const posts = new ConnectorPosts(harness.database, harness.now, 'slack')