From 0e73346c5f62465e6c39a90c2a55d086677a69f0 Mon Sep 17 00:00:00 2001 From: Hampus Date: Mon, 31 Aug 2026 14:23:11 +0200 Subject: [PATCH] fix(svc): treat a shard overload reply as retryable backpressure (#2237) --- deploy/self-hosting/docker-compose.yml | 2 ++ .../message/MessageResponseDataService.ts | 7 ++-- .../api/infrastructure/NatsUnfurlerService.ts | 4 ++- .../src/api/infrastructure/SvcErrorReply.ts | 28 +++++++++++++++ .../api/infrastructure/UsersServiceClient.ts | 7 ++-- .../tests/SvcErrorReply.test.ts | 34 +++++++++++++++++++ 6 files changed, 77 insertions(+), 5 deletions(-) create mode 100644 fluxer_api/src/api/infrastructure/SvcErrorReply.ts create mode 100644 fluxer_api/src/api/infrastructure/tests/SvcErrorReply.test.ts diff --git a/deploy/self-hosting/docker-compose.yml b/deploy/self-hosting/docker-compose.yml index 9656d0c26..eeb56df2b 100644 --- a/deploy/self-hosting/docker-compose.yml +++ b/deploy/self-hosting/docker-compose.yml @@ -466,6 +466,7 @@ services: environment: <<: *fluxer-env FLUXER_SVC_MODE: router + FLUXER_SVC_MAX_CONCURRENT_REQUESTS: "${FLUXER_SVC_MAX_CONCURRENT_REQUESTS:-20}" depends_on: nats: {condition: service_started} @@ -528,6 +529,7 @@ services: <<: *fluxer-env FLUXER_SVC_NAME: messages FLUXER_SVC_MODE: router + FLUXER_SVC_MAX_CONCURRENT_REQUESTS: "${FLUXER_SVC_MAX_CONCURRENT_REQUESTS:-20}" depends_on: nats: {condition: service_started} diff --git a/fluxer_api/src/api/channel/services/message/MessageResponseDataService.ts b/fluxer_api/src/api/channel/services/message/MessageResponseDataService.ts index b0cb265fc..0c95a70b9 100644 --- a/fluxer_api/src/api/channel/services/message/MessageResponseDataService.ts +++ b/fluxer_api/src/api/channel/services/message/MessageResponseDataService.ts @@ -7,10 +7,11 @@ import {StringCodec} from 'nats'; import type {ChannelID, GuildID, MessageID, UserID} from '../../../BrandedTypes'; import {createUserID} from '../../../BrandedTypes'; import {Config} from '../../../Config'; +import {throwForSvcErrorReply} from '../../../infrastructure/SvcErrorReply'; import {Logger} from '../../../Logger'; import type {Channel} from '../../../models/Channel'; import type {Message} from '../../../models/Message'; -import {isJsonRecord, parseJsonWithGuard} from '../../../utils/JsonBoundaryUtils'; +import {isJsonRecord, parseJsonRecord, parseJsonWithGuard} from '../../../utils/JsonBoundaryUtils'; const MESSAGE_RESPONSE_SERVICE_SUBJECT = 'svc.messages'; const MESSAGE_RESPONSE_SERVICE_TIMEOUT_MS = 6000; @@ -303,8 +304,10 @@ export class MessageResponseDataService { this.codec.encode(JSON.stringify(payload)), {timeout: MESSAGE_RESPONSE_SERVICE_TIMEOUT_MS}, ); - const parsed = parseJsonWithGuard(this.codec.decode(response.data), isMessageServiceResponse); + const decoded = this.codec.decode(response.data); + const parsed = parseJsonWithGuard(decoded, isMessageServiceResponse); if (!parsed) { + throwForSvcErrorReply('message-response-service', parseJsonRecord(decoded)); throw new Error('[message-response-service] invalid response payload'); } return parsed; diff --git a/fluxer_api/src/api/infrastructure/NatsUnfurlerService.ts b/fluxer_api/src/api/infrastructure/NatsUnfurlerService.ts index 0ced1f57c..176f0b834 100644 --- a/fluxer_api/src/api/infrastructure/NatsUnfurlerService.ts +++ b/fluxer_api/src/api/infrastructure/NatsUnfurlerService.ts @@ -4,9 +4,10 @@ import type {MessageEmbedResponse} from '@fluxer/schema/src/domains/message/Embe import type {INatsConnectionManager} from '@pkgs/nats/src/INatsConnectionManager'; import {StringCodec} from 'nats'; import {Logger} from '../Logger'; -import {isJsonRecord, parseJsonWithGuard} from '../utils/JsonBoundaryUtils'; +import {isJsonRecord, parseJsonRecord, parseJsonWithGuard} from '../utils/JsonBoundaryUtils'; import type {MediaProxyNsfwMode} from './IMediaService'; import {IUnfurlerService, type UnfurlOptions, type UnfurlResult} from './IUnfurlerService'; +import {throwForSvcErrorReply} from './SvcErrorReply'; const NATS_UNFURL_SUBJECT = 'svc.unfurl'; const NATS_UNFURL_TIMEOUT_MS = 12000; @@ -92,6 +93,7 @@ export class NatsUnfurlerService extends IUnfurlerService { const responseText = this.codec.decode(responseMsg.data); const response = parseJsonWithGuard(responseText, isNatsUnfurlResponse); if (!response) { + throwForSvcErrorReply('nats-unfurl', parseJsonRecord(responseText)); throw new Error(`[nats-unfurl] invalid response payload: ${responseText}`); } if ('Resolved' in response) { diff --git a/fluxer_api/src/api/infrastructure/SvcErrorReply.ts b/fluxer_api/src/api/infrastructure/SvcErrorReply.ts new file mode 100644 index 000000000..c3a6fe141 --- /dev/null +++ b/fluxer_api/src/api/infrastructure/SvcErrorReply.ts @@ -0,0 +1,28 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {ServiceUnavailableError} from '@fluxer/errors/src/domains/core/ServiceUnavailableError'; +import {isJsonRecord} from '../utils/JsonBoundaryUtils'; + +const OVERLOADED = 'overloaded'; + +export function svcErrorReplyReason(decoded: unknown): string | null { + if (!isJsonRecord(decoded)) { + return null; + } + const reason = decoded.error; + return typeof reason === 'string' && reason.length > 0 ? reason : null; +} + +export function throwForSvcErrorReply(service: string, decoded: unknown): void { + const reason = svcErrorReplyReason(decoded); + if (reason === null) { + return; + } + if (reason === OVERLOADED) { + throw new ServiceUnavailableError({ + message: `[${service}] shard rejected the request because it is at its concurrency limit`, + headers: {'Retry-After': '1'}, + }); + } + throw new ServiceUnavailableError({message: `[${service}] ${reason}`}); +} diff --git a/fluxer_api/src/api/infrastructure/UsersServiceClient.ts b/fluxer_api/src/api/infrastructure/UsersServiceClient.ts index dfb518535..75482e65e 100644 --- a/fluxer_api/src/api/infrastructure/UsersServiceClient.ts +++ b/fluxer_api/src/api/infrastructure/UsersServiceClient.ts @@ -7,7 +7,8 @@ import {StringCodec} from 'nats'; import {createUserID, type UserID} from '../BrandedTypes'; import {Config} from '../Config'; import {Logger} from '../Logger'; -import {isJsonRecord, parseJsonWithGuard} from '../utils/JsonBoundaryUtils'; +import {isJsonRecord, parseJsonRecord, parseJsonWithGuard} from '../utils/JsonBoundaryUtils'; +import {throwForSvcErrorReply} from './SvcErrorReply'; const USERS_SERVICE_SUBJECT = process.env.FLUXER_USERS_SERVICE_SUBJECT || 'svc.users'; const DEFAULT_USERS_SERVICE_TIMEOUT_MS = 6000; @@ -163,8 +164,10 @@ export class NatsUsersServiceClient implements IUsersServiceClient { const response = await connection.request(this.subject, this.codec.encode(JSON.stringify(payload)), { timeout: this.requestTimeoutMs, }); - const parsed = parseJsonWithGuard(this.codec.decode(response.data), isUsersServiceResponse); + const decoded = this.codec.decode(response.data); + const parsed = parseJsonWithGuard(decoded, isUsersServiceResponse); if (!parsed) { + throwForSvcErrorReply('users-service', parseJsonRecord(decoded)); throw new Error('[users-service] invalid response payload'); } return parsed; diff --git a/fluxer_api/src/api/infrastructure/tests/SvcErrorReply.test.ts b/fluxer_api/src/api/infrastructure/tests/SvcErrorReply.test.ts new file mode 100644 index 000000000..9b0f3cd65 --- /dev/null +++ b/fluxer_api/src/api/infrastructure/tests/SvcErrorReply.test.ts @@ -0,0 +1,34 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {ServiceUnavailableError} from '@fluxer/errors/src/domains/core/ServiceUnavailableError'; +import {describe, expect, it} from 'vitest'; +import {svcErrorReplyReason, throwForSvcErrorReply} from '../SvcErrorReply'; + +describe('SvcErrorReply', () => { + it('recognises the shard overload reply', () => { + expect(svcErrorReplyReason({error: 'overloaded'})).toBe('overloaded'); + }); + + it('maps an overload reply to a retryable service unavailable error', () => { + try { + throwForSvcErrorReply('users-service', {error: 'overloaded'}); + expect.unreachable('should have thrown'); + } catch (error) { + expect(error).toBeInstanceOf(ServiceUnavailableError); + expect((error as ServiceUnavailableError).status).toBe(503); + expect((error as ServiceUnavailableError).headers?.['Retry-After']).toBe('1'); + } + }); + + it('maps any other structured error reply to service unavailable', () => { + expect(() => throwForSvcErrorReply('users-service', {error: 'shard_unavailable'})).toThrow(ServiceUnavailableError); + }); + + it('ignores a reply that is not a structured error', () => { + expect(svcErrorReplyReason({user: {id: '1'}})).toBeNull(); + expect(svcErrorReplyReason(null)).toBeNull(); + expect(svcErrorReplyReason('overloaded')).toBeNull(); + expect(svcErrorReplyReason({error: ''})).toBeNull(); + expect(() => throwForSvcErrorReply('users-service', {user: {id: '1'}})).not.toThrow(); + }); +});