From f4b79f390c43d3830ef99a8d67c5a176741c2ccc Mon Sep 17 00:00:00 2001 From: natho080 Date: Wed, 26 Aug 2026 16:42:13 +0100 Subject: [PATCH 1/4] feat(prisma): add read replica routing via DATABASE_REPLICA_URL --- .env.example | 7 ++++++- src/prisma/prisma.service.ts | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/.env.example b/.env.example index 40acd49..3684b93 100644 --- a/.env.example +++ b/.env.example @@ -2,6 +2,11 @@ # PostgreSQL connection URL (used by Prisma) — REQUIRED DATABASE_URL="postgresql://parashield:secret@localhost:5432/parashield_db?schema=public" +# Read replica URL (#439) — optional. When set, all read-only queries are +# routed to this instance via PrismaService.reader; writes and transactions +# always go to the primary DATABASE_URL. Use the same format as DATABASE_URL. +# DATABASE_REPLICA_URL="postgresql://parashield:secret@replica-host:5432/parashield_db?schema=public" + # Prisma connection pool tuning (#381) — optional. # The backend applies these automatically to the datasource URL at runtime; # values already present as DATABASE_URL query params (e.g. ?connection_limit=10) @@ -110,5 +115,5 @@ CORS_ORIGIN="http://localhost:3000" # Optional tuning (defaults shown). See README "CORS configuration". # CORS_METHODS=GET,POST,PUT,DELETE,OPTIONS -# CORS_ALLOWED_HEADERS=Content-Type,Authorization,x-wallet-address,x-wallet-signature,x-wallet-message,x-api-key,x-admin-api-key +# CORS_ALLOWED_HEADERS=Content-Type,Authorization,x-wallet-address,x-wallet-signature,x-wallet-message,x-api-key,x-admin-api-key,Idempotency-Key # CORS_CREDENTIALS=false diff --git a/src/prisma/prisma.service.ts b/src/prisma/prisma.service.ts index 4fe38a3..017dc42 100644 --- a/src/prisma/prisma.service.ts +++ b/src/prisma/prisma.service.ts @@ -38,8 +38,33 @@ function withConnectionPoolParams(url: string | undefined): string | undefined { export class PrismaService extends PrismaClient implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(PrismaService.name); + // #439 — Read replica client. Instantiated only when DATABASE_REPLICA_URL + // is set; otherwise reader falls back to the primary so callers need no + // conditional logic and existing code paths are unchanged. + private readonly replicaClient: PrismaClient | null; + constructor() { super({ datasourceUrl: withConnectionPoolParams(process.env.DATABASE_URL) }); + + const replicaUrl = withConnectionPoolParams(process.env.DATABASE_REPLICA_URL); + this.replicaClient = replicaUrl + ? new PrismaClient({ datasourceUrl: replicaUrl }) + : null; + } + + /** + * Returns the read-replica PrismaClient when DATABASE_REPLICA_URL is + * configured, otherwise returns the primary client. + * + * Use this for all read-only queries (findMany, findFirst, findUnique, + * count, aggregate) to offload traffic from the primary. + * Always use `this.prisma` (the primary) for writes and $transactions. + * + * @example + * const items = await this.prisma.reader.policy.findMany({ ... }); + */ + get reader(): PrismaClient { + return this.replicaClient ?? this; } async onModuleInit() { @@ -48,6 +73,10 @@ export class PrismaService extends PrismaClient implements OnModuleInit, OnModul for (let attempt = 1; attempt <= maxRetries; attempt++) { try { await this.$connect(); + if (this.replicaClient) { + await this.replicaClient.$connect(); + this.logger.log('Read-replica connection established'); + } this.logger.log('Database connection established'); return; } catch (err) { @@ -68,6 +97,9 @@ export class PrismaService extends PrismaClient implements OnModuleInit, OnModul async onModuleDestroy() { await this.$disconnect(); + if (this.replicaClient) { + await this.replicaClient.$disconnect(); + } this.logger.log('Database connection closed'); } } From 6717d215955ff68e18b683fa8b1ffc76512ea94d Mon Sep 17 00:00:00 2001 From: natho080 Date: Wed, 26 Aug 2026 16:42:48 +0100 Subject: [PATCH 2/4] feat(middleware): add idempotency key deduplication for mutating requests --- .../middleware/idempotency.middleware.ts | 102 ++++++++++++++++++ src/main.ts | 10 ++ 2 files changed, 112 insertions(+) create mode 100644 src/common/middleware/idempotency.middleware.ts diff --git a/src/common/middleware/idempotency.middleware.ts b/src/common/middleware/idempotency.middleware.ts new file mode 100644 index 0000000..2749cf6 --- /dev/null +++ b/src/common/middleware/idempotency.middleware.ts @@ -0,0 +1,102 @@ +import { Injectable, NestMiddleware, Logger } from '@nestjs/common'; +import { Request, Response, NextFunction } from 'express'; +import Redis from 'ioredis'; + +// #438 — Idempotency keys prevent duplicate processing of the same request. +// A client that retries a POST (e.g. after a network timeout) sends the same +// Idempotency-Key header; we return the cached response instead of re-running +// the handler. Only applies to mutating methods (POST, PUT, PATCH). +const IDEMPOTENCY_TTL_SECONDS = 86_400; // 24 h — matches typical API conventions +const IDEMPOTENCY_KEY_MAX_LENGTH = 255; + +/** + * IdempotencyMiddleware (#438) + * + * Intercepts POST/PUT/PATCH requests that include an `Idempotency-Key` header. + * + * First call — processes normally, caches the response body + status in Redis + * for 24 hours under the key `idempotency:{method}:{path}:{key}`. + * Repeat call — returns the cached response immediately with a + * `X-Idempotent-Replayed: true` header so callers can detect it. + * + * Requests without the header pass through unchanged, so the middleware is + * completely opt-in and does not affect existing clients. + * + * Register in main.ts after body parsers: + * const idempotency = new IdempotencyMiddleware(redisClient); + * app.use((req, res, next) => idempotency.use(req, res, next)); + */ +@Injectable() +export class IdempotencyMiddleware implements NestMiddleware { + private readonly logger = new Logger(IdempotencyMiddleware.name); + + constructor(private readonly redis: Redis) {} + + use(req: Request, res: Response, next: NextFunction): void { + const MUTATING_METHODS = ['POST', 'PUT', 'PATCH']; + if (!MUTATING_METHODS.includes(req.method)) { + return next(); + } + + const rawKey = req.headers['idempotency-key'] as string | undefined; + if (!rawKey) { + return next(); + } + + // Basic validation — reject keys that are too long or contain newlines + // (could be used to construct arbitrary Redis keys). + if (rawKey.length > IDEMPOTENCY_KEY_MAX_LENGTH || /[\r\n]/.test(rawKey)) { + res.status(400).json({ + success: false, + errorCode: 'BAD_REQUEST', + error: 'Invalid Idempotency-Key header value', + statusCode: 400, + }); + return; + } + + const storeKey = `idempotency:${req.method}:${req.path}:${rawKey}`; + + this.redis.get(storeKey).then((cached) => { + if (cached) { + try { + const { status, body } = JSON.parse(cached) as { status: number; body: unknown }; + this.logger.debug(`Replaying idempotent response for key=${rawKey}`); + res.setHeader('X-Idempotent-Replayed', 'true'); + res.status(status).json(body); + return; + } catch { + // Corrupted cache entry — fall through to normal processing. + this.logger.warn(`Failed to parse cached idempotency entry for key=${rawKey}`); + } + } + + // Intercept the response so we can cache it before it is sent. + const originalJson = res.json.bind(res); + res.json = (body: unknown) => { + // Only cache successful responses (2xx) to avoid caching transient errors. + if (res.statusCode >= 200 && res.statusCode < 300) { + const entry = JSON.stringify({ status: res.statusCode, body }); + this.redis + .set(storeKey, entry, 'EX', IDEMPOTENCY_TTL_SECONDS) + .catch((err: unknown) => + this.logger.warn( + `Failed to store idempotency key=${rawKey}: ${err instanceof Error ? err.message : String(err)}`, + ), + ); + } + return originalJson(body); + }; + + next(); + }).catch((err: unknown) => { + // Redis unavailable — fail open (process the request normally) so a + // Redis outage doesn't take down the API. Log as a warning so ops can + // detect the degradation. + this.logger.warn( + `Idempotency Redis lookup failed for key=${rawKey}: ${err instanceof Error ? err.message : String(err)}. Processing request without idempotency check.`, + ); + next(); + }); + } +} diff --git a/src/main.ts b/src/main.ts index 158988a..60d3117 100644 --- a/src/main.ts +++ b/src/main.ts @@ -10,12 +10,14 @@ import { JsonLogger } from './common/logging/json-logger.service'; import { InputSanitizationMiddleware } from './common/middleware/input-sanitization.middleware'; import { RequestTimeoutMiddleware } from './common/middleware/request-timeout.middleware'; import { UsdcPrecisionValidationMiddleware } from './common/middleware/usdc-precision-validation.middleware'; +import { IdempotencyMiddleware } from './common/middleware/idempotency.middleware'; import { loadVaultSecrets } from './common/secrets/vault-secrets.loader'; import { applyRateLimitHeaders } from './common/swagger/rate-limit-headers'; import { initializeOpenTelemetry } from './common/telemetry/opentelemetry'; import helmet from 'helmet'; import { ConfigService } from '@nestjs/config'; import { json, urlencoded } from 'express'; +import Redis from 'ioredis'; const REQUEST_BODY_LIMIT = '1mb'; const SERVER_TIMEOUT_MS = 30_000; @@ -31,6 +33,7 @@ const DEFAULT_CORS_ALLOWED_HEADERS = [ 'x-wallet-message', 'x-api-key', 'x-admin-api-key', + 'Idempotency-Key', ]; function parseCsvEnv(value: string | undefined): string[] | undefined { @@ -79,6 +82,13 @@ async function bootstrap() { const usdcValidator = new UsdcPrecisionValidationMiddleware(); app.use((req, res, next) => usdcValidator.use(req, res, next)); + // #438 — idempotency key deduplication for mutating requests (POST/PUT/PATCH). + // Clients include an `Idempotency-Key` header; replayed requests with the + // same key get the cached response instead of re-executing the handler. + const redisClient = app.get('REDIS_CLIENT'); + const idempotency = new IdempotencyMiddleware(redisClient); + app.use((req, res, next) => idempotency.use(req, res, next)); + // Global exception filter app.useGlobalFilters(new GlobalExceptionFilter()); From b8f6beb1d4231404cf531658b5ce2280c3906e1e Mon Sep 17 00:00:00 2001 From: natho080 Date: Wed, 26 Aug 2026 16:43:49 +0100 Subject: [PATCH 3/4] feat(streaming): add NDJSON streaming interceptor for list endpoints --- src/claims/claims.controller.ts | 17 ++- .../interceptors/streaming.interceptor.ts | 123 ++++++++++++++++++ src/oracle/oracle.controller.ts | 22 +++- src/policy/policy.controller.ts | 16 +++ 4 files changed, 175 insertions(+), 3 deletions(-) create mode 100644 src/common/interceptors/streaming.interceptor.ts diff --git a/src/claims/claims.controller.ts b/src/claims/claims.controller.ts index 0314e82..255917e 100644 --- a/src/claims/claims.controller.ts +++ b/src/claims/claims.controller.ts @@ -1,4 +1,5 @@ -import { Body, Controller, ForbiddenException, Get, Param, Post, Query, Req, UseGuards, UnauthorizedException, NotFoundException, Throttle } from '@nestjs/common'; +import { Body, Controller, ForbiddenException, Get, Param, Post, Query, Req, UseGuards, UseInterceptors, UnauthorizedException, NotFoundException, Throttle } from '@nestjs/common'; +import { StreamingInterceptor } from '../common/interceptors/streaming.interceptor'; import { ApiTags, ApiOperation, @@ -55,11 +56,18 @@ export class ClaimsController { /** GET /api/v1/claims?wallet=... — get claim history for the authenticated wallet */ @Get() @UseGuards(JwtAuthGuard) + @UseInterceptors(StreamingInterceptor) @ApiBearerAuth() @ApiOperation({ summary: 'Get claim history for a wallet address (query param)' }) @ApiQuery({ name: 'wallet', required: true, description: 'Stellar wallet address' }) @ApiQuery({ name: 'page', required: false, description: 'Page number' }) @ApiQuery({ name: 'limit', required: false, description: 'Items per page' }) + @ApiQuery({ + name: 'stream', + required: false, + description: "Set to 'true' to receive the data array as NDJSON (one item per line). Alternatively send Accept: application/x-ndjson. Pagination metadata available in X-Total-Count, X-Page, X-Limit headers.", + example: 'true', + }) @ApiResponse({ status: 200, description: 'Returns paginated claim history — { success, data, total, page, limit }', @@ -138,11 +146,18 @@ export class ClaimsController { /** GET /api/v1/claims/history/:wallet — get all claims for a wallet address */ @Get('history/:wallet') @UseGuards(JwtAuthGuard) + @UseInterceptors(StreamingInterceptor) @ApiBearerAuth() @ApiOperation({ summary: 'Get all claims for a wallet address' }) @ApiParam({ name: 'wallet', description: 'Stellar wallet address' }) @ApiQuery({ name: 'page', required: false, description: 'Page number' }) @ApiQuery({ name: 'limit', required: false, description: 'Items per page' }) + @ApiQuery({ + name: 'stream', + required: false, + description: "Set to 'true' to receive the data array as NDJSON (one item per line). Alternatively send Accept: application/x-ndjson. Pagination metadata available in X-Total-Count, X-Page, X-Limit headers.", + example: 'true', + }) @ApiResponse({ status: 200, description: 'Returns paginated claim history — { success, data, total, page, limit }', diff --git a/src/common/interceptors/streaming.interceptor.ts b/src/common/interceptors/streaming.interceptor.ts new file mode 100644 index 0000000..c6d884d --- /dev/null +++ b/src/common/interceptors/streaming.interceptor.ts @@ -0,0 +1,123 @@ +import { + Injectable, + NestInterceptor, + ExecutionContext, + CallHandler, + StreamableFile, +} from '@nestjs/common'; +import { Observable } from 'rxjs'; +import { switchMap } from 'rxjs/operators'; +import { PassThrough } from 'stream'; +import { Request, Response } from 'express'; + +/** + * StreamingInterceptor — streams large list responses as NDJSON to reduce + * peak memory usage (#440). + * + * Instead of buffering the full result array and serialising it as a single + * JSON payload, the interceptor writes each item as an individual JSON line + * (newline-delimited JSON / NDJSON) through a PassThrough Node stream. The + * Express response is flushed chunk-by-chunk with Transfer-Encoding: chunked, + * so the server never holds the entire serialised payload in memory at once. + * + * Opt-in trigger (either condition activates streaming): + * • Request header: Accept: application/x-ndjson + * • Query parameter: ?stream=true + * + * When neither condition is present the interceptor is a no-op and the + * standard JSON response path is used unchanged. + * + * The NDJSON format uses one JSON line per item: + * {"id":"...","status":"ACTIVE",...}\n + * {"id":"...","status":"ACTIVE",...}\n + * ... + * + * Clients that need the full metadata envelope (success, total, page, limit) + * can either use the standard (non-streaming) response or read the custom + * response headers: + * X-Total-Count — total number of items in the list + * X-Page — page number returned (when applicable) + * X-Limit — page size (when applicable) + * + * Usage: + * @UseInterceptors(StreamingInterceptor) + * @Get('some-list') + * async list(...) { ... } + */ +@Injectable() +export class StreamingInterceptor implements NestInterceptor { + intercept(context: ExecutionContext, next: CallHandler): Observable { + const http = context.switchToHttp(); + const request = http.getRequest(); + const response = http.getResponse(); + + const wantsStream = + request.headers['accept'] === 'application/x-ndjson' || + request.query['stream'] === 'true'; + + if (!wantsStream) { + return next.handle(); + } + + return next.handle().pipe( + switchMap((payload: unknown) => { + // Only stream responses that carry a data array. + // Non-list responses (single objects, errors) pass through unchanged. + if ( + payload === null || + typeof payload !== 'object' || + !Array.isArray((payload as Record)['data']) + ) { + return new Observable(subscriber => { + subscriber.next(payload); + subscriber.complete(); + }); + } + + const envelope = payload as Record; + const items = envelope['data'] as unknown[]; + const total = envelope['total']; + const page = envelope['page']; + const limit = envelope['limit']; + + // Expose pagination metadata via response headers so NDJSON clients + // don't have to parse a wrapper object just to get count info. + response.setHeader('Content-Type', 'application/x-ndjson; charset=utf-8'); + if (total !== undefined) { + response.setHeader('X-Total-Count', String(total)); + } + if (page !== undefined) { + response.setHeader('X-Page', String(page)); + } + if (limit !== undefined) { + response.setHeader('X-Limit', String(limit)); + } + + // Build a PassThrough stream and write each item as a JSON line. + const passThrough = new PassThrough(); + + // Schedule writes asynchronously so the stream is returned to NestJS + // before we start pushing data (avoids blocking the event loop). + setImmediate(() => { + try { + for (const item of items) { + passThrough.write(JSON.stringify(item) + '\n'); + } + passThrough.end(); + } catch (err) { + passThrough.destroy(err instanceof Error ? err : new Error(String(err))); + } + }); + + // Return a StreamableFile so NestJS hands off the stream to Express + // and uses Transfer-Encoding: chunked automatically. + return new Observable(subscriber => { + subscriber.next(new StreamableFile(passThrough, { + type: 'application/x-ndjson', + })); + subscriber.complete(); + }); + }), + ); + } +} diff --git a/src/oracle/oracle.controller.ts b/src/oracle/oracle.controller.ts index b678cf4..1b49b94 100644 --- a/src/oracle/oracle.controller.ts +++ b/src/oracle/oracle.controller.ts @@ -7,7 +7,9 @@ import { Post, Query, UseGuards, + UseInterceptors, } from "@nestjs/common"; +import { StreamingInterceptor } from "../common/interceptors/streaming.interceptor"; import { ApiTags, ApiOperation, @@ -95,22 +97,38 @@ export class OracleController { * Oracle data is public and accessible to all users. * * Rate limited: 60 requests/minute per IP (global ThrottleGuard) + * + * #440 — supports NDJSON streaming to reduce peak memory for large result + * sets. Pass `?stream=true` or `Accept: application/x-ndjson` to activate. */ @Get("readings") @Throttle({ default: { limit: 60, ttl: 60000 } }) + @UseInterceptors(StreamingInterceptor) @ApiOperation({ summary: "List all stored oracle readings", description: - "Public endpoint. Returns latest oracle readings ordered by submission time (most recent first). Rate limited to 60 requests/minute per IP.", + "Public endpoint. Returns latest oracle readings ordered by submission time (most recent first). " + + "Pass `?stream=true` or `Accept: application/x-ndjson` to receive the data array as NDJSON (one item per line), " + + "which reduces server memory usage for large result sets. " + + "Rate limited to 60 requests/minute per IP.", }) @ApiQuery({ name: "limit", required: false, description: "Max rows to return (default 100, max 500)", }) + @ApiQuery({ + name: "stream", + required: false, + description: + "Set to 'true' to receive the response as NDJSON (one JSON object per line). " + + "Alternatively send Accept: application/x-ndjson. " + + "Pagination metadata is returned in X-Total-Count, X-Page, X-Limit headers.", + example: "true", + }) @ApiResponse({ status: 200, - description: "Array of oracle readings", + description: "Array of oracle readings (JSON envelope) or NDJSON stream when ?stream=true", schema: { example: { success: true, diff --git a/src/policy/policy.controller.ts b/src/policy/policy.controller.ts index cee71a9..1df778c 100644 --- a/src/policy/policy.controller.ts +++ b/src/policy/policy.controller.ts @@ -15,9 +15,11 @@ import { ForbiddenException, BadRequestException, UseGuards, + UseInterceptors, Req, UnauthorizedException, } from '@nestjs/common'; +import { StreamingInterceptor } from '../common/interceptors/streaming.interceptor'; import { Observable } from 'rxjs'; import { ApiTags, @@ -53,9 +55,16 @@ export class PolicyController { /** GET /api/v1/products — list all active insurance products with pagination */ @Get('products') + @UseInterceptors(StreamingInterceptor) @ApiOperation({ summary: 'List all active insurance products with pagination' }) @ApiQuery({ name: 'page', required: false, description: 'Page number (default 1)', example: 1 }) @ApiQuery({ name: 'limit', required: false, description: 'Items per page, max 100 (default 20)', example: 20 }) + @ApiQuery({ + name: 'stream', + required: false, + description: "Set to 'true' to receive the data array as NDJSON (one item per line). Alternatively send Accept: application/x-ndjson. Pagination metadata available in X-Total-Count, X-Page, X-Limit headers.", + example: 'true', + }) @ApiResponse({ status: 200, description: 'Returns paginated products — { success, data, total, page, limit }', @@ -83,10 +92,17 @@ export class PolicyController { /** GET /api/v1/policies/me?page=&limit= — get paginated policies for the authenticated wallet */ @Get('policies/me') @UseGuards(JwtAuthGuard) + @UseInterceptors(StreamingInterceptor) @ApiBearerAuth() @ApiOperation({ summary: 'Get paginated policies for the authenticated wallet' }) @ApiQuery({ name: 'page', required: false, description: 'Page number (default 1)', example: 1 }) @ApiQuery({ name: 'limit', required: false, description: 'Items per page, max 100 (default 20)', example: 20 }) + @ApiQuery({ + name: 'stream', + required: false, + description: "Set to 'true' to receive the data array as NDJSON (one item per line). Alternatively send Accept: application/x-ndjson. Pagination metadata available in X-Total-Count, X-Page, X-Limit headers.", + example: 'true', + }) @ApiResponse({ status: 200, description: 'Returns paginated policies — { success, data, total, page, limit }', From 1a57dc0a708fcd7915ccef22d005fea1083e8f2d Mon Sep 17 00:00:00 2001 From: natho080 Date: Wed, 26 Aug 2026 16:44:30 +0100 Subject: [PATCH 4/4] feat(health): add Stellar RPC connectivity check via getLatestLedger --- src/health/dto/health-response.dto.ts | 9 +++++++++ src/health/health.controller.ts | 27 ++++++++++++++++++++++++++- src/stellar/stellar.service.ts | 23 +++++++++++++++++++++++ 3 files changed, 58 insertions(+), 1 deletion(-) diff --git a/src/health/dto/health-response.dto.ts b/src/health/dto/health-response.dto.ts index d4414e7..f21c29a 100644 --- a/src/health/dto/health-response.dto.ts +++ b/src/health/dto/health-response.dto.ts @@ -43,6 +43,15 @@ export class StellarCheckDto { @ApiProperty({ description: 'Keeper account native XLM balance (7-decimal fixed point)', required: false }) keeperBalanceXlm?: string; + @ApiProperty({ description: 'Stellar RPC (Soroban) connectivity status', enum: ['ok', 'error'], required: false }) + rpcStatus?: 'ok' | 'error'; + + @ApiProperty({ description: 'Stellar RPC round-trip latency in milliseconds', required: false }) + rpcLatencyMs?: number; + + @ApiProperty({ description: 'Latest ledger sequence number returned by Stellar RPC', required: false }) + rpcLedger?: number; + @ApiProperty({ description: 'Error message when status is "error"', required: false }) error?: string; } diff --git a/src/health/health.controller.ts b/src/health/health.controller.ts index 7f11d09..d8db81b 100644 --- a/src/health/health.controller.ts +++ b/src/health/health.controller.ts @@ -61,6 +61,10 @@ export class HealthController { let stellarStatus: 'ok' | 'error' = 'ok'; let stellarError: string | undefined; let keeperBalanceXlm: string | undefined; + // #441 — Direct Soroban RPC connectivity check fields + let stellarRpcStatus: 'ok' | 'error' = 'ok'; + let stellarRpcLatencyMs: number | undefined; + let stellarRpcLedger: number | undefined; let queueStatus: 'ok' | 'error' = 'ok'; let queueError: string | undefined; @@ -96,6 +100,24 @@ export class HealthController { // Non-fatal: pg_stat_activity may be restricted on managed databases. } + // #441 — Direct Stellar RPC (Soroban) connectivity check. + // getLatestLedger is the lightest available probe: it requires no + // authentication, touches no account state, and always succeeds when + // the RPC node is reachable. We record latency so ops can distinguish + // a slow node from a fully unreachable one. A failure here is fatal + // (stellarStatus → 'error') because contract invocations — claims, + // policy submissions, oracle writes — all depend on the Soroban RPC. + try { + const rpcProbe = await this.stellar.checkRpcConnectivity(HEALTH_CHECK_RPC_TIMEOUT_MS); + stellarRpcLatencyMs = rpcProbe.latencyMs; + stellarRpcLedger = rpcProbe.ledger; + } catch (err) { + stellarRpcStatus = 'error'; + stellarStatus = 'error'; + stellarError = `Stellar RPC unreachable: ${err instanceof Error ? err.message : String(err)}`; + this.logger.error(`Health check: ${stellarError}`); + } + try { keeperBalanceXlm = await this.stellar.getAccountBalance( this.stellar.keeperKeypair.publicKey(), @@ -117,7 +139,7 @@ export class HealthController { } } catch (err) { stellarStatus = 'error'; - this.logger.error(`Health check Stellar RPC failed: ${err instanceof Error ? err.message : String(err)}`); + this.logger.error(`Health check Stellar keeper/Horizon failed: ${err instanceof Error ? err.message : String(err)}`); } // #403 — Redis/message queue connectivity check. @@ -229,6 +251,9 @@ export class HealthController { }, stellar: { status: stellarStatus, + rpcStatus: stellarRpcStatus, + ...(stellarRpcLatencyMs !== undefined ? { rpcLatencyMs: stellarRpcLatencyMs } : {}), + ...(stellarRpcLedger !== undefined ? { rpcLedger: stellarRpcLedger } : {}), ...(keeperBalanceXlm !== undefined ? { keeperBalanceXlm } : {}), ...(stellarError ? { error: stellarError } : {}), }, diff --git a/src/stellar/stellar.service.ts b/src/stellar/stellar.service.ts index 5ddf0c9..63ba053 100644 --- a/src/stellar/stellar.service.ts +++ b/src/stellar/stellar.service.ts @@ -456,6 +456,29 @@ export class StellarService { return nativeBalance.balance; } + /** + * #441 — Verify Stellar RPC (Soroban) connectivity by calling getLatestLedger, + * the lightest available RPC probe (no auth, no on-chain state required). + * Returns the ledger sequence number and round-trip latency so the health + * endpoint can surface both reachability and basic responsiveness. + * + * @param timeoutMs Maximum time to wait in milliseconds (default 10s). + * Health checks should pass HEALTH_CHECK_RPC_TIMEOUT_MS + * to keep probe latency bounded. + */ + async checkRpcConnectivity(timeoutMs?: number): Promise<{ latencyMs: number; ledger: number }> { + const start = Date.now(); + const result = await this.withTimeout( + this.rpc.getLatestLedger(), + 'getLatestLedger', + timeoutMs, + ); + return { + latencyMs: Date.now() - start, + ledger: result.sequence, + }; + } + /** Return the current network passphrase. */ get networkPassphrase(): string { return this.network;