fix: 前进迁移 Kafka outbox 索引

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;

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])),
   };
 }

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<string, string>)["008_v01_dsflash_go_backend_profile"] === "string" && (postgresContract.checksums as Record<string, string>)["008_v01_dsflash_go_backend_profile"].length > 0);
   assert.ok(typeof (postgresContract.checksums as Record<string, string>)["009_v01_dsflash_go_model_catalog"] === "string" && (postgresContract.checksums as Record<string, string>)["009_v01_dsflash_go_model_catalog"].length > 0);
   assert.ok(typeof (postgresContract.checksums as Record<string, string>)["010_v01_queue_session_ref"] === "string" && (postgresContract.checksums as Record<string, string>)["010_v01_queue_session_ref"].length > 0);
@@ -28,7 +30,11 @@ const selfTest: SelfTestCase = async () => {
   assert.ok(typeof (postgresContract.checksums as Record<string, string>)["012_v02_durable_dispatch_kafka_outbox"] === "string" && (postgresContract.checksums as Record<string, string>)["012_v02_durable_dispatch_kafka_outbox"].length > 0);
   assert.ok(typeof (postgresContract.checksums as Record<string, string>)["013_v02_runner_dispatch_outcome"] === "string" && (postgresContract.checksums as Record<string, string>)["013_v02_runner_dispatch_outcome"].length > 0);
   assert.ok(typeof (postgresContract.checksums as Record<string, string>)["014_v02_queue_attempt_history_retry"] === "string" && (postgresContract.checksums as Record<string, string>)["014_v02_queue_attempt_history_retry"].length > 0);
+  assert.ok(typeof (postgresContract.checksums as Record<string, string>)["015_v02_kafka_outbox_pending_key_order"] === "string" && (postgresContract.checksums as Record<string, string>)["015_v02_kafka_outbox_pending_key_order"].length > 0);
   assert.equal((postgresContract.checksums as Record<string, string>)["002_v01_backend_profiles"], "928b5c490cc4539cb64ecef34784557601b2724fa2870570f16a53576804e49c");
+  assert.equal((postgresContract.checksums as Record<string, string>)["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;

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<void> {
   assert.equal(store.kafkaEventOutboxStatus().backlogCount, 3);
 }
 
+async function assertRelayParallelismBoundedByBatchSize(): Promise<void> {
+  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<void> {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }

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<ReturnType<typeof createP
   assert.equal((await store.kafkaEventOutboxStatus()).backlogCount, 0);
 }
 
+async function assertForwardMigrationFromExistingV02Ledger(store: Awaited<ReturnType<typeof createPostgresAgentRunStore>>, pool: Pool): Promise<Awaited<ReturnType<typeof createPostgresAgentRunStore>>> {
+  const contract = postgresMigrationContract();
+  const checksums = contract.checksums as Record<string, string>;
+  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<void> {
   await removeGateTrigger(pool);
   await pool.query(`