feat(projectors): project runs into a Postgres event log without blocking append

.changeset/postgres-event-log.md

@@ -0,0 +1,7 @@
+---
+'@swirls/looms': minor
+---
+
+feat(projectors): project runs into a Postgres event log without blocking append
+
+Durable Object projectors are scheduled with `waitUntil` after the log commits. `postgresEventLog` stores each non-ephemeral envelope and a run index with status, launch links, and tenant attributes, and can rebuild that sink from the log.

docs/production.md

@@ -13,6 +13,31 @@ coalesced. A deployment, eviction, or alarm retry may repeat a wake; event
 preconditions make that safe. Cross-object queries and transactions are not
 provided. Build global run lists with a projector.
 
+On a Durable Object, projector delivery is scheduled with `waitUntil` after the
+append commits, so projector latency is not part of the run step. The cursor is
+durable. A delivery that fails, or an isolate that stops before `waitUntil`
+finishes, is retried on the next alarm or request.
+
+`postgresEventLog` is the generic sink: one row per non-ephemeral envelope,
+keyed by `(run_id, seq)`, and a `looms_run_index` row with status, timestamps,
+the parent run, the launching effect, and the run's attributes. `tenant`,
+`project`, and `org` are copied out of string attributes so row-level security
+can index them:
+
+```sql
+ALTER TABLE looms_run_index ENABLE ROW LEVEL SECURITY;
+CREATE POLICY tenant_isolation ON looms_run_index
+  USING (tenant = current_setting('looms.tenant', true));
+```
+
+Pass Hyperdrive's connection string as `postgresEventLog({ url })`. Each
+delivered batch is one transaction of about three statements, however many
+events and runs it holds: one multi-row event insert per 1,000 events, one
+run-index upsert, and one update for runs that only gained events. Writes are
+idempotent. `rebuild(source)` replays a run from seq 1; run that while live
+delivery is paused, then let the cursor catch up. Ephemeral events are left out
+of the log.
+
 Live subscribers on a Durable Object should open a WebSocket (`Upgrade: websocket` on `GET /runs/:id/events`). The object hibernates the socket, answers `ping` with `pong` without waking, and fans out events when the log is appended. SSE stays available, including on `looms.serve`, but its keepalive keeps a Durable Object awake. Writes stay on HTTP. A browser cannot send `Authorization` on the upgrade, so pass a read grant (see `grantAuth` below); `@looms/client` puts `grant` on the socket URL. An idle-socket billed-duration check on a real Cloudflare account has not been run.
 
 A workflow can launch children with `childExecution: 'run'`. Each child is its own run, addressed by a deterministic id, and reports `runtime.child.settled` back to the parent. Cancelling the parent cancels those children. `childClose: 'wait'` records `runtime.child.close.requested` and cancels the parent only after the children reach a terminal status. `childClose: 'terminate'` cancels them as the parent closes. Children stay threads in the parent run unless `childExecution` is set.

packages/cloudflare/src/event-store.ts

@@ -1,4 +1,4 @@
-import type { Effect } from 'effect'
+import { Effect } from 'effect'
 
 import { snapshotStoreOf, withBlobOffload, type BlobStore, type EventStore } from '@looms/core'
 import { sqliteEventStore, sqlStorageExec } from '@looms/core/sqlite-store'
@@ -23,7 +23,9 @@ export interface DurableObjectEventStore extends EventStore {
 
 /**
  * Creates an EventStore backed by a Durable Object's embedded SQLite database (`ctx.storage.sql`).
- * Optionally attaches asynchronous projectors (such as S2).
+ * Optionally attaches projectors. Their delivery is scheduled with `waitUntil`, so an append
+ * returns when the log commits. A crashed delivery is retried from the durable cursor on the next
+ * alarm or request.
  */
 export function durableObjectEventStore(
   ctx: DurableObjectState,
@@ -49,7 +51,13 @@ export function durableObjectEventStore(
       { scheduleRetry: options.scheduleProjectorRetry },
     )
 
-    const wrapped = withProjectors(store, options.projectors, { delivery })
+    const wrapped = withProjectors(store, options.projectors, {
+      delivery,
+      scheduleDelivery: (_runId, work) => {
+        ctx.waitUntil(Effect.runPromise(work))
+      },
+    })
+
     return Object.assign(wrapped, { projectorDelivery: delivery })
   }
 

packages/projectors/package.json

@@ -43,6 +43,8 @@
     "effect": "4.0.0-rc.112"
   },
   "devDependencies": {
+    "@electric-sql/pglite": "^0.5.11",
+    "@electric-sql/pglite-socket": "^0.2.14",
     "@types/bun": "^1.3.9",
     "postgres": "^3.4.0",
     "typescript": "7.0.2"

packages/projectors/src/event-log-postgres.test.ts

@@ -0,0 +1,198 @@
+import { afterAll, beforeAll, describe, expect, test } from 'bun:test'
+
+import { PGlite } from '@electric-sql/pglite'
+import { PGLiteSocketServer } from '@electric-sql/pglite-socket'
+import createPostgresClient, { type Sql } from 'postgres'
+
+import { createEvent, type EventEnvelope, type EventInput, type RunStatus } from '@looms/core'
+
+import { memoryEventLog, postgresEventLog, type RunLogRow } from './event-log'
+
+const PORT = 54_329
+const RUN_STATUSES: readonly RunStatus[] = ['running', 'completed', 'failed', 'cancelled']
+
+let db: PGlite
+let server: PGLiteSocketServer
+let sql: Sql
+const statements: string[] = []
+
+beforeAll(async () => {
+  db = await PGlite.create()
+  server = new PGLiteSocketServer({ db, port: PORT, host: '127.0.0.1' })
+  await server.start()
+
+  sql = createPostgresClient({
+    host: '127.0.0.1',
+    port: PORT,
+    max: 1,
+    onnotice: () => undefined,
+    debug: (_connection, query) => {
+      statements.push(query)
+    },
+  })
+})
+
+afterAll(async () => {
+  await sql.end({ timeout: 1 })
+  await server.stop()
+  await db.close()
+})
+
+function event(runId: string, seq: number, input: EventInput): EventEnvelope {
+  return { ...createEvent(runId, input), seq, ts: 1_700_000_000_000 + seq * 1_000 }
+}
+
+function started(runId: string): EventEnvelope {
+  return event(runId, 1, {
+    type: 'runtime.run.started',
+    payload: {
+      rootThreadId: 'thr_1',
+      kind: 'workflow',
+      definitionName: 'checkout',
+      definitionVersion: 'v1',
+      input: { text: 'it\'s "quoted"' },
+      attributes: { tenant: 'acme', project: 'billing' },
+      launchedBy: {
+        runId: 'run_parent',
+        threadId: 'thr_parent',
+        effectId: 'eff_1',
+        causationId: 'evt_cause',
+        causationSeq: 4,
+      },
+    },
+    correlationId: 'run_root',
+    threadId: null,
+    origin: { type: 'system' },
+  })
+}
+
+function step(runId: string, seq: number): EventEnvelope {
+  return event(runId, seq, {
+    type: 'workflow.node.completed',
+    payload: 'plain string payload',
+    threadId: 'thr_1',
+    origin: { type: 'system' },
+  })
+}
+
+function failed(runId: string, seq: number): EventEnvelope {
+  return event(runId, seq, {
+    type: 'runtime.run.completed',
+    payload: { output: null, error: 'nope' },
+    threadId: null,
+    origin: { type: 'system' },
+  })
+}
+
+interface IndexRow {
+  readonly run_id: string
+  readonly status: string
+  readonly kind: string | null
+  readonly definition_name: string | null
+  readonly definition_version: string | null
+  readonly root_thread_id: string | null
+  readonly parent_run_id: string | null
+  readonly launched_by_effect_id: string | null
+  readonly correlation_id: string | null
+  readonly attributes: { readonly [key: string]: string }
+  readonly tenant: string | null
+  readonly project: string | null
+  readonly org: string | null
+  readonly started_at: Date | null
+  readonly updated_at: Date
+  readonly completed_at: Date | null
+}
+
+function fromIndexRow(row: IndexRow): RunLogRow {
+  return {
+    runId: row.run_id,
+    status: RUN_STATUSES.find((status) => status === row.status) ?? 'running',
+    kind: row.kind,
+    definitionName: row.definition_name,
+    definitionVersion: row.definition_version,
+    rootThreadId: row.root_thread_id,
+    parentRunId: row.parent_run_id,
+    launchedByEffectId: row.launched_by_effect_id,
+    correlationId: row.correlation_id,
+    attributes: row.attributes,
+    tenant: row.tenant,
+    project: row.project,
+    org: row.org,
+    startedAt: row.started_at?.getTime() ?? null,
+    updatedAt: row.updated_at.getTime(),
+    completedAt: row.completed_at?.getTime() ?? null,
+  }
+}
+
+function indexRow(runId: string): Promise<RunLogRow | null> {
+  return sql<IndexRow[]>`SELECT * FROM looms_run_index WHERE run_id = ${runId}`.then(([row]) =>
+    row ? fromIndexRow(row) : null,
+  )
+}
+
+describe('postgres event log', () => {
+  test('matches the memory sink for starts, steps, a replay, and a finish', async () => {
+    const postgres = postgresEventLog({ sql })
+    const memory = memoryEventLog()
+    await postgres.init?.()
+
+    const batches = [
+      [started('run_a'), step('run_a', 2), started('run_b')],
+      [step('run_a', 3), failed('run_a', 4), step('run_b', 2)],
+      [started('run_a'), step('run_a', 3)],
+    ]
+
+    for (const batch of batches) {
+      await postgres.project(batch)
+      await memory.project(batch)
+    }
+
+    for (const runId of ['run_a', 'run_b']) {
+      expect(await indexRow(runId)).toEqual(memory.run(runId))
+    }
+
+    const events = await sql<{ seq: string; payload: unknown }[]>`
+      SELECT seq, payload FROM looms_run_events WHERE run_id = 'run_a' ORDER BY seq
+    `
+
+    expect(events.map((row) => Number(row.seq))).toEqual([1, 2, 3, 4])
+    expect(events[1]?.payload).toBe('plain string payload')
+    expect(events[0]?.payload).toMatchObject({ input: { text: 'it\'s "quoted"' } })
+  })
+
+  test('writes a batch with a fixed number of statements', async () => {
+    const postgres = postgresEventLog({ sql })
+    await postgres.init?.()
+
+    const batch = Array.from({ length: 50 }, (_, index) => [
+      started(`run_many_${index}`),
+      step(`run_many_${index}`, 2),
+      step(`run_many_${index}`, 3),
+    ]).flat()
+
+    statements.length = 0
+    await postgres.project(batch)
+
+    const writes = statements.filter((query) => /^\s*(INSERT|UPDATE)/i.test(query))
+
+    const [count] = await sql<{ count: string }[]>`
+      SELECT count(*) FROM looms_run_index WHERE run_id LIKE 'run_many_%'
+    `
+
+    expect(writes).toHaveLength(2)
+    expect(Number(count?.count)).toBe(50)
+  })
+
+  test('a touch-only batch updates existing runs and ignores unknown ones', async () => {
+    const postgres = postgresEventLog({ sql })
+    await postgres.init?.()
+    await postgres.project([started('run_touch')])
+
+    statements.length = 0
+    await postgres.project([step('run_touch', 9), step('run_unknown', 2)])
+
+    expect(statements.filter((query) => /^\s*UPDATE/i.test(query))).toHaveLength(1)
+    expect((await indexRow('run_touch'))?.updatedAt).toBe(1_700_000_009_000)
+    expect(await indexRow('run_unknown')).toBeNull()
+  })
+})