From e8cd287aad237b076497fc80a39aa470d96f35e3 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Tue, 28 Jul 2026 00:53:32 -0400 Subject: [PATCH] refactor(delivery): centralize durable queue routing metadata (#114928) * refactor(delivery): centralize durable queue routing metadata * fix(delivery): keep queue metadata private to its owner --- src/infra/delivery-queue-sqlite.test.ts | 152 +++++++++++++++++++ src/infra/delivery-queue-sqlite.ts | 8 +- src/infra/outbound/delivery-queue-storage.ts | 17 --- src/infra/session-delivery-queue-storage.ts | 16 -- 4 files changed, 156 insertions(+), 37 deletions(-) diff --git a/src/infra/delivery-queue-sqlite.test.ts b/src/infra/delivery-queue-sqlite.test.ts index 559587671938..a0fde30644dc 100644 --- a/src/infra/delivery-queue-sqlite.test.ts +++ b/src/infra/delivery-queue-sqlite.test.ts @@ -8,6 +8,8 @@ import { promoteDeliveryQueueEntryPlatformSend, } from "./delivery-queue-sqlite-claim.js"; import { + commitStagedDeliveryQueueEntry, + commitStagedDeliveryQueueEntryOnce, completeDeliveryQueueEntry, countFailedDeliveryQueueEntries, deleteDeliveryQueueEntry, @@ -140,6 +142,156 @@ describe("delivery-queue-sqlite corrupt JSON resilience", () => { expect(loaded).toMatchObject({ id: "rt-1", enqueuedAt: 1000, retryCount: 0 }); }); + it.each([ + { + name: "outbound delivery", + queueName: "outbound", + entry: { + id: "metadata-outbound", + channel: "discord", + to: "channel:123", + accountId: "bot-a", + session: { key: "agent:main:discord:channel:123" }, + }, + expected: { + entry_kind: "outbound", + session_key: "agent:main:discord:channel:123", + channel: "discord", + target: "channel:123", + account_id: "bot-a", + }, + }, + { + name: "routed session delivery", + queueName: "session", + entry: { + id: "metadata-session-route", + kind: "agentTurn", + sessionKey: "agent:main:discord:channel:123", + route: { channel: "discord", to: "channel:123", accountId: "bot-a" }, + deliveryContext: { channel: "telegram", to: "999", accountId: "bot-b" }, + }, + expected: { + entry_kind: "agentTurn", + session_key: "agent:main:discord:channel:123", + channel: "discord", + target: "channel:123", + account_id: "bot-a", + }, + }, + { + name: "context-only session delivery", + queueName: "session", + entry: { + id: "metadata-session-context", + kind: "systemEvent", + sessionKey: "agent:main:telegram:direct:123", + deliveryContext: { channel: "telegram", to: "123", accountId: "bot-a" }, + }, + expected: { + entry_kind: "systemEvent", + session_key: "agent:main:telegram:direct:123", + channel: "telegram", + target: "123", + account_id: "bot-a", + }, + }, + ])("indexes canonical $name metadata", ({ queueName, entry, expected }) => { + upsertDeliveryQueueEntry({ + queueName, + entry: { ...entry, enqueuedAt: 1000, retryCount: 0 }, + stateDir, + }); + + const { db } = openOpenClawStateDatabase({ + env: { ...process.env, OPENCLAW_STATE_DIR: stateDir }, + }); + const readMetadata = () => + db + .prepare( + `SELECT entry_kind, session_key, channel, target, account_id + FROM delivery_queue_entries WHERE queue_name = ? AND id = ?`, + ) + .get(queueName, entry.id); + expect(readMetadata()).toEqual(expected); + + updateDeliveryQueueEntry(queueName, entry.id, stateDir, (current) => ({ + ...current, + retryCount: current.retryCount + 1, + })); + expect(readMetadata()).toEqual(expected); + }); + + it("preserves explicit queue metadata ownership", () => { + upsertDeliveryQueueEntry({ + queueName: "outbound-media-staging", + entry: { id: "metadata-media-stage", enqueuedAt: 1000, retryCount: 0 }, + metadata: { entryKind: "outbound-media-stage" }, + stateDir, + }); + + const { db } = openOpenClawStateDatabase({ + env: { ...process.env, OPENCLAW_STATE_DIR: stateDir }, + }); + expect( + db + .prepare("SELECT entry_kind FROM delivery_queue_entries WHERE queue_name = ? AND id = ?") + .get("outbound-media-staging", "metadata-media-stage"), + ).toEqual({ entry_kind: "outbound-media-stage" }); + }); + + it.each([ + { name: "ordinary", commit: commitStagedDeliveryQueueEntry, expected: true }, + { name: "insert-only", commit: commitStagedDeliveryQueueEntryOnce, expected: "created" }, + ])("indexes $name staged outbound commits", ({ commit, expected }) => { + const stagingQueueName = "outbound-media-staging"; + const stagingId = "metadata-staged-media"; + const outboundEntry = { + id: "metadata-staged-outbound", + enqueuedAt: 1000, + retryCount: 0, + channel: "discord", + to: "channel:123", + accountId: "bot-a", + session: { key: "agent:main:discord:channel:123" }, + }; + upsertDeliveryQueueEntry({ + queueName: stagingQueueName, + entry: { id: stagingId, enqueuedAt: 1000, retryCount: 0 }, + metadata: { entryKind: "outbound-media-stage" }, + stateDir, + }); + + expect( + commit({ + queueName: "outbound", + entry: outboundEntry, + stagingId, + stagingQueueName, + stateDir, + }), + ).toBe(expected); + + const { db } = openOpenClawStateDatabase({ + env: { ...process.env, OPENCLAW_STATE_DIR: stateDir }, + }); + expect( + db + .prepare( + `SELECT entry_kind, session_key, channel, target, account_id + FROM delivery_queue_entries WHERE queue_name = ? AND id = ?`, + ) + .get("outbound", "metadata-staged-outbound"), + ).toEqual({ + entry_kind: "outbound", + session_key: "agent:main:discord:channel:123", + channel: "discord", + target: "channel:123", + account_id: "bot-a", + }); + expect(loadDeliveryQueueEntry(stagingQueueName, stagingId, stateDir)).toBeNull(); + }); + it("update increments retry count", () => { upsertDeliveryQueueEntry({ queueName: QUEUE, diff --git a/src/infra/delivery-queue-sqlite.ts b/src/infra/delivery-queue-sqlite.ts index 78e373dc1ae9..7207fabd0786 100644 --- a/src/infra/delivery-queue-sqlite.ts +++ b/src/infra/delivery-queue-sqlite.ts @@ -25,7 +25,7 @@ export type DeliveryQueueCompletionRetention = }>; /** Indexed metadata extracted from queue payloads for diagnostics and recovery. */ -export type DeliveryQueueRowMetadata = { +type DeliveryQueueRowMetadata = { entryKind?: string; sessionKey?: string; channel?: string; @@ -114,7 +114,7 @@ function inflate(row: QueueRow): DeliveryQueueEntryState | null { }; } -function metadata(entry: DeliveryQueueEntryState): DeliveryQueueRowMetadata { +function metadata(queueName: string, entry: DeliveryQueueEntryState): DeliveryQueueRowMetadata { const item = entry as DeliveryQueueEntryState & { kind?: string; sessionKey?: string; @@ -126,7 +126,7 @@ function metadata(entry: DeliveryQueueEntryState): DeliveryQueueRowMetadata { deliveryContext?: { channel?: string; to?: string; accountId?: string }; }; return { - entryKind: item.kind, + entryKind: item.kind ?? queueName, sessionKey: item.sessionKey ?? item.session?.key, channel: item.channel ?? item.route?.channel ?? item.deliveryContext?.channel, target: item.to ?? item.route?.to ?? item.deliveryContext?.to, @@ -140,7 +140,7 @@ function upsertDeliveryQueueEntryInDatabase( ): boolean { const now = Date.now(); const status = params.status ?? "pending"; - const meta = params.metadata ?? metadata(params.entry); + const meta = params.metadata ?? metadata(params.queueName, params.entry); const queueDb = getNodeSqliteKysely(database.db); const insert = queueDb.insertInto("delivery_queue_entries").values({ queue_name: params.queueName, diff --git a/src/infra/outbound/delivery-queue-storage.ts b/src/infra/outbound/delivery-queue-storage.ts index da6a8140ae53..82b988dba584 100644 --- a/src/infra/outbound/delivery-queue-storage.ts +++ b/src/infra/outbound/delivery-queue-storage.ts @@ -22,7 +22,6 @@ import { reserveDeliveryQueueEntryAttempt, updateDeliveryQueueEntry, upsertDeliveryQueueEntry, - type DeliveryQueueRowMetadata, type DeliveryQueueCompletionRetention, } from "../delivery-queue-sqlite.js"; import { generateSecureUuid } from "../secure-random.js"; @@ -120,16 +119,6 @@ export interface QueuedDelivery extends QueuedDeliveryPayload { recoveryState?: "producer_claimed" | "send_attempt_started" | "unknown_after_send"; } -function queuedDeliveryMetadata(entry: QueuedDelivery): DeliveryQueueRowMetadata { - return { - entryKind: "outbound", - sessionKey: entry.session?.key, - channel: entry.channel, - target: entry.to, - accountId: entry.accountId, - }; -} - function createQueuedDelivery(params: QueuedDeliveryPayload, id: string): QueuedDelivery { return { id, @@ -172,12 +161,10 @@ export async function enqueueDelivery( ): Promise { const id = generateSecureUuid(); const entry = createQueuedDelivery(params, id); - const metadata = queuedDeliveryMetadata(entry); if (mediaStageId) { const committed = commitStagedDeliveryQueueEntry({ queueName: OUTBOUND_DELIVERY_QUEUE_NAME, entry, - metadata, stagingId: mediaStageId, stagingQueueName: DELIVERY_QUEUE_MEDIA_STAGING_QUEUE_NAME, stateDir, @@ -189,7 +176,6 @@ export async function enqueueDelivery( upsertDeliveryQueueEntry({ queueName: OUTBOUND_DELIVERY_QUEUE_NAME, entry, - metadata, stateDir, }); } @@ -208,13 +194,11 @@ export async function enqueueDeliveryOnce( throw new Error("Stable delivery queue id is required"); } const entry = createQueuedDelivery(params, normalizedId); - const metadata = queuedDeliveryMetadata(entry); const created = mediaStageId ? (() => { const result = commitStagedDeliveryQueueEntryOnce({ queueName: OUTBOUND_DELIVERY_QUEUE_NAME, entry, - metadata, stagingId: mediaStageId, stagingQueueName: DELIVERY_QUEUE_MEDIA_STAGING_QUEUE_NAME, stateDir, @@ -227,7 +211,6 @@ export async function enqueueDeliveryOnce( : upsertDeliveryQueueEntry({ queueName: OUTBOUND_DELIVERY_QUEUE_NAME, entry, - metadata, stateDir, insertOnly: true, }); diff --git a/src/infra/session-delivery-queue-storage.ts b/src/infra/session-delivery-queue-storage.ts index 9f1e1500ad58..f533b6d7e6ac 100644 --- a/src/infra/session-delivery-queue-storage.ts +++ b/src/infra/session-delivery-queue-storage.ts @@ -12,7 +12,6 @@ import { updateDeliveryQueueEntry, upsertDeliveryQueueEntry, type DeliveryQueueCompletionRetention, - type DeliveryQueueRowMetadata, } from "./delivery-queue-sqlite.js"; import { generateSecureUuid } from "./secure-random.js"; @@ -113,17 +112,6 @@ function buildEntryId(idempotencyKey?: string): string { return sha256Hex(idempotencyKey); } -function queuedSessionDeliveryMetadata(entry: QueuedSessionDelivery): DeliveryQueueRowMetadata { - const route = entry.kind === "agentTurn" ? entry.route : undefined; - return { - entryKind: entry.kind, - sessionKey: entry.sessionKey, - channel: route?.channel ?? entry.deliveryContext?.channel, - target: route?.to ?? entry.deliveryContext?.to, - accountId: route?.accountId ?? entry.deliveryContext?.accountId, - }; -} - /** Enqueue a session delivery and return its durable id. */ export async function enqueueSessionDelivery( params: QueuedSessionDeliveryPayload, @@ -140,7 +128,6 @@ export async function enqueueSessionDelivery( upsertDeliveryQueueEntry({ queueName: QUEUE_NAME, entry, - metadata: queuedSessionDeliveryMetadata(entry), stateDir, ...(params.completionRetention === "permanent" ? { insertOnly: true } @@ -170,7 +157,6 @@ export async function enqueueClaimedSessionDelivery( const claimed = upsertDeliveryQueueEntry({ queueName: QUEUE_NAME, entry, - metadata: queuedSessionDeliveryMetadata(entry), stateDir, insertOnly: true, }); @@ -241,7 +227,6 @@ export async function markSessionDeliveryAttemptStarted( ...entry, deliveryStartedAt: entry.deliveryStartedAt ?? Date.now(), } as QueuedSessionDelivery, - metadata: queuedSessionDeliveryMetadata(entry), stateDir, updatePendingOnly: true, }); @@ -278,7 +263,6 @@ export async function markSessionDeliverySettlement( settlementOutcome: outcome, ...(outcome === "recovered" ? { acknowledgedAt: entry.acknowledgedAt ?? Date.now() } : {}), } as QueuedSessionDelivery, - metadata: queuedSessionDeliveryMetadata(entry), stateDir, updatePendingOnly: true, });