Merge pull request #56 from pinto-org/pp/sg-proxy-performance

Improve proxy performance

Author: PintoPirate

docker/docker-compose.yml

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

package-lock.json

@@ -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/datasources/subgraph-client.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/postgres/startup-seeders/apy-seeder.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/repository/subgraph/subgraph-cache.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;

src/scheduled/tasks/IndexingTask.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/apy/silo-apy.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/repository/seeders/deposit-seeder.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/scheduled/sunrise.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}`,