From 6959630ef10dea4f50c2295a48e8b3039af1f34a Mon Sep 17 00:00:00 2001 From: AgentRun Codex Date: Mon, 13 Jul 2026 20:29:23 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=89=8D=E8=BF=9B=E8=BF=81=E7=A7=BB=20K?= =?UTF-8?q?afka=20outbox=20=E7=B4=A2=E5=BC=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mgr/postgres-durable-queues.ts | 2 ++ src/mgr/postgres-store.ts | 13 ++++++++- src/selftest/cases/00-redaction-postgres.ts | 10 +++++-- src/selftest/cases/35-kafka-durable-outbox.ts | 27 ++++++++++++++++++- .../integration/postgres-kafka-order.ts | 24 +++++++++++++++-- 5 files changed, 70 insertions(+), 6 deletions(-) diff --git a/src/mgr/postgres-durable-queues.ts b/src/mgr/postgres-durable-queues.ts index 7d63c5f..69e3e98 100644 --- a/src/mgr/postgres-durable-queues.ts +++ b/src/mgr/postgres-durable-queues.ts @@ -49,7 +49,9 @@ CREATE TABLE IF NOT EXISTS agentrun_kafka_event_outbox ( CREATE INDEX IF NOT EXISTS agentrun_kafka_event_outbox_due_idx ON agentrun_kafka_event_outbox (delivered_at, next_attempt_at, lease_expires_at); +`; +export const kafkaOutboxPendingKeyOrderMigrationSql = ` CREATE INDEX IF NOT EXISTS agentrun_kafka_event_outbox_pending_key_order_idx ON agentrun_kafka_event_outbox (topic, partition_key, outbox_seq) WHERE delivered_at IS NULL; diff --git a/src/mgr/postgres-store.ts b/src/mgr/postgres-store.ts index 8de2067..a6cf049 100644 --- a/src/mgr/postgres-store.ts +++ b/src/mgr/postgres-store.ts @@ -11,7 +11,7 @@ import { backendCapabilitiesSqlValues, mergeBackendCapability } from "../common/ import { normalizeRunEventPayload, requireEventType, requireExternallyAppendableEventType, userMessagePayloadForCommand } from "../common/events.js"; import { buildKafkaEventOutboxRecord, canonicalAgentRunEventPartitionKey } from "./event-outbox.js"; import { assertRunnerDispatchReplay, attachRunnerDispatchIntent, newRunnerDispatchIntent } from "./runner-dispatch-intent.js"; -import { durableDispatchAndKafkaOutboxMigrationSql, durableQueueStatusFromRow, kafkaEventOutboxFromRow, runnerDispatchIntentFromRow, runnerDispatchOutcomeMigrationSql } from "./postgres-durable-queues.js"; +import { durableDispatchAndKafkaOutboxMigrationSql, durableQueueStatusFromRow, kafkaEventOutboxFromRow, kafkaOutboxPendingKeyOrderMigrationSql, runnerDispatchIntentFromRow, runnerDispatchOutcomeMigrationSql } from "./postgres-durable-queues.js"; import { fenceActiveRunnerDispatchIntents, fenceTerminalRunnerDispatchIntents, lockRunnerDispatchClaim, staleClaimError } from "./postgres-runner-dispatch.js"; interface PostgresStoreOptions { @@ -497,6 +497,11 @@ const postgresMigrations: MigrationDefinition[] = [ checksum: checksumSql(queueAttemptsMigrationSql), sql: queueAttemptsMigrationSql, }, + { + id: "015_v02_kafka_outbox_pending_key_order", + checksum: checksumSql(kafkaOutboxPendingKeyOrderMigrationSql), + sql: kafkaOutboxPendingKeyOrderMigrationSql, + }, ]; export function postgresMigrationContract(): JsonRecord { @@ -514,6 +519,12 @@ export function postgresMigrationContract(): JsonRecord { failureVisibility: ["failure_kind", "failure_message"], valuesPrinted: false, }, + kafkaOutbox: { + pendingKeyOrderIndexMigrationId: "015_v02_kafka_outbox_pending_key_order", + pendingKeyOrderIndex: "agentrun_kafka_event_outbox_pending_key_order_idx", + relayParallelismBound: "AGENTRUN_KAFKA_OUTBOX_BATCH_SIZE", + valuesPrinted: false, + }, checksums: Object.fromEntries(postgresMigrations.map((migration) => [migration.id, migration.checksum])), }; } diff --git a/src/selftest/cases/00-redaction-postgres.ts b/src/selftest/cases/00-redaction-postgres.ts index 9d1b0b2..898e109 100644 --- a/src/selftest/cases/00-redaction-postgres.ts +++ b/src/selftest/cases/00-redaction-postgres.ts @@ -3,6 +3,7 @@ import { openAgentRunStoreFromEnv } from "../../mgr/store.js"; import { postgresMigrationContract } from "../../mgr/postgres-store.js"; import { redactText } from "../../common/redaction.js"; import { AgentRunError } from "../../common/errors.js"; +import type { JsonRecord } from "../../common/types.js"; import type { SelfTestCase } from "../harness.js"; const selfTest: SelfTestCase = async () => { @@ -13,7 +14,7 @@ const selfTest: SelfTestCase = async () => { (error) => error instanceof AgentRunError && error.failureKind === "infra-failed" && error.message.includes("DATABASE_URL is required"), ); const postgresContract = postgresMigrationContract(); - assert.equal(postgresContract.latestMigrationId, "014_v02_queue_attempt_history_retry"); + assert.equal(postgresContract.latestMigrationId, "015_v02_kafka_outbox_pending_key_order"); assert.equal((postgresContract.migrationIds as string[]).includes("008_v01_dsflash_go_backend_profile"), true); assert.equal((postgresContract.migrationIds as string[]).includes("009_v01_dsflash_go_model_catalog"), true); assert.equal((postgresContract.migrationIds as string[]).includes("010_v01_queue_session_ref"), true); @@ -21,6 +22,7 @@ const selfTest: SelfTestCase = async () => { assert.equal((postgresContract.migrationIds as string[]).includes("012_v02_durable_dispatch_kafka_outbox"), true); assert.equal((postgresContract.migrationIds as string[]).includes("013_v02_runner_dispatch_outcome"), true); assert.equal((postgresContract.migrationIds as string[]).includes("014_v02_queue_attempt_history_retry"), true); + assert.equal((postgresContract.migrationIds as string[]).includes("015_v02_kafka_outbox_pending_key_order"), true); assert.ok(typeof (postgresContract.checksums as Record)["008_v01_dsflash_go_backend_profile"] === "string" && (postgresContract.checksums as Record)["008_v01_dsflash_go_backend_profile"].length > 0); assert.ok(typeof (postgresContract.checksums as Record)["009_v01_dsflash_go_model_catalog"] === "string" && (postgresContract.checksums as Record)["009_v01_dsflash_go_model_catalog"].length > 0); assert.ok(typeof (postgresContract.checksums as Record)["010_v01_queue_session_ref"] === "string" && (postgresContract.checksums as Record)["010_v01_queue_session_ref"].length > 0); @@ -28,7 +30,11 @@ const selfTest: SelfTestCase = async () => { assert.ok(typeof (postgresContract.checksums as Record)["012_v02_durable_dispatch_kafka_outbox"] === "string" && (postgresContract.checksums as Record)["012_v02_durable_dispatch_kafka_outbox"].length > 0); assert.ok(typeof (postgresContract.checksums as Record)["013_v02_runner_dispatch_outcome"] === "string" && (postgresContract.checksums as Record)["013_v02_runner_dispatch_outcome"].length > 0); assert.ok(typeof (postgresContract.checksums as Record)["014_v02_queue_attempt_history_retry"] === "string" && (postgresContract.checksums as Record)["014_v02_queue_attempt_history_retry"].length > 0); + assert.ok(typeof (postgresContract.checksums as Record)["015_v02_kafka_outbox_pending_key_order"] === "string" && (postgresContract.checksums as Record)["015_v02_kafka_outbox_pending_key_order"].length > 0); assert.equal((postgresContract.checksums as Record)["002_v01_backend_profiles"], "928b5c490cc4539cb64ecef34784557601b2724fa2870570f16a53576804e49c"); + assert.equal((postgresContract.checksums as Record)["012_v02_durable_dispatch_kafka_outbox"], "48c552aae6d9a70eb8f1c0638aeda6ac24dcb6efa4d9a9c56e471dbc17367dc5"); + assert.equal((postgresContract.kafkaOutbox as JsonRecord).pendingKeyOrderIndexMigrationId, "015_v02_kafka_outbox_pending_key_order"); + assert.equal((postgresContract.kafkaOutbox as JsonRecord).relayParallelismBound, "AGENTRUN_KAFKA_OUTBOX_BATCH_SIZE"); assert.ok(Array.isArray(postgresContract.requiredTables)); assert.ok(postgresContract.requiredTables.includes("agentrun_schema_migrations")); assert.ok(postgresContract.requiredTables.includes("agentrun_runs")); @@ -41,7 +47,7 @@ const selfTest: SelfTestCase = async () => { assert.ok(postgresContract.requiredTables.includes("agentrun_cancel_requests")); assert.ok(postgresContract.requiredTables.includes("agentrun_runner_dispatch_intents")); assert.ok(postgresContract.requiredTables.includes("agentrun_kafka_event_outbox")); - return { name: "redaction-postgres", tests: ["redaction", "postgres-store-contract"] }; + return { name: "redaction-postgres", tests: ["redaction", "postgres-store-contract", "historical-migration-checksum"] }; }; export default selfTest; diff --git a/src/selftest/cases/35-kafka-durable-outbox.ts b/src/selftest/cases/35-kafka-durable-outbox.ts index a9a759b..a117e0e 100644 --- a/src/selftest/cases/35-kafka-durable-outbox.ts +++ b/src/selftest/cases/35-kafka-durable-outbox.ts @@ -134,6 +134,7 @@ const selfTest: SelfTestCase = async () => { assert.equal(stdioMetadata.runId, run.id); assert.equal(stdioMetadata.commandId, command.id); await assertRelayParallelismAndFailureIsolation(); + await assertRelayParallelismBoundedByBatchSize(); assertWarmRunCommandTraceAuthority(); const claimedIntent = store.claimRunnerDispatchIntents({ owner: "terminalizer", leaseMs: 60_000, limit: 1 })[0]; @@ -173,7 +174,7 @@ const selfTest: SelfTestCase = async () => { await assertStaleClaimsCannotOverwrite(); await assertProducerRecreatedAfterDisconnected(); - return { name: "kafka-durable-outbox", tests: ["explicit-enable", "canonical-event", "warm-run-command-trace-authority", "run-level-trace-fallback", "stdio-command-trace-context", "kafka-tail-correlation-summary", "durable-key-batch", "cross-key-parallelism", "failed-key-isolation", "atomic-dispatch-terminal", "terminal-command-idempotent-replay", "terminal-command-replay-no-new-events-or-outbox", "terminal-command-same-key-payload-conflict", "terminal-command-different-key-fails-closed", "stale-claim-fencing", "producer-reconnect"] }; + return { name: "kafka-durable-outbox", tests: ["explicit-enable", "canonical-event", "warm-run-command-trace-authority", "run-level-trace-fallback", "stdio-command-trace-context", "kafka-tail-correlation-summary", "durable-key-batch", "cross-key-parallelism", "batch-bounded-parallelism", "failed-key-isolation", "atomic-dispatch-terminal", "terminal-command-idempotent-replay", "terminal-command-replay-no-new-events-or-outbox", "terminal-command-same-key-payload-conflict", "terminal-command-different-key-fails-closed", "stale-claim-fencing", "producer-reconnect"] }; }; function assertWarmRunCommandTraceAuthority(): void { @@ -363,6 +364,30 @@ async function assertRelayParallelismAndFailureIsolation(): Promise { assert.equal(store.kafkaEventOutboxStatus().backlogCount, 3); } +async function assertRelayParallelismBoundedByBatchSize(): Promise { + const store = new MemoryAgentRunStore({ eventOutbox: { enabled: true, topic: config.agentrunEventTopic, source: config.clientId } }); + for (let index = 0; index < 20; index++) { + const run = store.createRun(runInput(`ses_kafka_parallel_${index}`)); + store.appendEvent(run.id, "backend_status", { phase: `parallel-${index}` }); + } + let active = 0; + let maxActive = 0; + const result = await relayKafkaEventOutboxOnce({ + store, + options: { ...relayOptions, batchSize: 8 }, + publish: async () => { + active++; + maxActive = Math.max(maxActive, active); + await delay(10); + active--; + }, + }); + assert.equal(result.claimedCount, 8); + assert.equal(result.partitionKeyCount, 8); + assert.equal(maxActive, 8); + assert.equal(store.kafkaEventOutboxStatus().backlogCount, 12); +} + function delay(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } diff --git a/src/selftest/integration/postgres-kafka-order.ts b/src/selftest/integration/postgres-kafka-order.ts index 318ed7d..634bda3 100644 --- a/src/selftest/integration/postgres-kafka-order.ts +++ b/src/selftest/integration/postgres-kafka-order.ts @@ -3,7 +3,7 @@ import { Pool, type PoolClient } from "pg"; import type { CreateRunInput, JsonRecord, KafkaEventOutboxRecord } from "../../common/types.js"; import { validateCreateCommand } from "../../common/validation.js"; import { relayKafkaEventOutboxOnce, type KafkaOutboxRelayOptions } from "../../mgr/kafka-outbox-relay.js"; -import { createPostgresAgentRunStore } from "../../mgr/postgres-store.js"; +import { createPostgresAgentRunStore, postgresMigrationContract } from "../../mgr/postgres-store.js"; import { runCreatedEventPayload } from "../../mgr/store.js"; const connectionString = process.env.AGENTRUN_SELFTEST_DATABASE_URL?.trim(); @@ -13,7 +13,7 @@ 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({ +let store = await createPostgresAgentRunStore({ connectionString, poolMax: 6, eventOutbox: { enabled: true, topic, source: "agentrun-pg-order-test" }, @@ -60,6 +60,7 @@ try { await store.completeKafkaEventOutbox(second); await assertPostgresBatchRelay(store, controlPool); + store = await assertForwardMigrationFromExistingV02Ledger(store, controlPool); const userRun = await store.createRun(runInput("ses_pg_user_message_order")); const userCommandInput = validateCreateCommand({ @@ -239,6 +240,25 @@ async function assertPostgresBatchRelay(store: Awaited>, pool: Pool): Promise>> { + const contract = postgresMigrationContract(); + const checksums = contract.checksums as Record; + const legacyIds = ["012_v02_durable_dispatch_kafka_outbox", "013_v02_runner_dispatch_outcome", "014_v02_queue_attempt_history_retry"]; + const legacyLedger = await pool.query<{ id: string; checksum: string }>("SELECT id, checksum FROM agentrun_schema_migrations WHERE id = ANY($1::text[]) ORDER BY id", [legacyIds]); + assert.deepEqual(legacyLedger.rows.map((row) => row.id), legacyIds); + assert.deepEqual(legacyLedger.rows.map((row) => row.checksum), legacyIds.map((id) => checksums[id])); + await store.close(); + await pool.query("DROP INDEX IF EXISTS agentrun_kafka_event_outbox_pending_key_order_idx"); + await pool.query("DELETE FROM agentrun_schema_migrations WHERE id = $1", ["015_v02_kafka_outbox_pending_key_order"]); + const reopened = await createPostgresAgentRunStore({ connectionString, poolMax: 6, eventOutbox: { enabled: true, topic, source: "agentrun-pg-order-test" } }); + const applied = await pool.query<{ checksum: string }>("SELECT checksum FROM agentrun_schema_migrations WHERE id = $1", ["015_v02_kafka_outbox_pending_key_order"]); + assert.equal(applied.rows[0]?.checksum, checksums["015_v02_kafka_outbox_pending_key_order"]); + const index = await pool.query<{ present: boolean }>("SELECT to_regclass('agentrun_kafka_event_outbox_pending_key_order_idx') IS NOT NULL AS present"); + assert.equal(index.rows[0]?.present, true); + assert.equal((await reopened.health()).migrationId, "015_v02_kafka_outbox_pending_key_order"); + return reopened; +} + async function installGateTrigger(pool: Pool, advisoryGate: string): Promise { await removeGateTrigger(pool); await pool.query(`