From cb04a37b22dbddb3b83aad7203edef6090df31f7 Mon Sep 17 00:00:00 2001 From: Flegma Date: Fri, 10 Jul 2026 21:58:07 +0200 Subject: [PATCH 1/2] fix: api event-dedup ordering, news stream error handling, redis parse guard - MatchEventsGateway: write the dedup cache entry only after the processor succeeds, and wrap processing in try/catch that logs and rethrows. A transient failure no longer both throws and leaves a 'processed' marker that suppresses the game server's redelivery for the 10s TTL (#532). - NewsController.serveImage: handle 'error' on the S3 stream and destroy it on response close, so a stream error can't crash the process and a client disconnect can't leak the upstream fd (#546). - SocketsService: guard JSON.parse of pub/sub messages so a malformed payload logs and returns instead of throwing an uncaught exception in the message listener (#557, redis-handler half). Adds match-events.gateway.spec.ts covering the dedup ordering. --- src/matches/match-events.gateway.spec.ts | 76 ++++++++++++++++++++++++ src/matches/match-events.gateway.ts | 21 ++++--- src/news/news.controller.ts | 22 +++++++ src/sockets/sockets.service.ts | 21 +++++-- 4 files changed, 126 insertions(+), 14 deletions(-) create mode 100644 src/matches/match-events.gateway.spec.ts diff --git a/src/matches/match-events.gateway.spec.ts b/src/matches/match-events.gateway.spec.ts new file mode 100644 index 000000000..b675b84f6 --- /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 e8ca75cc7..754b1cb15 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 eb6ff2c38..fc8996b2d 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 230052e85..913261cb9 100644 --- a/src/sockets/sockets.service.ts +++ b/src/sockets/sockets.service.ts @@ -41,11 +41,22 @@ 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 { - steamId: string; - event: string; - data: unknown; - }; + let parsed: { steamId: string; event: string; data: 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; + } + + const { steamId, event, data } = parsed; switch (channel) { case "broadcast-message": From ee07ce0ebd06aed702bb755fc7ad941226680ca3 Mon Sep 17 00:00:00 2001 From: Flegma Date: Fri, 10 Jul 2026 22:21:59 +0200 Subject: [PATCH 2/2] fix: guard non-object pub/sub payloads before destructuring (review follow-up) JSON.parse succeeds for a valid non-object payload (e.g. literal null, which typeof-reports as object), so the destructure sat outside the try and would throw an uncaught TypeError. Validate parsed is a non-null object first. --- src/sockets/sockets.service.ts | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/src/sockets/sockets.service.ts b/src/sockets/sockets.service.ts index 913261cb9..554e31e2c 100644 --- a/src/sockets/sockets.service.ts +++ b/src/sockets/sockets.service.ts @@ -41,7 +41,7 @@ export class SocketsService { void sub.subscribe("broadcast-message"); void sub.subscribe("send-message-to-steam-id"); sub.on("message", (channel, message) => { - let parsed: { steamId: string; event: string; data: unknown }; + let parsed: unknown; try { parsed = JSON.parse(message); } catch (error) { @@ -56,7 +56,21 @@ export class SocketsService { return; } - const { steamId, event, data } = parsed; + // 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; + }; switch (channel) { case "broadcast-message":