diff --git a/docs/reference/slack-app-setup.md b/docs/reference/slack-app-setup.md index 9ad0cc0f9..607fbdad0 100644 --- a/docs/reference/slack-app-setup.md +++ b/docs/reference/slack-app-setup.md @@ -349,8 +349,7 @@ decisions, and posts its own replies. - **Changing forwarding entries is not a drain or cancellation:** relays already retained keep their original host and remain visible in `rig slack status`, even after the last entry is removed. An enabled connector retries them; disabling it pauses retry without deleting them. - Let relays finish before removing a forward entry. A zero dead-letter backlog alone does not - prove they have drained: it does not count relay HTTP requests still in flight. + Let relays finish before removing a forward entry; `rig slack status` shows local in-flight attempts separately from retained records, but zero does not prove global quiescence. - **Who answered:** every post starts `From on `, using the machine's host id (set it with `host.name`). - **The hub's own channels** are handled on the hub, as before. A channel can be relayed to one diff --git a/packages/cli/src/commands/slack.ts b/packages/cli/src/commands/slack.ts index d02dba294..4721350a2 100644 --- a/packages/cli/src/commands/slack.ts +++ b/packages/cli/src/commands/slack.ts @@ -252,7 +252,7 @@ export function slackCommand(deps: SlackDeps = {}): Command { const connector = observation.connector as { configurationDigest?: string; inboundReady?: boolean; outboundReady?: boolean; inboundRouting?: { fallback: string; destination: string }; settlement?: { pending: number; lastFailure: string | null }; - deadLetterBacklog?: number | null; deadLetterBacklogState?: string; deadLetterBacklogReason?: string; + deadLetterBacklog?: number | null; deadLetterBacklogState?: string; deadLetterBacklogReason?: string; relayAttemptsInFlight?: number; inbound?: InboundDeliveryStatus & { state?: string; generation?: number; lastEventAt?: string }; recovery?: { state?: string; reason?: string; lastScanAt?: string; acceptedThisProcess?: number; deadLetteredThisProcess?: number; coverage?: { coverageStart: string; coveredThrough: string; pending?: { upper: string; nextLatest: string }; nextRetryAt?: number } | null; @@ -291,6 +291,7 @@ export function slackCommand(deps: SlackDeps = {}): Command { + `${reasons(threads.backoff) ? `; backing off: ${reasons(threads.backoff)}` : ""}${reasons(threads.incomplete) ? `; incomplete: ${reasons(threads.incomplete)}` : ""}`); } log(` Recovery counts since connector start: accepted ${recovery?.acceptedThisProcess ?? "unknown"}; dead-lettered ${recovery?.deadLetteredThisProcess ?? "unknown"} (custody, not delivery)`); + if (connector.relayAttemptsInFlight !== undefined) log(` Relay attempts awaiting response: ${connector.relayAttemptsInFlight} (this process, point-in-time; zero does not prove global quiescence)`); if (connector.deadLetterBacklogState !== undefined) { const scope = "inbound messages, reactions and click answers"; if (connector.deadLetterBacklogState === "unknown") { diff --git a/packages/cli/test/slack-cutover-cli.test.ts b/packages/cli/test/slack-cutover-cli.test.ts index a9fb8e004..71f28e927 100644 --- a/packages/cli/test/slack-cutover-cli.test.ts +++ b/packages/cli/test/slack-cutover-cli.test.ts @@ -138,13 +138,15 @@ describe("S10 CLI cutover — config surfaces survive on the daemon-homed module { name: "enabled bot-only relay receiver", outboundReady: true, inboundReady: false, paused: false }, ])("shows configured routing and retained retry state on a $name", async ({ outboundReady, inboundReady, paused }) => { const { deps, logs } = makeDeps({ clientFactory: () => ({ get: async () => ({ status: 200, - data: { state: "active", connector: { outboundReady, inboundReady, inboundRouting: { fallback: "local", destination: "ops@hub" }, inbound: { state: "not-configured" }, deadLetterBacklog: 6, deadLetterBacklogState: "ok" } } as T }) }) }); + data: { state: "active", connector: { outboundReady, inboundReady, inboundRouting: { fallback: "local", destination: "ops@hub" }, inbound: { state: "not-configured" }, deadLetterBacklog: 6, deadLetterBacklogState: "ok", relayAttemptsInFlight: 2 } } as T }) }) }); await run(slackCommand(deps), ["status"]); expect(logs.join("\n")).toContain("Configured routing fallback: local to ops@hub (after forwarding, thread and channel routes)"); expect(logs.join("\n")).toContain(paused ? "6 retained record(s); retry paused" : "6 retained record(s) awaiting retry"); + expect(logs.join("\n")).toContain("Relay attempts awaiting response: 2 (this process, point-in-time; zero does not prove global quiescence)"); logs.length = 0; await run(slackCommand(deps), ["status", "--json"]); expect(JSON.parse(logs[0]!).observation.connector.inboundRouting).toEqual({ fallback: "local", destination: "ops@hub" }); + expect(JSON.parse(logs[0]!).observation.connector.relayAttemptsInFlight).toBe(2); }); }); diff --git a/packages/daemon/src/domain/gateway/slack/inbound.ts b/packages/daemon/src/domain/gateway/slack/inbound.ts index 4404fd99a..055113169 100644 --- a/packages/daemon/src/domain/gateway/slack/inbound.ts +++ b/packages/daemon/src/domain/gateway/slack/inbound.ts @@ -232,6 +232,7 @@ export interface InboundDeps { export class InboundRouter { private readonly inflight = new Set(); // same-channel message identity double-dispatch guard + private relayAttemptsInFlight = 0; private readonly replyingTo = new Set(); // click replies being posted now private retryPass: Promise<{ retried: number; landed: number }> | undefined; private readonly botApps = new Map(); // bot_id → its app, from bots.info @@ -247,6 +248,15 @@ export class InboundRouter { this.botLimiter = new BotLimiter(deps.now); } + /** Point-in-time local HTTP attempts, independent of durable retained records. */ + getRelayAttemptsInFlight(): number { return this.relayAttemptsInFlight; } + + private async sendForward(host: string, env: SocketEnvelope): Promise>> { + this.relayAttemptsInFlight++; + try { return await this.deps.forward!.send(host, env); } + finally { this.relayAttemptsInFlight--; } + } + /** The row's summary names the person who actually wrote (their registered address); its tags stay as they were. */ private summaryOf(ev: SlackEvent, sender: string, transfer?: InboundFileResult | null, correlationQitemId?: string, envelope?: string): { summary: string; body: string; attachmentsChars?: number } { const text = String(ev.text ?? ""); @@ -909,7 +919,7 @@ export class InboundRouter { /** Relay an envelope to the machine that owns its channel. The hub has already acknowledged it to * Slack, so a failed relay is kept with its host and retried, never dropped. */ async forwardEnvelope(host: string, env: SocketEnvelope): Promise<{ status: InboundDisposition; reason?: string }> { - const r = await this.deps.forward!.send(host, env); + const r = await this.sendForward(host, env); const channel = envelopeChannel(env) ?? "-"; if (r.ok) { this.deps.log?.(`forwarded ${env.type} channel=${channel} -> ${host}`); @@ -936,7 +946,7 @@ export class InboundRouter { for (const e of entries) { if (failedHosts.has(e.ev.host)) { stillFailing.push(e); continue; } sent++; - const r = await this.deps.forward.send(e.ev.host, e.ev.envelope); + const r = await this.sendForward(e.ev.host, e.ev.envelope); if (r.ok) landed++; else { failedHosts.add(e.ev.host); stillFailing.push({ ev: e.ev, at: e.at, attempts: e.attempts + 1 }); } } diff --git a/packages/daemon/src/domain/gateway/slack/slack-subsystem.ts b/packages/daemon/src/domain/gateway/slack/slack-subsystem.ts index 2d819ba67..bb3777dc7 100644 --- a/packages/daemon/src/domain/gateway/slack/slack-subsystem.ts +++ b/packages/daemon/src/domain/gateway/slack/slack-subsystem.ts @@ -1213,6 +1213,7 @@ export function buildSlackGatewayWire(opts: SlackWireOpts): GatewayWire { // Durable inbound custody across event, click and relay files, even when inert. // Count records, not distinct messages; unreadable storage is null/unknown. ...deadLetterBacklogFields(dead, deadActions, deadForwards), + relayAttemptsInFlight: router?.getRelayAttemptsInFlight() ?? 0, // A machine with no connection of its own can take events a hub relays here; whether a hub does is // the hub's configuration, so this says only that it can, and how many have arrived. ...(router && !inboundReady ? { relayedInbound: { accepting: true, received: relayedReceived } } : {}), diff --git a/packages/daemon/test/slack-shared-foundation-wire.test.ts b/packages/daemon/test/slack-shared-foundation-wire.test.ts index f6f3d5c08..03fad7e43 100644 --- a/packages/daemon/test/slack-shared-foundation-wire.test.ts +++ b/packages/daemon/test/slack-shared-foundation-wire.test.ts @@ -155,7 +155,7 @@ describe("one shared Slack app across two machines", () => { }); it("reports the local fallback and still routes an unmapped channel to that destination", async () => { - expect(hub.wire!.status()).toMatchObject({ inboundRouting: { fallback: "local", destination: "orch-advisor@v-build" } }); + expect(hub.wire!.status()).toMatchObject({ inboundRouting: { fallback: "local", destination: "orch-advisor@v-build" }, relayAttemptsInFlight: 0 }); await say("C0UNMAPPED", "735.1"); await vi.waitFor(() => expect(inbound(hub)).toHaveLength(1)); expect(inbound(hub)[0]!.destinationSession).toBe("orch-advisor@v-build"); diff --git a/packages/daemon/test/slack-shared-foundation.test.ts b/packages/daemon/test/slack-shared-foundation.test.ts index 663a23497..29c37cbe1 100644 --- a/packages/daemon/test/slack-shared-foundation.test.ts +++ b/packages/daemon/test/slack-shared-foundation.test.ts @@ -2,7 +2,7 @@ // rig's representative or a configured seat; a person's top-level message becomes the root its answer // lands under; the hub relays a channel another machine owns, keeps a failed relay and retries it; // the receiving route takes a relayed envelope; and every post names the machine it comes from. -import { describe, it, expect } from "vitest"; +import { describe, it, expect, vi } from "vitest"; import { Hono } from "hono"; import { createDb } from "../src/db/connection.js"; import { migrate } from "../src/db/migrate.js"; @@ -165,6 +165,31 @@ describe("the hub relays a channel another machine owns", () => { } const message = (channel: string): SocketEnvelope => ({ envelope_id: "e1", type: "events_api", payload: { event: { type: "message", user: "U1", text: "hi", ts: "700.1", channel } } }); + it("counts only pending live and retry relay attempts, including failed sends", async () => { + let finish!: (result: { ok: false; error: string } | { ok: true }) => void; + const h = hub(() => new Promise((resolve) => { finish = resolve; })); + const live = handleEnvelope(message("C0VPS04"), () => {}, h.r); + await vi.waitFor(() => expect(h.r.getRelayAttemptsInFlight()).toBe(1)); + expect(h.forwardDeadLetter.readAll()).toHaveLength(0); + finish({ ok: false, error: "synthetic failure" }); + await live; + expect(h.r.getRelayAttemptsInFlight()).toBe(0); + expect(h.forwardDeadLetter.readAll()).toHaveLength(1); + const retry = h.r.retryDeadLetters(); + await vi.waitFor(() => expect(h.r.getRelayAttemptsInFlight()).toBe(1)); + expect(h.forwardDeadLetter.readAll()).toHaveLength(1); + finish({ ok: true }); + await retry; + expect(h.r.getRelayAttemptsInFlight()).toBe(0); + expect(h.forwardDeadLetter.readAll()).toHaveLength(0); + }); + + it("clears the attempt count when the transport throws", async () => { + const h = hub(async () => { throw new Error("synthetic error"); }); + await expect(h.r.forwardEnvelope("vps04", message("C0VPS04"))).rejects.toThrow("synthetic error"); + expect(h.r.getRelayAttemptsInFlight()).toBe(0); + }); + it("sends a message, a reaction and a click for that channel there, untouched, and handles nothing here", async () => { const sent: Array<[string, SocketEnvelope]> = []; const h = hub(async (host, env) => { sent.push([host, env]); return { ok: true }; });