From eced1603597f805b91e229771d528cb3e788e2b3 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Tue, 20 Oct 2026 08:29:24 +0200 Subject: [PATCH 1/5] feat(alerts): a gateway sends chosen fleet events to generic webhook destinations (#719) Config, port, alerts module with signed generic format, wiring in the gateway composition root, docs and tests. --- .dependency-cruiser.mjs | 16 +- docs/CONFIGURATION.md | 68 ++- docs/internal/ARCHITECTURE.md | 4 +- src/alerts/AGENTS.md | 30 ++ src/alerts/CLAUDE.md | 1 + src/alerts/alerts.test.ts | 498 ++++++++++++++++++ src/alerts/formats/conformance.test.ts | 44 ++ src/alerts/formats/generic/index.ts | 22 + src/alerts/formats/index.ts | 7 + src/alerts/formats/types.ts | 12 + src/alerts/index.ts | 47 ++ src/alerts/queue.ts | 170 ++++++ src/cli/index.test.ts | 1 + src/contract/AGENTS.md | 2 +- src/contract/operations.test.ts | 2 + src/contract/protocol.test.ts | 4 +- src/contract/protocol.ts | 2 +- src/contract/schemas.ts | 14 + src/core/cleanup/idle-destroy.test.ts | 1 + src/core/cleanup/idle-shutdown.test.ts | 1 + src/core/config.test.ts | 180 +++++++ src/core/config.ts | 134 ++++- src/core/doctor.test.ts | 1 + src/core/index.ts | 6 + src/core/nuke.test.ts | 1 + src/core/reaper.test.ts | 1 + src/daemon-protocol/index.test.ts | 2 +- src/daemon/AGENTS.md | 10 +- src/daemon/dispatcher.test.ts | 1 + src/daemon/gateway-fleet.e2e.test.ts | 189 ++++++- src/daemon/main.ts | 23 + src/daemon/server.test.ts | 6 +- src/gateway/dispatcher.test.ts | 1 + src/gateway/service.test.ts | 2 +- src/http/test-fakes.ts | 1 + src/leasing/acquisition-planner.test.ts | 1 + src/leasing/create-leasing.test.ts | 1 + .../lease-acquisition-coordinator.test.ts | 1 + src/leasing/lease-end-participants.test.ts | 1 + src/leasing/lease-health-monitor.test.ts | 1 + src/leasing/physical-device-watcher.test.ts | 1 + src/leasing/physical-leasing.test.ts | 1 + src/ports/AGENTS.md | 1 + src/ports/index.ts | 1 + src/ports/testing.ts | 12 + src/ports/webhook-poster.test.ts | 60 +++ src/ports/webhook-poster.ts | 27 + 47 files changed, 1573 insertions(+), 39 deletions(-) create mode 100644 src/alerts/AGENTS.md create mode 100644 src/alerts/CLAUDE.md create mode 100644 src/alerts/alerts.test.ts create mode 100644 src/alerts/formats/conformance.test.ts create mode 100644 src/alerts/formats/generic/index.ts create mode 100644 src/alerts/formats/index.ts create mode 100644 src/alerts/formats/types.ts create mode 100644 src/alerts/index.ts create mode 100644 src/alerts/queue.ts create mode 100644 src/ports/webhook-poster.test.ts create mode 100644 src/ports/webhook-poster.ts diff --git a/.dependency-cruiser.mjs b/.dependency-cruiser.mjs index fbf9b2e6b..01098dcd7 100644 --- a/.dependency-cruiser.mjs +++ b/.dependency-cruiser.mjs @@ -105,8 +105,8 @@ export default { ), direction( "no-fs-or-process", - "core, leasing, agent-device, host-load and stream reach the filesystem and processes only through injected ports", - "^src/(core|leasing|agent-device|host-load|stream)/", + "core, leasing, agent-device, host-load, stream and alerts reach the filesystem and processes only through injected ports", + "^src/(core|leasing|agent-device|host-load|stream|alerts)/", { dependencyTypes: ["core"], path: fsOrProcess }, ), direction( @@ -139,6 +139,18 @@ export default { "^src/(core|leasing|drivers|gateway|contract)/", { path: "^src/host-load/" }, ), + direction( + "alerts-imports-bus-ports-core-only", + "alerts reads the bus, sends through the WebhookPoster port and takes config types from core, and imports no other component", + "^src/alerts/", + { path: "^src/", pathNot: "^src/(alerts|bus|ports|core)/" }, + ), + direction( + "not-alerts", + "only the composition root, main.ts, imports the alerts module", + "^src/(?!alerts/|daemon/main\\.ts$)", + { path: "^src/alerts/" }, + ), direction( "gateway-not-edges", "the gateway imports no driver, HTTP, CLI or MCP module", diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 9a82d48a8..4826e3fcc 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -45,6 +45,7 @@ edit yourself keeps the mode you give it until the next `config set`; run | `capabilities.stream.bitrateKbps` | **Worker side.** Target bitrate in kilobits per second, handed to the encoder as an average rather than a ceiling; a frame that needs more takes it. A whole number from `100` to `40000`; the iOS encoder takes nothing lower. | `1500` | | `exec.timeoutMs` | **Worker side.** How long one `device.exec` command (`simlock simctl` / `simlock adb` against a gateway or over HTTP) may run before the worker kills it and the operation fails with `EXEC_TIMEOUT`. Authoritative: it bounds the process that actually runs. | `10 minutes` | | `gateway.routing` | **Gateway side.** Which routing policy places a queued request on a worker. `warm-then-free` is the only policy in v1. Among the workers that can serve the request, in order: a worker whose health is not `running`, or that has requests of its own waiting, is left out; a worker holding a ready device that fits goes first (the device's mode must fit the request, a request with no mode fits a device that serves the worker's default mode, and its runtime must be the one requested or, with none requested, the one the worker would pick: the newest runtime the model pairs with); otherwise a worker with no free running slot for the platform or overall is left out (a running device that is not leased counts as free: the worker shuts it down to make room), a worker under its RAM budget for the platform is preferred over one at it (a worker at its budget is still asked when it is the only one left), and the one with the most free running capacity for the platform wins. A worker that is left out is busy, not unable: the request waits for it. Ties go to the lowest worker id. | `warm-then-free` | +| `gateway.alerts.destinations` | **Gateway side.** A list of at most 16 webhook destinations the gateway sends chosen fleet events to; see [Alerts](#alerts). A bad entry stops the daemon from starting, on a gateway and on a worker alike; a worker sends nothing whatever it lists. `simlock config` shows the URLs and secrets; the daemon log, the `daemon.started` event and the event history show `[redacted]`. | `[]` (no alerts) | | `gateway.disconnectedRetentionMs` | **Gateway side.** How long a disconnected worker is kept (greyed, never dispatched to) before the gateway forgets it. The clock is held while the gateway still knows of gateway-issued leases on that worker, and that hold ends when the last of those leases passes its deadline. | `24 hours` | | `gateway.execTimeoutMs` | **Gateway side.** How long the gateway waits on a proxied `device.exec` before giving up. A backstop for a worker that never answers at all — deliberately longer than the worker's own `exec.timeoutMs`, which is authoritative because that side owns the process and can kill it, so an ordinary timeout surfaces as the worker's `EXEC_TIMEOUT` rather than racing this one. | `11 minutes` | | `gateway.leaseRequestTimeoutMs` | **Gateway side.** How long the gateway waits on a forwarded `lease.request` before giving up on that worker for this request, answering `WORKER_UNREACHABLE`. Bounds the one uplink call that otherwise had no timeout of its own, so a wedged worker cannot park a request where neither a deadline nor `lease.cancel` could ever reach it again. Generous against a cold device provision-plus-boot; well below `gateway.execTimeoutMs`, since granting a lease should never take as long as a command run against the device afterward. | `5 minutes` | @@ -128,7 +129,7 @@ not apply in `mode: "gateway"`, where both keys are worker-side and are warned about and ignored like every other worker key — a gateway is not misconfigured by leftovers from the config it was flipped out of. `gateway.label` is a non-empty string, `gateway.routing` one of the -registered routing policies, and `exec.timeoutMs`, `gateway.execTimeoutMs`, +registered routing policies, `gateway.alerts.destinations` the list described under [Alerts](#alerts), and `exec.timeoutMs`, `gateway.execTimeoutMs`, `gateway.leaseRequestTimeoutMs`, and `gateway.disconnectedRetentionMs` positive numbers. **`mode: "gateway"` with `http.enabled: false` is rejected at load**, naming the key: a @@ -239,6 +240,7 @@ does not apply to it. It reads: |---|---| | `mode` | to be a gateway at all | | `gateway.routing`, `gateway.disconnectedRetentionMs`, `gateway.execTimeoutMs`, `gateway.leaseRequestTimeoutMs` | how to run the fleet | +| `gateway.alerts.destinations` | where to send fleet events | | `http.*` | it is the fleet's contact point | | `lease.*` | `defaultTtlMs`/`maxTtlMs` bound what its own clients may ask for, before a request is dispatched — see below | | `log.*`, `eventBuffer.*`, `eventLog.*`, `metrics.*` | logging, the event history and the usage figures kept beside it, as anywhere | @@ -312,6 +314,70 @@ because only one of them owns anything: `drivers.android.adbServerPort` values (see below), but a gateway plus a worker is one driver set, so there is nothing to split. +## Alerts + +A gateway can send chosen fleet events to a webhook, so n8n, Zapier or your own +receiver hears about them without anyone watching the console. Set +`gateway.alerts.destinations` to a list of destinations: + +```json +{ + "mode": "gateway", + "gateway": { + "alerts": { + "destinations": [ + { + "name": "audit", + "format": "generic", + "url": "https://n8n.example.com/hook", + "secret": "a-long-random-string", + "events": ["worker.connected", "worker.disconnected"] + } + ] + } + } +} +``` + +| Field | Rule | +|---|---| +| `name` | Lowercase letters, digits and dashes, starting with a letter, at most 32 characters; no two destinations share one. | +| `format` | `generic`, the only format so far. | +| `url` | `https://`, or `http://` to `localhost`, `127.0.0.1` or `[::1]`; no user name or password in it. | +| `events` | A non-empty list, each event once, from: `worker.connected`, `worker.disconnected`, `worker.rejected`, `worker.drain-started`, `worker.drain-ended`, `device.quarantine-abandoned`, `device.quarantine-stranded`, `device.recovery-failed`, `device.stalled-transition-detected`, `disk.pressure-detected`, `component.install-failed`. | +| `secret` | A non-empty string the gateway signs each message with. | + +An event a worker reports reaches the destination too, with `workerId` in its +payload. A key not listed above is warned about and ignored, and +`simlock config set` refuses it. + +**What the receiver gets.** One `POST` per event, with the event as the body: the +same JSON `simlock events --json` prints (`id`, `seq`, `timestamp`, `event`, +`payload`, `module`). Headers: + +| Header | Value | +|---|---| +| `Content-Type` | `application/json` | +| `User-Agent` | `simlock/` | +| `X-Simlock-Event` | the event name | +| `X-Simlock-Delivery` | the event `id`; the same on every try, so a receiver can drop a repeat | +| `X-Simlock-Signature` | `sha256=` and the hex HMAC-SHA256 of the raw body, keyed with the destination's `secret` | + +To check a message, compute the HMAC-SHA256 of the bytes you received with the +secret and compare it to the header value in constant time. Use the raw body, +not a re-serialised copy. + +**Delivery.** Each destination sends one message at a time, in the order the +gateway saw the events, and a slow destination never delays another. A try +ends after 10 seconds. A network error, a timeout, `429` or a `5xx` is tried +again 1 second after the failed try, then 5 seconds after the second: three tries +in all, then a warning in `daemon.log` ("Alert send gave up"). Any other status +that is not `2xx`, a redirect included, gives up at once; a redirect is never +followed. Up to 100 messages wait per destination; past that the oldest waiting +one is dropped with a warning. Nothing is written to disk, so a gateway restart +loses the waiting messages, and stopping the gateway abandons what is still +being sent. The log names a destination and never its URL or secret. + ## Downloads A missing iOS simulator runtime or Android system image is downloaded only diff --git a/docs/internal/ARCHITECTURE.md b/docs/internal/ARCHITECTURE.md index 586feb2c0..3fa61cfb8 100644 --- a/docs/internal/ARCHITECTURE.md +++ b/docs/internal/ARCHITECTURE.md @@ -45,6 +45,7 @@ worker daemon ── uplink (WebSocket to /v1/uplink, worker dials) ──┘ | core/usage | Usage figures from the history rows, read on a thread | [usage](../../src/core/usage/AGENTS.md) | | drivers | The `Driver` port's two implementations | [drivers](../../src/drivers/AGENTS.md) · [android](../../src/drivers/android/AGENTS.md) · [android screen](../../src/drivers/android/screen/AGENTS.md) · [ios](../../src/drivers/ios/AGENTS.md) | | gateway | Fleet queue, routing, forwarding to workers, relay | [gateway](../../src/gateway/AGENTS.md) · [routing](../../src/gateway/routing/AGENTS.md) | +| alerts | Gateway only: chosen fleet events POSTed to webhook destinations | [alerts](../../src/alerts/AGENTS.md) | | link | Byte channels over the uplink | [link](../../src/link/AGENTS.md) | | agent-device | Optional edge: the agent-device daemon beside a worker, its HTTP front | [agent-device](../../src/agent-device/AGENTS.md) | | stream | Optional edge: a leased device's live screen over HTTP, relayed by a gateway | [stream](../../src/stream/AGENTS.md) | @@ -73,13 +74,14 @@ Arrows point from the optional to the required (architecture rule 15). | `past-index` | Outside a component, import only its index file (tests may also import its testing file) | | `core-not-leasing` | Core declares a port; the composition root connects leasing to it | | `core-not-drivers` | Core and leasing reach a platform only through the `Driver` interface | -| `no-fs-or-process` | Core, leasing, agent-device, host-load and stream use ports, never `fs` or `child_process` | +| `no-fs-or-process` | Core, leasing, agent-device, host-load, stream and alerts use ports, never `fs` or `child_process` | | `definition-is-light` | A module's definition file imports its own module only for types or by dynamic import | | `auth-imports-little` | `src/auth` imports only ports and contract | | `modules-imports-little`, `modules-imports-bus-types-only` | `src/modules` imports only ports, contract and core, and the bus for types | | `not-modules` | Core, leasing, drivers, gateway and contract import nothing from modules or module-registry | | `not-agent-device` | Core, leasing and drivers import nothing from agent-device | | `host-load-imports-ports-only`, `not-host-load` | host-load stands alone | +| `alerts-imports-bus-ports-core-only`, `not-alerts` | alerts imports the bus, ports and core only; only `main.ts` imports it | | `not-stream`, `stream-not-edges` | Nothing required imports stream; stream imports no frontend, gateway or driver | | `gateway-not-edges`, `gateway-only-dispatch` | Gateway imports no driver or frontend, and from daemon only `src/daemon/dispatch.ts` | | `only-main-knows-gateway` | In daemon, only `src/daemon/main.ts` imports the gateway | diff --git a/src/alerts/AGENTS.md b/src/alerts/AGENTS.md new file mode 100644 index 000000000..9ee7c9c5b --- /dev/null +++ b/src/alerts/AGENTS.md @@ -0,0 +1,30 @@ +# Alerts + +Does: a gateway sends chosen fleet events to webhook destinations (ADR 0035). +Entry: `index.ts` (`createAlerts`). Built only by `../daemon/main.ts`, in the gateway branch, when +`gateway.alerts.destinations` is not empty; a worker never builds it. +Uses: bus (subscribe, types), ports (`Clock`, `Logger`, `WebhookPoster`), core (config types and +`ALERT_EVENTS`). Never imports anything else; `.dependency-cruiser.mjs` checks both directions. +Imported by: `../daemon/main.ts` only. + +## Files + +| File | Role | +| ------------------ | ------------------------------------------------------------------------ | +| `index.ts` | `createAlerts`: one subscription per listed event, `stop()` | +| `queue.ts` | `DestinationQueue`: one message at a time, tries, timeout, bounded wait | +| `formats/types.ts` | `AlertFormatDefinition`: `build` turns an envelope into headers and body | +| `formats/index.ts` | `ALERT_FORMAT_REGISTRY`, typed over core's `ALERT_FORMATS` | +| `formats/generic/` | The envelope as JSON, signed with HMAC-SHA256 | + +## How it works + +1. The bus handler puts the envelope on the queue of every destination that lists the event and returns. +2. A queue takes the oldest waiting message, builds its request once, then POSTs it: 10 s at most + per try, 3 tries, 1 s then 5 s between them. Network errors, timeouts, `429` and `5xx` are retried; + any other non-2xx status gives up at once. A queue holds 100 waiting messages and drops the oldest. +3. `stop()` aborts the try in flight, cancels timers and empties the queues; it waits for nothing. + +Logs name the destination, never its URL or secret; a failed try is logged by code, never by an +error's message. A format is added by a directory under `formats/`, an entry in the registry and in +core's `ALERT_FORMATS`; the queue never names a format. diff --git a/src/alerts/CLAUDE.md b/src/alerts/CLAUDE.md new file mode 100644 index 000000000..43c994c2d --- /dev/null +++ b/src/alerts/CLAUDE.md @@ -0,0 +1 @@ +@AGENTS.md diff --git a/src/alerts/alerts.test.ts b/src/alerts/alerts.test.ts new file mode 100644 index 000000000..2e1591814 --- /dev/null +++ b/src/alerts/alerts.test.ts @@ -0,0 +1,498 @@ +import { createHmac } from "node:crypto"; +import { createServer } from "node:net"; + +import { afterEach, describe, expect, it, vi } from "vitest"; + +import { EventBus, type EventMap, type EventName } from "../bus/index.js"; +import { type AlertDestination } from "../core/index.js"; +import { FakeClock, FetchWebhookPoster, JsonLinesLogger, MemoryLogSink } from "../ports/index.js"; +import { FakeWebhookPoster } from "../ports/testing.js"; +import { createAlerts, type Alerts } from "./index.js"; +import { ALERT_FORMAT_REGISTRY } from "./formats/index.js"; + +const URL_TOKEN = "hook-token-xyz"; + +function destination(overrides: Partial = {}): AlertDestination { + return { + name: "audit", + format: "generic", + url: `https://n8n.example.com/${URL_TOKEN}`, + events: ["worker.connected"], + secret: "s3cr3t", + ...overrides, + }; +} + +function settle(): Promise { + return new Promise((resolve) => setImmediate(resolve)); +} + +const connected = (workerId = "wrk_1") => ({ workerId }); + +class Harness { + readonly clock = new FakeClock(1_000); + readonly eventBus = new EventBus(this.clock); + readonly poster = new FakeWebhookPoster(); + readonly sink = new MemoryLogSink(); + alerts: Alerts; + + constructor(destinations: AlertDestination[] = [destination()]) { + this.alerts = createAlerts({ + destinations, + eventBus: this.eventBus, + poster: this.poster, + clock: this.clock, + logger: new JsonLinesLogger({ clock: this.clock, level: "debug", sink: this.sink }).child( + "alerts", + ), + version: "9.9.9", + }); + } + + emit(event: Event, payload: EventMap[Event]) { + return this.eventBus.emit(event, payload, "test"); + } + + logs(message: string) { + return this.sink.records.filter((record) => record.message === message); + } + + async advance(milliseconds: number): Promise { + this.clock.advance(milliseconds); + await settle(); + } + + failWith(kind: "refused" | "no-code" | 429 | 503): void { + this.poster.respond = () => { + if (kind === "refused") { + return Promise.reject( + Object.assign(new TypeError("fetch failed"), { cause: { code: "ECONNREFUSED" } }), + ); + } + if (kind === "no-code") return Promise.reject(new TypeError("fetch failed")); + return Promise.resolve({ status: kind }); + }; + } +} + +function deferred() { + let resolve!: (value: { status: number }) => void; + const promise = new Promise<{ status: number }>((done) => { + resolve = done; + }); + return { promise, resolve }; +} + +const harnesses: Harness[] = []; +function harness(destinations?: AlertDestination[]): Harness { + const created = new Harness(destinations); + harnesses.push(created); + return created; +} +afterEach(() => { + for (const each of harnesses.splice(0)) each.alerts.stop(); + vi.restoreAllMocks(); +}); + +describe("alerts", () => { + it.each([ + ["a gateway-own worker.connected", "own"], + ["a republished device.quarantine-abandoned", "relayed"], + ])( + "posts a listed event to a generic destination as the envelope with Content-Type, User-Agent, X-Simlock-Event, X-Simlock-Delivery and a signature that verifies with the secret, and logs Alert delivered at debug (%s)", + async (_name, kind) => { + const h = harness([ + destination({ events: ["worker.connected", "device.quarantine-abandoned"] }), + ]); + const envelope = + kind === "own" + ? h.emit("worker.connected", connected()) + : h.eventBus.republish({ + id: "evt_remote", + timestamp: 42, + event: "device.quarantine-abandoned", + payload: { deviceId: "dev_1", attempts: 3, workerId: "wrk_9" } as never, + module: "quarantine", + }); + await settle(); + + expect(h.poster.requests).toHaveLength(1); + const [request] = h.poster.requests; + expect(request?.url).toBe(destination().url); + expect(request?.body).toBe(JSON.stringify(envelope)); + expect(JSON.parse(request?.body ?? "")).toMatchObject({ + id: envelope.id, + event: envelope.event, + payload: envelope.payload, + }); + if (kind === "relayed") + expect(JSON.parse(request?.body ?? "").payload.workerId).toBe("wrk_9"); + expect(request?.headers).toEqual({ + "Content-Type": "application/json", + "User-Agent": "simlock/9.9.9", + "X-Simlock-Event": envelope.event, + "X-Simlock-Delivery": envelope.id, + "X-Simlock-Signature": `sha256=${createHmac("sha256", "s3cr3t") + .update(request?.body ?? "") + .digest("hex")}`, + }); + expect(h.logs("Alert delivered")).toHaveLength(1); + expect(h.logs("Alert delivered")[0]?.level).toBe("debug"); + }, + ); + + it("posts nothing for an event the destination did not list", async () => { + const h = harness(); + h.emit("worker.connected", connected()); + await settle(); + expect(h.poster.requests).toHaveLength(1); + + h.emit("worker.disconnected", { workerId: "wrk_1", leaseCount: 0 }); + h.emit("lease.granted", { + leaseId: "lse_1", + deviceId: "dev_1", + requester: "r", + requestId: "req_1", + source: "warm", + } as never); + await h.advance(60_000); + + expect(h.poster.requests).toHaveLength(1); + }); + + it("posts to every destination that lists the event, each signed with its own secret", async () => { + const h = harness([ + destination({ name: "one", url: "https://one.example.com/x", secret: "first" }), + destination({ name: "two", url: "https://two.example.com/x", secret: "second" }), + ]); + h.emit("worker.connected", connected()); + await settle(); + + const byUrl = new Map(h.poster.requests.map((request) => [request.url, request])); + expect([...byUrl.keys()].sort()).toEqual([ + "https://one.example.com/x", + "https://two.example.com/x", + ]); + for (const [url, secret] of [ + ["https://one.example.com/x", "first"], + ["https://two.example.com/x", "second"], + ] as const) { + const request = byUrl.get(url); + expect(request?.headers["X-Simlock-Signature"]).toBe( + `sha256=${createHmac("sha256", secret) + .update(request?.body ?? "") + .digest("hex")}`, + ); + } + }); + + describe("a failed try", () => { + const cases = [ + { name: "a network error", kind: "refused", endMs: 0, field: { error: "ECONNREFUSED" } }, + { + name: "an error without a code", + kind: "no-code", + endMs: 0, + field: { error: "network-error" }, + }, + { name: "429", kind: 429, endMs: 0, field: { status: 429 } }, + { name: "5xx", kind: 503, endMs: 0, field: { status: 503 } }, + { name: "a timeout", kind: "timeout", endMs: 10_000, field: { error: "timeout" } }, + ] as const; + + it.each(cases)( + "tries again after $name, 1 s then 5 s after the failed try, and gives up with a warn line after the third", + async ({ kind, endMs, field }) => { + const h = harness(); + if (kind === "timeout") h.poster.respond = () => new Promise(() => {}); + else h.failWith(kind); + const envelope = h.emit("worker.connected", connected()); + await settle(); + expect(h.poster.requests).toHaveLength(1); + + await h.advance(endMs + 999); + expect(h.poster.requests).toHaveLength(1); + await h.advance(1); + expect(h.poster.requests).toHaveLength(2); + + await h.advance(endMs + 4_999); + expect(h.poster.requests).toHaveLength(2); + expect(h.logs("Alert send gave up")).toHaveLength(0); + await h.advance(1); + expect(h.poster.requests).toHaveLength(3); + + await h.advance(endMs); + expect(h.logs("Alert send gave up")).toHaveLength(1); + const [line] = h.logs("Alert send gave up"); + expect(line?.level).toBe("warn"); + expect(line?.fields).toEqual({ + destination: "audit", + event: "worker.connected", + eventId: envelope.id, + attempt: 3, + ...field, + }); + await h.advance(60_000); + expect(h.poster.requests).toHaveLength(3); + }, + ); + }); + + it.each([301, 400, 404])( + "gives up at once on a status that is not 2xx, 429 or 5xx, and posts it once (%i)", + async (status) => { + const h = harness(); + h.poster.respond = () => Promise.resolve({ status }); + h.emit("worker.connected", connected()); + await settle(); + + expect(h.logs("Alert send gave up").map((line) => line.fields)).toEqual([ + expect.objectContaining({ attempt: 1, status }), + ]); + await h.advance(60_000); + expect(h.poster.requests).toHaveLength(1); + }, + ); + + it("aborts a try with no answer at 10 s, not before, and moves on without waiting for it to settle", async () => { + const h = harness(); + const first = deferred(); + h.poster.respond = (_request, index) => (index === 0 ? first.promise : new Promise(() => {})); + h.emit("worker.connected", connected()); + await settle(); + const signal = h.poster.requests[0]?.signal; + + await h.advance(9_999); + expect(signal?.aborted).toBe(false); + await h.advance(1); + expect(signal?.aborted).toBe(true); + + await h.advance(1_000); + expect(h.poster.requests).toHaveLength(2); + first.resolve({ status: 200 }); + await settle(); + expect(h.logs("Alert delivered")).toHaveLength(0); + expect(h.poster.requests).toHaveLength(2); + }); + + it("sends one message at a time per destination, in the order the bus emitted them, the second only after the first's last try", async () => { + const h = harness(); + h.failWith(503); + const a = h.emit("worker.connected", connected("wrk_a")); + const b = h.emit("worker.connected", connected("wrk_b")); + const c = h.emit("worker.connected", connected("wrk_c")); + await settle(); + await h.advance(1_000); + await h.advance(4_999); + const sent = () => h.poster.requests.map((request) => request.headers["X-Simlock-Delivery"]); + expect(sent()).toEqual([a.id, a.id]); + + await h.advance(1); + expect(sent()).toEqual([a.id, a.id, a.id, b.id]); + + h.poster.respond = () => Promise.resolve({ status: 200 }); + await h.advance(1_000); + expect(sent()).toEqual([a.id, a.id, a.id, b.id, b.id, c.id]); + }); + + it("a destination that does not answer does not delay another", async () => { + const h = harness([ + destination({ name: "stuck", url: "https://stuck.example.com/x" }), + destination({ name: "quick", url: "https://quick.example.com/x" }), + ]); + h.poster.respond = (request) => + request.url.includes("stuck") ? new Promise(() => {}) : Promise.resolve({ status: 200 }); + h.emit("worker.connected", connected("wrk_a")); + h.emit("worker.connected", connected("wrk_b")); + await settle(); + + const quick = h.poster.requests.filter((request) => request.url.includes("quick")); + expect(quick).toHaveLength(2); + expect(h.poster.requests.filter((request) => request.url.includes("stuck"))).toHaveLength(1); + }); + + it("keeps at most 100 waiting per destination and drops the oldest waiting one, never the one in flight, with a warn line", async () => { + const h = harness(); + const inFlight = deferred(); + h.poster.respond = (_request, index) => + index === 0 ? inFlight.promise : Promise.resolve({ status: 200 }); + const envelopes = [h.emit("worker.connected", connected("wrk_0"))]; + await settle(); + for (let index = 1; index <= 100; index += 1) { + envelopes.push(h.emit("worker.connected", connected(`wrk_${index}`))); + } + expect(h.logs("Alert dropped: queue full")).toHaveLength(0); + envelopes.push(h.emit("worker.connected", connected("wrk_101"))); + + const dropped = h.logs("Alert dropped: queue full"); + expect(dropped).toHaveLength(1); + expect(dropped[0]?.level).toBe("warn"); + expect(dropped[0]?.fields).toMatchObject({ destination: "audit", eventId: envelopes[1]?.id }); + + inFlight.resolve({ status: 200 }); + await settle(); + const sent = h.poster.requests.map((request) => request.headers["X-Simlock-Delivery"]); + expect(sent).toEqual( + envelopes.filter((_envelope, index) => index !== 1).map((each) => each.id), + ); + }); + + it("stop aborts the try in flight, posts nothing more, ignores later events and late settles, and logs the discarded count per destination as its last line", async () => { + const h = harness(); + const inFlight = deferred(); + h.poster.respond = () => inFlight.promise; + h.emit("worker.connected", connected("wrk_a")); + await settle(); + h.emit("worker.connected", connected("wrk_b")); + h.emit("worker.connected", connected("wrk_c")); + + h.alerts.stop(); + expect(h.poster.requests[0]?.signal.aborted).toBe(true); + const lines = h.sink.records.length; + expect(h.sink.records.at(-1)).toMatchObject({ + level: "info", + message: "Alerts discarded at stop", + fields: { discarded: { audit: 3 } }, + }); + + h.emit("worker.connected", connected("wrk_d")); + inFlight.resolve({ status: 200 }); + await h.advance(60_000); + expect(h.poster.requests).toHaveLength(1); + expect(h.sink.records).toHaveLength(lines); + }); + + it("stop counts a message waiting between tries as discarded and cancels its retry", async () => { + const h = harness(); + h.failWith(503); + h.emit("worker.connected", connected()); + await settle(); + + h.alerts.stop(); + await h.advance(60_000); + + expect(h.logs("Alerts discarded at stop")[0]?.fields).toEqual({ discarded: { audit: 1 } }); + expect(h.poster.requests).toHaveLength(1); + }); + + it("the bus handler returns before the post settles, and a poster that throws reaches neither the emitter nor other subscribers", async () => { + const h = harness(); + const seen: string[] = []; + h.eventBus.subscribe("worker.connected", (envelope) => seen.push(envelope.id)); + h.poster.post = () => { + throw new Error(`boom ${URL_TOKEN}`); + }; + + const envelope = h.emit("worker.connected", connected()); + + expect(h.poster.requests).toHaveLength(0); + expect(seen).toEqual([envelope.id]); + await settle(); + await h.advance(1_000); + await h.advance(5_000); + expect(h.logs("Alert send gave up")[0]?.fields).toMatchObject({ + attempt: 3, + error: "network-error", + }); + }); + + it("a request that throws while being built ends as gave up with build-failed and the next message is sent", async () => { + const h = harness(); + const original = ALERT_FORMAT_REGISTRY.generic.build; + vi.spyOn(ALERT_FORMAT_REGISTRY.generic, "build").mockImplementationOnce(() => { + throw new Error(`cannot build ${URL_TOKEN}`); + }); + vi.spyOn(ALERT_FORMAT_REGISTRY.generic, "build").mockImplementation(original); + const first = h.emit("worker.connected", connected("wrk_a")); + const second = h.emit("worker.connected", connected("wrk_b")); + await settle(); + + expect(h.logs("Alert send gave up")[0]?.fields).toMatchObject({ + eventId: first.id, + error: "build-failed", + }); + expect(h.poster.requests.map((request) => request.headers["X-Simlock-Delivery"])).toEqual([ + second.id, + ]); + }); + + it("a message is built once when it leaves the queue, and later tries resend the same headers and body", async () => { + const h = harness(); + let calls = 0; + const build = vi.spyOn(ALERT_FORMAT_REGISTRY.generic, "build").mockImplementation(() => { + calls += 1; + return { headers: { "X-Call": String(calls) }, body: `call-${calls}` }; + }); + h.failWith(503); + h.emit("worker.connected", connected()); + await settle(); + await h.advance(1_000); + await h.advance(5_000); + + expect(h.poster.requests).toHaveLength(3); + for (const request of h.poster.requests) { + expect(request.headers).toEqual({ "X-Call": "1" }); + expect(request.body).toBe("call-1"); + } + expect(build).toHaveBeenCalledTimes(1); + }); + + it("no log line holds a destination's url or secret", async () => { + const h = harness([destination({ events: ["worker.connected", "worker.disconnected"] })]); + h.poster.respond = () => + Promise.reject( + Object.assign(new TypeError(`fetch failed: ${destination().url} s3cr3t`), { + cause: { code: "ECONNREFUSED" }, + }), + ); + h.emit("worker.connected", connected()); + await settle(); + await h.advance(1_000); + await h.advance(5_000); + for (let index = 0; index < 102; index += 1) { + h.emit("worker.disconnected", { workerId: `wrk_${index}`, leaseCount: 0 }); + } + h.alerts.stop(); + + const text = h.sink.records.map((record) => JSON.stringify(record)).join("\n"); + expect(h.logs("Alert send gave up")).toHaveLength(1); + expect(h.logs("Alert dropped: queue full")).toHaveLength(1); + expect(h.logs("Alerts discarded at stop")).toHaveLength(1); + expect(text).not.toContain(URL_TOKEN); + expect(text).not.toContain("n8n.example.com"); + expect(text).not.toContain("s3cr3t"); + }); + + it("a refused connection through FetchWebhookPoster logs ECONNREFUSED and never the URL", async () => { + const closedPort = await new Promise((resolve) => { + const server = createServer(); + server.listen(0, "127.0.0.1", () => { + const { port } = server.address() as { port: number }; + server.close(() => resolve(port)); + }); + }); + const h = harness([]); + const alerts = createAlerts({ + destinations: [destination({ url: `http://127.0.0.1:${closedPort}/${URL_TOKEN}` })], + eventBus: h.eventBus, + poster: new FetchWebhookPoster(), + clock: h.clock, + logger: new JsonLinesLogger({ clock: h.clock, level: "debug", sink: h.sink }), + version: "9.9.9", + }); + h.emit("worker.connected", connected()); + + await vi.waitFor( + () => { + h.clock.advance(10_000); + expect(h.logs("Alert send gave up")).toHaveLength(1); + }, + { timeout: 5_000, interval: 50 }, + ); + alerts.stop(); + + expect(h.logs("Alert send gave up")[0]?.fields).toMatchObject({ error: "ECONNREFUSED" }); + expect(JSON.stringify(h.sink.records)).not.toContain(URL_TOKEN); + expect(JSON.stringify(h.sink.records)).not.toContain(String(closedPort)); + }); +}); diff --git a/src/alerts/formats/conformance.test.ts b/src/alerts/formats/conformance.test.ts new file mode 100644 index 000000000..821f8499c --- /dev/null +++ b/src/alerts/formats/conformance.test.ts @@ -0,0 +1,44 @@ +import { describe, expect, it } from "vitest"; + +import type { EventEnvelope } from "../../bus/index.js"; +import { ALERT_EVENTS, ALERT_FORMATS } from "../../core/index.js"; +import { ALERT_FORMAT_REGISTRY } from "./index.js"; + +describe.each(Object.entries(ALERT_FORMAT_REGISTRY))("format %s", (name, format) => { + it("every registered format passes the conformance suite", () => { + expect(name).toBe(format.name); + for (const event of ALERT_EVENTS) { + const envelope = { + id: "evt_1", + seq: 1, + timestamp: 5, + event, + payload: { workerId: "wrk_1" }, + module: "test", + } as unknown as EventEnvelope; + + const built = format.build({ + envelope, + destination: { + name: "d", + format: format.name, + url: "https://x.example", + events: [event], + secret: "s", + }, + version: "1.2.3", + workerLabel: (id) => id, + }); + + expect(() => JSON.parse(built.body)).not.toThrow(); + expect(built.headers["Content-Type"]).toBe("application/json"); + expect(built.headers["User-Agent"]).toBe("simlock/1.2.3"); + } + }); +}); + +describe("ALERT_FORMAT_REGISTRY", () => { + it("has exactly the formats core names", () => { + expect(Object.keys(ALERT_FORMAT_REGISTRY).sort()).toEqual([...ALERT_FORMATS].sort()); + }); +}); diff --git a/src/alerts/formats/generic/index.ts b/src/alerts/formats/generic/index.ts new file mode 100644 index 000000000..5e6a758bd --- /dev/null +++ b/src/alerts/formats/generic/index.ts @@ -0,0 +1,22 @@ +import { createHmac } from "node:crypto"; + +import type { AlertFormatDefinition } from "../types.js"; + +export const genericFormat: AlertFormatDefinition = { + name: "generic", + build({ envelope, destination, version }) { + if (destination.secret === undefined) throw new Error("a generic destination has no secret"); + const body = JSON.stringify(envelope); + const signature = createHmac("sha256", destination.secret).update(body).digest("hex"); + return { + body, + headers: { + "Content-Type": "application/json", + "User-Agent": `simlock/${version}`, + "X-Simlock-Event": envelope.event, + "X-Simlock-Delivery": envelope.id, + "X-Simlock-Signature": `sha256=${signature}`, + }, + }; + }, +}; diff --git a/src/alerts/formats/index.ts b/src/alerts/formats/index.ts new file mode 100644 index 000000000..d38911a6e --- /dev/null +++ b/src/alerts/formats/index.ts @@ -0,0 +1,7 @@ +import type { AlertFormatName } from "../../core/index.js"; +import { genericFormat } from "./generic/index.js"; +import type { AlertFormatDefinition } from "./types.js"; + +export const ALERT_FORMAT_REGISTRY: { readonly [N in AlertFormatName]: AlertFormatDefinition } = { + generic: genericFormat, +}; diff --git a/src/alerts/formats/types.ts b/src/alerts/formats/types.ts new file mode 100644 index 000000000..a6a000e98 --- /dev/null +++ b/src/alerts/formats/types.ts @@ -0,0 +1,12 @@ +import type { EventEnvelope } from "../../bus/index.js"; +import type { AlertDestination, AlertFormatName } from "../../core/index.js"; + +export interface AlertFormatDefinition { + readonly name: AlertFormatName; + build(input: { + envelope: EventEnvelope; + destination: AlertDestination; + version: string; + workerLabel(id: string): string; + }): { headers: Record; body: string }; +} diff --git a/src/alerts/index.ts b/src/alerts/index.ts new file mode 100644 index 000000000..c0c5af0f1 --- /dev/null +++ b/src/alerts/index.ts @@ -0,0 +1,47 @@ +import type { EventBus } from "../bus/index.js"; +import type { AlertDestination } from "../core/index.js"; +import { ALERT_EVENTS } from "../core/index.js"; +import type { Clock, Logger, WebhookPoster } from "../ports/index.js"; +import { DestinationQueue } from "./queue.js"; + +export interface Alerts { + stop(): void; +} + +export function createAlerts(options: { + destinations: readonly AlertDestination[]; + eventBus: EventBus; + poster: WebhookPoster; + clock: Clock; + logger: Logger; + version: string; +}): Alerts { + const { destinations, eventBus, poster, clock, logger, version } = options; + const queues = destinations.map((destination) => ({ + destination, + queue: new DestinationQueue({ destination, poster, clock, logger, version }), + })); + + const unsubscribes = ALERT_EVENTS.filter((name) => + destinations.some((destination) => destination.events.includes(name)), + ).map((name) => + eventBus.subscribe(name, (envelope) => { + for (const { destination, queue } of queues) { + if (!destination.events.includes(name)) continue; + queue.enqueue(envelope); + } + }), + ); + + return { + stop() { + for (const unsubscribe of unsubscribes) unsubscribe(); + const discarded: Record = {}; + for (const { destination, queue } of queues) { + const count = queue.stop(); + if (count > 0) discarded[destination.name] = count; + } + if (Object.keys(discarded).length > 0) logger.info("Alerts discarded at stop", { discarded }); + }, + }; +} diff --git a/src/alerts/queue.ts b/src/alerts/queue.ts new file mode 100644 index 000000000..0669d04bb --- /dev/null +++ b/src/alerts/queue.ts @@ -0,0 +1,170 @@ +import type { EventEnvelope } from "../bus/index.js"; +import type { AlertDestination } from "../core/index.js"; +import type { Clock, Logger, TimerHandle, WebhookPoster } from "../ports/index.js"; +import { ALERT_FORMAT_REGISTRY } from "./formats/index.js"; + +const TRY_TIMEOUT_MS = 10_000; +const MAX_TRIES = 3; +const RETRY_WAITS_MS = [1_000, 5_000] as const; +const MAX_WAITING = 100; + +interface InFlight { + readonly envelope: EventEnvelope; + request?: { headers: Record; body: string }; + controller?: AbortController; + timer?: TimerHandle; +} + +interface QueueOptions { + readonly destination: AlertDestination; + readonly poster: WebhookPoster; + readonly clock: Clock; + readonly logger: Logger; + readonly version: string; +} + +function errorCode(error: unknown): string { + const cause = (error as { cause?: { code?: unknown } } | null)?.cause; + return typeof cause?.code === "string" ? cause.code : "network-error"; +} + +export class DestinationQueue { + readonly #options: QueueOptions; + readonly #waiting: EventEnvelope[] = []; + #current: InFlight | undefined; + #stopped = false; + + constructor(options: QueueOptions) { + this.#options = options; + } + + enqueue(envelope: EventEnvelope): void { + if (this.#stopped) return; + if (this.#waiting.length >= MAX_WAITING) { + const dropped = this.#waiting.shift() as EventEnvelope; + this.#fields("warn", "Alert dropped: queue full", dropped); + } + this.#waiting.push(envelope); + this.#pump(); + } + + stop(): number { + this.#stopped = true; + const discarded = this.#waiting.length + (this.#current === undefined ? 0 : 1); + this.#waiting.length = 0; + const current = this.#current; + this.#current = undefined; + if (current?.timer !== undefined) this.#options.clock.cancel(current.timer); + current?.controller?.abort(); + return discarded; + } + + #fields( + level: "debug" | "warn", + message: string, + envelope: EventEnvelope, + extra: Record = {}, + ): void { + this.#options.logger[level](message, { + destination: this.#options.destination.name, + event: envelope.event, + eventId: envelope.id, + ...extra, + }); + } + + #pump(): void { + if (this.#current !== undefined || this.#stopped) return; + const envelope = this.#waiting.shift(); + if (envelope === undefined) return; + const message: InFlight = { envelope }; + this.#current = message; + // why: the bus handler only enqueues; building and the first POST happen after it returns. + queueMicrotask(() => this.#start(message)); + } + + #start(message: InFlight): void { + if (this.#current !== message) return; + const { envelope } = message; + try { + message.request = ALERT_FORMAT_REGISTRY[this.#options.destination.format].build({ + envelope, + destination: this.#options.destination, + version: this.#options.version, + workerLabel: (id) => id, + }); + } catch { + this.#fields("warn", "Alert send gave up", envelope, { attempt: 1, error: "build-failed" }); + this.#finish(message); + return; + } + this.#attempt(message, 1); + } + + #finish(message: InFlight): void { + if (this.#current !== message) return; + this.#current = undefined; + this.#pump(); + } + + #attempt(message: InFlight, attempt: number): void { + const { clock, poster, destination } = this.#options; + const request = message.request as { headers: Record; body: string }; + const controller = new AbortController(); + message.controller = controller; + let settled = false; + + // why: a try ends exactly once; whichever of the answer, the failure or the timeout comes + // first wins, and a late settle of an aborted try changes nothing. + const settle = (outcome: { status: number } | { error: string }): void => { + if (settled || this.#current !== message) return; + settled = true; + if (message.timer !== undefined) clock.cancel(message.timer); + this.#decide(message, attempt, outcome); + }; + + message.timer = clock.setTimer(TRY_TIMEOUT_MS, () => { + if (settled || this.#current !== message) return; + controller.abort(); + settle({ error: "timeout" }); + }); + + let pending: Promise<{ status: number }>; + try { + pending = poster.post({ + url: destination.url, + headers: request.headers, + body: request.body, + signal: controller.signal, + }); + } catch (error) { + pending = Promise.reject(error); + } + pending.then( + ({ status }) => settle({ status }), + (error: unknown) => settle({ error: errorCode(error) }), + ); + } + + #decide( + message: InFlight, + attempt: number, + outcome: { status: number } | { error: string }, + ): void { + const { envelope } = message; + if ("status" in outcome && outcome.status >= 200 && outcome.status < 300) { + this.#fields("debug", "Alert delivered", envelope, { attempt }); + this.#finish(message); + return; + } + const retryable = "error" in outcome || outcome.status === 429 || outcome.status >= 500; + if (!retryable || attempt >= MAX_TRIES) { + this.#fields("warn", "Alert send gave up", envelope, { attempt, ...outcome }); + this.#finish(message); + return; + } + message.timer = this.#options.clock.setTimer(RETRY_WAITS_MS[attempt - 1] as number, () => { + if (this.#current === message) this.#attempt(message, attempt + 1); + }); + } +} diff --git a/src/cli/index.test.ts b/src/cli/index.test.ts index 0b22c925f..b1eb39489 100644 --- a/src/cli/index.test.ts +++ b/src/cli/index.test.ts @@ -7102,6 +7102,7 @@ function testConfig(): Config { execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free" as const, + alerts: { destinations: [] }, }, drivers: {}, eventBuffer: { capacity: 100 }, diff --git a/src/contract/AGENTS.md b/src/contract/AGENTS.md index ef616c006..357fbac7d 100644 --- a/src/contract/AGENTS.md +++ b/src/contract/AGENTS.md @@ -55,7 +55,7 @@ Three identities: ## Protocol version -- `PROTOCOL_VERSION_RANGE` in `protocol.ts` holds it. `DAEMON_PROTOCOL_VERSION` (`src/daemon-protocol/`) is its `max`. +- `PROTOCOL_VERSION_RANGE` in `protocol.ts` holds it, now `{ min: 49, max: 49 }`. `DAEMON_PROTOCOL_VERSION` (`src/daemon-protocol/`) is its `max`. - Every change to a contract shape bumps it by one in the same PR. Move `min` too, unless a compatibility path stays side by side. `.agents/scripts/protocol-bump-check.mjs` checks it on push and in CI; it counts every file here as a shape but tests, `index.ts` and the diff --git a/src/contract/operations.test.ts b/src/contract/operations.test.ts index 2b38ee2d6..caa7c13f3 100644 --- a/src/contract/operations.test.ts +++ b/src/contract/operations.test.ts @@ -1082,6 +1082,7 @@ describe("operation input/output round trips", () => { execTimeoutMs: 660_000, leaseRequestTimeoutMs: 300_000, routing: "warm-then-free", + alerts: { destinations: [] }, }, capacity: { strategy: "fixed", config: { maxRunning: 4 } }, downloads: { policy: "on-request", acceptAndroidLicenses: false, timeoutMs: 1_000 }, @@ -1151,6 +1152,7 @@ describe("operation input/output round trips", () => { execTimeoutMs: 1, leaseRequestTimeoutMs: 1, routing: "warm-then-free", + alerts: { destinations: [] }, }, capacity: { strategy: "fixed", config: { maxRunning: 4 } }, downloads: { policy: "on-request", acceptAndroidLicenses: false, timeoutMs: 1 }, diff --git a/src/contract/protocol.test.ts b/src/contract/protocol.test.ts index 4ee598906..b827f59f0 100644 --- a/src/contract/protocol.test.ts +++ b/src/contract/protocol.test.ts @@ -47,7 +47,7 @@ describe("helloRequestSchema", () => { }); describe("PROTOCOL_VERSION_RANGE", () => { - it("is one above #713's, for http.proxyAuth", () => { - expect(PROTOCOL_VERSION_RANGE).toEqual({ max: 48, min: 48 }); + it("is one above the base branch's, for alerts destinations in config.get", () => { + expect(PROTOCOL_VERSION_RANGE).toEqual({ max: 49, min: 49 }); }); }); diff --git a/src/contract/protocol.ts b/src/contract/protocol.ts index 1ffe8bf1d..09ce670dd 100644 --- a/src/contract/protocol.ts +++ b/src/contract/protocol.ts @@ -10,7 +10,7 @@ export interface ProtocolRange { const protocolRangeSchema = z.object({ min: z.number().int(), max: z.number().int() }); /** Widen only once a second version is kept alive side by side with the first; a bump that leaves no shim behind moves both ends. */ -export const PROTOCOL_VERSION_RANGE: ProtocolRange = { min: 48, max: 48 }; +export const PROTOCOL_VERSION_RANGE: ProtocolRange = { min: 49, max: 49 }; const LEGACY_DAEMON_PROTOCOL_VERSION = 2; diff --git a/src/contract/schemas.ts b/src/contract/schemas.ts index 99fe12ea2..decd7ebc3 100644 --- a/src/contract/schemas.ts +++ b/src/contract/schemas.ts @@ -1110,6 +1110,20 @@ export const configSchema = z.object({ * header comment), gateway or worker alike. `"warm-then-free"` is the only policy in v1; * adding a second means adding it to both registries, exactly as a capacity strategy does. */ routing: z.enum(["warm-then-free"]), + /** + * why: `format` and `events` are re-declared, not imported from core; a test checks `format` against core's list. `secret` is optional so a `slack` destination parses. + */ + alerts: z.object({ + destinations: z.array( + z.object({ + name: z.string(), + format: z.enum(["generic"]), + url: z.string(), + events: z.array(z.string()), + secret: z.string().optional(), + }), + ), + }), }), stalledTransition: z.object({ thresholdMultiplier: z.number(), diff --git a/src/core/cleanup/idle-destroy.test.ts b/src/core/cleanup/idle-destroy.test.ts index b63230165..f50333a01 100644 --- a/src/core/cleanup/idle-destroy.test.ts +++ b/src/core/cleanup/idle-destroy.test.ts @@ -15,6 +15,7 @@ const config: Config = { execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free" as const, + alerts: { destinations: [] }, }, drivers: {}, eventBuffer: { capacity: 1 }, diff --git a/src/core/cleanup/idle-shutdown.test.ts b/src/core/cleanup/idle-shutdown.test.ts index efb404917..0405e4059 100644 --- a/src/core/cleanup/idle-shutdown.test.ts +++ b/src/core/cleanup/idle-shutdown.test.ts @@ -13,6 +13,7 @@ const config: Config = { execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free" as const, + alerts: { destinations: [] }, }, drivers: {}, eventBuffer: { capacity: 1 }, diff --git a/src/core/config.test.ts b/src/core/config.test.ts index 92512b650..8327c4079 100644 --- a/src/core/config.test.ts +++ b/src/core/config.test.ts @@ -7,6 +7,8 @@ import { MemoryFilesystem, FakeSystemStats } from "../ports/index.js"; import type { ResourceStrategyOptions } from "./capacity/index.js"; import { configSchema } from "../contract/schemas.js"; import { + ALERT_EVENTS, + ALERT_FORMATS, type Config, effectiveAllowDownload, loadConfig, @@ -1721,6 +1723,7 @@ describe("loadConfig modes (ADR 0005)", () => { execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free", + alerts: { destinations: [] }, }); }); @@ -1924,6 +1927,145 @@ describe("effectiveAllowDownload", () => { }); }); +describe("gateway.alerts.destinations", () => { + const validDestination = { + name: "audit", + format: "generic", + url: "https://n8n.example.com/x", + events: ["worker.connected"], + secret: "s3cr3t", + }; + const nameExpected = + "expected a name of lowercase letters, digits and dashes that starts with a letter, at most 32 characters"; + const urlExpected = + "expected an https:// URL, or an http:// URL to localhost, 127.0.0.1 or [::1], with no user name or password"; + const eventsExpected = `expected a non-empty list of alert events, each once: ${ALERT_EVENTS.join(", ")}`; + + async function load( + gateway: Record, + mode: "gateway" | "worker" = "gateway", + warn: (message: string) => void = () => {}, + ): Promise { + const filesystem = new MemoryFilesystem(); + await filesystem.mkdirp("/home/agent/.simlock"); + await filesystem.writeFileAtomic(configPath, JSON.stringify({ mode, gateway })); + return loadConfig({ configPath, filesystem, systemStats: createStats(), warn }); + } + + function loadDestinations(destinations: unknown, mode: "gateway" | "worker" = "gateway") { + return load({ alerts: { destinations } }, mode); + } + + it("loadConfig accepts a valid generic destination and defaults gateway.alerts.destinations to an empty list", async () => { + expect((await load({})).gateway.alerts.destinations).toEqual([]); + for (const url of [ + "https://n8n.example.com/x", + "http://localhost:5678/x", + "http://127.0.0.1/x", + "http://[::1]:80/x", + ]) { + const config = await loadDestinations([{ ...validDestination, url }]); + expect(config.gateway.alerts.destinations).toEqual([{ ...validDestination, url }]); + } + const names = await loadDestinations([ + { ...validDestination, name: "ops-1" }, + validDestination, + ]); + expect(names.gateway.alerts.destinations.map((each) => each.name)).toEqual(["ops-1", "audit"]); + }); + + const bad = (field: string, value: unknown) => ({ ...validDestination, [field]: value }); + const without = (field: string) => { + const { [field as "name"]: _omitted, ...rest } = validDestination; + return rest; + }; + + it.each([ + ["name", bad("name", "Audit"), nameExpected], + ["name", bad("name", "1ops"), nameExpected], + ["name", bad("name", "a_b"), nameExpected], + ["name", bad("name", ""), nameExpected], + ["name", bad("name", "a".repeat(33)), nameExpected], + ["name", without("name"), nameExpected], + ["format", bad("format", "slack"), 'expected one of "generic"'], + ["format", bad("format", "json"), 'expected one of "generic"'], + ["format", without("format"), 'expected one of "generic"'], + ["url", bad("url", "http://n8n.internal/x"), urlExpected], + ["url", bad("url", "ftp://example.com"), urlExpected], + ["url", bad("url", "n8n.example.com/x"), urlExpected], + ["url", bad("url", "https://user:pw@host/x"), urlExpected], + ["url", bad("url", "https://user@host/x"), urlExpected], + ["url", bad("url", ""), urlExpected], + ["url", without("url"), urlExpected], + ["events", bad("events", []), eventsExpected], + ["events", bad("events", ["lease.granted"]), eventsExpected], + ["events", bad("events", ["worker.connected", "worker.connected"]), eventsExpected], + ["events", without("events"), eventsExpected], + ["secret", bad("secret", ""), "expected a non-empty secret for a generic destination"], + ["secret", without("secret"), "expected a non-empty secret for a generic destination"], + ])("loadConfig refuses a bad %s with its message", async (field, destination, expected) => { + await expect(loadDestinations([validDestination])).resolves.toBeDefined(); + await expect(loadDestinations([destination])).rejects.toThrow( + `Invalid config value for "gateway.alerts.destinations[0].${field}": ${expected}`, + ); + }); + + it("loadConfig refuses a repeated name, a bad list and a bad entry, on a gateway and on a worker", async () => { + await expect(loadDestinations([validDestination, validDestination])).rejects.toThrow( + 'Invalid config value for "gateway.alerts.destinations[1].name": expected a name no other destination uses', + ); + const sixteen = Array.from({ length: 16 }, (_, index) => ({ + ...validDestination, + name: `d${index}`, + })); + await expect(loadDestinations(sixteen)).resolves.toBeDefined(); + for (const list of [[...sixteen, validDestination], {}, "x"]) { + await expect(loadDestinations(list)).rejects.toThrow( + 'Invalid config value for "gateway.alerts.destinations": expected a list of at most 16 destinations', + ); + } + for (const entry of ["x", null, []]) { + await expect(loadDestinations([entry])).rejects.toThrow( + 'Invalid config value for "gateway.alerts.destinations[0]": expected an object', + ); + } + await expect(loadDestinations([bad("url", "ftp://x")], "worker")).rejects.toThrow( + `Invalid config value for "gateway.alerts.destinations[0].url": ${urlExpected}`, + ); + }); + + it("loadConfig reports name before url when two fields are bad, and the first bad destination", async () => { + await expect( + loadDestinations([ + { ...validDestination, url: "ftp://x", name: "Bad" }, + { ...validDestination, name: "Worse" }, + ]), + ).rejects.toThrow(`"gateway.alerts.destinations[0].name": ${nameExpected}`); + }); + + it("loadConfig warns about an unknown key inside a destination and still loads", async () => { + const warnings: string[] = []; + const config = await load( + { alerts: { destinations: [{ ...validDestination, headers: {} }] } }, + "gateway", + (message) => warnings.push(message), + ); + expect(warnings).toEqual(['Unknown config key: "gateway.alerts.destinations[0].headers"']); + expect(config.gateway.alerts.destinations).toEqual([validDestination]); + }); + + it("config.get's format enum lists exactly core's ALERT_FORMATS", () => { + const destination = configSchema.shape.gateway.shape.alerts.shape.destinations.element; + expect(destination.shape.format.options).toEqual([...ALERT_FORMATS]); + }); + + it("config.get's schema keeps every field of a loaded destination", async () => { + const config = await loadDestinations([validDestination]); + const parsed = configSchema.parse(config); + expect(parsed.gateway.alerts).toEqual(config.gateway.alerts); + }); +}); + describe("redactConfig", () => { async function load(contents: unknown): Promise { const filesystem = new MemoryFilesystem(); @@ -1946,6 +2088,44 @@ describe("redactConfig", () => { expect(config.gateway.token).toBe("secret-join-token"); }); + it("masks every destination's url and secret, keeps name, format and events, and leaves its input untouched", async () => { + const destinations = [ + { + name: "audit", + format: "generic", + url: "https://n8n.example.com/hook-token", + events: ["worker.connected"], + secret: "s3cr3t", + }, + { + name: "ops", + format: "generic", + url: "http://localhost:5678/x", + events: ["worker.disconnected"], + secret: "other", + }, + ]; + for (const token of ["join-token", undefined]) { + const config = await load({ + mode: "gateway", + gateway: { token, alerts: { destinations } }, + }); + const before = structuredClone(config); + + const redacted = redactConfig(config); + + expect(redacted.gateway.alerts.destinations).toEqual( + destinations.map((destination) => ({ + ...destination, + url: REDACTED_VALUE, + secret: REDACTED_VALUE, + })), + ); + expect(redacted.gateway.token).toBe(token === undefined ? undefined : REDACTED_VALUE); + expect(config).toEqual(before); + } + }); + it("adds no gateway.token to a config that has none, so unset still reads as unset", async () => { const config = await load({}); diff --git a/src/core/config.ts b/src/core/config.ts index 8cf012fff..3aa40c446 100644 --- a/src/core/config.ts +++ b/src/core/config.ts @@ -94,6 +94,118 @@ const DEFAULT_ROUTING_POLICY = "warm-then-free"; * force. */ type RoutingPolicyName = (typeof ROUTING_POLICIES)[number]; +export const ALERT_EVENTS = [ + "worker.connected", + "worker.disconnected", + "worker.rejected", + "worker.drain-started", + "worker.drain-ended", + "device.quarantine-abandoned", + "device.quarantine-stranded", + "device.recovery-failed", + "device.stalled-transition-detected", + "disk.pressure-detected", + "component.install-failed", +] as const; +export type AlertEventName = (typeof ALERT_EVENTS)[number]; + +/** + * why: re-declared in the contract's `configSchema`, which imports nothing from core; a test checks the two agree. + */ +export const ALERT_FORMATS = ["generic"] as const; +export type AlertFormatName = (typeof ALERT_FORMATS)[number]; + +/** why: checked at config load, which cannot import the alerts module. */ +export const ALERT_FORMAT_SECRET: Record = { + generic: "required", +}; + +export interface AlertDestination { + readonly name: string; + readonly format: AlertFormatName; + readonly url: string; + readonly events: readonly AlertEventName[]; + readonly secret?: string; +} + +const MAX_ALERT_DESTINATIONS = 16; +const ALERT_NAME_PATTERN = /^[a-z][a-z0-9-]{0,31}$/; +const ALERT_LOCAL_HOSTS = ["localhost", "127.0.0.1", "[::1]"]; +const ALERT_URL_EXPECTED = + "an https:// URL, or an http:// URL to localhost, 127.0.0.1 or [::1], with no user name or password"; + +function isAlertUrl(value: unknown): value is string { + if (typeof value !== "string") return false; + let url: URL; + try { + url = new URL(value); + } catch { + return false; + } + if (url.username !== "" || url.password !== "") return false; + if (url.protocol === "https:") return true; + return url.protocol === "http:" && ALERT_LOCAL_HOSTS.includes(url.hostname); +} + +function alertDestinationValidator( + names: Set, +): (value: unknown, path: string, warn: Warn) => AlertDestination { + return (value, path, warn) => { + const entry = requireObject(value, path); + for (const key of Object.keys(entry)) { + if (!["name", "format", "url", "events", "secret"].includes(key)) { + warn(`Unknown config key: "${path}.${key}"`); + } + } + const { name, format, url, events, secret } = entry; + if (typeof name !== "string" || !ALERT_NAME_PATTERN.test(name)) { + throw invalidValue( + `${path}.name`, + "a name of lowercase letters, digits and dashes that starts with a letter, at most 32 characters", + ); + } + if (names.has(name)) throw invalidValue(`${path}.name`, "a name no other destination uses"); + if (typeof format !== "string" || !(ALERT_FORMATS as readonly string[]).includes(format)) { + throw invalidValue( + `${path}.format`, + `one of ${ALERT_FORMATS.map((each) => `"${each}"`).join(", ")}`, + ); + } + if (!isAlertUrl(url)) throw invalidValue(`${path}.url`, ALERT_URL_EXPECTED); + if ( + !Array.isArray(events) || + events.length === 0 || + new Set(events).size !== events.length || + !events.every((event) => (ALERT_EVENTS as readonly unknown[]).includes(event)) + ) { + throw invalidValue( + `${path}.events`, + `a non-empty list of alert events, each once: ${ALERT_EVENTS.join(", ")}`, + ); + } + const secretRule = ALERT_FORMAT_SECRET[format as AlertFormatName]; + if (secretRule === "required" && (typeof secret !== "string" || secret.length === 0)) { + throw invalidValue(`${path}.secret`, `a non-empty secret for a ${format} destination`); + } + names.add(name); + return { + name, + format: format as AlertFormatName, + url, + events: events as AlertEventName[], + ...(secret === undefined ? {} : { secret: secret as string }), + }; + }; +} + +function alertDestinationsValidator(value: unknown, path: string, warn: Warn): AlertDestination[] { + if (!Array.isArray(value) || value.length > MAX_ALERT_DESTINATIONS) { + throw invalidValue(path, `a list of at most ${MAX_ALERT_DESTINATIONS} destinations`); + } + const validate = alertDestinationValidator(new Set()); + return value.map((entry, index) => validate(entry, `${path}[${index}]`, warn)); +} + /** * `exec.timeoutMs`'s default (ADR 0005 §19e): how long a single `device.exec` command may run * on this worker before it is killed and the operation fails with `EXEC_TIMEOUT`. Ten minutes, @@ -323,6 +435,7 @@ export interface Config { * per-request choice. `"warm-then-free"` is the only value in v1 (`ROUTING_POLICIES` above). */ readonly routing: RoutingPolicyName; + readonly alerts: { readonly destinations: readonly AlertDestination[] }; }; readonly stalledTransition: { /** @@ -343,8 +456,23 @@ export const REDACTED_VALUE = "[redacted]"; // why: masked rather than deleted, so the copy keeps `Config`'s keys and a reader can tell "set" from "unset"; a copy, because the live config is what `config.get` answers and the uplink presents. export function redactConfig(config: Config): Config { - if (config.gateway.token === undefined) return config; - return { ...config, gateway: { ...config.gateway, token: REDACTED_VALUE } }; + const { token, alerts } = config.gateway; + if (token === undefined && alerts.destinations.length === 0) return config; + return { + ...config, + gateway: { + ...config.gateway, + ...(token === undefined ? {} : { token: REDACTED_VALUE }), + alerts: { + ...alerts, + destinations: alerts.destinations.map((destination) => ({ + ...destination, + url: REDACTED_VALUE, + ...(destination.secret === undefined ? {} : { secret: REDACTED_VALUE }), + })), + }, + }, + }; } /** @@ -820,6 +948,7 @@ function defaultConfig( execTimeoutMs: DEFAULT_GATEWAY_EXEC_TIMEOUT_MS, leaseRequestTimeoutMs: DEFAULT_LEASE_REQUEST_TIMEOUT_MS, routing: DEFAULT_ROUTING_POLICY, + alerts: { destinations: [] }, }, stalledTransition: { thresholdMultiplier: 3, @@ -893,6 +1022,7 @@ function configValidators( execTimeoutMs: positiveNumber, leaseRequestTimeoutMs: positiveNumber, routing: stringUnion(ROUTING_POLICIES), + alerts: objectValidator({ destinations: alertDestinationsValidator }), }), capacity: objectValidator({ strategy: stringUnion(capacityStrategyNames), diff --git a/src/core/doctor.test.ts b/src/core/doctor.test.ts index 8ef76df7e..337f204a4 100644 --- a/src/core/doctor.test.ts +++ b/src/core/doctor.test.ts @@ -3233,6 +3233,7 @@ function config(stalledTransitionOverrides: Partial execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free" as const, + alerts: { destinations: [] }, }, drivers: {}, eventBuffer: { capacity: 10 }, diff --git a/src/core/index.ts b/src/core/index.ts index 61bcac96e..bd7c3ca14 100644 --- a/src/core/index.ts +++ b/src/core/index.ts @@ -1,4 +1,10 @@ export { + ALERT_EVENTS, + ALERT_FORMATS, + ALERT_FORMAT_SECRET, + type AlertDestination, + type AlertEventName, + type AlertFormatName, type Config, type ConfigOverrides, // fallow-ignore-next-line unused-type -- public Config surface (config.mode, ADR 0005 §1); every in-tree reader gets it structurally off `Config` diff --git a/src/core/nuke.test.ts b/src/core/nuke.test.ts index bb9cdb40e..3ed914bd8 100644 --- a/src/core/nuke.test.ts +++ b/src/core/nuke.test.ts @@ -165,6 +165,7 @@ function config(): Config { execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free" as const, + alerts: { destinations: [] }, }, drivers: {}, eventBuffer: { capacity: 10 }, diff --git a/src/core/reaper.test.ts b/src/core/reaper.test.ts index a694f1a2c..3ac190876 100644 --- a/src/core/reaper.test.ts +++ b/src/core/reaper.test.ts @@ -61,6 +61,7 @@ function config(): Config { execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free" as const, + alerts: { destinations: [] }, }, drivers: {}, eventBuffer: { capacity: 100 }, diff --git a/src/daemon-protocol/index.test.ts b/src/daemon-protocol/index.test.ts index dc5abbb04..edd7d6c36 100644 --- a/src/daemon-protocol/index.test.ts +++ b/src/daemon-protocol/index.test.ts @@ -9,7 +9,7 @@ import { describe("daemon protocol", () => { it("keeps the current protocol version and newline framing", () => { - expect(DAEMON_PROTOCOL_VERSION).toBe(48); + expect(DAEMON_PROTOCOL_VERSION).toBe(49); expect(serializeFrame({ id: 1, type: "hello" })).toBe('{"id":1,"type":"hello"}\n'); }); diff --git a/src/daemon/AGENTS.md b/src/daemon/AGENTS.md index ac1ef1f47..c3b43684d 100644 --- a/src/daemon/AGENTS.md +++ b/src/daemon/AGENTS.md @@ -2,8 +2,8 @@ Does: one process per data directory — composes a worker or a gateway and serves the contract on a unix socket. Entry: `main.ts` → `startDaemon(options)` → a started `DaemonServer`. -Uses: contract, daemon-protocol, ports, bus, history, link, core, leasing, drivers, http, host-load, modules, module-registry; gateway from `main.ts` only. -Never imports: gateway outside `main.ts`. `dispatch.ts` imports no core, drivers or http. +Uses: contract, daemon-protocol, ports, bus, history, link, core, leasing, drivers, http, host-load, modules, module-registry; gateway and alerts from `main.ts` only. +Never imports: gateway or alerts outside `main.ts`. `dispatch.ts` imports no core, drivers or http. Imported by: http (`dispatcher.ts`, `error-code.ts`, `owner-routed-facts.ts`); gateway (`dispatch.ts` only). Started by: cli and mcp spawn `main.ts` as a detached process. Flows: [startup](../../docs/internal/flows/startup.md), [device.exec](../../docs/internal/flows/device-exec.md), [gateway](../gateway/AGENTS.md). @@ -37,7 +37,9 @@ Both modes: config (`config.json`) → `Logger` → host info → `EventBus` → HTTP starts only when `http.enabled`. - **Gateway** — instance id → `TokenStore`, `AdminSecretManager` → `FleetLeaseIndex` (requester prefix `gw::`) → `GatewayService` (uplink listener, `workers.json`) - → `FleetLeaseCoordinator` → `DaemonServer` with `GatewayDispatcher`. HTTP always + → `FleetLeaseCoordinator` → `createAlerts` when `gateway.alerts.destinations` is not + empty (before `start()`, so no worker joins first; stopped first in `stopAuxiliary`, before the + uplinks close) → `DaemonServer` with `GatewayDispatcher`. HTTP always starts; the uplink upgrades on the same listener. An HTTP bind failure after a good start stops the daemon (`http-start-failed`). @@ -51,7 +53,7 @@ An HTTP bind failure after a good start stops the daemon (`http-start-failed`). Principal: `hello.principal`, else `SIMLOCK_AGENT_ID`, else the daemon's pid. Both are fixed for the connection's life. 3. **Version** — ranges overlap → highest common version. `PROTOCOL_VERSION_RANGE` - (`src/contract/protocol.ts`) is the daemon's range. No overlap → the + is `{ min: 49, max: 49 }` (`src/contract/protocol.ts`). No overlap → the connection stays open, and every request but `daemon.stop` gets `PROTOCOL_VERSION_UNSUPPORTED`. 4. **Request** — one frame → one `dispatch()` call. While stopping: `DAEMON_STOPPING`. diff --git a/src/daemon/dispatcher.test.ts b/src/daemon/dispatcher.test.ts index 366a4f042..9ddc48f8c 100644 --- a/src/daemon/dispatcher.test.ts +++ b/src/daemon/dispatcher.test.ts @@ -3784,6 +3784,7 @@ function testConfig( execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free" as const, + alerts: { destinations: [] }, }, drivers: {}, exec: { timeoutMs: 600_000, ...execOverrides }, diff --git a/src/daemon/gateway-fleet.e2e.test.ts b/src/daemon/gateway-fleet.e2e.test.ts index 1f0d3184b..fff7772c6 100644 --- a/src/daemon/gateway-fleet.e2e.test.ts +++ b/src/daemon/gateway-fleet.e2e.test.ts @@ -21,6 +21,8 @@ * so it can flake under CPU contention alongside an unrelated file's own suite -- re-run this * file alone before treating a failure here as a real regression. */ +import { createHmac } from "node:crypto"; +import { createServer as createHttpServer, type Server } from "node:http"; import { mkdtemp, rm } from "node:fs/promises"; import { createServer } from "node:net"; import { tmpdir } from "node:os"; @@ -28,7 +30,7 @@ import { join } from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; -import type { PassthroughCommand } from "../core/index.js"; +import type { AlertDestination, PassthroughCommand } from "../core/index.js"; import { FakeDriver } from "../core/testing.js"; import type { Filesystem, IpcConnector, IpcListenerFactory } from "../ports/index.js"; import { @@ -38,6 +40,7 @@ import { NoopLogger, SystemClock, } from "../ports/index.js"; +import { FakeWebhookPoster } from "../ports/testing.js"; import type { DispatchSession } from "./dispatch.js"; import { startDaemon } from "./main.js"; import type { DaemonServer } from "./server.js"; @@ -127,7 +130,9 @@ describe("gateway fleet smoke (ADR 0005 §35)", () => { ); }); - async function startGateway(): Promise { + async function startGateway( + alerts: { destinations?: readonly AlertDestination[]; poster?: FakeWebhookPoster } = {}, + ): Promise { const directory = await mkdtemp(join(tmpdir(), "simlock-e2e-gateway-")); directories.push(directory); gatewayPort = await freeLoopbackPort(); @@ -135,6 +140,9 @@ describe("gateway fleet smoke (ADR 0005 §35)", () => { configOverrides: { mode: "gateway", http: { enabled: true, host: "127.0.0.1", port: gatewayPort }, + ...(alerts.destinations === undefined + ? {} + : { gateway: { alerts: { destinations: alerts.destinations } } }), }, dataDirectory: directory, filesystem: new MemoryFilesystem(), @@ -142,6 +150,7 @@ describe("gateway fleet smoke (ADR 0005 §35)", () => { logger: new NoopLogger(), statePath: join(directory, "state.json"), version: "1.0.0-e2e", + ...(alerts.poster === undefined ? {} : { webhookPoster: alerts.poster }), }); daemons.push(daemon); return daemon; @@ -195,45 +204,48 @@ describe("gateway fleet smoke (ADR 0005 §35)", () => { readonly existing?: { readonly filesystem: Filesystem; readonly directory: string }; /** The worker's own socket transport; omitted, the real one. */ readonly ipc?: IpcConnector & IpcListenerFactory; + readonly alerts?: { destinations: readonly AlertDestination[]; poster: FakeWebhookPoster }; + readonly onDriver?: (driver: FakeDriver) => void; }): Promise<{ daemon: DaemonServer; filesystem: Filesystem; directory: string }> { const directory = options.existing?.directory ?? (await mkdtemp(join(tmpdir(), `simlock-e2e-${options.label}-`))); if (options.existing === undefined) directories.push(directory); const filesystem = options.existing?.filesystem ?? new MemoryFilesystem(); + const driver = new FakeDriver({ + availableOsVersions: ["18.0"], + // why: a real command, so `device.exec` proves a real child process actually ran on + // *this* worker: it prints which worker answered and exits with a distinguishing + // code, both asserted below. + ...passthroughOptions({ + args: ["-e", `process.stdout.write(${JSON.stringify(options.stdout)}); process.exit(7);`], + command: process.execPath, + env: {}, + }), + clock: new SystemClock(), + knownModels: [options.model], + platform: "android", + }); + options.onDriver?.(driver); const daemon = await startDaemon({ configOverrides: { gateway: { url: options.gatewayUrl ?? gatewayUrl(), token: options.token, label: options.label, + ...(options.alerts === undefined + ? {} + : { alerts: { destinations: options.alerts.destinations } }), }, }, dataDirectory: directory, - drivers: [ - new FakeDriver({ - availableOsVersions: ["18.0"], - // A real command, so `device.exec` proves a real child process actually ran on - // *this* worker: it prints which worker answered and exits with a distinguishing - // code, both asserted below. - ...passthroughOptions({ - args: [ - "-e", - `process.stdout.write(${JSON.stringify(options.stdout)}); process.exit(7);`, - ], - command: process.execPath, - env: {}, - }), - clock: new SystemClock(), - knownModels: [options.model], - platform: "android", - }), - ], + drivers: [driver], filesystem, ...(options.ipc === undefined ? {} : { ipc: options.ipc }), logger: new NoopLogger(), statePath: join(directory, "state.json"), version: "1.0.0-e2e", + ...(options.alerts === undefined ? {} : { webhookPoster: options.alerts.poster }), }); daemons.push(daemon); return { daemon, directory, filesystem }; @@ -573,4 +585,139 @@ describe("gateway fleet smoke (ADR 0005 §35)", () => { agentSession({ principal: "fleet-owner" }), ); }, 55_000); + + const destinationFor = (overrides: Partial): AlertDestination => ({ + name: "audit", + format: "generic", + url: "https://alerts.example.com/hook", + events: ["component.install-failed"], + secret: "e2e-secret", + ...overrides, + }); + + async function workerToken(gateway: DaemonServer, label: string): Promise { + const { secret } = await gateway.dispatch( + "token.create", + { role: "worker", label }, + adminSession(), + ); + return secret; + } + + it("a worker with a destination posts nothing, while its gateway posts the relayed event", async () => { + const gatewayPoster = new FakeWebhookPoster(); + const workerPoster = new FakeWebhookPoster(); + const destinations = [destinationFor({})]; + const gateway = await startGateway({ destinations, poster: gatewayPoster }); + let driver: FakeDriver | undefined; + const worker = await startWorker({ + label: "worker-a", + model: "Pixel-A", + token: await workerToken(gateway, "worker-a"), + stdout: "a", + alerts: { destinations, poster: workerPoster }, + onDriver: (created) => { + driver = created; + }, + }); + + await vi.waitFor(async () => { + const events = (await gateway.dispatch("events.replay", {}, adminSession())) as Array<{ + readonly event: string; + }>; + expect(events.some((entry) => entry.event === "worker.connected")).toBe(true); + }); + driver?.failOn("installComponent", 1, new Error("installer exited 1")); + await expect( + worker.daemon.dispatch( + "component.install", + { platform: "android", version: "35" }, + adminSession(), + ), + ).rejects.toThrow(); + + await vi.waitFor(() => expect(gatewayPoster.requests).toHaveLength(1)); + const body = JSON.parse(gatewayPoster.requests[0]?.body ?? ""); + expect(body.event).toBe("component.install-failed"); + expect(body.payload.workerId).toEqual(expect.any(String)); + expect(workerPoster.requests).toEqual([]); + }, 30_000); + + it("stopping a gateway sends no worker.disconnected and aborts a try in flight", async () => { + const poster = new FakeWebhookPoster(); + poster.respond = (request) => + request.url.includes("stuck") ? new Promise(() => {}) : Promise.resolve({ status: 200 }); + const gateway = await startGateway({ + destinations: [ + destinationFor({ + name: "quick", + url: "https://quick.example.com/x", + events: ["worker.disconnected"], + }), + destinationFor({ + name: "stuck", + url: "https://stuck.example.com/x", + events: ["worker.connected"], + }), + ], + poster, + }); + await startWorker({ + label: "worker-a", + model: "Pixel-A", + token: await workerToken(gateway, "worker-a"), + stdout: "a", + }); + await vi.waitFor(() => expect(poster.requests).toHaveLength(1)); + const stuck = poster.requests[0]; + expect(stuck?.url).toBe("https://stuck.example.com/x"); + + await gateway.stop("test"); + + expect(stuck?.signal.aborted).toBe(true); + expect(poster.requests.filter((request) => request.url.includes("quick"))).toEqual([]); + }, 30_000); + + it("a worker joining a gateway with a generic destination makes a loopback receiver get one signed worker.connected naming that worker", async () => { + const received: Array<{ headers: Record; body: string }> = []; + const receiver: Server = createHttpServer((request, response) => { + const chunks: Buffer[] = []; + request.on("data", (chunk: Buffer) => chunks.push(chunk)); + request.on("end", () => { + received.push({ headers: request.headers, body: Buffer.concat(chunks).toString("utf8") }); + response.writeHead(200).end(); + }); + }); + await new Promise((resolve) => receiver.listen(0, "127.0.0.1", resolve)); + try { + const port = (receiver.address() as { port: number }).port; + const gateway = await startGateway({ + destinations: [ + destinationFor({ url: `http://127.0.0.1:${port}/hook`, events: ["worker.connected"] }), + ], + }); + await startWorker({ + label: "worker-a", + model: "Pixel-A", + token: await workerToken(gateway, "worker-a"), + stdout: "a", + }); + + await vi.waitFor(() => expect(received).toHaveLength(1)); + const [message] = received; + expect(message?.headers["x-simlock-signature"]).toBe( + `sha256=${createHmac("sha256", "e2e-secret") + .update(message?.body ?? "") + .digest("hex")}`, + ); + expect(message?.headers["x-simlock-event"]).toBe("worker.connected"); + const body = JSON.parse(message?.body ?? ""); + const { workers } = await gateway.dispatch("worker.list", {}, adminSession()); + expect(body.payload.workerId).toBe(workers[0]?.id); + expect(body.payload.label).toBe("worker-a"); + } finally { + receiver.closeAllConnections(); + await new Promise((resolve) => receiver.close(() => resolve())); + } + }, 30_000); }); diff --git a/src/daemon/main.ts b/src/daemon/main.ts index 9ac53ab9d..40abfc809 100644 --- a/src/daemon/main.ts +++ b/src/daemon/main.ts @@ -2,6 +2,7 @@ import { homedir } from "node:os"; import { join, resolve } from "node:path"; import { fileURLToPath, pathToFileURL } from "node:url"; +import { type Alerts, createAlerts } from "../alerts/index.js"; import { EventBus, type EventBusLogger } from "../bus/index.js"; import { fitHostFacts, fitHostLoad } from "../contract/index.js"; import { HostLoadSampler } from "../host-load/index.js"; @@ -92,6 +93,8 @@ import { type ProcessRunner, type ProcessSupervisor, type TcpProbe, + FetchWebhookPoster, + type WebhookPoster, } from "../ports/index.js"; import { createRoutingPolicy, @@ -145,6 +148,8 @@ export interface StartDaemonOptions { readonly systemStats?: SystemStats; readonly tcpProbe?: TcpProbe; readonly version?: string; + /** why: test-only, replaces the poster a gateway's alerts send through. */ + readonly webhookPoster?: WebhookPoster; } /** Constructs the daemon's real adapters once; all state remains in the daemon. */ @@ -241,6 +246,7 @@ export async function startDaemon(options: StartDaemonOptions = {}): Promise[]; readonly socketPath: string; readonly version: string; + readonly webhookPoster?: WebhookPoster; } /** @@ -822,6 +829,8 @@ async function startGatewayDaemon(options: GatewayDaemonOptions): Promise Promise.resolve(), clock, @@ -882,6 +891,8 @@ async function startGatewayDaemon(options: GatewayDaemonOptions): Promise { + // why: before the HTTP gateway closes the worker uplinks, so this gateway's own stop sends no worker.disconnected (ADR 0035 §4). + alerts?.stop(); await moduleHost.beforeDrain(); await httpStarted.catch(() => undefined); await stopHttpGateway?.(); @@ -950,6 +961,18 @@ async function startGatewayDaemon(options: GatewayDaemonOptions): Promise 0) { + alerts = createAlerts({ + clock, + destinations: config.gateway.alerts.destinations, + eventBus, + logger: logger.child("alerts"), + poster: options.webhookPoster ?? new FetchWebhookPoster(), + version: options.version, + }); + } + const daemonStarted = daemon.start().finally(() => { if (!socketClaimed) resolveHttpStarted?.(); }); diff --git a/src/daemon/server.test.ts b/src/daemon/server.test.ts index 73d254dc1..ace071f3d 100644 --- a/src/daemon/server.test.ts +++ b/src/daemon/server.test.ts @@ -219,8 +219,9 @@ describe("DaemonServer", () => { [38, "cannot read agent-device's section under config.get's capabilities"], [46, "cannot call identity.get"], [47, "cannot read config.get's http.proxyAuth"], + [48, "cannot read config.get's gateway.alerts"], ])( - "answers PROTOCOL_VERSION_UNSUPPORTED, naming protocol 48, to a client on protocol %i, which %s", + "answers PROTOCOL_VERSION_UNSUPPORTED, naming protocol 49, to a client on protocol %i, which %s", async (version) => { const harness = await createHarness(); const previous = await createClient(harness.socketPath); @@ -230,7 +231,7 @@ describe("DaemonServer", () => { ).resolves.toMatchObject({ error: { code: "PROTOCOL_VERSION_UNSUPPORTED", - details: { client: { min: version, max: version }, daemon: { min: 48, max: 48 } }, + details: { client: { min: version, max: version }, daemon: { min: 49, max: 49 } }, }, ok: false, }); @@ -3706,6 +3707,7 @@ function testConfig( execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free" as const, + alerts: { destinations: [] }, }, drivers: {}, eventBuffer: { capacity: 100 }, diff --git a/src/gateway/dispatcher.test.ts b/src/gateway/dispatcher.test.ts index 9c5e55375..ce0a3bbef 100644 --- a/src/gateway/dispatcher.test.ts +++ b/src/gateway/dispatcher.test.ts @@ -47,6 +47,7 @@ const gatewayConfig = { execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free" as const, + alerts: { destinations: [] }, }, health: { enabled: false, diff --git a/src/gateway/service.test.ts b/src/gateway/service.test.ts index 95644a387..4b5d27143 100644 --- a/src/gateway/service.test.ts +++ b/src/gateway/service.test.ts @@ -1265,7 +1265,7 @@ describe("GatewayService", () => { min: PROTOCOL_VERSION_RANGE.max - 1, max: PROTOCOL_VERSION_RANGE.max - 1, }; - expect(PROTOCOL_VERSION_RANGE.max).toBe(48); + expect(PROTOCOL_VERSION_RANGE.max).toBe(49); expect(negotiateProtocolVersion(PROTOCOL_VERSION_RANGE, previous)).toBeUndefined(); const harness = fleet(); await harness.service.start(); diff --git a/src/http/test-fakes.ts b/src/http/test-fakes.ts index 6568fef7d..3c0159424 100644 --- a/src/http/test-fakes.ts +++ b/src/http/test-fakes.ts @@ -37,6 +37,7 @@ export function testConfig( execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free" as const, + alerts: { destinations: [] }, }, drivers: {}, capacity: { diff --git a/src/leasing/acquisition-planner.test.ts b/src/leasing/acquisition-planner.test.ts index 3e132d2d6..8fce662e2 100644 --- a/src/leasing/acquisition-planner.test.ts +++ b/src/leasing/acquisition-planner.test.ts @@ -25,6 +25,7 @@ const config: Config = { execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free" as const, + alerts: { destinations: [] }, }, drivers: {}, eventBuffer: { capacity: 100 }, diff --git a/src/leasing/create-leasing.test.ts b/src/leasing/create-leasing.test.ts index c5d8364af..c0b489ebb 100644 --- a/src/leasing/create-leasing.test.ts +++ b/src/leasing/create-leasing.test.ts @@ -53,6 +53,7 @@ function config( execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free" as const, + alerts: { destinations: [] }, }, drivers: {}, downloads: { acceptAndroidLicenses: false, policy: "on-request", timeoutMs: 1_200_000 }, diff --git a/src/leasing/lease-acquisition-coordinator.test.ts b/src/leasing/lease-acquisition-coordinator.test.ts index ee5d05f7c..c953a221b 100644 --- a/src/leasing/lease-acquisition-coordinator.test.ts +++ b/src/leasing/lease-acquisition-coordinator.test.ts @@ -72,6 +72,7 @@ function config(maxDevices = 1, maxRunning = 1): Config { execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free" as const, + alerts: { destinations: [] }, }, drivers: {}, downloads: { acceptAndroidLicenses: false, policy: "on-request", timeoutMs: 1_200_000 }, diff --git a/src/leasing/lease-end-participants.test.ts b/src/leasing/lease-end-participants.test.ts index 8c9e56cec..5e277f9e0 100644 --- a/src/leasing/lease-end-participants.test.ts +++ b/src/leasing/lease-end-participants.test.ts @@ -120,6 +120,7 @@ function config(options: EngineOptions): Config { execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free" as const, + alerts: { destinations: [] }, }, drivers: {}, downloads: { acceptAndroidLicenses: false, policy: "on-request", timeoutMs: 1_200_000 }, diff --git a/src/leasing/lease-health-monitor.test.ts b/src/leasing/lease-health-monitor.test.ts index 51ca893ae..055919326 100644 --- a/src/leasing/lease-health-monitor.test.ts +++ b/src/leasing/lease-health-monitor.test.ts @@ -38,6 +38,7 @@ function config(overrides: Partial = {}): Config { execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free" as const, + alerts: { destinations: [] }, }, drivers: {}, eventBuffer: { capacity: 100 }, diff --git a/src/leasing/physical-device-watcher.test.ts b/src/leasing/physical-device-watcher.test.ts index bbdc3b492..e10713669 100644 --- a/src/leasing/physical-device-watcher.test.ts +++ b/src/leasing/physical-device-watcher.test.ts @@ -54,6 +54,7 @@ function config(options: HarnessOptions): Config { execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free" as const, + alerts: { destinations: [] }, }, drivers: {}, downloads: { acceptAndroidLicenses: false, policy: "on-request", timeoutMs: 1_200_000 }, diff --git a/src/leasing/physical-leasing.test.ts b/src/leasing/physical-leasing.test.ts index 9b977c49d..8a3f2c056 100644 --- a/src/leasing/physical-leasing.test.ts +++ b/src/leasing/physical-leasing.test.ts @@ -42,6 +42,7 @@ function config(leaseOverrides: object = {}): Config { execTimeoutMs: 11 * 60_000, leaseRequestTimeoutMs: 5 * 60_000, routing: "warm-then-free" as const, + alerts: { destinations: [] }, }, drivers: {}, downloads: { acceptAndroidLicenses: false, policy: "on-request", timeoutMs: 1_200_000 }, diff --git a/src/ports/AGENTS.md b/src/ports/AGENTS.md index c8622e69e..ce3208725 100644 --- a/src/ports/AGENTS.md +++ b/src/ports/AGENTS.md @@ -27,6 +27,7 @@ Uses: nothing from other components. Imported by: every component that touches t | `BytePipeLauncher` | `byte-pipe.ts` | `NodeBytePipeLauncher` | `FakeBytePipeLauncher` | A child with binary stdout and stdin as its liveness | | `KeySetFetcher` | `key-set-fetcher.ts` | `NodeKeySetFetcher` | `FakeKeySetFetcher` | One GET of a key-set URL: no redirects, body capped at 64 KiB | | `SqliteOpener` | `sqlite.ts` | `NodeSqliteOpener` | none; tests use the adapter on a temp file | Open a SQLite file; `exec`, `prepare`, `close` | +| `WebhookPoster` | `webhook-poster.ts` | `FetchWebhookPoster` | `FakeWebhookPoster` (`testing.ts`) | POST to an operator-named URL; a redirect is not followed | | `IdGenerator` | `id-generator.ts` | `CryptoIdGenerator` | none; tests inline one | Random ids | | `TokenSecrets` | `token-secrets.ts` | `CryptoTokenSecrets` | none; tests inline one | New secret, its hash | | Paths | `paths.ts` | `resolveSimlockHome`, `resolveDaemonSocketPath` | — | `SIMLOCK_HOME` (default `~/.simlock`), socket path length | diff --git a/src/ports/index.ts b/src/ports/index.ts index a13785369..d06cfb822 100644 --- a/src/ports/index.ts +++ b/src/ports/index.ts @@ -125,3 +125,4 @@ export { SqliteUnavailableError, } from "./sqlite.js"; export type { SqliteDatabase, SqliteOpener, SqliteStatement } from "./sqlite.js"; +export { FetchWebhookPoster, type WebhookPoster, type WebhookRequest } from "./webhook-poster.js"; diff --git a/src/ports/testing.ts b/src/ports/testing.ts index 57af6ad66..300e8e7e9 100644 --- a/src/ports/testing.ts +++ b/src/ports/testing.ts @@ -1,2 +1,14 @@ // why: the real WebSocket adapters stay out of the ports barrel so the CLI never loads `ws`; tests that need them import them from here. export { WebSocketUplinkConnector, WebSocketUplinkListenerFactory } from "./uplink-websocket.js"; +import type { WebhookPoster, WebhookRequest } from "./webhook-poster.js"; + +export class FakeWebhookPoster implements WebhookPoster { + readonly requests: WebhookRequest[] = []; + respond: (request: WebhookRequest, index: number) => Promise<{ status: number }> = () => + Promise.resolve({ status: 200 }); + + post(request: WebhookRequest): Promise<{ status: number }> { + this.requests.push(request); + return this.respond(request, this.requests.length - 1); + } +} diff --git a/src/ports/webhook-poster.test.ts b/src/ports/webhook-poster.test.ts new file mode 100644 index 000000000..5a0b7c350 --- /dev/null +++ b/src/ports/webhook-poster.test.ts @@ -0,0 +1,60 @@ +import { createServer, type Server, type ServerResponse } from "node:http"; +import type { AddressInfo } from "node:net"; + +import { afterEach, describe, expect, it } from "vitest"; + +import { FetchWebhookPoster } from "./webhook-poster.js"; + +const servers: Server[] = []; +afterEach(async () => { + await Promise.all( + servers.splice(0).map( + (server) => + new Promise((resolve) => { + server.closeAllConnections(); + server.close(() => resolve()); + }), + ), + ); +}); + +async function listen(handler: (url: string, response: ServerResponse) => void): Promise { + const server = createServer((request, response) => handler(request.url ?? "", response)); + servers.push(server); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + return `http://127.0.0.1:${(server.address() as AddressInfo).port}`; +} + +describe("FetchWebhookPoster", () => { + it("returns a 301 without following it, cancels the response body unread, and rejects when the signal aborts", async () => { + const hits: string[] = []; + let bodyClosed!: () => void; + const closed = new Promise((resolve) => { + bodyClosed = resolve; + }); + const base = await listen((url, response) => { + hits.push(url); + if (url === "/moved") { + response.writeHead(301, { Location: "/target" }); + response.write("x".repeat(1024)); + response.on("close", bodyClosed); + return; + } + if (url === "/silent") return; + response.writeHead(200).end("ok"); + }); + const poster = new FetchWebhookPoster(); + const controller = new AbortController(); + const post = (path: string, signal = controller.signal) => + poster.post({ url: `${base}${path}`, headers: { "X-A": "1" }, body: "{}", signal }); + + await expect(post("/ok")).resolves.toEqual({ status: 200 }); + await expect(post("/moved")).resolves.toEqual({ status: 301 }); + await closed; + expect(hits).toEqual(["/ok", "/moved"]); + + const pending = post("/silent"); + controller.abort(); + await expect(pending).rejects.toMatchObject({ name: "AbortError" }); + }); +}); diff --git a/src/ports/webhook-poster.ts b/src/ports/webhook-poster.ts new file mode 100644 index 000000000..3f292c08e --- /dev/null +++ b/src/ports/webhook-poster.ts @@ -0,0 +1,27 @@ +export interface WebhookRequest { + readonly url: string; + readonly headers: Readonly>; + readonly body: string; + readonly signal: AbortSignal; +} + +export interface WebhookPoster { + post(request: WebhookRequest): Promise<{ status: number }>; +} + +/** + * why: `redirect: "manual"` keeps a message on the URL the operator wrote, and the response body is cancelled unread because only the status is read. + */ +export class FetchWebhookPoster implements WebhookPoster { + async post(request: WebhookRequest): Promise<{ status: number }> { + const response = await fetch(request.url, { + method: "POST", + headers: request.headers, + body: request.body, + redirect: "manual", + signal: request.signal, + }); + await response.body?.cancel(); + return { status: response.status }; + } +} From cf4f1712b448e05759d429397e43963fbc1e9729 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Tue, 20 Oct 2026 08:34:09 +0200 Subject: [PATCH 2/5] refactor(alerts): split the destination check and keep fallow quiet (#719) --- docs/internal/ARCHITECTURE.md | 2 +- src/alerts/alerts.test.ts | 35 ++++++------- src/alerts/index.ts | 16 +++--- src/alerts/queue.ts | 2 + src/core/config.ts | 76 +++++++++++++++------------- src/core/index.ts | 2 - src/daemon/AGENTS.md | 7 ++- src/daemon/gateway-fleet.e2e.test.ts | 1 + src/ports/index.ts | 2 +- 9 files changed, 72 insertions(+), 71 deletions(-) diff --git a/docs/internal/ARCHITECTURE.md b/docs/internal/ARCHITECTURE.md index 3fa61cfb8..a0322b6e2 100644 --- a/docs/internal/ARCHITECTURE.md +++ b/docs/internal/ARCHITECTURE.md @@ -81,7 +81,7 @@ Arrows point from the optional to the required (architecture rule 15). | `not-modules` | Core, leasing, drivers, gateway and contract import nothing from modules or module-registry | | `not-agent-device` | Core, leasing and drivers import nothing from agent-device | | `host-load-imports-ports-only`, `not-host-load` | host-load stands alone | -| `alerts-imports-bus-ports-core-only`, `not-alerts` | alerts imports the bus, ports and core only; only `main.ts` imports it | +| `alerts-imports-bus-ports-core-only`, `not-alerts` | alerts imports the bus, ports and core only; only `src/daemon/main.ts` imports it | | `not-stream`, `stream-not-edges` | Nothing required imports stream; stream imports no frontend, gateway or driver | | `gateway-not-edges`, `gateway-only-dispatch` | Gateway imports no driver or frontend, and from daemon only `src/daemon/dispatch.ts` | | `only-main-knows-gateway` | In daemon, only `src/daemon/main.ts` imports the gateway | diff --git a/src/alerts/alerts.test.ts b/src/alerts/alerts.test.ts index 2e1591814..c04f54b39 100644 --- a/src/alerts/alerts.test.ts +++ b/src/alerts/alerts.test.ts @@ -95,38 +95,33 @@ afterEach(() => { }); describe("alerts", () => { + const gatewayOwn = (h: Harness) => h.emit("worker.connected", connected()); + const relayed = (h: Harness) => + h.eventBus.republish({ + id: "evt_remote", + timestamp: 42, + event: "device.quarantine-abandoned", + payload: { deviceId: "dev_1", attempts: 3, workerId: "wrk_9" } as never, + module: "quarantine", + }); + it.each([ - ["a gateway-own worker.connected", "own"], - ["a republished device.quarantine-abandoned", "relayed"], + ["a gateway-own worker.connected", gatewayOwn, "wrk_1"], + ["a republished device.quarantine-abandoned", relayed, "wrk_9"], ])( "posts a listed event to a generic destination as the envelope with Content-Type, User-Agent, X-Simlock-Event, X-Simlock-Delivery and a signature that verifies with the secret, and logs Alert delivered at debug (%s)", - async (_name, kind) => { + async (_name, emitEnvelope, workerId) => { const h = harness([ destination({ events: ["worker.connected", "device.quarantine-abandoned"] }), ]); - const envelope = - kind === "own" - ? h.emit("worker.connected", connected()) - : h.eventBus.republish({ - id: "evt_remote", - timestamp: 42, - event: "device.quarantine-abandoned", - payload: { deviceId: "dev_1", attempts: 3, workerId: "wrk_9" } as never, - module: "quarantine", - }); + const envelope = emitEnvelope(h); await settle(); expect(h.poster.requests).toHaveLength(1); const [request] = h.poster.requests; expect(request?.url).toBe(destination().url); expect(request?.body).toBe(JSON.stringify(envelope)); - expect(JSON.parse(request?.body ?? "")).toMatchObject({ - id: envelope.id, - event: envelope.event, - payload: envelope.payload, - }); - if (kind === "relayed") - expect(JSON.parse(request?.body ?? "").payload.workerId).toBe("wrk_9"); + expect(JSON.parse(request?.body ?? "").payload.workerId).toBe(workerId); expect(request?.headers).toEqual({ "Content-Type": "application/json", "User-Agent": "simlock/9.9.9", diff --git a/src/alerts/index.ts b/src/alerts/index.ts index c0c5af0f1..a3e53fceb 100644 --- a/src/alerts/index.ts +++ b/src/alerts/index.ts @@ -17,18 +17,16 @@ export function createAlerts(options: { version: string; }): Alerts { const { destinations, eventBus, poster, clock, logger, version } = options; - const queues = destinations.map((destination) => ({ - destination, - queue: new DestinationQueue({ destination, poster, clock, logger, version }), - })); + const queues = destinations.map( + (destination) => new DestinationQueue({ destination, poster, clock, logger, version }), + ); const unsubscribes = ALERT_EVENTS.filter((name) => destinations.some((destination) => destination.events.includes(name)), ).map((name) => eventBus.subscribe(name, (envelope) => { - for (const { destination, queue } of queues) { - if (!destination.events.includes(name)) continue; - queue.enqueue(envelope); + for (const queue of queues) { + if (queue.destination.events.includes(name)) queue.enqueue(envelope); } }), ); @@ -37,9 +35,9 @@ export function createAlerts(options: { stop() { for (const unsubscribe of unsubscribes) unsubscribe(); const discarded: Record = {}; - for (const { destination, queue } of queues) { + for (const queue of queues) { const count = queue.stop(); - if (count > 0) discarded[destination.name] = count; + if (count > 0) discarded[queue.destination.name] = count; } if (Object.keys(discarded).length > 0) logger.info("Alerts discarded at stop", { discarded }); }, diff --git a/src/alerts/queue.ts b/src/alerts/queue.ts index 0669d04bb..5b961cc08 100644 --- a/src/alerts/queue.ts +++ b/src/alerts/queue.ts @@ -30,12 +30,14 @@ function errorCode(error: unknown): string { export class DestinationQueue { readonly #options: QueueOptions; + readonly destination: AlertDestination; readonly #waiting: EventEnvelope[] = []; #current: InFlight | undefined; #stopped = false; constructor(options: QueueOptions) { this.#options = options; + this.destination = options.destination; } enqueue(envelope: EventEnvelope): void { diff --git a/src/core/config.ts b/src/core/config.ts index 3aa40c446..9140b0f01 100644 --- a/src/core/config.ts +++ b/src/core/config.ts @@ -116,7 +116,7 @@ export const ALERT_FORMATS = ["generic"] as const; export type AlertFormatName = (typeof ALERT_FORMATS)[number]; /** why: checked at config load, which cannot import the alerts module. */ -export const ALERT_FORMAT_SECRET: Record = { +const ALERT_FORMAT_SECRET: Record = { generic: "required", }; @@ -147,53 +147,61 @@ function isAlertUrl(value: unknown): value is string { return url.protocol === "http:" && ALERT_LOCAL_HOSTS.includes(url.hostname); } +const ALERT_KEYS = ["name", "format", "url", "events", "secret"]; + +function checkAlertName(name: unknown, path: string, names: Set): string { + if (typeof name !== "string" || !ALERT_NAME_PATTERN.test(name)) { + throw invalidValue( + `${path}.name`, + "a name of lowercase letters, digits and dashes that starts with a letter, at most 32 characters", + ); + } + if (names.has(name)) throw invalidValue(`${path}.name`, "a name no other destination uses"); + return name; +} + +function checkAlertEvents(events: unknown, path: string): AlertEventName[] { + if ( + !Array.isArray(events) || + events.length === 0 || + new Set(events).size !== events.length || + !events.every((event) => (ALERT_EVENTS as readonly unknown[]).includes(event)) + ) { + throw invalidValue( + `${path}.events`, + `a non-empty list of alert events, each once: ${ALERT_EVENTS.join(", ")}`, + ); + } + return events as AlertEventName[]; +} + function alertDestinationValidator( names: Set, ): (value: unknown, path: string, warn: Warn) => AlertDestination { return (value, path, warn) => { const entry = requireObject(value, path); for (const key of Object.keys(entry)) { - if (!["name", "format", "url", "events", "secret"].includes(key)) { - warn(`Unknown config key: "${path}.${key}"`); - } + if (!ALERT_KEYS.includes(key)) warn(`Unknown config key: "${path}.${key}"`); } - const { name, format, url, events, secret } = entry; - if (typeof name !== "string" || !ALERT_NAME_PATTERN.test(name)) { - throw invalidValue( - `${path}.name`, - "a name of lowercase letters, digits and dashes that starts with a letter, at most 32 characters", - ); - } - if (names.has(name)) throw invalidValue(`${path}.name`, "a name no other destination uses"); - if (typeof format !== "string" || !(ALERT_FORMATS as readonly string[]).includes(format)) { - throw invalidValue( - `${path}.format`, - `one of ${ALERT_FORMATS.map((each) => `"${each}"`).join(", ")}`, - ); - } - if (!isAlertUrl(url)) throw invalidValue(`${path}.url`, ALERT_URL_EXPECTED); + // why: the fields are checked in this order, so a destination with several bad ones reports the first. + const name = checkAlertName(entry.name, path, names); + const format = stringUnion(ALERT_FORMATS)(entry.format, `${path}.format`); + if (!isAlertUrl(entry.url)) throw invalidValue(`${path}.url`, ALERT_URL_EXPECTED); + const events = checkAlertEvents(entry.events, path); + const { secret } = entry; if ( - !Array.isArray(events) || - events.length === 0 || - new Set(events).size !== events.length || - !events.every((event) => (ALERT_EVENTS as readonly unknown[]).includes(event)) + ALERT_FORMAT_SECRET[format] === "required" && + (typeof secret !== "string" || secret.length === 0) ) { - throw invalidValue( - `${path}.events`, - `a non-empty list of alert events, each once: ${ALERT_EVENTS.join(", ")}`, - ); - } - const secretRule = ALERT_FORMAT_SECRET[format as AlertFormatName]; - if (secretRule === "required" && (typeof secret !== "string" || secret.length === 0)) { throw invalidValue(`${path}.secret`, `a non-empty secret for a ${format} destination`); } names.add(name); return { name, - format: format as AlertFormatName, - url, - events: events as AlertEventName[], - ...(secret === undefined ? {} : { secret: secret as string }), + format, + url: entry.url, + events, + ...(typeof secret === "string" ? { secret } : {}), }; }; } diff --git a/src/core/index.ts b/src/core/index.ts index bd7c3ca14..a72177c05 100644 --- a/src/core/index.ts +++ b/src/core/index.ts @@ -1,9 +1,7 @@ export { ALERT_EVENTS, ALERT_FORMATS, - ALERT_FORMAT_SECRET, type AlertDestination, - type AlertEventName, type AlertFormatName, type Config, type ConfigOverrides, diff --git a/src/daemon/AGENTS.md b/src/daemon/AGENTS.md index c3b43684d..02992ca77 100644 --- a/src/daemon/AGENTS.md +++ b/src/daemon/AGENTS.md @@ -37,10 +37,9 @@ Both modes: config (`config.json`) → `Logger` → host info → `EventBus` → HTTP starts only when `http.enabled`. - **Gateway** — instance id → `TokenStore`, `AdminSecretManager` → `FleetLeaseIndex` (requester prefix `gw::`) → `GatewayService` (uplink listener, `workers.json`) - → `FleetLeaseCoordinator` → `createAlerts` when `gateway.alerts.destinations` is not - empty (before `start()`, so no worker joins first; stopped first in `stopAuxiliary`, before the - uplinks close) → `DaemonServer` with `GatewayDispatcher`. HTTP always - starts; the uplink upgrades on the same listener. + → `FleetLeaseCoordinator` → `createAlerts` when `gateway.alerts.destinations` is not empty + (before `start()`; stopped first in `stopAuxiliary`) → `DaemonServer` with `GatewayDispatcher`. + HTTP always starts; the uplink upgrades on the same listener. An HTTP bind failure after a good start stops the daemon (`http-start-failed`). diff --git a/src/daemon/gateway-fleet.e2e.test.ts b/src/daemon/gateway-fleet.e2e.test.ts index fff7772c6..a799b45f6 100644 --- a/src/daemon/gateway-fleet.e2e.test.ts +++ b/src/daemon/gateway-fleet.e2e.test.ts @@ -188,6 +188,7 @@ describe("gateway fleet smoke (ADR 0005 §35)", () => { return { daemon, filesystem, directory }; } + // fallow-ignore-next-line complexity -- one test harness builder; each option is a plain pass-through to the daemon. async function startWorker(options: { readonly label: string; readonly model: string; diff --git a/src/ports/index.ts b/src/ports/index.ts index d06cfb822..27e3c0fcf 100644 --- a/src/ports/index.ts +++ b/src/ports/index.ts @@ -125,4 +125,4 @@ export { SqliteUnavailableError, } from "./sqlite.js"; export type { SqliteDatabase, SqliteOpener, SqliteStatement } from "./sqlite.js"; -export { FetchWebhookPoster, type WebhookPoster, type WebhookRequest } from "./webhook-poster.js"; +export { FetchWebhookPoster, type WebhookPoster } from "./webhook-poster.js"; From e2bdeaeb380446587008c1a3f381c3a6f7ad53d6 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Tue, 20 Oct 2026 08:34:37 +0200 Subject: [PATCH 3/5] chore(alerts): name the queue calls fallow cannot trace (#719) --- src/alerts/queue.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/alerts/queue.ts b/src/alerts/queue.ts index 5b961cc08..c2e3714a6 100644 --- a/src/alerts/queue.ts +++ b/src/alerts/queue.ts @@ -40,6 +40,7 @@ export class DestinationQueue { this.destination = options.destination; } + // fallow-ignore-next-line unused-class-member -- called from createAlerts on array elements. enqueue(envelope: EventEnvelope): void { if (this.#stopped) return; if (this.#waiting.length >= MAX_WAITING) { @@ -50,6 +51,7 @@ export class DestinationQueue { this.#pump(); } + // fallow-ignore-next-line unused-class-member -- called from createAlerts on array elements. stop(): number { this.#stopped = true; const discarded = this.#waiting.length + (this.#current === undefined ? 0 : 1); From 0708345e07acc52b7c3426363e7d0e044c9341bd Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Tue, 20 Oct 2026 08:45:57 +0200 Subject: [PATCH 4/5] =?UTF-8?q?test(alerts):=20probe=20fixes=20=E2=80=94?= =?UTF-8?q?=20500=20boundary,=20per-destination=20filter,=20handler=20defe?= =?UTF-8?q?rs=20the=20post,=20poster=20request=20and=20redirect,=20passwor?= =?UTF-8?q?d-only=20URL=20(#719)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/alerts/alerts.test.ts | 19 +++++++++++++----- src/core/config.test.ts | 1 + src/ports/webhook-poster.test.ts | 34 ++++++++++++++++++++++++++++---- 3 files changed, 45 insertions(+), 9 deletions(-) diff --git a/src/alerts/alerts.test.ts b/src/alerts/alerts.test.ts index c04f54b39..4a039719f 100644 --- a/src/alerts/alerts.test.ts +++ b/src/alerts/alerts.test.ts @@ -62,7 +62,7 @@ class Harness { await settle(); } - failWith(kind: "refused" | "no-code" | 429 | 503): void { + failWith(kind: "refused" | "no-code" | 429 | 500 | 503): void { this.poster.respond = () => { if (kind === "refused") { return Promise.reject( @@ -159,6 +159,11 @@ describe("alerts", () => { const h = harness([ destination({ name: "one", url: "https://one.example.com/x", secret: "first" }), destination({ name: "two", url: "https://two.example.com/x", secret: "second" }), + destination({ + name: "other", + url: "https://other.example.com/x", + events: ["worker.disconnected"], + }), ]); h.emit("worker.connected", connected()); await settle(); @@ -191,6 +196,7 @@ describe("alerts", () => { field: { error: "network-error" }, }, { name: "429", kind: 429, endMs: 0, field: { status: 429 } }, + { name: "500", kind: 500, endMs: 0, field: { status: 500 } }, { name: "5xx", kind: 503, endMs: 0, field: { status: 503 } }, { name: "a timeout", kind: "timeout", endMs: 10_000, field: { error: "timeout" } }, ] as const; @@ -374,14 +380,17 @@ describe("alerts", () => { const h = harness(); const seen: string[] = []; h.eventBus.subscribe("worker.connected", (envelope) => seen.push(envelope.id)); + const first = h.emit("worker.connected", connected("wrk_a")); + expect(h.poster.requests).toHaveLength(0); + await settle(); + expect(h.poster.requests).toHaveLength(1); + expect(seen).toEqual([first.id]); + h.poster.post = () => { throw new Error(`boom ${URL_TOKEN}`); }; - const envelope = h.emit("worker.connected", connected()); - - expect(h.poster.requests).toHaveLength(0); - expect(seen).toEqual([envelope.id]); + expect(seen).toEqual([first.id, envelope.id]); await settle(); await h.advance(1_000); await h.advance(5_000); diff --git a/src/core/config.test.ts b/src/core/config.test.ts index 8327c4079..e46ff9532 100644 --- a/src/core/config.test.ts +++ b/src/core/config.test.ts @@ -1995,6 +1995,7 @@ describe("gateway.alerts.destinations", () => { ["url", bad("url", "n8n.example.com/x"), urlExpected], ["url", bad("url", "https://user:pw@host/x"), urlExpected], ["url", bad("url", "https://user@host/x"), urlExpected], + ["url", bad("url", "https://:pw@host/x"), urlExpected], ["url", bad("url", ""), urlExpected], ["url", without("url"), urlExpected], ["events", bad("events", []), eventsExpected], diff --git a/src/ports/webhook-poster.test.ts b/src/ports/webhook-poster.test.ts index 5a0b7c350..83971d3ac 100644 --- a/src/ports/webhook-poster.test.ts +++ b/src/ports/webhook-poster.test.ts @@ -18,8 +18,26 @@ afterEach(async () => { ); }); -async function listen(handler: (url: string, response: ServerResponse) => void): Promise { - const server = createServer((request, response) => handler(request.url ?? "", response)); +interface Seen { + method: string | undefined; + header: string | string[] | undefined; + body: string; +} + +async function listen( + handler: (url: string, response: ServerResponse, seen: Seen) => void, +): Promise { + const server = createServer((request, response) => { + const chunks: Buffer[] = []; + request.on("data", (chunk: Buffer) => chunks.push(chunk)); + request.on("end", () => + handler(request.url ?? "", response, { + method: request.method, + header: request.headers["x-a"], + body: Buffer.concat(chunks).toString("utf8"), + }), + ); + }); servers.push(server); await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); return `http://127.0.0.1:${(server.address() as AddressInfo).port}`; @@ -32,8 +50,14 @@ describe("FetchWebhookPoster", () => { const closed = new Promise((resolve) => { bodyClosed = resolve; }); - const base = await listen((url, response) => { + const sent: Seen[] = []; + const base = await listen((url, response, seen) => { hits.push(url); + sent.push(seen); + if (url === "/moved-done") { + response.writeHead(301, { Location: "/target" }).end("moved"); + return; + } if (url === "/moved") { response.writeHead(301, { Location: "/target" }); response.write("x".repeat(1024)); @@ -49,9 +73,11 @@ describe("FetchWebhookPoster", () => { poster.post({ url: `${base}${path}`, headers: { "X-A": "1" }, body: "{}", signal }); await expect(post("/ok")).resolves.toEqual({ status: 200 }); + expect(sent[0]).toEqual({ method: "POST", header: "1", body: "{}" }); + await expect(post("/moved-done")).resolves.toEqual({ status: 301 }); await expect(post("/moved")).resolves.toEqual({ status: 301 }); await closed; - expect(hits).toEqual(["/ok", "/moved"]); + expect(hits).toEqual(["/ok", "/moved-done", "/moved"]); const pending = post("/silent"); controller.abort(); From d537522a2f7d944873d5220b04abce09b288aaa2 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Tue, 20 Oct 2026 09:00:28 +0200 Subject: [PATCH 5/5] test(alerts): log test drives delivered and build-failed, stop test covers same-tick enqueue, drop extra registry test (#719) --- src/alerts/alerts.test.ts | 18 +++++++++++++++++- src/alerts/formats/conformance.test.ts | 8 +------- 2 files changed, 18 insertions(+), 8 deletions(-) diff --git a/src/alerts/alerts.test.ts b/src/alerts/alerts.test.ts index 4a039719f..149e092b8 100644 --- a/src/alerts/alerts.test.ts +++ b/src/alerts/alerts.test.ts @@ -361,6 +361,12 @@ describe("alerts", () => { await h.advance(60_000); expect(h.poster.requests).toHaveLength(1); expect(h.sink.records).toHaveLength(lines); + + const same = harness(); + same.emit("worker.connected", connected("wrk_e")); + same.alerts.stop(); + await same.advance(60_000); + expect(same.poster.requests).toHaveLength(0); }); it("stop counts a message waiting between tries as discarded and cancels its retry", async () => { @@ -453,13 +459,23 @@ describe("alerts", () => { await settle(); await h.advance(1_000); await h.advance(5_000); + h.poster.respond = () => Promise.resolve({ status: 200 }); + h.emit("worker.connected", connected()); + await settle(); + vi.spyOn(ALERT_FORMAT_REGISTRY.generic, "build").mockImplementationOnce(() => { + throw new Error(`cannot build ${destination().url} s3cr3t`); + }); + h.emit("worker.connected", connected()); + await settle(); + h.poster.respond = () => new Promise(() => {}); for (let index = 0; index < 102; index += 1) { h.emit("worker.disconnected", { workerId: `wrk_${index}`, leaseCount: 0 }); } h.alerts.stop(); const text = h.sink.records.map((record) => JSON.stringify(record)).join("\n"); - expect(h.logs("Alert send gave up")).toHaveLength(1); + expect(h.logs("Alert delivered")).toHaveLength(1); + expect(h.logs("Alert send gave up")).toHaveLength(2); expect(h.logs("Alert dropped: queue full")).toHaveLength(1); expect(h.logs("Alerts discarded at stop")).toHaveLength(1); expect(text).not.toContain(URL_TOKEN); diff --git a/src/alerts/formats/conformance.test.ts b/src/alerts/formats/conformance.test.ts index 821f8499c..c76d78388 100644 --- a/src/alerts/formats/conformance.test.ts +++ b/src/alerts/formats/conformance.test.ts @@ -1,7 +1,7 @@ import { describe, expect, it } from "vitest"; import type { EventEnvelope } from "../../bus/index.js"; -import { ALERT_EVENTS, ALERT_FORMATS } from "../../core/index.js"; +import { ALERT_EVENTS } from "../../core/index.js"; import { ALERT_FORMAT_REGISTRY } from "./index.js"; describe.each(Object.entries(ALERT_FORMAT_REGISTRY))("format %s", (name, format) => { @@ -36,9 +36,3 @@ describe.each(Object.entries(ALERT_FORMAT_REGISTRY))("format %s", (name, format) } }); }); - -describe("ALERT_FORMAT_REGISTRY", () => { - it("has exactly the formats core names", () => { - expect(Object.keys(ALERT_FORMAT_REGISTRY).sort()).toEqual([...ALERT_FORMATS].sort()); - }); -});