diff --git a/ENVIRONMENT_VARIABLES_AND_SECRETS.md b/ENVIRONMENT_VARIABLES_AND_SECRETS.md index 85c34ac..b6adbf6 100644 --- a/ENVIRONMENT_VARIABLES_AND_SECRETS.md +++ b/ENVIRONMENT_VARIABLES_AND_SECRETS.md @@ -141,6 +141,7 @@ Both variables must be provided together or neither. | Variable | Default | Required | Description | |---|---|---|---| | `POLL_INTERVAL_MS` | `30000` | No | How often the listener polls Stellar for new contract events (ms). | +| `EVENT_BATCH_SIZE` | `100` | No | Maximum number of blockchain events fetched in each polling cycle. Must be at least `1`. | | `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. | diff --git a/listener/.env.example b/listener/.env.example index eed6f66..c5f610b 100644 --- a/listener/.env.example +++ b/listener/.env.example @@ -135,6 +135,9 @@ WEBHOOK_SECRETS=[{"id":"default","secret":"whsec_your_secret_here"}] # How often to poll Stellar for new contract events (ms). POLL_INTERVAL_MS=30000 +# Maximum number of blockchain events fetched in each polling cycle. +EVENT_BATCH_SIZE=100 + # Maximum number of reconnect attempts when the RPC endpoint fails. MAX_RECONNECT_ATTEMPTS=5 diff --git a/listener/src/config.test.ts b/listener/src/config.test.ts index 5b24153..96dad73 100644 --- a/listener/src/config.test.ts +++ b/listener/src/config.test.ts @@ -55,6 +55,36 @@ describe('Config validation', () => { expect(() => loadConfig()).toThrow('EVENTS_API_PORT must be a valid integer, got "eighty"'); }); + it('loads the default blockchain event batch size', () => { + delete process.env.EVENT_BATCH_SIZE; + + expect(loadConfig().eventBatchSize).toBe(100); + }); + + it('loads a configured blockchain event batch size', () => { + process.env.EVENT_BATCH_SIZE = '250'; + + expect(loadConfig().eventBatchSize).toBe(250); + }); + + it('rejects a non-integer blockchain event batch size', () => { + process.env.EVENT_BATCH_SIZE = 'many'; + + expect(() => loadConfig()).toThrow( + 'EVENT_BATCH_SIZE must be a valid integer, got "many"' + ); + }); + + it('rejects a non-positive blockchain event batch size', () => { + process.env.EVENT_BATCH_SIZE = '0'; + + const config = loadConfig(); + + expect(() => validateConfig(config)).toThrow( + 'EVENT_BATCH_SIZE must be >= 1 (received: 0).' + ); + }); + it('loads default values when optional environment variables are omitted', () => { process.env.CONTRACT_ADDRESSES = JSON.stringify([{ address: 'CXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX', events: ['*'] }]); delete process.env.STELLAR_NETWORK; diff --git a/listener/src/config.ts b/listener/src/config.ts index 7e42d1a..8afb78c 100644 --- a/listener/src/config.ts +++ b/listener/src/config.ts @@ -260,6 +260,7 @@ export function loadConfig(): Config { stellarNetworkPassphrase: trimEnv('STELLAR_NETWORK_PASSPHRASE') || 'Test SDF Network ; September 2015', contractAddresses: validateContractAddresses(rawContractAddresses), pollIntervalMs: parseIntegerEnv('POLL_INTERVAL_MS', '30000'), + eventBatchSize: parseIntegerEnv('EVENT_BATCH_SIZE', '100'), maxReconnectAttempts: parseIntegerEnv('MAX_RECONNECT_ATTEMPTS', '5'), reconnectDelayMs: parseIntegerEnv('RECONNECT_DELAY_MS', '5000'), eventsApiPort: parseIntegerEnv('EVENTS_API_PORT', '8787'), @@ -395,6 +396,10 @@ export function validateConfig(config: Config): void { ); } + if (config.eventBatchSize < 1) { + errors.push(`EVENT_BATCH_SIZE must be >= 1 (received: ${config.eventBatchSize}).`); + } + if (config.maxReconnectAttempts < 1) { errors.push( `MAX_RECONNECT_ATTEMPTS must be >= 1 (received: ${config.maxReconnectAttempts}).`, diff --git a/listener/src/services/event-subscriber-reorg.test.ts b/listener/src/services/event-subscriber-reorg.test.ts index c2038ec..d8ba5ff 100644 --- a/listener/src/services/event-subscriber-reorg.test.ts +++ b/listener/src/services/event-subscriber-reorg.test.ts @@ -54,6 +54,7 @@ const testConfig: Config = { stellarRpcUrl: 'https://soroban-testnet.stellar.org:443', contractAddresses: [contractConfig], pollIntervalMs: 30000, + eventBatchSize: 100, maxReconnectAttempts: 5, reconnectDelayMs: 100, eventsApiPort: 8787, diff --git a/listener/src/services/event-subscriber.test.ts b/listener/src/services/event-subscriber.test.ts index 3dc8215..1abd452 100644 --- a/listener/src/services/event-subscriber.test.ts +++ b/listener/src/services/event-subscriber.test.ts @@ -54,6 +54,7 @@ const testConfig: Config = { stellarRpcUrl: 'https://soroban-testnet.stellar.org:443', contractAddresses: [contractConfig], pollIntervalMs: 30000, + eventBatchSize: 100, maxReconnectAttempts: 5, reconnectDelayMs: 100, eventsApiPort: 8787, @@ -268,6 +269,20 @@ describe('EventSubscriber', () => { expect(mockGetEvents.mock.calls[1][0]).toMatchObject({ cursor: 'cursor-next' }); }); + it('uses the configured event batch size for RPC requests', async () => { + const configuredBatchSize = 25; + const subscriber = new EventSubscriber({ + ...testConfig, + eventBatchSize: configuredBatchSize, + }); + + await (subscriber as any).checkForEvents(); + + expect(mockGetEvents.mock.calls[0][0]).toMatchObject({ + limit: configuredBatchSize, + }); + }); + it('tracks cursors independently per contract', async () => { const secondContract: ContractConfig = { address: 'CBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB', diff --git a/listener/src/services/event-subscriber.ts b/listener/src/services/event-subscriber.ts index 225b331..82dcda7 100644 --- a/listener/src/services/event-subscriber.ts +++ b/listener/src/services/event-subscriber.ts @@ -339,6 +339,27 @@ export class EventSubscriber { contractConfig: ContractConfig ): Promise { const lastCursor = this.lastCursors.get(contractConfig.address); + const request: StellarSDK.rpc.Api.GetEventsRequest = lastCursor + ? { + filters: [ + { + contractIds: [contractConfig.address], + type: 'contract', + }, + ], + cursor: lastCursor, + limit: this.config.eventBatchSize, + } + : { + filters: [ + { + contractIds: [contractConfig.address], + type: 'contract', + }, + ], + startLedger: 1, + limit: this.config.eventBatchSize, + }; let request: StellarSDK.rpc.Api.GetEventsRequest; diff --git a/listener/src/types/index.ts b/listener/src/types/index.ts index 9f56069..1d212d1 100644 --- a/listener/src/types/index.ts +++ b/listener/src/types/index.ts @@ -46,6 +46,8 @@ export interface Config { stellarNetworkPassphrase: string; contractAddresses: ContractConfig[]; pollIntervalMs: number; + /** Maximum number of blockchain events fetched per polling cycle (default: 100). */ + eventBatchSize: number; maxReconnectAttempts: number; reconnectDelayMs: number; eventsApiPort: number;