diff --git a/src/constants/raw/pinto-base.js b/src/constants/raw/pinto-base.js index 2f76823..05829c9 100644 --- a/src/constants/raw/pinto-base.js +++ b/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]] diff --git a/src/datasources/alchemy.js b/src/datasources/alchemy.js index 37c4275..1e2d867 100644 --- a/src/datasources/alchemy.js +++ b/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; diff --git a/src/datasources/contracts/contracts.js b/src/datasources/contracts/contracts.js index 33fd7c4..af607cc 100644 --- a/src/datasources/contracts/contracts.js +++ b/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()) { diff --git a/src/datasources/contracts/multicall.js b/src/datasources/contracts/multicall.js new file mode 100644 index 0000000..2789fc8 --- /dev/null +++ b/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; diff --git a/src/datasources/rpc-cache.js b/src/datasources/rpc-cache.js new file mode 100644 index 0000000..ec9bfbd --- /dev/null +++ b/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; diff --git a/src/repository/postgres/queries/season-repository.js b/src/repository/postgres/queries/season-repository.js index f2d3771..14070fa 100644 --- a/src/repository/postgres/queries/season-repository.js +++ b/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); diff --git a/src/repository/postgres/startup-seeders/season-seeder.js b/src/repository/postgres/startup-seeders/season-seeder.js index 610f690..6f157c5 100644 --- a/src/repository/postgres/startup-seeders/season-seeder.js +++ b/src/repository/postgres/startup-seeders/season-seeder.js @@ -1,8 +1,13 @@ +const { C } = require('../../../constants/runtime-constants'); +const FilterLogs = require('../../../datasources/events/filter-logs'); +const SeasonRepository = require('../queries/season-repository'); const SeasonService = require('../../../service/season-service'); const Concurrent = require('../../../utils/async/concurrent'); const Log = require('../../../utils/logging'); const retryable = require('../../../utils/async/retryable'); +const SUNRISE_LOG_CHUNK_SIZE = 500000; + class SeasonSeeder { static __active = false; @@ -13,10 +18,34 @@ class SeasonSeeder { if (missingSeasons.length > 0) { Log.info(`Found ${missingSeasons.length} missing seasons`); } + if (missingSeasons.length === 0) { + return; + } + + const maxSeasonBlock = await SeasonRepository.getMaxSeasonBlock(); + const fromBlock = maxSeasonBlock ? maxSeasonBlock + 1 : (C().MILESTONE.startSeasonBlock ?? 22622961); + const sunriseEvents = await this.getSunriseEvents(fromBlock); + const sunriseBySeason = sunriseEvents.reduce((acc, event) => { + acc.set(Number(event.args.season), event); + return acc; + }, new Map()); + const missingWithoutBoundedEvent = []; - // Identify corresponding onchain season events - const TAG = Concurrent.tag('seasonSeeder'); for (const season of missingSeasons) { + const event = sunriseBySeason.get(season); + if (!event) { + missingWithoutBoundedEvent.push(season); + continue; + } + + await SeasonService.handleSunrise(event); + if (season % 100 === 0) { + Log.info(`Saved season ${season}...`); + } + } + + const TAG = Concurrent.tag('seasonSeeder'); + for (const season of missingWithoutBoundedEvent) { await Concurrent.run(TAG, 50, () => retryable(async () => { try { @@ -34,5 +63,21 @@ class SeasonSeeder { } await Concurrent.allSettled(TAG); } + + static async getSunriseEvents(fromBlock, toBlock = 'latest') { + const resolvedToBlock = toBlock === 'latest' ? (await C().RPC.getBlock()).number : toBlock; + const events = []; + + for (let chunkFrom = fromBlock; chunkFrom <= resolvedToBlock; chunkFrom += SUNRISE_LOG_CHUNK_SIZE + 1) { + events.push( + ...(await FilterLogs.getBeanstalkEvents(['Sunrise'], { + fromBlock: chunkFrom, + toBlock: Math.min(chunkFrom + SUNRISE_LOG_CHUNK_SIZE, resolvedToBlock) + })) + ); + } + + return events; + } } module.exports = SeasonSeeder; diff --git a/src/scheduled/tasks/inflows.js b/src/scheduled/tasks/inflows.js index d6bf41e..6cbd5e6 100644 --- a/src/scheduled/tasks/inflows.js +++ b/src/scheduled/tasks/inflows.js @@ -22,6 +22,21 @@ const FIELD_EVENTS = new Set(['Sow', 'Harvest', 'PodListingFilled', 'PodOrderFil const SILO_EVENTS = new Set(['AddDeposit', 'RemoveDeposit', 'RemoveDeposits', 'Plant', 'Convert', 'ClaimPlenty']); const ALL_EVENTS = [...FIELD_EVENTS, ...SILO_EVENTS]; +const getOrSet = async (map, key, valueFn) => { + if (!map.has(key)) { + map.set( + key, + Promise.resolve() + .then(valueFn) + .catch((e) => { + map.delete(key); + throw e; + }) + ); + } + return await map.get(key); +}; + class InflowsTask extends IndexingTask { static async handleLiveEvent(event) { // Inflows are only used for snapshots currently, therefore update on Sunrise only @@ -53,11 +68,15 @@ class InflowsTask extends IndexingTask { const siloInflowDtos = []; const fieldInflowDtos = []; + const beanPriceByBlock = new Map(); + const temperatureByBlock = new Map(); + const tokenPriceByTokenBlock = new Map(); const TAG = Concurrent.tag('inflows'); for (const txnHash in byTxn) { await Concurrent.run(TAG, 35, async () => { const txnEvents = byTxn[txnHash]; + const blockNumber = txnEvents[0].rawLog.blockNumber; const converts = txnEvents.filter((e) => e.name === 'Convert'); const plants = txnEvents.filter((e) => e.name === 'Plant'); const addRemoves = txnEvents.filter((e) => e.name.includes('Deposit')); @@ -69,19 +88,33 @@ class InflowsTask extends IndexingTask { SiloEvents.removeConvertRelatedEvents(addRemoves, converts); SiloEvents.removePlantRelatedEvents(addRemoves, plants); - const beanPrice = (await PriceService.getBeanPrice({ blockNumber: txnEvents[0].rawLog.blockNumber })).usdPrice; + const beanPrice = await getOrSet( + beanPriceByBlock, + blockNumber, + async () => (await PriceService.getBeanPrice({ blockNumber })).usdPrice + ); + const temperature = + sowEvents.length === 0 + ? null + : await getOrSet(temperatureByBlock, blockNumber, () => + FieldInflowsUtil.getTemperatureForBlock(blockNumber) + ); + const getTokenPrice = (token, block) => + getOrSet(tokenPriceByTokenBlock, `${token}|${block}`, () => + PriceService.getTokenPrice(token, { blockNumber: block }) + ); // Assign true beans sown to the sow events - await FieldInflowsUtil.assignTrueBeansSown(sowEvents, txnEvents[0].rawLog.blockNumber); + await FieldInflowsUtil.assignTrueBeansSown(sowEvents, blockNumber, temperature); // Attaches bdv/bean price to the plenty events - await SiloInflowsUtil.assignClaimPlentyBdvs(claimPlenties, beanPrice, txnEvents[0].rawLog.blockNumber); + await SiloInflowsUtil.assignClaimPlentyBdvs(claimPlenties, beanPrice, blockNumber, getTokenPrice); const netDeposits = SiloInflowsUtil.netDeposits(addRemoves); const netSilo = SiloInflowsUtil.netBdvInflows(netDeposits, claimPlenties); const netField = FieldInflowsUtil.netBdvInflows(fieldEvents); const txnMeta = { - block: txnEvents[0].rawLog.blockNumber, + block: blockNumber, timestamp: txnEvents[0].extra.timestamp, txnHash, beanPrice diff --git a/src/scheduled/tasks/tractor.js b/src/scheduled/tasks/tractor.js index a8246d7..f05d99c 100644 --- a/src/scheduled/tasks/tractor.js +++ b/src/scheduled/tasks/tractor.js @@ -23,6 +23,21 @@ const MAX_BLOCKS = 10000; const SNAPSHOT_SERVICES = [SnapshotSowService, SnapshotConvertUpService]; +const getOrSet = async (map, key, valueFn) => { + if (!map.has(key)) { + map.set( + key, + Promise.resolve() + .then(valueFn) + .catch((e) => { + map.delete(key); + throw e; + }) + ); + } + return await map.get(key); +}; + class TractorTask extends IndexingTask { static async handleLiveEvent(event) { if (['Sunrise', 'PublishRequisition', 'CancelBlueprint', 'Tractor'].includes(event.name)) { @@ -96,7 +111,7 @@ class TractorTask extends IndexingTask { TractorService.updateOrders.bind(TractorService), updateBlock, siloUpdateAccounts, - updateBlock === nextSunriseBlock + this.shouldForceUpdateAll(updateBlock, sunrise[0]) ); }) ); @@ -132,11 +147,41 @@ class TractorTask extends IndexingTask { await TractorService.cancelOrder(event.args.blueprintHash); } - static async handleTractor(event) { + static shouldForceUpdateAll(updateBlock, nextSunriseEvent) { + if (updateBlock !== nextSunriseEvent?.rawLog.blockNumber) { + return false; + } + + const season = Number(nextSunriseEvent.args?.season ?? nextSunriseEvent.args?.[0]); + if (!Number.isFinite(season)) { + Log.info(`Skipping Tractor forceUpdateAll; Sunrise event did not include a season number.`); + return false; + } + + return season % 24 === 0; + } + + static _tractorEventContracts() { + return [ + Contracts.getBeanstalk(), + Contracts.get(C().SOW_V0), + Contracts.get(C().SOW_V0_TRACTOR_HELPERS), + Contracts.get(C().CONVERT_UP_V0), + Contracts.get(C().CONVERT_UP_V0_TRACTOR_HELPERS) + ]; + } + + static async _getTractorTransactionEvents(receipt) { + return await FilterLogs.getTransactionEvents(this._tractorEventContracts(), receipt); + } + + static async handleTractor(event, context = {}) { const [receipt, order, ethPriceUsd] = await Promise.all([ - C().RPC.getTransactionReceipt(event.rawLog.transactionHash), + context.receipt ?? C().RPC.getTransactionReceipt(event.rawLog.transactionHash), (async () => (await TractorService.getOrders({ blueprintHash: event.args.blueprintHash })).orders[0])(), - PriceService.getTokenPrice(C().WETH, { blockNumber: event.rawLog.blockNumber }) + context.ethPriceUsd === undefined + ? PriceService.getTokenPrice(C().WETH, { blockNumber: event.rawLog.blockNumber }) + : { usdPrice: context.ethPriceUsd } ]); if (!order) { // Tractor event received for unpublished blueprint hash. Skip it @@ -144,16 +189,7 @@ class TractorTask extends IndexingTask { } // This should be refactored to not use the deprecated method; however in practice it is acceptable currently // because though the Convert event signature changed, it was not during the lifetime of any existing Convert blueprint. - const txnEvents = await FilterLogs.getTransactionEvents( - [ - Contracts.getBeanstalk(), - Contracts.get(C().SOW_V0), - Contracts.get(C().SOW_V0_TRACTOR_HELPERS), - Contracts.get(C().CONVERT_UP_V0), - Contracts.get(C().CONVERT_UP_V0_TRACTOR_HELPERS) - ], - receipt - ); + const txnEvents = context.txnEvents ?? (await this._getTractorTransactionEvents(receipt)); // Find events between TractorExecutionBegan and Tractor event indexes const began = txnEvents.find( @@ -193,6 +229,11 @@ class TractorTask extends IndexingTask { static async processEventsConcurrently(allEvents, eventName, handler) { const events = allEvents.filter((e) => e.name === eventName); + if (eventName === 'Tractor') { + await this.processTractorEventsConcurrently(events); + return; + } + const TAG = Concurrent.tag(eventName); for (const event of events) { await Concurrent.run(TAG, 50, async () => { @@ -201,5 +242,52 @@ class TractorTask extends IndexingTask { } await Concurrent.allResolved(TAG); } + + static async processTractorEventsConcurrently(events) { + if (events.length === 0) { + return; + } + + if (events.some((e) => !e.rawLog?.transactionHash)) { + const TAG = Concurrent.tag('Tractor'); + for (const event of events) { + await Concurrent.run(TAG, 50, async () => { + await this.handleTractor(event); + }); + } + await Concurrent.allResolved(TAG); + return; + } + + const ethPriceByBlock = new Map(); + const eventsByTxn = events.reduce((acc, event) => { + (acc[event.rawLog.transactionHash] ??= []).push(event); + return acc; + }, {}); + + const TAG = Concurrent.tag('Tractor'); + for (const [txnHash, txnEvents] of Object.entries(eventsByTxn)) { + await Concurrent.run(TAG, 50, async () => { + const blockNumber = txnEvents[0].rawLog.blockNumber; + const [receipt, ethPriceUsd] = await Promise.all([ + C().RPC.getTransactionReceipt(txnHash), + getOrSet(ethPriceByBlock, blockNumber, async () => { + const price = await PriceService.getTokenPrice(C().WETH, { blockNumber }); + return price.usdPrice; + }) + ]); + const parsedTxnEvents = await this._getTractorTransactionEvents(receipt); + + for (const event of txnEvents) { + await this.handleTractor(event, { + receipt, + txnEvents: parsedTxnEvents, + ethPriceUsd + }); + } + }); + } + await Concurrent.allResolved(TAG); + } } module.exports = TractorTask; diff --git a/src/scheduled/util/field-inflows.js b/src/scheduled/util/field-inflows.js index 22f8dd3..3198bbc 100644 --- a/src/scheduled/util/field-inflows.js +++ b/src/scheduled/util/field-inflows.js @@ -3,12 +3,20 @@ const FieldInflowDto = require('../../repository/dto/inflow/FieldInflowDto'); const { toBigInt, fromBigInt } = require('../../utils/number'); class FieldInflowsUtil { + static async getTemperatureForBlock(block) { + const beanstalk = Contracts.getBeanstalk(); + return await beanstalk.temperature({ blockTag: block }); + } + // Previously there was a bug in the Sow event emission such that the amount of beans indicated // was actually the amount of soil reduced. This occurred during the morning when above peg. // The true amount of beans sown is computed here and attached as a _beansSown field. - static async assignTrueBeansSown(sowEvents, block) { - const beanstalk = Contracts.getBeanstalk(); - const temperature = await beanstalk.temperature({ blockTag: block }); + static async assignTrueBeansSown(sowEvents, block, temperature = undefined) { + if (sowEvents.length === 0) { + return; + } + + temperature ??= await this.getTemperatureForBlock(block); for (const e of sowEvents) { e._beansSown = toBigInt(fromBigInt(BigInt(e.args.pods), 6) / (1 + fromBigInt(temperature, 6 + 2)), 6); diff --git a/src/scheduled/util/silo-inflows.js b/src/scheduled/util/silo-inflows.js index fc7737e..be2470e 100644 --- a/src/scheduled/util/silo-inflows.js +++ b/src/scheduled/util/silo-inflows.js @@ -8,12 +8,19 @@ const { bigintFloatMultiplier, bigintPercent, toBigInt, fromBigInt } = require(' class SiloInflowsUtil { // Assigns a pseudo bdv to each claim plenty event (the claimed tokens aren't whitelisted and don't have a bdv) - static async assignClaimPlentyBdvs(claimPlenties, beanPrice, block) { + static async assignClaimPlentyBdvs( + claimPlenties, + beanPrice, + block, + getTokenPrice = (token, blockNumber) => PriceService.getTokenPrice(token, { blockNumber }) + ) { + if (claimPlenties.length === 0) { + return; + } + // Price the value and bdvs of all claimed tokens const tokens = claimPlenties.map((e) => e.args.token.toLowerCase()); - const tokenPrices = ( - await Promise.all(tokens.map((t) => PriceService.getTokenPrice(t, { blockNumber: block }))) - ).map((p) => p.usdPrice); + const tokenPrices = (await Promise.all(tokens.map((t) => getTokenPrice(t, block)))).map((p) => p.usdPrice); const pseudoBdvs = tokenPrices.map((p) => p / beanPrice); for (let i = 0; i < claimPlenties.length; ++i) { diff --git a/src/service/silo-service.js b/src/service/silo-service.js index ef5e074..d487dda 100644 --- a/src/service/silo-service.js +++ b/src/service/silo-service.js @@ -1,12 +1,12 @@ const { C } = require('../constants/runtime-constants'); const Contracts = require('../datasources/contracts/contracts'); +const Multicall = require('../datasources/contracts/multicall'); const TokenRepository = require('../repository/postgres/queries/token-repository'); const BeanstalkSubgraphRepository = require('../repository/subgraph/beanstalk-subgraph'); const ArraysUtil = require('../utils/arrays'); const BlockUtil = require('../utils/block'); const Concurrent = require('../utils/async/concurrent'); const { createNumberSpread } = require('../utils/number'); -const PromiseUtil = require('../utils/async/promise'); const AsyncContext = require('../utils/async/context'); class SiloService { @@ -118,22 +118,26 @@ class SiloService { const tokenModels = await TokenRepository.findWhitelistedTokens(chain); const addresses = tokenModels.map((t) => t.address); - const results = await Promise.all( - addresses.map(async (t) => { - const [tokenSetting, stemTip, bdv] = await Promise.all([ - beanstalk.tokenSettings(t, { blockTag: block }), - beanstalk.stemTipForToken(t, { blockTag: block }), - PromiseUtil.defaultOnReject(1n)(beanstalk.bdv(t, BigInt(10 ** C().DECIMALS[t]), { blockTag: block })) - ]); - return { tokenSetting, stemTip, bdv }; - }) + const multicallResults = await Multicall.aggregate( + addresses.flatMap((t) => [ + { contract: beanstalk, method: 'tokenSettings', args: [t], blockTag: block }, + { contract: beanstalk, method: 'stemTipForToken', args: [t], blockTag: block }, + { + contract: beanstalk, + method: 'bdv', + args: [t, BigInt(10 ** C().DECIMALS[t])], + blockTag: block, + allowFailure: true + } + ]) ); return addresses.reduce((acc, next, idx) => { + const resultIndex = idx * 3; acc[next] = { - ...results[idx].tokenSetting, - stemTip: results[idx].stemTip, - bdv: results[idx].bdv + ...multicallResults[resultIndex], + stemTip: multicallResults[resultIndex + 1], + bdv: multicallResults[resultIndex + 2] ?? 1n }; return acc; }, {}); @@ -143,14 +147,23 @@ class SiloService { static async getMowStems(accountTokenPairs, blockNumber) { const results = {}; const beanstalk = Contracts.getBeanstalk(); - const TAG = Concurrent.tag('getMowStems'); - for (let i = 0; i < accountTokenPairs.length; ++i) { - const { account, token } = accountTokenPairs[i]; - await Concurrent.run(TAG, 50, async () => { - results[`${account}|${token}`] = await beanstalk.getLastMowedStem(account, token, { blockTag: blockNumber }); - }); + const pairChunks = ArraysUtil.toChunks(accountTokenPairs, 500); + + for (const pairChunk of pairChunks) { + const chunkResults = await Multicall.aggregate( + pairChunk.map(({ account, token }) => ({ + contract: beanstalk, + method: 'getLastMowedStem', + args: [account, token], + blockTag: blockNumber + })) + ); + + for (let i = 0; i < pairChunk.length; ++i) { + const { account, token } = pairChunk[i]; + results[`${account}|${token}`] = chunkResults[i]; + } } - await Concurrent.allResolved(TAG); return results; } @@ -227,22 +240,35 @@ class SiloService { const updatedTokens = []; await AsyncContext.sequelizeTransaction(async () => { - for (const tokenModel of tokenModels) { - const token = tokenModel.address; - const [supply, bdv, stalkEarnedPerSeason, stemTip, totalDeposited, totalDepositedBdv] = await Promise.all( - [ - Contracts.get(token).totalSupply(), - PromiseUtil.defaultOnReject(1n)(beanstalk.bdv(token, BigInt(10 ** tokenModel.decimals))), - (async () => { - const tokenSettings = await beanstalk.tokenSettings(token); - return tokenSettings.stalkEarnedPerSeason; - })(), - beanstalk.stemTipForToken(token), - beanstalk.getTotalDeposited(token), - beanstalk.getTotalDepositedBdv(token) - ].map(PromiseUtil.defaultOnReject(null)) - ); + const multicallResults = await Multicall.aggregate( + tokenModels.flatMap((tokenModel) => { + const token = tokenModel.address; + return [ + { contract: Contracts.get(token), method: 'totalSupply', allowFailure: true }, + { + contract: beanstalk, + method: 'bdv', + args: [token, BigInt(10 ** tokenModel.decimals)], + allowFailure: true + }, + { contract: beanstalk, method: 'tokenSettings', args: [token], allowFailure: true }, + { contract: beanstalk, method: 'stemTipForToken', args: [token], allowFailure: true }, + { contract: beanstalk, method: 'getTotalDeposited', args: [token], allowFailure: true }, + { contract: beanstalk, method: 'getTotalDepositedBdv', args: [token], allowFailure: true } + ]; + }) + ); + for (let i = 0; i < tokenModels.length; ++i) { + const tokenModel = tokenModels[i]; + const token = tokenModel.address; + const resultIndex = i * 6; + const supply = multicallResults[resultIndex] ?? null; + const bdv = multicallResults[resultIndex + 1] ?? 1n; + const stalkEarnedPerSeason = multicallResults[resultIndex + 2]?.stalkEarnedPerSeason ?? null; + const stemTip = multicallResults[resultIndex + 3] ?? null; + const totalDeposited = multicallResults[resultIndex + 4] ?? null; + const totalDepositedBdv = multicallResults[resultIndex + 5] ?? null; updatedTokens.push( ...(await TokenRepository.updateToken(token, chain, { supply, diff --git a/src/service/tractor/blueprints/convert-up.js b/src/service/tractor/blueprints/convert-up.js index 092fb24..009ca04 100644 --- a/src/service/tractor/blueprints/convert-up.js +++ b/src/service/tractor/blueprints/convert-up.js @@ -11,6 +11,7 @@ const { C } = require('../../../constants/runtime-constants'); const ConvertUpOrderDto = require('../../../repository/dto/tractor/ConvertUpOrderDto'); const Concurrent = require('../../../utils/async/concurrent'); const BlockUtil = require('../../../utils/block'); +const Log = require('../../../utils/logging'); const BeanstalkPrice = require('../../../datasources/contracts/upgradeable/beanstalk-price'); class TractorConvertUpService extends Blueprint { @@ -69,10 +70,22 @@ class TractorConvertUpService extends Blueprint { } await TractorService_updateOrders(ordersToUpdate); + const totalOrders = orders.length; + let skippedPublishers = []; if (!forceUpdateAll) { // Only update orders with recent silo activity + skippedPublishers = orders + .filter((o) => !siloUpdateAccounts.has(o.publisher.toLowerCase())) + .map((o) => o.publisher.toLowerCase()); orders = orders.filter((o) => siloUpdateAccounts.has(o.publisher.toLowerCase())); } + Log.info(`Tractor ${this.orderType} periodicUpdate`, { + blockNumber, + forceUpdateAll, + evaluatedOrders: orders.length, + skippedOrders: totalOrders - orders.length, + skippedPublishers: [...new Set(skippedPublishers)] + }); // Sort orders that can be executed first orders.sort((a, b) => { diff --git a/src/service/tractor/blueprints/sow.js b/src/service/tractor/blueprints/sow.js index 165e593..8a07726 100644 --- a/src/service/tractor/blueprints/sow.js +++ b/src/service/tractor/blueprints/sow.js @@ -9,6 +9,7 @@ const SowOrderAssembler = require('../../../repository/postgres/models/assembler const { TractorOrderType } = require('../../../repository/postgres/models/types/types'); const Concurrent = require('../../../utils/async/concurrent'); const BlockUtil = require('../../../utils/block'); +const Log = require('../../../utils/logging'); const Blueprint = require('./blueprint'); const BlueprintConstants = require('./blueprint-constants'); @@ -68,10 +69,22 @@ class TractorSowService extends Blueprint { } await TractorService_updateOrders(ordersToUpdate); + const totalOrders = orders.length; + let skippedPublishers = []; if (!forceUpdateAll) { // Only update orders with recent silo activity + skippedPublishers = orders + .filter((o) => !siloUpdateAccounts.has(o.publisher.toLowerCase())) + .map((o) => o.publisher.toLowerCase()); orders = orders.filter((o) => siloUpdateAccounts.has(o.publisher.toLowerCase())); } + Log.info(`Tractor ${this.orderType} periodicUpdate`, { + blockNumber, + forceUpdateAll, + evaluatedOrders: orders.length, + skippedOrders: totalOrders - orders.length, + skippedPublishers: [...new Set(skippedPublishers)] + }); // Sort orders that can be executed first orders.sort((a, b) => { diff --git a/src/utils/block.js b/src/utils/block.js index 9f084ea..6ea675a 100644 --- a/src/utils/block.js +++ b/src/utils/block.js @@ -27,26 +27,49 @@ class BlockUtil { // Performs a binary search lookup to find the ethereum block number closest to this timestamp static async findBlockByTimestamp(timestamp) { const provider = C().RPC; - let upper = await provider.getBlockNumber(); - let lower = 22622961; // Pinto did not exist prior to this block - let bestBlock = null; + const startBlock = C().MILESTONE.startSeasonBlock ?? 22622961; + const startTimestamp = C().MILESTONE.startSeasonBlockTimestamp ?? 1732035269; + const requestedTimestamp = Number(timestamp); + const estimatedBlockNumber = Math.max( + startBlock, + startBlock + Math.floor((requestedTimestamp - startTimestamp) / 2) + ); + let bestBlock = await provider.getBlock(estimatedBlockNumber); - while (lower <= upper) { - const mid = lower + Math.floor((upper - lower) / 2); - bestBlock = await provider.getBlock(mid); + if (!bestBlock) { + return await provider.getBlock('latest'); + } + if (bestBlock.timestamp === requestedTimestamp || estimatedBlockNumber === startBlock) { + return bestBlock; + } - if (bestBlock.timestamp == timestamp) { + const direction = bestBlock.timestamp < requestedTimestamp ? 1 : -1; + let cursorBlockNumber = bestBlock.number; + for (let i = 1; i <= 3; ++i) { + cursorBlockNumber += direction; + if (cursorBlockNumber < startBlock) { + break; + } + const nextBlock = await provider.getBlock(cursorBlockNumber); + if (!nextBlock) { break; } - if (bestBlock.timestamp < timestamp) { - lower = mid + 1; - } else { - upper = mid - 1; + + bestBlock = this._closestBlock(requestedTimestamp, bestBlock, nextBlock); + if ( + (direction > 0 && nextBlock.timestamp >= requestedTimestamp) || + (direction < 0 && nextBlock.timestamp <= requestedTimestamp) + ) { + break; } } return bestBlock; } + static _closestBlock(timestamp, a, b) { + return Math.abs(a.timestamp - timestamp) <= Math.abs(b.timestamp - timestamp) ? a : b; + } + // Transforms a block during a pause to the block right before the pause. static pauseGuard(block, c = C()) { for (const [pauseBlock, unpauseBlock] of c.MILESTONE.pauseBlocks) { diff --git a/test/datasources/multicall.test.js b/test/datasources/multicall.test.js new file mode 100644 index 0000000..22158cc --- /dev/null +++ b/test/datasources/multicall.test.js @@ -0,0 +1,74 @@ +const { ethers } = require('ethers'); + +jest.mock('../../src/datasources/contracts/contracts', () => ({ + makeContract: jest.fn() +})); + +const Contracts = require('../../src/datasources/contracts/contracts'); +const Multicall = require('../../src/datasources/contracts/multicall'); + +describe('Multicall', () => { + beforeEach(() => { + Contracts.makeContract.mockReset(); + }); + + test('decodes aggregate3 responses in request order', async () => { + const tokenInterface = new ethers.Interface(['function balanceOf(address account) view returns (uint256)']); + const token = { + address: '0x0000000000000000000000000000000000000001', + interface: tokenInterface + }; + const multicall = { + aggregate3: jest.fn().mockResolvedValue([ + { + success: true, + returnData: tokenInterface.encodeFunctionResult('balanceOf', [123n]) + } + ]) + }; + Contracts.makeContract.mockReturnValue(multicall); + + const results = await Multicall.aggregate([ + { + contract: token, + method: 'balanceOf', + args: ['0x0000000000000000000000000000000000000002'], + blockTag: 100 + } + ]); + + expect(multicall.aggregate3).toHaveBeenCalledWith( + { target: 'SuperContract', skipTransform: true }, + [ + { + target: token.address, + allowFailure: false, + callData: tokenInterface.encodeFunctionData('balanceOf', ['0x0000000000000000000000000000000000000002']) + } + ], + { blockTag: 100 } + ); + expect(results).toEqual([123n]); + }); + + test('returns null for allowed failed calls', async () => { + const tokenInterface = new ethers.Interface(['function symbol() view returns (string)']); + const multicall = { + aggregate3: jest.fn().mockResolvedValue([{ success: false, returnData: '0x' }]) + }; + Contracts.makeContract.mockReturnValue(multicall); + + const results = await Multicall.aggregate([ + { + contract: { + address: '0x0000000000000000000000000000000000000001', + interface: tokenInterface + }, + method: 'symbol', + allowFailure: true + } + ]); + + expect(results).toEqual([null]); + }); +}); diff --git a/test/datasources/rpc-cache.test.js b/test/datasources/rpc-cache.test.js new file mode 100644 index 0000000..1c62b50 --- /dev/null +++ b/test/datasources/rpc-cache.test.js @@ -0,0 +1,53 @@ +const store = new Map(); +const mockRedisClient = { + get: jest.fn((key) => Promise.resolve(store.get(key) ?? null)), + set: jest.fn((key, value) => { + store.set(key, value); + return Promise.resolve(); + }) +}; + +jest.mock('../../src/datasources/redis-client', () => mockRedisClient); + +const RpcCache = require('../../src/datasources/rpc-cache'); + +describe('RpcCache', () => { + beforeEach(() => { + store.clear(); + jest.clearAllMocks(); + }); + + test('Caches historical contract reads by method, args, and blockTag', async () => { + const contract = { + balanceOf: jest.fn().mockResolvedValue(123n) + }; + const cached = RpcCache.wrapContract(contract, '0x0000000000000000000000000000000000000001'); + + await expect(cached.balanceOf('0x0000000000000000000000000000000000000002', { blockTag: 100 })).resolves.toEqual( + 123n + ); + await expect(cached.balanceOf('0x0000000000000000000000000000000000000002', { blockTag: 100 })).resolves.toEqual( + 123n + ); + + expect(contract.balanceOf).toHaveBeenCalledTimes(1); + expect(mockRedisClient.get).toHaveBeenCalledTimes(2); + expect(mockRedisClient.set).toHaveBeenCalledTimes(1); + }); + + test('Skips cache for moving block tags', async () => { + const contract = { + balanceOf: jest.fn().mockResolvedValue(123n) + }; + const cached = RpcCache.wrapContract(contract, '0x0000000000000000000000000000000000000001'); + + await cached.balanceOf('0x0000000000000000000000000000000000000002', { blockTag: 'latest' }); + await cached.balanceOf('0x0000000000000000000000000000000000000002', { blockTag: 'safe' }); + await cached.balanceOf('0x0000000000000000000000000000000000000002', { blockTag: 'pending' }); + await cached.balanceOf('0x0000000000000000000000000000000000000002'); + + expect(contract.balanceOf).toHaveBeenCalledTimes(4); + expect(mockRedisClient.get).not.toHaveBeenCalled(); + expect(mockRedisClient.set).not.toHaveBeenCalled(); + }); +}); diff --git a/test/repository/seeders/season-seeder.test.js b/test/repository/seeders/season-seeder.test.js new file mode 100644 index 0000000..a1f9bca --- /dev/null +++ b/test/repository/seeders/season-seeder.test.js @@ -0,0 +1,59 @@ +const AlchemyUtil = require('../../../src/datasources/alchemy'); +const FilterLogs = require('../../../src/datasources/events/filter-logs'); +const SeasonRepository = require('../../../src/repository/postgres/queries/season-repository'); +const SeasonSeeder = require('../../../src/repository/postgres/startup-seeders/season-seeder'); +const SeasonService = require('../../../src/service/season-service'); + +describe('SeasonSeeder', () => { + beforeEach(() => { + jest.restoreAllMocks(); + }); + + test('Fetches missing sunrise events with one bounded getLogs call', async () => { + const season2 = { args: { season: 2 }, rawLog: { blockNumber: 101 } }; + const season3 = { args: { season: 3 }, rawLog: { blockNumber: 201 } }; + jest.spyOn(SeasonService, 'findMissingSeasons').mockResolvedValue([2, 3]); + jest.spyOn(SeasonRepository, 'getMaxSeasonBlock').mockResolvedValue(100); + jest.spyOn(AlchemyUtil, 'providerForChain').mockReturnValue({ + getBlock: jest.fn().mockResolvedValue({ number: 300 }) + }); + jest.spyOn(FilterLogs, 'getBeanstalkEvents').mockResolvedValue([season2, season3]); + const handleSpy = jest.spyOn(SeasonService, 'handleSunrise').mockResolvedValue(); + const insertSpy = jest.spyOn(SeasonService, 'insertSeason').mockResolvedValue(); + + await SeasonSeeder.run(); + + expect(FilterLogs.getBeanstalkEvents).toHaveBeenCalledTimes(1); + expect(FilterLogs.getBeanstalkEvents).toHaveBeenCalledWith(['Sunrise'], { + fromBlock: 101, + toBlock: 300 + }); + expect(handleSpy).toHaveBeenCalledTimes(2); + expect(handleSpy).toHaveBeenNthCalledWith(1, season2); + expect(handleSpy).toHaveBeenNthCalledWith(2, season3); + expect(insertSpy).not.toHaveBeenCalled(); + }); + + test('Chunks large sunrise ranges below FilterLogs max range', async () => { + jest.spyOn(AlchemyUtil, 'providerForChain').mockReturnValue({ + getBlock: jest.fn().mockResolvedValue({ number: 1200100 }) + }); + jest.spyOn(FilterLogs, 'getBeanstalkEvents').mockResolvedValue([]); + + await SeasonSeeder.getSunriseEvents(100); + + expect(FilterLogs.getBeanstalkEvents).toHaveBeenCalledTimes(3); + expect(FilterLogs.getBeanstalkEvents).toHaveBeenNthCalledWith(1, ['Sunrise'], { + fromBlock: 100, + toBlock: 500100 + }); + expect(FilterLogs.getBeanstalkEvents).toHaveBeenNthCalledWith(2, ['Sunrise'], { + fromBlock: 500101, + toBlock: 1000101 + }); + expect(FilterLogs.getBeanstalkEvents).toHaveBeenNthCalledWith(3, ['Sunrise'], { + fromBlock: 1000102, + toBlock: 1200100 + }); + }); +}); diff --git a/test/scheduled/inflows.test.js b/test/scheduled/inflows.test.js new file mode 100644 index 0000000..b85b529 --- /dev/null +++ b/test/scheduled/inflows.test.js @@ -0,0 +1,87 @@ +const EventsUtils = require('../../src/datasources/events/util'); +const FilterLogs = require('../../src/datasources/events/filter-logs'); +const SiloEvents = require('../../src/datasources/events/silo-events'); +const SiloInflowService = require('../../src/service/inflow/silo-inflow-service'); +const SiloInflowSnapshotService = require('../../src/service/inflow/silo-inflow-snapshot-service'); +const FieldInflowService = require('../../src/service/inflow/field-inflow-service'); +const FieldInflowSnapshotService = require('../../src/service/inflow/field-inflow-snapshot-service'); +const AppMetaService = require('../../src/service/meta-service'); +const PriceService = require('../../src/service/price-service'); +const InflowsTask = require('../../src/scheduled/tasks/inflows'); +const TaskRangeUtil = require('../../src/scheduled/util/task-range'); +const FieldInflowsUtil = require('../../src/scheduled/util/field-inflows'); +const SiloInflowsUtil = require('../../src/scheduled/util/silo-inflows'); +const AsyncContext = require('../../src/utils/async/context'); +const { C } = require('../../src/constants/runtime-constants'); + +describe('InflowsTask', () => { + beforeEach(() => { + jest.restoreAllMocks(); + }); + + test('Memoizes block-scoped price reads during an update run', async () => { + const events = [ + { + name: 'Sow', + args: { account: '0xaaa', pods: '1000000' }, + rawLog: { logIndex: 1, blockNumber: 1000, transactionHash: '0xtx1' }, + extra: { timestamp: 1 } + }, + { + name: 'ClaimPlenty', + args: { account: '0xaaa', token: C().WETH, plenty: '1000' }, + rawLog: { logIndex: 2, blockNumber: 1000, transactionHash: '0xtx1' }, + extra: { timestamp: 1 } + }, + { + name: 'Sow', + args: { account: '0xbbb', pods: '2000000' }, + rawLog: { logIndex: 3, blockNumber: 1000, transactionHash: '0xtx2' }, + extra: { timestamp: 1 } + }, + { + name: 'ClaimPlenty', + args: { account: '0xbbb', token: C().WETH, plenty: '2000' }, + rawLog: { logIndex: 4, blockNumber: 1000, transactionHash: '0xtx2' }, + extra: { timestamp: 1 } + } + ]; + + jest.spyOn(AppMetaService, 'getInflowMeta').mockResolvedValue({}); + jest.spyOn(TaskRangeUtil, 'getUpdateInfo').mockResolvedValue({ + isInitialized: true, + lastUpdate: 999, + updateBlock: 1000, + isCaughtUp: true + }); + jest.spyOn(FilterLogs, 'getBeanstalkEvents').mockResolvedValue(events); + jest.spyOn(EventsUtils, 'attachTimestamps').mockResolvedValue(); + jest.spyOn(EventsUtils, 'groupByTransaction').mockResolvedValue({ + '0xtx1': [events[0], events[1]], + '0xtx2': [events[2], events[3]] + }); + jest.spyOn(SiloEvents, 'removeConvertRelatedEvents').mockImplementation(() => {}); + jest.spyOn(SiloEvents, 'removePlantRelatedEvents').mockImplementation(() => {}); + const beanPriceSpy = jest.spyOn(PriceService, 'getBeanPrice').mockResolvedValue({ usdPrice: 1 }); + const temperatureSpy = jest.spyOn(FieldInflowsUtil, 'getTemperatureForBlock').mockResolvedValue(0n); + const tokenPriceSpy = jest.spyOn(PriceService, 'getTokenPrice').mockResolvedValue({ usdPrice: 1 }); + jest.spyOn(SiloInflowsUtil, 'netDeposits').mockReturnValue({}); + jest.spyOn(SiloInflowsUtil, 'netBdvInflows').mockReturnValue({}); + jest.spyOn(FieldInflowsUtil, 'netBdvInflows').mockReturnValue({}); + jest.spyOn(SiloInflowsUtil, 'inflowsFromNetDeposits').mockResolvedValue([]); + jest.spyOn(SiloInflowsUtil, 'inflowsFromClaimPlenties').mockReturnValue([]); + jest.spyOn(FieldInflowsUtil, 'inflowsFromFieldEvents').mockReturnValue([]); + jest.spyOn(AsyncContext, 'sequelizeTransaction').mockImplementation((cb) => cb()); + jest.spyOn(SiloInflowService, 'insertInflows').mockResolvedValue(); + jest.spyOn(FieldInflowService, 'insertInflows').mockResolvedValue(); + jest.spyOn(SiloInflowSnapshotService, 'takeMissingSnapshots').mockResolvedValue(); + jest.spyOn(FieldInflowSnapshotService, 'takeMissingSnapshots').mockResolvedValue(); + jest.spyOn(AppMetaService, 'setLastInflowUpdate').mockResolvedValue(); + + await InflowsTask.update(); + + expect(beanPriceSpy).toHaveBeenCalledTimes(1); + expect(temperatureSpy).toHaveBeenCalledTimes(1); + expect(tokenPriceSpy).toHaveBeenCalledTimes(1); + }); +}); diff --git a/test/setup/jest.setup.js b/test/setup/jest.setup.js index f3e1a2b..d6e502e 100644 --- a/test/setup/jest.setup.js +++ b/test/setup/jest.setup.js @@ -10,6 +10,7 @@ jest.mock('../../src/utils/env', () => { getDeploymentEnv: jest.fn(), getDiscordWebhooks: jest.fn(), getDiscordPrefix: jest.fn(), + getRedisUrl: jest.fn().mockReturnValue('redis://localhost:6379'), getSG: jest.fn().mockImplementation(() => ({ BEANSTALK: 'a', BEAN: 'b', diff --git a/test/tractor/tractor.test.js b/test/tractor/tractor.test.js index 9016703..8b3b127 100644 --- a/test/tractor/tractor.test.js +++ b/test/tractor/tractor.test.js @@ -54,7 +54,7 @@ describe('TractorTask', () => { }); jest .spyOn(FilterLogs, 'getBeanstalkEvents') - .mockResolvedValueOnce([{ name: 'Sunrise', rawLog: { blockNumber: 1000 } }]) + .mockResolvedValueOnce([{ name: 'Sunrise', args: { season: 24 }, rawLog: { blockNumber: 1000 } }]) .mockResolvedValueOnce([ { name: 'PublishRequisition', value: 1 }, { name: 'CancelBlueprint', value: 2 }, @@ -244,5 +244,59 @@ describe('TractorTask', () => { expect(executionDbSpy).toHaveBeenNthCalledWith(2, [insertedDto]); expect(insertedDto.tipUsd).toBe(1.25); }); + + test('Groups same-transaction events for shared receipt and WETH price', async () => { + const receiptSpy = jest.fn().mockResolvedValue('receipt'); + jest.spyOn(AlchemyUtil, 'providerForChain').mockReturnValue({ + getTransactionReceipt: receiptSpy + }); + const priceSpy = jest.spyOn(PriceService, 'getTokenPrice').mockResolvedValue({ usdPrice: 1500 }); + const parsedEvents = [ + { + name: 'TractorExecutionBegan', + args: { gasleft: 5000, blueprintHash: 123, nonce: 5 }, + rawLog: { index: 0 } + }, + { name: 'Tractor', args: { gasleft: 4750, blueprintHash: 123, nonce: 5 }, rawLog: { index: 100 } }, + { + name: 'TractorExecutionBegan', + args: { gasleft: 7000, blueprintHash: 456, nonce: 6 }, + rawLog: { index: 101 } + }, + { name: 'Tractor', args: { gasleft: 6500, blueprintHash: 456, nonce: 6 }, rawLog: { index: 200 } } + ]; + const parseSpy = jest.spyOn(TractorTask, '_getTractorTransactionEvents').mockResolvedValue(parsedEvents); + const handleSpy = jest.spyOn(TractorTask, 'handleTractor').mockResolvedValue(); + const groupedEvents = [ + { + name: 'Tractor', + args: { gasleft: 4750, blueprintHash: 123, nonce: 5 }, + rawLog: { transactionHash: '0xtx', blockNumber: 1000, index: 100 } + }, + { + name: 'Tractor', + args: { gasleft: 6500, blueprintHash: 456, nonce: 6 }, + rawLog: { transactionHash: '0xtx', blockNumber: 1000, index: 200 } + } + ]; + + await TractorTask.processTractorEventsConcurrently(groupedEvents); + + expect(receiptSpy).toHaveBeenCalledTimes(1); + expect(receiptSpy).toHaveBeenCalledWith('0xtx'); + expect(priceSpy).toHaveBeenCalledTimes(1); + expect(parseSpy).toHaveBeenCalledTimes(1); + expect(handleSpy).toHaveBeenCalledTimes(2); + expect(handleSpy).toHaveBeenNthCalledWith(1, groupedEvents[0], { + receipt: 'receipt', + txnEvents: parsedEvents, + ethPriceUsd: 1500 + }); + expect(handleSpy).toHaveBeenNthCalledWith(2, groupedEvents[1], { + receipt: 'receipt', + txnEvents: parsedEvents, + ethPriceUsd: 1500 + }); + }); }); }); diff --git a/test/util/mock-constants.js b/test/util/mock-constants.js index 4bfa05e..9397aa0 100644 --- a/test/util/mock-constants.js +++ b/test/util/mock-constants.js @@ -14,6 +14,7 @@ function mockBeanstalkConstants() { } function mockPintoConstants() { + jest.spyOn(EnvUtil, 'defaultChain').mockReturnValue('base'); jest.spyOn(RuntimeConstants, '_getMapping').mockReturnValue({ base: PintoBase }); diff --git a/test/utils.test.js b/test/utils.test.js index 8d60baa..0af5ed7 100644 --- a/test/utils.test.js +++ b/test/utils.test.js @@ -12,7 +12,7 @@ const { allToBigInt, fromBigInt, percentDiff } = require('../src/utils/number'); const CommonSubgraphRepository = require('../src/repository/subgraph/common-subgraph'); const { BigInt_applyPercent } = require('../src/utils/bigint'); const { C } = require('../src/constants/runtime-constants'); -const { mockBeanstalkConstants } = require('./util/mock-constants'); +const { mockBeanstalkConstants, mockPintoConstants } = require('./util/mock-constants'); describe('Utils', () => { beforeEach(() => { @@ -53,6 +53,27 @@ describe('Utils', () => { expect(result.number).toEqual(19500000); }); + test('Finds Base block by timestamp with arithmetic estimate', async () => { + mockPintoConstants(); + const getBlockSpy = jest.spyOn(alchemy.providerForChain(), 'getBlock').mockImplementation((blockNumber) => { + const base = { number: 22622961, timestamp: 1732035269 }; + if (blockNumber === 'latest') { + return Promise.resolve({ number: 22622963, timestamp: 1732035273 }); + } + return Promise.resolve({ + number: blockNumber, + timestamp: base.timestamp + (blockNumber - base.number) * 2 + }); + }); + getBlockSpy.mockClear(); + + const result = await BlockUtil.findBlockByTimestamp(1732035273); + + expect(result).toEqual({ number: 22622963, timestamp: 1732035273 }); + expect(getBlockSpy).toHaveBeenCalledTimes(1); + expect(getBlockSpy).toHaveBeenCalledWith(22622963); + }); + describe('BigInt', () => { test('Converts all strings to BigInt', () => { const obj = allToBigInt({