feat(listener): make blockchain event batch size configurable (#627)

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. |

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
 

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;

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}).`,

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,

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',

listener/src/services/event-subscriber.ts

@@ -339,6 +339,27 @@ export class EventSubscriber {
     contractConfig: ContractConfig
   ): Promise<StellarSDK.rpc.Api.GetEventsResponse> {
     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;
 

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;