From ef1c83274d7f9382a9a068e588a4e9fed1c732ab Mon Sep 17 00:00:00 2001 From: Masato Hoshino Date: Mon, 6 Jul 2026 17:22:16 +0900 Subject: [PATCH] improve(health): surface dead-lettered delivery queue entries (#99842) * improve(health): surface dead-lettered delivery queue entries Deliveries that exhaust retries are moved to the failed status in the SQLite delivery queue for diagnostics, but no health surface ever read them back: openclaw health stayed all-green while messages sat dead-lettered, visible only in gateway logs. Add a per-queue failed count accessor and report dead-lettered entries in the health snapshot (JSON field plus a warning line), following the existing plugin and context-engine health section pattern. Observer-only: no retry, purge, or behavior change. * test(health): cover snapshot and CLI dead-letter reporting Add getHealthSnapshot coverage against an isolated state dir, an openclaw health text-output test for the warning line, exact oldest-failure timestamp assertions via fake timers, and a debug log when the delivery-queue health read fails. * fix(health): recompute dead-letter counts for cached gateway responses The gateway health handler can serve a cached snapshot for up to a refresh interval, so a delivery dead-lettered after the cache was filled stayed hidden from openclaw health. Recompute the delivery queue summary in mergeCachedHealthRuntimeState alongside the existing context-engine and model-pricing live merges, and cover the cached handler path with a regression test. --- src/commands/health.snapshot.test.ts | 33 +++++++++ src/commands/health.test.ts | 51 ++++++++++++++ src/commands/health.ts | 49 ++++++++++++++ src/commands/health.types.ts | 10 +++ src/gateway/server-methods/health.ts | 12 +++- .../server-methods/server-methods.test.ts | 67 +++++++++++++++++++ src/infra/delivery-queue-sqlite.test.ts | 63 ++++++++++++++++- src/infra/delivery-queue-sqlite.ts | 35 ++++++++++ 8 files changed, 318 insertions(+), 2 deletions(-) diff --git a/src/commands/health.snapshot.test.ts b/src/commands/health.snapshot.test.ts index 9d8d190ba79..0de105a03c9 100644 --- a/src/commands/health.snapshot.test.ts +++ b/src/commands/health.snapshot.test.ts @@ -552,6 +552,39 @@ describe("getHealthSnapshot", () => { ]); }); + it("includes dead-lettered delivery queue entries in the health snapshot", async () => { + testConfig = { session: { store: "/tmp/x" } }; + testStore = {}; + setActivePluginRegistry(createTestRegistry([])); + const tmpStateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-health-dq-")); + const previousStateDir = process.env.OPENCLAW_STATE_DIR; + process.env.OPENCLAW_STATE_DIR = tmpStateDir; + try { + const { moveDeliveryQueueEntryToFailed, upsertDeliveryQueueEntry } = + await import("../infra/delivery-queue-sqlite.js"); + const clean = await getHealthSnapshot({ timeoutMs: 10, probe: false }); + expect(clean.deliveryQueues).toBeUndefined(); + + upsertDeliveryQueueEntry({ + queueName: "outbound", + entry: { id: "dead-1", enqueuedAt: 1_000, retryCount: 5 }, + }); + moveDeliveryQueueEntryToFailed("outbound", "dead-1"); + + const snap = await getHealthSnapshot({ timeoutMs: 10, probe: false }); + expect(snap.deliveryQueues?.failed).toEqual([ + { queueName: "outbound", count: 1, oldestFailedAt: expect.any(Number) }, + ]); + } finally { + if (previousStateDir === undefined) { + delete process.env.OPENCLAW_STATE_DIR; + } else { + process.env.OPENCLAW_STATE_DIR = previousStateDir; + } + fs.rmSync(tmpStateDir, { recursive: true, force: true }); + } + }); + it("skips telegram probe when not configured", async () => { testConfig = { session: { store: "/tmp/x" } }; testStore = { diff --git a/src/commands/health.test.ts b/src/commands/health.test.ts index 32263fdd64f..9a89ac197d8 100644 --- a/src/commands/health.test.ts +++ b/src/commands/health.test.ts @@ -10,6 +10,7 @@ import { formatHealthCheckFailure } from "./health-format.js"; import type { HealthSummary } from "./health.js"; import { formatContextEngineHealthLine, + formatDeliveryQueueHealthLine, formatHealthChannelLines, formatModelPricingHealthLine, healthCommand, @@ -184,6 +185,26 @@ describe("healthCommand", () => { expect(parsed.sessions.count).toBe(1); }); + it("prints the delivery queue warning line when the gateway reports dead-letters", async () => { + const snapshot = createHealthSummary({ + channels: {}, + channelOrder: [], + channelLabels: {}, + }); + snapshot.deliveryQueues = { + failed: [{ queueName: "outbound", count: 2, oldestFailedAt: Date.now() - 7_200_000 }], + }; + callGatewayMock.mockResolvedValueOnce(snapshot); + + await healthCommand({ json: false, timeoutMs: 1000, config: {} }, runtime as never); + + expect(runtime.exit).not.toHaveBeenCalled(); + const output = stripAnsi(runtime.log.mock.calls.map((c) => String(c[0])).join("\n")); + expect(output).toContain( + "Delivery queue: warning (dead-lettered entries — outbound: 2; oldest 2h ago)", + ); + }); + it("prints the rich text summary and verbose gateway details", async () => { const recent = [ { key: "main", updatedAt: Date.now() - 60_000, age: 60_000 }, @@ -486,6 +507,36 @@ describe("formatContextEngineHealthLine", () => { }); }); +describe("formatDeliveryQueueHealthLine", () => { + it("summarizes dead-lettered delivery queue entries with the oldest age", () => { + const summary = createHealthSummary({ + channels: {}, + channelOrder: [], + channelLabels: {}, + }); + summary.deliveryQueues = { + failed: [ + { queueName: "outbound", count: 3, oldestFailedAt: 90_000 }, + { queueName: "session", count: 1 }, + ], + }; + + expect(formatDeliveryQueueHealthLine(summary, 7_290_000)).toBe( + "Delivery queue: warning (dead-lettered entries — outbound: 3, session: 1; oldest 2h ago)", + ); + }); + + it("returns null when no dead-lettered entries are reported", () => { + const summary = createHealthSummary({ + channels: {}, + channelOrder: [], + channelLabels: {}, + }); + + expect(formatDeliveryQueueHealthLine(summary)).toBeNull(); + }); +}); + describe("formatHealthCheckFailure", () => { it("keeps non-rich output stable", () => { const err = new Error("gateway closed (1006 abnormal closure): no close reason"); diff --git a/src/commands/health.ts b/src/commands/health.ts index e32f734b8ba..6888a4eeb71 100644 --- a/src/commands/health.ts +++ b/src/commands/health.ts @@ -36,8 +36,10 @@ import { getGatewayModelPricingHealth } from "../gateway/model-pricing-cache-sta import { isGatewayModelPricingEnabled } from "../gateway/model-pricing-config.js"; import type { ChannelRuntimeSnapshot } from "../gateway/server-channel-runtime.types.js"; import { info } from "../globals.js"; +import { countFailedDeliveryQueueEntries } from "../infra/delivery-queue-sqlite.js"; import { isTruthyEnvValue } from "../infra/env.js"; import { formatErrorMessage } from "../infra/errors.js"; +import { formatDurationHuman } from "../infra/format-time/format-duration.js"; import { resolveHeartbeatSummaryForAgent } from "../infra/heartbeat-summary.js"; import { getActivePluginRegistry } from "../plugins/runtime.js"; import { buildChannelAccountBindings, resolvePreferredAccountId } from "../routing/bindings.js"; @@ -54,6 +56,7 @@ import type { ChannelAccountHealthSummary, ChannelHealthSummary, ContextEngineHealthSummary, + DeliveryQueueHealthSummary, HealthSummary, PluginHealthErrorSummary, PluginHealthSummary, @@ -242,6 +245,46 @@ export function formatContextEngineHealthLine(summary: HealthSummary): string | return `Context engine: warning (${quarantined.length} quarantined; downgraded to legacy: ${engines})`; } +/** Builds dead-lettered delivery queue health; shared with cached gateway responses. */ +export function buildDeliveryQueueHealthSummary(): DeliveryQueueHealthSummary | undefined { + // Dead-lettered deliveries are retained in SQLite for diagnostics but had no + // health surface; a storage read failure must not take health down with it. + try { + const failed = countFailedDeliveryQueueEntries().map((queue) => { + const entry: DeliveryQueueHealthSummary["failed"][number] = { + queueName: queue.queueName, + count: queue.count, + }; + if (queue.oldestFailedAt != null) { + entry.oldestFailedAt = queue.oldestFailedAt; + } + return entry; + }); + return failed.length > 0 ? { failed } : undefined; + } catch (error) { + debugHealth("delivery queue health read failed", error); + return undefined; + } +} + +/** Formats dead-lettered delivery queue entries for text health output. */ +export function formatDeliveryQueueHealthLine( + summary: HealthSummary, + now = Date.now(), +): string | null { + const failed = summary.deliveryQueues?.failed ?? []; + if (failed.length === 0) { + return null; + } + const counts = failed.map((queue) => `${queue.queueName}: ${queue.count}`).join(", "); + const oldest = failed + .map((queue) => queue.oldestFailedAt) + .filter((value): value is number => typeof value === "number"); + const oldestNote = + oldest.length > 0 ? `; oldest ${formatDurationHuman(now - Math.min(...oldest))} ago` : ""; + return `Delivery queue: warning (dead-lettered entries — ${counts}${oldestNote})`; +} + const resolveHeartbeatSummary = (cfg: OpenClawConfig, agentId: string) => resolveHeartbeatSummaryForAgent(cfg, agentId); @@ -658,6 +701,7 @@ export async function getHealthSnapshot(params?: { const pluginHealth = buildPluginHealthSummary(); const contextEngineHealth = buildContextEngineHealthSummary(); + const deliveryQueueHealth = buildDeliveryQueueHealthSummary(); const summary: HealthSummary = { ok: true, ts: Date.now(), @@ -665,6 +709,7 @@ export async function getHealthSnapshot(params?: { ...(params?.eventLoop ? { eventLoop: params.eventLoop } : {}), ...(pluginHealth ? { plugins: pluginHealth } : {}), ...(contextEngineHealth ? { contextEngines: contextEngineHealth } : {}), + ...(deliveryQueueHealth ? { deliveryQueues: deliveryQueueHealth } : {}), modelPricing: getGatewayModelPricingHealth({ enabled: isGatewayModelPricingEnabled(cfg) }), channels, channelOrder, @@ -900,6 +945,10 @@ export async function healthCommand( if (contextEngineLine) { runtime.log(styleHealthChannelLine(contextEngineLine, rich)); } + const deliveryQueueLine = formatDeliveryQueueHealthLine(summary); + if (deliveryQueueLine) { + runtime.log(styleHealthChannelLine(deliveryQueueLine, rich)); + } for (const plugin of displayPlugins) { const channelSummary = summary.channels?.[plugin.id]; if (!channelSummary || channelSummary.linked !== true) { diff --git a/src/commands/health.types.ts b/src/commands/health.types.ts index b685e0d2f4b..65db433e224 100644 --- a/src/commands/health.types.ts +++ b/src/commands/health.types.ts @@ -55,6 +55,15 @@ export type ContextEngineHealthSummary = { quarantined: ContextEngineHealthQuarantineSummary[]; }; +/** Dead-lettered delivery queue entries surfaced in health output. */ +export type DeliveryQueueHealthSummary = { + failed: Array<{ + queueName: string; + count: number; + oldestFailedAt?: number; + }>; +}; + /** Optional model pricing cache health reported by the gateway. */ type ModelPricingHealthSummary = import("../gateway/model-pricing-cache-state.js").GatewayModelPricingHealth; @@ -67,6 +76,7 @@ export type HealthSummary = { eventLoop?: import("../gateway/server/event-loop-health.js").GatewayEventLoopHealth; plugins?: PluginHealthSummary; contextEngines?: ContextEngineHealthSummary; + deliveryQueues?: DeliveryQueueHealthSummary; modelPricing?: ModelPricingHealthSummary; channels: Record; channelOrder: string[]; diff --git a/src/gateway/server-methods/health.ts b/src/gateway/server-methods/health.ts index ae042f08e49..7a31278cd6d 100644 --- a/src/gateway/server-methods/health.ts +++ b/src/gateway/server-methods/health.ts @@ -2,6 +2,7 @@ // detecting stale channel runtime state against live gateway snapshots. import { ErrorCodes, errorShape } from "../../../packages/gateway-protocol/src/index.js"; import type { ChannelAccountSnapshot } from "../../channels/plugins/types.public.js"; +import { buildDeliveryQueueHealthSummary } from "../../commands/health.js"; import type { ChannelHealthSummary, HealthSummary } from "../../commands/health.types.js"; import { getStatusSummary } from "../../commands/status.js"; import { listContextEngineQuarantines } from "../../context-engine/registry.js"; @@ -92,7 +93,15 @@ function mergeCachedHealthRuntimeState(params: { cached: HealthSummary; eventLoop?: HealthSummary["eventLoop"]; }): HealthSummary { - const { contextEngines: _cachedContextEngines, ...cached } = params.cached; + const { + contextEngines: _cachedContextEngines, + deliveryQueues: _cachedDeliveryQueues, + ...cached + } = params.cached; + // Dead-letter counts are cheap SQLite reads; recompute them like context + // engines so a delivery that failed after the cache was filled is not hidden + // for a refresh interval. + const deliveryQueues = buildDeliveryQueueHealthSummary(); const quarantinedContextEngines: NonNullable["quarantined"] = []; for (const entry of listContextEngineQuarantines()) { const summary: NonNullable["quarantined"][number] = { @@ -112,6 +121,7 @@ function mergeCachedHealthRuntimeState(params: { ...(quarantinedContextEngines.length > 0 ? { contextEngines: { quarantined: quarantinedContextEngines } } : {}), + ...(deliveryQueues ? { deliveryQueues } : {}), modelPricing: getGatewayModelPricingHealth({ enabled: params.cached.modelPricing?.state !== "disabled", }), diff --git a/src/gateway/server-methods/server-methods.test.ts b/src/gateway/server-methods/server-methods.test.ts index d545cd12e6f..9dd2ded9ca5 100644 --- a/src/gateway/server-methods/server-methods.test.ts +++ b/src/gateway/server-methods/server-methods.test.ts @@ -4895,6 +4895,73 @@ describe("gateway healthHandlers.health cache freshness", () => { } }); + it("merges live dead-lettered delivery queue counts into cached health responses", async () => { + const tmpStateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-health-cached-dq-")); + const previousStateDir = process.env.OPENCLAW_STATE_DIR; + process.env.OPENCLAW_STATE_DIR = tmpStateDir; + try { + const { moveDeliveryQueueEntryToFailed, upsertDeliveryQueueEntry } = + await import("../../infra/delivery-queue-sqlite.js"); + // The cached snapshot was built before this delivery dead-lettered. + const cached = { + ok: true, + ts: Date.now(), + durationMs: 1, + channels: {}, + channelOrder: [], + channelLabels: {}, + heartbeatSeconds: 0, + defaultAgentId: "main", + agents: [], + sessions: { path: "/tmp/sessions.json", count: 0, recent: [] }, + }; + upsertDeliveryQueueEntry({ + queueName: "outbound", + entry: { id: "dead-1", enqueuedAt: 1_000, retryCount: 5 }, + }); + moveDeliveryQueueEntryToFailed("outbound", "dead-1"); + + const respond = vi.fn(); + const refreshHealthSnapshot = vi.fn().mockResolvedValue(cached); + + await healthHandlers.health({ + req: {} as never, + params: {} as never, + respond: respond as never, + context: { + getHealthCache: () => cached, + refreshHealthSnapshot, + getRuntimeSnapshot: () => ({ channels: {}, channelAccounts: {} }), + logHealth: { error: vi.fn() }, + } as never, + client: { connect: { role: "operator", scopes: ["operator.read"] } } as never, + isWebchatConnect: () => false, + }); + + const payload = mockCallArg(respond, 0, 1) as + | { + deliveryQueues?: { + failed?: Array<{ queueName?: string; count?: number; oldestFailedAt?: number }>; + }; + } + | undefined; + expect(payload?.deliveryQueues?.failed).toHaveLength(1); + expect(payload?.deliveryQueues?.failed?.[0]).toMatchObject({ + queueName: "outbound", + count: 1, + }); + expect(typeof payload?.deliveryQueues?.failed?.[0]?.oldestFailedAt).toBe("number"); + expect(mockCallArg(respond, 0, 3)).toEqual({ cached: true }); + } finally { + if (previousStateDir === undefined) { + delete process.env.OPENCLAW_STATE_DIR; + } else { + process.env.OPENCLAW_STATE_DIR = previousStateDir; + } + fs.rmSync(tmpStateDir, { recursive: true, force: true }); + } + }); + it("refreshes cached health when a runtime account is missing from the cached account summary", async () => { const cached = { ok: true, diff --git a/src/infra/delivery-queue-sqlite.test.ts b/src/infra/delivery-queue-sqlite.test.ts index 63444d3fe4b..831dadbbc94 100644 --- a/src/infra/delivery-queue-sqlite.test.ts +++ b/src/infra/delivery-queue-sqlite.test.ts @@ -1,9 +1,10 @@ // Validates SQLite delivery queue inflate guards against corrupted entry_json. import fs from "node:fs"; import path from "node:path"; -import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { openOpenClawStateDatabase } from "../state/openclaw-state-db.js"; import { + countFailedDeliveryQueueEntries, deleteDeliveryQueueEntry, loadDeliveryQueueEntries, loadDeliveryQueueEntry, @@ -160,3 +161,63 @@ describe("delivery-queue-sqlite corrupt JSON resilience", () => { }); }); }); + +describe("countFailedDeliveryQueueEntries", () => { + let tmpDir: string; + let stateDir: string; + + beforeEach(() => { + tmpDir = fs.mkdtempSync(path.join(resolvePreferredOpenClawTmpDir(), "openclaw-dq-count-")); + stateDir = path.join(tmpDir, "state"); + fs.mkdirSync(stateDir, { recursive: true }); + }); + + afterEach(() => { + fs.rmSync(tmpDir, { recursive: true, force: true }); + }); + + function enqueue(queueName: string, id: string, enqueuedAt: number) { + upsertDeliveryQueueEntry({ + queueName, + entry: { id, enqueuedAt, retryCount: 0 }, + stateDir, + }); + } + + it("returns an empty list when nothing is dead-lettered", () => { + enqueue("outbound", "pending-1", 1_000); + + expect(countFailedDeliveryQueueEntries(stateDir)).toEqual([]); + }); + + it("counts dead-lettered entries per queue with the oldest failure timestamp", () => { + enqueue("outbound", "dead-1", 1_000); + enqueue("outbound", "dead-2", 2_000); + enqueue("outbound", "still-pending", 3_000); + enqueue("session", "dead-3", 4_000); + vi.useFakeTimers(); + try { + vi.setSystemTime(50_000); + moveDeliveryQueueEntryToFailed("outbound", "dead-1", stateDir); + vi.setSystemTime(60_000); + moveDeliveryQueueEntryToFailed("outbound", "dead-2", stateDir); + vi.setSystemTime(70_000); + moveDeliveryQueueEntryToFailed("session", "dead-3", stateDir); + } finally { + vi.useRealTimers(); + } + + const counts = countFailedDeliveryQueueEntries(stateDir); + + expect(counts).toHaveLength(2); + const outbound = counts.find((queue) => queue.queueName === "outbound"); + expect(outbound?.count).toBe(2); + expect(outbound?.oldestFailedAt).toBe(50_000); + const session = counts.find((queue) => queue.queueName === "session"); + expect(session?.count).toBe(1); + expect(session?.oldestFailedAt).toBe(70_000); + expect(loadDeliveryQueueEntries("outbound", stateDir).map((entry) => entry.id)).toEqual([ + "still-pending", + ]); + }); +}); diff --git a/src/infra/delivery-queue-sqlite.ts b/src/infra/delivery-queue-sqlite.ts index c706626cb5c..ee94f753973 100644 --- a/src/infra/delivery-queue-sqlite.ts +++ b/src/infra/delivery-queue-sqlite.ts @@ -242,6 +242,41 @@ export function updateDeliveryQueueEntry( upsertDeliveryQueueEntry({ queueName, entry: update(current), stateDir }); } +/** Dead-lettered entry counts for one queue namespace. */ +export type FailedDeliveryQueueCount = { + queueName: string; + count: number; + oldestFailedAt: number | null; +}; + +/** Count dead-lettered (failed) entries per queue namespace for health reporting. */ +export function countFailedDeliveryQueueEntries(stateDir?: string): FailedDeliveryQueueCount[] { + const database = openStateDatabase(stateDir); + const queueDb = getNodeSqliteKysely(database.db); + const rows = executeSqliteQuerySync( + database.db, + queueDb + .selectFrom("delivery_queue_entries") + .select((eb) => [ + "queue_name", + eb.fn.countAll().as("failed_count"), + eb.fn.min("failed_at").as("oldest_failed_at"), + ]) + .where("status", "=", "failed") + .groupBy("queue_name") + .orderBy("queue_name", "asc"), + ).rows as Array<{ + queue_name: string; + failed_count: number | bigint; + oldest_failed_at: number | bigint | null; + }>; + return rows.map((row) => ({ + queueName: row.queue_name, + count: Number(row.failed_count), + oldestFailedAt: row.oldest_failed_at == null ? null : Number(row.oldest_failed_at), + })); +} + /** Mark a pending delivery queue entry as failed for later diagnostics. */ export function moveDeliveryQueueEntryToFailed( queueName: string,