Optimize RPC usage

src/constants/raw/pinto-base.js

@@ -61,6 +61,8 @@ const DECIMALS = {
 
 const MILESTONE = {
   startSeason: 1,
+  startSeasonBlock: 22622961,
+  startSeasonBlockTimestamp: 1732035269,
   endSeason: 99999999,
   isGaugeEnabled: ({ season, block }) => true,
   // [[pauseBlock, unpauseBlock]]

src/datasources/alchemy.js

@@ -1,6 +1,18 @@
 const { Alchemy } = require('alchemy-sdk');
 const EnvUtil = require('../utils/env');
 const { ethers } = require('ethers');
+const Log = require('../utils/logging');
+
+const RPC_BATCH_OPTIONS = {
+  batchMaxCount: Number(process.env.RPC_BATCH_MAX_COUNT ?? 10),
+  batchStallTime: Number(process.env.RPC_BATCH_STALL_TIME_MS ?? 10),
+  batchMaxSize: Number(process.env.RPC_BATCH_MAX_SIZE ?? 1024 * 1024)
+};
+const ETHERS_NETWORKS = {
+  eth: { name: 'mainnet', chainId: 1 },
+  arb: { name: 'arbitrum', chainId: 42161 },
+  base: { name: 'base', chainId: 8453 }
+};
 
 class AlchemyUtil {
   // Contains alchemy object
@@ -10,22 +22,22 @@ class AlchemyUtil {
   // Access to the underlying promise whos execution populates _providers.
   // Allows flexibility in awaiting when necessary (i.e. once at application startup)
   static _providerPromises = {};
+  static _rpcCounters = {};
 
   static {
     for (const chain of EnvUtil.getEnabledChains()) {
       if (EnvUtil.getCustomRpcUrl(chain)) {
-        this._providers[chain] = new ethers.JsonRpcProvider(EnvUtil.getCustomRpcUrl(chain));
-        // Needed to get Contract constructor to work
-        this._providers[chain]._isProvider = true;
+        this._providers[chain] = this._makeProvider(chain, EnvUtil.getCustomRpcUrl(chain));
       } else {
         const settings = {
           apiKey: EnvUtil.getAlchemyKey(),
           network: `${chain}-mainnet` // Of type alchemy-sdk.Network
         };
         this._alchemies[chain] = new Alchemy(settings);
-        this._providerPromises[chain] = this._alchemies[chain].config.getProvider().then((p) => {
-          this._providers[chain] = p;
-        });
+        this._providers[chain] = this._makeProvider(
+          chain,
+          `https://${chain}-mainnet.g.alchemy.com/v2/${EnvUtil.getAlchemyKey()}`
+        );
       }
     }
   }
@@ -38,10 +50,63 @@ class AlchemyUtil {
     return AlchemyUtil._providers[chain];
   }
 
+  static rpcCounts(chain) {
+    return AlchemyUtil._rpcCounters[chain];
+  }
+
   // Returns immediately if already resolved (and _providers is populated)
   static async ready(chain) {
     await AlchemyUtil._providerPromises[chain];
   }
+
+  static _makeProvider(chain, url) {
+    const provider = new ethers.JsonRpcProvider(url, ETHERS_NETWORKS[chain], RPC_BATCH_OPTIONS);
+    // Needed to get the alchemy-sdk Contract constructor to work with an ethers v6 provider.
+    provider._isProvider = true;
+    return AlchemyUtil._withRequestCounter(chain, provider);
+  }
+
+  static _withRequestCounter(chain, provider) {
+    if (process.env.LOG_RPC !== '1' || !provider._send) {
+      return provider;
+    }
+
+    const counter = {
+      total: 0,
+      methods: {}
+    };
+    AlchemyUtil._rpcCounters[chain] = counter;
+
+    const originalSend = provider._send.bind(provider);
+    provider._send = async (payload) => {
+      const requests = Array.isArray(payload) ? payload : [payload];
+      for (const request of requests) {
+        counter.total++;
+        counter.methods[request.method] = (counter.methods[request.method] ?? 0) + 1;
+      }
+      return await originalSend(payload);
+    };
+
+    let didLog = false;
+    const logCounts = () => {
+      if (didLog) {
+        return;
+      }
+      didLog = true;
+      Log.info(`RPC request counts for ${chain}`, JSON.stringify(counter));
+    };
+    process.once('beforeExit', logCounts);
+    process.once('exit', logCounts);
+    process.once('SIGINT', () => {
+      logCounts();
+      process.exit(0);
+    });
+    process.once('SIGTERM', () => {
+      logCounts();
+      process.exit(0);
+    });
+    return provider;
+  }
 }
 
 module.exports = AlchemyUtil;

src/datasources/contracts/contracts.js

@@ -1,5 +1,6 @@
 const { Contract: AlchemyContract } = require('alchemy-sdk');
 const { C } = require('../../constants/runtime-constants');
+const RpcCache = require('../rpc-cache');
 const SuperContract = require('./super-contract');
 const wellFunctionAbi = require('../../datasources/abi/basin/WellFunction.json');
 
@@ -20,7 +21,7 @@ class Contracts {
 
   static makeContract(address, abi, provider) {
     const underlyingContract = new AlchemyContract(address, abi, provider);
-    return new SuperContract(underlyingContract);
+    return RpcCache.wrapContract(new SuperContract(underlyingContract), address);
   }
 
   static _getDefaultContract(address, c = C()) {

src/datasources/contracts/multicall.js

@@ -0,0 +1,91 @@
+const { C } = require('../../constants/runtime-constants');
+const retryable = require('../../utils/async/retryable');
+const SuperContract = require('./super-contract');
+const Contracts = require('./contracts');
+
+const MULTICALL3_ADDRESS = '0xcA11bde05977b3631167028862bE2a173976CA11';
+const MULTICALL3_ABI = [
+  {
+    inputs: [
+      {
+        components: [
+          { internalType: 'address', name: 'target', type: 'address' },
+          { internalType: 'bool', name: 'allowFailure', type: 'bool' },
+          { internalType: 'bytes', name: 'callData', type: 'bytes' }
+        ],
+        internalType: 'struct Multicall3.Call3[]',
+        name: 'calls',
+        type: 'tuple[]'
+      }
+    ],
+    name: 'aggregate3',
+    outputs: [
+      {
+        components: [
+          { internalType: 'bool', name: 'success', type: 'bool' },
+          { internalType: 'bytes', name: 'returnData', type: 'bytes' }
+        ],
+        internalType: 'struct Multicall3.Result[]',
+        name: 'returnData',
+        type: 'tuple[]'
+      }
+    ],
+    stateMutability: 'view',
+    type: 'function'
+  }
+];
+
+class Multicall {
+  static async aggregate(calls, { blockTag = 'latest', c = C() } = {}) {
+    if (calls.length === 0) {
+      return [];
+    }
+
+    const results = new Array(calls.length);
+    const callsByBlock = calls.reduce((acc, call, index) => {
+      const callBlock = call.blockTag ?? blockTag;
+      if (!acc.has(callBlock)) {
+        acc.set(callBlock, []);
+      }
+      acc.get(callBlock).push({ call, index });
+      return acc;
+    }, new Map());
+
+    await Promise.all(
+      [...callsByBlock.entries()].map(async ([callBlock, entries]) => {
+        const multicall = Contracts.makeContract(MULTICALL3_ADDRESS, MULTICALL3_ABI, c.RPC);
+        const aggregateCalls = entries.map(({ call }) => ({
+          target: call.contract.address,
+          allowFailure: call.allowFailure ?? false,
+          callData: call.contract.interface.encodeFunctionData(call.method, call.args ?? [])
+        }));
+
+        const aggregateResults = await retryable(() =>
+          multicall.aggregate3({ target: 'SuperContract', skipTransform: true }, aggregateCalls, {
+            blockTag: callBlock
+          })
+        );
+
+        for (let i = 0; i < entries.length; ++i) {
+          const { call, index } = entries[i];
+          const result = aggregateResults[i];
+          if (!result.success) {
+            if (call.allowFailure) {
+              results[index] = null;
+              continue;
+            }
+            throw new Error(`Multicall failed for ${call.contract.address}.${call.method}`);
+          }
+
+          const fragment = call.contract.interface.getFunction(call.method);
+          const decoded = call.contract.interface.decodeFunctionResult(fragment, result.returnData);
+          results[index] = SuperContract._transformAll(fragment.outputs.length === 1 ? decoded[0] : decoded);
+        }
+      })
+    );
+
+    return results;
+  }
+}
+
+module.exports = Multicall;

src/datasources/rpc-cache.js

@@ -0,0 +1,100 @@
+const crypto = require('crypto');
+
+const BIGINT_PREFIX = 'bigint:';
+
+class RpcCache {
+  static _redisClient;
+
+  static wrapContract(contract, address) {
+    return new Proxy(contract, {
+      get(target, property, receiver) {
+        if (typeof property !== 'string') {
+          return target[property];
+        }
+        if (['interface', 'provider', 'filters', 'address', 'then'].includes(property)) {
+          return target[property];
+        }
+
+        return async (...args) => {
+          const cache = RpcCache._cacheRequest(address, property, args);
+          if (!cache) {
+            return await target[property](...args);
+          }
+
+          const cachedValue = await RpcCache._get(cache.key);
+          if (cachedValue !== null) {
+            return RpcCache._deserialize(cachedValue);
+          }
+
+          const result = await target[property](...args);
+          await RpcCache._set(cache.key, RpcCache._serialize(result));
+          return result;
+        };
+      }
+    });
+  }
+
+  static _cacheRequest(address, method, args) {
+    const superArgsIndex = args.findIndex((a) => a?.target === 'SuperContract');
+    const superOptions = superArgsIndex === -1 ? null : args[superArgsIndex];
+    if (superOptions?.skipTransform) {
+      return null;
+    }
+
+    const contractArgs = args.filter((_, idx) => idx !== superArgsIndex);
+    const callOptions = contractArgs.find((a) => a && typeof a === 'object' && a.blockTag !== undefined);
+    const blockTag = callOptions?.blockTag;
+    if (!RpcCache._isHistoricalBlockTag(blockTag)) {
+      return null;
+    }
+
+    return {
+      key: `rpc:${address.toLowerCase()}:${method}:${RpcCache._hash({ args: contractArgs, blockTag })}`
+    };
+  }
+
+  static _isHistoricalBlockTag(blockTag) {
+    if (typeof blockTag === 'number' || typeof blockTag === 'bigint') {
+      return true;
+    }
+    return typeof blockTag === 'string' && /^0x[0-9a-f]+$/i.test(blockTag);
+  }
+
+  static async _get(key) {
+    try {
+      return await RpcCache._client().get(key);
+    } catch (e) {
+      return null;
+    }
+  }
+
+  static async _set(key, value) {
+    try {
+      await RpcCache._client().set(key, value);
+    } catch (e) {}
+  }
+
+  static _client() {
+    RpcCache._redisClient ??= require('./redis-client');
+    return RpcCache._redisClient;
+  }
+
+  static _hash(value) {
+    return crypto.createHash('sha256').update(RpcCache._serialize(value)).digest('hex');
+  }
+
+  static _serialize(value) {
+    return JSON.stringify(value, (_, v) => (typeof v === 'bigint' ? `${BIGINT_PREFIX}${v.toString()}` : v));
+  }
+
+  static _deserialize(value) {
+    return JSON.parse(value, (_, v) => {
+      if (typeof v === 'string' && v.startsWith(BIGINT_PREFIX)) {
+        return BigInt(v.slice(BIGINT_PREFIX.length));
+      }
+      return v;
+    });
+  }
+}
+
+module.exports = RpcCache;

src/repository/postgres/queries/season-repository.js

@@ -20,6 +20,17 @@ class SeasonRepository {
     return maxSeason;
   }
 
+  static async findMaxSeason() {
+    return await sequelize.models.Season.findOne({
+      order: [['season', 'DESC']],
+      transaction: AsyncContext.getOrUndef('transaction')
+    });
+  }
+
+  static async getMaxSeasonBlock() {
+    return (await this.findMaxSeason())?.block;
+  }
+
   // Returns a list of all seasons that are missing
   static async findMissingSeasons(maxSeason) {
     return await SharedRepository.findMissingSeasons(SEASON_TABLE.env, maxSeason);