feat(durable): Postgres storage at one round trip per commit, reads from memory

packages/durable/CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## [Unreleased]
 
+### Changed
+
+- Postgres storage is built for a database that is a network hop away. A commit is two round trips: one statement reads the session row, the batch's claimed ids, its documents, and their latest revision versions as JSON; one data-modifying CTE statement writes every table's rows, clears and adds revisions, and advances the session, each part gated on the session still carrying this owner at the sequence the read observed. A single statement is atomic, so no transaction brackets a commit, and a takeover by another process writes nothing and is reported as before.
+- Postgres storage serves repeated reads from memory. It is the Session's only writer, so a record it has read stays valid until it writes that record: conversation records, tasks and submissions by id and request, each conversation's newest entries and head markers, current document scans and addresses, and the materialized current content of documents are cached and kept current by every commit. Measured on one faux-model turn, admitting an input drops from 22 queries to 3, creating a conversation from 37 to 11, a plain turn from 61 to 9, and a turn with one tool call from 189 to 31.
+- `PostgresStorage.open` takes `migrate: false` to skip the schema check for deployments that migrate in an explicit step; an open is then two statements.
+
+
 ### Breaking Changes
 
 - The portable SQLite facade in `@earendil-works/pi-durable/storage/sqlite` is asynchronous: `SqliteDatabase` extends the new `SqliteExecutor` (`exec`, `run`, `get`, `all` by SQL text), `prepare` and `SqliteStatement` are removed, `transaction` takes an async callback that receives a transaction handle, and `close()` returns a promise. Custom adapters must be rewritten ([#10232](https://github.com/earendil-works/pi/pull/10232) by [@christianklotz](https://github.com/christianklotz)).

packages/durable/README.md

@@ -523,7 +523,7 @@ const usage = await harness.usage(context); // { models: { "openai/gpt-6-sol": U
 | Memory | `MemoryStorage` from the package root | Nothing is persisted. |
 | SQLite | `openNodeSqliteStorage(file)` from `@earendil-works/pi-durable/storage/sqlite/node` | One database file. WAL mode with `synchronous = NORMAL`: commits survive process crashes; the newest may be lost on power or host failure. |
 | JSONL | `openNodeJsonlStorage(directory, context)` from `@earendil-works/pi-durable/storage/jsonl/node` | Append-only files in one directory. Pass `{ fsync: true }` to flush before each commit marker. |
-| Postgres | `PostgresStorage.open(pool, { session })` from `@earendil-works/pi-durable/storage/postgres` | One Session per `session` name; many Sessions share one schema. Locked to one process. |
+| Postgres | `PostgresStorage.open(pool, { session })` from `@earendil-works/pi-durable/storage/postgres` | One Session per `session` name; many Sessions share one schema. Locked to one process. Built for a remote database: a commit is two round trips and repeated reads are served from memory. Pass `migrate: false` when the schema is migrated in an explicit step, so an open costs two statements. |
 
 One process owns a storage at a time; SQLite and JSONL do no cross-process locking. The portable SQLite and JSONL cores (`/storage/sqlite`, `/storage/jsonl`) run without Node APIs, for example on Bun or in Cloudflare Durable Objects, given an asynchronous `SqliteDatabase` facade or a `FileSystem` from `@earendil-works/pi-durable/env`.
 

packages/durable/src/storage/postgres/migrations.ts

@@ -101,8 +101,170 @@ const INITIAL_SCHEMA: readonly string[] = [
 	"CREATE INDEX document_revisions_by_kind ON {schema}.document_revisions (session_id, document_id, kind, seq DESC)",
 ];
 
+// One commit is one call. The function locks the session row, checks that the caller still owns the session at
+// the sequence it expects, validates the batch against the stored state, writes every table, and advances the
+// session. Each rule raises the adapter's message with its own SQLSTATE so the adapter can map it back; see
+// `commitErrorCodes` in storage.ts.
+//
+// Records and content travel inside the batch as JSON text and are cast to `json` on insert. Postgres accepts a
+// lone UTF-16 surrogate escape when it stores `json` but rejects it in any function that decodes string values,
+// such as `json_to_recordset` and `->>`; so nothing here decodes a record, and the current-only test matches the
+// stored record's text.
+const COMMIT_FUNCTION = `CREATE OR REPLACE FUNCTION {schema}.commit_batch(
+	p_session TEXT, p_owner TEXT, p_seq BIGINT, p_next_id BIGINT, p_batch JSON
+) RETURNS BIGINT LANGUAGE plpgsql AS $fn$
+DECLARE
+	v_owner TEXT;
+	v_seq BIGINT;
+	v_id BIGINT;
+	v_type TEXT;
+	v_exists BOOLEAN;
+	v_retired BIGINT;
+	v_version INTEGER;
+	v_key TEXT;
+	v_live JSONB := '{}'::jsonb;
+	v_count INTEGER;
+	doc RECORD;
+	stored RECORD;
+	live RECORD;
+BEGIN
+	SELECT owner, next_seq INTO v_owner, v_seq FROM {schema}.sessions WHERE session_id = p_session FOR UPDATE;
+	IF NOT FOUND THEN
+		RAISE EXCEPTION 'Durable Postgres session metadata is missing' USING ERRCODE = 'DU001';
+	END IF;
+	IF v_owner <> p_owner OR v_seq <> p_seq THEN
+		RAISE EXCEPTION 'Durable Postgres session % was opened by another process', p_session USING ERRCODE = 'DU002';
+	END IF;
+
+	SELECT c.id, r.record_type INTO v_id, v_type
+		FROM json_to_recordset(p_batch->'claims') AS c(id BIGINT, type TEXT)
+		JOIN {schema}.record_ids r ON r.session_id = p_session AND r.id = c.id
+		WHERE c.type IN ('conversation', 'entry', 'document') OR r.record_type <> c.type
+		LIMIT 1;
+	IF FOUND THEN
+		RAISE EXCEPTION 'ID % already belongs to %', v_id, v_type USING ERRCODE = 'DU003';
+	END IF;
+
+	FOR doc IN SELECT * FROM json_to_recordset(p_batch->'documents') AS x(
+		id BIGINT, is_create BOOLEAN, retire BOOLEAN, kind TEXT, family SMALLINT, key_value TEXT,
+		scope_kind TEXT, owner_id BIGINT, content_kind TEXT, version INTEGER)
+	LOOP
+		SELECT x.retired_at, x.kind, x.family, x.key_value, x.scope_kind, x.owner_id
+			INTO stored FROM {schema}.documents x WHERE x.session_id = p_session AND x.id = doc.id;
+		v_exists := FOUND;
+		IF NOT doc.is_create AND NOT v_exists THEN
+			RAISE EXCEPTION 'Unknown document: %', doc.id USING ERRCODE = 'DU004';
+		END IF;
+		IF doc.is_create AND v_exists THEN
+			RAISE EXCEPTION 'Document % already exists', doc.id USING ERRCODE = 'DU005';
+		END IF;
+		IF v_exists AND stored.retired_at IS NOT NULL THEN
+			RAISE EXCEPTION 'Document % is retired', doc.id USING ERRCODE = 'DU006';
+		END IF;
+		IF doc.content_kind = 'delta' THEN
+			SELECT r.version INTO v_version FROM {schema}.document_revisions r
+				WHERE r.session_id = p_session AND r.document_id = doc.id ORDER BY r.seq DESC LIMIT 1;
+			IF NOT FOUND THEN
+				RAISE EXCEPTION 'Document % delta has no base', doc.id USING ERRCODE = 'DU007';
+			END IF;
+			IF v_version <> doc.version THEN
+				RAISE EXCEPTION 'Document % version transition requires a base', doc.id USING ERRCODE = 'DU008';
+			END IF;
+		END IF;
+		IF doc.is_create OR doc.retire THEN
+			IF doc.is_create THEN
+				v_key := concat_ws('|', doc.kind, doc.scope_kind, doc.owner_id, doc.family, doc.key_value);
+			ELSE
+				v_key := concat_ws('|', stored.kind, stored.scope_kind, stored.owner_id, stored.family, stored.key_value);
+			END IF;
+			IF NOT v_live ? v_key THEN
+				IF doc.is_create THEN
+					SELECT count(*) INTO v_count FROM {schema}.documents x WHERE x.session_id = p_session
+						AND x.kind = doc.kind AND x.scope_kind = doc.scope_kind AND x.owner_id = doc.owner_id
+						AND x.family = doc.family AND x.key_value = doc.key_value AND x.retired_at IS NULL;
+				ELSE
+					SELECT count(*) INTO v_count FROM {schema}.documents x WHERE x.session_id = p_session
+						AND x.kind = stored.kind AND x.scope_kind = stored.scope_kind AND x.owner_id = stored.owner_id
+						AND x.family = stored.family AND x.key_value = stored.key_value AND x.retired_at IS NULL;
+				END IF;
+				v_count := LEAST(v_count, 1);
+			ELSE
+				v_count := (v_live->>v_key)::integer;
+			END IF;
+			IF doc.retire AND v_exists THEN v_count := v_count - 1; END IF;
+			IF doc.is_create AND NOT doc.retire THEN v_count := v_count + 1; END IF;
+			v_live := v_live || jsonb_build_object(v_key, v_count);
+		END IF;
+	END LOOP;
+	FOR live IN SELECT value FROM jsonb_each_text(v_live) LOOP
+		IF live.value::integer > 1 THEN
+			RAISE EXCEPTION 'Document address already has a current incarnation' USING ERRCODE = 'DU009';
+		END IF;
+	END LOOP;
+
+	INSERT INTO {schema}.record_ids (session_id, id, record_type)
+		SELECT p_session, c.id, c.type FROM json_to_recordset(p_batch->'claims') AS c(id BIGINT, type TEXT)
+		ON CONFLICT (session_id, id) DO NOTHING;
+	INSERT INTO {schema}.conversations (session_id, id, owner_conversation_id, owner_task_id, record)
+		SELECT p_session, x.id, x.owner_conversation_id, x.owner_task_id, x.record::json
+		FROM json_to_recordset(p_batch->'conversations')
+			AS x(id BIGINT, owner_conversation_id BIGINT, owner_task_id BIGINT, record TEXT);
+	INSERT INTO {schema}.entries (session_id, id, conversation_id, head, commit_seq, record)
+		SELECT p_session, x.id, x.conversation_id, x.head, p_seq, x.record::json
+		FROM json_to_recordset(p_batch->'entries') AS x(id BIGINT, conversation_id BIGINT, head BIGINT, record TEXT);
+	INSERT INTO {schema}.tasks (session_id, id, conversation_id, kind, status, abort_requested, background, record)
+		SELECT p_session, x.id, x.conversation_id, x.kind, x.status, x.abort_requested, x.background, x.record::json
+		FROM json_to_recordset(p_batch->'tasks') AS x(id BIGINT, conversation_id BIGINT, kind TEXT, status TEXT,
+			abort_requested BOOLEAN, background BOOLEAN, record TEXT)
+		ON CONFLICT (session_id, id) DO UPDATE SET conversation_id = excluded.conversation_id, kind = excluded.kind,
+			status = excluded.status, abort_requested = excluded.abort_requested, background = excluded.background,
+			record = excluded.record;
+	INSERT INTO {schema}.submissions (session_id, id, conversation_id, request_id, status, record)
+		SELECT p_session, x.id, x.conversation_id, x.request_id, x.status, x.record::json
+		FROM json_to_recordset(p_batch->'submissions')
+			AS x(id BIGINT, conversation_id BIGINT, request_id TEXT, status TEXT, record TEXT)
+		ON CONFLICT (session_id, id) DO UPDATE SET conversation_id = excluded.conversation_id,
+			request_id = excluded.request_id, status = excluded.status, record = excluded.record;
+
+	-- A retired record gains retiredAt by appending to its text, which keeps the rest of the record verbatim.
+	UPDATE {schema}.documents x SET retired_at = p_seq,
+		record = (left(x.record::text, length(x.record::text) - 1) || ',"retiredAt":' || p_seq || '}')::json
+		FROM json_to_recordset(p_batch->'documents') AS d(id BIGINT, is_create BOOLEAN, retire BOOLEAN)
+		WHERE x.session_id = p_session AND x.id = d.id AND d.retire AND NOT d.is_create;
+	INSERT INTO {schema}.documents
+		(session_id, id, kind, family, key_value, scope_kind, owner_id, created_at, retired_at, record)
+		SELECT p_session, d.id, d.kind, d.family, d.key_value, d.scope_kind, d.owner_id, p_seq,
+			CASE WHEN d.retire THEN p_seq END, d.record::json
+		FROM json_to_recordset(p_batch->'documents') AS d(id BIGINT, is_create BOOLEAN, retire BOOLEAN, kind TEXT,
+			family SMALLINT, key_value TEXT, scope_kind TEXT, owner_id BIGINT, record TEXT)
+		WHERE d.is_create;
+	-- A current-only document keeps no history: a base replaces its revisions, and a retirement clears them. The
+	-- record is matched as text; its only object values are the adapter's own, so the pair cannot appear elsewhere.
+	DELETE FROM {schema}.document_revisions r
+		USING json_to_recordset(p_batch->'documents') AS d(id BIGINT, retire BOOLEAN, content_kind TEXT)
+		JOIN {schema}.documents x ON x.session_id = p_session AND x.id = d.id
+		WHERE r.session_id = p_session AND r.document_id = d.id
+			AND (x.scope_kind <> 'conversation' OR x.record::text LIKE '%"history":"latest"%')
+			AND (d.retire OR d.content_kind = 'base');
+	INSERT INTO {schema}.document_revisions (session_id, document_id, seq, kind, version, content)
+		SELECT p_session, d.id, p_seq, d.content_kind, d.version, d.content::json
+		FROM json_to_recordset(p_batch->'documents')
+			AS d(id BIGINT, retire BOOLEAN, content_kind TEXT, version INTEGER, content TEXT)
+		JOIN {schema}.documents x ON x.session_id = p_session AND x.id = d.id
+		WHERE d.content_kind IS NOT NULL
+			AND NOT (d.retire AND (x.scope_kind <> 'conversation' OR x.record::text LIKE '%"history":"latest"%'));
+
+	UPDATE {schema}.sessions SET next_id = GREATEST(next_id, p_next_id), next_seq = p_seq + 1
+		WHERE session_id = p_session;
+	RETURN p_seq;
+END
+$fn$`;
+
 /** Immutable, ordered schema history. Append new migrations after the initial schema ships. */
-export const POSTGRES_MIGRATIONS: readonly PostgresMigration[] = [{ version: 1, statements: INITIAL_SCHEMA }];
+export const POSTGRES_MIGRATIONS: readonly PostgresMigration[] = [
+	{ version: 1, statements: INITIAL_SCHEMA },
+	{ version: 2, statements: [COMMIT_FUNCTION] },
+];
 
 export const CURRENT_POSTGRES_SCHEMA_VERSION = POSTGRES_MIGRATIONS.at(-1)?.version ?? 0;