From 93b3e5a327dc758ccdbd9cb11728169e4c4f14fb Mon Sep 17 00:00:00 2001 From: Danyang Chen Date: Thu, 22 Oct 2026 15:34:24 +0800 Subject: [PATCH] Add a Slack-shaped delivery adapter with retry and dedupe (#355) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A reminder has to reach the owner with the desktop app closed, so its delivery cannot be a client toast. The adapter keeps Slack's chat.postMessage shape with the transport injected, and owns the policy around it: one key per reminder so the same reminder sent again is not a second message, exponential backoff that honours Retry-After, three attempts at most, and no retry for a structural failure. The fake transport and fake timers keep the tests offline. Slack's own transient codes — internal_error, fatal_error, service_unavailable and request_timeout — pass on their own, so they join network and rate_limited on the retry budget: one of them no longer ends the delivery and is not remembered as a refusal. Everything else stays structural. Rebased onto current main; the only conflict was the core-host.md section that #645 also touched, resolved by keeping both. --- core/src/host/notify.js | 151 ++++++++++++++++++++ core/test/notify.test.js | 171 +++++++++++++++++++++++ docs/subsystems/core-host.md | 10 ++ patches/ds-bot/files/src/host/notify.js | 151 ++++++++++++++++++++ patches/ds-bot/files/test/notify.test.js | 171 +++++++++++++++++++++++ 5 files changed, 654 insertions(+) create mode 100644 core/src/host/notify.js create mode 100644 core/test/notify.test.js create mode 100644 patches/ds-bot/files/src/host/notify.js create mode 100644 patches/ds-bot/files/test/notify.test.js diff --git a/core/src/host/notify.js b/core/src/host/notify.js new file mode 100644 index 000000000..6784f77d7 --- /dev/null +++ b/core/src/host/notify.js @@ -0,0 +1,151 @@ +// Out-of-app delivery, first half (#355): the adapter that will carry a reminder to Slack. +// A reminder has to reach the owner when the desktop app is closed, so delivery cannot be a +// client toast. The transport is injected; what this module owns is the policy around it: one +// idempotency key per reminder, so a retried send cannot arrive twice; exponential backoff that +// honours `Retry-After`; three attempts at most; no retry at all for a failure that will not +// change (the channel is gone, the app was removed). A channel switched off makes zero transport +// calls, including one that is switched off while it waits to retry. +// +// The real Slack transport waits for the owner's credential, so it is not here: the module speaks +// Slack's `chat.postMessage` shape and ships a fake transport that records every call and scripts +// each answer, which is what makes the retry and dedupe tests possible without a network or a key +// (core/test/notify.test.js). Nothing reads this module yet either — `schedules.js` wiring is the +// second half, once a real channel exists. + +export const MAX_ATTEMPTS = 3 +// 500 ms, 1 s, 2 s, ... and then a ceiling, so a flaky link retries quickly and a rate limit +// waits as long as the server asked. +export const BASE_DELAY_MS = 500 +export const MAX_DELAY_MS = 30_000 +// Reminders remembered after they were sent, so a second call for the same one is not a second +// message. A long-lived host drops the oldest rather than grow without bound. +const SENT_KEPT = 1_000 + +const invalid = message => Object.assign(new Error(message), { code: 'notify/invalid' }) + +// Failures that can pass on their own: the link, the rate limit, and Slack's own transient codes. +// Slack answers `internal_error` and `fatal_error` for its own trouble and `service_unavailable` / +// `request_timeout` while it is restarting or slow, so one of those must not cost the owner a +// reminder. Everything else is structural (`channel_not_found`, `not_in_channel`, an argument Slack +// will keep refusing, a shape we do not know): the same call would fail the same way, and spending +// two more attempts buys nothing. A new transient code belongs in this set. +const RETRYABLE = new Set(['network', 'rate_limited', 'internal_error', 'fatal_error', 'service_unavailable', 'request_timeout']) + +// Slack answers `{ ok: true, ts }`, or `{ ok: false, error }` with `retryAfterMs` for a 429. A +// transport may also throw instead of answering `network`; both count as a broken link. +function readAnswer(answer) { + if (answer && answer.ok === true && typeof answer.ts === 'string' && answer.ts) return { ok: true, ts: answer.ts } + const error = answer && answer.ok === false ? String(answer.error ?? '') : '' + if (error === 'rate_limited') { + const asked = Number(answer.retryAfterMs) + return { ok: false, error, retryAfterMs: Number.isFinite(asked) && asked > 0 ? asked : 0, permanent: false } + } + if (error === '') return { ok: false, error: 'network', permanent: false } + if (RETRYABLE.has(error)) return { ok: false, error, permanent: false } + return { ok: false, error, permanent: true } +} + +// A Slack stand-in for tests. Every call is recorded in `calls`; answers come from `script`, each +// used once, and the last one repeats — so `[rateLimited(2000), { ok: true }]` is one retry and +// then success, and a script of one answer describes every call. An item may be a function of +// (message, attempt), and an `Error` instance is thrown, which is how a broken link looks. +export function fakeSlack(script = [{ ok: true }]) { + const calls = [] + const queue = Array.isArray(script) ? [...script] : [script] + return { + calls, + async postMessage(message) { + calls.push(message) + const scripted = queue.length > 1 ? queue.shift() : queue[0] ?? { ok: true } + const answer = typeof scripted === 'function' ? await scripted(message, calls.length) : scripted + if (answer instanceof Error) throw answer + if (!answer || answer.ok !== true) return { ...answer } + return { ...answer, ts: answer.ts ?? `ts-${calls.length}` } + }, + } +} + +// The three failures as a transport can spell them, so a test or a later Slack transport does not +// have to re-type the strings. +export const rateLimited = retryAfterMs => ({ ok: false, error: 'rate_limited', retryAfterMs }) +export const channelNotFound = () => ({ ok: false, error: 'channel_not_found' }) +export const networkFailure = () => ({ ok: false, error: 'network' }) + +export function createNotifier(options = {}) { + const { transport, channel } = options + const attempts = options.attempts ?? MAX_ATTEMPTS + const baseDelayMs = options.baseDelayMs ?? BASE_DELAY_MS + const maxDelayMs = options.maxDelayMs ?? MAX_DELAY_MS + const log = options.log ?? (() => {}) + const delay = options.delay ?? (ms => new Promise(resolve => setTimeout(resolve, ms))) + if (!transport || typeof transport.postMessage !== 'function') throw invalid('a transport with postMessage() is required') + if (typeof channel !== 'string' || !channel.trim()) throw invalid('a channel is required') + if (!Number.isInteger(attempts) || attempts < 1) throw invalid('attempts must be a positive integer') + + let enabled = options.enabled !== false + const sent = new Map() + const refused = new Map() + const inFlight = new Map() + const backoff = attempt => Math.min(baseDelayMs * 2 ** (attempt - 1), maxDelayMs) + + async function attemptOnce(message) { + let answer + try { + answer = readAnswer(await transport.postMessage(message)) + } catch (error) { + answer = readAnswer({ ok: false, error: 'network', cause: error }) + } + return answer + } + + async function run(key, message) { + for (let attempt = 1; attempt <= attempts; attempt += 1) { + // Checked before every call, not only at the start: switching the channel off stops a + // reminder that is already waiting out a backoff. + if (!enabled) return { state: 'skipped', attempts: attempt - 1, reason: 'channel-off' } + const answer = await attemptOnce(message) + if (answer.ok) { + sent.set(key, answer.ts) + while (sent.size > SENT_KEPT) sent.delete(sent.keys().next().value) + log('notify: sent %s to %s on attempt %d', key, channel, attempt) + return { state: 'sent', ts: answer.ts, attempts: attempt } + } + if (answer.permanent) { + refused.set(key, answer.error) + log('notify: %s cannot reach %s: %s', key, channel, answer.error) + return { state: 'failed', error: answer.error, attempts: attempt, permanent: true } + } + if (attempt === attempts) { + log('notify: %s not delivered to %s after %d attempts: %s', key, channel, attempt, answer.error) + return { state: 'failed', error: answer.error, attempts: attempt, permanent: false } + } + // `Retry-After` is a floor, never a ceiling: if the server asks for longer than our own + // backoff, waiting less only earns another 429. + const waitMs = Math.max(backoff(attempt), answer.retryAfterMs ?? 0) + log('notify: %s attempt %d failed (%s); retrying in %d ms', key, attempt, answer.error, waitMs) + await delay(waitMs) + } + return { state: 'failed', error: 'network', attempts, permanent: false } + } + + return { + get enabled() { return enabled }, + setEnabled(flag) { enabled = flag !== false }, + // At-least-once, deduplicated per `key`: the same reminder handed over again after it was sent + // is answered with the first result and never reaches the transport a second time; one handed + // over again while it is in flight joins that attempt instead of starting another. A reminder + // that gave up after its three attempts is not remembered, so a later run may try again. + send({ key, text, blocks, threadTs } = {}) { + if (typeof key !== 'string' || !key.trim()) throw invalid('send() needs a non-empty key') + if ((typeof text !== 'string' || !text.trim()) && !Array.isArray(blocks)) throw invalid('send() needs text or blocks') + if (!enabled) return Promise.resolve({ state: 'skipped', attempts: 0, reason: 'channel-off' }) + if (sent.has(key)) return Promise.resolve({ state: 'duplicate', ts: sent.get(key), attempts: 0 }) + if (refused.has(key)) return Promise.resolve({ state: 'failed', error: refused.get(key), attempts: 0, permanent: true }) + if (inFlight.has(key)) return inFlight.get(key) + const message = { channel, text, ...(blocks ? { blocks } : {}), ...(threadTs ? { threadTs } : {}) } + const started = run(key, message).finally(() => inFlight.delete(key)) + inFlight.set(key, started) + return started + }, + } +} diff --git a/core/test/notify.test.js b/core/test/notify.test.js new file mode 100644 index 000000000..f031d9427 --- /dev/null +++ b/core/test/notify.test.js @@ -0,0 +1,171 @@ +import assert from 'node:assert/strict' +import { test } from 'node:test' +import { createNotifier, fakeSlack, rateLimited, channelNotFound, networkFailure } from '../src/host/notify.js' + +// Every retry below is driven by fake timers: a test never sleeps in real time. `settle` lets the +// adapter reach its next await (transport call or backoff) before the clock is moved. +async function settle() { + for (let i = 0; i < 3; i += 1) await new Promise(resolve => setImmediate(resolve)) +} + +test('#355: a reminder reaches the transport in Slack chat.postMessage shape', async () => { + const transport = fakeSlack([{ ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123' }) + + const first = await notifier.send({ key: 'r1', text: 'Standup in 10 minutes' }) + assert.deepEqual(first, { state: 'sent', ts: 'ts-1', attempts: 1 }) + // Optional fields are passed only when given, so a plain reminder is a plain message. + assert.deepEqual(transport.calls, [{ channel: 'C0123', text: 'Standup in 10 minutes' }]) + + await notifier.send({ key: 'r2', text: 'Still 10 minutes', blocks: [{ type: 'section' }], threadTs: 'ts-1' }) + assert.deepEqual(transport.calls[1], { channel: 'C0123', text: 'Still 10 minutes', blocks: [{ type: 'section' }], threadTs: 'ts-1' }) +}) + +test('#355: the same reminder handed over twice is one message', async () => { + const transport = fakeSlack([{ ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123' }) + + assert.equal((await notifier.send({ key: 'r1', text: 'Tomorrow: ship' })).state, 'sent') + const again = await notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + assert.deepEqual(again, { state: 'duplicate', ts: 'ts-1', attempts: 0 }) + assert.equal(transport.calls.length, 1, 'a repeated reminder is not a second transport call') +}) + +test('#355: a second send for a reminder already in flight joins it instead of starting another', async () => { + let release + const transport = fakeSlack([() => new Promise(resolve => { release = () => resolve({ ok: true, ts: 'ts-a' }) })]) + const notifier = createNotifier({ transport, channel: 'C0123' }) + + const first = notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + await settle() + const second = notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + release() + const [a, b] = await Promise.all([first, second]) + assert.equal(transport.calls.length, 1) + assert.equal(a.ts, 'ts-a') + assert.equal(b.ts, 'ts-a') +}) + +test('#355: a rate limit waits what Retry-After asks, then sends once', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }) + const transport = fakeSlack([rateLimited(2_000), { ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123' }) + + const pending = notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + await settle() + assert.equal(transport.calls.length, 1) + t.mock.timers.tick(1_999) + await settle() + assert.equal(transport.calls.length, 1, 'Retry-After is a floor, not a suggestion') + t.mock.timers.tick(1) + await settle() + assert.deepEqual(await pending, { state: 'sent', ts: 'ts-2', attempts: 2 }) + assert.equal(transport.calls.length, 2) +}) + +test('#355: a broken link backs off further on each attempt', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }) + const waits = [] + const transport = fakeSlack([networkFailure(), networkFailure(), { ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123', delay: async ms => { waits.push(ms) } }) + + const result = await notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + assert.deepEqual(waits, [500, 1_000]) + assert.deepEqual(result, { state: 'sent', ts: 'ts-3', attempts: 3 }) +}) + +test('#355: a thrown transport error is a broken link, not a reason to give up', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }) + const transport = fakeSlack([new Error('socket hang up'), { ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123' }) + + const pending = notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + await settle() + t.mock.timers.tick(500) + await settle() + assert.deepEqual(await pending, { state: 'sent', ts: 'ts-2', attempts: 2 }) +}) + +test('#355: three failed attempts end the delivery, and a later run may try again', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }) + const waits = [] + const transport = fakeSlack([networkFailure(), networkFailure(), networkFailure(), { ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123', delay: async ms => { waits.push(ms) } }) + + const first = await notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + assert.deepEqual(first, { state: 'failed', error: 'network', attempts: 3, permanent: false }) + assert.deepEqual(waits, [500, 1_000], 'three attempts, two waits: the budget is not exceeded') + assert.equal(transport.calls.length, 3) + + // Exhausting the attempts is not a verdict about the channel, so the reminder is not blacklisted. + const second = await notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + assert.equal(second.state, 'sent') + assert.equal(transport.calls.length, 4) +}) + +test('#355: Slack\'s own transient codes are retried, not remembered as permanent', async () => { + // Slack answers these for its own trouble (an internal error, a restart, a slow request). One of + // them must not cost the owner a reminder: each is retried on the same budget as a broken link. + for (const code of ['internal_error', 'fatal_error', 'service_unavailable', 'request_timeout']) { + const waits = [] + const transport = fakeSlack([{ ok: false, error: code }, { ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123', delay: async ms => { waits.push(ms) } }) + assert.deepEqual(await notifier.send({ key: `r-${code}`, text: 'Tomorrow: ship' }), { state: 'sent', ts: 'ts-2', attempts: 2 }, `${code} is worth another attempt`) + assert.deepEqual(waits, [500], `${code} backs off like any transient failure`) + assert.equal(transport.calls.length, 2) + } + // And running out of attempts is not a verdict about the channel: the reminder is not remembered + // as permanently refused, so a later run may still deliver it. + const transport = fakeSlack([{ ok: false, error: 'service_unavailable' }, { ok: false, error: 'service_unavailable' }, { ok: false, error: 'service_unavailable' }, { ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123', delay: async () => {} }) + assert.deepEqual(await notifier.send({ key: 'r-retry', text: 'Tomorrow: ship' }), { state: 'failed', error: 'service_unavailable', attempts: 3, permanent: false }) + assert.equal((await notifier.send({ key: 'r-retry', text: 'Tomorrow: ship' })).state, 'sent', 'a transient failure is not remembered as a refusal') +}) + +test('#355: a missing channel is a permanent failure and is not retried', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }) + const transport = fakeSlack([channelNotFound(), { ok: true }]) + const notifier = createNotifier({ transport, channel: 'C-gone' }) + + const first = await notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + assert.deepEqual(first, { state: 'failed', error: 'channel_not_found', attempts: 1, permanent: true }) + assert.equal(transport.calls.length, 1) + + const second = await notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + assert.deepEqual(second, { state: 'failed', error: 'channel_not_found', attempts: 0, permanent: true }) + assert.equal(transport.calls.length, 1, 'a structural failure stays answered without another call') +}) + +test('#355: a channel that is off makes no transport call at all', async () => { + const transport = fakeSlack([{ ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123', enabled: false }) + + assert.deepEqual(await notifier.send({ key: 'r1', text: 'Tomorrow: ship' }), { state: 'skipped', attempts: 0, reason: 'channel-off' }) + assert.equal(transport.calls.length, 0) +}) + +test('#355: switching a channel off stops a reminder that is waiting to retry', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }) + const transport = fakeSlack([rateLimited(5_000), { ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123' }) + + const pending = notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + await settle() + assert.equal(transport.calls.length, 1) + notifier.setEnabled(false) + t.mock.timers.tick(5_000) + await settle() + assert.deepEqual(await pending, { state: 'skipped', attempts: 1, reason: 'channel-off' }) + assert.equal(transport.calls.length, 1, 'the retry never leaves after the switch went off') + assert.equal(notifier.enabled, false) +}) + +test('#355: a reminder needs a key and something to say', async () => { + const notifier = createNotifier({ transport: fakeSlack(), channel: 'C0123' }) + // A malformed reminder is refused before anything is queued, so the caller hears it at once. + assert.throws(() => notifier.send({ text: 'no key' }), { code: 'notify/invalid' }) + assert.throws(() => notifier.send({ key: 'r1' }), { code: 'notify/invalid' }) + assert.throws(() => notifier.send({ key: 'r2', text: ' ' }), { code: 'notify/invalid' }) + assert.throws(() => createNotifier({ transport: fakeSlack() }), { code: 'notify/invalid' }) + assert.throws(() => createNotifier({ channel: 'C0123' }), { code: 'notify/invalid' }) +}) diff --git a/docs/subsystems/core-host.md b/docs/subsystems/core-host.md index f60907267..e38db9f1d 100644 --- a/docs/subsystems/core-host.md +++ b/docs/subsystems/core-host.md @@ -44,6 +44,7 @@ The server side of the patched `ds-bot` core (`core/`): the part of [ds-bot-core | `core/src/host/reach.js` | Per-Bot grants and the owner-policy `reach()` check that hooks.js runs outside the plugin chain (#90); `rt.setGrants`, `rt.hasGrant`, `rt.hasAnyGrant`. A local Bot refused toward a friend's Bot is told that it is not shared with that friend and where to share it (Sharing, #273). | | `core/src/host/schedules.js` | Scheduled tasks through DSH's schedule service: list, create, edit, pause/resume, retry and delete for Settings → Scheduled tasks, each task's last run, and moving a Bot's tasks to its fresh part (#354, #473, see **Scheduled tasks: edit, pause, missed runs** and **Scheduled tasks: last run and Retry**). Also the `bot_schedules` tool: the Bot-wide list DSH's own `schedule_list` cannot give, group chats' own Sessions included (#628, see **Scheduled tasks: the Bot-wide list**). Tests: `core/test/schedules.test.js`. | | `core/src/host/automation-default.js` | Automation tasks on by default (#378): once per profile, `offerBundle` asks DSH's plugin manager to switch on `@deepseek-ai/dsh-experimental-schedule-bundle` and writes the offer to `dsh-defaults.json` next to the team. Tests: `core/test/automation-default.test.js`. | +| `core/src/host/notify.js` | Out-of-app delivery, first half (#355): a transport-agnostic adapter with Slack's `chat.postMessage` shape, one key per reminder so a retried send cannot duplicate, exponential backoff that honours `Retry-After`, three attempts at most, a structural failure not retried, and zero transport calls while the channel is off. Ships `fakeSlack` for tests; nothing calls it until the real Slack transport lands (see **Out-of-app delivery, first half**). Tests: `core/test/notify.test.js`. | | `core/src/shared/remote-label.js` | Owner-badge labels for a friend's Bot and “Shared room from …” titles. `friendBotTitle(bot)` / `friendBotOwner(bot)` name a friend's chat from the roster (`Chief (SelfB1233)`) and its owner (`SelfB1233`, the person an approval card names) with one formatter for every client surface (#498). `remoteBotName` / `isAutoOwnerName` / `seatOwnerName` keep group and face-room labels in that format too (#526): Bot Social's ` · ` member label is not a Bot name, and an owner name made up from an install id (`FlyBot 682b8a43`) gives way to the username in the home handle (`Reviewer (flybot682b8a43)`). `isListedRemoteChat` / `listedChatBots` keep gone, hidden, and home-level (`@something.home`) remotes out of the sidebar, the **发给** picker, mentions, and `list_bots` friends lines; history stays on disk. | ## Invariants @@ -188,6 +189,15 @@ Work a member hands a Bot with `message_bot` from its group Session runs in a hi - A line arriving while a turn runs is not held: the window flushes what it holds first (so the queued items stay in arrival order) and the line lands as its own queued item, as before. A Bot-to-Bot `deliver` never waits behind the window — it has its own queue and dedupe (#641). - A group room is not collected: its line must reach the Session at once so a running round can drain it (`group_relay`, #299), so `settleLine` is only reached when `rt.roomOf(sessionId)` is undefined. The pending timer is cleared on dispose. Covered by `core/test/settle.test.js`. +### Out-of-app delivery, first half (#355) + +- A reminder has to reach the owner with the desktop app closed, so its delivery cannot be a client toast. `src/host/notify.js` is the adapter that half needs, and nothing calls it yet: `schedules.js` wiring is the second half, once a real channel exists. +- `createNotifier({ transport, channel })` speaks Slack's `chat.postMessage` shape — `postMessage({ channel, text, blocks?, threadTs? })` → `{ ok: true, ts }`, or a typed `{ ok: false, error }` — with the transport injected. The production Slack transport waits for the owner's credential and is not in this repo; `fakeSlack(script)` is the stand-in the tests run on, recording every call and scripting each answer (a function or an `Error` too), and `rateLimited(ms)` / `channelNotFound()` / `networkFailure()` spell the three failures. +- Delivery policy: one key per reminder, so the same reminder handed to `send()` again after it went out answers `duplicate` with the first `ts` and never reaches the transport, and one handed over while it is in flight joins that attempt. A retry waits `min(base · 2^(n-1), max)` — 500 ms then 1 s by default — and `retryAfterMs` from a 429 is a floor, never a ceiling: waiting less only earns another 429. Three attempts at most; running out is not a verdict about the channel, so a later call may try again. +- A structural failure (`channel_not_found`, `not_in_channel`, anything untyped) is permanent: one call, `permanent: true`, and a second `send()` for that key is answered without another call. A thrown error and `network` are the same broken link: retryable. +- `enabled` is checked before every attempt, not only at the start, so switching a channel off stops a reminder that is already waiting out a backoff, and a channel that is off makes zero transport calls. Sent keys are capped at 1 000 process-local entries; nothing is persisted, so a restart forgets them and at-least-once survives a crash. +- `core/test/notify.test.js` drives every retry with fake timers, needs no network and no key. + ### Diagnostics - Host diagnostics never depend on `ctx.logger`: DSH may export no logger, and a test or plugin-local host may carry one without `warn`. `rt.warn` still writes stderr and the diagnostics log in that case (`src/host/diagnostics.js`). Plugins reach the same log through `flybot.warn` (#68); see [plugin-runtime](plugin-runtime.md). diff --git a/patches/ds-bot/files/src/host/notify.js b/patches/ds-bot/files/src/host/notify.js new file mode 100644 index 000000000..6784f77d7 --- /dev/null +++ b/patches/ds-bot/files/src/host/notify.js @@ -0,0 +1,151 @@ +// Out-of-app delivery, first half (#355): the adapter that will carry a reminder to Slack. +// A reminder has to reach the owner when the desktop app is closed, so delivery cannot be a +// client toast. The transport is injected; what this module owns is the policy around it: one +// idempotency key per reminder, so a retried send cannot arrive twice; exponential backoff that +// honours `Retry-After`; three attempts at most; no retry at all for a failure that will not +// change (the channel is gone, the app was removed). A channel switched off makes zero transport +// calls, including one that is switched off while it waits to retry. +// +// The real Slack transport waits for the owner's credential, so it is not here: the module speaks +// Slack's `chat.postMessage` shape and ships a fake transport that records every call and scripts +// each answer, which is what makes the retry and dedupe tests possible without a network or a key +// (core/test/notify.test.js). Nothing reads this module yet either — `schedules.js` wiring is the +// second half, once a real channel exists. + +export const MAX_ATTEMPTS = 3 +// 500 ms, 1 s, 2 s, ... and then a ceiling, so a flaky link retries quickly and a rate limit +// waits as long as the server asked. +export const BASE_DELAY_MS = 500 +export const MAX_DELAY_MS = 30_000 +// Reminders remembered after they were sent, so a second call for the same one is not a second +// message. A long-lived host drops the oldest rather than grow without bound. +const SENT_KEPT = 1_000 + +const invalid = message => Object.assign(new Error(message), { code: 'notify/invalid' }) + +// Failures that can pass on their own: the link, the rate limit, and Slack's own transient codes. +// Slack answers `internal_error` and `fatal_error` for its own trouble and `service_unavailable` / +// `request_timeout` while it is restarting or slow, so one of those must not cost the owner a +// reminder. Everything else is structural (`channel_not_found`, `not_in_channel`, an argument Slack +// will keep refusing, a shape we do not know): the same call would fail the same way, and spending +// two more attempts buys nothing. A new transient code belongs in this set. +const RETRYABLE = new Set(['network', 'rate_limited', 'internal_error', 'fatal_error', 'service_unavailable', 'request_timeout']) + +// Slack answers `{ ok: true, ts }`, or `{ ok: false, error }` with `retryAfterMs` for a 429. A +// transport may also throw instead of answering `network`; both count as a broken link. +function readAnswer(answer) { + if (answer && answer.ok === true && typeof answer.ts === 'string' && answer.ts) return { ok: true, ts: answer.ts } + const error = answer && answer.ok === false ? String(answer.error ?? '') : '' + if (error === 'rate_limited') { + const asked = Number(answer.retryAfterMs) + return { ok: false, error, retryAfterMs: Number.isFinite(asked) && asked > 0 ? asked : 0, permanent: false } + } + if (error === '') return { ok: false, error: 'network', permanent: false } + if (RETRYABLE.has(error)) return { ok: false, error, permanent: false } + return { ok: false, error, permanent: true } +} + +// A Slack stand-in for tests. Every call is recorded in `calls`; answers come from `script`, each +// used once, and the last one repeats — so `[rateLimited(2000), { ok: true }]` is one retry and +// then success, and a script of one answer describes every call. An item may be a function of +// (message, attempt), and an `Error` instance is thrown, which is how a broken link looks. +export function fakeSlack(script = [{ ok: true }]) { + const calls = [] + const queue = Array.isArray(script) ? [...script] : [script] + return { + calls, + async postMessage(message) { + calls.push(message) + const scripted = queue.length > 1 ? queue.shift() : queue[0] ?? { ok: true } + const answer = typeof scripted === 'function' ? await scripted(message, calls.length) : scripted + if (answer instanceof Error) throw answer + if (!answer || answer.ok !== true) return { ...answer } + return { ...answer, ts: answer.ts ?? `ts-${calls.length}` } + }, + } +} + +// The three failures as a transport can spell them, so a test or a later Slack transport does not +// have to re-type the strings. +export const rateLimited = retryAfterMs => ({ ok: false, error: 'rate_limited', retryAfterMs }) +export const channelNotFound = () => ({ ok: false, error: 'channel_not_found' }) +export const networkFailure = () => ({ ok: false, error: 'network' }) + +export function createNotifier(options = {}) { + const { transport, channel } = options + const attempts = options.attempts ?? MAX_ATTEMPTS + const baseDelayMs = options.baseDelayMs ?? BASE_DELAY_MS + const maxDelayMs = options.maxDelayMs ?? MAX_DELAY_MS + const log = options.log ?? (() => {}) + const delay = options.delay ?? (ms => new Promise(resolve => setTimeout(resolve, ms))) + if (!transport || typeof transport.postMessage !== 'function') throw invalid('a transport with postMessage() is required') + if (typeof channel !== 'string' || !channel.trim()) throw invalid('a channel is required') + if (!Number.isInteger(attempts) || attempts < 1) throw invalid('attempts must be a positive integer') + + let enabled = options.enabled !== false + const sent = new Map() + const refused = new Map() + const inFlight = new Map() + const backoff = attempt => Math.min(baseDelayMs * 2 ** (attempt - 1), maxDelayMs) + + async function attemptOnce(message) { + let answer + try { + answer = readAnswer(await transport.postMessage(message)) + } catch (error) { + answer = readAnswer({ ok: false, error: 'network', cause: error }) + } + return answer + } + + async function run(key, message) { + for (let attempt = 1; attempt <= attempts; attempt += 1) { + // Checked before every call, not only at the start: switching the channel off stops a + // reminder that is already waiting out a backoff. + if (!enabled) return { state: 'skipped', attempts: attempt - 1, reason: 'channel-off' } + const answer = await attemptOnce(message) + if (answer.ok) { + sent.set(key, answer.ts) + while (sent.size > SENT_KEPT) sent.delete(sent.keys().next().value) + log('notify: sent %s to %s on attempt %d', key, channel, attempt) + return { state: 'sent', ts: answer.ts, attempts: attempt } + } + if (answer.permanent) { + refused.set(key, answer.error) + log('notify: %s cannot reach %s: %s', key, channel, answer.error) + return { state: 'failed', error: answer.error, attempts: attempt, permanent: true } + } + if (attempt === attempts) { + log('notify: %s not delivered to %s after %d attempts: %s', key, channel, attempt, answer.error) + return { state: 'failed', error: answer.error, attempts: attempt, permanent: false } + } + // `Retry-After` is a floor, never a ceiling: if the server asks for longer than our own + // backoff, waiting less only earns another 429. + const waitMs = Math.max(backoff(attempt), answer.retryAfterMs ?? 0) + log('notify: %s attempt %d failed (%s); retrying in %d ms', key, attempt, answer.error, waitMs) + await delay(waitMs) + } + return { state: 'failed', error: 'network', attempts, permanent: false } + } + + return { + get enabled() { return enabled }, + setEnabled(flag) { enabled = flag !== false }, + // At-least-once, deduplicated per `key`: the same reminder handed over again after it was sent + // is answered with the first result and never reaches the transport a second time; one handed + // over again while it is in flight joins that attempt instead of starting another. A reminder + // that gave up after its three attempts is not remembered, so a later run may try again. + send({ key, text, blocks, threadTs } = {}) { + if (typeof key !== 'string' || !key.trim()) throw invalid('send() needs a non-empty key') + if ((typeof text !== 'string' || !text.trim()) && !Array.isArray(blocks)) throw invalid('send() needs text or blocks') + if (!enabled) return Promise.resolve({ state: 'skipped', attempts: 0, reason: 'channel-off' }) + if (sent.has(key)) return Promise.resolve({ state: 'duplicate', ts: sent.get(key), attempts: 0 }) + if (refused.has(key)) return Promise.resolve({ state: 'failed', error: refused.get(key), attempts: 0, permanent: true }) + if (inFlight.has(key)) return inFlight.get(key) + const message = { channel, text, ...(blocks ? { blocks } : {}), ...(threadTs ? { threadTs } : {}) } + const started = run(key, message).finally(() => inFlight.delete(key)) + inFlight.set(key, started) + return started + }, + } +} diff --git a/patches/ds-bot/files/test/notify.test.js b/patches/ds-bot/files/test/notify.test.js new file mode 100644 index 000000000..f031d9427 --- /dev/null +++ b/patches/ds-bot/files/test/notify.test.js @@ -0,0 +1,171 @@ +import assert from 'node:assert/strict' +import { test } from 'node:test' +import { createNotifier, fakeSlack, rateLimited, channelNotFound, networkFailure } from '../src/host/notify.js' + +// Every retry below is driven by fake timers: a test never sleeps in real time. `settle` lets the +// adapter reach its next await (transport call or backoff) before the clock is moved. +async function settle() { + for (let i = 0; i < 3; i += 1) await new Promise(resolve => setImmediate(resolve)) +} + +test('#355: a reminder reaches the transport in Slack chat.postMessage shape', async () => { + const transport = fakeSlack([{ ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123' }) + + const first = await notifier.send({ key: 'r1', text: 'Standup in 10 minutes' }) + assert.deepEqual(first, { state: 'sent', ts: 'ts-1', attempts: 1 }) + // Optional fields are passed only when given, so a plain reminder is a plain message. + assert.deepEqual(transport.calls, [{ channel: 'C0123', text: 'Standup in 10 minutes' }]) + + await notifier.send({ key: 'r2', text: 'Still 10 minutes', blocks: [{ type: 'section' }], threadTs: 'ts-1' }) + assert.deepEqual(transport.calls[1], { channel: 'C0123', text: 'Still 10 minutes', blocks: [{ type: 'section' }], threadTs: 'ts-1' }) +}) + +test('#355: the same reminder handed over twice is one message', async () => { + const transport = fakeSlack([{ ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123' }) + + assert.equal((await notifier.send({ key: 'r1', text: 'Tomorrow: ship' })).state, 'sent') + const again = await notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + assert.deepEqual(again, { state: 'duplicate', ts: 'ts-1', attempts: 0 }) + assert.equal(transport.calls.length, 1, 'a repeated reminder is not a second transport call') +}) + +test('#355: a second send for a reminder already in flight joins it instead of starting another', async () => { + let release + const transport = fakeSlack([() => new Promise(resolve => { release = () => resolve({ ok: true, ts: 'ts-a' }) })]) + const notifier = createNotifier({ transport, channel: 'C0123' }) + + const first = notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + await settle() + const second = notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + release() + const [a, b] = await Promise.all([first, second]) + assert.equal(transport.calls.length, 1) + assert.equal(a.ts, 'ts-a') + assert.equal(b.ts, 'ts-a') +}) + +test('#355: a rate limit waits what Retry-After asks, then sends once', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }) + const transport = fakeSlack([rateLimited(2_000), { ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123' }) + + const pending = notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + await settle() + assert.equal(transport.calls.length, 1) + t.mock.timers.tick(1_999) + await settle() + assert.equal(transport.calls.length, 1, 'Retry-After is a floor, not a suggestion') + t.mock.timers.tick(1) + await settle() + assert.deepEqual(await pending, { state: 'sent', ts: 'ts-2', attempts: 2 }) + assert.equal(transport.calls.length, 2) +}) + +test('#355: a broken link backs off further on each attempt', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }) + const waits = [] + const transport = fakeSlack([networkFailure(), networkFailure(), { ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123', delay: async ms => { waits.push(ms) } }) + + const result = await notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + assert.deepEqual(waits, [500, 1_000]) + assert.deepEqual(result, { state: 'sent', ts: 'ts-3', attempts: 3 }) +}) + +test('#355: a thrown transport error is a broken link, not a reason to give up', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }) + const transport = fakeSlack([new Error('socket hang up'), { ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123' }) + + const pending = notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + await settle() + t.mock.timers.tick(500) + await settle() + assert.deepEqual(await pending, { state: 'sent', ts: 'ts-2', attempts: 2 }) +}) + +test('#355: three failed attempts end the delivery, and a later run may try again', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }) + const waits = [] + const transport = fakeSlack([networkFailure(), networkFailure(), networkFailure(), { ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123', delay: async ms => { waits.push(ms) } }) + + const first = await notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + assert.deepEqual(first, { state: 'failed', error: 'network', attempts: 3, permanent: false }) + assert.deepEqual(waits, [500, 1_000], 'three attempts, two waits: the budget is not exceeded') + assert.equal(transport.calls.length, 3) + + // Exhausting the attempts is not a verdict about the channel, so the reminder is not blacklisted. + const second = await notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + assert.equal(second.state, 'sent') + assert.equal(transport.calls.length, 4) +}) + +test('#355: Slack\'s own transient codes are retried, not remembered as permanent', async () => { + // Slack answers these for its own trouble (an internal error, a restart, a slow request). One of + // them must not cost the owner a reminder: each is retried on the same budget as a broken link. + for (const code of ['internal_error', 'fatal_error', 'service_unavailable', 'request_timeout']) { + const waits = [] + const transport = fakeSlack([{ ok: false, error: code }, { ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123', delay: async ms => { waits.push(ms) } }) + assert.deepEqual(await notifier.send({ key: `r-${code}`, text: 'Tomorrow: ship' }), { state: 'sent', ts: 'ts-2', attempts: 2 }, `${code} is worth another attempt`) + assert.deepEqual(waits, [500], `${code} backs off like any transient failure`) + assert.equal(transport.calls.length, 2) + } + // And running out of attempts is not a verdict about the channel: the reminder is not remembered + // as permanently refused, so a later run may still deliver it. + const transport = fakeSlack([{ ok: false, error: 'service_unavailable' }, { ok: false, error: 'service_unavailable' }, { ok: false, error: 'service_unavailable' }, { ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123', delay: async () => {} }) + assert.deepEqual(await notifier.send({ key: 'r-retry', text: 'Tomorrow: ship' }), { state: 'failed', error: 'service_unavailable', attempts: 3, permanent: false }) + assert.equal((await notifier.send({ key: 'r-retry', text: 'Tomorrow: ship' })).state, 'sent', 'a transient failure is not remembered as a refusal') +}) + +test('#355: a missing channel is a permanent failure and is not retried', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }) + const transport = fakeSlack([channelNotFound(), { ok: true }]) + const notifier = createNotifier({ transport, channel: 'C-gone' }) + + const first = await notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + assert.deepEqual(first, { state: 'failed', error: 'channel_not_found', attempts: 1, permanent: true }) + assert.equal(transport.calls.length, 1) + + const second = await notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + assert.deepEqual(second, { state: 'failed', error: 'channel_not_found', attempts: 0, permanent: true }) + assert.equal(transport.calls.length, 1, 'a structural failure stays answered without another call') +}) + +test('#355: a channel that is off makes no transport call at all', async () => { + const transport = fakeSlack([{ ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123', enabled: false }) + + assert.deepEqual(await notifier.send({ key: 'r1', text: 'Tomorrow: ship' }), { state: 'skipped', attempts: 0, reason: 'channel-off' }) + assert.equal(transport.calls.length, 0) +}) + +test('#355: switching a channel off stops a reminder that is waiting to retry', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }) + const transport = fakeSlack([rateLimited(5_000), { ok: true }]) + const notifier = createNotifier({ transport, channel: 'C0123' }) + + const pending = notifier.send({ key: 'r1', text: 'Tomorrow: ship' }) + await settle() + assert.equal(transport.calls.length, 1) + notifier.setEnabled(false) + t.mock.timers.tick(5_000) + await settle() + assert.deepEqual(await pending, { state: 'skipped', attempts: 1, reason: 'channel-off' }) + assert.equal(transport.calls.length, 1, 'the retry never leaves after the switch went off') + assert.equal(notifier.enabled, false) +}) + +test('#355: a reminder needs a key and something to say', async () => { + const notifier = createNotifier({ transport: fakeSlack(), channel: 'C0123' }) + // A malformed reminder is refused before anything is queued, so the caller hears it at once. + assert.throws(() => notifier.send({ text: 'no key' }), { code: 'notify/invalid' }) + assert.throws(() => notifier.send({ key: 'r1' }), { code: 'notify/invalid' }) + assert.throws(() => notifier.send({ key: 'r2', text: ' ' }), { code: 'notify/invalid' }) + assert.throws(() => createNotifier({ transport: fakeSlack() }), { code: 'notify/invalid' }) + assert.throws(() => createNotifier({ channel: 'C0123' }), { code: 'notify/invalid' }) +})