Improve proxy performance

docker/docker-compose.yml

@@ -32,4 +32,3 @@ services:
     command: ["redis-server", "--appendonly", "yes"]
     volumes:
       - ./.data/${DOCKER_ENV}/redis:/data
-    restart: unless-stopped

src/datasources/subgraph-client.js

@@ -16,9 +16,16 @@ class SubgraphClients {
 
   // let callNumber = 1;
   static fromUrl(url) {
-    return async (query) => {
+    const requestFunction = async (query) => {
       const client = this._getClient(url);
-      const response = await client.request(query);
+      const res = await client.rawRequest(query);
+
+      // Attach response metadata to the client function
+      requestFunction.meta = {
+        version: res.headers.get('x-version'),
+        deployment: res.headers.get('x-deployment'),
+        indexedBlock: res.headers.get('x-indexed-block')
+      };
 
       // if (EnvUtil.getDeploymentEnv().includes('local')) {
       //   // Use this to assist in mocking. Should be commented in/out as needed.
@@ -28,8 +35,9 @@ class SubgraphClients {
       //   );
       //   console.log('wrote subgraph output to test directory');
       // }
-      return response;
+      return res.data;
     };
+    return requestFunction;
   }
 
   static _getClient(url) {

src/repository/postgres/startup-seeders/apy-seeder.js

@@ -22,7 +22,7 @@ class ApySeeder {
     // Calculate and save all vapys for each season (this will take a long time for many seasons)
     const TAG = Concurrent.tag('apySeeder');
     for (const season of missingSeasons) {
-      await Concurrent.run(TAG, 3, async () => {
+      await Concurrent.run(TAG, 1, async () => {
         try {
           await YieldService.saveSeasonalApys({ season });
         } catch (e) {

src/repository/subgraph/subgraph-cache.js

@@ -1,16 +1,23 @@
 const { C } = require('../../constants/runtime-constants');
 const redisClient = require('../../datasources/redis-client');
+const { sendWebhookMessage } = require('../../utils/discord');
 const Log = require('../../utils/logging');
 const SubgraphQueryUtil = require('../../utils/subgraph-query');
 const { SG_CACHE_CONFIG } = require('./cache-config');
 const CommonSubgraphRepository = require('./common-subgraph');
 
 // Caches past season results for configured queries, enabling retrieval of the full history to be fast
 class SubgraphCache {
+  // Introspection is required at runtime to build the schema.
+  // If the schema of an underlying subgraph changes, the API must be redeployed (or apollo restarted).
+  // Therefore the schema can be cached here rather than retrieved at runtime on each request.
+  static initialIntrospection = {};
+  static introspectionDeployment = {};
+
   static async get(cacheQueryName, where) {
     const sgName = SG_CACHE_CONFIG[cacheQueryName].subgraph;
 
-    const introspection = await this.introspect(sgName);
+    const introspection = this.initialIntrospection[sgName];
 
     const { latest, cache } = await this._getCachedResults(cacheQueryName, where);
     const freshResults = await this._queryFreshResults(cacheQueryName, where, latest, introspection);
@@ -64,14 +71,17 @@ class SubgraphCache {
     }
 
     if (!fromCache) {
-      Log.info(`New deployment detected; clearing subgraph cache for ${sgName}`);
-      await this.clear(sgName);
-
-      await redisClient.set(`sg-deployment:${sgName}`, deployment);
-      await redisClient.set(`sg-introspection:${sgName}`, JSON.stringify(queryInfo));
+      await this._newDeploymentDetected(sgName, deployment);
     }
 
-    return queryInfo;
+    this.introspectionDeployment[sgName] = deployment;
+    return (this.initialIntrospection[sgName] = queryInfo);
+  }
+
+  static async _newDeploymentDetected(sgName, deployment) {
+    Log.info(`New deployment detected; clearing subgraph cache for ${sgName}`);
+    await this.clear(sgName);
+    await redisClient.set(`sg-deployment:${sgName}`, deployment);
   }
 
   // Recursively build a type string to use in the re-exported schema
@@ -88,7 +98,8 @@ class SubgraphCache {
 
   static async _getCachedResults(cacheQueryName, where) {
     const cfg = SG_CACHE_CONFIG[cacheQueryName];
-    const cachedResults = JSON.parse(await redisClient.get(`sg:${cfg.subgraph}:${cacheQueryName}:${where}`)) ?? [];
+    const redisResult = await redisClient.get(`sg:${cfg.subgraph}:${cacheQueryName}:${where}`);
+    const cachedResults = JSON.parse(redisResult) ?? [];
 
     return {
       latest:
@@ -101,8 +112,9 @@ class SubgraphCache {
 
   static async _queryFreshResults(cacheQueryName, where, latestValue, introspection, c = C()) {
     const cfg = SG_CACHE_CONFIG[cacheQueryName];
+    const sgClient = cfg.client(c);
     const results = await SubgraphQueryUtil.allPaginatedSG(
-      cfg.client(c),
+      sgClient,
       `{ ${cfg.queryName} { ${introspection[cacheQueryName].fields
         .filter((f) => !cfg.omitFields?.includes(f.name))
         .concat(cfg.syntheticFields?.map((f) => ({ name: f.queryAccessor })) ?? [])
@@ -113,6 +125,14 @@ class SubgraphCache {
       { ...cfg.paginationSettings, lastValue: latestValue }
     );
 
+    // If new deployment detected, clear the cache and send an alert that API might need restarting
+    if (sgClient.meta.deployment !== this.introspectionDeployment[cfg.subgraph]) {
+      sendWebhookMessage(
+        `New deployment detected for ${cfg.subgraph}, the API might need to be restarted (if the schema changed).`
+      );
+      await this._newDeploymentDetected(cfg.subgraph, sgClient.meta.deployment);
+    }
+
     for (const result of results) {
       for (const syntheticField of cfg.syntheticFields ?? []) {
         result[syntheticField.objectRewritePath] = syntheticField.objectAccessor(result);

src/scheduled/tasks/IndexingTask.js

@@ -40,7 +40,7 @@ class IndexingTask {
         return {
           countEvents,
           queuedCallersBehind: this._queueCounter > localCount,
-          canExecuteAgain: !this.isCaughtUp()
+          canExecuteAgain: !this.isCaughtUp() && countEvents !== false // false indicates task skipped
         };
       } finally {
         this._running = false;

test/apy/silo-apy.test.js

@@ -8,7 +8,7 @@ const GaugeApyUtil = require('../../src/service/utils/apy/gauge');
 const PreGaugeApyUtil = require('../../src/service/utils/apy/pre-gauge');
 const { toBigInt } = require('../../src/utils/number');
 const { mockBeanstalkConstants } = require('../util/mock-constants');
-const { mockBeanstalkSG } = require('../util/mock-sg');
+const { mockBeanstalkSG, mockWrappedSgReturnData } = require('../util/mock-sg');
 
 describe('Window EMA', () => {
   beforeEach(() => {
@@ -20,7 +20,7 @@ describe('Window EMA', () => {
 
   it('should calculate window EMA', async () => {
     const rewardMintResponse = require('../mock-responses/subgraph/silo-apy/siloHourlyRewardMints_1.json');
-    jest.spyOn(mockBeanstalkSG, 'request').mockResolvedValue(rewardMintResponse);
+    jest.spyOn(mockBeanstalkSG, 'rawRequest').mockResolvedValue(mockWrappedSgReturnData(rewardMintResponse));
 
     const emaResult = await SiloApyService.calcWindowEMA(21816, [24, 168, 720]);
 
@@ -45,7 +45,7 @@ describe('Window EMA', () => {
 
   it('should use up to as many season as are available', async () => {
     const rewardMintResponse = require('../mock-responses/subgraph/silo-apy/siloHourlyRewardMints_2.json');
-    jest.spyOn(mockBeanstalkSG, 'request').mockResolvedValue(rewardMintResponse);
+    jest.spyOn(mockBeanstalkSG, 'rawRequest').mockResolvedValue(mockWrappedSgReturnData(rewardMintResponse));
 
     const emaResult = await SiloApyService.calcWindowEMA(6100, [10000, 20000]);
 
@@ -291,9 +291,9 @@ describe('SiloApyService Orchestration', () => {
 
   it('pre-gauge should supply appropriate parameters', async () => {
     const seasonBlockResponse = require('../mock-responses/subgraph/silo-apy/preGaugeApyInputs_1.json');
-    jest.spyOn(mockBeanstalkSG, 'request').mockResolvedValueOnce(seasonBlockResponse);
+    jest.spyOn(mockBeanstalkSG, 'rawRequest').mockResolvedValueOnce(mockWrappedSgReturnData(seasonBlockResponse));
     const preGaugeApyInputsResponse = require('../mock-responses/subgraph/silo-apy/preGaugeApyInputs_2.json');
-    jest.spyOn(mockBeanstalkSG, 'request').mockResolvedValueOnce(preGaugeApyInputsResponse);
+    jest.spyOn(mockBeanstalkSG, 'rawRequest').mockResolvedValueOnce(mockWrappedSgReturnData(preGaugeApyInputsResponse));
 
     const spy = jest.spyOn(PreGaugeApyUtil, 'calcApy');
     spy.mockReturnValueOnce({
@@ -329,9 +329,9 @@ describe('SiloApyService Orchestration', () => {
 
   it('gauge should supply appropriate parameters', async () => {
     const seasonBlockResponse = require('../mock-responses/subgraph/silo-apy/gaugeApyInputs_1.json');
-    jest.spyOn(mockBeanstalkSG, 'request').mockResolvedValueOnce(seasonBlockResponse);
+    jest.spyOn(mockBeanstalkSG, 'rawRequest').mockResolvedValueOnce(mockWrappedSgReturnData(seasonBlockResponse));
     const gaugeApyInputsResponse = require('../mock-responses/subgraph/silo-apy/gaugeApyInputs_2.json');
-    jest.spyOn(mockBeanstalkSG, 'request').mockResolvedValueOnce(gaugeApyInputsResponse);
+    jest.spyOn(mockBeanstalkSG, 'rawRequest').mockResolvedValueOnce(mockWrappedSgReturnData(gaugeApyInputsResponse));
 
     const spy = jest.spyOn(GaugeApyUtil, 'calcApy');
     spy.mockReturnValueOnce({

test/repository/seeders/deposit-seeder.test.js

@@ -7,7 +7,7 @@ const AsyncContext = require('../../../src/utils/async/context');
 const Log = require('../../../src/utils/logging');
 const { allToBigInt } = require('../../../src/utils/number');
 const { mockBeanstalkConstants } = require('../../util/mock-constants');
-const { mockBeanstalkSG } = require('../../util/mock-sg');
+const { mockBeanstalkSG, mockWrappedSgReturnData } = require('../../util/mock-sg');
 
 describe('Deposit Seeder', () => {
   beforeEach(() => {
@@ -21,7 +21,7 @@ describe('Deposit Seeder', () => {
   });
   test('Seeds all deposits', async () => {
     const depositsResponse = require('../../mock-responses/subgraph/silo-service/allDeposits.json');
-    jest.spyOn(mockBeanstalkSG, 'request').mockResolvedValueOnce(depositsResponse);
+    jest.spyOn(mockBeanstalkSG, 'rawRequest').mockResolvedValueOnce(mockWrappedSgReturnData(depositsResponse));
 
     const whitelistInfoResponse = allToBigInt(require('../../mock-responses/service/whitelistedTokenInfo.json'));
     jest.spyOn(SiloService, 'getWhitelistedTokenInfo').mockResolvedValue(whitelistInfoResponse);

test/scheduled/sunrise.test.js

@@ -1,6 +1,6 @@
 const OnSunriseUtil = require('../../src/scheduled/util/on-sunrise');
 const { mockBeanstalkConstants } = require('../util/mock-constants');
-const { mockBeanSG, mockBasinSG, mockBeanstalkSG } = require('../util/mock-sg');
+const { mockBeanSG, mockBasinSG, mockBeanstalkSG, mockWrappedSgReturnData } = require('../util/mock-sg');
 
 async function checkLastPromiseResult(spy, expected) {
   const lastCallResult = await spy.mock.results[spy.mock.results.length - 1].value;
@@ -22,19 +22,21 @@ describe('OnSunrise', () => {
 
   it('identifies when the subgraphs have processed the new season', async () => {
     const seasonResponse = require('../mock-responses/subgraph/scheduled/sunrise/beanstalkSeason_1.json');
-    const beanstalkSGSpy = jest.spyOn(mockBeanstalkSG, 'request').mockResolvedValue(seasonResponse);
+    const beanstalkSGSpy = jest
+      .spyOn(mockBeanstalkSG, 'rawRequest')
+      .mockResolvedValue(mockWrappedSgReturnData(seasonResponse));
 
     const metaNotReady = require('../mock-responses/subgraph/scheduled/sunrise/metaNotReady.json');
-    const beanSGSpy = jest.spyOn(mockBeanSG, 'request').mockResolvedValue(metaNotReady);
-    const basinSGSpy = jest.spyOn(mockBasinSG, 'request').mockResolvedValue(metaNotReady);
+    const beanSGSpy = jest.spyOn(mockBeanSG, 'rawRequest').mockResolvedValue(mockWrappedSgReturnData(metaNotReady));
+    const basinSGSpy = jest.spyOn(mockBasinSG, 'rawRequest').mockResolvedValue(mockWrappedSgReturnData(metaNotReady));
 
     const checkSpy = jest.spyOn(OnSunriseUtil, 'checkSubgraphsForSunrise');
 
     const waitPromise = OnSunriseUtil.waitForSunrise(17501, 5 * 60 * 1000);
     await checkLastPromiseResult(checkSpy, false);
 
     const seasonResponse2 = require('../mock-responses/subgraph/scheduled/sunrise/beanstalkSeason_2.json');
-    beanstalkSGSpy.mockResolvedValue(seasonResponse2);
+    beanstalkSGSpy.mockResolvedValue(mockWrappedSgReturnData(seasonResponse2));
     // Fast-forward timers and continue
     jest.advanceTimersByTime(5000);
     jest.runAllTimers();
@@ -55,7 +57,7 @@ describe('OnSunrise', () => {
 
   test('fails to identify a new season within the time limit', async () => {
     const seasonResponse = require('../mock-responses/subgraph/scheduled/sunrise/beanstalkSeason_1.json');
-    jest.spyOn(mockBeanstalkSG, 'request').mockResolvedValue(seasonResponse);
+    jest.spyOn(mockBeanstalkSG, 'rawRequest').mockResolvedValue(mockWrappedSgReturnData(seasonResponse));
 
     const checkSpy = jest.spyOn(OnSunriseUtil, 'checkSubgraphsForSunrise');
 

test/service/exchange-service.test.js

@@ -5,7 +5,7 @@ const { getTickers, getWellPriceStats, getTrades } = require('../../src/service/
 const {
   ADDRESSES: { BEANWETH, BEANWSTETH, WETH, BEAN }
 } = require('../../src/constants/raw/beanstalk-eth');
-const { mockBasinSG } = require('../util/mock-sg');
+const { mockBasinSG, mockWrappedSgReturnData } = require('../util/mock-sg');
 const LiquidityUtil = require('../../src/service/utils/pool/liquidity');
 const ExchangeService = require('../../src/service/exchange-service');
 const { mockBeanstalkConstants } = require('../util/mock-constants');
@@ -23,7 +23,7 @@ describe('ExchangeService', () => {
 
   it('should return all Basin tickers in the expected format', async () => {
     const wellsResponse = require('../mock-responses/subgraph/basin/wells.json');
-    jest.spyOn(mockBasinSG, 'request').mockResolvedValue(wellsResponse);
+    jest.spyOn(mockBasinSG, 'rawRequest').mockResolvedValue(mockWrappedSgReturnData(wellsResponse));
     // In practice these 2 values are not necessary since the subsequent getWellPriceRange is also mocked.
     jest.spyOn(BasinSubgraphRepository, 'getAllTrades').mockResolvedValue(undefined);
     jest.spyOn(ExchangeService, 'priceEventsByWell').mockReturnValueOnce(undefined);
@@ -79,7 +79,9 @@ describe('ExchangeService', () => {
   });
 
   test('Returns swap history', async () => {
-    jest.spyOn(mockBasinSG, 'request').mockResolvedValue(require('../mock-responses/subgraph/basin/swapHistory.json'));
+    jest
+      .spyOn(mockBasinSG, 'rawRequest')
+      .mockResolvedValue(mockWrappedSgReturnData(require('../mock-responses/subgraph/basin/swapHistory.json')));
 
     const options = {
       ticker_id: `${BEAN}_${WETH}`,

test/service/field-service.test.js

@@ -1,14 +1,14 @@
 const Contracts = require('../../src/datasources/contracts/contracts');
 const FieldService = require('../../src/service/field-service');
 const { toBigInt } = require('../../src/utils/number');
-const { mockBeanstalkSG } = require('../util/mock-sg');
+const { mockBeanstalkSG, mockWrappedSgReturnData } = require('../util/mock-sg');
 
 describe('FieldService', () => {
   describe('Plot summary', () => {
     beforeEach(async () => {
       jest.restoreAllMocks();
       const allPlots = require('../mock-responses/service/field/allPlots.json');
-      jest.spyOn(mockBeanstalkSG, 'request').mockResolvedValueOnce(allPlots);
+      jest.spyOn(mockBeanstalkSG, 'rawRequest').mockResolvedValueOnce(mockWrappedSgReturnData(allPlots));
       jest.spyOn(Contracts, 'getBeanstalk').mockReturnValue({
         harvestableIndex: jest.fn().mockResolvedValue(5000)
       });
@@ -85,7 +85,7 @@ describe('FieldService', () => {
             result: 'result'
           }
         });
-        const sgSpy = jest.spyOn(mockBeanstalkSG, 'request');
+        const sgSpy = jest.spyOn(mockBeanstalkSG, 'rawRequest');
 
         const result = await FieldService.getAggregatePlotSummary({ bucketSize: 10000 });
 
@@ -100,7 +100,7 @@ describe('FieldService', () => {
             result: 'result'
           }
         });
-        const sgSpy = jest.spyOn(mockBeanstalkSG, 'request');
+        const sgSpy = jest.spyOn(mockBeanstalkSG, 'rawRequest');
 
         jest.setSystemTime(Date.now() + 1000 * 60 * 300 + 1);
         const result = await FieldService.getAggregatePlotSummary({ bucketSize: 10000 });

test/service/silo-service.test.js

@@ -6,7 +6,7 @@ const {
 } = require('../../src/constants/raw/beanstalk-eth');
 const Contracts = require('../../src/datasources/contracts/contracts');
 const whitelistedSGResponse = require('../mock-responses/subgraph/silo-service/whitelistedTokens.json');
-const { mockBeanstalkSG } = require('../util/mock-sg');
+const { mockBeanstalkSG, mockWrappedSgReturnData } = require('../util/mock-sg');
 const { mockBeanstalkConstants } = require('../util/mock-constants');
 const SiloService = require('../../src/service/silo-service');
 const { C } = require('../../src/constants/runtime-constants');
@@ -35,7 +35,7 @@ describe('SiloService', () => {
       })
     };
 
-    jest.spyOn(mockBeanstalkSG, 'request').mockResolvedValueOnce(whitelistedSGResponse);
+    jest.spyOn(mockBeanstalkSG, 'rawRequest').mockResolvedValueOnce(mockWrappedSgReturnData(whitelistedSGResponse));
     jest.spyOn(Contracts, 'getBeanstalk').mockReturnValue(mockBeanstalk);
 
     const grownStalk = await getMigratedGrownStalk(accounts, defaultOptions);
@@ -49,8 +49,8 @@ describe('SiloService', () => {
     const accounts = ['0xabcd', '0x1234'];
 
     const siloSGResponse = require('../mock-responses/subgraph/silo-service/depositedBdvs.json');
-    jest.spyOn(mockBeanstalkSG, 'request').mockResolvedValueOnce(siloSGResponse);
-    jest.spyOn(mockBeanstalkSG, 'request').mockResolvedValueOnce(whitelistedSGResponse);
+    jest.spyOn(mockBeanstalkSG, 'rawRequest').mockResolvedValueOnce(mockWrappedSgReturnData(siloSGResponse));
+    jest.spyOn(mockBeanstalkSG, 'rawRequest').mockResolvedValueOnce(mockWrappedSgReturnData(whitelistedSGResponse));
 
     const mockBeanstalk = {
       stemTipForToken: jest.fn().mockImplementation((token, options) => {

test/util/mock-sg.js

@@ -4,5 +4,15 @@ const EnvUtil = require('../../src/utils/env');
 module.exports = {
   mockBeanstalkSG: SubgraphClients._getClient(`https://graph.pinto.money/${EnvUtil.getSG('eth').BEANSTALK}`),
   mockBeanSG: SubgraphClients._getClient(`https://graph.pinto.money/${EnvUtil.getSG('eth').BEAN}`),
-  mockBasinSG: SubgraphClients._getClient(`https://graph.pinto.money/${EnvUtil.getSG('eth').BASIN}`)
+  mockBasinSG: SubgraphClients._getClient(`https://graph.pinto.money/${EnvUtil.getSG('eth').BASIN}`),
+  mockWrappedSgReturnData: (data) => {
+    return {
+      data,
+      headers: new Map([
+        ['x-version', '1.0.0'],
+        ['x-deployment', 'Qmcfyemdsh6Gw22mLZA797kswQiCYfQyXBNEEmFxYi72HN'],
+        ['x-indexed-block', 41375750]
+      ])
+    };
+  }
 };