diff --git a/src/matches/match-events.gateway.spec.ts b/src/matches/match-events.gateway.spec.ts new file mode 100644 index 00000000..b675b84f --- /dev/null +++ b/src/matches/match-events.gateway.spec.ts @@ -0,0 +1,76 @@ +// Isolate the gateway from its heavy DI imports; we only exercise the +// dedup/processing ordering logic in handleMatchEvent. +jest.mock("./events", () => ({ MatchEvents: { testEvent: class {} } })); +jest.mock("src/hasura/hasura.service", () => ({ HasuraService: class {} })); +jest.mock("src/cache/cache.service", () => ({ CacheService: class {} })); + +import { MatchEventsGateway } from "./match-events.gateway"; + +function makeGateway(opts: { cacheHit?: boolean; processImpl?: () => any }) { + const cache = { + has: jest.fn().mockResolvedValue(opts.cacheHit ?? false), + put: jest.fn().mockResolvedValue(undefined), + }; + const processor = { + setData: jest.fn(), + process: jest.fn().mockImplementation(opts.processImpl ?? (async () => {})), + }; + const moduleRef = { resolve: jest.fn().mockResolvedValue(processor) }; + const logger = { + warn: jest.fn(), + error: jest.fn(), + log: jest.fn(), + debug: jest.fn(), + }; + const gateway = new MatchEventsGateway( + logger as any, + moduleRef as any, + {} as any, + cache as any, + ); + return { gateway, cache, processor, logger }; +} + +const message = { + matchId: "match-1", + messageId: "msg-1", + data: { event: "testEvent", data: {} }, +}; + +describe("MatchEventsGateway.handleMatchEvent dedup", () => { + it("marks the event processed only after process() succeeds", async () => { + const { gateway, cache, processor } = makeGateway({}); + const result = await gateway.handleMatchEvent(message as any); + + expect(processor.process).toHaveBeenCalledTimes(1); + expect(cache.put).toHaveBeenCalledTimes(1); + // Ordering: process resolved before the dedup entry was written. + expect(processor.process.mock.invocationCallOrder[0]).toBeLessThan( + cache.put.mock.invocationCallOrder[0], + ); + expect(result).toBe("msg-1"); + }); + + it("does NOT write the dedup entry when process() throws (so redelivery retries)", async () => { + const { gateway, cache, processor } = makeGateway({ + processImpl: async () => { + throw new Error("hasura down"); + }, + }); + + await expect(gateway.handleMatchEvent(message as any)).rejects.toThrow( + "hasura down", + ); + expect(processor.process).toHaveBeenCalledTimes(1); + expect(cache.put).not.toHaveBeenCalled(); + }); + + it("short-circuits on a dedup hit without processing", async () => { + const { gateway, cache, processor } = makeGateway({ cacheHit: true }); + const result = await gateway.handleMatchEvent(message as any); + + expect(result).toBe("msg-1"); + expect(processor.process).not.toHaveBeenCalled(); + expect(cache.put).not.toHaveBeenCalled(); + }); +}); diff --git a/src/matches/match-events.gateway.ts b/src/matches/match-events.gateway.ts index e8ca75cc..754b1cb1 100644 --- a/src/matches/match-events.gateway.ts +++ b/src/matches/match-events.gateway.ts @@ -115,8 +115,6 @@ export class MatchEventsGateway { return messageId; } - await this.cache.put(cacheKey, true, 10); - const { data, event } = message.data; this.logger.debug( @@ -127,7 +125,7 @@ export class MatchEventsGateway { if (!Processor) { this.logger.warn("unable to find event handler", event); - return; + return messageId; } const processor = @@ -138,16 +136,21 @@ export class MatchEventsGateway { try { await processor.process(); } catch (error) { + // Do NOT write the dedup entry on failure: leave the key absent so the + // game server's redelivery of this messageId is reprocessed instead of + // being silently swallowed by the dedup short-circuit for its TTL. this.logger.error( - `[${matchId}] error processing game event ${event} (messageId=${messageId}): ${error.message}`, - error.stack, + `[${matchId}] error processing game event ${event} (messageId=${messageId}): ${ + (error as Error)?.message + }`, + (error as Error)?.stack, ); - // withhold the ack and clear the dedup key so the game server's retry - // re-processes instead of being swallowed by the cache - await this.cache.forget(cacheKey); - return; + throw error; } + // Mark processed only after success. + await this.cache.put(cacheKey, true, 10); + return messageId; } } diff --git a/src/news/news.controller.ts b/src/news/news.controller.ts index eb6ff2c3..fc8996b2 100644 --- a/src/news/news.controller.ts +++ b/src/news/news.controller.ts @@ -12,6 +12,7 @@ import { FileTypeValidator, ForbiddenException, NotFoundException, + Logger, } from "@nestjs/common"; import { FileInterceptor } from "@nestjs/platform-express"; import { Request, Response } from "express"; @@ -25,6 +26,8 @@ import { NewsService } from "./news.service"; @Controller("news") export class NewsController { + private readonly logger = new Logger(NewsController.name); + constructor( private readonly news: NewsService, private readonly system: SystemService, @@ -126,6 +129,25 @@ export class NewsController { res.setHeader("ETag", result.etag); } + // Without these handlers an S3 stream error would throw an unhandled + // 'error' event (crashing the process), and a client that disconnects + // mid-download would leave the upstream stream open (fd leak). + result.stream.on("error", (error: Error) => { + this.logger.error( + `error streaming news image ${filename}: ${error?.message}`, + error?.stack, + ); + result.stream.destroy(); + if (!res.headersSent) { + res.status(500).end(); + } else { + res.destroy(); + } + }); + res.on("close", () => { + result.stream.destroy(); + }); + result.stream.pipe(res); } diff --git a/src/sockets/sockets.service.ts b/src/sockets/sockets.service.ts index 230052e8..554e31e2 100644 --- a/src/sockets/sockets.service.ts +++ b/src/sockets/sockets.service.ts @@ -41,7 +41,32 @@ export class SocketsService { void sub.subscribe("broadcast-message"); void sub.subscribe("send-message-to-steam-id"); sub.on("message", (channel, message) => { - const { steamId, event, data } = JSON.parse(message) as { + let parsed: unknown; + try { + parsed = JSON.parse(message); + } catch (error) { + // A malformed payload on the pub/sub channel must not take the pod + // down: this handler runs outside any request lifecycle, so a throw + // here is an uncaught exception. + this.logger.error( + `failed to parse pub/sub message on ${channel}: ${ + (error as Error)?.message + }`, + ); + return; + } + + // JSON.parse succeeds for non-objects too (e.g. the literal `null`, which + // typeof-reports as "object"); guard before destructuring so a valid but + // non-object payload can't throw a TypeError here. + if (parsed === null || typeof parsed !== "object") { + this.logger.error( + `ignoring non-object pub/sub message on ${channel}`, + ); + return; + } + + const { steamId, event, data } = parsed as { steamId: string; event: string; data: unknown;