diff --git a/doc/api.md b/doc/api.md index 82d32399..e83c3718 100644 --- a/doc/api.md +++ b/doc/api.md @@ -437,6 +437,7 @@ bound for long-lived bots that talk to many distinct chats. | Method | Params | Returns | Description | | --- | --- | --- | --- | +| `close` | - | void | Close the client, but only if this store opened it (a passed-in `client` and the shared `redis` are the caller's / runtime's). Closing ends this store's life - a later use throws and you construct a new one. With a shared or caller-supplied client this is a no-op. | | `delete` | `key`: string | Promise | - | | `read` | `key`: string | Promise | The stored string for `key`, or `undefined` when there is none. | | `touch` | `key`: string, `ttlSeconds`: number | Promise | Refresh a key's expiry without rewriting it - what the middleware calls when an update changed nothing, so an active chat is not evicted mid-conversation. | @@ -1109,12 +1110,13 @@ out, whether the loop stopped or threw. A fatal poll-stop is also written to stderr before being re-thrown: a bare rejection can be dropped (fire-and-forget, or a swallowing `.catch`), which would leave the process alive but no longer polling - the silent hang #1350 -describes. +describes. Pass `exitOnError: true` to also exit the process non-zero after +teardown, so a supervisor restarts the bot instead of relying on the caller. | Param | Type | | --- | --- | | `bot` | [Bot](#bot) | -| `options?` | [LongPollOptions](#longpolloptions) | +| `options` | [RunOptions](#runoptions) | **Returns:** Promise @@ -1136,9 +1138,10 @@ comparison always inspects every position, so it leaks no information about ### `startWebhook()` Managed webhook runner: create a `node:http` webhook server, start listening, -and resolve when it shuts down. Installs `SIGINT`/`SIGTERM` handlers that close -the server for a graceful exit (cleaned up in a `finally`), mirroring `run()` for -long polling. Rejects if the server fails (e.g. the port is in use). +and resolve when it shuts down. Once listening, installs `SIGINT`/`SIGTERM` +handlers that close the server for a graceful exit (cleaned up in a `finally`), +mirroring `run()` for long polling. Rejects if the server fails to start (e.g. +the port is in use). Shutdown cannot hang: `server.close()` waits for existing connections to end, so we also drop idle keep-alive sockets at once and force-close anything still @@ -6344,6 +6347,7 @@ type RedisSessionStorageOptions = { client?: RedisClient; prefix?: string; ttlSeconds?: number; + url?: string; }; ``` @@ -7211,6 +7215,14 @@ type RichTextUrl = { }; ``` +### `RunOptions` + +```ts +type RunOptions = [LongPollOptions](#longpolloptions) & { + exitOnError?: boolean; +}; +``` + ### `SavePreparedInlineMessageParams` ```ts diff --git a/examples/01-polling-bot.ts b/examples/01-polling-bot.ts index 3cb5e065..cc5d63d2 100644 --- a/examples/01-polling-bot.ts +++ b/examples/01-polling-bot.ts @@ -47,4 +47,7 @@ bot.catch((err, ctx) => { }); // `run()` resolves when the process receives SIGINT/SIGTERM and the loop drains. -await run(bot); +// `exitOnError` makes a fatal poll-stop exit non-zero (after logging) so a +// supervisor (systemd, Docker, pm2, ...) restarts the bot instead of leaving a +// live-but-silent process. +await run(bot, { exitOnError: true }); diff --git a/examples/06-keyboards.ts b/examples/06-keyboards.ts index 79aa7799..4ebd7731 100644 --- a/examples/06-keyboards.ts +++ b/examples/06-keyboards.ts @@ -53,4 +53,4 @@ bot.on("callback_query", async (ctx) => { if (data) await ctx.reply(`You pressed: ${data}`); }); -await run(bot); +await run(bot, { exitOnError: true }); diff --git a/examples/09-middleware.ts b/examples/09-middleware.ts index 1e289624..72462f70 100644 --- a/examples/09-middleware.ts +++ b/examples/09-middleware.ts @@ -60,4 +60,4 @@ bot.on("message", (ctx) => { return ctx.reply(`You've sent ${session.count} message(s) this session.`); }); -await run(bot); +await run(bot, { exitOnError: true }); diff --git a/examples/12-conversation.ts b/examples/12-conversation.ts index 7810ca0c..eb22b60d 100644 --- a/examples/12-conversation.ts +++ b/examples/12-conversation.ts @@ -74,4 +74,4 @@ bot.on("message", (ctx) => { } }); -await run(bot); +await run(bot, { exitOnError: true }); diff --git a/examples/16-sessions.ts b/examples/16-sessions.ts index 6d3fbd37..517e257b 100644 --- a/examples/16-sessions.ts +++ b/examples/16-sessions.ts @@ -84,4 +84,4 @@ bot.on("message", async (ctx, next) => { await ctx.reply(`Thanks ${data.name}! Saved ${data.email}. Send /me to see it, /start to redo.`); }); -await run(bot); +await run(bot, { exitOnError: true }); diff --git a/examples/17-callback-tracking.ts b/examples/17-callback-tracking.ts index 85e8d2bf..d435b9bf 100644 --- a/examples/17-callback-tracking.ts +++ b/examples/17-callback-tracking.ts @@ -160,4 +160,4 @@ bot.on("callback_query", async (ctx, next) => { await ctx.reply(`Ran ${pending.op} on ${pending.targets.join(", ")} (asked at ${pending.requestedAt}).`); }); -await run(bot); +await run(bot, { exitOnError: true }); diff --git a/src/bun/redis-storage.ts b/src/bun/redis-storage.ts index 40def566..7ca173a3 100644 --- a/src/bun/redis-storage.ts +++ b/src/bun/redis-storage.ts @@ -13,12 +13,14 @@ * `./node` (a CI guard enforces it). */ -import { redis, type RedisClient } from "bun"; +import { RedisClient, redis } from "bun"; import type { SessionStore, SessionWriteOptions } from "../core/session.js"; export type RedisSessionStorageOptions = { /** Bun `RedisClient` to use. Defaults to Bun's shared `redis` (REDIS_URL / VALKEY_URL). */ client?: RedisClient; + /** Connect a client the store owns (and closes on teardown) to this URL, instead of the shared `redis`. Ignored when `client` is given. */ + url?: string; /** Prefix prepended to every key. Default `"session:"`. */ prefix?: string; /** @@ -30,25 +32,57 @@ export type RedisSessionStorageOptions = { export class RedisSessionStorage implements SessionStore { private readonly client: RedisClient; + /** True when this store opened the client itself, so `close()` may close it. */ + private readonly owned: boolean; + /** Set once `close()` closed a client this store owned - the store is then spent. */ + private closed = false; private readonly prefix: string; private readonly ttlSeconds?: number; constructor(options: RedisSessionStorageOptions = {}) { - this.client = options.client ?? redis; + this.owned = options.client === undefined && options.url !== undefined; + this.client = + options.client ?? (options.url !== undefined ? this.createClient(options.url) : redis); this.prefix = options.prefix ?? "session:"; this.ttlSeconds = options.ttlSeconds; } + /** Construct the owned client for a `url`. A seam so tests can supply a fake. */ + protected createClient(url: string): RedisClient { + return new RedisClient(url); + } + + private open(): RedisClient { + if (this.closed) { + throw new Error("RedisSessionStorage: this store was closed; construct a new one"); + } + return this.client; + } + + /** + * Close the client, but only if this store opened it (a passed-in `client` and + * the shared `redis` are the caller's / runtime's). Closing ends this store's + * life - a later use throws and you construct a new one. With a shared or + * caller-supplied client this is a no-op. + */ + close(): void { + if (this.owned && !this.closed) { + this.closed = true; + this.client.close(); + } + } + async read(key: string): Promise { - return (await this.client.get(this.prefix + key)) ?? undefined; + return (await this.open().get(this.prefix + key)) ?? undefined; } async write(key: string, value: string, options?: SessionWriteOptions): Promise { + const client = this.open(); const k = this.prefix + key; - await this.client.set(k, value); + await client.set(k, value); const ttl = options?.ttlSeconds ?? this.ttlSeconds; if (ttl !== undefined) { - await this.client.expire(k, ttl); + await client.expire(k, ttl); } } @@ -57,10 +91,10 @@ export class RedisSessionStorage implements SessionStore { * an update changed nothing, so an active chat is not evicted mid-conversation. */ async touch(key: string, ttlSeconds: number): Promise { - await this.client.expire(this.prefix + key, ttlSeconds); + await this.open().expire(this.prefix + key, ttlSeconds); } async delete(key: string): Promise { - await this.client.del(this.prefix + key); + await this.open().del(this.prefix + key); } } diff --git a/src/node/run.ts b/src/node/run.ts index 4d68da69..0993266d 100644 --- a/src/node/run.ts +++ b/src/node/run.ts @@ -14,6 +14,22 @@ import type { Bot } from "../core/bot.js"; import type { LongPollOptions } from "../core/longpoll.js"; import { withShutdownSignals } from "./signals.js"; +/** Write to stderr and resolve once the chunk is flushed - a following `process.exit()` would otherwise truncate it. */ +function writeStderr(line: string): Promise { + return new Promise((resolve) => { + process.stderr.write(line, () => resolve()); + }); +} + +export type RunOptions = LongPollOptions & { + /** + * On a fatal poll-stop, exit the process with a non-zero code (after teardown + * and the stderr log) so a supervisor restarts the bot. Default false - `run` + * re-throws instead, leaving the exit policy to the caller. + */ + exitOnError?: boolean; +}; + /** * Start the bot's long-poll loop and resolve when it stops. Installs * `SIGINT`/`SIGTERM` handlers that trigger `bot.stop()` for a clean shutdown, @@ -25,24 +41,40 @@ import { withShutdownSignals } from "./signals.js"; * A fatal poll-stop is also written to stderr before being re-thrown: a bare * rejection can be dropped (fire-and-forget, or a swallowing `.catch`), which * would leave the process alive but no longer polling - the silent hang #1350 - * describes. + * describes. Pass `exitOnError: true` to also exit the process non-zero after + * teardown, so a supervisor restarts the bot instead of relying on the caller. */ -export async function run(bot: Bot, options?: LongPollOptions): Promise { +export async function run(bot: Bot, options: RunOptions = {}): Promise { + const { exitOnError = false, ...pollOptions } = options; // Only a call that actually owns the pump may close. `startPolling` refuses // when another run is already active (as a rejection - it is async - so the // loser cannot be told apart after the fact); checking first is what keeps this // call from closing stores under the run that is still using them. const owned = !bot.isRunning(); + let failed = false; try { return await withShutdownSignals( () => bot.stop(), - () => bot.startPolling(undefined, options), + () => bot.startPolling(undefined, pollOptions), ); } catch (err) { - // Never let a fatal poll-stop be silent (see the doc comment above). - process.stderr.write(`node-telegram-bot-api: polling stopped on a fatal error: ${String(err)}\n`); + // Only the call that owns the pump reports and acts on the stop; a call that + // lost the "already running" race just re-throws to its own caller - it must + // not log a misleading "polling stopped" line or exit a healthy pump's process. + if (owned) { + failed = true; + // Await the flush so the exit below can't truncate this diagnostic. + await writeStderr(`node-telegram-bot-api: polling stopped on a fatal error: ${String(err)}\n`); + } throw err; } finally { - if (owned) await bot.close(); + try { + if (owned) await bot.close(); + } finally { + // Nested so it runs even when teardown throws: `exitOnError` must not be + // defeated by the very stuck resource it exists to escape. Gated on + // `failed` (which implies `owned`), so a losing double-run never exits. + if (failed && exitOnError) process.exit(1); + } } } diff --git a/test/unit/bun/redis-storage.test.ts b/test/unit/bun/redis-storage.test.ts index b1df0582..ce216d79 100644 --- a/test/unit/bun/redis-storage.test.ts +++ b/test/unit/bun/redis-storage.test.ts @@ -8,12 +8,13 @@ import { RedisSessionStorage } from "../../../src/bun/redis-storage.js"; // Node runner skips it; `bun test test/unit` runs it. /** Minimal in-memory stand-in for the RedisClient methods the store uses. */ -function fakeRedis(): RedisClient & { store: Map; expires: Map } { +function fakeRedis(): RedisClient & { store: Map; expires: Map; closes: number } { const store = new Map(); const expires = new Map(); const client = { store, expires, + closes: 0, async get(key: string): Promise { return store.has(key) ? (store.get(key) as string) : null; }, @@ -28,12 +29,29 @@ function fakeRedis(): RedisClient & { store: Map; expires: Map; expires: Map }; + return client as unknown as RedisClient & { store: Map; expires: Map; closes: number }; } const envelope = JSON.stringify({ v: 1, data: { n: 1 } }); +/** + * A store that takes the `{ url }` (owned) path but whose owned client is `client`. + * `createClient` is overridden via a closure (not an instance field), so the fake + * is available during `super()` - a field initializer would run too late. + */ +function ownedStore(client: RedisClient): RedisSessionStorage { + class Owned extends RedisSessionStorage { + protected override createClient(): RedisClient { + return client; + } + } + return new Owned({ url: "redis://fake" }); +} + describe("RedisSessionStorage", () => { test("prefixes keys and round-trips the encoded string", async () => { const client = fakeRedis(); @@ -78,4 +96,29 @@ describe("RedisSessionStorage", () => { await new RedisSessionStorage({ client, ttlSeconds: 3600 }).write("k", envelope, { ttlSeconds: 60 }); assert.equal(client.expires.get("session:k"), 60); }); + + test("close() does not close an injected client (the caller owns it) and the store stays usable", async () => { + const client = fakeRedis(); + const store = new RedisSessionStorage({ client }); + store.close(); + assert.equal(client.closes, 0); // not ours to close + await store.write("k", envelope); // still usable + assert.equal(await store.read("k"), envelope); + }); + + test("a url-owned client is closed on close(), idempotently, and reuse throws", async () => { + const client = fakeRedis(); + const store = ownedStore(client); + + await store.write("k", envelope); // usable while open + assert.equal(await store.read("k"), envelope); + + store.close(); + assert.equal(client.closes, 1); // owned -> closed + store.close(); + assert.equal(client.closes, 1); // idempotent, not double-closed + + await assert.rejects(store.read("k"), /was closed/); + await assert.rejects(store.write("k", envelope), /was closed/); + }); }); diff --git a/test/unit/run.test.ts b/test/unit/run.test.ts index ac129360..dcc97145 100644 --- a/test/unit/run.test.ts +++ b/test/unit/run.test.ts @@ -14,8 +14,12 @@ function envelopeFetch(body: unknown): typeof fetch { async function captureStderr(fn: () => Promise): Promise { const original = process.stderr.write.bind(process.stderr); let captured = ""; - process.stderr.write = ((chunk: string | Uint8Array) => { + // `run` writes with a flush callback (so `process.exit` can't truncate it), so + // the stub must invoke it - otherwise the write's promise never resolves. + process.stderr.write = ((chunk: string | Uint8Array, ...rest: unknown[]) => { captured += typeof chunk === "string" ? chunk : Buffer.from(chunk).toString(); + const cb = rest.find((a) => typeof a === "function") as (() => void) | undefined; + cb?.(); return true; }) as typeof process.stderr.write; try { @@ -26,6 +30,22 @@ async function captureStderr(fn: () => Promise): Promise { return captured; } +/** Run `fn` with `process.exit` stubbed, returning the exit codes it requested. */ +async function captureExit(fn: () => Promise): Promise { + const original = process.exit; + const codes: number[] = []; + process.exit = ((code?: number) => { + codes.push(code ?? 0); + // Do not actually exit; let `run` finish so the test can assert. + }) as typeof process.exit; + try { + await fn(); + } finally { + process.exit = original; + } + return codes; +} + describe("run", () => { test("surfaces a fatal poll-stop to stderr and re-throws", async () => { // 401 is non-retriable; retry:false makes longPoll throw on the first poll. @@ -64,4 +84,74 @@ describe("run", () => { assert.strictEqual(stderr, ""); assert.strictEqual(bot.isRunning(), false); }); + + test("exitOnError exits non-zero after a fatal poll-stop", async () => { + const bot = new Bot("123:abc", { + fetch: envelopeFetch({ ok: false, error_code: 401, description: "Unauthorized" }), + }); + + let codes: number[] = []; + // Suppress the stderr line so it doesn't clutter the test output. + await captureStderr(async () => { + codes = await captureExit(async () => { + // Still re-throws (exit is stubbed, so control returns to the caller). + await assert.rejects(run(bot, { retry: false, exitOnError: true }), TelegramApiError); + }); + }); + + assert.deepStrictEqual(codes, [1]); + }); + + test("exitOnError does not exit on a clean stop", async () => { + const bot = new Bot("123:abc", { + fetch: envelopeFetch({ ok: true, result: [] }), + }); + + const codes = await captureExit(async () => { + const running = run(bot, { exitOnError: true }); + bot.stop(); + await running; + }); + + assert.deepStrictEqual(codes, []); + }); + + test("exitOnError still exits when teardown (bot.close) throws", async () => { + const bot = new Bot("123:abc", { + fetch: envelopeFetch({ ok: false, error_code: 401, description: "Unauthorized" }), + }); + // A wedged middleware whose close() rejects is exactly the case exitOnError + // must survive - the exit can't be defeated by the stuck resource. + bot.close = async () => { + throw new Error("teardown boom"); + }; + + let codes: number[] = []; + await captureStderr(async () => { + codes = await captureExit(async () => { + await assert.rejects(run(bot, { retry: false, exitOnError: true })); + }); + }); + + assert.deepStrictEqual(codes, [1]); + }); + + test("exitOnError does not exit (or kill) a run that lost the already-running race", async () => { + const bot = new Bot("123:abc", { + fetch: envelopeFetch({ ok: true, result: [] }), + }); + + const first = run(bot); // becomes the owning pump + for (let i = 0; i < 50 && !bot.isRunning(); i++) await new Promise((r) => setTimeout(r, 5)); + assert.strictEqual(bot.isRunning(), true); + + const codes = await captureExit(async () => { + // Loses the race -> rejects, but must NOT exit the healthy pump's process. + await assert.rejects(run(bot, { exitOnError: true })); + }); + assert.deepStrictEqual(codes, []); + + bot.stop(); + await first; + }); });