diff --git a/package.json b/package.json index 8f13bc8..fe3010a 100644 --- a/package.json +++ b/package.json @@ -6,6 +6,7 @@ "scripts": { "check": "tsc --noEmit", "self-test": "bun run src/selftest/run.ts", + "self-test:postgres-kafka-order": "bun run src/selftest/integration/postgres-kafka-order.ts", "test": "bun run src/selftest/run.ts", "cli": "bun scripts/agentrun-cli.ts" }, diff --git a/src/selftest/integration/postgres-kafka-order.ts b/src/selftest/integration/postgres-kafka-order.ts new file mode 100644 index 0000000..7d8fc65 --- /dev/null +++ b/src/selftest/integration/postgres-kafka-order.ts @@ -0,0 +1,148 @@ +import assert from "node:assert/strict"; +import { Pool, type PoolClient } from "pg"; +import type { CreateRunInput, JsonRecord, KafkaEventOutboxRecord } from "../../common/types.js"; +import { createPostgresAgentRunStore } from "../../mgr/postgres-store.js"; + +const connectionString = process.env.AGENTRUN_SELFTEST_DATABASE_URL?.trim(); +if (!connectionString) throw new Error("AGENTRUN_SELFTEST_DATABASE_URL is required; this integration test never falls back to memory"); + +const sessionId = "ses_pg_kafka_commit_order"; +const topic = "agentrun.event.v1"; +const gate = "746173865219047"; +const controlPool = new Pool({ connectionString, application_name: "agentrun-pg-order-control", max: 2 }); +const store = await createPostgresAgentRunStore({ + connectionString, + poolMax: 6, + eventOutbox: { enabled: true, topic, source: "agentrun-pg-order-test" }, +}); +let gateClient: PoolClient | null = null; +let gateHeld = false; + +try { + const runA = await store.createRun(runInput(sessionId)); + const runB = await store.createRun(runInput(sessionId)); + await controlPool.query("DELETE FROM agentrun_kafka_event_outbox"); + await controlPool.query("ALTER SEQUENCE agentrun_kafka_event_outbox_seq RESTART WITH 1"); + await installGateTrigger(controlPool, gate); + + gateClient = await controlPool.connect(); + await gateClient.query("SELECT pg_advisory_lock($1::bigint)", [gate]); + gateHeld = true; + + const appendA = store.appendEvent(runA.id, "backend_status", { phase: "pg-order-a", orderMarker: "A" }); + await waitForBlockedOutboxInsert(controlPool); + + let appendBSettled = false; + const appendB = store.appendEvent(runB.id, "backend_status", { phase: "pg-order-b", orderMarker: "B" }) + .finally(() => { appendBSettled = true; }); + await delay(250); + assert.equal(appendBSettled, false, "same-session event B committed before blocked event A"); + assert.deepEqual(await store.claimKafkaEventOutbox({ owner: "pre-release", leaseMs: 60_000, limit: 1 }), [], "uncommitted event A must not be bypassed by event B"); + + await gateClient.query("SELECT pg_advisory_unlock($1::bigint)", [gate]); + gateHeld = false; + await Promise.all([appendA, appendB]); + + const rows = await markerRows(controlPool); + assert.deepEqual(rows.map((row) => row.marker), ["A", "B"]); + assert.deepEqual(rows.map((row) => row.outboxSeq), [1, 2]); + assert.deepEqual(rows.map((row) => row.partitionKey), [sessionId, sessionId]); + + const first = singleClaim(await store.claimKafkaEventOutbox({ owner: "relay-a", leaseMs: 60_000, limit: 1 })); + assert.equal(marker(first), "A"); + await store.completeKafkaEventOutbox(first); + const second = singleClaim(await store.claimKafkaEventOutbox({ owner: "relay-b", leaseMs: 60_000, limit: 1 })); + assert.equal(marker(second), "B"); + await store.completeKafkaEventOutbox(second); + + console.log(JSON.stringify({ ok: true, test: "postgres-kafka-same-session-commit-order", outboxSeq: rows.map((row) => row.outboxSeq), publishOrder: [marker(first), marker(second)], partitionKey: sessionId, valuesPrinted: false })); +} finally { + if (gateClient && gateHeld) await gateClient.query("SELECT pg_advisory_unlock($1::bigint)", [gate]).catch(() => undefined); + gateClient?.release(); + await removeGateTrigger(controlPool).catch(() => undefined); + await store.close(); + await controlPool.end(); +} + +function runInput(targetSessionId: string): CreateRunInput { + return { + tenantId: "unidesk", + projectId: "pikasTech/agentrun", + workspaceRef: { kind: "host-path", path: "/tmp/agentrun-pg-order" }, + sessionRef: { sessionId: targetSessionId }, + resourceBundleRef: null, + providerId: "NC01", + backendProfile: "codex", + executionPolicy: { + sandbox: "workspace-write", + approval: "never", + timeoutMs: 60_000, + network: "default", + secretScope: { allowCredentialEcho: false, providerCredentials: [{ profile: "codex", secretRef: { name: "agentrun-v02-provider-codex", keys: ["auth.json", "config.toml"] } }] }, + }, + traceSink: null, + }; +} + +async function installGateTrigger(pool: Pool, advisoryGate: string): Promise { + await removeGateTrigger(pool); + await pool.query(` + CREATE FUNCTION agentrun_selftest_gate_outbox_a() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + IF NEW.value #>> '{event,payload,orderMarker}' = 'A' THEN + PERFORM pg_advisory_xact_lock(${advisoryGate}::bigint); + END IF; + RETURN NEW; + END; + $$; + CREATE TRIGGER agentrun_selftest_gate_outbox_a + BEFORE INSERT ON agentrun_kafka_event_outbox + FOR EACH ROW EXECUTE FUNCTION agentrun_selftest_gate_outbox_a(); + `); +} + +async function removeGateTrigger(pool: Pool): Promise { + await pool.query("DROP TRIGGER IF EXISTS agentrun_selftest_gate_outbox_a ON agentrun_kafka_event_outbox; DROP FUNCTION IF EXISTS agentrun_selftest_gate_outbox_a()"); +} + +async function waitForBlockedOutboxInsert(pool: Pool): Promise { + const deadline = Date.now() + 5_000; + while (Date.now() < deadline) { + const result = await pool.query<{ blocked: boolean }>(` + SELECT EXISTS ( + SELECT 1 FROM pg_stat_activity + WHERE application_name LIKE 'agentrun-mgr%' + AND wait_event_type = 'Lock' + AND query LIKE 'INSERT INTO agentrun_kafka_event_outbox%' + ) AS blocked + `); + if (result.rows[0]?.blocked) return; + await delay(20); + } + throw new Error("event A did not block inside the Kafka outbox insert trigger"); +} + +async function markerRows(pool: Pool): Promise> { + const result = await pool.query<{ outbox_seq: string | number; marker: string; partition_key: string }>(` + SELECT outbox_seq, value #>> '{event,payload,orderMarker}' AS marker, partition_key + FROM agentrun_kafka_event_outbox + WHERE value #>> '{event,payload,orderMarker}' IN ('A', 'B') + ORDER BY outbox_seq ASC + `); + return result.rows.map((row) => ({ outboxSeq: Number(row.outbox_seq), marker: row.marker, partitionKey: row.partition_key })); +} + +function singleClaim(items: KafkaEventOutboxRecord[]): KafkaEventOutboxRecord { + assert.equal(items.length, 1); + return items[0] as KafkaEventOutboxRecord; +} + +function marker(item: KafkaEventOutboxRecord): string | null { + const event = item.value.event as JsonRecord | undefined; + const payload = event?.payload as JsonRecord | undefined; + return typeof payload?.orderMarker === "string" ? payload.orderMarker : null; +} + +async function delay(ms: number): Promise { + await new Promise((resolve) => setTimeout(resolve, ms)); +}