diff --git a/.env.example b/.env.example index 377ed0c..51a80d2 100644 --- a/.env.example +++ b/.env.example @@ -46,6 +46,11 @@ JOBS_ENABLED=true INDEXER_INTERVAL=30 TX_STATUS_INTERVAL=15 REMINDER_CRON=0 9 * * * +JOB_STALE_THRESHOLD_INDEXER_SECONDS=180 +JOB_STALE_THRESHOLD_TRANSACTION_STATUS_CHECKER_SECONDS=360 +JOB_STALE_THRESHOLD_LOAN_PAYMENT_REMINDER_SECONDS=93600 +JOB_STALE_THRESHOLD_NONCE_CLEANUP_SECONDS=7200 +JOB_STALE_THRESHOLD_SUPABASE_KEEPALIVE_SECONDS=345600 # Sentry SENTRY_DSN= diff --git a/context/progress-tracker.md b/context/progress-tracker.md index 3a5dd01..dfe4028 100644 --- a/context/progress-tracker.md +++ b/context/progress-tracker.md @@ -6,6 +6,15 @@ pure chore/docs commits). Direct pushes to main must also be logged here. --- +## 2026-07-22 + +- Added shared background-job monitoring for the indexer, transaction status + checker, loan payment reminders, nonce cleanup, and Supabase keep-alive. + Successful runs now publish Prometheus timestamps, failed runs track + consecutive failures and report to Sentry at power-of-two intervals to avoid + alert floods, and `/health` degrades when a job exceeds its independently + configurable staleness threshold. + ## 2026-07-19 - Wired `LiquidityContractClient` (restored under `src/blockchain/contracts/liquidity-contract.client.ts`) into `LiquidityService` constructor. diff --git a/src/app.module.ts b/src/app.module.ts index 65d297a..9d70147 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -29,10 +29,12 @@ import { CreditScoringModule } from './modules/credit-scoring/credit-scoring.mod import { AdminModule } from './modules/admin/admin.module'; import { CorrelationIdMiddleware } from './common/logger/correlation-id.middleware'; import { StateReconciliationModule } from './jobs/state-reconciliation/state-reconciliation.module'; +import { validateEnvironment } from './config/env'; +import { JobMonitorModule } from './jobs/monitoring/job-monitor.module'; @Module({ imports: [ - ConfigModule.forRoot({ isGlobal: true }), + ConfigModule.forRoot({ isGlobal: true, validate: validateEnvironment }), SentryModule.forRoot(), ScheduleModule.forRoot(), LoggerModule, @@ -45,6 +47,7 @@ import { StateReconciliationModule } from './jobs/state-reconciliation/state-rec AuthModule, HealthModule, MetricsModule, + JobMonitorModule, LoansModule, ReputationModule, UsersModule, diff --git a/src/config/env.ts b/src/config/env.ts index e69de29..b3b383c 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -0,0 +1,48 @@ +export const JOB_STALE_THRESHOLD_CONFIG = { + indexer: { + env: 'JOB_STALE_THRESHOLD_INDEXER_SECONDS', + defaultSeconds: 180, + }, + transactionStatusChecker: { + env: 'JOB_STALE_THRESHOLD_TRANSACTION_STATUS_CHECKER_SECONDS', + defaultSeconds: 360, + }, + loanPaymentReminder: { + env: 'JOB_STALE_THRESHOLD_LOAN_PAYMENT_REMINDER_SECONDS', + defaultSeconds: 93_600, + }, + nonceCleanup: { + env: 'JOB_STALE_THRESHOLD_NONCE_CLEANUP_SECONDS', + defaultSeconds: 7_200, + }, + supabaseKeepAlive: { + env: 'JOB_STALE_THRESHOLD_SUPABASE_KEEPALIVE_SECONDS', + defaultSeconds: 345_600, + }, +} as const; + +export type MonitoredJobName = keyof typeof JOB_STALE_THRESHOLD_CONFIG; + +export function validateEnvironment( + config: Record, +): Record { + const validated = { ...config }; + + for (const { env, defaultSeconds } of Object.values( + JOB_STALE_THRESHOLD_CONFIG, + )) { + const rawValue = config[env]; + const parsed = + rawValue === undefined || rawValue === '' + ? defaultSeconds + : Number(rawValue); + + if (!Number.isInteger(parsed) || parsed <= 0) { + throw new Error(`${env} must be a positive integer number of seconds`); + } + + validated[env] = parsed; + } + + return validated; +} diff --git a/src/indexer/indexer.service.ts b/src/indexer/indexer.service.ts index 1a2c28e..eac9190 100644 --- a/src/indexer/indexer.service.ts +++ b/src/indexer/indexer.service.ts @@ -5,6 +5,7 @@ import * as StellarSdk from 'stellar-sdk'; import { SupabaseService } from '../database/supabase.client'; import { SorobanService } from '../blockchain/soroban/soroban.service'; import { EventParserService } from './event-parser.service'; +import { JobMonitorService } from '../jobs/monitoring/job-monitor.service'; import { ParsedContractEvent, LoanEventType, @@ -31,6 +32,7 @@ export class IndexerService { private readonly sorobanService: SorobanService, private readonly supabaseService: SupabaseService, private readonly eventParser: EventParserService, + private readonly jobMonitorService: JobMonitorService, ) { this.loanContractId = this.configService.get('CREDIT_LINE_CONTRACT_ID') || ''; @@ -49,10 +51,18 @@ export class IndexerService { this.logger.log('Blockchain indexer cycle started'); try { - await this.indexLoanContract(); - await this.indexReputationContract(); + const loanError = await this.indexLoanContract(); + const reputationError = await this.indexReputationContract(); + const runError = loanError ?? reputationError; + + if (runError) { + this.jobMonitorService.recordFailure('indexer', runError); + } else { + this.jobMonitorService.recordSuccess('indexer'); + } } catch (error) { this.logger.error({ error: error.message, stack: error.stack }, 'Indexer cycle failed'); + this.jobMonitorService.recordFailure('indexer', error); } finally { this.isRunning = false; } @@ -60,25 +70,29 @@ export class IndexerService { this.logger.log('Blockchain indexer cycle completed'); } - private async indexLoanContract(): Promise { + private async indexLoanContract(): Promise { try { await this.indexContract(this.loanContractId, 'loan'); + return null; } catch (error) { this.logger.error( { error: error.message, stack: error.stack }, 'Failed to index loan contract events — will retry next cycle', ); + return error instanceof Error ? error : new Error(String(error)); } } - private async indexReputationContract(): Promise { + private async indexReputationContract(): Promise { try { await this.indexContract(this.reputationContractId, 'reputation'); + return null; } catch (error) { this.logger.error( { error: error.message, stack: error.stack }, 'Failed to index reputation contract events — will retry next cycle', ); + return error instanceof Error ? error : new Error(String(error)); } } diff --git a/src/jobs/loan-payment-reminder/loan-payment-reminder.service.ts b/src/jobs/loan-payment-reminder/loan-payment-reminder.service.ts index a8bbb96..1bfcdba 100644 --- a/src/jobs/loan-payment-reminder/loan-payment-reminder.service.ts +++ b/src/jobs/loan-payment-reminder/loan-payment-reminder.service.ts @@ -8,6 +8,7 @@ import { ReminderSummary, ReminderType, } from './interfaces/reminder.interfaces'; +import { JobMonitorService } from '../monitoring/job-monitor.service'; const REMINDER_TITLES: Record = { payment_reminder_3d: 'Payment Due in 3 Days', @@ -37,12 +38,16 @@ export class LoanPaymentReminderService { private readonly logger = new Logger(LoanPaymentReminderService.name); private isRunning = false; - constructor(private readonly supabaseService: SupabaseService) {} + constructor( + private readonly supabaseService: SupabaseService, + private readonly jobMonitorService: JobMonitorService, + ) {} @Cron('0 9 * * *') async sendPaymentReminders(): Promise { if (this.isRunning) return; this.isRunning = true; + let runFailure: unknown; const summary: ReminderSummary = { total: 0, @@ -86,13 +91,23 @@ export class LoanPaymentReminderService { summary.created++; summary.breakdown[candidate.reminderType]++; } catch (error) { + runFailure ??= error; summary.failed++; this.logger.error(`Failed to process reminder candidate for loan ${candidate.loan.loan_id}: ${error.message}`); } } } catch (error) { + runFailure = error; this.logger.error(`Fatal error during reminder job: ${error.message}`); } finally { + if (runFailure) { + this.jobMonitorService.recordFailure( + 'loanPaymentReminder', + runFailure, + ); + } else { + this.jobMonitorService.recordSuccess('loanPaymentReminder'); + } this.logger.log(`Reminder job complete — created: ${summary.created}, skipped: ${summary.skipped}, failed: ${summary.failed}`); this.isRunning = false; } diff --git a/src/jobs/monitoring/job-monitor.module.ts b/src/jobs/monitoring/job-monitor.module.ts new file mode 100644 index 0000000..20c11f8 --- /dev/null +++ b/src/jobs/monitoring/job-monitor.module.ts @@ -0,0 +1,9 @@ +import { Global, Module } from '@nestjs/common'; +import { JobMonitorService } from './job-monitor.service'; + +@Global() +@Module({ + providers: [JobMonitorService], + exports: [JobMonitorService], +}) +export class JobMonitorModule {} diff --git a/src/jobs/monitoring/job-monitor.service.spec.ts b/src/jobs/monitoring/job-monitor.service.spec.ts new file mode 100644 index 0000000..0a4b0b6 --- /dev/null +++ b/src/jobs/monitoring/job-monitor.service.spec.ts @@ -0,0 +1,90 @@ +import { ConfigService } from '@nestjs/config'; +import * as Sentry from '@sentry/nestjs'; +import { + JOB_STALE_THRESHOLD_CONFIG, + validateEnvironment, +} from '../../config/env'; +import { MetricsService } from '../../modules/metrics/metrics.service'; +import { JobMonitorService } from './job-monitor.service'; + +describe('JobMonitorService', () => { + const metricsService = { + setJobLastSuccessTimestamp: jest.fn(), + setJobConsecutiveFailures: jest.fn(), + }; + const configService = { + get: jest.fn((_key: string, defaultValue: number) => defaultValue), + }; + + let service: JobMonitorService; + + beforeEach(() => { + jest.clearAllMocks(); + service = new JobMonitorService( + configService as unknown as ConfigService, + metricsService as unknown as MetricsService, + ); + }); + + it('records success timestamps and resets the failure gauge', () => { + service.recordFailure('indexer', new Error('temporary failure')); + service.recordSuccess('indexer'); + + expect(metricsService.setJobLastSuccessTimestamp).toHaveBeenCalledWith( + 'indexer', + expect.any(Number), + ); + expect(metricsService.setJobConsecutiveFailures).toHaveBeenLastCalledWith( + 'indexer', + 0, + ); + }); + + it('captures the first failure and then powers of two to avoid alert noise', () => { + const captureException = jest + .spyOn(Sentry, 'captureException') + .mockReturnValue('event-id'); + + for (let failure = 1; failure <= 8; failure++) { + service.recordFailure('transactionStatusChecker', new Error('down')); + } + + expect(captureException).toHaveBeenCalledTimes(4); + expect(captureException).toHaveBeenLastCalledWith( + expect.any(Error), + expect.objectContaining({ + tags: { job: 'transactionStatusChecker' }, + contexts: { + job: { + name: 'transactionStatusChecker', + consecutiveFailures: 8, + }, + }, + }), + ); + }); + + it('reports a job as stale after its configured threshold', () => { + service.recordSuccess('indexer'); + const statuses = service.getHealthStatuses( + new Date(Date.now() + 181_000), + ); + + expect(statuses.find((status) => status.job === 'indexer')).toEqual( + expect.objectContaining({ status: 'stale', thresholdSeconds: 180 }), + ); + }); + + it('validates and applies job threshold defaults at startup', () => { + const validated = validateEnvironment({}); + expect( + validated[JOB_STALE_THRESHOLD_CONFIG.indexer.env], + ).toBe(JOB_STALE_THRESHOLD_CONFIG.indexer.defaultSeconds); + + expect(() => + validateEnvironment({ + [JOB_STALE_THRESHOLD_CONFIG.indexer.env]: '0', + }), + ).toThrow('must be a positive integer'); + }); +}); diff --git a/src/jobs/monitoring/job-monitor.service.ts b/src/jobs/monitoring/job-monitor.service.ts new file mode 100644 index 0000000..85842ed --- /dev/null +++ b/src/jobs/monitoring/job-monitor.service.ts @@ -0,0 +1,127 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import * as Sentry from '@sentry/nestjs'; +import { + JOB_STALE_THRESHOLD_CONFIG, + MonitoredJobName, +} from '../../config/env'; +import { MetricsService } from '../../modules/metrics/metrics.service'; + +interface JobState { + lastSuccessAt: Date | null; + consecutiveFailures: number; +} + +export interface JobHealthStatus { + job: MonitoredJobName; + status: 'ok' | 'stale'; + lastSuccessAt: string | null; + ageSeconds: number; + thresholdSeconds: number; + consecutiveFailures: number; +} + +@Injectable() +export class JobMonitorService { + private readonly logger = new Logger(JobMonitorService.name); + private readonly startedAt = new Date(); + private readonly states = new Map(); + + constructor( + private readonly configService: ConfigService, + private readonly metricsService: MetricsService, + ) { + for (const job of Object.keys( + JOB_STALE_THRESHOLD_CONFIG, + ) as MonitoredJobName[]) { + this.states.set(job, { lastSuccessAt: null, consecutiveFailures: 0 }); + this.metricsService.setJobConsecutiveFailures(job, 0); + } + } + + recordSuccess(job: MonitoredJobName): void { + const now = new Date(); + this.states.set(job, { lastSuccessAt: now, consecutiveFailures: 0 }); + this.metricsService.setJobLastSuccessTimestamp( + job, + Math.floor(now.getTime() / 1000), + ); + this.metricsService.setJobConsecutiveFailures(job, 0); + } + + recordFailure(job: MonitoredJobName, error: unknown): void { + const current = this.getState(job); + const consecutiveFailures = current.consecutiveFailures + 1; + this.states.set(job, { ...current, consecutiveFailures }); + this.metricsService.setJobConsecutiveFailures(job, consecutiveFailures); + + const capturedError = + error instanceof Error ? error : new Error(String(error)); + + this.logger.error( + { + job, + consecutiveFailures, + error: capturedError.message, + stack: capturedError.stack, + }, + 'Background job run failed', + ); + + if (this.shouldCaptureFailure(consecutiveFailures)) { + Sentry.captureException(capturedError, { + tags: { job }, + contexts: { + job: { name: job, consecutiveFailures }, + }, + }); + } + } + + getHealthStatuses(now = new Date()): JobHealthStatus[] { + return (Object.keys( + JOB_STALE_THRESHOLD_CONFIG, + ) as MonitoredJobName[]).map((job) => { + const state = this.getState(job); + const referenceTime = state.lastSuccessAt ?? this.startedAt; + const ageSeconds = Math.max( + 0, + Math.floor((now.getTime() - referenceTime.getTime()) / 1000), + ); + const thresholdSeconds = this.getThresholdSeconds(job); + + return { + job, + status: ageSeconds > thresholdSeconds ? 'stale' : 'ok', + lastSuccessAt: state.lastSuccessAt?.toISOString() ?? null, + ageSeconds, + thresholdSeconds, + consecutiveFailures: state.consecutiveFailures, + }; + }); + } + + private getState(job: MonitoredJobName): JobState { + return ( + this.states.get(job) ?? { + lastSuccessAt: null, + consecutiveFailures: 0, + } + ); + } + + private getThresholdSeconds(job: MonitoredJobName): number { + const definition = JOB_STALE_THRESHOLD_CONFIG[job]; + return this.configService.get( + definition.env, + definition.defaultSeconds, + ); + } + + private shouldCaptureFailure(consecutiveFailures: number): boolean { + return ( + consecutiveFailures > 0 && + (consecutiveFailures & (consecutiveFailures - 1)) === 0 + ); + } +} diff --git a/src/jobs/nonce-cleanup/nonce-cleanup.service.ts b/src/jobs/nonce-cleanup/nonce-cleanup.service.ts index c42aea7..dd3ed85 100644 --- a/src/jobs/nonce-cleanup/nonce-cleanup.service.ts +++ b/src/jobs/nonce-cleanup/nonce-cleanup.service.ts @@ -1,15 +1,26 @@ import { Injectable, Logger } from '@nestjs/common'; import { Cron, CronExpression } from '@nestjs/schedule'; import { SupabaseService } from '../../database/supabase.client'; +import { JobMonitorService } from '../monitoring/job-monitor.service'; @Injectable() export class NonceCleanupService { private readonly logger = new Logger(NonceCleanupService.name); + private isRunning = false; - constructor(private readonly supabaseService: SupabaseService) {} + constructor( + private readonly supabaseService: SupabaseService, + private readonly jobMonitorService: JobMonitorService, + ) {} @Cron(CronExpression.EVERY_HOUR) async cleanupExpiredNonces(): Promise { + if (this.isRunning) { + this.logger.debug('Nonce cleanup already running, skipping'); + return; + } + this.isRunning = true; + try { const client = this.supabaseService.getServiceRoleClient(); @@ -26,8 +37,12 @@ export class NonceCleanupService { } this.logger.log(`Deleted ${count ?? 0} expired nonces`); + this.jobMonitorService.recordSuccess('nonceCleanup'); } catch (error) { this.logger.error('Nonce cleanup failed', error); + this.jobMonitorService.recordFailure('nonceCleanup', error); + } finally { + this.isRunning = false; } } } diff --git a/src/jobs/supabase-keepalive/supabase-keepalive.service.ts b/src/jobs/supabase-keepalive/supabase-keepalive.service.ts index 8227afe..1c69269 100644 --- a/src/jobs/supabase-keepalive/supabase-keepalive.service.ts +++ b/src/jobs/supabase-keepalive/supabase-keepalive.service.ts @@ -1,5 +1,6 @@ import { Injectable, OnModuleInit, Logger } from '@nestjs/common'; import { SupabaseService } from '../../database/supabase.client'; +import { JobMonitorService } from '../monitoring/job-monitor.service'; /** * Keeps the Supabase project active by issuing a lightweight read every 3 days. @@ -10,8 +11,12 @@ import { SupabaseService } from '../../database/supabase.client'; @Injectable() export class SupabaseKeepAliveService implements OnModuleInit { private readonly logger = new Logger(SupabaseKeepAliveService.name); + private isRunning = false; - constructor(private readonly supabaseService: SupabaseService) {} + constructor( + private readonly supabaseService: SupabaseService, + private readonly jobMonitorService: JobMonitorService, + ) {} onModuleInit(): void { // Run every 3 days in milliseconds @@ -21,6 +26,12 @@ export class SupabaseKeepAliveService implements OnModuleInit { } async ping(): Promise { + if (this.isRunning) { + this.logger.debug('Supabase keep-alive already running, skipping'); + return; + } + this.isRunning = true; + try { // Lightweight read (equivalent to `SELECT 1 FROM users LIMIT 1`) that keeps // the Supabase project active so the free tier does not pause it for @@ -36,8 +47,12 @@ export class SupabaseKeepAliveService implements OnModuleInit { } this.logger.log('Supabase keep-alive ping successful'); + this.jobMonitorService.recordSuccess('supabaseKeepAlive'); } catch (error) { this.logger.error('Keep-alive ping failed', error); + this.jobMonitorService.recordFailure('supabaseKeepAlive', error); + } finally { + this.isRunning = false; } } } diff --git a/src/jobs/transaction-status-checker/transaction-status-checker.service.ts b/src/jobs/transaction-status-checker/transaction-status-checker.service.ts index fcc8dee..696d937 100644 --- a/src/jobs/transaction-status-checker/transaction-status-checker.service.ts +++ b/src/jobs/transaction-status-checker/transaction-status-checker.service.ts @@ -4,6 +4,7 @@ import { ConfigService } from '@nestjs/config'; import * as StellarSdk from 'stellar-sdk'; import { SupabaseService } from '../../database/supabase.client'; import { TransactionType } from '../../modules/transactions/dto/submit-transaction-request.dto'; +import { JobMonitorService } from '../monitoring/job-monitor.service'; interface PendingTransaction { id: string; @@ -56,6 +57,7 @@ export class TransactionStatusCheckerService { constructor( private readonly configService: ConfigService, private readonly supabaseService: SupabaseService, + private readonly jobMonitorService: JobMonitorService, ) { const horizonUrl = this.configService.get('STELLAR_HORIZON_URL') || @@ -76,6 +78,7 @@ export class TransactionStatusCheckerService { return; } this.isRunning = true; + let runFailure: unknown; this.logger.log( { @@ -140,6 +143,7 @@ export class TransactionStatusCheckerService { ); } } catch (error) { + runFailure ??= error; this.logger.error( { context: 'TransactionStatusCheckerService', @@ -153,6 +157,7 @@ export class TransactionStatusCheckerService { } } } catch (error) { + runFailure = error; this.logger.error( { context: 'TransactionStatusCheckerService', @@ -164,6 +169,14 @@ export class TransactionStatusCheckerService { ); } finally { await this.cleanupOldTransactions(); + if (runFailure) { + this.jobMonitorService.recordFailure( + 'transactionStatusChecker', + runFailure, + ); + } else { + this.jobMonitorService.recordSuccess('transactionStatusChecker'); + } this.isRunning = false; this.logger.log( { diff --git a/src/modules/health/health.service.ts b/src/modules/health/health.service.ts index f212928..a9aebb6 100644 --- a/src/modules/health/health.service.ts +++ b/src/modules/health/health.service.ts @@ -1,6 +1,7 @@ import { Injectable, Logger } from '@nestjs/common'; import { SupabaseService } from '../../database/supabase.client'; import { HorizonClientService } from '../../stellar/horizon-client.service'; +import { JobMonitorService } from '../../jobs/monitoring/job-monitor.service'; @Injectable() export class HealthService { @@ -9,6 +10,7 @@ export class HealthService { constructor( private readonly horizonClientService: HorizonClientService, private readonly supabaseService: SupabaseService, + private readonly jobMonitorService: JobMonitorService, ) {} async check() { @@ -17,8 +19,12 @@ export class HealthService { this.checkHorizon(), this.checkIndexerLag(), ]); + const jobs = this.jobMonitorService.getHealthStatuses(); + const jobsStatus = jobs.some((job) => job.status === 'stale') + ? 'degraded' + : 'ok'; - const allOk = [db, horizon, indexer].every( + const allOk = [db, horizon, indexer, { status: jobsStatus }].every( (c) => c.status === 'ok', ); @@ -30,6 +36,10 @@ export class HealthService { database: db, horizon: horizon, indexer: indexer, + jobs: { + status: jobsStatus, + details: jobs, + }, }, }; } @@ -121,4 +131,4 @@ export class HealthService { return 0; } } -} \ No newline at end of file +} diff --git a/src/modules/metrics/metrics.service.ts b/src/modules/metrics/metrics.service.ts index 5f35f5b..1aef016 100644 --- a/src/modules/metrics/metrics.service.ts +++ b/src/modules/metrics/metrics.service.ts @@ -13,6 +13,9 @@ export const INDEXER_LAG = 'indexer_lag_ledgers'; export const HORIZON_HEALTH = 'horizon_up'; export const DB_POOL_OPEN = 'db_pool_open'; export const RECONCILIATION_DRIFT = 'state_reconciliation_drift'; +export const JOB_LAST_SUCCESS_TIMESTAMP_SECONDS = + 'job_last_success_timestamp_seconds'; +export const JOB_CONSECUTIVE_FAILURES = 'job_consecutive_failures'; export const metricProviders = [ makeCounterProvider({ @@ -43,6 +46,16 @@ export const metricProviders = [ help: 'Current reconciliation drift count by mismatch type', labelNames: ['type'] as const, }), + makeGaugeProvider({ + name: JOB_LAST_SUCCESS_TIMESTAMP_SECONDS, + help: 'Unix timestamp of the last successful run for a background job', + labelNames: ['job'] as const, + }), + makeGaugeProvider({ + name: JOB_CONSECUTIVE_FAILURES, + help: 'Number of consecutive failed runs for a background job', + labelNames: ['job'] as const, + }), ]; @Injectable() @@ -60,6 +73,10 @@ export class MetricsService { private readonly dbPoolOpen: Gauge, @InjectMetric(RECONCILIATION_DRIFT) private readonly reconciliationDrift: Gauge, + @InjectMetric(JOB_LAST_SUCCESS_TIMESTAMP_SECONDS) + private readonly jobLastSuccessTimestamp: Gauge, + @InjectMetric(JOB_CONSECUTIVE_FAILURES) + private readonly jobConsecutiveFailures: Gauge, ) {} async getMetrics(): Promise { @@ -90,4 +107,12 @@ export class MetricsService { setReconciliationDrift(type: string, count: number): void { this.reconciliationDrift.labels(type).set(count); } + + setJobLastSuccessTimestamp(job: string, timestampSeconds: number): void { + this.jobLastSuccessTimestamp.labels(job).set(timestampSeconds); + } + + setJobConsecutiveFailures(job: string, count: number): void { + this.jobConsecutiveFailures.labels(job).set(count); + } } diff --git a/test/unit/jobs/blockchain-indexer/blockchain-indexer.processor.spec.ts b/test/unit/jobs/blockchain-indexer/blockchain-indexer.processor.spec.ts index b6f6485..2afebc3 100644 --- a/test/unit/jobs/blockchain-indexer/blockchain-indexer.processor.spec.ts +++ b/test/unit/jobs/blockchain-indexer/blockchain-indexer.processor.spec.ts @@ -4,6 +4,7 @@ import { IndexerService } from '../../../../src/indexer/indexer.service'; import { EventParserService } from '../../../../src/indexer/event-parser.service'; import { SupabaseService } from '../../../../src/database/supabase.client'; import { SorobanService } from '../../../../src/blockchain/soroban/soroban.service'; +import { JobMonitorService } from '../../../../src/jobs/monitoring/job-monitor.service'; import { LoanEventType, ReputationEventType, @@ -99,6 +100,13 @@ describe('IndexerService', () => { EventParserService, { provide: SupabaseService, useValue: mockSupabaseService }, { provide: SorobanService, useValue: mockSorobanService }, + { + provide: JobMonitorService, + useValue: { + recordSuccess: jest.fn(), + recordFailure: jest.fn(), + }, + }, { provide: ConfigService, useValue: { @@ -202,6 +210,13 @@ describe('IndexerService', () => { EventParserService, { provide: SupabaseService, useValue: mockSupabaseService }, { provide: SorobanService, useValue: mockSorobanService }, + { + provide: JobMonitorService, + useValue: { + recordSuccess: jest.fn(), + recordFailure: jest.fn(), + }, + }, { provide: ConfigService, useValue: { get: jest.fn().mockReturnValue('') }, diff --git a/test/unit/jobs/monitoring/job-monitor.service.spec.ts b/test/unit/jobs/monitoring/job-monitor.service.spec.ts new file mode 100644 index 0000000..ab829f9 --- /dev/null +++ b/test/unit/jobs/monitoring/job-monitor.service.spec.ts @@ -0,0 +1 @@ +import '../../../../src/jobs/monitoring/job-monitor.service.spec'; diff --git a/test/unit/modules/health/health.service.spec.ts b/test/unit/modules/health/health.service.spec.ts index f766a12..9be28c6 100644 --- a/test/unit/modules/health/health.service.spec.ts +++ b/test/unit/modules/health/health.service.spec.ts @@ -3,6 +3,7 @@ import { HealthService } from '../../../../src/modules/health/health.service'; import { SupabaseService } from '../../../../src/database/supabase.client'; import { ConfigService } from '@nestjs/config'; import { HorizonClientService } from '../../../../src/stellar/horizon-client.service'; +import { JobMonitorService } from '../../../../src/jobs/monitoring/job-monitor.service'; describe('HealthService', () => { let service: HealthService; @@ -34,6 +35,19 @@ describe('HealthService', () => { getTransaction: jest.fn(), }; + const mockJobMonitorService = { + getHealthStatuses: jest.fn().mockReturnValue([ + { + job: 'indexer', + status: 'ok', + lastSuccessAt: null, + ageSeconds: 0, + thresholdSeconds: 180, + consecutiveFailures: 0, + }, + ]), + }; + beforeEach(async () => { const module: TestingModule = await Test.createTestingModule({ providers: [ @@ -46,6 +60,10 @@ describe('HealthService', () => { provide: HorizonClientService, useValue: mockHorizonClientService, }, + { + provide: JobMonitorService, + useValue: mockJobMonitorService, + }, ], }).compile(); @@ -78,8 +96,35 @@ describe('HealthService', () => { expect(result).toHaveProperty('timestamp'); expect(result).toHaveProperty('service', 'StepFi API'); expect(result).toHaveProperty('checks'); + expect(result.checks.jobs).toHaveProperty('status', 'ok'); expect(result.timestamp).toBeDefined(); }); + + it('should report degraded when a monitored job is stale', async () => { + jest.spyOn(service, 'checkDatabase').mockResolvedValue({ + status: 'ok', + database: 'connected', + message: 'Supabase reachable', + timestamp: new Date().toISOString(), + }); + jest.spyOn(service, 'checkHorizon').mockResolvedValue({ status: 'ok' }); + jest.spyOn(service, 'checkIndexerLag').mockResolvedValue({ status: 'ok' }); + mockJobMonitorService.getHealthStatuses.mockReturnValueOnce([ + { + job: 'indexer', + status: 'stale', + lastSuccessAt: '2026-07-22T00:00:00.000Z', + ageSeconds: 181, + thresholdSeconds: 180, + consecutiveFailures: 2, + }, + ]); + + const result = await service.check(); + + expect(result.status).toBe('degraded'); + expect(result.checks.jobs.status).toBe('degraded'); + }); }); describe('checkDatabase', () => {