test: verify PostgreSQL Kafka commit ordering

2026-07-10 03:54:33 +02:00
parent 51aaafd21d
commit 5cc3cc862e
2 changed files with 149 additions and 0 deletions
@@ -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"
  },
@@ -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<void> {
+  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<void> {
+  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<void> {
+  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<Array<{ outboxSeq: number; marker: string; partitionKey: string }>> {
+  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<void> {
+  await new Promise<void>((resolve) => setTimeout(resolve, ms));
+}