fix(svc): treat a shard overload reply as retryable backpressure (#2237)

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}
 

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;

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) {

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}`});
+}

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;

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();
+	});
+});