diff --git a/listener/src/services/event-subscriber.ts b/listener/src/services/event-subscriber.ts index 8c3d9f2a..788d699e 100644 --- a/listener/src/services/event-subscriber.ts +++ b/listener/src/services/event-subscriber.ts @@ -14,6 +14,7 @@ import { NotificationRetryQueue } from './notification-retry-queue'; import { EventDeduplicationService } from './event-deduplication-service'; import { EventProcessingQueue } from './event-processing-queue'; import { NotificationExpirationService } from './notification-expiration'; +import { pollingMetrics } from './polling-metrics'; export class EventSubscriber { private config: Config; @@ -83,17 +84,23 @@ export class EventSubscriber { await this.checkForEvents(requestId); this.reconnectAttempts = 0; + const durationMs = Date.now() - pollStart; + pollingMetrics.record(durationMs, true); + logger.info('Poll cycle complete', { requestId, - durationMs: Date.now() - pollStart, + durationMs, }); await this.delay(this.config.pollIntervalMs); } catch (error) { + const durationMs = Date.now() - pollStart; + pollingMetrics.record(durationMs, false); + logger.error('Error polling for events', { requestId, error, - durationMs: Date.now() - pollStart, + durationMs, }); await this.handleReconnection(requestId); } diff --git a/listener/src/services/notification-health-monitor.ts b/listener/src/services/notification-health-monitor.ts index a7eb2807..2dd1106a 100644 --- a/listener/src/services/notification-health-monitor.ts +++ b/listener/src/services/notification-health-monitor.ts @@ -3,6 +3,7 @@ import { EventProcessingQueue } from './event-processing-queue'; import { WorkerManager } from './worker-manager'; import { eventRegistry } from '../store/event-registry'; import { ScheduledNotificationRepository } from './scheduled-notification-repository'; +import { pollingMetrics } from './polling-metrics'; export type ComponentStatus = 'healthy' | 'degraded' | 'unhealthy'; @@ -26,12 +27,32 @@ export interface RegistryHealth { processingDelayMs: number | null; } +export interface PollingHealth { + /** ISO timestamp of the most recent successful poll, or null if none yet. */ + lastSuccessAt: string | null; + /** ISO timestamp of the most recent failed poll, or null if none yet. */ + lastFailureAt: string | null; + /** ISO timestamp of the most recent poll cycle (successful or not). */ + lastPollAt: string | null; + /** Duration of the most recent poll cycle in milliseconds. */ + lastPollDurationMs: number | null; + /** Whether the most recent poll cycle completed without error. */ + lastPollSucceeded: boolean | null; + /** Total number of recorded poll cycles. */ + totalPolls: number; + /** Number of successful poll cycles. */ + successfulPolls: number; + /** Number of failed poll cycles. */ + failedPolls: number; +} + export interface HealthReport { status: ComponentStatus; timestamp: string; queue: QueueHealth; workers: WorkerHealth; registry: RegistryHealth; + polling: PollingHealth; } export interface NotificationHealthMonitorOptions { @@ -116,6 +137,7 @@ export class NotificationHealthMonitor { const queueHealth = this.checkQueue(); const workerHealth = this.checkWorkers(); const registryHealth = this.checkRegistry(); + const pollingHealth = this.checkPolling(); const overallStatus = this.deriveOverallStatus( queueHealth.status, @@ -129,6 +151,7 @@ export class NotificationHealthMonitor { queue: queueHealth, workers: workerHealth, registry: registryHealth, + polling: pollingHealth, }; this.lastReport = report; @@ -218,6 +241,21 @@ export class NotificationHealthMonitor { return { status, eventCount, lastIngestedAt, processingDelayMs }; } + private checkPolling(): PollingHealth { + const snapshot = pollingMetrics.snapshot(); + + return { + lastSuccessAt: snapshot.lastSuccessAt, + lastFailureAt: snapshot.lastFailureAt, + lastPollAt: snapshot.lastPollAt, + lastPollDurationMs: snapshot.lastPollDurationMs, + lastPollSucceeded: snapshot.lastPollSucceeded, + totalPolls: snapshot.totalPolls, + successfulPolls: snapshot.successfulPolls, + failedPolls: snapshot.failedPolls, + }; + } + private deriveOverallStatus(...statuses: ComponentStatus[]): ComponentStatus { if (statuses.includes('unhealthy')) return 'unhealthy'; if (statuses.includes('degraded')) return 'degraded'; diff --git a/listener/src/services/polling-metrics.ts b/listener/src/services/polling-metrics.ts new file mode 100644 index 00000000..ddb458fd --- /dev/null +++ b/listener/src/services/polling-metrics.ts @@ -0,0 +1,127 @@ +/** + * In-memory telemetry for blockchain polling cycles. + * + * Records the outcome of every polling cycle (duration and success) so the + * listener's synchronization state can be inspected through the existing + * health/diagnostic interface. State is held in memory only, so recorded + * values reset naturally when the process restarts. + */ + +export interface PollingMetricsSnapshot { + /** ISO timestamp of the most recent successful poll, or null if none yet. */ + lastSuccessAt: string | null; + /** ISO timestamp of the most recent failed poll, or null if none yet. */ + lastFailureAt: string | null; + /** ISO timestamp of the most recent poll cycle (successful or not). */ + lastPollAt: string | null; + /** Duration of the most recent poll cycle in milliseconds. */ + lastPollDurationMs: number | null; + /** Whether the most recent poll cycle completed without error. */ + lastPollSucceeded: boolean | null; + /** Total number of recorded poll cycles. */ + totalPolls: number; + /** Number of successful poll cycles. */ + successfulPolls: number; + /** Number of failed poll cycles. */ + failedPolls: number; + /** Success ratio (0-100) rounded to one decimal place, or null when idle. */ + successRate: number | null; + /** Average duration of recent poll cycles in milliseconds. */ + averageDurationMs: number | null; + /** Durations of the most recent poll cycles (oldest first, size-capped). */ + recentDurationsMs: number[]; +} + +/** Maximum number of recent cycle durations retained in memory. */ +const RECENT_DURATIONS_LIMIT = 100; + +/** + * Collects minimal, in-memory telemetry about blockchain polling cycles. + * + * Recording a cycle is O(1) apart from a size-capped ring of recent + * durations, so the overhead on the hot polling path remains negligible. + */ +export class PollingMetrics { + private lastSuccessAt: number | null = null; + private lastFailureAt: number | null = null; + private lastPollAt: number | null = null; + private lastPollDurationMs: number | null = null; + private lastPollSucceeded: boolean | null = null; + private totalPolls = 0; + private successfulPolls = 0; + private failedPolls = 0; + private recentDurationsMs: number[] = []; + + /** + * Record the outcome of a single polling cycle. + * + * Failed cycles never update the last-successful-poll timestamp, keeping + * the listener's "actively synchronizing" signal accurate. + */ + record(durationMs: number, success: boolean): void { + const now = Date.now(); + const normalizedDuration = Math.max(0, durationMs); + + this.totalPolls++; + this.lastPollAt = now; + this.lastPollDurationMs = normalizedDuration; + this.lastPollSucceeded = success; + + if (success) { + this.successfulPolls++; + this.lastSuccessAt = now; + } else { + this.failedPolls++; + this.lastFailureAt = now; + } + + this.recentDurationsMs.push(normalizedDuration); + if (this.recentDurationsMs.length > RECENT_DURATIONS_LIMIT) { + this.recentDurationsMs.shift(); + } + } + + /** Returns a serializable snapshot of the recorded poll telemetry. */ + snapshot(): PollingMetricsSnapshot { + const recentCount = this.recentDurationsMs.length; + const totalDurationMs = this.recentDurationsMs.reduce( + (sum, duration) => sum + duration, + 0, + ); + const averageDurationMs = + recentCount > 0 ? Math.round((totalDurationMs / recentCount) * 10) / 10 : null; + + return { + lastSuccessAt: + this.lastSuccessAt !== null ? new Date(this.lastSuccessAt).toISOString() : null, + lastFailureAt: + this.lastFailureAt !== null ? new Date(this.lastFailureAt).toISOString() : null, + lastPollAt: this.lastPollAt !== null ? new Date(this.lastPollAt).toISOString() : null, + lastPollDurationMs: this.lastPollDurationMs, + lastPollSucceeded: this.lastPollSucceeded, + totalPolls: this.totalPolls, + successfulPolls: this.successfulPolls, + failedPolls: this.failedPolls, + successRate: + this.totalPolls > 0 ? Math.round((this.successfulPolls / this.totalPolls) * 1000) / 10 : null, + averageDurationMs, + recentDurationsMs: [...this.recentDurationsMs], + }; + } + + /** Clears all recorded telemetry, returning to the post-restart state. */ + reset(): void { + this.lastSuccessAt = null; + this.lastFailureAt = null; + this.lastPollAt = null; + this.lastPollDurationMs = null; + this.lastPollSucceeded = null; + this.totalPolls = 0; + this.successfulPolls = 0; + this.failedPolls = 0; + this.recentDurationsMs = []; + } +} + +/** Process-wide polling telemetry collector. */ +export const pollingMetrics = new PollingMetrics(); \ No newline at end of file