Merge PR #260: fix/137-event-listener-health-retry

Author: Calebux

backend/src/index.ts

@@ -129,6 +129,7 @@ app.get('/api/admin/health', createAdminLimiter(), adminAuth, async (req, res) =
   try {
     const includeHistory = req.query.history !== 'false';
     const health = await healthService.getAdminHealth(includeHistory);
+    const health = await healthService.getAdminHealth(includeHistory, eventListener.getHealth());
     const statusCode = health.status === 'unhealthy' ? 503 : 200;
     res.status(statusCode).json(health);
   } catch (error) {
@@ -779,6 +780,38 @@ function startHealthSnapshotInterval() {
   setTimeout(() => healthService.recordSnapshot().catch(() => {}), 5000);
 }
 
+app.post('/api/admin/expiry/process', createAdminLimiter(), adminAuth, async (req, res) => {
+  try {
+    const result = await expiryService.processExpiries();
+    res.json({ success: true, data: result });
+  } catch (error) {
+    logger.error('Error processing expiries:', error);
+    res.status(500).json({
+      success: false,
+      error: error instanceof Error ? error.message : String(error),
+    });
+  }
+});
+
+// Start server
+const server = app.listen(PORT, async () => {
+  logger.info(`Server running on port ${PORT}`);
+  logger.info(`Environment: ${process.env.NODE_ENV || 'development'}`);
+
+  // Validate critical env vars at startup — warn clearly so operators know what's missing
+  const criticalEnvVars = ['SOROBAN_CONTRACT_ADDRESS', 'STELLAR_NETWORK_URL'];
+  for (const envVar of criticalEnvVars) {
+    if (!process.env[envVar]) {
+      logger.warn(`${envVar} not configured — EventListener will be disabled`);
+    }
+  }
+
+  // Initialize rate limiting Redis store
+  try {
+    await RateLimiterFactory.initializeRedisStore();
+    logger.info('Rate limiting initialized successfully');
+  } catch (error) {
+    logger.warn('Rate limiting initialization failed, using memory store:', error);
 /**
  * @openapi
  * /api/admin/expiry/process:
@@ -818,5 +851,42 @@ export function validateMnemonic(mnemonic: string): boolean {
     return false;
   }
 
+  // Start health metrics snapshot loop
+  startHealthSnapshotInterval();
+
+  // Start event listener (no-op if disabled due to missing config)
+  await eventListener.start();
+  const elHealth = eventListener.getHealth();
+  if (elHealth.status === 'disabled') {
+    logger.warn('EventListener is disabled', { reason: elHealth.reason });
+  } else {
+    logger.info('EventListener started', { status: elHealth.status });
+  }
+
+  scheduleAutoResume();
+});
+
+
+
+// Graceful shutdown
+process.on('SIGTERM', () => {
+  logger.info('SIGTERM received, shutting down gracefully');
+  schedulerService.stop();
+  eventListener.stop();
+  server.close(() => {
+    logger.info('Server closed');
+    process.exit(0);
+  });
+});
+
+process.on('SIGINT', () => {
+  logger.info('SIGINT received, shutting down gracefully');
+  schedulerService.stop();
+  eventListener.stop();
+  server.close(() => {
+    logger.info('Server closed');
+    process.exit(0);
+  });
+});
   return bip39.validateMnemonic(words.join(' '));
-}
+}

backend/src/services/event-listener.ts

@@ -3,6 +3,7 @@ import { supabase } from '../config/database';
 import { reorgHandler } from './reorg-handler';
 import { generateCycleId } from '../utils/cycle-id';
 import { renewalCooldownService } from './renewal-cooldown-service';
+import { calculateBackoffDelay, NonRetryableError } from '../utils/retry';
 
 interface ContractEvent {
   type: string;
@@ -31,6 +32,15 @@ export interface EventListenerHealth {
 
 const ALERT_THRESHOLD = 10;
 const MAX_BACKOFF_MS = 300_000; // 5 minutes
+export type EventListenerStatus = 'running' | 'stopped' | 'disabled' | 'retrying' | 'failed';
+  status: EventListenerStatus;
+  reason?: string;
+  lastProcessedLedger: number | null;
+  retryCount?: number;
+  nextRetryAt?: string | null;
+const MAX_RETRY_ATTEMPTS = 10;
+const RETRY_INITIAL_DELAY_MS = 5000;
+const RETRY_MAX_DELAY_MS = 5 * 60 * 1000; // 5 minutes
 
 export class EventListener {
   private contractId: string;
@@ -49,20 +59,49 @@ export class EventListener {
   private consecutiveErrors: number = 0;
   private lastSuccessfulPoll: Date | null = null;
 
+  // Health tracking
+  private _status: EventListenerStatus = 'stopped';
+  private _disabledReason?: string;
+  private _retryCount: number = 0;
+  private _nextRetryAt: Date | null = null;
+
   constructor() {
     this.contractId = process.env.SOROBAN_CONTRACT_ADDRESS || '';
     this.rpcUrl =
       process.env.STELLAR_NETWORK_URL || 'https://soroban-testnet.stellar.org';
 
     if (!this.contractId) {
-      throw new Error('SOROBAN_CONTRACT_ADDRESS not configured');
+      // Don't throw — mark as disabled so the process can still start
+      this._status = 'disabled';
+      this._disabledReason = 'SOROBAN_CONTRACT_ADDRESS not configured';
+      logger.warn('EventListener disabled: SOROBAN_CONTRACT_ADDRESS not configured');
     }
   }
 
+  getHealth(): EventListenerHealth {
+    return {
+      status: this._status,
+      reason: this._disabledReason,
+      lastProcessedLedger: this.lastProcessedLedger || null,
+      retryCount: this._retryCount,
+      nextRetryAt: this._nextRetryAt?.toISOString() ?? null,
+    };
+  }
+
   async start() {
+    if (this._status === 'disabled') {
+      logger.warn('EventListener.start() called but listener is disabled', {
+        reason: this._disabledReason,
+      });
+      return;
+    }
+
     if (this.isRunning) return;
 
     this.isRunning = true;
+    this._status = 'running';
+    this._retryCount = 0;
+    this._nextRetryAt = null;
     this.lastProcessedLedger = await this.getLastProcessedLedger();
     logger.info('Event listener started', { lastLedger: this.lastProcessedLedger });
 
@@ -71,6 +110,9 @@ export class EventListener {
 
   stop() {
     this.isRunning = false;
+    if (this._status !== 'disabled') {
+      this._status = 'stopped';
+    }
     logger.info('Event listener stopped');
   }
 
@@ -141,6 +183,15 @@ export class EventListener {
       } finally {
         // Always release the mutex, even if fetchAndProcessEvents throws
         this.isProcessing = false;
+        // Reset retry count on success
+        if (this._retryCount > 0) {
+          logger.info('EventListener recovered after retries', { retryCount: this._retryCount });
+          this._retryCount = 0;
+          this._nextRetryAt = null;
+        this._status = 'running';
+        logger.error('Event polling error:', error);
+        await this.handlePollError(error);
+        if (!this.isRunning) break;
       }
 
       await this.sleep(backoffMs);
@@ -150,6 +201,40 @@ export class EventListener {
 private async fetchAndProcessEvents() {
   logger.info('Polling for events...');
   const currentLedger = await this.getCurrentLedger();
+  private async handlePollError(error: unknown) {
+    this._retryCount++;
+
+    if (this._retryCount >= MAX_RETRY_ATTEMPTS) {
+      this._status = 'failed';
+      this._disabledReason = `Exceeded max retry attempts (${MAX_RETRY_ATTEMPTS}). Last error: ${error instanceof Error ? error.message : String(error)}`;
+      logger.error('EventListener permanently failed after max retries', {
+        retryCount: this._retryCount,
+        error: this._disabledReason,
+      });
+      this.isRunning = false;
+      return;
+    }
+
+    const delay = calculateBackoffDelay(this._retryCount, {
+      initialDelay: RETRY_INITIAL_DELAY_MS,
+      maxDelay: RETRY_MAX_DELAY_MS,
+      multiplier: 2,
+      jitter: true,
+    });
+
+    this._status = 'retrying';
+    this._nextRetryAt = new Date(Date.now() + delay);
+    logger.warn('EventListener will retry', {
+      attempt: this._retryCount,
+      delayMs: delay,
+      nextRetryAt: this._nextRetryAt.toISOString(),
+    });
+
+    await this.sleep(delay);
+  }
+
+  private async fetchAndProcessEvents() {
+    const currentLedger = await this.getCurrentLedger();
 
   // Check for reorg
   if (currentLedger < this.lastProcessedLedger) {

backend/src/services/health-service.ts

@@ -2,6 +2,7 @@ import { supabase } from '../config/database';
 import logger from '../config/logger';
 import { monitoringService } from './monitoring-service';
 import { eventListener, EventListenerHealth } from './event-listener';
+import type { EventListenerHealth } from './event-listener';
 
 export interface HealthThresholds {
   failedRenewalsPerHour: number;
@@ -250,8 +251,9 @@ export class HealthService {
    *   listener unhealthy + base healthy  → degraded
    *   listener unhealthy + base degraded → unhealthy
    *   base unhealthy (any listener)      → unhealthy
+   * Full admin health: current metrics, alerts, status, event listener state, optional history.
    */
-  async getAdminHealth(includeHistory: boolean = true): Promise<AdminHealthResponse> {
+  async getAdminHealth(includeHistory: boolean = true, eventListenerHealth?: EventListenerHealth): Promise<AdminHealthResponse> {
     const metrics = await this.getCurrentMetrics();
     const alerts = this.evaluateAlerts(metrics);
     const listenerHealth = eventListener.getHealth();
@@ -264,6 +266,31 @@ export class HealthService {
         ? 'unhealthy'
         : baseStatus;
 
+    // Degrade overall status if event listener is not running
+    const elHealth: EventListenerHealth = eventListenerHealth ?? {
+      status: 'stopped',
+      lastProcessedLedger: null,
+    };
+    if (elHealth.status === 'disabled' || elHealth.status === 'failed') {
+      alerts.push({
+        id: 'event_listener',
+        message: `EventListener is ${elHealth.status}: ${elHealth.reason ?? 'unknown reason'}`,
+        severity: elHealth.status === 'failed' ? 'critical' : 'warning',
+        value: 0,
+        threshold: 0,
+        triggeredAt: new Date().toISOString(),
+      });
+    } else if (elHealth.status === 'retrying') {
+      alerts.push({
+        id: 'event_listener',
+        message: `EventListener is retrying (attempt ${elHealth.retryCount ?? '?'})`,
+        severity: 'warning',
+        value: elHealth.retryCount ?? 0,
+        threshold: 0,
+        triggeredAt: new Date().toISOString(),
+      });
+    }
+    const status = this.getStatus(alerts);
     const history = includeHistory ? await this.getHistory(24) : undefined;
 
     return {
@@ -273,6 +300,7 @@ export class HealthService {
       alerts,
       thresholds: this.getThresholds(),
       eventListener: listenerHealth,
+      eventListener: elHealth,
       history,
     };
   }