From eadc650b38c991048b5a816f114ecb0fa760fccb Mon Sep 17 00:00:00 2001 From: RedStar Date: Sat, 17 Oct 2026 11:39:19 +0200 Subject: [PATCH] feat(plugin-gateway): add RegisterAsDispatchListener for one-process dispatch listeners Add a decorator that runs a dispatch listener on the workers while the gateway client forwards its dispatches, and on the gateway client otherwise. forwardGatewayDispatches marks the client through the new optional GatewayDispatchEmitterLike#markDispatchesForwarded (ref-counted, undone by the stop function). The decorator checks the flag on every emit, so the order forwarding and the pieces load in does not matter. Gateway-only events are rejected by the types and throw at decoration time. Refs #244 --- .../broker-mark-forwarded-dispatches.md | 5 + .../gateway-register-as-dispatch-listener.md | 5 + packages/plugin-broker/README.md | 17 +++ packages/plugin-broker/src/lib/gateway.ts | 15 ++- .../tests/gateway-replay.test.ts | 50 ++++++++- packages/plugin-broker/tests/gateway.test.ts | 13 +++ packages/plugin-gateway/README.md | 33 ++++++ packages/plugin-gateway/src/GatewayClient.ts | 28 +++++ .../plugin-gateway/src/util/decorators.ts | 71 ++++++++++++- packages/plugin-gateway/src/util/events.ts | 33 ++++++ .../tests/EventGatewayListener.test.ts | 100 ++++++++++++++++++ 11 files changed, 366 insertions(+), 4 deletions(-) create mode 100644 .changeset/broker-mark-forwarded-dispatches.md create mode 100644 .changeset/gateway-register-as-dispatch-listener.md diff --git a/.changeset/broker-mark-forwarded-dispatches.md b/.changeset/broker-mark-forwarded-dispatches.md new file mode 100644 index 00000000..13fb8bf0 --- /dev/null +++ b/.changeset/broker-mark-forwarded-dispatches.md @@ -0,0 +1,5 @@ +--- +"@wolfstar/plugin-broker": minor +--- + +`forwardGatewayDispatches` marks a `GatewayClient` as forwarding (`forwardsDispatches`) until its stop function is called, through the new optional `GatewayDispatchEmitterLike#markDispatchesForwarded`, so listeners registered with `RegisterAsDispatchListener` from `@wolfstar/plugin-gateway` run on the workers only. Emitters without it forward as before. diff --git a/.changeset/gateway-register-as-dispatch-listener.md b/.changeset/gateway-register-as-dispatch-listener.md new file mode 100644 index 00000000..a9ac7283 --- /dev/null +++ b/.changeset/gateway-register-as-dispatch-listener.md @@ -0,0 +1,5 @@ +--- +"@wolfstar/plugin-gateway": minor +--- + +Add `RegisterAsDispatchListener`, a `RegisterAsGatewayListener` whose listener runs on one process only: the workers replaying the dispatches while the gateway client forwards them, the gateway client otherwise, so a bot loading the same pieces on both no longer handles each dispatch twice. `GatewayClient` gains `forwardsDispatches` and `markDispatchesForwarded()` for `forwardGatewayDispatches` to set, and `DispatchEventName`, `GatewayOnlyEventName` and `gatewayOnlyEvents` tell the events a worker receives from the ones only the gateway client emits, which the decorator rejects. diff --git a/packages/plugin-broker/README.md b/packages/plugin-broker/README.md index f2f73781..01459b40 100644 --- a/packages/plugin-broker/README.md +++ b/packages/plugin-broker/README.md @@ -232,6 +232,23 @@ entries and run them concurrently, so two events of the same guild can be replay across workers, and the relations a listener resolves from the cache can differ from the gateway process's. The previous state shipped with each entry is a snapshot, so `old` stays right. +A bot that loads the same pieces on the gateway process and on the workers would handle each dispatch +twice. `forwardGatewayDispatches` marks a `GatewayClient` as forwarding (`forwardsDispatches`) until +its stop function is called, and `RegisterAsDispatchListener` from +[`@wolfstar/plugin-gateway`](https://www.npmjs.com/package/@wolfstar/plugin-gateway) uses it to run +a listener on the workers only, and on the gateway client when nothing is forwarded, with no +environment variable to read: + +```ts +@RegisterAsDispatchListener("messageCreate") +export class ModerationListener extends EventGatewayListener<"messageCreate"> {} +``` + +It rejects the events a worker never receives (`clientReady`, `shardReady`, …), and a `guildDelete` +emitted by `READY` is not forwarded +([#243](https://github.com/wolfstar-project/plugins/issues/243)), so such a listener on the workers +misses it. + ### `BrokerListener` piece Like `EventGatewayListener` in `@wolfstar/plugin-gateway`, a `BrokerListener` piece typed by event diff --git a/packages/plugin-broker/src/lib/gateway.ts b/packages/plugin-broker/src/lib/gateway.ts index ed5573af..afc39f54 100644 --- a/packages/plugin-broker/src/lib/gateway.ts +++ b/packages/plugin-broker/src/lib/gateway.ts @@ -37,6 +37,12 @@ export interface GatewayDispatchEmitterLike { * {@link GatewayReplayTargetLike.reviveDispatchState}. An emitter without it forwards dispatches without state. */ serializeDispatchState?(type: string, state: unknown): unknown; + /** + * Marks the emitter as forwarding its dispatches, until the returned function is called, so that listeners meant to + * run on one process only (`RegisterAsDispatchListener`) stay idle on it. An emitter without it forwards without + * telling. + */ + markDispatchesForwarded?(): () => void; on(event: "dispatch", listener: GatewayDispatchListener): unknown; off(event: "dispatch", listener: GatewayDispatchListener): unknown; } @@ -63,7 +69,8 @@ export interface ForwardGatewayDispatchesOptions { * sharing that cache (a `@wolfstar/plugin-cache` Redis cache) see the state the dispatch left when they receive it. * Publishing does not hold up the gateway, and the entries are added in the order the dispatches were processed. * For a `GatewayClient` the entry also carries the dispatch's previous state and shard, which - * `replayGatewayDispatches` uses on workers. + * `replayGatewayDispatches` uses on workers, and the client is marked as forwarding until forwarding stops, so the + * listeners registered with `RegisterAsDispatchListener` leave the dispatches to the workers. * * @example * ```typescript @@ -125,7 +132,11 @@ export function forwardGatewayDispatches( }; client.on("dispatch", listener); - return () => void client.off("dispatch", listener); + const unmark = client.markDispatchesForwarded?.(); + return () => { + client.off("dispatch", listener); + unmark?.(); + }; } /** diff --git a/packages/plugin-broker/tests/gateway-replay.test.ts b/packages/plugin-broker/tests/gateway-replay.test.ts index 3378ec0c..e9df0571 100644 --- a/packages/plugin-broker/tests/gateway-replay.test.ts +++ b/packages/plugin-broker/tests/gateway-replay.test.ts @@ -1,5 +1,11 @@ import { createInMemoryCache } from "@wolfstar/plugin-cache"; -import { GatewayClient } from "@wolfstar/plugin-gateway"; +import { container } from "@wolfstar/http-framework"; +import { + EventGatewayListener, + GatewayClient, + RegisterAsDispatchListener, + type Message, +} from "@wolfstar/plugin-gateway"; import { afterEach, describe, expect, test, vi } from "vitest"; import { BrokerConsumer, @@ -299,4 +305,46 @@ describe("replayGatewayDispatches", () => { expect(replayDispatch).not.toHaveBeenCalled(); }); + + test("GIVEN a gateway process and a worker loading the same dispatch listener THEN only the worker runs it", async () => { + const { broker, consumer } = setup(); + const received: string[] = []; + + @RegisterAsDispatchListener("messageCreate") + class LogMessagesListener extends EventGatewayListener<"messageCreate"> { + public override run(_message: Message) { + received.push(this.name); + } + } + const store = container.stores.get("listeners"); + const context = (name: string): EventGatewayListener.LoaderContext => ({ + name, + path: `/virtual/${name}.js`, + root: "/virtual", + store, + }); + + const { producer, worker } = clients(); + try { + // A listener binds to `container.client`, so point it at each client in turn. + container.client = producer; + await store.insert(new LogMessagesListener(context("gateway"), {} as never)); + container.client = worker; + await store.insert(new LogMessagesListener(context("worker"), {} as never)); + + const stopForwarding = forwardGatewayDispatches(producer, broker); + replayGatewayDispatches(consumer, worker); + await consumer.start(); + + await feed(producer, "MESSAGE_CREATE", message()); + await vi.waitFor(() => expect(received).toEqual(["worker"])); + + // Once forwarding stops, the gateway process handles its dispatches itself again. + stopForwarding(); + await feed(producer, "MESSAGE_CREATE", message({ id: "1200000000000000001" })); + expect(received).toEqual(["worker", "gateway"]); + } finally { + await store.unloadAll(); + } + }); }); diff --git a/packages/plugin-broker/tests/gateway.test.ts b/packages/plugin-broker/tests/gateway.test.ts index b1cec3e5..ded0c2e6 100644 --- a/packages/plugin-broker/tests/gateway.test.ts +++ b/packages/plugin-broker/tests/gateway.test.ts @@ -86,6 +86,19 @@ describe("forwardGatewayDispatches", () => { expect(client.listenerCount("dispatch")).toBe(0); }); + test("GIVEN an emitter that can be marked THEN it is marked until forwarding stops", () => { + const unmark = vi.fn(); + const markDispatchesForwarded = vi.fn(() => unmark); + const client = Object.assign(gateway(), { markDispatchesForwarded }); + + const stop = forwardGatewayDispatches(client, { publish: vi.fn() }); + expect(markDispatchesForwarded).toHaveBeenCalledOnce(); + expect(unmark).not.toHaveBeenCalled(); + + stop(); + expect(unmark).toHaveBeenCalledOnce(); + }); + test("GIVEN an emitter that serializes state THEN the serialized state is published with the dispatch", () => { const publish = vi.fn().mockResolvedValue("1-0"); const serializeDispatchState = vi.fn((_type: string, state: unknown) => ({ raw: state })); diff --git a/packages/plugin-gateway/README.md b/packages/plugin-gateway/README.md index c9913388..93a5ee33 100644 --- a/packages/plugin-gateway/README.md +++ b/packages/plugin-gateway/README.md @@ -283,6 +283,39 @@ export class LogMessagesListener extends EventGatewayListener<"messageCreate"> { With `once: true`, the listener unloads itself after its first run. +With [`@wolfstar/plugin-broker`](https://www.npmjs.com/package/@wolfstar/plugin-broker), a +dispatch is handled by the gateway process and, replayed, by every worker, so pieces loaded on both +would run twice. `RegisterAsDispatchListener` takes the same arguments and runs the listener on +one process only: the workers while the gateway client forwards its dispatches +(`forwardGatewayDispatches` marks it as `client.forwardsDispatches`), the gateway client itself +otherwise, so it also works on a bot that does not use the broker: + +```ts +import { + EventGatewayListener, + RegisterAsDispatchListener, + type Message, +} from "@wolfstar/plugin-gateway"; + +@RegisterAsDispatchListener("messageCreate") +export class LogMessagesListener extends EventGatewayListener<"messageCreate"> { + public override run(message: Message) { + console.log(`${message.author.username}: ${message.content}`); + } +} +``` + +The events only the gateway client emits (`clientReady`, `shardReady`, `shardResume`, +`shardClose`, `shardError`, `dispatch`, `cacheError`, `cacheSweep`) never reach a worker: the +types reject them and the decorator throws, so keep `RegisterAsGatewayListener` for those. The +decision is made on every emit, so the order forwarding starts and the pieces load in does not +matter. Two limits come with it: a `guildDelete` that `READY` emits for the guilds the client no +longer lists is not forwarded +([#243](https://github.com/wolfstar-project/plugins/issues/243)), so a listener on the workers +misses it, and with `events` set on `forwardGatewayDispatches`, a listener of an event built from +a dispatch left out runs nowhere. Workers receive dispatches at least once, so keep the listener +idempotent. + Any piece (listener, command, interaction handler) reaches the client through `this.container.gatewayClient`, typed as `GatewayClient`, so its managers need no cast: diff --git a/packages/plugin-gateway/src/GatewayClient.ts b/packages/plugin-gateway/src/GatewayClient.ts index 1ade6b37..c647b9ac 100644 --- a/packages/plugin-gateway/src/GatewayClient.ts +++ b/packages/plugin-gateway/src/GatewayClient.ts @@ -458,6 +458,9 @@ export class GatewayClient extends Client { // guilds proceed concurrently. readonly #queue = new DispatchQueue(); + // How many forwarders of this client's dispatches are running, see `markDispatchesForwarded`. + #dispatchForwarders = 0; + // One cache per entity, shared by every manager of that entity. readonly #caches = new Map>(); @@ -940,6 +943,31 @@ export class GatewayClient extends Client { return emitAndWait(this, event, args); } + /** + * Whether the dispatches of this client are forwarded to workers, which replay them on their own listeners, see + * `forwardGatewayDispatches` from `@wolfstar/plugin-broker`. Listeners registered with + * {@link RegisterAsDispatchListener} stay idle on a client that does. + */ + public get forwardsDispatches(): boolean { + return this.#dispatchForwarders > 0; + } + + /** + * Marks this client as forwarding its dispatches, until the returned function is called. Called by + * `forwardGatewayDispatches`, it counts, so the client keeps forwarding while any forwarder is left. + * + * @returns A function to undo the mark, which does nothing when called again. + */ + public markDispatchesForwarded(): () => void { + this.#dispatchForwarders++; + let marked = true; + return () => { + if (!marked) return; + marked = false; + this.#dispatchForwarders--; + }; + } + /** * The dispatch types {@link GatewayClient.replayDispatch} turns into events. */ diff --git a/packages/plugin-gateway/src/util/decorators.ts b/packages/plugin-gateway/src/util/decorators.ts index 664a5343..b05c8ff6 100644 --- a/packages/plugin-gateway/src/util/decorators.ts +++ b/packages/plugin-gateway/src/util/decorators.ts @@ -1,4 +1,4 @@ -import type { GatewayEventName } from "./events.js"; +import { gatewayOnlyEvents, type DispatchEventName, type GatewayEventName } from "./events.js"; import type { EventGatewayListener } from "../EventGatewayListener.js"; type ListenerConstructor = new (...args: any[]) => EventGatewayListener; @@ -40,3 +40,72 @@ export function RegisterAsGatewayListener( } as unknown as T; }; } + +/** + * Like {@link RegisterAsGatewayListener}, for an event that comes from a dispatch, but the listener only runs on one + * process: the workers replaying the dispatches while the gateway client forwards them (see `forwardGatewayDispatches` + * from `@wolfstar/plugin-broker`), the gateway client itself otherwise. A bot loading the same pieces on both + * processes therefore handles each dispatch once, without reading its environment to tell which process it is on. + * + * @remarks + * The decision is made on every emit, so it does not matter whether the forwarding starts before or after the + * listener is loaded, and stopping it brings the listener back. A `once` listener is not spent while it idles. + * + * The events only the gateway client emits (`clientReady`, `shardReady`, `shardResume`, `shardClose`, `shardError`, + * `dispatch`, `cacheError`, `cacheSweep`) are never replayed on a worker: they are rejected by the types, and the + * decorator throws for them. Use {@link RegisterAsGatewayListener} for those. + * + * Only the dispatches that are forwarded reach the workers: with `forwardGatewayDispatches`'s `events` set, a listener + * of an event built from another dispatch runs nowhere. A `guildDelete` that `READY` emits for the guilds it no longer + * lists is not forwarded either (wolfstar-project/plugins#243), so a listener on the workers misses it. Delivery to the + * workers is at-least-once, so the listener should be idempotent. + * + * @example + * ```typescript + * import { EventGatewayListener, RegisterAsDispatchListener, type Message } from '@wolfstar/plugin-gateway'; + * + * @RegisterAsDispatchListener('messageCreate') + * export class LogMessagesListener extends EventGatewayListener<'messageCreate'> { + * public override run(message: Message) { + * console.log(`${message.author.username}: ${message.content}`); + * } + * } + * ``` + * + * @param event The dispatch event to listen to. + * @param options The additional listener options. + * @throws {TypeError} When the event is one only the gateway client emits. + */ +export function RegisterAsDispatchListener( + event: Event, + options: Omit, "event"> = {}, +) { + if (gatewayOnlyEvents.has(event)) { + throw new TypeError( + `The "${event}" event is only emitted by the gateway client, and never reaches a worker: use RegisterAsGatewayListener.`, + ); + } + + const register = RegisterAsGatewayListener(event, options); + return function decorate(target: T): T { + const base = register(target) as unknown as new (...args: any[]) => DispatchListenerInternals; + return class extends base { + public constructor(...args: any[]) { + super(...args); + + // The outermost wrapper, so a `once` listener is only unloaded by a run that happens. + // oxlint-disable-next-line no-underscore-dangle + const listener = this._listener; + // oxlint-disable-next-line no-underscore-dangle + this._listener = (...listenerArgs: readonly unknown[]) => + this.emitter.forwardsDispatches ? undefined : listener(...listenerArgs); + } + } as unknown as T; + }; +} + +interface DispatchListenerInternals { + // Anything but a `GatewayClient` has no flag, and so never forwards. + emitter: { forwardsDispatches?: boolean }; + _listener: (...args: readonly unknown[]) => unknown; +} diff --git a/packages/plugin-gateway/src/util/events.ts b/packages/plugin-gateway/src/util/events.ts index 81a2520f..3c65b91e 100644 --- a/packages/plugin-gateway/src/util/events.ts +++ b/packages/plugin-gateway/src/util/events.ts @@ -398,6 +398,39 @@ export interface GatewayEventMap { */ export type GatewayEventName = keyof GatewayEventMap; +/** + * The events only the process that connects to Discord emits: a worker replaying forwarded dispatches never sees + * them. Every other event comes from a dispatch, `raw` included, which {@link GatewayClient.replayDispatch} emits too. + */ +export type GatewayOnlyEventName = + | "dispatch" + | "shardReady" + | "clientReady" + | "shardResume" + | "shardClose" + | "shardError" + | "cacheError" + | "cacheSweep"; + +/** + * The name of any event that comes from a dispatch, and so reaches a worker replaying the forwarded ones. + */ +export type DispatchEventName = Exclude; + +/** + * The {@link GatewayOnlyEventName} events, for telling them from {@link DispatchEventName} ones at runtime. + */ +export const gatewayOnlyEvents: ReadonlySet = new Set([ + "dispatch", + "shardReady", + "clientReady", + "shardResume", + "shardClose", + "shardError", + "cacheError", + "cacheSweep", +]); + /** * The name of every event a {@link GatewayClient} emits, mirroring the keys of {@link GatewayEventMap}, like * `@wolfstar/http-framework`'s own `Events`. diff --git a/packages/plugin-gateway/tests/EventGatewayListener.test.ts b/packages/plugin-gateway/tests/EventGatewayListener.test.ts index 351fa92e..735825a4 100644 --- a/packages/plugin-gateway/tests/EventGatewayListener.test.ts +++ b/packages/plugin-gateway/tests/EventGatewayListener.test.ts @@ -10,7 +10,10 @@ import { afterEach, describe, expect, test } from "vitest"; import { EventGatewayListener, GatewayClient, + RegisterAsDispatchListener, RegisterAsGatewayListener, + gatewayOnlyEvents, + GatewayEvents, type User, } from "../src/index.js"; @@ -123,3 +126,100 @@ describe("EventGatewayListener", () => { expect(store.has("once")).toBe(false); }); }); + +describe("RegisterAsDispatchListener", () => { + afterEach(async () => { + await container.stores.get("listeners").unloadAll(); + }); + + function decorated(received: string[], options: { once?: boolean } = {}) { + @RegisterAsDispatchListener("userUpdate", options) + class UserUpdateListener extends EventGatewayListener<"userUpdate"> { + public override run(_previous: User | null, current: User) { + received.push(current.username); + } + } + return new UserUpdateListener(context("dispatch"), {} as never); + } + + test("GIVEN a client that does not forward THEN the listener runs", async () => { + const client = createClient(); + const received: string[] = []; + const listener = decorated(received); + + expect(listener.emitter).toBe(client); + expect(listener.event).toBe("userUpdate"); + + await container.stores.get("listeners").insert(listener); + await sendUserUpdate(client, "standalone"); + + expect(received).toEqual(["standalone"]); + }); + + test("GIVEN a client that forwards THEN the listener stays idle until forwarding stops", async () => { + const client = createClient(); + const received: string[] = []; + await container.stores.get("listeners").insert(decorated(received)); + + const unmark = client.markDispatchesForwarded(); + expect(client.forwardsDispatches).toBe(true); + await sendUserUpdate(client, "forwarded"); + expect(received).toEqual([]); + + unmark(); + expect(client.forwardsDispatches).toBe(false); + await sendUserUpdate(client, "again"); + expect(received).toEqual(["again"]); + }); + + test("GIVEN forwarding that starts before the listener loads THEN the listener stays idle", async () => { + const client = createClient(); + const received: string[] = []; + client.markDispatchesForwarded(); + + await container.stores.get("listeners").insert(decorated(received)); + await sendUserUpdate(client, "forwarded"); + + expect(received).toEqual([]); + }); + + test("GIVEN several forwarders THEN the client forwards until the last stops", () => { + const client = createClient(); + const first = client.markDispatchesForwarded(); + const second = client.markDispatchesForwarded(); + + first(); + first(); + expect(client.forwardsDispatches).toBe(true); + second(); + expect(client.forwardsDispatches).toBe(false); + }); + + test("GIVEN once THEN an idle listener is not spent", async () => { + const client = createClient(); + const received: string[] = []; + const store = container.stores.get("listeners"); + await store.insert(decorated(received, { once: true })); + + const unmark = client.markDispatchesForwarded(); + await sendUserUpdate(client, "idle"); + expect(store.has("dispatch")).toBe(true); + + unmark(); + await sendUserUpdate(client, "first"); + await sendUserUpdate(client, "second"); + + expect(received).toEqual(["first"]); + expect(store.has("dispatch")).toBe(false); + }); + + test.each([...gatewayOnlyEvents])("GIVEN the gateway-only %s event THEN it throws", (event) => { + expect(() => RegisterAsDispatchListener(event as never)).toThrow(TypeError); + }); + + test("GIVEN the gateway-only events THEN they are events of the client, and raw is not one", () => { + const events = new Set(Object.values(GatewayEvents)); + for (const event of gatewayOnlyEvents) expect(events.has(event)).toBe(true); + expect(gatewayOnlyEvents.has("raw")).toBe(false); + }); +});