From 64bba690adbeda116a75a6bb978c3d583ec6ac4b Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Mon, 26 Oct 2026 01:27:26 -0400 Subject: [PATCH] feat(connector): inbox with held long poll, two confirmations and expiry, on connector tokens (#33734) --- .github/workflows/cloud-verify.yml | 1 + cloud/apps/connector/Dockerfile | 11 +- cloud/apps/connector/package.json | 3 +- .../connector/src/connector-background.ts | 58 +++ .../apps/connector/src/connector-database.ts | 7 +- .../connector/src/connector-events.test.ts | 295 ++++++++++++ cloud/apps/connector/src/connector-hosts.ts | 114 +++++ .../connector/src/connector-inbox.test.ts | 432 ++++++++++++++++++ cloud/apps/connector/src/connector-inbox.ts | 209 +++++++++ cloud/apps/connector/src/connector-schema.ts | 65 +++ .../connector-server-harness.test-fixture.ts | 149 +++++- cloud/apps/connector/src/connector-server.ts | 79 +++- cloud/apps/connector/src/connector-threads.ts | 138 ++++++ cloud/apps/connector/src/dev-enqueue-route.ts | 64 +++ cloud/apps/connector/src/inbox-waker.test.ts | 43 ++ cloud/apps/connector/src/inbox-waker.ts | 73 +++ cloud/apps/connector/src/index.ts | 21 +- cloud/apps/connector/src/machine-auth.ts | 30 ++ cloud/apps/connector/src/machine-routes.ts | 164 +++++++ cloud/apps/connector/src/queue-delivery.ts | 36 ++ .../src/cloud-notifications.test.ts | 97 ++++ .../cloud-database/src/cloud-notifications.ts | 138 ++++++ cloud/packages/cloud-database/src/index.ts | 5 + cloud/pnpm-lock.yaml | 3 + 24 files changed, 2214 insertions(+), 21 deletions(-) create mode 100644 cloud/apps/connector/src/connector-background.ts create mode 100644 cloud/apps/connector/src/connector-events.test.ts create mode 100644 cloud/apps/connector/src/connector-hosts.ts create mode 100644 cloud/apps/connector/src/connector-inbox.test.ts create mode 100644 cloud/apps/connector/src/connector-inbox.ts create mode 100644 cloud/apps/connector/src/connector-schema.ts create mode 100644 cloud/apps/connector/src/connector-threads.ts create mode 100644 cloud/apps/connector/src/dev-enqueue-route.ts create mode 100644 cloud/apps/connector/src/inbox-waker.test.ts create mode 100644 cloud/apps/connector/src/inbox-waker.ts create mode 100644 cloud/apps/connector/src/machine-auth.ts create mode 100644 cloud/apps/connector/src/machine-routes.ts create mode 100644 cloud/apps/connector/src/queue-delivery.ts create mode 100644 cloud/packages/cloud-database/src/cloud-notifications.test.ts create mode 100644 cloud/packages/cloud-database/src/cloud-notifications.ts diff --git a/.github/workflows/cloud-verify.yml b/.github/workflows/cloud-verify.yml index 7a12a4d3184..80d24392106 100644 --- a/.github/workflows/cloud-verify.yml +++ b/.github/workflows/cloud-verify.yml @@ -189,6 +189,7 @@ jobs: ORCA_BACKGROUND_LAUNCH: '1' ORCA_PUSH_TEST_DATABASE_URL: postgres://relay_test:relay_test@127.0.0.1:5432/orca_relay_test ORCA_CONNECTOR_TEST_DATABASE_URL: postgres://relay_test:relay_test@127.0.0.1:5432/orca_relay_test + ORCA_CLOUD_DATABASE_TEST_URL: postgres://relay_test:relay_test@127.0.0.1:5432/orca_relay_test ORCA_RELAY_TEST_POSTGRES_URL: postgres://relay_test:relay_test@127.0.0.1:5432/orca_relay_test steps: - uses: actions/checkout@v4 diff --git a/cloud/apps/connector/Dockerfile b/cloud/apps/connector/Dockerfile index 3bb28d5c02e..1eb1c340034 100644 --- a/cloud/apps/connector/Dockerfile +++ b/cloud/apps/connector/Dockerfile @@ -4,14 +4,19 @@ ENV PATH="/opt/pnpm/node_modules/.bin:$PATH" COPY package.json pnpm-lock.yaml pnpm-workspace.yaml tsconfig.base.json ./ RUN npm install --prefix /opt/pnpm --ignore-scripts --no-audit --no-fund "$(node -p 'require("./package.json").packageManager.split("+")[0]')" \ && npm audit signatures --prefix /opt/pnpm +COPY packages/push-contract/package.json packages/push-contract/package.json +COPY packages/connector-contract/package.json packages/connector-contract/package.json COPY packages/postgres-schema/package.json packages/postgres-schema/package.json COPY packages/cloud-database/package.json packages/cloud-database/package.json COPY apps/connector/package.json apps/connector/package.json RUN pnpm install --frozen-lockfile +COPY packages/push-contract packages/push-contract +COPY packages/connector-contract packages/connector-contract COPY packages/postgres-schema packages/postgres-schema COPY packages/cloud-database packages/cloud-database COPY apps/connector apps/connector -RUN pnpm --filter @orca-cloud/postgres-schema build && pnpm --filter @orca-cloud/cloud-database build \ +RUN pnpm --filter @orca-cloud/postgres-schema build && pnpm --filter @orca-cloud/push-contract build \ + && pnpm --filter @orca-cloud/connector-contract build && pnpm --filter @orca-cloud/cloud-database build \ && pnpm --filter @orca-cloud/connector build FROM node:24.22.1-alpine@sha256:feb76650b7195319268c59cb2d419c732d6c499b13bf5296f16ab59eb1f0a4f3 AS runtime @@ -21,9 +26,13 @@ WORKDIR /app COPY --from=build /opt/pnpm /opt/pnpm ENV PATH="/opt/pnpm/node_modules/.bin:$PATH" COPY package.json pnpm-lock.yaml pnpm-workspace.yaml ./ +COPY packages/push-contract/package.json packages/push-contract/package.json +COPY packages/connector-contract/package.json packages/connector-contract/package.json COPY packages/postgres-schema/package.json packages/postgres-schema/package.json COPY packages/cloud-database/package.json packages/cloud-database/package.json COPY apps/connector/package.json apps/connector/package.json +COPY --from=build /app/packages/push-contract/dist packages/push-contract/dist +COPY --from=build /app/packages/connector-contract/dist packages/connector-contract/dist COPY --from=build /app/packages/postgres-schema/dist packages/postgres-schema/dist COPY --from=build /app/packages/cloud-database/dist packages/cloud-database/dist COPY --from=build /app/apps/connector/dist apps/connector/dist diff --git a/cloud/apps/connector/package.json b/cloud/apps/connector/package.json index 3dbe175e985..5ece592b894 100644 --- a/cloud/apps/connector/package.json +++ b/cloud/apps/connector/package.json @@ -9,7 +9,7 @@ "clean": "node -e \"require('fs').rmSync('dist', { recursive: true, force: true })\"", "dev": "tsx watch src/index.ts", "lint": "tsc -p tsconfig.json --noEmit", - "pretest": "pnpm --filter @orca-cloud/postgres-schema build && pnpm --filter @orca-cloud/cloud-database build", + "pretest": "pnpm --filter @orca-cloud/postgres-schema build && pnpm --filter @orca-cloud/push-contract build && pnpm --filter @orca-cloud/connector-contract build && pnpm --filter @orca-cloud/cloud-database build", "start": "node dist/index.js", "test": "vitest run", "typecheck": "tsc -p tsconfig.json --noEmit" @@ -17,6 +17,7 @@ "dependencies": { "@hono/node-server": "^2.1.5", "@orca-cloud/cloud-database": "workspace:*", + "@orca-cloud/connector-contract": "workspace:*", "hono": "^4.13.13", "jose": "^6.2.13", "zod": "^4.6.7" diff --git a/cloud/apps/connector/src/connector-background.ts b/cloud/apps/connector/src/connector-background.ts new file mode 100644 index 00000000000..1e9b7c80332 --- /dev/null +++ b/cloud/apps/connector/src/connector-background.ts @@ -0,0 +1,58 @@ +// Chains each sweep after the previous one settles, so a slow database never stacks them. +export function every(label: string, intervalMs: number, run: () => Promise): () => void { + let timer: NodeJS.Timeout | undefined + let stopped = false + const schedule = (): void => { + if (stopped) { + return + } + timer = setTimeout(() => { + void run() + .catch((error: unknown) => { + console.warn( + JSON.stringify({ + event: 'orca_connector_sweep_failed', + target: label, + error: error instanceof Error ? error.name : 'unknown' + }) + ) + }) + .finally(schedule) + }, intervalMs) + timer.unref() + } + schedule() + return () => { + stopped = true + clearTimeout(timer) + } +} + +const EXPIRY_SWEEP_INTERVAL_MS = 15_000 +const INBOX_PRUNE_INTERVAL_MS = 60_000 +// Settled rows keep only ids and results; this bounds how long even those stay. +const INBOX_RETENTION_MS = 7 * 24 * 60 * 60_000 + +export function startConnectorBackground(runtime: { + inbox: { + expireDue(): Promise<{ hostId: string; externalId: string }[]> + pruneSettled(retentionMs: number): Promise + } +}): () => void { + const stops = [ + every('inbox_expiry', EXPIRY_SWEEP_INTERVAL_MS, async () => { + const threads = await runtime.inbox.expireDue() + if (threads.length > 0) { + console.log(JSON.stringify({ event: 'orca_connector_deliveries_expired', threads: threads.length })) + } + }), + every('inbox_retention', INBOX_PRUNE_INTERVAL_MS, () => + runtime.inbox.pruneSettled(INBOX_RETENTION_MS) + ) + ] + return () => { + for (const stop of stops) { + stop() + } + } +} diff --git a/cloud/apps/connector/src/connector-database.ts b/cloud/apps/connector/src/connector-database.ts index f2976dfb863..3177e5acec6 100644 --- a/cloud/apps/connector/src/connector-database.ts +++ b/cloud/apps/connector/src/connector-database.ts @@ -4,8 +4,7 @@ import { type CloudDatabase, type OpenCloudDatabaseInput } from '@orca-cloud/cloud-database' - -const CONNECTOR_SCHEMA_STATEMENTS: string[] = [] +import { connectorSchemaStatements } from './connector-schema.js' export async function openConnectorDatabase( input: Omit @@ -13,10 +12,10 @@ export async function openConnectorDatabase( return await openCloudDatabase({ ...input, service: 'connector', - schemaStatements: CONNECTOR_SCHEMA_STATEMENTS + schemaStatements: connectorSchemaStatements() }) } export async function openInMemoryConnectorDatabase(): Promise { - return await openInMemoryCloudDatabase(CONNECTOR_SCHEMA_STATEMENTS) + return await openInMemoryCloudDatabase(connectorSchemaStatements()) } diff --git a/cloud/apps/connector/src/connector-events.test.ts b/cloud/apps/connector/src/connector-events.test.ts new file mode 100644 index 00000000000..c7ed62c269a --- /dev/null +++ b/cloud/apps/connector/src/connector-events.test.ts @@ -0,0 +1,295 @@ +import { CONNECTOR_PATHS, ConnectorPollResponseSchema } from '@orca-cloud/connector-contract' +import { afterEach, describe, expect, it } from 'vitest' +import { listenHostname } from './connector-server.js' +import { + cleanupConnectorHarnesses, + createConnectorHarness, + pauseBefore +} from './connector-server-harness.test-fixture.js' + +afterEach(cleanupConnectorHarnesses) + +const conversation = { source: 'slack', externalId: 'T1:C1:1.0:U1' } + +// A loopback origin turns on the dev enqueue route, which also creates the thread. +async function setup(options: Parameters[1] = {}) { + const harness = await createConnectorHarness({ publicUrl: 'http://127.0.0.1:8080' }, options) + const machine = await harness.machine() + await machine.hello() + const enqueue = (hostId: string) => + harness.json('/dev/enqueue', { + hostId, + externalId: conversation.externalId, + delivery: { kind: 'start', agent: 'claude', placeId: 'pl_1', prompt: 'fix it' } + }) + let nextSeq = 0 + const events = (list: unknown[], seq = nextSeq++, helloRev = 1) => + machine.call(CONNECTOR_PATHS.events, { v: 1, helloRev, seq, events: list }) + const thread = async () => + ( + await harness.database.query( + `SELECT pane_key, agent, open_turn_id, turn_state, turn_stop_reason, + pending_prompt FROM connector_threads` + ) + )[0] + return { harness, machine, enqueue, events, thread } +} + +describe('connector turn events', () => { + it('queues through the dev route only on a loopback origin', async () => { + const remote = await createConnectorHarness() + expect((await remote.json('/dev/enqueue', {})).status).toBe(404) + const { machine, enqueue } = await setup() + const queued = await enqueue(machine.hostId) + expect(await queued.json()).toMatchObject({ status: 'queued' }) + const reply = ConnectorPollResponseSchema.parse((await machine.poll()).body) + expect(reply.deliveries[0]).toMatchObject({ kind: 'start', conversation }) + }) + + it('listens only on loopback when the dev route is on', async () => { + expect(listenHostname('http://127.0.0.1:8080')).toBe('127.0.0.1') + expect(listenHostname('http://[::1]:8080')).toBe('::1') + expect(listenHostname('https://connector.onorca.dev')).toBeUndefined() + const { harness } = await setup() + const { server } = harness.runtime + await new Promise((resolve) => server.listen(0, listenHostname('http://127.0.0.1:8080'), resolve)) + expect(server.address()).toMatchObject({ address: '127.0.0.1' }) + await new Promise((resolve) => server.close(resolve)) + }) + + it('applies turn events once, whatever order and repeats they arrive in', async () => { + const { machine, enqueue, events, thread } = await setup() + await enqueue(machine.hostId) + const linked = { conversation, paneKey: 'pane-1', kind: 'linked', agent: 'claude' } + const started = { conversation, paneKey: 'pane-1', kind: 'turn-started', at: 100 } + const ended = { + conversation, + paneKey: 'pane-1', + kind: 'turn-ended', + turnId: 100, + at: 160, + stopReason: 'end_turn', + lastAssistantMessage: 'done' + } + expect((await events([linked, started, ended])).status).toBe(204) + const settled = { + pane_key: 'pane-1', + agent: 'claude', + open_turn_id: 100, + turn_state: 'ended', + turn_stop_reason: 'end_turn', + pending_prompt: null + } + const row = await thread() + // Postgres returns BIGINT as text. + expect({ ...row, open_turn_id: Number(row?.open_turn_id) }).toEqual(settled) + // A resent start for the same turn changes nothing. + await events([started, ended, started]) + expect(await thread()).toMatchObject({ turn_state: 'ended', turn_stop_reason: 'end_turn' }) + }) + + it('reads an open turn the snapshot no longer shows as unverifiable, never as done', async () => { + const { machine, enqueue, events, thread } = await setup() + await enqueue(machine.hostId) + await events([{ conversation, paneKey: 'pane-1', kind: 'turn-started', at: 100 }]) + const prompt = { kind: 'approval', promptId: 'p1', tool: 'Bash', options: [] } + await events([ + { + kind: 'snapshot', + conversations: [{ conversation, paneKey: 'pane-1', agent: 'claude', working: false, prompt }] + } + ]) + expect(await thread()).toMatchObject({ + turn_state: 'unverifiable', + pending_prompt: JSON.stringify(prompt) + }) + }) + + it('keeps a replayed turn end of an old pane off the pane a snapshot rebound', async () => { + const { machine, enqueue, events, thread } = await setup() + await enqueue(machine.hostId) + await events([{ conversation, paneKey: 'old-pane', kind: 'turn-started', at: 100 }]) + await events([ + { + kind: 'snapshot', + conversations: [ + { conversation, paneKey: 'new-pane', agent: 'claude', working: false, prompt: null } + ] + } + ]) + const oldEnd = { conversation, paneKey: 'old-pane', kind: 'turn-ended', turnId: 100, at: 160 } + expect((await events([{ ...oldEnd, stopReason: 'end_turn' }])).status).toBe(204) + expect(await thread()).toMatchObject({ pane_key: 'new-pane', turn_state: 'unverifiable' }) + // The new pane's own turn applies, whatever its start time against the old pane's. + await events([{ conversation, paneKey: 'new-pane', kind: 'turn-started', at: 90 }]) + expect(await thread()).toMatchObject({ pane_key: 'new-pane', turn_state: 'working' }) + }) + + it("drops the old pane's prompt when a conversation links to another pane", async () => { + const { machine, enqueue, events, thread } = await setup() + await enqueue(machine.hostId) + const prompt = { kind: 'approval', promptId: 'p1', tool: 'Bash', options: [] } + await events([{ conversation, paneKey: 'old-pane', kind: 'prompt', prompt }]) + await events([{ conversation, paneKey: 'old-pane', kind: 'linked', agent: 'claude' }]) + expect(await thread()).toMatchObject({ pending_prompt: JSON.stringify(prompt) }) + await events([{ conversation, paneKey: 'new-pane', kind: 'linked', agent: 'claude' }]) + expect(await thread()).toMatchObject({ pane_key: 'new-pane', pending_prompt: null }) + }) + + it("drops the old pane's completed turn when a conversation moves to another pane", async () => { + const { machine, enqueue, events, thread } = await setup() + await enqueue(machine.hostId) + const ended = { conversation, paneKey: 'old-pane', kind: 'turn-ended', turnId: 100, at: 160 } + await events([{ conversation, paneKey: 'old-pane', kind: 'turn-started', at: 100 }]) + await events([{ ...ended, stopReason: 'end_turn' }]) + await events([{ conversation, paneKey: 'old-pane', kind: 'linked', agent: 'claude' }]) + expect(await thread()).toMatchObject({ turn_state: 'ended', turn_stop_reason: 'end_turn' }) + await events([{ conversation, paneKey: 'new-pane', kind: 'linked', agent: 'claude' }]) + const cleared = { open_turn_id: null, turn_state: null, turn_stop_reason: null } + expect(await thread()).toMatchObject({ pane_key: 'new-pane', ...cleared }) + await events([{ conversation, paneKey: 'new-pane', kind: 'turn-started', at: 200 }]) + await events([ + { + kind: 'snapshot', + conversations: [ + { conversation, paneKey: 'newer-pane', agent: 'claude', working: true, prompt: null } + ] + } + ]) + // The snapshot moved it off a working turn: unverifiable, never the old verdict. + expect(await thread()).toMatchObject({ + pane_key: 'newer-pane', + open_turn_id: null, + turn_state: 'unverifiable', + turn_stop_reason: null + }) + }) + + it('clears the prompt of a conversation the snapshot no longer lists', async () => { + const { machine, enqueue, events, thread } = await setup() + await enqueue(machine.hostId) + const prompt = { kind: 'approval', promptId: 'p1', tool: 'Bash', options: [] } + await events([{ conversation, paneKey: 'pane-1', kind: 'prompt', prompt }]) + await events([{ kind: 'snapshot', conversations: [] }]) + expect(await thread()).toMatchObject({ pending_prompt: null }) + }) + + it('reads an open turn missing from the snapshot as unverifiable', async () => { + const { machine, enqueue, events, thread } = await setup() + await enqueue(machine.hostId) + await events([{ conversation, paneKey: 'pane-1', kind: 'turn-started', at: 100 }]) + await events([{ kind: 'snapshot', conversations: [] }]) + expect(await thread()).toMatchObject({ turn_state: 'unverifiable' }) + }) + + it("drops a batch at or below the host's high-water mark", async () => { + const { machine, enqueue, events, thread } = await setup() + await enqueue(machine.hostId) + const prompt = { kind: 'approval', promptId: 'p1', tool: 'Bash', options: [] } + const promptEvent = { conversation, paneKey: 'pane-1', kind: 'prompt', prompt } + const ended = { + conversation, + paneKey: 'pane-1', + kind: 'turn-ended', + turnId: 100, + at: 160, + stopReason: 'end_turn' + } + // Batch 1 (the prompt) lands late, after batch 2 cleared it. + expect((await events([ended], 2)).status).toBe(204) + expect((await events([promptEvent], 1)).status).toBe(204) + expect(await thread()).toMatchObject({ pending_prompt: null, turn_state: 'ended' }) + // A retry of the applied batch changes nothing either. + await events([promptEvent], 2) + expect(await thread()).toMatchObject({ pending_prompt: null }) + }) + + it('keeps the high-water mark across tokens for the same machine', async () => { + const { harness, machine, enqueue, events, thread } = await setup() + await enqueue(machine.hostId) + await events([{ conversation, paneKey: 'pane-1', kind: 'turn-started', at: 100 }], 5) + const refreshed = await harness.machine('user-1', { machineId: machine.machineId }) + const batch = (seq: number, paneKey: string) => + refreshed.call(CONNECTOR_PATHS.events, { + v: 1, + helloRev: 1, + seq, + events: [{ conversation, paneKey, kind: 'linked', agent: 'claude' }] + }) + expect((await batch(5, 'pane-2')).status).toBe(204) + expect(await thread()).toMatchObject({ pane_key: 'pane-1' }) + await batch(6, 'pane-3') + expect(await thread()).toMatchObject({ pane_key: 'pane-3' }) + }) + + it('applies events only for the current hello and restarts the mark with a new one', async () => { + const { machine, enqueue, events, thread } = await setup() + await enqueue(machine.hostId) + const linked = (paneKey: string) => [{ conversation, paneKey, kind: 'linked', agent: 'claude' }] + await events(linked('pane-1'), 5) + expect((await events(linked('pane-x'), 0, 2)).status).toBe(409) + await machine.hello(2) + // The new hello's batch 0 applies; a late batch from the old hello changes nothing. + expect((await events(linked('pane-2'), 0, 2)).status).toBe(204) + expect((await events(linked('pane-old'), 6, 1)).status).toBe(204) + expect(await thread()).toMatchObject({ pane_key: 'pane-2' }) + // A repeated hello for the same revision keeps the mark. + await machine.hello(2) + await events(linked('pane-retry'), 0, 2) + expect(await thread()).toMatchObject({ pane_key: 'pane-2' }) + }) + + it('keeps the newer hello when an older one arrives late', async () => { + const { harness, machine, enqueue, events, thread } = await setup() + await enqueue(machine.hostId) + await machine.hello(2) + const linked = (paneKey: string) => [{ conversation, paneKey, kind: 'linked', agent: 'claude' }] + await events(linked('new-pane'), 3, 2) + expect((await machine.hello(1)).status).toBe(204) + expect((await events(linked('old-pane'), 0, 1)).status).toBe(204) + expect(await thread()).toMatchObject({ pane_key: 'new-pane' }) + const [host] = await harness.database.query('SELECT hello_rev, events_seq FROM connector_hosts') + expect({ helloRev: Number(host?.hello_rev), eventsSeq: Number(host?.events_seq) }).toEqual({ + helloRev: 2, + eventsSeq: 3 + }) + expect((await machine.poll(2)).status).toBe(200) + }) + + // SQLite runs one transaction at a time, so only Postgres can race a batch with a hello. + it.skipIf(!process.env.ORCA_CONNECTOR_TEST_DATABASE_URL)( + 'ignores a batch from the old hello that waited on a newer hello', + async () => { + let gate!: ReturnType + const { machine, enqueue, events, thread } = await setup({ + wrapDatabase: (database) => { + gate = pauseBefore(database, 'INSERT INTO connector_hosts', 1) + return gate.database + } + }) + await enqueue(machine.hostId) + const hello = machine.hello(2) + await gate.atStatement + const late = events([{ conversation, paneKey: 'pane-old', kind: 'linked', agent: 'claude' }], 0, 1) + await new Promise((resolve) => setTimeout(resolve, 200)) + gate.release() + expect((await hello).status).toBe(204) + expect((await late).status).toBe(204) + expect(await thread()).toMatchObject({ pane_key: null }) + } + ) + + it("ignores events for another host's thread", async () => { + const { harness, machine, enqueue, thread } = await setup() + await enqueue(machine.hostId) + const other = await harness.machine('user-1') + await other.hello() + await other.call(CONNECTOR_PATHS.events, { + v: 1, + helloRev: 1, + seq: 0, + events: [{ conversation, paneKey: 'pane-9', kind: 'turn-started', at: 100 }] + }) + expect(await thread()).toMatchObject({ pane_key: null, turn_state: null }) + }) +}) diff --git a/cloud/apps/connector/src/connector-hosts.ts b/cloud/apps/connector/src/connector-hosts.ts new file mode 100644 index 00000000000..6e694e24e11 --- /dev/null +++ b/cloud/apps/connector/src/connector-hosts.ts @@ -0,0 +1,114 @@ +import type { CloudDatabase } from '@orca-cloud/cloud-database' +import type { ConnectorHelloRequest } from '@orca-cloud/connector-contract' +import type { ConnectorInbox } from './connector-inbox.js' + +// Coarse on purpose: last-seen is written at most once a minute per host. +const LAST_SEEN_RESOLUTION_MS = 60_000 + +// Every machine request and queueing take this, so a newer hello lands wholly before or after each. +export async function lockConnectorHost(transaction: CloudDatabase, hostId: string): Promise { + await transaction.lockQuotaScope(`orca-connector-host:${hostId}`) +} + +// A host is one machine of one account, (sub, mid) from its token; its first hello records it. +export class ConnectorHosts { + constructor( + private readonly database: CloudDatabase, + private readonly inbox: ConnectorInbox, + private readonly now: () => number + ) {} + + // Under the host lock. The highest revision wins (a delayed older hello changes nothing); held polls + // wake to find theirs stale, the next poll lists what was received but never settled, and a new + // revision restarts the events mark. + async recordHello( + transaction: CloudDatabase, + host: { hostId: string; accountId: string }, + hello: ConnectorHelloRequest + ): Promise { + const now = this.now() + const [recorded] = await transaction.query( + `INSERT INTO connector_hosts + (host_id, account_id, host_name, platform, runtime, orca_version, hello_rev, hello, + unsettled_due, first_seen_at, last_seen_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?) + ON CONFLICT (host_id) DO UPDATE SET host_name = excluded.host_name, + platform = excluded.platform, runtime = excluded.runtime, + orca_version = excluded.orca_version, hello_rev = excluded.hello_rev, + hello = excluded.hello, unsettled_due = 1, last_seen_at = excluded.last_seen_at, + events_seq = CASE WHEN connector_hosts.hello_rev = excluded.hello_rev + THEN connector_hosts.events_seq END + WHERE connector_hosts.hello_rev <= excluded.hello_rev`, + [ + host.hostId, + host.accountId, + hello.hostName, + hello.platform, + hello.runtime, + hello.orcaVersion, + hello.helloRev, + JSON.stringify(hello), + now, + now + ] + ) + if (Number(recorded?.changes ?? 0) > 0) { + await this.inbox.wake(transaction, host.hostId) + } + } + + // Under the host lock: a request's hello is the machine's current one, an older one, or one the + // cloud has no record of (the machine must send it). + async helloStatus( + transaction: CloudDatabase, + hostId: string, + helloRev: number + ): Promise<'current' | 'older' | 'unknown'> { + const [row] = await transaction.query( + 'SELECT hello_rev FROM connector_hosts WHERE host_id = ?', + [hostId] + ) + const current = row ? Number(row.hello_rev) : null + if (current === helloRev) { + return 'current' + } + return current !== null && helloRev < current ? 'older' : 'unknown' + } + + // Under the host lock: true once per hello, for the poll that lists unsettled deliveries. + async takeUnsettledDue(transaction: CloudDatabase, hostId: string): Promise { + const [taken] = await transaction.query( + 'UPDATE connector_hosts SET unsettled_due = 0 WHERE host_id = ? AND unsettled_due = 1', + [hostId] + ) + return Number(taken?.changes ?? 0) > 0 + } + + async accountOf(transaction: CloudDatabase, hostId: string): Promise { + const [row] = await transaction.query( + 'SELECT account_id FROM connector_hosts WHERE host_id = ?', + [hostId] + ) + return row ? String(row.account_id) : null + } + + // Under the host lock, for the current hello: events apply only above its high-water mark, so a + // late original never undoes newer state. + async advanceEventsSeq(transaction: CloudDatabase, hostId: string, seq: number): Promise { + const [advanced] = await transaction.query( + `UPDATE connector_hosts SET events_seq = ? + WHERE host_id = ? AND (events_seq IS NULL OR events_seq < ?)`, + [seq, hostId, seq] + ) + return Number(advanced?.changes ?? 0) > 0 + } + + async touch(hostId: string): Promise { + const now = this.now() + await this.database.query( + `UPDATE connector_hosts SET last_seen_at = ? + WHERE host_id = ? AND last_seen_at < ?`, + [now, hostId, now - LAST_SEEN_RESOLUTION_MS] + ) + } +} diff --git a/cloud/apps/connector/src/connector-inbox.test.ts b/cloud/apps/connector/src/connector-inbox.test.ts new file mode 100644 index 00000000000..034ed15c135 --- /dev/null +++ b/cloud/apps/connector/src/connector-inbox.test.ts @@ -0,0 +1,432 @@ +import { CONNECTOR_LIMITS, ConnectorPollResponseSchema } from '@orca-cloud/connector-contract' +import { afterEach, describe, expect, it } from 'vitest' +import type { CloudDatabase } from '@orca-cloud/cloud-database' +import type { DeliveryBody } from './connector-inbox.js' +import { + cleanupConnectorHarnesses, + createConnectorHarness, + pauseBefore +} from './connector-server-harness.test-fixture.js' + +afterEach(cleanupConnectorHarnesses) + +const START: DeliveryBody = { kind: 'start', agent: 'claude', placeId: 'pl_1', prompt: 'fix it' } +const SEND: DeliveryBody = { kind: 'send', text: 'and the tests' } + +// Fails the first statement starting with `prefix` once armed, as a transient database fault. +function failOnce(database: CloudDatabase, prefix: string) { + let armed = false + const wrap = (inner: CloudDatabase): CloudDatabase => ({ + dialect: inner.dialect, + query: async (sql, params) => { + if (armed && sql.trimStart().startsWith(prefix)) { + armed = false + throw new Error('injected') + } + return await inner.query(sql, params) + }, + transaction: (run) => inner.transaction((transaction) => run(wrap(transaction))), + lockQuotaScope: (key) => inner.lockQuotaScope(key), + tryLockScope: (key) => inner.tryLockScope(key), + tryLockSharedScope: (key) => inner.tryLockSharedScope(key), + close: () => inner.close() + }) + return { database: wrap(database), arm: () => (armed = true) } +} + +async function setup( + options: Parameters[1] = {}, + machineOptions: Parameters>['machine']>[1] = {} +) { + const harness = await createConnectorHarness({}, options) + const machine = await harness.machine('user-1', machineOptions) + expect((await machine.hello()).status).toBe(204) + const enqueue = (externalId: string, deliveryId: string, body: DeliveryBody) => + harness.runtime.queueDelivery({ + thread: { + externalId, + teamId: 'T1', + channelId: 'C1', + threadTs: externalId, + ownerUserId: 'U1', + hostId: machine.hostId, + placeId: 'pl_1' + }, + deliveryId, + body + }) + const poll = async () => { + const { status, body } = await machine.poll() + expect(status).toBe(200) + return ConnectorPollResponseSchema.parse(body) + } + const ids = async () => (await poll()).deliveries.map((delivery) => delivery.deliveryId) + return { harness, machine, enqueue, poll, ids } +} + +describe('connector inbox', () => { + it('refuses a request without a connector token', async () => { + const harness = await createConnectorHarness() + expect((await harness.json('/v1/connector/poll', { v: 1, helloRev: 1 })).status).toBe(401) + }) + + it('names a host by account and machine, whichever token it presents', async () => { + const { harness, machine, enqueue, ids } = await setup() + await enqueue('a', 'a1', START) + const sameMachine = await harness.machine('user-1', { machineId: machine.machineId }) + expect(ConnectorPollResponseSchema.parse((await sameMachine.poll()).body).deliveries).toHaveLength(1) + const otherAccount = await harness.machine('user-2', { machineId: machine.machineId }) + expect((await otherAccount.poll()).status).toBe(409) + expect(await ids()).toEqual([]) + }) + + it('requires a hello for the revision a poll names', async () => { + const harness = await createConnectorHarness() + const machine = await harness.machine() + expect((await machine.poll()).status).toBe(409) + await machine.hello(1) + expect((await machine.poll(2)).status).toBe(409) + expect((await machine.poll(1)).status).toBe(200) + }) + + it('hands out a queued delivery with its envelope', async () => { + const { harness, enqueue, poll } = await setup() + await expect(enqueue('T1:C1:1.0:U1', 'd1', START)).resolves.toBe('queued') + const reply = await poll() + expect(reply.holdMs).toBe(0) + expect(reply.serverNow).toBe(harness.now()) + expect(reply.deliveries).toEqual([ + { + ...START, + deliveryId: 'd1', + seq: 1, + expiresAt: harness.now() + CONNECTOR_LIMITS.deliveryTtlMs, + conversation: { source: 'slack', externalId: 'T1:C1:1.0:U1' } + } + ]) + await expect(enqueue('T1:C1:1.0:U1', 'd1', START)).resolves.toBe('duplicate') + }) + + it('wakes a held poll as soon as a delivery is queued', async () => { + const { enqueue, poll } = await setup({ pollHoldMs: 10_000 }) + const held = poll() + await new Promise((resolve) => setTimeout(resolve, 100)) + const startedAt = Date.now() + await enqueue('T1:C1:1.0:U1', 'd1', START) + expect((await held).deliveries.map((delivery) => delivery.deliveryId)).toEqual(['d1']) + expect(Date.now() - startedAt).toBeLessThan(5_000) + }) + + it('sends one delivery per conversation until it is received', async () => { + const { machine, enqueue, ids } = await setup() + await enqueue('a', 'a1', START) + await enqueue('a', 'a2', SEND) + await enqueue('b', 'b1', START) + expect(await ids()).toEqual(['a1', 'b1']) + expect(await ids()).toEqual([]) + await machine.ack([{ deliveryId: 'a1', step: 'received' }]) + expect(await ids()).toEqual(['a2']) + }) + + it('resends a delivery with the same id once its lease runs out unreceived', async () => { + const { harness, enqueue, ids } = await setup() + await enqueue('a', 'a1', START) + expect(await ids()).toEqual(['a1']) + harness.advance(CONNECTOR_LIMITS.leaseMs - 1) + expect(await ids()).toEqual([]) + harness.advance(1) + expect(await ids()).toEqual(['a1']) + }) + + it('lists received-but-not-done deliveries on the first poll after each hello only', async () => { + const { machine, enqueue, poll } = await setup() + await enqueue('a', 'a1', START) + await enqueue('b', 'b1', START) + await poll() + await machine.ack([ + { deliveryId: 'a1', step: 'received' }, + { deliveryId: 'b1', step: 'received' }, + { deliveryId: 'b1', step: 'done', result: { outcome: 'accepted' } } + ]) + expect((await poll()).unsettled).toBeUndefined() + + // The machine restarts: a new hello, then its first poll. + await machine.hello(2) + const first = ConnectorPollResponseSchema.parse((await machine.poll(2)).body) + expect(first.unsettled).toEqual([ + { deliveryId: 'a1', conversation: { source: 'slack', externalId: 'a' } } + ]) + const second = ConnectorPollResponseSchema.parse((await machine.poll(2)).body) + expect(second.unsettled).toBeUndefined() + }) + + // SQLite runs one transaction at a time, so only Postgres can interleave the sweep with a claim. + it.skipIf(!process.env.ORCA_CONNECTOR_TEST_DATABASE_URL)( + 'keeps an expiry that lands while a resend is being claimed', + async () => { + let gate!: ReturnType + const { harness, enqueue, poll } = await setup({ + wrapDatabase: (database) => { + gate = pauseBefore(database, "UPDATE connector_inbox SET state = 'leased'", 1) + return gate.database + } + }) + await enqueue('a', 'a1', START) + await poll() + harness.advance(CONNECTOR_LIMITS.leaseMs) + const resend = poll() + await gate.atStatement + // The sweep takes no host lock, so it can expire the row between the claim's read and write. + harness.advance(CONNECTOR_LIMITS.deliveryTtlMs) + expect(await harness.runtime.inbox.expireDue()).toHaveLength(1) + gate.release() + expect((await resend).deliveries).toEqual([]) + const [row] = await harness.database.query( + "SELECT state FROM connector_inbox WHERE delivery_id = 'a1'" + ) + expect(row?.state).toBe('expired') + } + ) + + it('keeps recovery for the next poll when the first one fails', async () => { + let fault!: ReturnType + const { machine, enqueue, poll } = await setup({ + wrapDatabase: (database) => { + fault = failOnce(database, 'SELECT delivery_id, external_id, seq') + return fault.database + } + }) + await enqueue('a', 'a1', START) + await poll() + await machine.ack([{ deliveryId: 'a1', step: 'received' }]) + await machine.hello(2) + fault.arm() + expect((await machine.poll(2)).status).toBe(500) + const retry = ConnectorPollResponseSchema.parse((await machine.poll(2)).body) + expect(retry.unsettled).toEqual([ + { deliveryId: 'a1', conversation: { source: 'slack', externalId: 'a' } } + ]) + }) + + it('sends a poll held across a newer hello back to hello', async () => { + const { machine } = await setup({ pollHoldMs: 10_000 }) + const held = machine.poll(1) + await new Promise((resolve) => setTimeout(resolve, 100)) + const startedAt = Date.now() + expect((await machine.hello(2)).status).toBe(204) + expect((await held).status).toBe(409) + expect(Date.now() - startedAt).toBeLessThan(5_000) + }) + + it('ignores acks from an older hello and asks for an unrecorded one', async () => { + const { harness, machine, enqueue, poll } = await setup() + await enqueue('a', 'a1', START) + await poll() + expect((await machine.ack([{ deliveryId: 'a1', step: 'received' }], 2)).status).toBe(409) + await machine.hello(2) + expect((await machine.ack([{ deliveryId: 'a1', step: 'received' }], 1)).status).toBe(204) + const [row] = await harness.database.query( + "SELECT state FROM connector_inbox WHERE delivery_id = 'a1'" + ) + expect(row?.state).toBe('leased') + }) + + // SQLite runs one transaction at a time, so only Postgres can race an ack with a hello. + it.skipIf(!process.env.ORCA_CONNECTOR_TEST_DATABASE_URL)( + 'ignores an ack from the old hello that waited on a newer hello; the lease resends it', + async () => { + let gate!: ReturnType + const { harness, machine, enqueue, poll } = await setup({ + wrapDatabase: (database) => { + gate = pauseBefore(database, 'INSERT INTO connector_hosts', 1) + return gate.database + } + }) + await enqueue('a', 'a1', START) + await poll() + const hello = machine.hello(2) + await gate.atStatement + const late = machine.ack([{ deliveryId: 'a1', step: 'received' }], 1) + await new Promise((resolve) => setTimeout(resolve, 200)) + gate.release() + expect((await hello).status).toBe(204) + expect((await late).status).toBe(204) + harness.advance(CONNECTOR_LIMITS.leaseMs) + const first = ConnectorPollResponseSchema.parse((await machine.poll(2)).body) + expect(first.deliveries.map((delivery) => delivery.deliveryId)).toEqual(['a1']) + } + ) + + // SQLite runs one transaction at a time, so only Postgres can race two hellos. + it.skipIf(!process.env.ORCA_CONNECTOR_TEST_DATABASE_URL)( + 'keeps the newer hello when an older one waited on it', + async () => { + let gate!: ReturnType + const { harness, machine } = await setup({ + wrapDatabase: (database) => { + gate = pauseBefore(database, 'INSERT INTO connector_hosts', 1) + return gate.database + } + }) + const newer = machine.hello(3) + await gate.atStatement + const older = machine.hello(2) + await new Promise((resolve) => setTimeout(resolve, 200)) + gate.release() + expect([(await newer).status, (await older).status]).toEqual([204, 204]) + const [host] = await harness.database.query('SELECT hello_rev FROM connector_hosts') + expect(Number(host?.hello_rev)).toBe(3) + expect((await machine.poll(3)).status).toBe(200) + } + ) + + it('keeps the first done', async () => { + const { harness, machine, enqueue, poll } = await setup() + await enqueue('a', 'a1', START) + await poll() + await machine.ack([{ deliveryId: 'a1', step: 'done', result: { outcome: 'accepted' } }]) + await machine.ack([{ deliveryId: 'a1', step: 'done', result: { outcome: 'unsupported' } }]) + const [row] = await harness.database.query( + "SELECT state, result, payload FROM connector_inbox WHERE delivery_id = 'a1'" + ) + expect(row).toEqual({ state: 'done', result: '{"outcome":"accepted"}', payload: null }) + }) + + it('drops a start the machine never received when it is cancelled', async () => { + const { enqueue, ids } = await setup() + await enqueue('a', 'a1', START) + await enqueue('a', 'a2', SEND) + await expect(enqueue('a', 'a3', { kind: 'cancel' })).resolves.toBe('dropped') + await expect(enqueue('a', 'a3', { kind: 'cancel' })).resolves.toBe('duplicate') + expect(await ids()).toEqual([]) + }) + + it('lets a cancel skip the queue once its start is received', async () => { + const { machine, enqueue, ids } = await setup() + await enqueue('a', 'a1', START) + expect(await ids()).toEqual(['a1']) + await machine.ack([{ deliveryId: 'a1', step: 'received' }]) + await enqueue('a', 'a2', SEND) + expect(await ids()).toEqual(['a2']) + await enqueue('a', 'a3', SEND) + await expect(enqueue('a', 'a4', { kind: 'cancel' })).resolves.toBe('queued') + expect(await ids()).toEqual(['a4']) + }) + + it('still lets a cancel skip the queue after its settled start is pruned', async () => { + const { harness, machine, enqueue, ids } = await setup() + await enqueue('a', 'a1', START) + await ids() + await machine.ack([{ deliveryId: 'a1', step: 'done', result: { outcome: 'accepted' } }]) + // What retention leaves of a long conversation: its settled start is gone. + await harness.database.query("DELETE FROM connector_inbox WHERE delivery_id = 'a1'") + await enqueue('a', 'a2', SEND) + expect(await ids()).toEqual(['a2']) + await enqueue('a', 'a3', { kind: 'cancel' }) + expect(await ids()).toEqual(['a3']) + }) + + it('queues nothing for a machine that never said hello', async () => { + const { harness, enqueue } = await setup() + const silent = await harness.machine('user-1') + await expect( + harness.runtime.queueDelivery({ + thread: { + externalId: 'a', + teamId: 'T1', + channelId: 'C1', + threadTs: 'a', + ownerUserId: 'U1', + hostId: silent.hostId, + placeId: 'pl_1' + }, + deliveryId: 'a1', + body: START + }) + ).resolves.toBe('unknown_host') + await expect(enqueue('b', 'b1', START)).resolves.toBe('queued') + }) + + it('expires pending work at its deadline and leased work 30 s after it', async () => { + const { harness, enqueue, ids } = await setup() + await enqueue('a', 'a1', START) + expect(await ids()).toEqual(['a1']) + await enqueue('b', 'b1', START) + harness.advance(CONNECTOR_LIMITS.deliveryTtlMs) + expect(await harness.runtime.inbox.expireDue()).toEqual([ + { hostId: expect.any(String), externalId: 'b' } + ]) + expect(await ids()).toEqual([]) + harness.advance(CONNECTOR_LIMITS.leasedExpiryGraceMs) + expect(await harness.runtime.inbox.expireDue()).toEqual([ + { hostId: expect.any(String), externalId: 'a' } + ]) + }) + + it('accepts a late received for expired work', async () => { + const { harness, machine, enqueue, ids } = await setup() + await enqueue('a', 'a1', START) + await ids() + harness.advance(CONNECTOR_LIMITS.deliveryTtlMs + CONNECTOR_LIMITS.leasedExpiryGraceMs) + await harness.runtime.inbox.expireDue() + expect((await machine.ack([{ deliveryId: 'a1', step: 'received' }])).status).toBe(204) + const [row] = await harness.database.query( + "SELECT state FROM connector_inbox WHERE delivery_id = 'a1'" + ) + expect(row?.state).toBe('received') + }) + + it('ends a held poll by its token expiry', async () => { + const { machine, poll } = await setup({ pollHoldMs: 10_000 }, { ttlSeconds: 2 }) + const reply = await poll() + expect(reply.deliveries).toEqual([]) + expect(Date.now()).toBeGreaterThanOrEqual(machine.expiresAt) + expect(Date.now() - machine.expiresAt).toBeLessThan(1_000) + }) + + it('rolls back a poll pass whose token expired while it ran', async () => { + let gate!: ReturnType + const { harness, machine, enqueue } = await setup( + { + wrapDatabase: (database) => { + gate = pauseBefore(database, 'SELECT delivery_id, external_id, seq') + return gate.database + } + }, + { ttlSeconds: 2 } + ) + await enqueue('a', 'a1', START) + const reply = machine.poll() + await gate.atStatement + await new Promise((resolve) => setTimeout(resolve, machine.expiresAt - Date.now() + 10)) + gate.release() + expect(ConnectorPollResponseSchema.parse((await reply).body).deliveries).toEqual([]) + const [row] = await harness.database.query( + "SELECT state FROM connector_inbox WHERE delivery_id = 'a1'" + ) + expect(row?.state).toBe('pending') + const [host] = await harness.database.query('SELECT unsettled_due FROM connector_hosts') + expect(Number(host?.unsettled_due)).toBe(1) + }) + + it('wakes a held poll when a lease runs out', async () => { + const { harness, enqueue, ids, poll } = await setup({ pollHoldMs: 10_000 }) + await enqueue('a', 'a1', START) + expect(await ids()).toEqual(['a1']) + harness.advance(CONNECTOR_LIMITS.leaseMs - 100) + const startedAt = Date.now() + const held = poll() + setTimeout(() => harness.advance(100), 50) + expect((await held).deliveries.map((delivery) => delivery.deliveryId)).toEqual(['a1']) + expect(Date.now() - startedAt).toBeLessThan(5_000) + }) + + it('answers held polls at once with a reconnect when the instance drains', async () => { + const { harness, machine } = await setup({ pollHoldMs: 10_000 }) + const held = machine.poll() + await new Promise((resolve) => setTimeout(resolve, 100)) + harness.runtime.waker.drain() + const reply = ConnectorPollResponseSchema.parse((await held).body) + expect(reply.control).toEqual({ reconnect: true }) + }) +}) diff --git a/cloud/apps/connector/src/connector-inbox.ts b/cloud/apps/connector/src/connector-inbox.ts new file mode 100644 index 00000000000..c52c62d789c --- /dev/null +++ b/cloud/apps/connector/src/connector-inbox.ts @@ -0,0 +1,209 @@ +import type { CloudDatabase, CloudNotifications } from '@orca-cloud/cloud-database' +import { + CONNECTOR_LIMITS, + ConnectorDeliverySchema, + type ConnectorAcksRequest, + type ConnectorDelivery +} from '@orca-cloud/connector-contract' + +// NOTIFY channel; the payload is the host id whose inbox changed. +export const INBOX_CHANNEL = 'connector_inbox' + +type DeliveryEnvelope = 'deliveryId' | 'seq' | 'expiresAt' | 'conversation' +// The kind-specific part of a delivery; the inbox assigns the envelope. +export type DeliveryBody = ConnectorDelivery extends infer D + ? D extends ConnectorDelivery + ? Omit + : never + : never + +export type EnqueueResult = 'queued' | 'duplicate' | 'dropped' +export type Unsettled = { deliveryId: string; conversation: { source: 'slack'; externalId: string } } + +const conversation = (externalId: string) => ({ source: 'slack' as const, externalId }) + +// States: pending -> leased -> received -> done, or expired / dropped. Text leaves with settlement. +export class ConnectorInbox { + constructor( + private readonly database: CloudDatabase, + private readonly notifications: CloudNotifications, + private readonly now: () => number + ) {} + + // Under the host lock, after the caller found the host and bound the thread (queueDelivery). + async enqueue( + transaction: CloudDatabase, + input: { hostId: string; externalId: string; deliveryId: string; body: DeliveryBody } + ): Promise { + const { hostId, externalId, deliveryId, body } = input + const now = this.now() + const { kind, ...payload } = body + const [inserted] = await transaction.query( + `INSERT INTO connector_inbox + (host_id, delivery_id, external_id, seq, kind, payload, state, expires_at, created_at) + SELECT ?, ?, ?, COALESCE(MAX(seq), 0) + 1, ?, ?, 'pending', ?, ? + FROM connector_inbox WHERE host_id = ? + ON CONFLICT (host_id, delivery_id) DO NOTHING`, + [ + hostId, + deliveryId, + externalId, + kind, + JSON.stringify(payload), + now + CONNECTOR_LIMITS.deliveryTtlMs, + now, + hostId + ] + ) + if (Number(inserted?.changes ?? 0) === 0) { + return 'duplicate' + } + if (body.kind === 'cancel') { + // A start the machine never received is dropped, with everything queued behind it, this + // cancel included, so a retried cancel reads as a duplicate. + const [dropped] = await transaction.query( + `UPDATE connector_inbox SET state = 'dropped', payload = NULL, settled_at = ? + WHERE host_id = ? AND external_id = ? AND state = 'pending' + AND EXISTS (SELECT 1 FROM connector_inbox s WHERE s.host_id = connector_inbox.host_id + AND s.external_id = connector_inbox.external_id AND s.kind = 'start' + AND s.state = 'pending')`, + [now, hostId, externalId] + ) + if (Number(dropped?.changes ?? 0) > 0) { + return 'dropped' + } + } + await this.wake(transaction, hostId) + return 'queued' + } + + // Under the host lock (the poll's pass). Each conversation's head goes out; the next waits + // until the head is received. A cancel skips the queue unless its start is still unreceived. + async claim( + transaction: CloudDatabase, + hostId: string + ): Promise<{ deliveries: ConnectorDelivery[]; nextLeaseDueAt: number | null }> { + const now = this.now() + const rows = await transaction.query( + `SELECT delivery_id, external_id, seq, kind, payload, expires_at FROM connector_inbox i + WHERE i.host_id = ? AND i.expires_at > ? + AND (i.state = 'pending' OR (i.state = 'leased' AND i.lease_until <= ?)) + AND ( + NOT EXISTS (SELECT 1 FROM connector_inbox e WHERE e.host_id = i.host_id + AND e.external_id = i.external_id AND e.seq < i.seq + AND e.state IN ('pending', 'leased')) + OR (i.kind = 'cancel' AND NOT EXISTS (SELECT 1 FROM connector_inbox s + WHERE s.host_id = i.host_id AND s.external_id = i.external_id + AND s.kind = 'start' AND s.state IN ('pending', 'leased'))) + ) + ORDER BY i.seq LIMIT ?`, + [hostId, now, now, CONNECTOR_LIMITS.maxDeliveriesPerPoll] + ) + const deliveries: ConnectorDelivery[] = [] + for (const row of rows) { + // The expiry sweep doesn't take the host lock; the state guard keeps its verdict. + const [leased] = await transaction.query( + `UPDATE connector_inbox SET state = 'leased', lease_until = ? + WHERE host_id = ? AND delivery_id = ? + AND (state = 'pending' OR (state = 'leased' AND lease_until <= ?))`, + [now + CONNECTOR_LIMITS.leaseMs, hostId, row.delivery_id, now] + ) + if (Number(leased?.changes ?? 0) === 0) { + continue + } + deliveries.push( + ConnectorDeliverySchema.parse({ + ...JSON.parse(String(row.payload)), + kind: row.kind, + deliveryId: row.delivery_id, + seq: Number(row.seq), + expiresAt: Number(row.expires_at), + conversation: conversation(String(row.external_id)) + }) + ) + } + const [next] = await transaction.query( + `SELECT MIN(lease_until) AS due FROM connector_inbox + WHERE host_id = ? AND state = 'leased' AND expires_at > ?`, + [hostId, now] + ) + return { deliveries, nextLeaseDueAt: next?.due === null ? null : Number(next?.due) } + } + + // What an earlier process received but never settled; the first poll after each hello lists it. + async listUnsettled(transaction: CloudDatabase, hostId: string): Promise { + const rows = await transaction.query( + `SELECT delivery_id, external_id FROM connector_inbox + WHERE host_id = ? AND state = 'received' ORDER BY seq`, + [hostId] + ) + return rows.map((row) => ({ + deliveryId: String(row.delivery_id), + conversation: conversation(String(row.external_id)) + })) + } + + async wake(transaction: CloudDatabase, hostId: string): Promise { + await this.notifications.notify(transaction, INBOX_CHANNEL, hostId) + } + + // A late received is accepted (it did arrive); the first done wins. + async acknowledge( + transaction: CloudDatabase, + hostId: string, + acks: ConnectorAcksRequest['acks'] + ): Promise { + const now = this.now() + let changed = 0 + for (const ack of acks) { + const [result] = + ack.step === 'received' + ? await transaction.query( + `UPDATE connector_inbox SET state = 'received' + WHERE host_id = ? AND delivery_id = ? AND state IN ('pending', 'leased', 'expired')`, + [hostId, ack.deliveryId] + ) + : await transaction.query( + `UPDATE connector_inbox SET state = 'done', result = ?, payload = NULL, settled_at = ? + WHERE host_id = ? AND delivery_id = ? AND state IN ('pending', 'leased', 'received', 'expired')`, + [JSON.stringify(ack.result), now, hostId, ack.deliveryId] + ) + changed += Number(result?.changes ?? 0) + } + // A received head lets the conversation's next delivery go out. + if (changed > 0) { + await this.wake(transaction, hostId) + } + } + + // Pending past its deadline, or leased and never received 30 s after it. Returns the threads hit. + async expireDue(): Promise<{ hostId: string; externalId: string }[]> { + const now = this.now() + return await this.database.transaction(async (transaction) => { + const rows = await transaction.query( + `UPDATE connector_inbox SET state = 'expired', payload = NULL, settled_at = ? + WHERE (state = 'pending' AND expires_at <= ?) OR (state = 'leased' AND expires_at <= ?) + RETURNING host_id, external_id`, + [now, now, now - CONNECTOR_LIMITS.leasedExpiryGraceMs] + ) + const threads = new Map() + for (const row of rows) { + const thread = { hostId: String(row.host_id), externalId: String(row.external_id) } + threads.set(`${thread.hostId}\0${thread.externalId}`, thread) + } + for (const hostId of new Set([...threads.values()].map((thread) => thread.hostId))) { + await this.wake(transaction, hostId) + } + return [...threads.values()] + }) + } + + async pruneSettled(retentionMs: number): Promise { + const [result] = await this.database.query( + `DELETE FROM connector_inbox + WHERE state IN ('done', 'expired', 'dropped') AND settled_at < ?`, + [this.now() - retentionMs] + ) + return Number(result?.changes ?? 0) + } +} diff --git a/cloud/apps/connector/src/connector-schema.ts b/cloud/apps/connector/src/connector-schema.ts new file mode 100644 index 00000000000..80a349266b7 --- /dev/null +++ b/cloud/apps/connector/src/connector-schema.ts @@ -0,0 +1,65 @@ +// Applied at startup for both dialects, so every column type reads the same in SQLite and PostgreSQL. +const CONNECTOR_SCHEMA = ` +CREATE TABLE IF NOT EXISTS connector_hosts ( + host_id TEXT PRIMARY KEY, + account_id TEXT NOT NULL, + host_name TEXT NOT NULL, + platform TEXT NOT NULL, + runtime TEXT NOT NULL, + orca_version TEXT NOT NULL, + hello_rev BIGINT NOT NULL, + hello TEXT NOT NULL, + unsettled_due INTEGER NOT NULL, + events_seq BIGINT, + first_seen_at BIGINT NOT NULL, + last_seen_at BIGINT NOT NULL +); +CREATE INDEX IF NOT EXISTS connector_hosts_account ON connector_hosts(account_id); + +CREATE TABLE IF NOT EXISTS connector_threads ( + external_id TEXT PRIMARY KEY, + team_id TEXT NOT NULL, + channel_id TEXT NOT NULL, + thread_ts TEXT NOT NULL, + owner_user_id TEXT NOT NULL, + account_id TEXT NOT NULL, + host_id TEXT NOT NULL, + place_id TEXT NOT NULL, + pane_key TEXT, + agent TEXT, + open_turn_id BIGINT, + turn_state TEXT, + turn_stop_reason TEXT, + pending_prompt TEXT, + created_at BIGINT NOT NULL +); +CREATE UNIQUE INDEX IF NOT EXISTS connector_threads_slack + ON connector_threads(team_id, channel_id, thread_ts, owner_user_id); +CREATE INDEX IF NOT EXISTS connector_threads_host ON connector_threads(host_id); + +CREATE TABLE IF NOT EXISTS connector_inbox ( + host_id TEXT NOT NULL, + delivery_id TEXT NOT NULL, + external_id TEXT NOT NULL, + seq BIGINT NOT NULL, + kind TEXT NOT NULL, + payload TEXT, + state TEXT NOT NULL, + lease_until BIGINT, + expires_at BIGINT NOT NULL, + result TEXT, + created_at BIGINT NOT NULL, + settled_at BIGINT, + PRIMARY KEY (host_id, delivery_id) +); +CREATE INDEX IF NOT EXISTS connector_inbox_host_state ON connector_inbox(host_id, state, seq); +CREATE INDEX IF NOT EXISTS connector_inbox_conversation + ON connector_inbox(host_id, external_id, seq); +CREATE INDEX IF NOT EXISTS connector_inbox_expiry ON connector_inbox(state, expires_at); +` + +export function connectorSchemaStatements(): string[] { + return CONNECTOR_SCHEMA.split(';') + .map((statement) => statement.trim()) + .filter((statement) => statement.length > 0) +} diff --git a/cloud/apps/connector/src/connector-server-harness.test-fixture.ts b/cloud/apps/connector/src/connector-server-harness.test-fixture.ts index 6e8bd24567d..c8fb335ff33 100644 --- a/cloud/apps/connector/src/connector-server-harness.test-fixture.ts +++ b/cloud/apps/connector/src/connector-server-harness.test-fixture.ts @@ -1,12 +1,16 @@ -import { randomUUID } from 'node:crypto' -import type { CloudDatabase } from '@orca-cloud/cloud-database' +import { randomBytes, randomUUID } from 'node:crypto' +import { openCloudNotifications, type CloudDatabase } from '@orca-cloud/cloud-database' +import { CONNECTOR_PATHS } from '@orca-cloud/connector-contract' +import { createLocalJWKSet, exportJWK, generateKeyPair, SignJWT } from 'jose' import pg from 'pg' import type { ConnectorConfig } from './config.js' import { openConnectorDatabase, openInMemoryConnectorDatabase } from './connector-database.js' import { createConnectorServer } from './connector-server.js' +import { connectorHostId } from './connector-token-verifier.js' export const PUBLIC_URL = 'https://connector.onorca.dev' export const AUTH_ISSUER = 'https://login.onorca.dev' +export const HOST_IP = '203.0.113.7' // Real Postgres when the CI service is present; each harness gets its own schema. const testDatabaseUrl = process.env.ORCA_CONNECTOR_TEST_DATABASE_URL @@ -29,11 +33,11 @@ export async function cleanupConnectorHarnesses(): Promise { await Promise.all(cleanups.splice(0).map((cleanup) => cleanup())) } -async function openTestDatabase(): Promise { +async function openTestDatabase(): Promise<{ database: CloudDatabase; databaseUrl?: string }> { if (!testDatabaseUrl) { const database = await openInMemoryConnectorDatabase() cleanups.push(() => database.close()) - return database + return { database } } const admin = new pg.Client({ connectionString: testDatabaseUrl }) await admin.connect() @@ -47,13 +51,140 @@ async function openTestDatabase(): Promise { await admin.query(`DROP SCHEMA IF EXISTS ${schema} CASCADE`) await admin.end() }) - return database + return { database, databaseUrl: url.toString() } } -export async function createConnectorHarness(config: Partial = {}) { - const database = await openTestDatabase() - const runtime = createConnectorServer({ ...testConnectorConfig(), ...config }, database) +// Stands in for the auth service's connector token route. +async function testTokenIssuer() { + const { privateKey, publicKey } = await generateKeyPair('ES256') + const jwk = { ...(await exportJWK(publicKey)), kid: 'test-key', alg: 'ES256' } + return { + jwks: createLocalJWKSet({ keys: [jwk] }), + sign: async (claims: { sub: string; mid: string }, expiresAtSeconds: number) => + await new SignJWT({ prof: 'profile-1', sid: 'family-1', mid: claims.mid, purpose: 'connector-host' }) + .setProtectedHeader({ alg: 'ES256', kid: 'test-key' }) + .setIssuer(AUTH_ISSUER) + .setAudience('orca-connector') + .setSubject(claims.sub) + .setIssuedAt() + .setExpirationTime(expiresAtSeconds) + .sign(privateKey) + } +} + +// Holds the first transaction statement starting with `prefix` until release(), locks still held. +// `skip` lets that many matching statements through first. +export function pauseBefore(database: CloudDatabase, prefix: string, skip = 0) { + let reached!: () => void + const atStatement = new Promise((resolve) => (reached = resolve)) + let release!: () => void + const released = new Promise((resolve) => (release = resolve)) + let seen = 0 + const wrap = (inner: CloudDatabase): CloudDatabase => ({ + dialect: inner.dialect, + query: async (sql, params) => { + if (sql.trimStart().startsWith(prefix) && seen++ === skip) { + reached() + await released + } + return await inner.query(sql, params) + }, + transaction: (run) => inner.transaction((transaction) => run(wrap(transaction))), + lockQuotaScope: (key) => inner.lockQuotaScope(key), + tryLockScope: (key) => inner.tryLockScope(key), + tryLockSharedScope: (key) => inner.tryLockSharedScope(key), + close: () => inner.close() + }) + return { database: wrap(database), atStatement, release } +} + +export async function createConnectorHarness( + config: Partial = {}, + options: { + pollHoldMs?: number + wrapDatabase?: (database: CloudDatabase) => CloudDatabase + } = {} +) { + let now = 1_800_000_000_000 + const issuer = await testTokenIssuer() + const { database, databaseUrl } = await openTestDatabase() + const notifications = openCloudNotifications(databaseUrl ? { databaseUrl } : {}) + const runtime = createConnectorServer( + { ...testConnectorConfig(), ...config }, + options.wrapDatabase?.(database) ?? database, + notifications, + { now: () => now, jwks: issuer.jwks, pollHoldMs: options.pollHoldMs ?? 0 } + ) + await runtime.waker.start() + cleanups.push(() => runtime.waker.close()) const request = (path: string, init?: RequestInit): Promise => Promise.resolve(runtime.app.request(path, init)) - return { runtime, database, request } + const json = (path: string, body: unknown, headers: Record = {}) => + request(path, { + method: 'POST', + headers: { 'content-type': 'application/json', 'x-forwarded-for': HOST_IP, ...headers }, + body: JSON.stringify(body) + }) + + // A signed-in machine holding a connector token, with helpers for the machine API. + async function machine( + account = 'user-1', + { machineId = randomBytes(16).toString('base64url'), ttlSeconds = 300 } = {} + ) { + const claims = { sub: account, mid: machineId } + const expiresAt = (Math.floor(Date.now() / 1000) + ttlSeconds) * 1000 + const token = await issuer.sign(claims, expiresAt / 1000) + const call = (path: string, body: unknown, method = 'POST') => + request(path, { + method, + headers: { + 'content-type': 'application/json', + 'x-forwarded-for': HOST_IP, + authorization: `Bearer ${token}` + }, + body: JSON.stringify(body) + }) + return { + hostId: connectorHostId(claims), + machineId, + expiresAt, + call, + hello: (helloRev = 1) => + call( + CONNECTOR_PATHS.hello, + { + v: 1, + helloRev, + contract: 1, + hostName: 'build-box', + platform: 'linux', + runtime: 'serve', + orcaVersion: '1.4.230', + agents: ['claude'], + promptKinds: ['approval'], + features: ['answer', 'cancel'], + projects: [] + }, + 'PUT' + ), + poll: async (helloRev = 1) => { + const response = await call(CONNECTOR_PATHS.poll, { v: 1, helloRev }) + const body: unknown = await response.json() + return { status: response.status, body } + }, + ack: (acks: unknown[], helloRev = 1) => call(CONNECTOR_PATHS.acks, { v: 1, helloRev, acks }) + } + } + + return { + runtime, + database, + machine, + request, + json, + advance: (ms: number) => { + now += ms + }, + now: () => now + } } diff --git a/cloud/apps/connector/src/connector-server.ts b/cloud/apps/connector/src/connector-server.ts index 0ca2238045c..0d5c757aafb 100644 --- a/cloud/apps/connector/src/connector-server.ts +++ b/cloud/apps/connector/src/connector-server.ts @@ -1,28 +1,71 @@ import { createAdaptorServer } from '@hono/node-server' -import type { CloudDatabase } from '@orca-cloud/cloud-database' +import type { CloudDatabase, CloudNotifications } from '@orca-cloud/cloud-database' +import { CONNECTOR_LIMITS } from '@orca-cloud/connector-contract' +import type { JWTVerifyGetKey } from 'jose' import { Hono } from 'hono' +import { bodyLimit } from 'hono/body-limit' import type { ConnectorConfig } from './config.js' +import { ConnectorHosts } from './connector-hosts.js' +import { ConnectorInbox } from './connector-inbox.js' +import { ConnectorThreads } from './connector-threads.js' +import { createConnectorTokenVerifier } from './connector-token-verifier.js' +import { registerDevEnqueueRoute } from './dev-enqueue-route.js' +import { InboxWaker } from './inbox-waker.js' +import { machineAuth, type MachineEnv } from './machine-auth.js' +import { registerMachineRoutes } from './machine-routes.js' +import { queueDelivery } from './queue-delivery.js' import { limitByClientIp, RateLimiter } from './rate-limiter.js' import { RequestDrain } from './request-drain.js' +const MAX_BODY_BYTES = 64 * 1024 const UNAUTHENTICATED_PER_MINUTE_PER_IP = 30 +const AUTHENTICATED_PER_MINUTE_PER_IP = 6_000 +const AUTHENTICATED_PER_MINUTE_PER_HOST = 600 export type ConnectorServerOptions = { now?: () => number + // The auth service's signing keys; tests pass a local set. + jwks?: JWTVerifyGetKey + pollHoldMs?: number +} + +function isLoopbackOrigin(origin: string): boolean { + return ['127.0.0.1', 'localhost', '[::1]'].includes(new URL(origin).hostname) +} + +// Local development listens on its loopback origin only, so the unauthenticated dev route can't be +// reached from another machine; anything else listens on every interface, as Cloud Run needs. +export function listenHostname(publicUrl: string): string | undefined { + return isLoopbackOrigin(publicUrl) ? new URL(publicUrl).hostname.replace(/^\[|\]$/g, '') : undefined } export function createConnectorServer( config: ConnectorConfig, database: CloudDatabase, + notifications: CloudNotifications, options: ConnectorServerOptions = {} ) { const now = options.now ?? Date.now + const inbox = new ConnectorInbox(database, notifications, now) + const hosts = new ConnectorHosts(database, inbox, now) + const threads = new ConnectorThreads(now) + const waker = new InboxWaker(notifications) + const queue = (input: Parameters[1]) => + queueDelivery({ database, inbox, threads, hosts }, input) const limitUnauthenticated = limitByClientIp( new RateLimiter(UNAUTHENTICATED_PER_MINUTE_PER_IP, now), config.trustedProxyHops ) + const limitAuthenticated = limitByClientIp( + new RateLimiter(AUTHENTICATED_PER_MINUTE_PER_IP, now), + config.trustedProxyHops + ) + const limitBody = bodyLimit({ + maxSize: MAX_BODY_BYTES, + onError: (context) => context.json({ error: 'request_too_large' }, 413) + }) - const app = new Hono() + const app = new Hono() const requestDrain = new RequestDrain() app.use('*', requestDrain.middleware) // Default logging can carry row detail; log only the error's name. @@ -50,5 +93,35 @@ export function createConnectorServer( app.use('*', async (context) => context.json({ error: 'validation_only' }, 503)) } - return { app, requestDrain, server: createAdaptorServer(app) } + app.use( + '/v1/connector/*', + limitAuthenticated, + machineAuth({ + verifyToken: createConnectorTokenVerifier(config, options.jwks), + hostRequests: new RateLimiter(AUTHENTICATED_PER_MINUTE_PER_HOST, now) + }) + ) + registerMachineRoutes(app, { + database, + inbox, + threads, + hosts, + waker, + now, + holdMs: options.pollHoldMs ?? CONNECTOR_LIMITS.initialPollHoldMs, + limitBody + }) + // Local development only: a stand-in for Slack ingress until it exists. + if (isLoopbackOrigin(config.publicUrl)) { + registerDevEnqueueRoute(app, { queue, limitBody }) + } + + return { + app, + requestDrain, + server: createAdaptorServer(app), + inbox, + waker, + queueDelivery: queue + } } diff --git a/cloud/apps/connector/src/connector-threads.ts b/cloud/apps/connector/src/connector-threads.ts new file mode 100644 index 00000000000..53bccd07554 --- /dev/null +++ b/cloud/apps/connector/src/connector-threads.ts @@ -0,0 +1,138 @@ +import type { CloudDatabase } from '@orca-cloud/cloud-database' +import type { ConnectorEvent } from '@orca-cloud/connector-contract' + +// Linked and snapshot (re)bind a conversation's pane. Moving to another pane forgets the old pane's +// turn and its verdict; one it left open (working or unverifiable) stays unverifiable, never done. +const REBIND_PANE = `pane_key = ?, + open_turn_id = CASE WHEN pane_key = ? THEN open_turn_id END, + turn_state = CASE WHEN pane_key IS NULL OR pane_key = ? THEN turn_state + WHEN turn_state IN ('working', 'unverifiable') THEN 'unverifiable' END, + turn_stop_reason = CASE WHEN pane_key IS NULL OR pane_key = ? THEN turn_stop_reason END` +const rebindParams = (paneKey: string) => [paneKey, paneKey, paneKey, paneKey] +// Other events only touch the pane the conversation is bound to. +const ON_BOUND_PANE = 'host_id = ? AND external_id = ? AND (pane_key IS NULL OR pane_key = ?)' + +export type NewThread = { + externalId: string + teamId: string + channelId: string + threadTs: string + ownerUserId: string + accountId: string + hostId: string + placeId: string +} + +// Each Slack thread is bound to one host for life; events only ever touch that host's threads. +export class ConnectorThreads { + constructor(private readonly now: () => number) {} + + // Binds the thread to its host on first sight; false when it already belongs to another host. + async bind(transaction: CloudDatabase, thread: NewThread): Promise { + await transaction.query( + `INSERT INTO connector_threads + (external_id, team_id, channel_id, thread_ts, owner_user_id, account_id, host_id, place_id, + created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (external_id) DO NOTHING`, + [ + thread.externalId, + thread.teamId, + thread.channelId, + thread.threadTs, + thread.ownerUserId, + thread.accountId, + thread.hostId, + thread.placeId, + this.now() + ] + ) + const [row] = await transaction.query( + 'SELECT host_id FROM connector_threads WHERE external_id = ?', + [thread.externalId] + ) + return row?.host_id === thread.hostId + } + + // Idempotent by each event's identity: a resend writes the same state again or matches nothing. + async applyEvents( + transaction: CloudDatabase, + hostId: string, + events: ConnectorEvent[] + ): Promise { + for (const event of events) { + if (event.kind === 'snapshot') { + // Not working never means done: an open turn the snapshot doesn't show working (or doesn't + // list; it lists every linked conversation) is unverifiable, and an unlisted one has no prompt. + const notIn = (ids: string[]) => + ids.length > 0 ? ` AND external_id NOT IN (${ids.map(() => '?').join(', ')})` : '' + const working = event.conversations + .filter((entry) => entry.working) + .map((entry) => entry.conversation.externalId) + await transaction.query( + `UPDATE connector_threads SET turn_state = 'unverifiable' + WHERE host_id = ? AND turn_state = 'working'${notIn(working)}`, + [hostId, ...working] + ) + const listed = event.conversations.map((entry) => entry.conversation.externalId) + await transaction.query( + `UPDATE connector_threads SET pending_prompt = NULL WHERE host_id = ?${notIn(listed)}`, + [hostId, ...listed] + ) + for (const entry of event.conversations) { + await transaction.query( + `UPDATE connector_threads SET ${REBIND_PANE}, agent = ?, pending_prompt = ? + WHERE host_id = ? AND external_id = ?`, + [ + ...rebindParams(entry.paneKey), + entry.agent, + entry.prompt ? JSON.stringify(entry.prompt) : null, + hostId, + entry.conversation.externalId + ] + ) + } + continue + } + const where = [hostId, event.conversation.externalId] + const onPane = [...where, event.paneKey] + switch (event.kind) { + case 'linked': + await transaction.query( + `UPDATE connector_threads SET ${REBIND_PANE}, agent = ?, + pending_prompt = CASE WHEN pane_key = ? THEN pending_prompt END + WHERE host_id = ? AND external_id = ?`, + [...rebindParams(event.paneKey), event.agent, event.paneKey, ...where] + ) + break + case 'prompt': + await transaction.query( + `UPDATE connector_threads SET pane_key = ?, pending_prompt = ? WHERE ${ON_BOUND_PANE}`, + [event.paneKey, JSON.stringify(event.prompt), ...onPane] + ) + break + case 'turn-started': + await transaction.query( + `UPDATE connector_threads SET pane_key = ?, open_turn_id = ?, turn_state = 'working', + turn_stop_reason = NULL, pending_prompt = NULL + WHERE ${ON_BOUND_PANE} AND (open_turn_id IS NULL OR open_turn_id < ?)`, + [event.paneKey, event.at, ...onPane, event.at] + ) + break + case 'turn-ended': + await transaction.query( + `UPDATE connector_threads SET pane_key = ?, open_turn_id = ?, turn_state = 'ended', + turn_stop_reason = ?, pending_prompt = NULL + WHERE ${ON_BOUND_PANE} AND (open_turn_id IS NULL OR open_turn_id <= ?)`, + [ + event.paneKey, + event.turnId, + event.stopReason, + ...onPane, + event.turnId + ] + ) + break + } + } + } +} diff --git a/cloud/apps/connector/src/dev-enqueue-route.ts b/cloud/apps/connector/src/dev-enqueue-route.ts new file mode 100644 index 00000000000..803d4a73fd5 --- /dev/null +++ b/cloud/apps/connector/src/dev-enqueue-route.ts @@ -0,0 +1,64 @@ +import { randomUUID } from 'node:crypto' +import { ConnectorDeliverySchema } from '@orca-cloud/connector-contract' +import type { Env, Hono, MiddlewareHandler } from 'hono' +import { z } from 'zod' +import type { queueDelivery } from './queue-delivery.js' + +const DevEnqueueSchema = z.object({ + hostId: z.string().min(1), + externalId: z.string().min(1), + placeId: z.string().min(1).default('dev-place'), + deliveryId: z.string().min(1).optional(), + // The kind-specific delivery fields; the envelope is filled in below and checked by the contract. + delivery: z.record(z.string(), z.unknown()) +}) + +// Queues a delivery without Slack for a host that said hello; only on a loopback origin. +export function registerDevEnqueueRoute( + app: Hono, + deps: { + queue: (input: Parameters[1]) => ReturnType + limitBody: MiddlewareHandler + } +): void { + app.post('/dev/enqueue', deps.limitBody, async (context) => { + const body = DevEnqueueSchema.safeParse(await context.req.json().catch(() => null)) + if (!body.success) { + return context.json({ error: 'invalid_request' }, 400) + } + const { hostId, externalId, placeId } = body.data + const deliveryId = body.data.deliveryId ?? randomUUID() + const delivery = ConnectorDeliverySchema.safeParse({ + ...body.data.delivery, + deliveryId, + seq: 1, + expiresAt: 1, + conversation: { source: 'slack', externalId } + }) + if (!delivery.success) { + return context.json({ error: 'invalid_request' }, 400) + } + const { deliveryId: _id, seq: _seq, expiresAt: _expiresAt, conversation: _c, ...rest } = + delivery.data + const status = await deps.queue({ + thread: { + externalId, + teamId: 'dev', + channelId: 'dev', + threadTs: externalId, + ownerUserId: 'dev', + hostId, + placeId + }, + deliveryId, + body: rest + }) + if (status === 'unknown_host') { + return context.json({ error: status }, 404) + } + if (status === 'not_this_hosts_thread') { + return context.json({ error: status }, 409) + } + return context.json({ deliveryId, status }) + }) +} diff --git a/cloud/apps/connector/src/inbox-waker.test.ts b/cloud/apps/connector/src/inbox-waker.test.ts new file mode 100644 index 00000000000..80dc5ee7909 --- /dev/null +++ b/cloud/apps/connector/src/inbox-waker.test.ts @@ -0,0 +1,43 @@ +import type { CloudNotifications, ListenHandlers } from '@orca-cloud/cloud-database' +import { describe, expect, it } from 'vitest' +import { InboxWaker } from './inbox-waker.js' + +function fakeNotifications() { + let handlers: ListenHandlers | null = null + const notifications: CloudNotifications = { + notify: async () => {}, + listen: async (_channel, next) => { + handlers = next + return async () => {} + } + } + return { notifications, handlers: () => handlers! } +} + +describe('inbox waker', () => { + it('wakes only the named host, and every waiter after the listener reconnects', async () => { + const fake = fakeNotifications() + const waker = new InboxWaker(fake.notifications) + await waker.start() + const woken: string[] = [] + const a = waker.register('a').wait(5_000).then(() => woken.push('a')) + const b = waker.register('b').wait(5_000).then(() => woken.push('b')) + fake.handlers().onNotify('a') + await a + expect(woken).toEqual(['a']) + fake.handlers().onReconnect() + await b + expect(woken).toEqual(['a', 'b']) + }) + + it('keeps a notification that lands between registering and waiting', async () => { + const fake = fakeNotifications() + const waker = new InboxWaker(fake.notifications) + await waker.start() + const waiter = waker.register('a') + fake.handlers().onNotify('a') + const startedAt = Date.now() + await waiter.wait(5_000) + expect(Date.now() - startedAt).toBeLessThan(1_000) + }) +}) diff --git a/cloud/apps/connector/src/inbox-waker.ts b/cloud/apps/connector/src/inbox-waker.ts new file mode 100644 index 00000000000..ff89252cd4a --- /dev/null +++ b/cloud/apps/connector/src/inbox-waker.ts @@ -0,0 +1,73 @@ +import type { CloudNotifications } from '@orca-cloud/cloud-database' +import { INBOX_CHANNEL } from './connector-inbox.js' + +export type InboxWaiter = { wait(ms: number): Promise; cancel(): void } + +// Held polls wait here. One LISTEN per instance; a reconnect wakes everyone to re-read, since +// notifications sent while the listener was down are lost. +export class InboxWaker { + private readonly waiters = new Map void>>() + private stopListening: (() => Promise) | null = null + private isDraining = false + + constructor(private readonly notifications: CloudNotifications) {} + + async start(): Promise { + this.stopListening = await this.notifications.listen(INBOX_CHANNEL, { + onNotify: (hostId) => this.wake(hostId), + onReconnect: () => this.wakeAll() + }) + } + + // Register before reading the inbox, so a change between the read and the wait still wakes it. + register(hostId: string): InboxWaiter { + let wake!: () => void + const woken = new Promise((resolve) => (wake = resolve)) + const waiters = this.waiters.get(hostId) ?? new Set() + waiters.add(wake) + this.waiters.set(hostId, waiters) + const cancel = (): void => { + waiters.delete(wake) + if (waiters.size === 0 && this.waiters.get(hostId) === waiters) { + this.waiters.delete(hostId) + } + } + return { + wait: async (ms) => { + let timer: NodeJS.Timeout | undefined + await Promise.race([woken, new Promise((resolve) => (timer = setTimeout(resolve, ms)))]) + clearTimeout(timer) + cancel() + }, + cancel + } + } + + // Shutdown: every held poll answers now and tells its machine to reconnect elsewhere. + drain(): void { + this.isDraining = true + this.wakeAll() + } + + get draining(): boolean { + return this.isDraining + } + + async close(): Promise { + await this.stopListening?.() + } + + private wake(hostId: string): void { + for (const wake of this.waiters.get(hostId) ?? []) { + wake() + } + } + + private wakeAll(): void { + for (const waiters of this.waiters.values()) { + for (const wake of waiters) { + wake() + } + } + } +} diff --git a/cloud/apps/connector/src/index.ts b/cloud/apps/connector/src/index.ts index 12f921fb41f..3016347cd23 100644 --- a/cloud/apps/connector/src/index.ts +++ b/cloud/apps/connector/src/index.ts @@ -1,6 +1,8 @@ +import { openCloudNotifications } from '@orca-cloud/cloud-database' import { loadConnectorConfig } from './config.js' +import { startConnectorBackground } from './connector-background.js' import { openConnectorDatabase } from './connector-database.js' -import { createConnectorServer } from './connector-server.js' +import { createConnectorServer, listenHostname } from './connector-server.js' const config = loadConnectorConfig() const database = await openConnectorDatabase({ @@ -10,9 +12,18 @@ const database = await openConnectorDatabase({ applicationName: 'orca-connector', readOnly: config.mode === 'validation' }) -const runtime = createConnectorServer(config, database) +const notifications = openCloudNotifications({ + ...(config.databaseUrl === undefined ? {} : { databaseUrl: config.databaseUrl }), + applicationName: 'orca-connector' +}) +const runtime = createConnectorServer(config, database, notifications) +const active = config.mode === 'active' +if (active) { + await runtime.waker.start() +} +const stopBackground = active ? startConnectorBackground(runtime) : () => {} -runtime.server.listen(config.port, () => { +runtime.server.listen(config.port, listenHostname(config.publicUrl), () => { console.log(`[orca-connector] listening on ${config.publicUrl} (port ${config.port})`) }) @@ -25,10 +36,14 @@ const shutdown = (): void => { // Cloud Run sends SIGKILL after ten seconds; leave time for explicit cleanup. const deadline = setTimeout(() => process.exit(1), 9_000) deadline.unref() + stopBackground() + // Held polls answer at once with a reconnect, so the drain is not held up by them. + runtime.waker.drain() const requests = runtime.requestDrain.begin() const connections = new Promise((resolve) => runtime.server.close(() => resolve())) void Promise.all([requests, connections]) .then(async () => { + await runtime.waker.close() await database.close() clearTimeout(deadline) }) diff --git a/cloud/apps/connector/src/machine-auth.ts b/cloud/apps/connector/src/machine-auth.ts new file mode 100644 index 00000000000..89f6dae3bdb --- /dev/null +++ b/cloud/apps/connector/src/machine-auth.ts @@ -0,0 +1,30 @@ +import type { MiddlewareHandler } from 'hono' +import { + connectorHostId, + readBearer, + type ConnectorTokenVerifier +} from './connector-token-verifier.js' +import type { RateLimiter } from './rate-limiter.js' + +// One machine of one account, as its connector token names it; expiresAt bounds what it may hold. +export type MachineHost = { hostId: string; accountId: string; expiresAt: number } +export type MachineEnv = { Variables: { host: MachineHost } } + +export function machineAuth(deps: { + verifyToken: ConnectorTokenVerifier + hostRequests: RateLimiter +}): MiddlewareHandler { + return async (context, next) => { + const bearer = readBearer(context.req.header('authorization')) + const claims = bearer ? await deps.verifyToken(bearer) : null + if (!claims) { + return context.json({ error: 'invalid_token' }, 401) + } + const hostId = connectorHostId(claims) + if (!deps.hostRequests.allow(hostId)) { + return context.json({ error: 'rate_limited' }, 429) + } + context.set('host', { hostId, accountId: claims.sub, expiresAt: claims.exp * 1000 }) + await next() + } +} diff --git a/cloud/apps/connector/src/machine-routes.ts b/cloud/apps/connector/src/machine-routes.ts new file mode 100644 index 00000000000..1df23bbac85 --- /dev/null +++ b/cloud/apps/connector/src/machine-routes.ts @@ -0,0 +1,164 @@ +import type { CloudDatabase } from '@orca-cloud/cloud-database' +import { + CONNECTOR_CONTRACT_VERSION, + CONNECTOR_PATHS, + ConnectorAcksRequestSchema, + ConnectorEventsRequestSchema, + ConnectorHelloRequestSchema, + ConnectorPollRequestSchema, + type ConnectorDelivery, + type ConnectorPollResponse +} from '@orca-cloud/connector-contract' +import type { Context, Hono, MiddlewareHandler } from 'hono' +import { lockConnectorHost, type ConnectorHosts } from './connector-hosts.js' +import type { ConnectorInbox, Unsettled } from './connector-inbox.js' +import type { ConnectorThreads } from './connector-threads.js' +import type { InboxWaker } from './inbox-waker.js' +import type { MachineEnv } from './machine-auth.js' + +class TokenExpiredDuringPoll extends Error {} + +// The machine API behind a connector token: hello, the held poll, confirmations and turn events. +export function registerMachineRoutes( + app: Hono, + deps: { + database: CloudDatabase + inbox: ConnectorInbox + threads: ConnectorThreads + hosts: ConnectorHosts + waker: InboxWaker + now: () => number + holdMs: number + limitBody: MiddlewareHandler + } +): void { + const { database, inbox, threads, hosts, waker, now, holdMs, limitBody } = deps + + // Acks and events from an older hello change nothing: the first poll after the newer hello lists + // what is unsettled, and its batches start a new events mark. An unrecorded hello must be sent. + async function onCurrentHello( + context: Context, + helloRev: number, + run: (transaction: CloudDatabase, hostId: string) => Promise + ): Promise { + const { hostId } = context.get('host') + const status = await database.transaction(async (transaction) => { + await lockConnectorHost(transaction, hostId) + const status = await hosts.helloStatus(transaction, hostId, helloRev) + if (status === 'current') { + await run(transaction, hostId) + } + return status + }) + return status === 'unknown' + ? context.json({ error: 'hello_required' }, 409) + : context.body(null, 204) + } + + app.put(CONNECTOR_PATHS.hello, limitBody, async (context) => { + const body = ConnectorHelloRequestSchema.safeParse(await context.req.json().catch(() => null)) + if (!body.success) { + return context.json({ error: 'invalid_request' }, 400) + } + if (body.data.contract < CONNECTOR_CONTRACT_VERSION) { + return context.json({ error: 'client_too_old' }, 426) + } + const host = context.get('host') + await database.transaction(async (transaction) => { + await lockConnectorHost(transaction, host.hostId) + await hosts.recordHello(transaction, host, body.data) + }) + return context.body(null, 204) + }) + + app.post(CONNECTOR_PATHS.poll, limitBody, async (context) => { + const body = ConnectorPollRequestSchema.safeParse(await context.req.json().catch(() => null)) + if (!body.success) { + return context.json({ error: 'invalid_request' }, 400) + } + const { hostId, expiresAt } = context.get('host') + const { helloRev } = body.data + await hosts.touch(hostId) + const holdUntil = Math.min(Date.now() + holdMs, expiresAt) + let deliveries: ConnectorDelivery[] = [] + let unsettled: Unsettled[] = [] + for (;;) { + const waiter = waker.register(hostId) + try { + // Each pass decides under the host lock, so a newer hello either lands before it (and this + // refuses) or after it has committed what it hands out. + const pass = await database + .transaction(async (transaction) => { + await lockConnectorHost(transaction, hostId) + if ((await hosts.helloStatus(transaction, hostId, helloRev)) !== 'current') { + return null + } + const recovered = (await hosts.takeUnsettledDue(transaction, hostId)) + ? await inbox.listUnsettled(transaction, hostId) + : [] + const claimed = await inbox.claim(transaction, hostId) + // Last moment before commit: a token past its expiry rolls the pass back, so nothing + // goes out on it and the next token's poll sees the same inbox. + if (Date.now() >= expiresAt) { + throw new TokenExpiredDuringPoll() + } + return { recovered, ...claimed } + }) + .catch((error: unknown) => { + if (error instanceof TokenExpiredDuringPoll) { + return { recovered: [], deliveries: [], nextLeaseDueAt: null } + } + throw error + }) + if (pass === null) { + return context.json({ error: 'hello_required' }, 409) + } + deliveries = pass.deliveries + unsettled = pass.recovered + const remaining = holdUntil - Date.now() + const ready = deliveries.length > 0 || unsettled.length > 0 + if (ready || remaining <= 0 || waker.draining) { + break + } + // A lease running out sends no notification, so wake for the next one too. + const leaseDueMs = + pass.nextLeaseDueAt === null ? remaining : Math.max(0, pass.nextLeaseDueAt - now()) + await waiter.wait(Math.min(remaining, leaseDueMs)) + } finally { + waiter.cancel() + } + } + const response: ConnectorPollResponse = { + v: 1, + serverNow: now(), + holdMs, + ...(unsettled.length > 0 ? { unsettled } : {}), + deliveries, + ...(waker.draining ? { control: { reconnect: true as const } } : {}) + } + return context.json(response) + }) + + app.post(CONNECTOR_PATHS.acks, limitBody, async (context) => { + const body = ConnectorAcksRequestSchema.safeParse(await context.req.json().catch(() => null)) + if (!body.success) { + return context.json({ error: 'invalid_request' }, 400) + } + return await onCurrentHello(context, body.data.helloRev, (transaction, hostId) => + inbox.acknowledge(transaction, hostId, body.data.acks) + ) + }) + + app.post(CONNECTOR_PATHS.events, limitBody, async (context) => { + const body = ConnectorEventsRequestSchema.safeParse(await context.req.json().catch(() => null)) + if (!body.success) { + return context.json({ error: 'invalid_request' }, 400) + } + // A stale batch still answers 204: the machine has nothing to retry. + return await onCurrentHello(context, body.data.helloRev, async (transaction, hostId) => { + if (await hosts.advanceEventsSeq(transaction, hostId, body.data.seq)) { + await threads.applyEvents(transaction, hostId, body.data.events) + } + }) + }) +} diff --git a/cloud/apps/connector/src/queue-delivery.ts b/cloud/apps/connector/src/queue-delivery.ts new file mode 100644 index 00000000000..c75fab277f3 --- /dev/null +++ b/cloud/apps/connector/src/queue-delivery.ts @@ -0,0 +1,36 @@ +import type { CloudDatabase } from '@orca-cloud/cloud-database' +import { lockConnectorHost, type ConnectorHosts } from './connector-hosts.js' +import type { ConnectorInbox, DeliveryBody, EnqueueResult } from './connector-inbox.js' +import type { ConnectorThreads, NewThread } from './connector-threads.js' + +export type QueueResult = EnqueueResult | 'unknown_host' | 'not_this_hosts_thread' + +// One host-locked transaction finds the host, binds the thread and queues, so per-host seq is +// assigned in order. +export async function queueDelivery( + deps: { + database: CloudDatabase + inbox: ConnectorInbox + threads: ConnectorThreads + hosts: ConnectorHosts + }, + input: { thread: Omit; deliveryId: string; body: DeliveryBody } +): Promise { + const { hostId, externalId } = input.thread + return await deps.database.transaction(async (transaction) => { + await lockConnectorHost(transaction, hostId) + const accountId = await deps.hosts.accountOf(transaction, hostId) + if (!accountId) { + return 'unknown_host' + } + if (!(await deps.threads.bind(transaction, { ...input.thread, accountId }))) { + return 'not_this_hosts_thread' + } + return await deps.inbox.enqueue(transaction, { + hostId, + externalId, + deliveryId: input.deliveryId, + body: input.body + }) + }) +} diff --git a/cloud/packages/cloud-database/src/cloud-notifications.test.ts b/cloud/packages/cloud-database/src/cloud-notifications.test.ts new file mode 100644 index 00000000000..875192b0f24 --- /dev/null +++ b/cloud/packages/cloud-database/src/cloud-notifications.test.ts @@ -0,0 +1,97 @@ +import { randomUUID } from 'node:crypto' +import { describe, expect, it } from 'vitest' +import { openCloudDatabase, openInMemoryCloudDatabase } from './cloud-database.js' +import { openCloudNotifications } from './cloud-notifications.js' + +const postgresUrl = process.env.ORCA_CLOUD_DATABASE_TEST_URL + +function collector() { + const payloads: string[] = [] + let reconnects = 0 + let wake: (() => void) | null = null + return { + payloads, + reconnects: () => reconnects, + handlers: { + onNotify: (payload: string) => { + payloads.push(payload) + wake?.() + }, + onReconnect: () => { + reconnects++ + wake?.() + } + }, + until: async (done: () => boolean, timeoutMs = 10_000) => { + const deadline = Date.now() + timeoutMs + while (!done()) { + if (Date.now() > deadline) { + throw new Error('timed out') + } + await new Promise((resolve) => { + wake = resolve + setTimeout(resolve, 50) + }) + } + } + } +} + +describe('cloud database notifications', () => { + it('wakes SQLite listeners in this process', async () => { + const database = await openInMemoryCloudDatabase([]) + const notifications = openCloudNotifications({}) + const seen = collector() + const stop = await notifications.listen('inbox', seen.handlers) + await database.transaction((tx) => notifications.notify(tx, 'inbox', 'host-1')) + await stop() + await notifications.notify(database, 'inbox', 'host-2') + expect(seen.payloads).toEqual(['host-1']) + await database.close() + }) + + it('refuses a channel that is not a plain identifier', async () => { + const database = await openInMemoryCloudDatabase([]) + await expect( + openCloudNotifications({}).notify(database, 'inbox; DROP', 'x') + ).rejects.toThrow('invalid_listen_channel') + await database.close() + }) + + it.skipIf(!postgresUrl)( + 'delivers a Postgres notification at commit and re-listens after losing its connection', + async () => { + const applicationName = `cloud-db-listen-${randomUUID()}` + const database = await openCloudDatabase({ + service: 'test', + schemaStatements: [], + databaseUrl: postgresUrl, + dataDir: '', + applicationName + }) + const notifications = openCloudNotifications({ databaseUrl: postgresUrl, applicationName }) + const seen = collector() + const stop = await notifications.listen('inbox', seen.handlers) + try { + await database.transaction(async (tx) => { + await notifications.notify(tx, 'inbox', 'committed') + await new Promise((resolve) => setTimeout(resolve, 100)) + expect(seen.payloads).toEqual([]) + }) + await seen.until(() => seen.payloads.length === 1) + + await database.query( + 'SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE application_name = ?', + [`${applicationName}/listen`] + ) + await seen.until(() => seen.reconnects() === 1) + await notifications.notify(database, 'inbox', 'after-reconnect') + await seen.until(() => seen.payloads.length === 2) + expect(seen.payloads).toEqual(['committed', 'after-reconnect']) + } finally { + await stop() + await database.close() + } + } + ) +}) diff --git a/cloud/packages/cloud-database/src/cloud-notifications.ts b/cloud/packages/cloud-database/src/cloud-notifications.ts new file mode 100644 index 00000000000..8136af965eb --- /dev/null +++ b/cloud/packages/cloud-database/src/cloud-notifications.ts @@ -0,0 +1,138 @@ +import { EventEmitter } from 'node:events' +import pg from 'pg' +import type { CloudDatabase } from './cloud-database.js' + +const LISTEN_CONNECTION_TIMEOUT_MS = 2_000 +const LISTEN_RETRY_BASE_MS = 1_000 +const LISTEN_RETRY_MAX_MS = 30_000 +// LISTEN takes no bind parameters, so a channel is a plain identifier. +const CHANNEL_NAME = /^[a-z][a-z0-9_]{0,62}$/ + +export type ListenHandlers = { onNotify: (payload: string) => void; onReconnect: () => void } + +export type CloudNotifications = { + // Postgres: NOTIFY on the caller's connection, so inside a transaction it is sent at commit. + // SQLite: wakes this process's listeners at once; they re-read, so a rollback costs one query. + notify(database: CloudDatabase, channel: string, payload: string): Promise + // Postgres holds one dedicated connection and calls onReconnect after listening again, + // because notifications sent while it was down are lost. + listen(channel: string, handlers: ListenHandlers): Promise<() => Promise> +} + +function requireChannel(channel: string): string { + if (!CHANNEL_NAME.test(channel)) { + throw new Error('invalid_listen_channel') + } + return channel +} + +// Pass the same databaseUrl as openCloudDatabase; without one, notifications stay in this process. +export function openCloudNotifications(input: { + databaseUrl?: string + applicationName?: string +}): CloudNotifications { + const local = new EventEmitter() + const { databaseUrl } = input + return { + notify: async (database, channel, payload) => { + requireChannel(channel) + if (database.dialect === 'postgres') { + await database.query('SELECT pg_notify(?, ?)', [channel, payload]) + } else { + local.emit(channel, payload) + } + }, + listen: async (channel, handlers) => { + requireChannel(channel) + if (!databaseUrl) { + const onNotify = (payload: string): void => handlers.onNotify(payload) + local.on(channel, onNotify) + return async () => { + local.off(channel, onNotify) + } + } + const listener = new PostgresListener( + { + connectionString: databaseUrl, + application_name: input.applicationName ? `${input.applicationName}/listen` : undefined, + connectionTimeoutMillis: LISTEN_CONNECTION_TIMEOUT_MS + }, + channel, + handlers + ) + await listener.connect() + return () => listener.close() + } + } +} + +// One LISTEN connection outside any pool, reconnected with doubling backoff until closed. +class PostgresListener { + private client: pg.Client | null = null + private retry: NodeJS.Timeout | undefined + private closed = false + private retryMs = LISTEN_RETRY_BASE_MS + + constructor( + private readonly connection: pg.ClientConfig, + private readonly channel: string, + private readonly handlers: ListenHandlers + ) {} + + async connect(): Promise { + const client = new pg.Client(this.connection) + client.on('notification', (message) => { + if (message.channel === this.channel) { + this.handlers.onNotify(message.payload ?? '') + } + }) + client.on('error', () => this.drop(client)) + client.on('end', () => this.drop(client)) + try { + await client.connect() + await client.query(`LISTEN ${this.channel}`) + } catch (error) { + await client.end().catch(() => undefined) + throw error + } + this.client = client + this.retryMs = LISTEN_RETRY_BASE_MS + } + + async close(): Promise { + this.closed = true + clearTimeout(this.retry) + await this.client?.end().catch(() => undefined) + } + + private drop(client: pg.Client): void { + if (this.client !== client) { + return + } + this.client = null + void client.end().catch(() => undefined) + this.scheduleReconnect() + } + + private scheduleReconnect(): void { + if (this.closed) { + return + } + this.retry = setTimeout(() => { + this.connect().then( + () => { + if (this.closed) { + void this.client?.end().catch(() => undefined) + } else { + this.handlers.onReconnect() + } + }, + () => { + this.retryMs = Math.min(this.retryMs * 2, LISTEN_RETRY_MAX_MS) + this.scheduleReconnect() + } + ) + }, this.retryMs) + this.retry.unref() + } +} diff --git a/cloud/packages/cloud-database/src/index.ts b/cloud/packages/cloud-database/src/index.ts index 3b7f278cbc2..03cb7f7b7b8 100644 --- a/cloud/packages/cloud-database/src/index.ts +++ b/cloud/packages/cloud-database/src/index.ts @@ -5,3 +5,8 @@ export { type OpenCloudDatabaseInput, type SqlRow } from './cloud-database.js' +export { + openCloudNotifications, + type CloudNotifications, + type ListenHandlers +} from './cloud-notifications.js' diff --git a/cloud/pnpm-lock.yaml b/cloud/pnpm-lock.yaml index cb1b20c4483..6da464431ca 100644 --- a/cloud/pnpm-lock.yaml +++ b/cloud/pnpm-lock.yaml @@ -193,6 +193,9 @@ importers: '@orca-cloud/cloud-database': specifier: workspace:* version: link:../../packages/cloud-database + '@orca-cloud/connector-contract': + specifier: workspace:* + version: link:../../packages/connector-contract hono: specifier: ^4.13.13 version: 4.13.13