Compute stagger time correctly, process batches in parallel

Author: Siegrift

src/update-feeds/update-feeds.test.ts

@@ -22,8 +22,6 @@ const provider = new ethers.providers.StaticJsonRpcProvider(rpcUrl, {
   name: chainId,
 });
 
-jest.mock('../state');
-
 describe(updateFeedsModule.startUpdateFeedsLoops.name, () => {
   it('starts staggered update loops for a chain', async () => {
     jest.spyOn(stateModule, 'getState').mockReturnValue(
@@ -41,6 +39,7 @@ describe(updateFeedsModule.startUpdateFeedsLoops.name, () => {
         },
       })
     );
+    jest.spyOn(stateModule, 'updateState').mockImplementation();
     jest.spyOn(updateFeedsModule, 'runUpdateFeeds').mockImplementation();
     const intervalCalls = [] as number[];
     jest.spyOn(global, 'setInterval').mockImplementation((() => {
@@ -100,6 +99,7 @@ describe(updateFeedsModule.startUpdateFeedsLoops.name, () => {
         },
       })
     );
+    jest.spyOn(stateModule, 'updateState').mockImplementation();
     jest.spyOn(updateFeedsModule, 'runUpdateFeeds').mockImplementation();
     const intervalCalls = [] as number[];
     jest.spyOn(global, 'setInterval').mockImplementation((() => {
@@ -171,7 +171,7 @@ describe(updateFeedsModule.runUpdateFeeds.name, () => {
     expect(logger.error).toHaveBeenCalledWith('Failed to get first active dAPIs batch', new Error('provider-error'));
   });
 
-  it('fetches other batches in a staggered way and logs errors', async () => {
+  it('fetches and processes other batches in a staggered way and logs errors', async () => {
     // Prepare the mocked contract so it returns three batches (of size 1) of dAPIs and the second batch fails to load.
     const firstDapi = generateReadDapiWithIndexResponse();
     const thirdDapi = generateReadDapiWithIndexResponse();
@@ -206,13 +206,21 @@ describe(updateFeedsModule.runUpdateFeeds.name, () => {
         gasPriceStore: {},
       })
     );
+    jest.spyOn(stateModule, 'updateState').mockImplementation();
     jest
       .spyOn(checkFeedsModule, 'getUpdatableFeeds')
-      .mockResolvedValue([
-        allowPartial<updateTransactionModule.UpdatableDapi>({ dapiInfo: firstDapi }),
-        allowPartial<updateTransactionModule.UpdatableDapi>({ dapiInfo: thirdDapi }),
-      ]);
+      .mockResolvedValueOnce([allowPartial<updateTransactionModule.UpdatableDapi>({ dapiInfo: firstDapi })])
+      .mockResolvedValueOnce([allowPartial<updateTransactionModule.UpdatableDapi>({ dapiInfo: thirdDapi })]);
     jest.spyOn(updateTransactionModule, 'updateFeeds').mockResolvedValue([null, null]);
+    const processBatchCalls = [] as number[];
+    // eslint-disable-next-line @typescript-eslint/require-await
+    const originalProcessBatch = updateFeedsModule.processBatch;
+    jest
+      .spyOn(updateFeedsModule, 'processBatch')
+      .mockImplementation(async (...args: Parameters<typeof originalProcessBatch>) => {
+        processBatchCalls.push(Date.now());
+        return originalProcessBatch(...args);
+      });
 
     await updateFeedsModule.runUpdateFeeds(
       'provider-name',
@@ -231,7 +239,11 @@ describe(updateFeedsModule.runUpdateFeeds.name, () => {
     expect(utilsModule.sleep).toHaveBeenCalledTimes(3);
     expect(sleepCalls[0]).toBeGreaterThan(0); // The first stagger time is computed dynamically (the execution time is subtracted from the interval time) which is slow on CI, so we just check it's non-zero.
     expect(sleepCalls[1]).toBe(0);
-    expect(sleepCalls[2]).toBe(49.999_999_999_999_99); // Stagger time is actually 150 / 3 = 50, but there is a rounding error.
+    expect(sleepCalls[2]).toBe(50);
+
+    // Expect the call times of processBatch to be staggered as well.
+    expect(updateFeedsModule.processBatch).toHaveBeenCalledTimes(2);
+    expect(processBatchCalls[1]! - processBatchCalls[0]!).toBeGreaterThanOrEqual(100); // The stagger time is 50ms, but second batch fails to load which means the third second processBatch call needs to happen after we wait for 2 stagger times.
 
     // Expect the logs to be called with the correct context.
     expect(logger.error).toHaveBeenCalledTimes(1);
@@ -244,7 +256,7 @@ describe(updateFeedsModule.runUpdateFeeds.name, () => {
     expect(logger.debug).toHaveBeenNthCalledWith(2, 'Processing batch of active dAPIs', expect.anything());
     expect(logger.debug).toHaveBeenNthCalledWith(3, 'Fetching batches of active dAPIs', {
       batchesCount: 3,
-      staggerTime: 49.999_999_999_999_99,
+      staggerTime: 50,
     });
     expect(logger.debug).toHaveBeenNthCalledWith(4, 'Fetching batch of active dAPIs', {
       batchIndex: 1,
@@ -254,9 +266,9 @@ describe(updateFeedsModule.runUpdateFeeds.name, () => {
     });
     expect(logger.debug).toHaveBeenNthCalledWith(6, 'Processing batch of active dAPIs', expect.anything());
     expect(logger.debug).toHaveBeenNthCalledWith(7, 'Finished processing batches of active dAPIs', {
-      batchesCount: 3,
-      errorCount: 4,
-      successCount: 0,
+      dapiUpdateFailures: 2,
+      dapiUpdates: 0,
+      skippedBatchesCount: 1,
     });
   });
 
@@ -280,6 +292,7 @@ describe(updateFeedsModule.runUpdateFeeds.name, () => {
         gasPriceStore: {},
       })
     );
+    jest.spyOn(stateModule, 'updateState').mockImplementation();
     jest.spyOn(logger, 'error');
     jest.spyOn(checkFeedsModule, 'getUpdatableFeeds').mockRejectedValueOnce(new Error('unexpected-unhandled-error'));
 
@@ -364,3 +377,35 @@ describe(updateFeedsModule.processBatch.name, () => {
     await expect(feeds).resolves.toStrictEqual([]);
   });
 });
+
+describe(updateFeedsModule.calculateStaggerTime.name, () => {
+  it('calculates zero stagger time for specific edge cases', () => {
+    expect(updateFeedsModule.calculateStaggerTime(1, 10_000, 60_000)).toBe(0); // When there is only a single batch.
+    expect(updateFeedsModule.calculateStaggerTime(2, 25_000, 30_000)).toBe(0); // When there are just two batches and fetching the first batch takes too long.
+  });
+
+  it('uses remaining time to calculate stagger time when fetching batch takes too long', () => {
+    expect(updateFeedsModule.calculateStaggerTime(3, 15_000, 30_000)).toBe(7500);
+    expect(updateFeedsModule.calculateStaggerTime(10, 10_000, 50_000)).toBe(4444);
+    expect(updateFeedsModule.calculateStaggerTime(10, 20_000, 20_000)).toBe(0);
+  });
+
+  it('staggers the batches evenly', () => {
+    const firstBatchDuration = 10_000;
+    const batchCount = 11;
+    const staggerTime = updateFeedsModule.calculateStaggerTime(batchCount, firstBatchDuration, 50_000);
+
+    const fetchTimes = [0, firstBatchDuration];
+    for (let i = 1; i < batchCount - 1; i++) {
+      fetchTimes.push(fetchTimes[1]! + staggerTime * i);
+    }
+
+    expect(fetchTimes).toStrictEqual([
+      0, 10_000, 14_000, 18_000, 22_000, 26_000, 30_000, 34_000, 38_000, 42_000, 46_000,
+    ]);
+  });
+
+  it('returns zero if first batch takes more than the full update interval', () => {
+    expect(updateFeedsModule.calculateStaggerTime(3, 60_000, 30_000)).toBe(0);
+  });
+});

src/update-feeds/update-feeds.ts

@@ -8,7 +8,7 @@ import { clearSponsorLastUpdateTimestampMs, initializeGasStore, hasPendingTransa
 import { logger } from '../logger';
 import { getState, updateState } from '../state';
 import type { ChainId, ProviderName } from '../types';
-import { isFulfilled, sleep, deriveSponsorWallet } from '../utils';
+import { sleep, deriveSponsorWallet } from '../utils';
 
 import { getApi3ServerV1 } from './api3-server-v1';
 import { getUpdatableFeeds } from './check-feeds';
@@ -54,6 +54,26 @@ export const startUpdateFeedsLoops = async () => {
   );
 };
 
+export const calculateStaggerTime = (
+  batchesCount: number,
+  firstBatchDuration: number,
+  dataFeedUpdateIntervalMs: number
+) => {
+  // First batch duration should not be longer than the update interval because we have a timeout in place. However, it
+  // may happen if the fetching resolves very close to the end of timeout and the duration ends up slightly larger. It's
+  // arguably a bit better to let the function return 0 instead of throwing an error.
+  if (batchesCount <= 1 || firstBatchDuration >= dataFeedUpdateIntervalMs) return 0;
+
+  // Calculate the optimal stagger time between the the batches.
+  const optimalStaggerTime = Math.round(dataFeedUpdateIntervalMs / batchesCount);
+  // If the first batch took longer than the optimal stagger time, we use the remaining time to stagger the rest of
+  // the batches.
+  if (firstBatchDuration > optimalStaggerTime) {
+    return batchesCount === 2 ? 0 : Math.round((dataFeedUpdateIntervalMs - firstBatchDuration) / (batchesCount - 1));
+  }
+  return optimalStaggerTime;
+};
+
 export const runUpdateFeeds = async (providerName: ProviderName, chain: Chain, chainId: ChainId) => {
   await logger.runWithContext({ chainId, providerName, updateFeedsCoordinatorId: Date.now().toString() }, async () => {
     // We do not expect this function to throw, but its possible that some execution path is incorrectly handled and we
@@ -100,10 +120,6 @@ export const runUpdateFeeds = async (providerName: ProviderName, chain: Chain, c
         logger.error(`Failed to get first active dAPIs batch`, goFirstBatch.error);
         return;
       }
-      if (Date.now() >= firstBatchStartTime + dataFeedUpdateIntervalMs) {
-        logger.warn(`Fetching the first batch took the whole interval. Skipping updates.`);
-        return;
-      }
 
       const { firstBatch, dapisCount } = goFirstBatch.data;
       if (dapisCount === 0) {
@@ -120,22 +136,22 @@ export const runUpdateFeeds = async (providerName: ProviderName, chain: Chain, c
         chainId
       ).catch((error) => error);
 
-      // Calculate the stagger time between the rest of the batches.
+      // Calculate the stagger time.
       const batchesCount = Math.ceil(dapisCount / dataFeedBatchSize);
-      const staggerTime = batchesCount <= 1 ? 0 : (dataFeedUpdateInterval / batchesCount) * 1000;
+      const firstBatchDuration = Date.now() - firstBatchStartTime;
+      const staggerTime = calculateStaggerTime(batchesCount, firstBatchDuration, dataFeedUpdateIntervalMs);
 
       // Wait the remaining stagger time required after fetching the first batch.
-      const firstBatchDuration = Date.now() - firstBatchStartTime;
       await sleep(Math.max(0, staggerTime - firstBatchDuration));
 
-      // Fetch the rest of the batches in parallel in a staggered way.
+      // Fetch the rest of the batches in parallel in a staggered way and process them.
       if (batchesCount > 1) {
         logger.debug('Fetching batches of active dAPIs', { batchesCount, staggerTime });
       }
-      const otherBatches = await Promise.allSettled(
-        range(1, batchesCount).map(async (batchIndex) => {
-          await sleep((batchIndex - 1) * staggerTime);
+      const processOtherBatchesPromises = range(1, batchesCount).map(async (batchIndex) => {
+        await sleep((batchIndex - 1) * staggerTime);
 
+        const goBatch = await go(async () => {
           logger.debug(`Fetching batch of active dAPIs`, { batchIndex });
           const dapiBatchIndexStart = batchIndex * dataFeedBatchSize;
           const dapiBatchIndexEnd = Math.min(dapisCount, dapiBatchIndexStart + dataFeedBatchSize);
@@ -147,24 +163,18 @@ export const runUpdateFeeds = async (providerName: ProviderName, chain: Chain, c
           );
 
           return returndata.map((returndata) => decodeReadDapiWithIndexResponse(dapiDataRegistry, returndata));
-        })
-      );
-      for (const batch of otherBatches.filter((batch) => !isFulfilled(batch))) {
-        logger.error(`Failed to get active dAPIs batch`, (batch as PromiseRejectedResult).reason);
-      }
-      const processOtherBatchesPromises = otherBatches
-        .filter((result) => isFulfilled(result))
-        .map(async (result) =>
-          processBatch(
-            (result as PromiseFulfilledResult<DecodedReadDapiWithIndexResponse[]>).value,
-            providerName,
-            provider,
-            chainId
-          )
-        );
+        });
+        if (!goBatch.success) {
+          logger.error(`Failed to get active dAPIs batch`, goBatch.error);
+          return;
+        }
+        const batch = goBatch.data;
+
+        return processBatch(batch, providerName, provider, chainId);
+      });
 
       // Wait for all the batches to be processed and print stats from this run.
-      const processingResult = await Promise.all([
+      const processedBatches = await Promise.all([
         // eslint-disable-next-line @typescript-eslint/promise-function-async
         new Promise<Awaited<ReturnType<typeof processBatch>>>((resolve, reject) => {
           return processFirstBatchPromise.then((result) => {
@@ -175,9 +185,16 @@ export const runUpdateFeeds = async (providerName: ProviderName, chain: Chain, c
         }),
         ...processOtherBatchesPromises,
       ]);
-      const successCount = processingResult.reduce((acc, { successCount }) => acc + successCount, 0);
-      const errorCount = processingResult.reduce((acc, { errorCount }) => acc + errorCount, 0);
-      logger.debug(`Finished processing batches of active dAPIs`, { batchesCount, successCount, errorCount });
+
+      // Print stats from this run.
+      const skippedBatchesCount = processedBatches.filter((batch) => !batch).length;
+      const dapiUpdates = processedBatches.reduce((acc, batch) => acc + (batch ? batch.successCount : 0), 0);
+      const dapiUpdateFailures = processedBatches.reduce((acc, batch) => acc + (batch ? batch.errorCount : 0), 0);
+      logger.debug(`Finished processing batches of active dAPIs`, {
+        skippedBatchesCount,
+        dapiUpdates,
+        dapiUpdateFailures,
+      });
     });
 
     if (!goRunUpdateFeeds.success) {

src/utils.ts

@@ -5,10 +5,6 @@ import { AIRSEEKER_PROTOCOL_ID } from './constants';
 
 export const sleep = async (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
 
-export function isFulfilled<T>(item: PromiseSettledResult<T>): item is PromiseFulfilledResult<T> {
-  return item.status === 'fulfilled';
-}
-
 export function deriveBeaconId(airnodeAddress: string, templateId: string) {
   return goSync(() => ethers.utils.solidityKeccak256(['address', 'bytes32'], [airnodeAddress, templateId])).data;
 }