diff --git a/ENVIRONMENT_VARIABLES_AND_SECRETS.md b/ENVIRONMENT_VARIABLES_AND_SECRETS.md index 4f527416..0b0bdf52 100644 --- a/ENVIRONMENT_VARIABLES_AND_SECRETS.md +++ b/ENVIRONMENT_VARIABLES_AND_SECRETS.md @@ -128,6 +128,7 @@ Both variables must be provided together or neither. | `POLL_INTERVAL_MS` | `30000` | No | How often the listener polls Stellar for new contract events (ms). | | `MAX_RECONNECT_ATTEMPTS` | `5` | No | Maximum number of reconnect attempts when the RPC endpoint fails. | | `RECONNECT_DELAY_MS` | `5000` | No | Delay between reconnect attempts (ms). | +| `PROCESSED_EVENT_RETENTION_MS` | `2592000000` (30 days) | No | How long processed event metadata is retained for persistent deduplication and operations. Expired records are removed during database cleanup; minimum `60000` ms. | ### 2.10 Retry queue (in-memory) diff --git a/listener/.env.example b/listener/.env.example index 1f6a7641..042ad75b 100644 --- a/listener/.env.example +++ b/listener/.env.example @@ -120,6 +120,10 @@ MAX_RECONNECT_ATTEMPTS=5 # Delay between reconnect attempts (ms). RECONNECT_DELAY_MS=5000 +# How long to retain processed event metadata for persistent deduplication (ms). +# Expired records are removed during database cleanup. Default: 30 days. +PROCESSED_EVENT_RETENTION_MS=2592000000 + # ----------------------------------------------------------------------------- # Retry Queue (in-memory, fast retries) # ----------------------------------------------------------------------------- diff --git a/listener/src/config.test.ts b/listener/src/config.test.ts index 8ebb90ac..f4f76a54 100644 --- a/listener/src/config.test.ts +++ b/listener/src/config.test.ts @@ -1,4 +1,4 @@ -import { ConfigError, loadConfig } from './config'; +import { ConfigError, loadConfig, validateConfig } from './config'; describe('Config validation', () => { const originalEnv = process.env; @@ -98,11 +98,36 @@ describe('Config validation', () => { notificationRetentionMs: 604800000, rateLimitEventRetentionMs: 86400000, eventRetentionMs: 86400000, + processedEventRetentionMs: 2592000000, executionLogRetentionMs: 7776000000, }, }); }); + it('loads a configured processed event retention duration', () => { + process.env.PROCESSED_EVENT_RETENTION_MS = '3600000'; + + expect(loadConfig().cleanup?.processedEventRetentionMs).toBe(3600000); + }); + + it('rejects invalid processed event retention configuration', () => { + process.env.PROCESSED_EVENT_RETENTION_MS = 'not-a-duration'; + + expect(() => loadConfig()).toThrow( + 'PROCESSED_EVENT_RETENTION_MS must be a valid integer, got "not-a-duration"' + ); + }); + + it('rejects processed event retention shorter than one minute', () => { + process.env.PROCESSED_EVENT_RETENTION_MS = '59999'; + + const config = loadConfig(); + + expect(() => validateConfig(config)).toThrow( + 'PROCESSED_EVENT_RETENTION_MS must be >= 60000 ms (received: 59999).' + ); + }); + it('loads notification deduplication settings when Discord is configured', () => { process.env.DISCORD_WEBHOOK_URL = 'https://discord.com/api/webhooks/123/abc'; process.env.DISCORD_WEBHOOK_ID = '123'; diff --git a/listener/src/config.ts b/listener/src/config.ts index 2d4e9e62..1d936be1 100644 --- a/listener/src/config.ts +++ b/listener/src/config.ts @@ -159,6 +159,10 @@ function loadCleanupConfig(): AppCleanupConfig { notificationRetentionMs: parseIntegerEnv('NOTIFICATION_RETENTION_MS', String(7 * 24 * 60 * 60 * 1000)), rateLimitEventRetentionMs: parseIntegerEnv('RATE_LIMIT_EVENT_RETENTION_MS', String(24 * 60 * 60 * 1000)), eventRetentionMs: parseIntegerEnv('EVENT_RETENTION_MS', String(24 * 60 * 60 * 1000)), + processedEventRetentionMs: parseIntegerEnv( + 'PROCESSED_EVENT_RETENTION_MS', + String(30 * 24 * 60 * 60 * 1000), + ), executionLogRetentionMs: parseIntegerEnv( 'EXECUTION_LOG_RETENTION_MS', String(90 * 24 * 60 * 60 * 1000), @@ -465,6 +469,12 @@ export function validateConfig(config: Config): void { `(received: ${config.cleanup.notificationRetentionMs}).`, ); } + if (config.cleanup.processedEventRetentionMs < 60_000) { + errors.push( + `PROCESSED_EVENT_RETENTION_MS must be >= 60000 ms ` + + `(received: ${config.cleanup.processedEventRetentionMs}).`, + ); + } } if (errors.length > 0) { diff --git a/listener/src/services/cleanup-service.test.ts b/listener/src/services/cleanup-service.test.ts index fafb9462..5cf64832 100644 --- a/listener/src/services/cleanup-service.test.ts +++ b/listener/src/services/cleanup-service.test.ts @@ -98,8 +98,8 @@ describe('CleanupService', () => { const result = await service.runDbCleanup(); - // Three DELETE calls: scheduled_notifications + rate_limit_events + execution_log - expect(db.run).toHaveBeenCalledTimes(3); + // Four DELETE calls, including processed event metadata. + expect(db.run).toHaveBeenCalledTimes(4); expect(db.run).toHaveBeenCalledWith( expect.stringContaining('DELETE FROM scheduled_notifications'), expect.any(Array), @@ -112,9 +112,29 @@ describe('CleanupService', () => { expect.stringContaining('DELETE FROM notification_execution_log'), expect.any(Array), ); + expect(db.run).toHaveBeenCalledWith( + expect.stringContaining('DELETE FROM processed_events'), + expect.any(Array), + ); expect(result.notifications).toBe(5); expect(result.rateLimitEvents).toBe(5); expect(result.executionLogs).toBe(5); + expect(result.processedEvents).toBe(5); + }); + + it('uses the configured retention duration for processed event metadata', async () => { + const db = makeDb(); + const registry = new EventRegistry(); + const service = new CleanupService(db, registry, { + processedEventRetentionMs: 60_000, + }); + + await service.runDbCleanup(); + + expect(db.run).toHaveBeenCalledWith( + expect.stringContaining('DELETE FROM processed_events'), + [60], + ); }); it('stop clears the interval and stops registry cleanup', async () => { diff --git a/listener/src/services/cleanup-service.ts b/listener/src/services/cleanup-service.ts index beadaec0..3679617a 100644 --- a/listener/src/services/cleanup-service.ts +++ b/listener/src/services/cleanup-service.ts @@ -11,6 +11,8 @@ export interface CleanupConfig { rateLimitEventRetentionMs: number; /** Retain notification execution log rows for this long (ms). Default: 90 days. */ executionLogRetentionMs: number; + /** Retain processed event metadata for this long (ms). Default: 30 days. */ + processedEventRetentionMs: number; } const DEFAULTS: CleanupConfig = { @@ -18,6 +20,7 @@ const DEFAULTS: CleanupConfig = { notificationRetentionMs: 7 * 24 * 60 * 60 * 1000, rateLimitEventRetentionMs: 24 * 60 * 60 * 1000, executionLogRetentionMs: 90 * 24 * 60 * 60 * 1000, + processedEventRetentionMs: 30 * 24 * 60 * 60 * 1000, }; export class CleanupService { @@ -48,12 +51,18 @@ export class CleanupService { logger.info('CleanupService stopped'); } - async runDbCleanup(): Promise<{ notifications: number; executionLogs: number; rateLimitEvents: number }> { + async runDbCleanup(): Promise<{ + notifications: number; + executionLogs: number; + rateLimitEvents: number; + processedEvents: number; + }> { const notificationCutoff = new Date(Date.now() - this.config.notificationRetentionMs).toISOString(); const rateLimitCutoff = new Date(Date.now() - this.config.rateLimitEventRetentionMs).toISOString(); const executionLogCutoff = new Date(Date.now() - this.config.executionLogRetentionMs).toISOString(); + const processedEventRetentionSeconds = this.config.processedEventRetentionMs / 1000; - const [notifResult, rateLimitResult, executionLogResult] = await Promise.all([ + const [notifResult, rateLimitResult, executionLogResult, processedEventResult] = await Promise.all([ this.db.run( `DELETE FROM scheduled_notifications WHERE status IN ('COMPLETED','FAILED','CANCELLED') @@ -68,12 +77,18 @@ export class CleanupService { `DELETE FROM notification_execution_log WHERE execution_time < ?`, [executionLogCutoff], ), + this.db.run( + `DELETE FROM processed_events + WHERE processed_at < datetime('now', '-' || ? || ' seconds')`, + [processedEventRetentionSeconds], + ), ]); const result = { notifications: notifResult.changes, executionLogs: executionLogResult.changes, rateLimitEvents: rateLimitResult.changes, + processedEvents: processedEventResult.changes, }; logger.info('DB cleanup completed', result); diff --git a/listener/src/types/index.ts b/listener/src/types/index.ts index 5ff0c023..24fe1aa0 100644 --- a/listener/src/types/index.ts +++ b/listener/src/types/index.ts @@ -93,6 +93,8 @@ export interface AppCleanupConfig { rateLimitEventRetentionMs: number; /** Retain in-memory events for this long (ms). */ eventRetentionMs: number; + /** Retain processed event metadata for this long (ms). Default: 30 days. */ + processedEventRetentionMs: number; /** Retain notification execution log rows for this long (ms). */ executionLogRetentionMs: number; }