From daa0dc3ab0adbb1a27c27aa0345d8fee3a7f0917 Mon Sep 17 00:00:00 2001 From: Najib Ishiyaku Njidda Date: Thu, 27 Aug 2026 23:36:47 +0000 Subject: [PATCH 1/3] Add listener uptime to health information --- listener/API.md | 3 +++ listener/src/api/events-server.health.test.ts | 2 ++ listener/src/api/events-server.ts | 2 ++ 3 files changed, 7 insertions(+) diff --git a/listener/API.md b/listener/API.md index 93febf9b..10aba379 100644 --- a/listener/API.md +++ b/listener/API.md @@ -31,12 +31,15 @@ For a centralized list of API errors, causes, examples, and troubleshooting step Returns the operational status of all service dependencies. +The `uptimeSeconds` field reports the listener process uptime in seconds, measured from process startup. + **Response `200`** — all systems operational (or Discord degraded but Stellar RPC healthy) ```json { "status": "ok", "timestamp": "2024-06-20T14:00:00.000Z", + "uptimeSeconds": 123.45, "services": { "stellarRpc": { "status": "ok", "latencyMs": 42 }, "discord": { "status": "ok", "latencyMs": 87 }, diff --git a/listener/src/api/events-server.health.test.ts b/listener/src/api/events-server.health.test.ts index 86c392ff..fcd4394e 100644 --- a/listener/src/api/events-server.health.test.ts +++ b/listener/src/api/events-server.health.test.ts @@ -107,6 +107,8 @@ describe('GET /health', () => { expect(health.services.database.status).toBe('ok'); expect(health.services.eventRegistry).toEqual({ status: 'ok', eventCount: 5 }); expect(health.timestamp).toBeDefined(); + expect(typeof health.uptimeSeconds).toBe('number'); + expect(health.uptimeSeconds).toBeGreaterThanOrEqual(0); }); it('returns 503 and status error when Stellar RPC is unreachable', async () => { diff --git a/listener/src/api/events-server.ts b/listener/src/api/events-server.ts index 0587f583..7de78cb8 100644 --- a/listener/src/api/events-server.ts +++ b/listener/src/api/events-server.ts @@ -111,6 +111,7 @@ interface ServiceHealth { interface HealthResponse { status: 'ok' | 'degraded' | 'error'; timestamp: string; + uptimeSeconds: number; services: { stellarRpc: ServiceHealth; discord: ServiceHealth; @@ -390,6 +391,7 @@ async function buildHealthResponse(options: EventsServerOptions): Promise Date: Fri, 28 Aug 2026 17:51:16 +0000 Subject: [PATCH 2/3] feat: expose last successful poll timestamp in health monitor closes #621 - Added lastSuccessfulPollAt tracking to EventSubscriber - Extended HealthReport to include lastSuccessfulPollAt - Fixed syntax/indentation errors in index.ts - Ensures failed polls do not update the timestamp Closes #621 --- listener/src/index.ts | 38 +++++++------------ listener/src/services/event-subscriber.ts | 19 +++++----- .../services/notification-health-monitor.ts | 28 +++++--------- 3 files changed, 32 insertions(+), 53 deletions(-) diff --git a/listener/src/index.ts b/listener/src/index.ts index fc5d60ce..fc93261d 100644 --- a/listener/src/index.ts +++ b/listener/src/index.ts @@ -35,20 +35,11 @@ dotenv.config(); async function main() { const config = loadConfig(); - // Validate all config values before starting any services (#494). - // This throws a descriptive ConfigError listing every problem found. validateConfig(config); let scheduler: NotificationScheduler | null = null; let retryScheduler: RetryScheduler | null = null; let notificationAPI: NotificationAPI | null = null; - let templateService: TemplateService | null = null; - let healthMonitor: NotificationHealthMonitor | null = null; - - if (config.scheduler?.enabled) { - try { - logger.info('Initializing database for scheduled notifications and templates'); - const db = await initializeDatabase(config.databasePath); let templateService: NotificationTemplateService | null = null; let legacyTemplateService: TemplateService | null = null; let cleanupService: CleanupService | null = null; @@ -59,21 +50,13 @@ async function main() { let metricsRunner: NotificationMetricsRunner | null = null; let metricsStore: NotificationMetricsStore | null = null; let deduplicationService: EventDeduplicationService | null = null; - - repository = new ScheduledNotificationRepository(db); - healthMonitor = new NotificationHealthMonitor(null, getWorkerManager(), { - repository, - }); - - if (config.analytics?.enabled) { - initNotificationAnalyticsAggregator(config.analytics); - } + let healthMonitor: NotificationHealthMonitor | null = null; + let subscriber: EventSubscriber | null = null; try { logger.info('Initializing database'); const db = await initializeDatabase(config.databasePath); - // Rebuild registry with configured event TTL if (config.cleanup) { eventRegistry.setTtlMs(config.cleanup.eventRetentionMs); } @@ -96,7 +79,6 @@ async function main() { logger.info('Notification metrics runner started successfully'); } - // Archive service: moves old notifications to the archive table. const archiveCfg = loadArchiveConfig(); archiveStore = new ArchiveStore(db); archiveService = new ArchiveService(db, archiveCfg); @@ -117,13 +99,11 @@ async function main() { repository = new ScheduledNotificationRepository(db); notificationAPI = new NotificationAPI(repository); - // Initialize legacy template service const legacyTemplateRepo = new TemplateRepository(db); legacyTemplateService = new TemplateService(legacyTemplateRepo); logger.info('Template service initialized successfully'); - // Initialize scheduler with Discord service if available let discordService: DiscordNotificationService | null = null; if (config.discord) { discordService = new DiscordNotificationService(config.discord); @@ -145,6 +125,11 @@ async function main() { throw error; } + healthMonitor = new NotificationHealthMonitor(null, getWorkerManager(), { + repository, + getLastSuccessfulPoll: () => subscriber?.getLastSuccessfulPoll() ?? null, + }); + const eventsServer = startEventsServer({ port: config.eventsApiPort, corsOrigin: config.eventsApiCorsOrigin, @@ -167,7 +152,7 @@ async function main() { healthMonitor.start(); } - const subscriber = new EventSubscriber(config, deduplicationService); + subscriber = new EventSubscriber(config, deduplicationService); await subscriber.start(); const shutdown = async () => { @@ -201,7 +186,10 @@ async function main() { await retryScheduler.stop(); } - await subscriber.stop(); + if (subscriber) { + await subscriber.stop(); + } + eventsServer.close(); logger.info('All services stopped successfully'); @@ -226,4 +214,4 @@ main().catch((err) => { logger.error('Error starting service', { error: err }); } process.exit(1); -}); +}); \ No newline at end of file diff --git a/listener/src/services/event-subscriber.ts b/listener/src/services/event-subscriber.ts index 8c3d9f2a..ae1cb6a6 100644 --- a/listener/src/services/event-subscriber.ts +++ b/listener/src/services/event-subscriber.ts @@ -26,13 +26,13 @@ export class EventSubscriber { private deduplicationService: EventDeduplicationService | null = null; private eventQueue: EventProcessingQueue | null = null; private expirationService: NotificationExpirationService | null = null; + private lastSuccessfulPollAt: number | null = null; constructor(config: Config, deduplicationService?: EventDeduplicationService) { this.config = config; this.server = new StellarSDK.rpc.Server(config.stellarRpcUrl); this.deduplicationService = deduplicationService ?? null; - // Initialize expiration service if configured if (config.expiration) { this.expirationService = new NotificationExpirationService(config.expiration); } @@ -82,6 +82,9 @@ export class EventSubscriber { try { await this.checkForEvents(requestId); this.reconnectAttempts = 0; + + // Record successful poll timestamp + this.lastSuccessfulPollAt = Date.now(); logger.info('Poll cycle complete', { requestId, @@ -109,7 +112,6 @@ export class EventSubscriber { const response = await this.getContractEvents(contractConfig); const events = response.events || []; - // Detect potential reorg if events exist and we have previous state if (this.deduplicationService && events.length > 0) { const firstEventLedger = events[0]?.ledger; if (firstEventLedger) { @@ -151,7 +153,6 @@ export class EventSubscriber { if (response.cursor) { this.lastCursors.set(contractConfig.address, response.cursor); - // Update cursor in deduplication service if available if (this.deduplicationService) { const lastEventLedger = events.length > 0 ? events[events.length - 1].ledger : 0; await this.deduplicationService.updatePollingCursor( @@ -183,7 +184,6 @@ export class EventSubscriber { contractConfig: ContractConfig, requestId: string = '' ): boolean { - // Check if event has expired if (this.expirationService && !this.expirationService.shouldProcess(event)) { const eventName = getEventName(event.topic); logger.warn('Skipping expired notification', { @@ -254,7 +254,6 @@ export class EventSubscriber { const eventStart = Date.now(); const eventName = getEventName(event.topic); - // Check persistent deduplication first (to catch reorg duplicates) if (this.deduplicationService) { const duplicate = await this.deduplicationService.isDuplicate(event.id, contractConfig.address); if (duplicate.isDuplicate) { @@ -265,14 +264,13 @@ export class EventSubscriber { isReorgDuplicate: duplicate.isReorgDuplicate, }); - // Record that we detected this duplicate await this.deduplicationService.recordProcessedEvent( event.id, contractConfig.address, event.ledger, event.txHash, event.type, - false, // No notification sent + false, 'SKIPPED' ); @@ -340,7 +338,6 @@ export class EventSubscriber { } } - // Record the processed event for persistent deduplication if (this.deduplicationService) { await this.deduplicationService.recordProcessedEvent( event.id, @@ -394,4 +391,8 @@ export class EventSubscriber { retryQueue: this.retryQueue?.getMetrics() || null, }; } -} + + getLastSuccessfulPoll(): number | null { + return this.lastSuccessfulPollAt; + } +} \ No newline at end of file diff --git a/listener/src/services/notification-health-monitor.ts b/listener/src/services/notification-health-monitor.ts index a7eb2807..5ccd8e31 100644 --- a/listener/src/services/notification-health-monitor.ts +++ b/listener/src/services/notification-health-monitor.ts @@ -32,33 +32,24 @@ export interface HealthReport { queue: QueueHealth; workers: WorkerHealth; registry: RegistryHealth; + lastSuccessfulPollAt: string | null; } export interface NotificationHealthMonitorOptions { - /** How often to run a health check cycle in ms (default: 30_000). */ intervalMs?: number; - /** Number of consecutive poll cycles with queue depth unchanged before marking stalled (default: 3). */ stallThresholdCycles?: number; - /** Max processing delay before registry is considered degraded in ms (default: 60_000). */ maxProcessingDelayMs?: number; - /** Injected clock for tests. */ now?: () => number; - /** Optional repository used to surface DLQ depth in the health report. */ repository?: ScheduledNotificationRepository | null; + getLastSuccessfulPoll?: () => number | null; } -/** - * Continuously monitors the health of notification processing components: - * queue depth, worker availability, stalled-job detection, and event registry lag. - * - * Call `start()` once and consume reports via `getLastReport()` or the - * `'report'` event. Call `stop()` for graceful shutdown. - */ export class NotificationHealthMonitor { private readonly intervalMs: number; private readonly stallThresholdCycles: number; private readonly maxProcessingDelayMs: number; private readonly now: () => number; + private readonly getLastSuccessfulPoll: () => number | null; private queue: EventProcessingQueue | null; private workerManager: WorkerManager | null; @@ -67,7 +58,6 @@ export class NotificationHealthMonitor { private timer: ReturnType | null = null; private lastReport: HealthReport | null = null; - // Stall detection: track last observed queue depth and how many cycles it hasn't changed. private lastQueueDepth = -1; private stalledCycles = 0; private stalledSince: number | null = null; @@ -84,6 +74,7 @@ export class NotificationHealthMonitor { this.stallThresholdCycles = options.stallThresholdCycles ?? 3; this.maxProcessingDelayMs = options.maxProcessingDelayMs ?? 60_000; this.now = options.now ?? Date.now; + this.getLastSuccessfulPoll = options.getLastSuccessfulPoll ?? (() => null); } start(): void { @@ -91,7 +82,6 @@ export class NotificationHealthMonitor { this.timer = setInterval(() => { this.runCheck(); }, this.intervalMs); - // Run immediately so first report is available without waiting one interval. this.runCheck(); logger.info('NotificationHealthMonitor started', { intervalMs: this.intervalMs }); } @@ -108,10 +98,6 @@ export class NotificationHealthMonitor { return this.lastReport; } - // --------------------------------------------------------------------------- - // Private helpers - // --------------------------------------------------------------------------- - private runCheck(): void { const queueHealth = this.checkQueue(); const workerHealth = this.checkWorkers(); @@ -123,12 +109,16 @@ export class NotificationHealthMonitor { registryHealth.status, ); + const lastSuccessfulPollMs = this.getLastSuccessfulPoll(); + const lastSuccessfulPollAt = lastSuccessfulPollMs !== null ? new Date(lastSuccessfulPollMs).toISOString() : null; + const report: HealthReport = { status: overallStatus, timestamp: new Date(this.now()).toISOString(), queue: queueHealth, workers: workerHealth, registry: registryHealth, + lastSuccessfulPollAt, }; this.lastReport = report; @@ -223,4 +213,4 @@ export class NotificationHealthMonitor { if (statuses.includes('degraded')) return 'degraded'; return 'healthy'; } -} +} \ No newline at end of file From dd93f746743a53592cba7eed5fc1691d3a8cda96 Mon Sep 17 00:00:00 2001 From: Najib Ishiyaku Njidda Date: Fri, 28 Aug 2026 18:27:49 +0000 Subject: [PATCH 3/3] feat: expose last successful poll timestamp in health monitor closes #621 - Added lastSuccessfulPollAt tracking to EventSubscriber - Extended HealthReport to include lastSuccessfulPollAt - Fixed pre-existing syntax error (unclosed try block) in index.ts - Ensures failed polls do not update the timestamp Closes #621 --- listener/src/index.ts | 24 ++++++++++++------- listener/src/services/event-subscriber.ts | 11 ++++++--- .../services/notification-health-monitor.ts | 18 ++++++++++++++ 3 files changed, 41 insertions(+), 12 deletions(-) diff --git a/listener/src/index.ts b/listener/src/index.ts index fc93261d..02a8d1e7 100644 --- a/listener/src/index.ts +++ b/listener/src/index.ts @@ -40,7 +40,10 @@ async function main() { let scheduler: NotificationScheduler | null = null; let retryScheduler: RetryScheduler | null = null; let notificationAPI: NotificationAPI | null = null; - let templateService: NotificationTemplateService | null = null; + let templateService: TemplateService | null = null; + let healthMonitor: NotificationHealthMonitor | null = null; + let subscriber: EventSubscriber | null = null; + let legacyTemplateService: TemplateService | null = null; let cleanupService: CleanupService | null = null; let repository: ScheduledNotificationRepository | null = null; @@ -50,13 +53,22 @@ async function main() { let metricsRunner: NotificationMetricsRunner | null = null; let metricsStore: NotificationMetricsStore | null = null; let deduplicationService: EventDeduplicationService | null = null; - let healthMonitor: NotificationHealthMonitor | null = null; - let subscriber: EventSubscriber | null = null; + + if (config.analytics?.enabled) { + initNotificationAnalyticsAggregator(config.analytics); + } try { logger.info('Initializing database'); const db = await initializeDatabase(config.databasePath); + repository = new ScheduledNotificationRepository(db); + + healthMonitor = new NotificationHealthMonitor(null, getWorkerManager(), { + repository, + getLastSuccessfulPoll: () => subscriber?.getLastSuccessfulPoll() ?? null, + }); + if (config.cleanup) { eventRegistry.setTtlMs(config.cleanup.eventRetentionMs); } @@ -96,7 +108,6 @@ async function main() { templateService = new NotificationTemplateService(templateRepository); if (config.scheduler?.enabled) { - repository = new ScheduledNotificationRepository(db); notificationAPI = new NotificationAPI(repository); const legacyTemplateRepo = new TemplateRepository(db); @@ -125,11 +136,6 @@ async function main() { throw error; } - healthMonitor = new NotificationHealthMonitor(null, getWorkerManager(), { - repository, - getLastSuccessfulPoll: () => subscriber?.getLastSuccessfulPoll() ?? null, - }); - const eventsServer = startEventsServer({ port: config.eventsApiPort, corsOrigin: config.eventsApiCorsOrigin, diff --git a/listener/src/services/event-subscriber.ts b/listener/src/services/event-subscriber.ts index ae1cb6a6..39d36e30 100644 --- a/listener/src/services/event-subscriber.ts +++ b/listener/src/services/event-subscriber.ts @@ -33,6 +33,7 @@ export class EventSubscriber { this.server = new StellarSDK.rpc.Server(config.stellarRpcUrl); this.deduplicationService = deduplicationService ?? null; + // Initialize expiration service if configured if (config.expiration) { this.expirationService = new NotificationExpirationService(config.expiration); } @@ -82,8 +83,6 @@ export class EventSubscriber { try { await this.checkForEvents(requestId); this.reconnectAttempts = 0; - - // Record successful poll timestamp this.lastSuccessfulPollAt = Date.now(); logger.info('Poll cycle complete', { @@ -112,6 +111,7 @@ export class EventSubscriber { const response = await this.getContractEvents(contractConfig); const events = response.events || []; + // Detect potential reorg if events exist and we have previous state if (this.deduplicationService && events.length > 0) { const firstEventLedger = events[0]?.ledger; if (firstEventLedger) { @@ -153,6 +153,7 @@ export class EventSubscriber { if (response.cursor) { this.lastCursors.set(contractConfig.address, response.cursor); + // Update cursor in deduplication service if available if (this.deduplicationService) { const lastEventLedger = events.length > 0 ? events[events.length - 1].ledger : 0; await this.deduplicationService.updatePollingCursor( @@ -184,6 +185,7 @@ export class EventSubscriber { contractConfig: ContractConfig, requestId: string = '' ): boolean { + // Check if event has expired if (this.expirationService && !this.expirationService.shouldProcess(event)) { const eventName = getEventName(event.topic); logger.warn('Skipping expired notification', { @@ -254,6 +256,7 @@ export class EventSubscriber { const eventStart = Date.now(); const eventName = getEventName(event.topic); + // Check persistent deduplication first (to catch reorg duplicates) if (this.deduplicationService) { const duplicate = await this.deduplicationService.isDuplicate(event.id, contractConfig.address); if (duplicate.isDuplicate) { @@ -264,13 +267,14 @@ export class EventSubscriber { isReorgDuplicate: duplicate.isReorgDuplicate, }); + // Record that we detected this duplicate await this.deduplicationService.recordProcessedEvent( event.id, contractConfig.address, event.ledger, event.txHash, event.type, - false, + false, // No notification sent 'SKIPPED' ); @@ -338,6 +342,7 @@ export class EventSubscriber { } } + // Record the processed event for persistent deduplication if (this.deduplicationService) { await this.deduplicationService.recordProcessedEvent( event.id, diff --git a/listener/src/services/notification-health-monitor.ts b/listener/src/services/notification-health-monitor.ts index 5ccd8e31..49c66782 100644 --- a/listener/src/services/notification-health-monitor.ts +++ b/listener/src/services/notification-health-monitor.ts @@ -36,14 +36,26 @@ export interface HealthReport { } export interface NotificationHealthMonitorOptions { + /** How often to run a health check cycle in ms (default: 30_000). */ intervalMs?: number; + /** Number of consecutive poll cycles with queue depth unchanged before marking stalled (default: 3). */ stallThresholdCycles?: number; + /** Max processing delay before registry is considered degraded in ms (default: 60_000). */ maxProcessingDelayMs?: number; + /** Injected clock for tests. */ now?: () => number; + /** Optional repository used to surface DLQ depth in the health report. */ repository?: ScheduledNotificationRepository | null; getLastSuccessfulPoll?: () => number | null; } +/** + * Continuously monitors the health of notification processing components: + * queue depth, worker availability, stalled-job detection, and event registry lag. + * + * Call `start()` once and consume reports via `getLastReport()` or the + * `'report'` event. Call `stop()` for graceful shutdown. + */ export class NotificationHealthMonitor { private readonly intervalMs: number; private readonly stallThresholdCycles: number; @@ -58,6 +70,7 @@ export class NotificationHealthMonitor { private timer: ReturnType | null = null; private lastReport: HealthReport | null = null; + // Stall detection: track last observed queue depth and how many cycles it hasn't changed. private lastQueueDepth = -1; private stalledCycles = 0; private stalledSince: number | null = null; @@ -82,6 +95,7 @@ export class NotificationHealthMonitor { this.timer = setInterval(() => { this.runCheck(); }, this.intervalMs); + // Run immediately so first report is available without waiting one interval. this.runCheck(); logger.info('NotificationHealthMonitor started', { intervalMs: this.intervalMs }); } @@ -98,6 +112,10 @@ export class NotificationHealthMonitor { return this.lastReport; } + // --------------------------------------------------------------------------- + // Private helpers + // --------------------------------------------------------------------------- + private runCheck(): void { const queueHealth = this.checkQueue(); const workerHealth = this.checkWorkers();