fix: 消除 Kafka outbox 同 key 队头阻塞
This commit is contained in:
AgentRun Codex
@@ -13,44 +13,60 @@ export interface KafkaOutboxRelayOptions {
|
||||
config: AgentRunKafkaConfig;
|
||||
}
|
||||
|
||||
let lastRelayCycle: JsonRecord | null = null;
|
||||
|
||||
export function kafkaOutboxRelayStatus(): JsonRecord {
|
||||
return { lastCycle: lastRelayCycle, valuesPrinted: false };
|
||||
}
|
||||
|
||||
export async function relayKafkaEventOutboxOnce(input: { store: AgentRunStore; options: KafkaOutboxRelayOptions; publish?: (config: AgentRunKafkaConfig, message: AgentRunKafkaMessage) => Promise<void> }): Promise<JsonRecord> {
|
||||
if (!input.options.enabled) return { action: "kafka-event-outbox-relay", enabled: false, claimedCount: 0, deliveredCount: 0, retryCount: 0, staleCount: 0, valuesPrinted: false };
|
||||
const claimed = await input.store.claimKafkaEventOutbox({ owner: input.options.owner, leaseMs: input.options.leaseMs, limit: input.options.batchSize });
|
||||
const items = claimed.slice().sort((a, b) => a.outboxSeq - b.outboxSeq);
|
||||
const blockedKeys = new Set<string>();
|
||||
const startedAt = Date.now();
|
||||
const groups = groupByPartitionKey(items);
|
||||
let deliveredCount = 0;
|
||||
let retryCount = 0;
|
||||
let staleCount = 0;
|
||||
let firstError: JsonRecord | null = null;
|
||||
|
||||
for (const item of items) {
|
||||
if (blockedKeys.has(item.partitionKey)) {
|
||||
if (await retry(input.store, input.options, item, { code: "partition-key-blocked", message: "an earlier outbox record for this partition key failed", valuesPrinted: false })) retryCount++;
|
||||
else staleCount++;
|
||||
continue;
|
||||
await Promise.all(groups.map(async (group) => {
|
||||
let blocked = false;
|
||||
for (const item of group) {
|
||||
if (blocked) {
|
||||
if (await retry(input.store, input.options, item, { code: "partition-key-blocked", message: "an earlier outbox record for this partition key failed", valuesPrinted: false })) retryCount++;
|
||||
else staleCount++;
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
await (input.publish ?? publishAgentRunKafkaMessage)(input.options.config, {
|
||||
topic: item.topic,
|
||||
key: item.partitionKey,
|
||||
value: item.value,
|
||||
headers: stringHeaders(item.headers),
|
||||
});
|
||||
await input.store.completeKafkaEventOutbox(item);
|
||||
deliveredCount++;
|
||||
} catch (error) {
|
||||
blocked = true;
|
||||
const typedError = {
|
||||
code: "kafka-produce-failed",
|
||||
message: redactText(error instanceof Error ? error.message : String(error)).slice(0, 500),
|
||||
topic: item.topic,
|
||||
partitionKey: item.partitionKey,
|
||||
outboxSeq: item.outboxSeq,
|
||||
valuesPrinted: false,
|
||||
};
|
||||
firstError ??= typedError;
|
||||
const retried = await retry(input.store, input.options, item, typedError);
|
||||
if (retried) retryCount++;
|
||||
else staleCount++;
|
||||
}
|
||||
}
|
||||
try {
|
||||
await (input.publish ?? publishAgentRunKafkaMessage)(input.options.config, {
|
||||
topic: item.topic,
|
||||
key: item.partitionKey,
|
||||
value: item.value,
|
||||
headers: stringHeaders(item.headers),
|
||||
});
|
||||
await input.store.completeKafkaEventOutbox(item);
|
||||
deliveredCount++;
|
||||
} catch (error) {
|
||||
blockedKeys.add(item.partitionKey);
|
||||
const retried = await retry(input.store, input.options, item, {
|
||||
code: "kafka-produce-failed",
|
||||
message: redactText(error instanceof Error ? error.message : String(error)).slice(0, 500),
|
||||
topic: item.topic,
|
||||
outboxSeq: item.outboxSeq,
|
||||
valuesPrinted: false,
|
||||
});
|
||||
if (retried) retryCount++;
|
||||
else staleCount++;
|
||||
}
|
||||
}
|
||||
return { action: "kafka-event-outbox-relay", enabled: true, claimedCount: items.length, deliveredCount, retryCount, staleCount, firstOutboxSeq: items[0]?.outboxSeq ?? null, lastOutboxSeq: items.at(-1)?.outboxSeq ?? null, valuesPrinted: false };
|
||||
}));
|
||||
const durationMs = Math.max(1, Date.now() - startedAt);
|
||||
lastRelayCycle = { at: new Date().toISOString(), claimedCount: items.length, partitionKeyCount: groups.length, deliveredCount, retryCount, staleCount, durationMs, claimRatePerSecond: Number((items.length * 1_000 / durationMs).toFixed(2)), publishRatePerSecond: Number((deliveredCount * 1_000 / durationMs).toFixed(2)), firstError, firstOutboxSeq: items[0]?.outboxSeq ?? null, lastOutboxSeq: items.at(-1)?.outboxSeq ?? null, valuesPrinted: false };
|
||||
return { action: "kafka-event-outbox-relay", enabled: true, ...lastRelayCycle };
|
||||
}
|
||||
|
||||
export function startKafkaEventOutboxRelay(input: { store: AgentRunStore; options: KafkaOutboxRelayOptions; onError?: (error: unknown) => void }): () => Promise<void> {
|
||||
@@ -96,3 +112,14 @@ async function retry(store: AgentRunStore, options: KafkaOutboxRelayOptions, ite
|
||||
function stringHeaders(headers: JsonRecord): Record<string, string> {
|
||||
return Object.fromEntries(Object.entries(headers).map(([key, value]) => [key, typeof value === "string" ? value : JSON.stringify(value)]));
|
||||
}
|
||||
|
||||
function groupByPartitionKey(items: KafkaEventOutboxRecord[]): KafkaEventOutboxRecord[][] {
|
||||
const groups = new Map<string, KafkaEventOutboxRecord[]>();
|
||||
for (const item of items) {
|
||||
const key = `${item.topic}\u0000${item.partitionKey}`;
|
||||
const group = groups.get(key) ?? [];
|
||||
group.push(item);
|
||||
groups.set(key, group);
|
||||
}
|
||||
return Array.from(groups.values());
|
||||
}
|
||||
|
||||
@@ -49,6 +49,10 @@ 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);
|
||||
|
||||
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;
|
||||
`;
|
||||
|
||||
export const runnerDispatchOutcomeMigrationSql = `
|
||||
|
||||
@@ -816,8 +816,9 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
async claimKafkaEventOutbox(input: { owner: string; leaseMs: number; limit: number }): Promise<KafkaEventOutboxRecord[]> {
|
||||
const leaseExpiresAt = new Date(Date.now() + input.leaseMs).toISOString();
|
||||
const result = await this.pool.query(
|
||||
`WITH due AS (
|
||||
SELECT candidate.id FROM agentrun_kafka_event_outbox candidate
|
||||
`WITH heads AS MATERIALIZED (
|
||||
SELECT candidate.id, candidate.topic, candidate.partition_key
|
||||
FROM agentrun_kafka_event_outbox candidate
|
||||
WHERE candidate.delivered_at IS NULL
|
||||
AND candidate.next_attempt_at <= now()
|
||||
AND (candidate.lease_expires_at IS NULL OR candidate.lease_expires_at <= now())
|
||||
@@ -831,6 +832,29 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
ORDER BY candidate.outbox_seq ASC
|
||||
FOR UPDATE OF candidate SKIP LOCKED
|
||||
LIMIT $1
|
||||
), ranked AS MATERIALIZED (
|
||||
SELECT candidate.id,
|
||||
row_number() OVER (PARTITION BY candidate.topic, candidate.partition_key ORDER BY candidate.outbox_seq ASC) AS key_offset
|
||||
FROM agentrun_kafka_event_outbox candidate
|
||||
JOIN heads ON heads.topic = candidate.topic AND heads.partition_key = candidate.partition_key
|
||||
WHERE candidate.delivered_at IS NULL
|
||||
AND candidate.next_attempt_at <= now()
|
||||
AND (candidate.lease_expires_at IS NULL OR candidate.lease_expires_at <= now())
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM agentrun_kafka_event_outbox blocker
|
||||
WHERE blocker.topic = candidate.topic
|
||||
AND blocker.partition_key = candidate.partition_key
|
||||
AND blocker.delivered_at IS NULL
|
||||
AND blocker.outbox_seq < candidate.outbox_seq
|
||||
AND (blocker.next_attempt_at > now() OR blocker.lease_expires_at > now())
|
||||
)
|
||||
), due AS MATERIALIZED (
|
||||
SELECT candidate.id
|
||||
FROM agentrun_kafka_event_outbox candidate
|
||||
JOIN ranked ON ranked.id = candidate.id
|
||||
ORDER BY ranked.key_offset ASC, candidate.outbox_seq ASC
|
||||
FOR UPDATE OF candidate SKIP LOCKED
|
||||
LIMIT $1
|
||||
)
|
||||
UPDATE agentrun_kafka_event_outbox o
|
||||
SET attempt_count = o.attempt_count + 1, lease_owner = $2, lease_expires_at = $3, updated_at = now()
|
||||
@@ -871,10 +895,11 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
count(*) FILTER (WHERE delivered_at IS NULL)::int AS backlog_count,
|
||||
count(*) FILTER (WHERE delivered_at IS NULL AND attempt_count > 0)::int AS retrying_count,
|
||||
min(created_at) FILTER (WHERE delivered_at IS NULL) AS oldest_pending_at,
|
||||
COALESCE(max(attempt_count), 0)::int AS max_attempt_count
|
||||
COALESCE(max(attempt_count), 0)::int AS max_attempt_count,
|
||||
(array_agg(last_error ORDER BY outbox_seq ASC) FILTER (WHERE delivered_at IS NULL AND last_error IS NOT NULL))[1] AS first_error
|
||||
FROM agentrun_kafka_event_outbox`,
|
||||
);
|
||||
return { ...durableQueueStatusFromRow(result.rows[0], "kafka-event-outbox"), enabled: this.eventOutboxConfig.enabled };
|
||||
return { ...durableQueueStatusFromRow(result.rows[0], "kafka-event-outbox"), firstError: result.rows[0]?.first_error ?? null, enabled: this.eventOutboxConfig.enabled };
|
||||
}
|
||||
|
||||
async registerRunner(input: Partial<RunnerRecord>): Promise<RunnerRecord> {
|
||||
|
||||
+3
-3
@@ -16,7 +16,7 @@ import { buildRunResult } from "./result.js";
|
||||
import { runnerJobStatusSummary } from "./runner-job-status.js";
|
||||
import { reconcileRunnerJobsOnce, startRunnerJobReconciler } from "./runner-reconciler.js";
|
||||
import { assertRunnerDispatcherAttemptTiming, dispatchRunnerIntentsOnce, startRunnerDispatcher, type RunnerDispatcherOptions } from "./runner-dispatcher.js";
|
||||
import { relayKafkaEventOutboxOnce, startKafkaEventOutboxRelay, type KafkaOutboxRelayOptions } from "./kafka-outbox-relay.js";
|
||||
import { kafkaOutboxRelayStatus, relayKafkaEventOutboxOnce, startKafkaEventOutboxRelay, type KafkaOutboxRelayOptions } from "./kafka-outbox-relay.js";
|
||||
import { createSessionPvc, deleteSessionPvc, getSessionPvcSummary, refreshSessionPvcSummary, runSessionStorageGc } from "./session-pvc.js";
|
||||
import type { SessionPvcSummary } from "./session-pvc.js";
|
||||
import type { SessionPvcOptions } from "./session-pvc.js";
|
||||
@@ -646,11 +646,11 @@ async function route({ method, url, body, store, sourceCommit, authSummary, diag
|
||||
if (method === "GET" && (path === "/health" || path === "/health/live" || path === "/health/readiness")) {
|
||||
const [database, runnerDispatch, kafkaEventOutbox] = await Promise.all([store.health(), store.runnerDispatchStatus(), store.kafkaEventOutboxStatus()]);
|
||||
const ready = path === "/health/live" ? true : database.ready;
|
||||
return { serviceId: "agentrun-mgr", live: true, ready, database, sourceCommit, auth: authSummary ?? null, runnerWorkReady: staticWorkReadyCapabilitySummary(), durableQueues: { runnerDispatch: { ...runnerDispatch, enabled: runnerDispatcherOptions.enabled }, kafkaEventOutbox }, secretRefs: { databaseUrl: database.adapter === "postgres" ? "redacted" : "not-used", valuesPrinted: false } };
|
||||
return { serviceId: "agentrun-mgr", live: true, ready, database, sourceCommit, auth: authSummary ?? null, runnerWorkReady: staticWorkReadyCapabilitySummary(), durableQueues: { runnerDispatch: { ...runnerDispatch, enabled: runnerDispatcherOptions.enabled }, kafkaEventOutbox: { ...kafkaEventOutbox, relay: kafkaOutboxRelayStatus() } }, secretRefs: { databaseUrl: database.adapter === "postgres" ? "redacted" : "not-used", valuesPrinted: false } };
|
||||
}
|
||||
if (method === "GET" && path === "/api/v1/operations/durable-queues") {
|
||||
const [runnerDispatch, kafkaEventOutbox] = await Promise.all([store.runnerDispatchStatus(), store.kafkaEventOutboxStatus()]);
|
||||
return { runnerDispatch: { ...runnerDispatch, enabled: runnerDispatcherOptions.enabled }, kafkaEventOutbox, relayEnabled: kafkaOutboxRelayOptions.enabled, valuesPrinted: false };
|
||||
return { runnerDispatch: { ...runnerDispatch, enabled: runnerDispatcherOptions.enabled }, kafkaEventOutbox: { ...kafkaEventOutbox, relay: kafkaOutboxRelayStatus() }, relayEnabled: kafkaOutboxRelayOptions.enabled, valuesPrinted: false };
|
||||
}
|
||||
if (method === "GET" && path === "/api/v1/backends") return await listBackendCapabilities(providerProfileDefaults) as JsonValue;
|
||||
if (method === "GET" && path === "/api/v1/tool-credentials") return await listToolCredentials(toolCredentialDefaults) as JsonValue;
|
||||
|
||||
+28
-6
@@ -516,11 +516,27 @@ export class MemoryAgentRunStore implements AgentRunStore {
|
||||
claimKafkaEventOutbox(input: { owner: string; leaseMs: number; limit: number }): KafkaEventOutboxRecord[] {
|
||||
const now = Date.now();
|
||||
const all = Array.from(this.kafkaEventOutbox.values());
|
||||
const claimed = all
|
||||
.filter((item) => !item.deliveredAt && Date.parse(item.nextAttemptAt) <= now && (!item.leaseExpiresAt || Date.parse(item.leaseExpiresAt) <= now))
|
||||
.filter((item) => !all.some((earlier) => !earlier.deliveredAt && earlier.topic === item.topic && earlier.partitionKey === item.partitionKey && earlier.outboxSeq < item.outboxSeq))
|
||||
.sort((a, b) => a.outboxSeq - b.outboxSeq)
|
||||
.slice(0, Math.max(1, Math.min(input.limit, 500)));
|
||||
const limit = Math.max(1, Math.min(input.limit, 500));
|
||||
const claimed: KafkaEventOutboxRecord[] = [];
|
||||
const pendingByKey = new Map<string, KafkaEventOutboxRecord[]>();
|
||||
for (const item of all.filter((candidate) => !candidate.deliveredAt).sort((a, b) => a.outboxSeq - b.outboxSeq)) {
|
||||
const key = `${item.topic}\u0000${item.partitionKey}`;
|
||||
const pending = pendingByKey.get(key) ?? [];
|
||||
pending.push(item);
|
||||
pendingByKey.set(key, pending);
|
||||
}
|
||||
const eligible = Array.from(pendingByKey.values())
|
||||
.filter((items) => isDueKafkaOutboxItem(items[0], now))
|
||||
.flatMap((items) => {
|
||||
const prefix: KafkaEventOutboxRecord[] = [];
|
||||
for (const item of items) {
|
||||
if (!isDueKafkaOutboxItem(item, now)) break;
|
||||
prefix.push(item);
|
||||
}
|
||||
return prefix.map((item, keyOffset) => ({ item, keyOffset }));
|
||||
})
|
||||
.sort((a, b) => a.keyOffset - b.keyOffset || a.item.outboxSeq - b.item.outboxSeq);
|
||||
claimed.push(...eligible.slice(0, limit).map(({ item }) => item));
|
||||
return claimed.map((item) => {
|
||||
const updated: KafkaEventOutboxRecord = { ...item, attemptCount: item.attemptCount + 1, leaseOwner: input.owner, leaseExpiresAt: new Date(now + input.leaseMs).toISOString(), updatedAt: nowIso() };
|
||||
this.kafkaEventOutbox.set(item.id, updated);
|
||||
@@ -546,7 +562,9 @@ export class MemoryAgentRunStore implements AgentRunStore {
|
||||
}
|
||||
|
||||
kafkaEventOutboxStatus(): JsonRecord {
|
||||
return { ...workQueueStatus(Array.from(this.kafkaEventOutbox.values()), "kafka-event-outbox", (item) => Boolean((item as KafkaEventOutboxRecord).deliveredAt)), enabled: this.eventOutboxConfig.enabled };
|
||||
const items = Array.from(this.kafkaEventOutbox.values());
|
||||
const firstError = items.filter((item) => !item.deliveredAt && item.lastError).sort((a, b) => a.outboxSeq - b.outboxSeq)[0]?.lastError ?? null;
|
||||
return { ...workQueueStatus(items, "kafka-event-outbox", (item) => Boolean((item as KafkaEventOutboxRecord).deliveredAt)), firstError, enabled: this.eventOutboxConfig.enabled };
|
||||
}
|
||||
|
||||
getCommand(commandId: string): CommandRecord {
|
||||
@@ -1905,6 +1923,10 @@ function workQueueStatus<T extends JsonRecord>(items: T[], component: string, co
|
||||
};
|
||||
}
|
||||
|
||||
function isDueKafkaOutboxItem(item: KafkaEventOutboxRecord | undefined, now: number): boolean {
|
||||
return Boolean(item && Date.parse(item.nextAttemptAt) <= now && (!item.leaseExpiresAt || Date.parse(item.leaseExpiresAt) <= now));
|
||||
}
|
||||
|
||||
export function isSessionOutputEvent(event: RunEvent): boolean {
|
||||
return event.type === "assistant_progress" || event.type === "assistant_message" || event.type === "command_output" || event.type === "diff" || event.type === "error" || event.type === "terminal_status";
|
||||
}
|
||||
|
||||
@@ -91,7 +91,7 @@ const selfTest: SelfTestCase = async () => {
|
||||
const normalizedDispatch = validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { transientEnv: [{ name: "HWLAB_RUNTIME_LANE", value: "v02", sensitive: false }] } } }).dispatch?.input;
|
||||
assert.deepEqual(normalizedDispatch, { transientEnv: [{ name: "HWLAB_RUNTIME_LANE", value: "v02", sensitive: false }] });
|
||||
|
||||
const firstClaim = store.claimKafkaEventOutbox({ owner: "first", leaseMs: 60_000, limit: 100 });
|
||||
const firstClaim = store.claimKafkaEventOutbox({ owner: "first", leaseMs: 60_000, limit: 1 });
|
||||
assert.equal(firstClaim.length, 1);
|
||||
assert.equal(firstClaim[0]?.outboxSeq, 1);
|
||||
store.retryKafkaEventOutbox(firstClaim[0]!, { code: "broker-unavailable" }, new Date(Date.now() + 20).toISOString());
|
||||
@@ -103,9 +103,10 @@ const selfTest: SelfTestCase = async () => {
|
||||
const firstCycle = await relayKafkaEventOutboxOnce({ store, options: relayOptions, publish });
|
||||
const secondCycle = await relayKafkaEventOutboxOnce({ store, options: relayOptions, publish });
|
||||
const thirdCycle = await relayKafkaEventOutboxOnce({ store, options: relayOptions, publish });
|
||||
assert.equal(firstCycle.deliveredCount, 1);
|
||||
assert.equal(secondCycle.deliveredCount, 1);
|
||||
assert.equal(thirdCycle.deliveredCount, 1);
|
||||
assert.equal(firstCycle.deliveredCount, 3);
|
||||
assert.equal(firstCycle.partitionKeyCount, 1);
|
||||
assert.equal(secondCycle.deliveredCount, 0);
|
||||
assert.equal(thirdCycle.deliveredCount, 0);
|
||||
assert.deepEqual(published.map((item) => item.outboxSeq), [1, 2, 3]);
|
||||
assert.deepEqual(published.map((item) => item.sessionId), ["ses_kafka_durable", "ses_kafka_durable", "ses_kafka_durable"]);
|
||||
assert.equal(published[2]?.schema, "agentrun.event.v1");
|
||||
@@ -132,6 +133,7 @@ const selfTest: SelfTestCase = async () => {
|
||||
assert.equal(stdioMetadata.frameSeq, 7);
|
||||
assert.equal(stdioMetadata.runId, run.id);
|
||||
assert.equal(stdioMetadata.commandId, command.id);
|
||||
await assertRelayParallelismAndFailureIsolation();
|
||||
assertWarmRunCommandTraceAuthority();
|
||||
|
||||
const claimedIntent = store.claimRunnerDispatchIntents({ owner: "terminalizer", leaseMs: 60_000, limit: 1 })[0];
|
||||
@@ -171,7 +173,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-head-of-line", "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", "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 {
|
||||
@@ -321,6 +323,46 @@ function staleClaim(error: unknown): boolean {
|
||||
return error instanceof AgentRunError && error.failureKind === "runner-lease-conflict";
|
||||
}
|
||||
|
||||
async function assertRelayParallelismAndFailureIsolation(): Promise<void> {
|
||||
const store = new MemoryAgentRunStore({ eventOutbox: { enabled: true, topic: config.agentrunEventTopic, source: config.clientId } });
|
||||
const blockedRun = store.createRun(runInput("ses_kafka_blocked"));
|
||||
store.appendEvent(blockedRun.id, "backend_status", { phase: "blocked-1" });
|
||||
store.appendEvent(blockedRun.id, "backend_status", { phase: "blocked-2" });
|
||||
store.appendEvent(blockedRun.id, "backend_status", { phase: "blocked-3" });
|
||||
const healthyRun = store.createRun(runInput("ses_kafka_healthy"));
|
||||
store.appendEvent(healthyRun.id, "backend_status", { phase: "healthy-1" });
|
||||
store.appendEvent(healthyRun.id, "backend_status", { phase: "healthy-2" });
|
||||
store.appendEvent(healthyRun.id, "backend_status", { phase: "healthy-3" });
|
||||
const publishedByKey = new Map<string, number[]>();
|
||||
let active = 0;
|
||||
let maxActive = 0;
|
||||
const result = await relayKafkaEventOutboxOnce({
|
||||
store,
|
||||
options: { ...relayOptions, batchSize: 6 },
|
||||
publish: async (_config, message) => {
|
||||
active++;
|
||||
maxActive = Math.max(maxActive, active);
|
||||
try {
|
||||
await delay(10);
|
||||
const key = String(message.value.sessionId);
|
||||
publishedByKey.set(key, [...(publishedByKey.get(key) ?? []), Number(message.value.outboxSeq)]);
|
||||
if (key === "ses_kafka_blocked") throw new Error("selftest broker rejection");
|
||||
} finally {
|
||||
active--;
|
||||
}
|
||||
},
|
||||
});
|
||||
assert.equal(result.claimedCount, 6);
|
||||
assert.equal(result.partitionKeyCount, 2);
|
||||
assert.equal(result.deliveredCount, 3);
|
||||
assert.equal(result.retryCount, 3);
|
||||
assert.equal((result.firstError as JsonRecord).code, "kafka-produce-failed");
|
||||
assert.equal(maxActive, 2);
|
||||
assert.deepEqual(publishedByKey.get("ses_kafka_blocked"), [1]);
|
||||
assert.deepEqual(publishedByKey.get("ses_kafka_healthy"), [4, 5, 6]);
|
||||
assert.equal(store.kafkaEventOutboxStatus().backlogCount, 3);
|
||||
}
|
||||
|
||||
function delay(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ import assert from "node:assert/strict";
|
||||
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 { runCreatedEventPayload } from "../../mgr/store.js";
|
||||
|
||||
@@ -58,6 +59,8 @@ try {
|
||||
assert.equal(marker(second), "B");
|
||||
await store.completeKafkaEventOutbox(second);
|
||||
|
||||
await assertPostgresBatchRelay(store, controlPool);
|
||||
|
||||
const userRun = await store.createRun(runInput("ses_pg_user_message_order"));
|
||||
const userCommandInput = validateCreateCommand({
|
||||
type: "turn",
|
||||
@@ -182,6 +185,60 @@ function runInput(targetSessionId: string): CreateRunInput {
|
||||
};
|
||||
}
|
||||
|
||||
async function assertPostgresBatchRelay(store: Awaited<ReturnType<typeof createPostgresAgentRunStore>>, pool: Pool): Promise<void> {
|
||||
await pool.query("DELETE FROM agentrun_kafka_event_outbox");
|
||||
await pool.query("ALTER SEQUENCE agentrun_kafka_event_outbox_seq RESTART WITH 1");
|
||||
const blockedRun = await store.createRun(runInput("ses_pg_kafka_blocked"));
|
||||
const healthyRun = await store.createRun(runInput("ses_pg_kafka_healthy"));
|
||||
for (let index = 1; index <= 100; index++) {
|
||||
await store.appendEvent(blockedRun.id, "backend_status", { phase: `blocked-${index}` });
|
||||
await store.appendEvent(healthyRun.id, "backend_status", { phase: `healthy-${index}` });
|
||||
}
|
||||
let active = 0;
|
||||
let maxActive = 0;
|
||||
const publishedByKey = new Map<string, number[]>();
|
||||
const options: KafkaOutboxRelayOptions = {
|
||||
enabled: true,
|
||||
intervalMs: 1_000,
|
||||
batchSize: 200,
|
||||
leaseMs: 60_000,
|
||||
retryBackoffMs: 1,
|
||||
owner: "postgres-batch-relay",
|
||||
config: { brokers: ["127.0.0.1:9092"], clientId: "postgres-batch-relay", agentrunEventTopic: topic, codexStdioTopic: "codex-stdio.raw.v1", enabled: true, valuesPrinted: false },
|
||||
};
|
||||
const result = await relayKafkaEventOutboxOnce({
|
||||
store,
|
||||
options,
|
||||
publish: async (_config, message) => {
|
||||
active++;
|
||||
maxActive = Math.max(maxActive, active);
|
||||
try {
|
||||
await delay(1);
|
||||
const key = String(message.value.sessionId);
|
||||
publishedByKey.set(key, [...(publishedByKey.get(key) ?? []), Number(message.value.outboxSeq)]);
|
||||
if (key === "ses_pg_kafka_blocked") throw new Error("postgres selftest broker rejection");
|
||||
} finally {
|
||||
active--;
|
||||
}
|
||||
},
|
||||
});
|
||||
assert.equal(result.claimedCount, 200);
|
||||
assert.equal(result.partitionKeyCount, 2);
|
||||
assert.equal(result.deliveredCount, 100);
|
||||
assert.equal(result.retryCount, 100);
|
||||
assert.equal(maxActive, 2);
|
||||
assert.deepEqual(publishedByKey.get("ses_pg_kafka_blocked"), [1]);
|
||||
assert.deepEqual(publishedByKey.get("ses_pg_kafka_healthy"), Array.from({ length: 100 }, (_, index) => (index + 1) * 2));
|
||||
const status = await store.kafkaEventOutboxStatus();
|
||||
assert.equal(status.backlogCount, 100);
|
||||
assert.equal((status.firstError as JsonRecord).code, "kafka-produce-failed");
|
||||
await delay(5);
|
||||
const retryResult = await relayKafkaEventOutboxOnce({ store, options, publish: async () => undefined });
|
||||
assert.equal(retryResult.claimedCount, 100);
|
||||
assert.equal(retryResult.deliveredCount, 100);
|
||||
assert.equal((await store.kafkaEventOutboxStatus()).backlogCount, 0);
|
||||
}
|
||||
|
||||
async function installGateTrigger(pool: Pool, advisoryGate: string): Promise<void> {
|
||||
await removeGateTrigger(pool);
|
||||
await pool.query(`
|
||||
|
||||
Reference in New Issue
Block a user