From 09df336ca190d64de5d13dd8092e3f793d00c53f Mon Sep 17 00:00:00 2001 From: Antonio Antenore Date: Sun, 19 Jul 2026 03:13:02 +0200 Subject: [PATCH] feat(storage): add PostgreSQL multi-replica store --- .github/workflows/ci.yml | 39 ++ CHANGELOG.md | 25 + README.md | 74 +- docs/adr/0003-postgres-multireplica-store.md | 53 ++ docs/delivery-contract.md | 47 +- docs/postgres-runbook.md | 110 +++ package.json | 9 +- pnpm-lock.yaml | 116 ++++ scripts/package-smoke.mjs | 14 + src/adapters/http/app.ts | 31 + src/adapters/storage/in-memory-event-store.ts | 5 +- src/adapters/storage/postgres-event-store.ts | 602 +++++++++++++++++ src/adapters/storage/sqlite-event-store.ts | 7 +- src/cli.ts | 6 +- src/ports/index.ts | 1 + src/ports/readiness.ts | 9 + src/postgres.ts | 1 + tests/http.test.ts | 58 +- tests/postgres-event-store.test.ts | 631 ++++++++++++++++++ tests/postgres.integration.test.ts | 198 ++++++ tests/storage.test.ts | 8 + 21 files changed, 2023 insertions(+), 21 deletions(-) create mode 100644 docs/adr/0003-postgres-multireplica-store.md create mode 100644 docs/postgres-runbook.md create mode 100644 src/adapters/storage/postgres-event-store.ts create mode 100644 src/ports/readiness.ts create mode 100644 src/postgres.ts create mode 100644 tests/postgres-event-store.test.ts create mode 100644 tests/postgres.integration.test.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 56519bf..5ffd871 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -34,3 +34,42 @@ jobs: run: pnpm install --frozen-lockfile - name: Run quality gate run: pnpm check + + postgres-integration: + name: postgres integration + runs-on: ubuntu-latest + services: + postgres: + image: postgres:17-alpine@sha256:742f40ea20b9ff2ff31db5458d127452988a2164df9e17441e191f3b72252193 + env: + POSTGRES_DB: postgres + POSTGRES_HOST_AUTH_METHOD: trust + POSTGRES_USER: postgres + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U postgres -d postgres" + --health-interval 5s + --health-timeout 5s + --health-retries 12 + env: + PAUSEMESH_REQUIRE_POSTGRES_TEST: "1" + PAUSEMESH_TEST_POSTGRES_URL: postgresql://postgres@127.0.0.1:5432/postgres + steps: + - name: Check out source + uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + persist-credentials: false + - name: Install pnpm + uses: pnpm/action-setup@0ebf47130e4866e96fce0953f49152a61190b271 # v6.0.9 + with: + version: 11.8.0 + - name: Install Node.js + uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0 + with: + node-version: 24 + cache: pnpm + - name: Install dependencies + run: pnpm install --frozen-lockfile + - name: Run PostgreSQL integration + run: pnpm test:postgres diff --git a/CHANGELOG.md b/CHANGELOG.md index b426a8e..16752d7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,31 @@ All notable changes to PauseMesh are documented here. The project follows semantic versioning; protocol adapter contracts remain experimental until a stable `1.0.0` release. +## 0.3.0-alpha.1 - 2026-07-19 + +### Added + +- Optional `pausemesh/postgres` entry point with an externally owned, structurally typed pool and + no PostgreSQL client dependency in the root runtime path. +- Explicit, checksum-validated PostgreSQL schema migration with a serialized migration lock, + append-only database guards, and no constructor-time DDL. +- Multi-replica compare-and-swap persistence using a stream-head fence and one checked-out client + per transaction. +- Bounded, fully validated PostgreSQL replay that detects missing events, version gaps, identity + mismatches, malformed snapshots, and unsafe `BIGINT` values. +- Host-owned readiness port and `/readyz`; SQLite, in-memory, and PostgreSQL stores implement the + same probe while `/healthz` remains an independent liveness endpoint. +- Scripted fake-pool coverage plus an opt-in real PostgreSQL test and mandatory Linux PostgreSQL CI + job covering concurrent replicas, exact retry, append-only enforcement, and restart replay. + +### Security + +- PostgreSQL schema identifiers are restricted and quoted; values remain parameterized. +- Migration metadata, physical tables, expected trigger/table/function tuples, trigger activation, + and runtime read access fail readiness closed without leaking database errors over HTTP. +- The adapter never owns or logs a connection string and never closes the injected pool. Deployment + code must separate migration and runtime roles and own timeout, idle-error, and shutdown policy. + ## 0.2.0-alpha.1 - 2026-07-17 ### Breaking diff --git a/README.md b/README.md index cc186d4..e4b16c4 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,7 @@ orchestrator. - A continuation can resume once. Concurrent attempts produce one winner. - Exact retries use an independent idempotency key and return the first outcome. - SQLite/WAL recovery after process restart, behind a replaceable `EventStore` port. +- Optional PostgreSQL multi-replica persistence with explicit migrations and transactional CAS. - Redacted observations that never include tokens or continuation payloads. - Versioned, intentionally narrow MCP 2025-11-25, A2A 1.0, and AG-UI adapters. - Explicit protocol-owned request/task/run bindings; continuation IDs are used only in adapter @@ -51,11 +52,14 @@ flowchart LR S --> P["EventStore port"] P --> M["In-memory adapter"] P --> Q[("SQLite / WAL")] + P --> PG[("PostgreSQL / replicas")] S --> O["Redacted observer"] ``` Dependencies point inward. Domain code imports no Hono, SQLite, protocol SDK, or logger package. Protocol drift stays in adapters; storage replacement stays behind one atomic append contract. +PostgreSQL lives behind the optional `pausemesh/postgres` entry point, so importing the root package +does not load or require a database client. ## Quick start @@ -174,6 +178,54 @@ const continuations = new ContinuationService({ }); ``` +### PostgreSQL multi-replica deployment + +The PostgreSQL adapter accepts an externally owned pool-like object. Migration is explicit and +must run from a deployment process with a DDL-capable role; `PostgresEventStore` constructors never +change the schema. The host owns TLS, credentials, timeouts, pool sizing, idle-client errors, and +shutdown. + +```ts +import { Pool } from "pg"; +import { + PostgresEventStore, + migratePostgresEventStore, +} from "pausemesh/postgres"; + +const migrationPool = new Pool({ + connectionString: process.env.PAUSEMESH_MIGRATION_DATABASE_URL, + connectionTimeoutMillis: 5_000, + max: 1, + statement_timeout: 10_000, +}); +await migratePostgresEventStore(migrationPool, { schema: "pausemesh" }); +await migrationPool.end(); + +const runtimePool = new Pool({ + connectionString: process.env.DATABASE_URL, + connectionTimeoutMillis: 5_000, + idleTimeoutMillis: 30_000, + max: 10, + query_timeout: 5_000, + statement_timeout: 5_000, +}); +runtimePool.on("error", () => { + // Report through protected host telemetry; never log the pool configuration or URL. +}); + +const store = new PostgresEventStore(runtimePool, { + schema: "pausemesh", + maxEventsPerStream: 32, +}); +``` + +Pass `store` as both the `ContinuationService` event store and +`CreateHttpAppOptions.readinessProbe`. `GET /healthz` remains process liveness and never touches the +database; `GET /readyz` verifies the exact migration and required PostgreSQL objects and returns a +bounded 503 body on failure. The bundled CLI intentionally remains SQLite-only because a production +PostgreSQL composition root must own provider-specific identity and lifecycle policy. See the +[PostgreSQL runbook](docs/postgres-runbook.md) and [ADR 0003](docs/adr/0003-postgres-multireplica-store.md). + Protocol adapters intentionally require their negotiated context: ```ts @@ -260,8 +312,8 @@ protocol parsing; hosts can tighten the default 64-level and 50,000-node limits. ```text src/domain/ versioned events, envelope, reducer, typed errors src/application/ lifecycle use cases and concurrency resolution -src/ports/ EventStore, Clock, TokenIssuer, Observer -src/adapters/storage/ in-memory and SQLite/WAL stores +src/ports/ EventStore, ReadinessProbe, Clock, TokenIssuer, Observer +src/adapters/storage/ in-memory, SQLite/WAL, and PostgreSQL stores src/adapters/{mcp,a2a,agui} protocol projections src/adapters/http/ local Hono reference API tests/ domain, storage, concurrency, conformance, HTTP, config tests @@ -277,6 +329,12 @@ protect payload data at rest, terminate TLS, and retain the same one-shot/CAS in `PAUSEMESH_MAX_PAYLOAD_BYTES` is enforced while streaming each JSON request, before parsing, even when no trustworthy content length is available. +For PostgreSQL, use separate migration and runtime roles. The runtime role needs schema usage, +read access, event inserts, and stream-head inserts/updates only; it must not receive DDL, delete, or +truncate privileges. Readiness validates migration metadata and the expected active trigger +bindings, but deployment access control remains the boundary against schema tampering. Connection +URLs and provider credentials are owned by the host and never accepted or logged by the adapter. + Resume tokens are bearer credentials. PauseMesh hashes them before persistence and redacts them from its observer, but the caller is responsible for safe delivery and storage. A SHA-256 hash is appropriate here because tokens are generated with 256 bits of entropy; it is not being used to @@ -284,11 +342,11 @@ hash human passwords or API keys. ## Status -`0.2.0-alpha.1` is an experimental protocol-conformance prerelease. The canonical state machine -and storage contract remain unchanged; the adapter surface now requires explicit upstream -bindings and negotiated capabilities. Protocol adapters will continue to evolve as upstream -specifications do. See [ADR 0002](docs/adr/0002-protocol-adapter-conformance.md) for the boundary -decision and [the delivery contract](docs/delivery-contract.md) for acceptance evidence and -explicit non-goals. +`0.3.0-alpha.1` is an experimental multi-replica persistence prerelease. The canonical state +machine, protocol mappings, token handling, and `EventStore` contract remain unchanged. PostgreSQL +adds a production-oriented storage option, not identity, tenancy, database HA, or orchestration. +See [ADR 0002](docs/adr/0002-protocol-adapter-conformance.md), +[ADR 0003](docs/adr/0003-postgres-multireplica-store.md), and +[the delivery contract](docs/delivery-contract.md) for boundaries and acceptance evidence. Apache-2.0 © 2026 Antonio Antenore. diff --git a/docs/adr/0003-postgres-multireplica-store.md b/docs/adr/0003-postgres-multireplica-store.md new file mode 100644 index 0000000..9c8ba19 --- /dev/null +++ b/docs/adr/0003-postgres-multireplica-store.md @@ -0,0 +1,53 @@ +# ADR 0003: Optional PostgreSQL store for multi-replica continuity + +- Status: accepted +- Date: 2026-07-19 +- Target: `0.3.0-alpha.1` + +## Context + +SQLite/WAL provides durable local recovery, but a callback may reach a different application +replica. The `EventStore` contract already requires an atomic compare-and-swap append, so the +multi-replica capability belongs in a storage adapter rather than in the continuation domain or a +new coordinator. + +PostgreSQL integration must not make the root package import a database driver, hide DDL in an +application constructor, expose connection credentials, or let a pool-backed transaction move +between physical connections. + +## Decision + +PauseMesh provides `PostgresEventStore` and `migratePostgresEventStore` from the optional +`pausemesh/postgres` entry point. + +- The host injects a minimal `PostgresPoolLike` object and retains ownership of connection limits, + timeouts, idle-client error handling, and shutdown. +- Migration is an explicit deployment operation. Version 1 is recorded with a stable SHA-256 + checksum and serialized with a transaction-scoped advisory lock. Constructors perform no DDL. +- A dedicated stream-head row holds `current_version`. Append creates the head only for a new + non-empty stream, advances it with `UPDATE ... WHERE current_version = expected`, and inserts the + full event batch on the same checked-out client and `READ COMMITTED` transaction. +- A lost fence becomes `VersionConflictError`; unexpected uniqueness, data, or database failures + are not mislabeled as concurrency. +- Event JSON remains text so replay observes the exact persisted snapshot. Load validates the head, + contiguous physical versions, embedded identity/version, schema, and a configurable event bound. +- Physical event rows and migration metadata are protected by database triggers. Stream-head rows + cannot be deleted or truncated; their version remains updateable by the runtime adapter. +- Readiness requires the exact migration version/checksum, ordinary tables, expected active + trigger/table/function tuples, and runtime read access. Liveness remains independent. +- The bundled CLI stays SQLite-only. A PostgreSQL deployment is a library composition root because + it must own provider-specific credentials, TLS, pool policy, roles, and lifecycle. + +## Consequences + +- Multiple stateless application replicas can share continuation state without changing protocol + adapters, lifecycle events, token handling, or application-service reconciliation. +- `pg` is used only for development and the real integration test; consumers may inject node-postgres + or another compatible pool without leaking its types through public declarations. +- Migration and runtime roles must be distinct. Runtime roles need no DDL, delete, or truncate + privilege. +- PostgreSQL availability, regional failover, backup/restore, tenant authorization, payload + encryption, and secret distribution remain deployment responsibilities. +- Readiness verifies migration metadata and the required object identities, not the entire function + body or every physical constraint. Removing DDL from the runtime role is therefore a required + defense, not an optional hardening step. diff --git a/docs/delivery-contract.md b/docs/delivery-contract.md index b8fea5b..fefb275 100644 --- a/docs/delivery-contract.md +++ b/docs/delivery-contract.md @@ -1,9 +1,9 @@ # Delivery Contract: PauseMesh Date: 2026-07-15 -Last updated: 2026-07-17 +Last updated: 2026-07-19 Mode: new-project -Status: delivered through `0.2.0-alpha.1` +Status: delivered through `0.3.0-alpha.1` ## Objective @@ -19,8 +19,10 @@ Must: - Issue opaque, one-shot resume tokens while persisting only their SHA-256 hashes. - Make retries idempotent and reject stale, wrong, expired, cancelled, or reused tokens. - Supply replaceable in-memory and SQLite/WAL event-store adapters. +- Supply an optional PostgreSQL adapter with explicit migrations and multi-replica CAS semantics. - Map a minimal continuation contract across MCP, A2A, and AG-UI adapters. - Expose a small HTTP/CLI demo and machine-readable errors. +- Keep liveness independent from a fail-closed, host-owned dependency readiness probe. Should: @@ -43,6 +45,7 @@ Out of scope: - Node.js 24+ is available. - SQLite is a reference store; consumers may implement the event-store port elsewhere. +- PostgreSQL consumers own their pool, credentials, TLS, timeouts, runtime role, and shutdown. - The MVP HTTP server is for local evaluation and must not be internet-exposed without authentication. - Protocol drafts can change, so adapter contracts are versioned independently from the core envelope. @@ -63,6 +66,10 @@ Out of scope: | R11 | AG-UI resume contract | Must | A whole resume batch is bound to an immutable issued receipt/current cohort and invalid input becomes `RUN_ERROR` | Official schema, receipt, and replay fixtures | | R12 | Adapter migration | Must | HTTP/demo/docs use the explicit 0.2 bindings with no legacy ID fallback | Build, demo, and HTTP tests | | R13 | Consumable package | Must | A clean source copy builds a tarball that installs in a real pnpm consumer with working ESM exports, declarations, docs, config, and CLI | Package smoke test | +| R14 | PostgreSQL multi-replica CAS | Must | Two independent pools share a stream, one conflicting append wins, the loser receives a typed version conflict, and exact service retry reconciles | Real PostgreSQL integration test | +| R15 | Explicit PostgreSQL migration | Must | Version/checksum mismatch, future schema, incomplete objects, or missing runtime access fail closed; constructors perform no DDL | Fake-pool and real PostgreSQL tests | +| R16 | Liveness/readiness split | Must | `/healthz` is unchanged and independent; `/readyz` reports configured dependency readiness without leaking underlying errors | HTTP and storage tests | +| R17 | Optional package boundary | Must | `pausemesh/postgres` works from the packed consumer while PostgreSQL symbols do not leak through the root export | Clean package smoke test | ## Acceptance Threshold @@ -72,8 +79,9 @@ the local API demo completes, and security/operational limitations are explicit. ## Architecture Approach A modular TypeScript library: pure domain state machine, application service, storage/clock/token -ports, protocol adapters at the edge, SQLite/WAL reference adapter, and a thin Hono HTTP adapter. -Dependencies point inward; protocol and storage packages never leak into the domain. +ports, protocol adapters at the edge, SQLite/WAL reference adapter, optional PostgreSQL adapter, +and a thin Hono HTTP adapter. Dependencies point inward; protocol and storage packages never leak +into the domain. The PostgreSQL client is injected structurally and owned by the host. ## Test Plan @@ -81,6 +89,8 @@ Critical: - State transition, stale version, token mismatch/reuse, idempotency, cancel, and expiry tests. - SQLite close/reopen replay and concurrent-resume tests. +- Scripted PostgreSQL migration, transaction, rollback, corruption, limit, and readiness tests. +- Real PostgreSQL two-pool CAS, idempotency, append-only, and restart replay test. - MCP/A2A/AG-UI projection contract tests. - Strict MCP primitive/enum schemas, original-request receipt binding, trusted URL policy, URL consent vs completion, and invalid result fixtures. @@ -93,10 +103,11 @@ Recommended: - HTTP error-shape and payload-limit integration tests. - Log redaction and configuration validation tests. -Not tested in the MVP: +Not covered by this library release: -- Multi-node consensus, hostile internet traffic, or full upstream conformance suites; risk is high - outside local evaluation, so the server is explicitly non-production until those gates exist. +- PostgreSQL regional failover/HA, backup restore, tenant isolation, hostile internet traffic, or + full upstream conformance suites. The bare CLI server remains non-production until a host adds + identity, authorization, TLS, rate limits, and deployment controls. ## Delivery Policy @@ -118,6 +129,9 @@ Selected mode: create public GitHub repository and push `main`, as explicitly re | 2026-07-17 | Prove the npm artifact from clean source | Release gate expanded | Prepack build and import/bin/content smoke included in `pnpm check` | | 2026-07-17 | Bind MCP results/completion to the exact emitted request | Added host-side issuance state | Request SHA-256 receipt is mandatory on inbound helpers | | 2026-07-17 | Remove consumer native-build approval | Storage implementation swap | `SqliteEventStore` now uses built-in `node:sqlite` behind the unchanged port | +| 2026-07-19 | Support callbacks reaching another replica | Added optional PostgreSQL adapter | Stream-head CAS on one checked-out `READ COMMITTED` transaction | +| 2026-07-19 | Keep deployment credentials and lifecycle out of the library | Pool remains externally owned | Optional subpath and structural pool contract; CLI remains SQLite-only | +| 2026-07-19 | Separate process liveness from dependency readiness | Added `ReadinessProbe` and `/readyz` | Omission/error fails closed without changing `/healthz` | ## Incident Register @@ -159,3 +173,22 @@ Selected mode: create public GitHub repository and push `main`, as explicitly re import, and the installed CLI. The production dependency audit reported no known vulnerabilities. - Boundary result: the continuation core and persisted envelope did not change. Protocol drift is isolated to adapters, and legacy task/run ID fallbacks are absent. + +### PostgreSQL multi-replica release-candidate evidence + +- Release: `0.3.0-alpha.1`, consolidated on the protected default branch after required CI. +- Architecture decision: [ADR 0003](adr/0003-postgres-multireplica-store.md); operations: + [PostgreSQL adapter runbook](postgres-runbook.md). +- PostgreSQL unit coverage uses a scripted pool/client boundary to verify statement ordering, + pre-validation, compare-and-swap conflicts, rollback/release behavior, migration checksums, + schema completeness, bounded replay, and corruption classification. +- The opt-in real PostgreSQL suite compiles against `pg.Pool` and is mandatory in its Linux CI job. + It runs concurrent migrations and replicas, exact retry, append-only guards, and restart replay. +- Package smoke installs the tarball into a clean consumer and verifies `pausemesh/postgres` without + leaking PostgreSQL symbols into the root export. +- Local PostgreSQL 17 integration: one real test passed against an ephemeral container and the + container was removed after verification. +- Local quality gate: `pnpm check` passed with 193 tests passed and one opt-in PostgreSQL test + skipped in the ordinary coverage run; statements 86.38%, branches 82.35%, functions 92.51%, and + lines 87.20%. Build and clean-consumer package smoke produced and verified + `pausemesh-0.3.0-alpha.1.tgz`. diff --git a/docs/postgres-runbook.md b/docs/postgres-runbook.md new file mode 100644 index 0000000..8b9cad7 --- /dev/null +++ b/docs/postgres-runbook.md @@ -0,0 +1,110 @@ +# PostgreSQL adapter runbook + +This runbook covers the optional `pausemesh/postgres` adapter in `0.3.0-alpha.1`. The built-in CLI +remains a SQLite reference server; PostgreSQL is composed by the production host. + +## 1. Own the pool at the host boundary + +Install the PostgreSQL client chosen by the host and pass a compatible pool to PauseMesh. Configure +TLS, connection and statement timeouts, a bounded pool size, and provider-specific options outside +the adapter. Register an idle-client error listener and close the pool during graceful shutdown. +Never log the connection string. + +```ts +import { Pool } from "pg"; +import { PostgresEventStore } from "pausemesh/postgres"; + +const pool = new Pool({ + connectionString: process.env.DATABASE_URL, + connectionTimeoutMillis: 5_000, + idleTimeoutMillis: 30_000, + max: 10, + query_timeout: 5_000, + statement_timeout: 5_000, +}); + +pool.on("error", () => { + // Mark the host unhealthy through its protected telemetry path; do not log pool config. +}); + +const eventStore = new PostgresEventStore(pool, { + schema: "pausemesh", + maxEventsPerStream: 32, +}); +``` + +`PostgresEventStore` deliberately has no `end()` method. The host must await `pool.end()` after it +has stopped accepting requests and drained in-flight work. + +## 2. Separate migration and runtime roles + +Run `migratePostgresEventStore` from a deployment process with a DDL-capable role before starting +or updating replicas. Do not call it in an application constructor or on every startup. + +```ts +import { migratePostgresEventStore } from "pausemesh/postgres"; + +await migratePostgresEventStore(migrationPool, { schema: "pausemesh" }); +``` + +For the default schema, the runtime role needs only: + +- `USAGE` on schema `pausemesh`; +- `SELECT` on `pausemesh_schema_migrations`; +- `SELECT`, `INSERT`, and `UPDATE` on `continuation_streams`; +- `SELECT` and `INSERT` on `continuation_events`. + +Do not grant the runtime role DDL, `DELETE`, or `TRUNCATE`. The adapter's readiness check validates +the migration version/checksum, physical tables, expected active trigger bindings, and read access, +but intentionally does not attest the complete trigger function body or every table constraint. +Only the controlled migration role may modify schema objects. + +## 3. Wire liveness and readiness separately + +Pass the store as the host-owned readiness probe: + +```ts +const app = createHttpApp({ + maxPayloadBytes: 1_048_576, + readinessProbe: eventStore, + service: continuations, +}); +``` + +- `GET /healthz` answers `200 {"status":"ok"}` while the process can serve HTTP. It never probes + PostgreSQL. +- `GET /readyz` answers 200 only when the migration and required PostgreSQL objects are valid and + accessible. Missing configuration or any probe error returns a bounded 503 response without the + database host, SQL text, credentials, or underlying error. + +The orchestrator should remove a replica from service on `/readyz` failure but restart it only from +an independent liveness policy. Configure pool/query timeouts so readiness has a deployment-specific +upper bound. + +## 4. Deploy and verify + +1. Back up according to the provider policy and stop schema-changing jobs. +2. Run the explicit migration with the migration role. Concurrent runners are serialized, but the + deployment pipeline should still have one migration stage. +3. Start one canary replica with the runtime role. +4. Verify `/healthz`, `/readyz`, create, inspect through another replica, exact resume retry, and + event history. +5. Roll out remaining replicas and monitor connection saturation, readiness failures, CAS conflicts, + and database latency without recording payloads or tokens. + +The v1 migration only creates a dedicated schema. There is no automatic downgrade. If a first-time +deployment must be abandoned before data is accepted, stop all replicas and remove that dedicated +schema with the migration role. Once continuations exist, restore from backup or deploy a forward +migration; never delete event rows to roll back application code. + +## 5. Test locally + +Set `PAUSEMESH_TEST_POSTGRES_URL` to an isolated database and run: + +```bash +pnpm test:postgres +``` + +The test creates and removes a random dedicated schema. CI also sets +`PAUSEMESH_REQUIRE_POSTGRES_TEST=1`, which makes a missing URL a hard failure rather than a skipped +test. diff --git a/package.json b/package.json index a18eb25..9f6ffd4 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "pausemesh", - "version": "0.2.0-alpha.1", + "version": "0.3.0-alpha.1", "description": "Protocol-neutral durable pause and one-shot resume primitives for agent workflows.", "author": "Antonio Antenore", "type": "module", @@ -30,6 +30,10 @@ ".": { "types": "./dist/index.d.ts", "import": "./dist/index.js" + }, + "./postgres": { + "types": "./dist/postgres.d.ts", + "import": "./dist/postgres.js" } }, "bin": { @@ -56,6 +60,7 @@ "test": "vitest run", "test:coverage": "vitest run --coverage", "test:package": "node scripts/package-smoke.mjs", + "test:postgres": "vitest run tests/postgres.integration.test.ts", "test:watch": "vitest", "typecheck": "tsc --noEmit" }, @@ -69,7 +74,9 @@ "devDependencies": { "@biomejs/biome": "2.5.4", "@types/node": "24.13.3", + "@types/pg": "8.20.0", "@vitest/coverage-v8": "4.1.10", + "pg": "8.22.0", "tsx": "4.23.1", "typescript": "7.0.2", "vitest": "4.1.10" diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7d5c34e..198c098 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -30,9 +30,15 @@ importers: '@types/node': specifier: 24.13.3 version: 24.13.3 + '@types/pg': + specifier: 8.20.0 + version: 8.20.0 '@vitest/coverage-v8': specifier: 4.1.10 version: 4.1.10(vitest@4.1.10) + pg: + specifier: 8.22.0 + version: 8.22.0 tsx: specifier: 4.23.1 version: 4.23.1 @@ -435,6 +441,9 @@ packages: '@types/node@24.13.3': resolution: {integrity: sha512-Dh8vAsV36ig5wa9OX4pXvMc9D3Veibfw2wix0CUwYODLD8nkj9UsLjASr49nPg+2eKzxhBV+v7L8pXvT4e639Q==} + '@types/pg@8.20.0': + resolution: {integrity: sha512-bEPFOaMAHTEP1EzpvHTbmwR8UsFyHSKsRisLIHVMXnpNefSbGA1bD6CVy+qKjGSqmZqNqBDV2azOBo8TgkcVow==} + '@typescript/typescript-aix-ppc64@7.0.2': resolution: {integrity: sha512-MTKKkWB7p/0E9xi1d1tHtZ5PiLkGEMIq88pK2CubZjOsLtYTLqhgIgi6zepFa+9GHZ6h05NMCkQxGKiPXMxXtQ==} engines: {node: '>=16.20.0'} @@ -770,6 +779,40 @@ packages: pathe@2.0.3: resolution: {integrity: sha512-WUjGcAqP1gQacoQe+OBJsFA7Ld4DyXuUIjZ5cc75cLHvJ7dtNsTugphxIADwspS+AraAUePCKrSVtPLFj/F88w==} + pg-cloudflare@1.4.0: + resolution: {integrity: sha512-Vo7z/6rrQYxpNRylp4Tlob2elzbh+N/MOQbxFVWCxS7oEx6jF53GTJFxK2WWpKuBRkmiin4Mt+xofFDjx09R0A==} + + pg-connection-string@2.14.0: + resolution: {integrity: sha512-XwWDGcLRGCXAR8F/AM5bG7Q+A3Wm2s6QeEjlOKZLlH3UYcguiqCWKyWXVag5TLTIjR7oOJUY8kcADaZgWPyLeg==} + + pg-int8@1.0.1: + resolution: {integrity: sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw==} + engines: {node: '>=4.0.0'} + + pg-pool@3.14.0: + resolution: {integrity: sha512-gKtPkFdQPU3DksooVLi9LsjZxrsBUZIpa+7aVx+LV5pNh0KzP4Zleud2po+ConrxbuXGBJ6Hfer6hdgpIBpBaw==} + peerDependencies: + pg: '>=8.0' + + pg-protocol@1.15.0: + resolution: {integrity: sha512-cq9sECI5s0+uPUXjbz8ioyPJni6RzsRib0US67i5IoTZKw8fNeYlVE7u8F4dG7vEJJtc5wdD1K189lCCUwqWTQ==} + + pg-types@2.2.0: + resolution: {integrity: sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA==} + engines: {node: '>=4'} + + pg@8.22.0: + resolution: {integrity: sha512-8wih1vVIBMxoUM2oB4soJsD9tDnDpLv4OXBJ+EJzFsvycD+lfyIreC2gGHq78f8jbLLt+bvlPTFdFZfJkOuzAA==} + engines: {node: '>= 16.0.0'} + peerDependencies: + pg-native: '>=3.0.1' + peerDependenciesMeta: + pg-native: + optional: true + + pgpass@1.0.5: + resolution: {integrity: sha512-FdW9r/jQZhSeohs1Z3sI1yxFQNFvMcnmfuj4WBMUTxOrAyLMaTcE1aAMBiTlbMNaXvBCQuVi0R7hd8udDSP7ug==} + picocolors@1.1.1: resolution: {integrity: sha512-xceH2snhtb5M9liqDsmEw56le376mTZkEX/jEb/RxNFyegNul7eNslCXP9FDj/Lcu0X8KEyMceP2ntpaHrDEVA==} @@ -791,6 +834,22 @@ packages: resolution: {integrity: sha512-Mz8SaolMd8nB+G13WkORcxQKHZ/NE4xXevtkJHVuG+guo9/wYKlIMTKAqGdEmYOXR2ijPjTYNHssizdaVSUNdQ==} engines: {node: ^10 || ^12 || >=14} + postgres-array@2.0.0: + resolution: {integrity: sha512-VpZrUqU5A69eQyW2c5CA1jtLecCsN2U/bD6VilrFDWq5+5UIEVO7nazS3TEcHf1zuPYO/sqGvUvW62g86RXZuA==} + engines: {node: '>=4'} + + postgres-bytea@1.0.1: + resolution: {integrity: sha512-5+5HqXnsZPE65IJZSMkZtURARZelel2oXUEO8rH83VS/hxH5vv1uHquPg5wZs8yMAfdv971IU+kcPUczi7NVBQ==} + engines: {node: '>=0.10.0'} + + postgres-date@1.0.7: + resolution: {integrity: sha512-suDmjLVQg78nMK2UZ454hAG+OAW+HQPZ6n++TNDUX+L0+uUlLywnoxJKDou51Zm+zTCjrCl0Nq6J9C5hP9vK/Q==} + engines: {node: '>=0.10.0'} + + postgres-interval@1.2.0: + resolution: {integrity: sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ==} + engines: {node: '>=0.10.0'} + process-warning@5.0.0: resolution: {integrity: sha512-a39t9ApHNx2L4+HBnQKqxxHNs1r7KF+Intd8Q/g1bUh6q0WIp9voPXJ/x0j+ZL45KF1pJd9+q2jLIRMfvEshkA==} @@ -966,6 +1025,10 @@ packages: engines: {node: '>=8'} hasBin: true + xtend@4.0.2: + resolution: {integrity: sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ==} + engines: {node: '>=0.4'} + zod@3.25.76: resolution: {integrity: sha512-gzUt/qt81nXsFGKIFcC3YnfEAx5NkunCfnDlvuBSSFS02bcXu4Lmea0AFIUwbLWxWPx3d9p8S5QoaujKcNQxcQ==} @@ -1217,6 +1280,12 @@ snapshots: dependencies: undici-types: 7.18.2 + '@types/pg@8.20.0': + dependencies: + '@types/node': 24.13.3 + pg-protocol: 1.15.0 + pg-types: 2.2.0 + '@typescript/typescript-aix-ppc64@7.0.2': optional: true @@ -1484,6 +1553,41 @@ snapshots: pathe@2.0.3: {} + pg-cloudflare@1.4.0: + optional: true + + pg-connection-string@2.14.0: {} + + pg-int8@1.0.1: {} + + pg-pool@3.14.0(pg@8.22.0): + dependencies: + pg: 8.22.0 + + pg-protocol@1.15.0: {} + + pg-types@2.2.0: + dependencies: + pg-int8: 1.0.1 + postgres-array: 2.0.0 + postgres-bytea: 1.0.1 + postgres-date: 1.0.7 + postgres-interval: 1.2.0 + + pg@8.22.0: + dependencies: + pg-connection-string: 2.14.0 + pg-pool: 3.14.0(pg@8.22.0) + pg-protocol: 1.15.0 + pg-types: 2.2.0 + pgpass: 1.0.5 + optionalDependencies: + pg-cloudflare: 1.4.0 + + pgpass@1.0.5: + dependencies: + split2: 4.2.0 + picocolors@1.1.1: {} picomatch@4.0.5: {} @@ -1514,6 +1618,16 @@ snapshots: picocolors: 1.1.1 source-map-js: 1.2.1 + postgres-array@2.0.0: {} + + postgres-bytea@1.0.1: {} + + postgres-date@1.0.7: {} + + postgres-interval@1.2.0: + dependencies: + xtend: 4.0.2 + process-warning@5.0.0: {} quick-format-unescaped@4.0.4: {} @@ -1660,6 +1774,8 @@ snapshots: siginfo: 2.0.0 stackback: 0.0.2 + xtend@4.0.2: {} + zod@3.25.76: {} zod@4.4.3: {} diff --git a/scripts/package-smoke.mjs b/scripts/package-smoke.mjs index 6c95c7c..956a60e 100644 --- a/scripts/package-smoke.mjs +++ b/scripts/package-smoke.mjs @@ -83,6 +83,8 @@ function verifyPackedFiles(packageRoot) { const expectedPaths = [ ["dist/index.js", "the ESM entry point"], ["dist/index.d.ts", "the public type declarations"], + ["dist/postgres.js", "the optional PostgreSQL entry point"], + ["dist/postgres.d.ts", "the optional PostgreSQL declarations"], ["dist/cli.js", "the CLI entry point"], ["docs/delivery-contract.md", "the linked delivery contract"], ["docs/adr", "the linked architecture decisions"], @@ -133,6 +135,7 @@ function verifyConsumerInstall(tarballPath) { writeFileSync( smokeModule, 'import * as pausemesh from "pausemesh";\n' + + 'import * as postgres from "pausemesh/postgres";\n' + "const expected = [\n" + ' "ContinuationService", "SqliteEventStore", "issueMcpElicitation",\n' + ' "parseMcpElicitResult", "toMcpElicitRequest",\n' + @@ -145,6 +148,17 @@ function verifyConsumerInstall(tarballPath) { ' if (typeof pausemesh[name] !== "function") {\n' + ' throw new Error("Package export " + name + " is unavailable");\n' + " }\n" + + "}\n" + + 'for (const name of ["PostgresEventStore", "migratePostgresEventStore"]) {\n' + + ' if (typeof postgres[name] !== "function") {\n' + + ' throw new Error("PostgreSQL package export " + name + " is unavailable");\n' + + " }\n" + + "}\n" + + "if (postgres.POSTGRES_EVENT_STORE_SCHEMA_VERSION !== 1) {\n" + + ' throw new Error("PostgreSQL schema version export is unavailable");\n' + + "}\n" + + 'if ("PostgresEventStore" in pausemesh) {\n' + + ' throw new Error("Optional PostgreSQL adapter leaked into the root export");\n' + "}\n", ); run(process.execPath, [smokeModule], consumerRoot); diff --git a/src/adapters/http/app.ts b/src/adapters/http/app.ts index c478d07..4d964fc 100644 --- a/src/adapters/http/app.ts +++ b/src/adapters/http/app.ts @@ -15,6 +15,7 @@ import { TokenMismatchError, VersionConflictError, } from "../../domain/index.js"; +import type { ReadinessProbe } from "../../ports/readiness.js"; import { toA2AInterruptedTask } from "../a2a/task.js"; import { issueAguiInterrupts } from "../agui/interrupt.js"; import { issueMcpElicitation, type McpProjectionPolicy } from "../mcp/elicitation.js"; @@ -74,6 +75,8 @@ export interface CreateHttpAppOptions { maxPayloadBytes: number; /** Trusted host policy; never populated from an HTTP request body. */ mcpProjectionPolicy?: McpProjectionPolicy; + /** Host-owned dependency probe. Omission makes readiness fail closed. */ + readinessProbe?: ReadinessProbe; service: ContinuationService; } @@ -182,6 +185,34 @@ export function createHttpApp(options: CreateHttpAppOptions): Hono { app.get("/healthz", (c) => c.json({ status: "ok" })); + app.get("/readyz", async (c) => { + if (options.readinessProbe === undefined) { + return c.json( + { + status: "not_ready", + checks: [{ component: "event-store", status: "not_configured" }], + }, + 503, + ); + } + + try { + await options.readinessProbe.checkReadiness(); + return c.json({ + status: "ready", + checks: [{ component: "event-store", status: "ready" }], + }); + } catch { + return c.json( + { + status: "not_ready", + checks: [{ component: "event-store", status: "unavailable" }], + }, + 503, + ); + } + }); + app.post("/v1/continuations", boundedJsonBody, async (c) => { const body = CreateContinuationSchema.parse(await readBoundedJson(c, options.maxPayloadBytes)); const result = await options.service.create({ diff --git a/src/adapters/storage/in-memory-event-store.ts b/src/adapters/storage/in-memory-event-store.ts index 517e4dd..4ace51d 100644 --- a/src/adapters/storage/in-memory-event-store.ts +++ b/src/adapters/storage/in-memory-event-store.ts @@ -2,13 +2,14 @@ import { VersionConflictError } from "../../domain/errors.js"; import type { ContinuationEventV1 } from "../../domain/events.js"; import type { ContinuationId, ContinuationVersion } from "../../domain/types.js"; import type { EventStore } from "../../ports/event-store.js"; +import type { ReadinessProbe } from "../../ports/readiness.js"; import { deserializeEvent, serializeAppendBatch } from "./event-serialization.js"; /** * Process-local event store with the same JSON snapshot and optimistic concurrency semantics as * the durable SQLite adapter. */ -export class InMemoryEventStore implements EventStore { +export class InMemoryEventStore implements EventStore, ReadinessProbe { readonly #streams = new Map(); async load(continuationId: ContinuationId): Promise { @@ -31,4 +32,6 @@ export class InMemoryEventStore implements EventStore { this.#streams.set(continuationId, [...stream, ...serializedEvents]); } } + + async checkReadiness(): Promise {} } diff --git a/src/adapters/storage/postgres-event-store.ts b/src/adapters/storage/postgres-event-store.ts new file mode 100644 index 0000000..6cdfbb7 --- /dev/null +++ b/src/adapters/storage/postgres-event-store.ts @@ -0,0 +1,602 @@ +import { createHash } from "node:crypto"; +import { VersionConflictError } from "../../domain/errors.js"; +import type { ContinuationEventV1 } from "../../domain/events.js"; +import type { ContinuationId, ContinuationVersion } from "../../domain/types.js"; +import type { EventStore } from "../../ports/event-store.js"; +import type { ReadinessProbe } from "../../ports/readiness.js"; +import { deserializeEvent, serializeAppendBatch } from "./event-serialization.js"; + +const DEFAULT_SCHEMA = "pausemesh"; +const DEFAULT_MAX_EVENTS_PER_STREAM = 32; +const MAX_SAFE_VERSION = Number.MAX_SAFE_INTEGER; +const MIGRATION_LOCK_NAMESPACE = 1_347_566_917; +const MIGRATION_LOCK_KEY = 1; +const SCHEMA_NAME_PATTERN = /^[A-Za-z_][A-Za-z0-9_]{0,62}$/; + +export const POSTGRES_EVENT_STORE_SCHEMA_VERSION = 1 as const; + +export interface PostgresQueryResultLike { + readonly rowCount: number | null; + readonly rows: readonly Record[]; +} + +export interface PostgresQueryableLike { + query(text: string, values?: unknown[]): Promise; +} + +export interface PostgresClientLike extends PostgresQueryableLike { + release(error?: Error | boolean): void; +} + +/** + * Minimal structural contract accepted by the adapter. The application composition root owns the + * actual pool, its idle-error listener, timeouts, and shutdown; the store never calls `end()`. + */ +export interface PostgresPoolLike extends PostgresQueryableLike { + connect(): Promise; +} + +export interface PostgresEventStoreOptions { + /** Maximum number of lifecycle events reconstructed for one continuation. */ + readonly maxEventsPerStream?: number; + /** Dedicated PostgreSQL schema. ASCII identifiers only; defaults to `pausemesh`. */ + readonly schema?: string; +} + +export interface MigratePostgresEventStoreOptions { + /** Dedicated PostgreSQL schema. ASCII identifiers only; defaults to `pausemesh`. */ + readonly schema?: string; +} + +export type PostgresEventStoreErrorCode = + | "CORRUPT_STREAM" + | "MIGRATION_CHECKSUM_MISMATCH" + | "MIGRATION_REQUIRED" + | "MIGRATION_VERSION_AHEAD" + | "SCHEMA_INCOMPLETE" + | "STREAM_TOO_LARGE"; + +export class PostgresEventStoreError extends Error { + constructor( + readonly code: PostgresEventStoreErrorCode, + message: string, + options?: ErrorOptions, + ) { + super(message, options); + this.name = "PostgresEventStoreError"; + } +} + +interface MigrationDefinition { + readonly checksum: string; + readonly name: string; + readonly statements: readonly string[]; + readonly version: number; +} + +interface MigrationRow { + readonly checksum: unknown; + readonly name: unknown; + readonly version: unknown; +} + +interface SchemaNames { + readonly migrationTable: string; + readonly schema: string; + readonly schemaSql: string; + readonly streamsTable: string; + readonly eventsTable: string; +} + +function quoteIdentifier(identifier: string): string { + return `"${identifier.replaceAll('"', '""')}"`; +} + +function schemaNames(schemaOption: string | undefined): SchemaNames { + const schema = schemaOption ?? DEFAULT_SCHEMA; + if (!SCHEMA_NAME_PATTERN.test(schema)) { + throw new TypeError( + "PostgreSQL schema must be a 1-63 character ASCII identifier starting with a letter or underscore", + ); + } + + const schemaSql = quoteIdentifier(schema); + return { + schema, + schemaSql, + migrationTable: `${schemaSql}."pausemesh_schema_migrations"`, + streamsTable: `${schemaSql}."continuation_streams"`, + eventsTable: `${schemaSql}."continuation_events"`, + }; +} + +function canonicalMigrationStatements(): readonly string[] { + const schema = "{{schema}}"; + return [ + `CREATE TABLE ${schema}."continuation_streams" ( + continuation_id TEXT PRIMARY KEY, + current_version BIGINT NOT NULL CHECK ( + current_version BETWEEN 0 AND ${MAX_SAFE_VERSION} + ) + )`, + `CREATE TABLE ${schema}."continuation_events" ( + continuation_id TEXT NOT NULL REFERENCES ${schema}."continuation_streams" (continuation_id), + version BIGINT NOT NULL CHECK (version BETWEEN 1 AND ${MAX_SAFE_VERSION}), + event_json TEXT NOT NULL CHECK (jsonb_typeof(event_json::jsonb) = 'object'), + PRIMARY KEY (continuation_id, version) + )`, + `CREATE FUNCTION ${schema}."pausemesh_reject_mutation"() + RETURNS trigger + LANGUAGE plpgsql + AS $$ + BEGIN + RAISE EXCEPTION 'PauseMesh persistence is append-only'; + END; + $$`, + `CREATE TRIGGER pausemesh_events_no_update_delete + BEFORE UPDATE OR DELETE ON ${schema}."continuation_events" + FOR EACH ROW EXECUTE FUNCTION ${schema}."pausemesh_reject_mutation"()`, + `CREATE TRIGGER pausemesh_events_no_truncate + BEFORE TRUNCATE ON ${schema}."continuation_events" + FOR EACH STATEMENT EXECUTE FUNCTION ${schema}."pausemesh_reject_mutation"()`, + `CREATE TRIGGER pausemesh_streams_no_delete + BEFORE DELETE ON ${schema}."continuation_streams" + FOR EACH ROW EXECUTE FUNCTION ${schema}."pausemesh_reject_mutation"()`, + `CREATE TRIGGER pausemesh_streams_no_truncate + BEFORE TRUNCATE ON ${schema}."continuation_streams" + FOR EACH STATEMENT EXECUTE FUNCTION ${schema}."pausemesh_reject_mutation"()`, + `CREATE TRIGGER pausemesh_migrations_no_update_delete + BEFORE UPDATE OR DELETE ON ${schema}."pausemesh_schema_migrations" + FOR EACH ROW EXECUTE FUNCTION ${schema}."pausemesh_reject_mutation"()`, + `CREATE TRIGGER pausemesh_migrations_no_truncate + BEFORE TRUNCATE ON ${schema}."pausemesh_schema_migrations" + FOR EACH STATEMENT EXECUTE FUNCTION ${schema}."pausemesh_reject_mutation"()`, + ]; +} + +function migrationFor(names: SchemaNames): MigrationDefinition { + const canonicalStatements = canonicalMigrationStatements(); + return { + version: POSTGRES_EVENT_STORE_SCHEMA_VERSION, + name: "postgres_multi_replica_event_store", + checksum: createHash("sha256").update(canonicalStatements.join(";\n")).digest("hex"), + statements: canonicalStatements.map((statement) => + statement.replaceAll("{{schema}}", names.schemaSql), + ), + }; +} + +export const POSTGRES_EVENT_STORE_SCHEMA_CHECKSUM = migrationFor( + schemaNames(DEFAULT_SCHEMA), +).checksum; + +function parseDatabaseVersion(value: unknown, label: string): number { + if ( + (typeof value !== "number" && typeof value !== "string") || + (typeof value === "string" && !/^\d+$/.test(value)) + ) { + throw new PostgresEventStoreError("CORRUPT_STREAM", `${label} is not an integer`); + } + + const parsed = typeof value === "number" ? value : Number(value); + if (!Number.isSafeInteger(parsed) || parsed < 0) { + throw new PostgresEventStoreError("CORRUPT_STREAM", `${label} is outside the safe range`); + } + return parsed; +} + +function validateExpectedVersion(expectedVersion: number, eventCount: number): void { + if (!Number.isSafeInteger(expectedVersion) || expectedVersion < 0) { + throw new RangeError("expectedVersion must be a non-negative safe integer"); + } + if (!Number.isSafeInteger(expectedVersion + eventCount)) { + throw new RangeError("The appended stream version exceeds the safe integer range"); + } +} + +function actualVersionFromRows(rows: readonly Record[]): number { + const row = rows[0]; + return row === undefined ? 0 : parseDatabaseVersion(row.current_version, "current_version"); +} + +function releaseFailure(error: unknown): Error | boolean { + return error instanceof Error ? error : true; +} + +async function rollbackPreservingPrimaryError( + client: PostgresClientLike, + primaryError: unknown, +): Promise { + try { + await client.query("ROLLBACK"); + return undefined; + } catch (rollbackError) { + if (primaryError instanceof Error && rollbackError instanceof Error) { + Object.defineProperty(primaryError, "rollbackError", { + configurable: true, + enumerable: false, + value: rollbackError, + }); + } + return releaseFailure(rollbackError); + } +} + +function requireMigrationRows( + rows: readonly Record[], + migration: MigrationDefinition, +): void { + const migrations = rows as readonly unknown[] as readonly MigrationRow[]; + const future = migrations.find( + (row) => parseDatabaseVersion(row.version, "migration version") > migration.version, + ); + if (future !== undefined) { + throw new PostgresEventStoreError( + "MIGRATION_VERSION_AHEAD", + "PostgreSQL event-store schema is newer than this PauseMesh adapter", + ); + } + + const applied = migrations.find( + (row) => parseDatabaseVersion(row.version, "migration version") === migration.version, + ); + if (applied === undefined) { + throw new PostgresEventStoreError( + "MIGRATION_REQUIRED", + "PostgreSQL event-store migration is not applied", + ); + } + if (applied.name !== migration.name || applied.checksum !== migration.checksum) { + throw new PostgresEventStoreError( + "MIGRATION_CHECKSUM_MISMATCH", + "PostgreSQL event-store migration checksum does not match the adapter", + ); + } +} + +async function readMigrationRows( + queryable: PostgresQueryableLike, + names: SchemaNames, +): Promise[]> { + const result = await queryable.query( + `SELECT version::text AS version, name, checksum + FROM ${names.migrationTable} + ORDER BY version ASC`, + ); + return result.rows; +} + +async function requireSchemaObjects( + queryable: PostgresQueryableLike, + names: SchemaNames, +): Promise { + const result = await queryable.query( + `WITH expected_triggers(table_name, trigger_name) AS ( + VALUES + ('continuation_events', 'pausemesh_events_no_update_delete'), + ('continuation_events', 'pausemesh_events_no_truncate'), + ('continuation_streams', 'pausemesh_streams_no_delete'), + ('continuation_streams', 'pausemesh_streams_no_truncate'), + ('pausemesh_schema_migrations', 'pausemesh_migrations_no_update_delete'), + ('pausemesh_schema_migrations', 'pausemesh_migrations_no_truncate') + ), valid_triggers AS ( + SELECT expected.trigger_name + FROM expected_triggers expected + JOIN pg_catalog.pg_namespace table_namespace ON table_namespace.nspname = $1 + JOIN pg_catalog.pg_class table_class + ON table_class.relnamespace = table_namespace.oid + AND table_class.relname = expected.table_name + AND table_class.relkind = 'r' + JOIN pg_catalog.pg_trigger trigger + ON trigger.tgrelid = table_class.oid + AND trigger.tgname = expected.trigger_name + AND NOT trigger.tgisinternal + AND trigger.tgenabled IN ('O', 'A') + JOIN pg_catalog.pg_proc trigger_function ON trigger_function.oid = trigger.tgfoid + JOIN pg_catalog.pg_namespace function_namespace + ON function_namespace.oid = trigger_function.pronamespace + AND function_namespace.nspname = $1 + WHERE trigger_function.proname = 'pausemesh_reject_mutation' + AND trigger_function.pronargs = 0 + ) + SELECT + EXISTS ( + SELECT 1 FROM pg_catalog.pg_class c + JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = $1 AND c.relname = 'continuation_streams' AND c.relkind = 'r' + ) AS streams_exists, + EXISTS ( + SELECT 1 FROM pg_catalog.pg_class c + JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = $1 AND c.relname = 'continuation_events' AND c.relkind = 'r' + ) AS events_exists, + EXISTS ( + SELECT 1 FROM pg_catalog.pg_class c + JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = $1 + AND c.relname = 'pausemesh_schema_migrations' + AND c.relkind = 'r' + ) AS migrations_exists, + ( + SELECT COUNT(*)::text FROM expected_triggers + WHERE trigger_name NOT IN (SELECT trigger_name FROM valid_triggers) + ) AS missing_trigger_count`, + [names.schema], + ); + const row = result.rows[0]; + const missingTriggerCount = row?.missing_trigger_count; + if ( + row?.streams_exists !== true || + row.events_exists !== true || + row.migrations_exists !== true || + parseDatabaseVersion(missingTriggerCount, "missing_trigger_count") !== 0 + ) { + throw new PostgresEventStoreError( + "SCHEMA_INCOMPLETE", + "PostgreSQL event-store schema objects are incomplete", + ); + } + + // These zero-row reads verify that the runtime role can access both tables. + await queryable.query(`SELECT current_version FROM ${names.streamsTable} WHERE FALSE`); + await queryable.query(`SELECT event_json FROM ${names.eventsTable} WHERE FALSE`); +} + +/** + * Applies and validates the explicit PostgreSQL schema migration. Run this with a DDL-capable + * migration role before starting application replicas; constructors deliberately never migrate. + */ +export async function migratePostgresEventStore( + pool: PostgresPoolLike, + options: MigratePostgresEventStoreOptions = {}, +): Promise { + const names = schemaNames(options.schema); + const migration = migrationFor(names); + const client = await pool.connect(); + let releaseError: Error | boolean | undefined; + + try { + await client.query("BEGIN ISOLATION LEVEL READ COMMITTED"); + await client.query("SELECT pg_advisory_xact_lock($1, $2)", [ + MIGRATION_LOCK_NAMESPACE, + MIGRATION_LOCK_KEY, + ]); + await client.query(`CREATE SCHEMA IF NOT EXISTS ${names.schemaSql}`); + await client.query( + `CREATE TABLE IF NOT EXISTS ${names.migrationTable} ( + version INTEGER PRIMARY KEY CHECK (version > 0), + name TEXT NOT NULL, + checksum TEXT NOT NULL CHECK (checksum ~ '^[a-f0-9]{64}$'), + applied_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP + )`, + ); + + const before = await readMigrationRows(client, names); + const future = before.find( + (row) => parseDatabaseVersion(row.version, "migration version") > migration.version, + ); + if (future !== undefined) { + throw new PostgresEventStoreError( + "MIGRATION_VERSION_AHEAD", + "PostgreSQL event-store schema is newer than this PauseMesh adapter", + ); + } + + const existing = before.find( + (row) => parseDatabaseVersion(row.version, "migration version") === migration.version, + ); + if (existing !== undefined) { + requireMigrationRows(before, migration); + } else { + for (const statement of migration.statements) { + await client.query(statement); + } + await client.query( + `INSERT INTO ${names.migrationTable} (version, name, checksum) + VALUES ($1, $2, $3)`, + [migration.version, migration.name, migration.checksum], + ); + } + + requireMigrationRows(await readMigrationRows(client, names), migration); + await requireSchemaObjects(client, names); + await client.query("COMMIT"); + } catch (error) { + releaseError = await rollbackPreservingPrimaryError(client, error); + throw error; + } finally { + client.release(releaseError); + } +} + +/** Multi-replica append-only event store backed by an externally owned PostgreSQL pool. */ +export class PostgresEventStore implements EventStore, ReadinessProbe { + readonly #maxEventsPerStream: number; + readonly #migration: MigrationDefinition; + readonly #names: SchemaNames; + readonly #pool: PostgresPoolLike; + + constructor(pool: PostgresPoolLike, options: PostgresEventStoreOptions = {}) { + const maxEventsPerStream = options.maxEventsPerStream ?? DEFAULT_MAX_EVENTS_PER_STREAM; + if (!Number.isSafeInteger(maxEventsPerStream) || maxEventsPerStream <= 0) { + throw new RangeError("maxEventsPerStream must be a positive safe integer"); + } + + this.#pool = pool; + this.#names = schemaNames(options.schema); + this.#migration = migrationFor(this.#names); + this.#maxEventsPerStream = maxEventsPerStream; + } + + async load(continuationId: ContinuationId): Promise { + const result = await this.#pool.query( + `SELECT + streams.current_version::text AS current_version, + events.version::text AS version, + events.event_json + FROM ${this.#names.streamsTable} AS streams + LEFT JOIN LATERAL ( + SELECT version, event_json + FROM ${this.#names.eventsTable} + WHERE continuation_id = streams.continuation_id + ORDER BY version ASC + LIMIT $2 + ) AS events ON TRUE + WHERE streams.continuation_id = $1 + ORDER BY events.version ASC`, + [continuationId, this.#maxEventsPerStream + 1], + ); + + if (result.rows.length === 0) return []; + + const currentVersion = parseDatabaseVersion(result.rows[0]?.current_version, "current_version"); + if ( + currentVersion > this.#maxEventsPerStream || + result.rows.length > this.#maxEventsPerStream + ) { + throw new PostgresEventStoreError( + "STREAM_TOO_LARGE", + `Continuation stream exceeds the configured ${this.#maxEventsPerStream}-event limit`, + ); + } + if (currentVersion === 0 || result.rows.length !== currentVersion) { + throw new PostgresEventStoreError( + "CORRUPT_STREAM", + "Continuation stream head does not match its event count", + ); + } + + return result.rows.map((row, index) => { + const expectedVersion = index + 1; + const storedVersion = parseDatabaseVersion(row.version, "event version"); + if (storedVersion !== expectedVersion) { + throw new PostgresEventStoreError( + "CORRUPT_STREAM", + "Stored continuation stream contains a version gap", + ); + } + if (typeof row.event_json !== "string") { + throw new PostgresEventStoreError( + "CORRUPT_STREAM", + "Stored continuation event is not a JSON text snapshot", + ); + } + + let event: ContinuationEventV1; + try { + event = deserializeEvent(row.event_json); + } catch (cause) { + throw new PostgresEventStoreError( + "CORRUPT_STREAM", + "Stored continuation event does not match the persistence contract", + { cause }, + ); + } + if (event.continuationId !== continuationId) { + throw new PostgresEventStoreError( + "CORRUPT_STREAM", + "Stored continuation event belongs to a different stream", + ); + } + if (event.version !== storedVersion) { + throw new PostgresEventStoreError( + "CORRUPT_STREAM", + "Stored continuation event embeds a different version", + ); + } + return event; + }); + } + + async append( + continuationId: ContinuationId, + expectedVersion: ContinuationVersion, + events: readonly ContinuationEventV1[], + ): Promise { + validateExpectedVersion(expectedVersion, events.length); + if (expectedVersion + events.length > this.#maxEventsPerStream) { + throw new PostgresEventStoreError( + "STREAM_TOO_LARGE", + `Continuation stream exceeds the configured ${this.#maxEventsPerStream}-event limit`, + ); + } + const serializedEvents = serializeAppendBatch(continuationId, expectedVersion, events); + + // A no-op still compares the head, but does not create a ghost stream or open a transaction. + if (serializedEvents.length === 0) { + const current = await this.#pool.query( + `SELECT current_version::text AS current_version + FROM ${this.#names.streamsTable} + WHERE continuation_id = $1`, + [continuationId], + ); + const actualVersion = actualVersionFromRows(current.rows); + if (actualVersion !== expectedVersion) { + throw new VersionConflictError(continuationId, expectedVersion, actualVersion); + } + return; + } + + const nextVersion = expectedVersion + serializedEvents.length; + const client = await this.#pool.connect(); + let releaseError: Error | boolean | undefined; + + try { + await client.query("BEGIN ISOLATION LEVEL READ COMMITTED"); + if (expectedVersion === 0) { + await client.query( + `INSERT INTO ${this.#names.streamsTable} (continuation_id, current_version) + VALUES ($1, 0) + ON CONFLICT (continuation_id) DO NOTHING`, + [continuationId], + ); + } + + const fenced = await client.query( + `UPDATE ${this.#names.streamsTable} + SET current_version = $3 + WHERE continuation_id = $1 AND current_version = $2 + RETURNING current_version::text AS current_version`, + [continuationId, expectedVersion, nextVersion], + ); + if (fenced.rowCount !== 1) { + const current = await client.query( + `SELECT current_version::text AS current_version + FROM ${this.#names.streamsTable} + WHERE continuation_id = $1`, + [continuationId], + ); + throw new VersionConflictError( + continuationId, + expectedVersion, + actualVersionFromRows(current.rows), + ); + } + + const versions = serializedEvents.map((_, index) => String(expectedVersion + index + 1)); + const inserted = await client.query( + `INSERT INTO ${this.#names.eventsTable} (continuation_id, version, event_json) + SELECT $1, batch.version, batch.event_json + FROM UNNEST($2::bigint[], $3::text[]) AS batch(version, event_json)`, + [continuationId, versions, [...serializedEvents]], + ); + if (inserted.rowCount !== serializedEvents.length) { + throw new PostgresEventStoreError( + "CORRUPT_STREAM", + "PostgreSQL did not persist the complete continuation event batch", + ); + } + await client.query("COMMIT"); + } catch (error) { + releaseError = await rollbackPreservingPrimaryError(client, error); + throw error; + } finally { + client.release(releaseError); + } + } + + async checkReadiness(): Promise { + requireMigrationRows(await readMigrationRows(this.#pool, this.#names), this.#migration); + await requireSchemaObjects(this.#pool, this.#names); + } +} diff --git a/src/adapters/storage/sqlite-event-store.ts b/src/adapters/storage/sqlite-event-store.ts index b8cf203..ae551df 100644 --- a/src/adapters/storage/sqlite-event-store.ts +++ b/src/adapters/storage/sqlite-event-store.ts @@ -3,6 +3,7 @@ import { ContinuationIdMismatchError, VersionConflictError } from "../../domain/ import type { ContinuationEventV1 } from "../../domain/events.js"; import type { ContinuationId, ContinuationVersion } from "../../domain/types.js"; import type { EventStore } from "../../ports/event-store.js"; +import type { ReadinessProbe } from "../../ports/readiness.js"; import { deserializeEvent, serializeAppendBatch } from "./event-serialization.js"; const DEFAULT_BUSY_TIMEOUT_MS = 5_000; @@ -43,7 +44,7 @@ export interface SqliteEventStoreOptions { } /** Durable append-only event store backed by SQLite in WAL mode. */ -export class SqliteEventStore implements EventStore { +export class SqliteEventStore implements EventStore, ReadinessProbe { readonly #database: DatabaseSync; constructor(databasePath: string, options: SqliteEventStoreOptions = {}) { @@ -133,6 +134,10 @@ export class SqliteEventStore implements EventStore { } } + async checkReadiness(): Promise { + this.#database.prepare("SELECT 1").get(); + } + close(): void { if (this.#database.isOpen) { this.#database.close(); diff --git a/src/cli.ts b/src/cli.ts index 9b60ab9..41a2061 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -72,7 +72,11 @@ export function runCli(argv: string[] = process.argv.slice(2)): void { tokenIssuer: new Sha256TokenIssuer(), tokenTtlSeconds: config.tokenTtlSeconds, }); - const app = createHttpApp({ maxPayloadBytes: config.maxPayloadBytes, service }); + const app = createHttpApp({ + maxPayloadBytes: config.maxPayloadBytes, + readinessProbe: eventStore, + service, + }); const server = serve( { fetch: app.fetch, hostname: config.host, port: config.port }, ({ address, port }) => logger.info({ address, port }, "pausemesh.listening"), diff --git a/src/ports/index.ts b/src/ports/index.ts index 44d6f2a..1a6dfd5 100644 --- a/src/ports/index.ts +++ b/src/ports/index.ts @@ -1,4 +1,5 @@ export * from "./clock.js"; export * from "./event-store.js"; export * from "./observer.js"; +export * from "./readiness.js"; export * from "./token-issuer.js"; diff --git a/src/ports/readiness.ts b/src/ports/readiness.ts new file mode 100644 index 0000000..478dd77 --- /dev/null +++ b/src/ports/readiness.ts @@ -0,0 +1,9 @@ +/** + * A dependency readiness check owned by the application composition root. + * + * Implementations either resolve or throw. HTTP adapters deliberately project + * only the bounded ready/unavailable state and never expose the thrown error. + */ +export interface ReadinessProbe { + checkReadiness(): Promise; +} diff --git a/src/postgres.ts b/src/postgres.ts new file mode 100644 index 0000000..1986f70 --- /dev/null +++ b/src/postgres.ts @@ -0,0 +1 @@ +export * from "./adapters/storage/postgres-event-store.js"; diff --git a/tests/http.test.ts b/tests/http.test.ts index d9acdd5..18dc208 100644 --- a/tests/http.test.ts +++ b/tests/http.test.ts @@ -5,11 +5,16 @@ import type { McpProjectionPolicy } from "../src/adapters/mcp/elicitation.js"; import { InMemoryEventStore } from "../src/adapters/storage/index.js"; import { ContinuationService } from "../src/application/index.js"; import { NoopObserver } from "../src/observability/pino-observer.js"; +import type { ReadinessProbe } from "../src/ports/readiness.js"; -function app(mcpProjectionPolicy?: McpProjectionPolicy) { +function app( + mcpProjectionPolicy?: McpProjectionPolicy, + readiness: ReadinessProbe | "default" | "omitted" = "default", +) { + const eventStore = new InMemoryEventStore(); const service = new ContinuationService({ clock: { now: () => new Date("2026-07-15T08:00:00.000Z") }, - eventStore: new InMemoryEventStore(), + eventStore, generateId: (() => { let id = 0; return () => `id-${++id}`; @@ -22,10 +27,59 @@ function app(mcpProjectionPolicy?: McpProjectionPolicy) { maxPayloadBytes: 4_096, service, ...(mcpProjectionPolicy === undefined ? {} : { mcpProjectionPolicy }), + ...(readiness === "omitted" + ? {} + : { readinessProbe: readiness === "default" ? eventStore : readiness }), }); } describe("HTTP adapter", () => { + it("keeps liveness independent and reports a ready event store", async () => { + const api = app(); + + const health = await api.request("/healthz"); + const readiness = await api.request("/readyz"); + + expect(health.status).toBe(200); + expect(await health.json()).toEqual({ status: "ok" }); + expect(readiness.status).toBe(200); + expect(await readiness.json()).toEqual({ + status: "ready", + checks: [{ component: "event-store", status: "ready" }], + }); + }); + + it("fails readiness without leaking probe errors while liveness remains healthy", async () => { + const api = app(undefined, { + checkReadiness: async () => { + throw new Error("postgresql://secret-user:secret-password@private-db/internal"); + }, + }); + + const readiness = await api.request("/readyz"); + const readinessBody = JSON.stringify(await readiness.json()); + + expect(readiness.status).toBe(503); + expect(readinessBody).toBe( + JSON.stringify({ + status: "not_ready", + checks: [{ component: "event-store", status: "unavailable" }], + }), + ); + expect(readinessBody).not.toContain("secret"); + expect((await api.request("/healthz")).status).toBe(200); + }); + + it("fails closed when the composition root omits a readiness probe", async () => { + const response = await app(undefined, "omitted").request("/readyz"); + + expect(response.status).toBe(503); + expect(await response.json()).toEqual({ + status: "not_ready", + checks: [{ component: "event-store", status: "not_configured" }], + }); + }); + it("creates, projects, resumes, and inspects a continuation", async () => { const api = app(); const createdResponse = await api.request("/v1/continuations", { diff --git a/tests/postgres-event-store.test.ts b/tests/postgres-event-store.test.ts new file mode 100644 index 0000000..cc255f8 --- /dev/null +++ b/tests/postgres-event-store.test.ts @@ -0,0 +1,631 @@ +import { describe, expect, it } from "vitest"; +import type { ContinuationCreatedV1, ContinuationResumedV1 } from "../src/domain/events.js"; +import { + migratePostgresEventStore, + POSTGRES_EVENT_STORE_SCHEMA_CHECKSUM, + type PostgresClientLike, + PostgresEventStore, + PostgresEventStoreError, + type PostgresPoolLike, + type PostgresQueryResultLike, +} from "../src/postgres.js"; + +const CONTINUATION_ID = "continuation-postgres-1"; + +function createdEvent(overrides: Partial = {}): ContinuationCreatedV1 { + return { + type: "continuation.created", + schemaVersion: 1, + continuationId: CONTINUATION_ID, + correlationId: "correlation-postgres-1", + version: 1, + occurredAt: "2026-07-19T08:00:00.000Z", + expiresAt: "2026-07-19T08:15:00.000Z", + payload: { action: "approve" }, + metadata: { source: "postgres-unit-test" }, + resumeTokenHash: "a".repeat(64), + ...overrides, + }; +} + +function resumedEvent(overrides: Partial = {}): ContinuationResumedV1 { + return { + type: "continuation.resumed", + schemaVersion: 1, + continuationId: CONTINUATION_ID, + version: 2, + occurredAt: "2026-07-19T08:01:00.000Z", + idempotencyKey: "postgres-request-1", + resumePayload: { approved: true }, + ...overrides, + }; +} + +function result( + rows: readonly Record[] = [], + rowCount: number | null = rows.length, +): PostgresQueryResultLike { + return { rowCount, rows }; +} + +interface QueryStep { + readonly error?: unknown; + readonly result?: PostgresQueryResultLike; + readonly text: RegExp | string; + readonly values?: readonly unknown[]; + readonly verifyValues?: (values: readonly unknown[] | undefined) => void; +} + +interface RecordedQuery { + readonly text: string; + readonly values?: readonly unknown[]; +} + +function normalized(text: string): string { + return text.replace(/\s+/g, " ").trim(); +} + +class ScriptedQueryable { + readonly queries: RecordedQuery[] = []; + readonly #steps: QueryStep[]; + + constructor(steps: readonly QueryStep[] = []) { + this.#steps = [...steps]; + } + + async query(text: string, values?: unknown[]): Promise { + this.queries.push({ text, ...(values === undefined ? {} : { values }) }); + const step = this.#steps.shift(); + if (step === undefined) { + throw new Error(`Unexpected query: ${normalized(text)}`); + } + + const actual = normalized(text); + if (typeof step.text === "string") { + expect(actual).toContain(step.text); + } else { + expect(actual).toMatch(step.text); + } + if (step.values !== undefined) expect(values).toEqual(step.values); + step.verifyValues?.(values); + if ("error" in step) throw step.error; + return step.result ?? result(); + } + + expectConsumed(): void { + expect(this.#steps, "unconsumed scripted queries").toEqual([]); + } +} + +class ScriptedClient implements PostgresClientLike { + readonly script: ScriptedQueryable; + readonly releases: (Error | boolean | undefined)[] = []; + + constructor(steps: readonly QueryStep[]) { + this.script = new ScriptedQueryable(steps); + } + + query(text: string, values?: unknown[]): Promise { + return this.script.query(text, values); + } + + release(error?: Error | boolean): void { + this.releases.push(error); + } +} + +class ScriptedPool implements PostgresPoolLike { + readonly direct: ScriptedQueryable; + readonly clients: ScriptedClient[]; + connectCount = 0; + + constructor(directSteps: readonly QueryStep[] = [], clients: readonly ScriptedClient[] = []) { + this.direct = new ScriptedQueryable(directSteps); + this.clients = [...clients]; + } + + query(text: string, values?: unknown[]): Promise { + return this.direct.query(text, values); + } + + async connect(): Promise { + this.connectCount += 1; + const client = this.clients.shift(); + if (client === undefined) throw new Error("Unexpected pool.connect() call"); + return client; + } +} + +function migrationRow( + overrides: Partial> = {}, +): Record { + return { + version: "1", + name: "postgres_multi_replica_event_store", + checksum: POSTGRES_EVENT_STORE_SCHEMA_CHECKSUM, + ...overrides, + }; +} + +function schemaValidationSteps(missingTriggerCount = "0"): readonly QueryStep[] { + return [ + { + text: "FROM pg_catalog.pg_class", + values: ["pausemesh"], + result: result([ + { + streams_exists: true, + events_exists: true, + migrations_exists: true, + missing_trigger_count: missingTriggerCount, + }, + ]), + }, + { text: 'SELECT current_version FROM "pausemesh"."continuation_streams" WHERE FALSE' }, + { text: 'SELECT event_json FROM "pausemesh"."continuation_events" WHERE FALSE' }, + ]; +} + +function migrationPrefix(rows: readonly Record[]): readonly QueryStep[] { + return [ + { text: "BEGIN" }, + { + text: "SELECT pg_advisory_xact_lock($1, $2)", + values: [1_347_566_917, 1], + }, + { text: 'CREATE SCHEMA IF NOT EXISTS "pausemesh"' }, + { text: 'CREATE TABLE IF NOT EXISTS "pausemesh"."pausemesh_schema_migrations"' }, + { + text: 'FROM "pausemesh"."pausemesh_schema_migrations" ORDER BY version ASC', + result: result(rows), + }, + ]; +} + +function newMigrationSteps(): readonly QueryStep[] { + return [ + ...migrationPrefix([]), + { text: 'CREATE TABLE "pausemesh"."continuation_streams"' }, + { text: 'CREATE TABLE "pausemesh"."continuation_events"' }, + { text: 'CREATE FUNCTION "pausemesh"."pausemesh_reject_mutation"()' }, + { text: "CREATE TRIGGER pausemesh_events_no_update_delete" }, + { text: "CREATE TRIGGER pausemesh_events_no_truncate" }, + { text: "CREATE TRIGGER pausemesh_streams_no_delete" }, + { text: "CREATE TRIGGER pausemesh_streams_no_truncate" }, + { text: "CREATE TRIGGER pausemesh_migrations_no_update_delete" }, + { text: "CREATE TRIGGER pausemesh_migrations_no_truncate" }, + { + text: 'INSERT INTO "pausemesh"."pausemesh_schema_migrations"', + values: [1, "postgres_multi_replica_event_store", POSTGRES_EVENT_STORE_SCHEMA_CHECKSUM], + }, + { + text: 'FROM "pausemesh"."pausemesh_schema_migrations" ORDER BY version ASC', + result: result([migrationRow()]), + }, + ...schemaValidationSteps(), + { text: "COMMIT" }, + ]; +} + +function existingMigrationSteps(): readonly QueryStep[] { + return [ + ...migrationPrefix([migrationRow()]), + { + text: 'FROM "pausemesh"."pausemesh_schema_migrations" ORDER BY version ASC', + result: result([migrationRow()]), + }, + ...schemaValidationSteps(), + { text: "COMMIT" }, + ]; +} + +describe("PostgreSQL migration", () => { + it("applies the versioned schema on one checked-out client", async () => { + const client = new ScriptedClient(newMigrationSteps()); + const pool = new ScriptedPool([], [client]); + + await migratePostgresEventStore(pool); + + expect(pool.connectCount).toBe(1); + expect(client.releases).toEqual([undefined]); + client.script.expectConsumed(); + }); + + it("is idempotent when the exact migration checksum is already present", async () => { + const client = new ScriptedClient(existingMigrationSteps()); + + await migratePostgresEventStore(new ScriptedPool([], [client])); + + expect( + client.script.queries.some((query) => normalized(query.text).startsWith("CREATE TRIGGER")), + ).toBe(false); + expect(client.releases).toEqual([undefined]); + client.script.expectConsumed(); + }); + + it.each([ + { + code: "MIGRATION_CHECKSUM_MISMATCH", + row: migrationRow({ checksum: "0".repeat(64) }), + }, + { + code: "MIGRATION_VERSION_AHEAD", + row: migrationRow({ version: "2" }), + }, + ])("rolls back incompatible migration metadata: $code", async ({ code, row }) => { + const client = new ScriptedClient([...migrationPrefix([row]), { text: "ROLLBACK" }]); + + await expect(migratePostgresEventStore(new ScriptedPool([], [client]))).rejects.toMatchObject({ + code, + }); + expect(client.releases).toEqual([undefined]); + client.script.expectConsumed(); + }); + + it("fails closed and rolls back when required schema objects are absent", async () => { + const client = new ScriptedClient([ + ...migrationPrefix([migrationRow()]), + { + text: 'FROM "pausemesh"."pausemesh_schema_migrations" ORDER BY version ASC', + result: result([migrationRow()]), + }, + { + text: "FROM pg_catalog.pg_class", + result: result([ + { + streams_exists: true, + events_exists: false, + migrations_exists: true, + missing_trigger_count: "0", + }, + ]), + }, + { text: "ROLLBACK" }, + ]); + + await expect(migratePostgresEventStore(new ScriptedPool([], [client]))).rejects.toMatchObject({ + code: "SCHEMA_INCOMPLETE", + }); + client.script.expectConsumed(); + }); + + it("preserves the primary error and discards a client when rollback also fails", async () => { + const rollbackError = new Error("rollback connection failure"); + const client = new ScriptedClient([ + ...migrationPrefix([migrationRow({ version: "2" })]), + { text: "ROLLBACK", error: rollbackError }, + ]); + + const failure = migratePostgresEventStore(new ScriptedPool([], [client])); + await expect(failure).rejects.toMatchObject({ code: "MIGRATION_VERSION_AHEAD" }); + expect(client.releases).toEqual([rollbackError]); + client.script.expectConsumed(); + }); + + it("rejects unsafe configurable schema names before acquiring a connection", async () => { + const pool = new ScriptedPool(); + + await expect( + migratePostgresEventStore(pool, { schema: 'tenant";DROP SCHEMA public' }), + ).rejects.toThrow(/ASCII identifier/); + expect(pool.connectCount).toBe(0); + }); +}); + +describe("PostgresEventStore", () => { + it("rejects invalid bounds and schema configuration", () => { + const pool = new ScriptedPool(); + + expect(() => new PostgresEventStore(pool, { maxEventsPerStream: 0 })).toThrow( + /positive safe integer/, + ); + expect(() => new PostgresEventStore(pool, { schema: "invalid-schema" })).toThrow( + /ASCII identifier/, + ); + }); + + it("pre-validates the complete batch before acquiring a client", async () => { + const pool = new ScriptedPool(); + const store = new PostgresEventStore(pool); + + await expect( + store.append(CONTINUATION_ID, 0, [createdEvent(), resumedEvent({ version: 3 })]), + ).rejects.toMatchObject({ code: "VERSION_CONFLICT" }); + await expect(store.append(CONTINUATION_ID, -1, [])).rejects.toThrow(/non-negative/); + await expect( + store.append(CONTINUATION_ID, Number.MAX_SAFE_INTEGER, [createdEvent()]), + ).rejects.toThrow(/exceeds the safe integer range/); + await expect( + new PostgresEventStore(pool, { maxEventsPerStream: 1 }).append(CONTINUATION_ID, 1, [ + resumedEvent(), + ]), + ).rejects.toMatchObject({ code: "STREAM_TOO_LARGE" }); + expect(pool.connectCount).toBe(0); + }); + + it("compares an empty batch without a transaction or ghost stream", async () => { + const pool = new ScriptedPool([ + { + text: 'SELECT current_version::text AS current_version FROM "pausemesh"."continuation_streams"', + values: [CONTINUATION_ID], + result: result([]), + }, + { + text: 'SELECT current_version::text AS current_version FROM "pausemesh"."continuation_streams"', + values: [CONTINUATION_ID], + result: result([{ current_version: "2" }]), + }, + ]); + const store = new PostgresEventStore(pool); + + await store.append(CONTINUATION_ID, 0, []); + await expect(store.append(CONTINUATION_ID, 1, [])).rejects.toMatchObject({ + code: "VERSION_CONFLICT", + expectedVersion: 1, + actualVersion: 2, + }); + expect(pool.connectCount).toBe(0); + expect(pool.direct.queries.every((query) => !normalized(query.text).includes("INSERT"))).toBe( + true, + ); + pool.direct.expectConsumed(); + }); + + it("atomically creates a head, fences it, and inserts one ordered batch", async () => { + const client = new ScriptedClient([ + { text: "BEGIN" }, + { + text: 'INSERT INTO "pausemesh"."continuation_streams"', + values: [CONTINUATION_ID], + }, + { + text: 'UPDATE "pausemesh"."continuation_streams" SET current_version = $3', + values: [CONTINUATION_ID, 0, 2], + result: result([{ current_version: "2" }], 1), + }, + { + text: 'INSERT INTO "pausemesh"."continuation_events"', + verifyValues: (values) => { + expect(values).toBeDefined(); + if (values === undefined) throw new Error("Expected query values"); + expect(values?.[0]).toBe(CONTINUATION_ID); + expect(values?.[1]).toEqual(["1", "2"]); + expect((values[2] as string[]).map((event) => JSON.parse(event))).toEqual([ + createdEvent(), + resumedEvent(), + ]); + }, + result: result([], 2), + }, + { text: "COMMIT" }, + ]); + const pool = new ScriptedPool([], [client]); + + await new PostgresEventStore(pool).append(CONTINUATION_ID, 0, [createdEvent(), resumedEvent()]); + + expect(client.releases).toEqual([undefined]); + client.script.expectConsumed(); + }); + + it("does not recreate a head when appending to an existing stream", async () => { + const client = new ScriptedClient([ + { text: "BEGIN" }, + { + text: 'UPDATE "pausemesh"."continuation_streams" SET current_version = $3', + values: [CONTINUATION_ID, 1, 2], + result: result([{ current_version: "2" }], 1), + }, + { + text: 'INSERT INTO "pausemesh"."continuation_events"', + result: result([], 1), + }, + { text: "COMMIT" }, + ]); + + await new PostgresEventStore(new ScriptedPool([], [client])).append(CONTINUATION_ID, 1, [ + resumedEvent(), + ]); + + client.script.expectConsumed(); + }); + + it("translates a lost CAS fence to VersionConflictError and rolls back", async () => { + const client = new ScriptedClient([ + { text: "BEGIN" }, + { text: 'INSERT INTO "pausemesh"."continuation_streams"' }, + { + text: 'UPDATE "pausemesh"."continuation_streams" SET current_version = $3', + result: result([], 0), + }, + { + text: 'SELECT current_version::text AS current_version FROM "pausemesh"."continuation_streams"', + result: result([{ current_version: "1" }]), + }, + { text: "ROLLBACK" }, + ]); + + await expect( + new PostgresEventStore(new ScriptedPool([], [client])).append(CONTINUATION_ID, 0, [ + createdEvent(), + ]), + ).rejects.toMatchObject({ + code: "VERSION_CONFLICT", + expectedVersion: 0, + actualVersion: 1, + }); + expect(client.releases).toEqual([undefined]); + client.script.expectConsumed(); + }); + + it("rolls back when PostgreSQL reports a partial batch insert", async () => { + const client = new ScriptedClient([ + { text: "BEGIN" }, + { text: 'INSERT INTO "pausemesh"."continuation_streams"' }, + { + text: 'UPDATE "pausemesh"."continuation_streams" SET current_version = $3', + result: result([{ current_version: "2" }], 1), + }, + { text: 'INSERT INTO "pausemesh"."continuation_events"', result: result([], 1) }, + { text: "ROLLBACK" }, + ]); + + await expect( + new PostgresEventStore(new ScriptedPool([], [client])).append(CONTINUATION_ID, 0, [ + createdEvent(), + resumedEvent(), + ]), + ).rejects.toMatchObject({ code: "CORRUPT_STREAM" }); + client.script.expectConsumed(); + }); + + it("loads and validates a complete ordered stream", async () => { + const pool = new ScriptedPool([ + { + text: 'FROM "pausemesh"."continuation_streams" AS streams LEFT JOIN LATERAL', + values: [CONTINUATION_ID, 33], + result: result([ + { + current_version: "2", + version: "1", + event_json: JSON.stringify(createdEvent()), + }, + { + current_version: "2", + version: "2", + event_json: JSON.stringify(resumedEvent()), + }, + ]), + }, + { + text: 'FROM "pausemesh"."continuation_streams" AS streams LEFT JOIN LATERAL', + result: result([]), + }, + ]); + const store = new PostgresEventStore(pool); + + expect(await store.load(CONTINUATION_ID)).toEqual([createdEvent(), resumedEvent()]); + expect(await store.load("missing")).toEqual([]); + pool.direct.expectConsumed(); + }); + + it.each([ + { + label: "head count mismatch", + rows: [{ current_version: "2", version: "1", event_json: JSON.stringify(createdEvent()) }], + expected: { code: "CORRUPT_STREAM" }, + }, + { + label: "version gap", + rows: [{ current_version: "1", version: "2", event_json: JSON.stringify(createdEvent()) }], + expected: { code: "CORRUPT_STREAM" }, + }, + { + label: "non-text snapshot", + rows: [{ current_version: "1", version: "1", event_json: createdEvent() }], + expected: { code: "CORRUPT_STREAM" }, + }, + { + label: "embedded continuation mismatch", + rows: [ + { + current_version: "1", + version: "1", + event_json: JSON.stringify(createdEvent({ continuationId: "other" })), + }, + ], + expected: { code: "CORRUPT_STREAM" }, + }, + { + label: "embedded version mismatch", + rows: [ + { + current_version: "1", + version: "1", + event_json: JSON.stringify(createdEvent({ version: 2 })), + }, + ], + expected: { code: "CORRUPT_STREAM" }, + }, + ])("rejects corrupted storage: $label", async ({ rows, expected }) => { + const store = new PostgresEventStore( + new ScriptedPool([{ text: "LEFT JOIN LATERAL", result: result(rows) }]), + ); + + await expect(store.load(CONTINUATION_ID)).rejects.toMatchObject(expected); + }); + + it("bounds reconstruction and rejects unsafe database BIGINT values", async () => { + const oversized = new PostgresEventStore( + new ScriptedPool([ + { + text: "LEFT JOIN LATERAL", + result: result([{ current_version: "3", version: "1", event_json: "{}" }]), + }, + ]), + { maxEventsPerStream: 2 }, + ); + await expect(oversized.load(CONTINUATION_ID)).rejects.toMatchObject({ + code: "STREAM_TOO_LARGE", + }); + + const unsafe = new PostgresEventStore( + new ScriptedPool([ + { + text: "LEFT JOIN LATERAL", + result: result([ + { + current_version: "9007199254740992", + version: "1", + event_json: JSON.stringify(createdEvent()), + }, + ]), + }, + ]), + ); + await expect(unsafe.load(CONTINUATION_ID)).rejects.toMatchObject({ code: "CORRUPT_STREAM" }); + }); + + it("reports ready only for the exact migration and complete accessible schema", async () => { + const pool = new ScriptedPool([ + { + text: 'FROM "pausemesh"."pausemesh_schema_migrations" ORDER BY version ASC', + result: result([migrationRow()]), + }, + ...schemaValidationSteps(), + ]); + + await new PostgresEventStore(pool).checkReadiness(); + + pool.direct.expectConsumed(); + }); + + it.each([ + { rows: [], code: "MIGRATION_REQUIRED" }, + { + rows: [migrationRow({ checksum: "f".repeat(64) })], + code: "MIGRATION_CHECKSUM_MISMATCH", + }, + { rows: [migrationRow({ version: "2" })], code: "MIGRATION_VERSION_AHEAD" }, + ])("fails readiness for incompatible ledger state: $code", async ({ rows, code }) => { + const store = new PostgresEventStore( + new ScriptedPool([ + { + text: 'FROM "pausemesh"."pausemesh_schema_migrations" ORDER BY version ASC', + result: result(rows), + }, + ]), + ); + + await expect(store.checkReadiness()).rejects.toBeInstanceOf(PostgresEventStoreError); + await expect( + new PostgresEventStore( + new ScriptedPool([ + { + text: 'FROM "pausemesh"."pausemesh_schema_migrations" ORDER BY version ASC', + result: result(rows), + }, + ]), + ).checkReadiness(), + ).rejects.toMatchObject({ code }); + }); +}); diff --git a/tests/postgres.integration.test.ts b/tests/postgres.integration.test.ts new file mode 100644 index 0000000..be6b99f --- /dev/null +++ b/tests/postgres.integration.test.ts @@ -0,0 +1,198 @@ +import { randomUUID } from "node:crypto"; +import { Pool } from "pg"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { Sha256TokenIssuer } from "../src/adapters/crypto/sha256-token-issuer.js"; +import { ContinuationService } from "../src/application/continuation-service.js"; +import type { ContinuationCreatedV1, ContinuationResumedV1 } from "../src/domain/events.js"; +import { NoopObserver } from "../src/observability/pino-observer.js"; +import { + migratePostgresEventStore, + PostgresEventStore, + type PostgresPoolLike, +} from "../src/postgres.js"; + +const connectionString = process.env.PAUSEMESH_TEST_POSTGRES_URL; +if (process.env.PAUSEMESH_REQUIRE_POSTGRES_TEST === "1" && connectionString === undefined) { + throw new Error("PAUSEMESH_TEST_POSTGRES_URL is required for the PostgreSQL CI integration test"); +} + +const describeWithPostgres = connectionString === undefined ? describe.skip : describe; + +describeWithPostgres("PostgresEventStore real multi-replica integration", () => { + const schema = `pausemesh_test_${randomUUID().replaceAll("-", "_")}`; + const activePools = new Set(); + const idleErrors: Error[] = []; + let firstPool: Pool; + let secondPool: Pool; + + function openPool(): Pool { + if (connectionString === undefined) throw new Error("PostgreSQL test URL is unavailable"); + const pool = new Pool({ + connectionString, + connectionTimeoutMillis: 5_000, + idleTimeoutMillis: 5_000, + max: 4, + query_timeout: 5_000, + statement_timeout: 5_000, + }); + pool.on("error", (error) => idleErrors.push(error)); + activePools.add(pool); + return pool; + } + + async function closePool(pool: Pool): Promise { + if (activePools.delete(pool)) await pool.end(); + } + + function asPoolLike(pool: Pool): PostgresPoolLike { + return pool; + } + + function service(eventStore: PostgresEventStore): ContinuationService { + return new ContinuationService({ + clock: { now: () => new Date("2026-07-19T08:00:00.000Z") }, + eventStore, + observer: new NoopObserver(), + tokenIssuer: new Sha256TokenIssuer(), + tokenTtlSeconds: 900, + }); + } + + function directCreatedEvent(continuationId: string): ContinuationCreatedV1 { + return { + type: "continuation.created", + schemaVersion: 1, + continuationId, + correlationId: `correlation-${continuationId}`, + version: 1, + occurredAt: "2026-07-19T08:00:00.000Z", + expiresAt: "2026-07-19T08:15:00.000Z", + payload: { action: "choose" }, + metadata: {}, + resumeTokenHash: "b".repeat(64), + }; + } + + function directResumedEvent( + continuationId: string, + idempotencyKey: string, + ): ContinuationResumedV1 { + return { + type: "continuation.resumed", + schemaVersion: 1, + continuationId, + version: 2, + occurredAt: "2026-07-19T08:01:00.000Z", + idempotencyKey, + resumePayload: { winner: idempotencyKey }, + }; + } + + beforeAll(() => { + firstPool = openPool(); + secondPool = openPool(); + }); + + afterAll(async () => { + const cleanupPool = [...activePools][0]; + if (cleanupPool !== undefined) { + await cleanupPool.query(`DROP SCHEMA IF EXISTS "${schema}" CASCADE`); + } + await Promise.all([...activePools].map((pool) => closePool(pool))); + expect(idleErrors).toEqual([]); + }); + + it("migrates concurrently and preserves CAS, idempotency, restart replay, and append-only data", async () => { + await Promise.all([ + migratePostgresEventStore(asPoolLike(firstPool), { schema }), + migratePostgresEventStore(asPoolLike(secondPool), { schema }), + ]); + // Explicit reruns validate the same version and checksum without replaying DDL. + await migratePostgresEventStore(asPoolLike(firstPool), { schema }); + + const firstStore = new PostgresEventStore(asPoolLike(firstPool), { schema }); + const secondStore = new PostgresEventStore(asPoolLike(secondPool), { schema }); + await expect(firstStore.checkReadiness()).resolves.toBeUndefined(); + await expect(secondStore.checkReadiness()).resolves.toBeUndefined(); + + const directId = "postgres-direct-cas"; + await firstStore.append(directId, 0, [directCreatedEvent(directId)]); + const directOutcomes = await Promise.allSettled([ + firstStore.append(directId, 1, [directResumedEvent(directId, "replica-a")]), + secondStore.append(directId, 1, [directResumedEvent(directId, "replica-b")]), + ]); + expect(directOutcomes.filter((outcome) => outcome.status === "fulfilled")).toHaveLength(1); + const directLoser = directOutcomes.find((outcome) => outcome.status === "rejected"); + expect(directLoser?.reason).toMatchObject({ + code: "VERSION_CONFLICT", + expectedVersion: 1, + actualVersion: 2, + }); + expect(await secondStore.load(directId)).toHaveLength(2); + + const firstService = service(firstStore); + const secondService = service(secondStore); + const continuationId = "postgres-service-multireplica"; + const created = await firstService.create({ + continuationId, + correlationId: "thread-postgres-integration", + payload: { kind: "input", message: "Choose" }, + }); + await expect(secondService.inspect(continuationId)).resolves.toMatchObject({ + status: "pending", + version: 1, + }); + + const resumeInputs = [ + { + continuationId, + idempotencyKey: "resume-from-a", + resumePayload: { choice: "a" }, + resumeToken: created.resumeToken, + }, + { + continuationId, + idempotencyKey: "resume-from-b", + resumePayload: { choice: "b" }, + resumeToken: created.resumeToken, + }, + ] as const; + const outcomes = await Promise.allSettled([ + firstService.resume(resumeInputs[0]), + secondService.resume(resumeInputs[1]), + ]); + const winnerIndex = outcomes.findIndex((outcome) => outcome.status === "fulfilled"); + expect(winnerIndex).toBeGreaterThanOrEqual(0); + expect(outcomes.filter((outcome) => outcome.status === "fulfilled")).toHaveLength(1); + expect(outcomes.filter((outcome) => outcome.status === "rejected")).toHaveLength(1); + expect(outcomes.find((outcome) => outcome.status === "rejected")?.reason).toMatchObject({ + code: "TOKEN_ALREADY_USED", + }); + + const winner = outcomes[winnerIndex]; + if (winner?.status !== "fulfilled") throw new Error("Expected one successful resume"); + const oppositeService = winnerIndex === 0 ? secondService : firstService; + const winningInput = resumeInputs[winnerIndex]; + if (winningInput === undefined) throw new Error("Expected a winning resume input"); + await expect(oppositeService.resume(winningInput)).resolves.toEqual(winner.value); + await expect( + oppositeService.resume({ ...winningInput, resumePayload: { choice: "changed" } }), + ).rejects.toMatchObject({ code: "IDEMPOTENCY_CONFLICT" }); + expect(await firstService.history(continuationId)).toHaveLength(2); + + await expect( + firstPool.query( + `UPDATE "${schema}"."continuation_events" SET event_json = event_json WHERE continuation_id = $1`, + [continuationId], + ), + ).rejects.toMatchObject({ code: "P0001" }); + expect(await secondService.history(continuationId)).toHaveLength(2); + + await closePool(firstPool); + await closePool(secondPool); + const restartedPool = openPool(); + const restartedStore = new PostgresEventStore(asPoolLike(restartedPool), { schema }); + await expect(restartedStore.checkReadiness()).resolves.toBeUndefined(); + await expect(service(restartedStore).inspect(continuationId)).resolves.toEqual(winner.value); + }); +}); diff --git a/tests/storage.test.ts b/tests/storage.test.ts index 716fca8..cb51e6f 100644 --- a/tests/storage.test.ts +++ b/tests/storage.test.ts @@ -184,6 +184,14 @@ describe("SqliteEventStore", () => { expect(() => reopened.close()).not.toThrow(); }); + it("reports readiness only while the SQLite dependency is open", async () => { + const store = openStore(await temporaryDatabasePath()); + + await expect(store.checkReadiness()).resolves.toBeUndefined(); + store.close(); + await expect(store.checkReadiness()).rejects.toThrow(); + }); + it("allows only one winner when two instances append at the same expected version", async () => { const databasePath = await temporaryDatabasePath(); const first = openStore(databasePath);