Merge branch 'rpc-optimization'

This commit is contained in:
De Facto 2021-02-25 11:49:29 +08:00
commit 5e49bbafec
4 changed files with 150 additions and 43 deletions

5
.gitignore vendored
View File

@ -9,4 +9,7 @@ dist
deploy.json
program/Cargo.lock
integration-example/Cargo.lock
.DS_Store
.DS_Store
.vscode/launch.json
config/setup.private.json
deploy.private.json

View File

@ -5,6 +5,7 @@ import { loadJSONFile } from "./json"
import { coinbase } from "./feeds"
import { Submitter, SubmitterConfig } from "./Submitter"
import { log } from "./log"
import { conn } from "./context"
// Look at all the available aggregators and submit to those that the wallet can
// act as an oracle.
@ -12,15 +13,19 @@ export class PriceFeeder {
constructor(
private deployInfo: AggregatorDeployFile,
private wallet: Wallet
) {
}
) {}
async start() {
// find aggregators that this wallet can act as oracle
this.startAccessibleAggregators()
}
private startAccessibleAggregators() {
private async startAccessibleAggregators() {
let slot = await conn.getSlot()
conn.onSlotChange((slotInfo) => {
slot = slotInfo.slot
})
for (let [name, aggregatorInfo] of Object.entries(
this.deployInfo.aggregators
)) {
@ -46,7 +51,8 @@ export class PriceFeeder {
// TODO: errrrr... probably make configurable on chain. hardwire for
// now, don't submit value unless btc changes at least a dollar
minValueChangeForNewRound: 100,
}
},
() => slot
)
submitter.start()

View File

@ -1,10 +1,10 @@
import { Connection } from "@solana/web3.js"
import { AccountInfo, Connection, EpochInfo } from "@solana/web3.js"
import { PublicKey, Wallet } from "solray"
import { conn } from "./context"
import { Aggregator, Submissions, Oracle } from "./schema"
import BN from "bn.js"
import { sleep } from "./utils"
import { getAccounts, getMultipleAccounts, sleep } from "./utils"
import FluxAggregator from "./FluxAggregator"
import { createLogger, Logger } from "winston"
@ -28,6 +28,8 @@ export class Submitter {
public program: FluxAggregator
public logger!: Logger
public currentValue: BN
private epoch?: EpochInfo
private refreshAccounts: () => Promise<void> = async () => {}
public reportedRound: BN
@ -38,7 +40,8 @@ export class Submitter {
private oracleOwnerWallet: Wallet,
private priceFeed: IPriceFeed,
private cfg: SubmitterConfig
private cfg: SubmitterConfig,
private getSlot: () => number
) {
this.program = new FluxAggregator(this.oracleOwnerWallet, programID)
@ -49,14 +52,13 @@ export class Submitter {
// TODO: harvest rewards if > n
public async start() {
// make sure the states are initialized
await this.reloadState()
await this.observeAggregatorState()
this.logger = log.child({
aggregator: this.aggregator.config.description,
})
await Promise.all([this.observeAggregatorState(), this.observePriceFeed()])
await this.observePriceFeed()
}
public async withdrawRewards() {
@ -74,32 +76,42 @@ export class Submitter {
// })
}
private async reloadState(loadAggregator = true) {
if (loadAggregator) {
private async updateStates() {
if (!this.aggregator) {
this.aggregator = await Aggregator.load(this.aggregatorPK)
}
this.roundSubmissions = await Submissions.load(
this.aggregator.roundSubmissions
)
this.answerSubmissions = await Submissions.load(
this.aggregator.answerSubmissions
)
const [
oracle,
roundSubmissions,
answerSubmissions,
] = await getAccounts(conn, [
this.oraclePK,
this.aggregator.roundSubmissions,
this.aggregator.answerSubmissions,
])
this.oracle = await Oracle.load(this.oraclePK)
this.oracle = Oracle.deserialize(oracle.data)
this.answerSubmissions = Submissions.deserialize(answerSubmissions.data)
this.roundSubmissions = Submissions.deserialize(roundSubmissions.data)
}
private isRoundReported(roundID: BN): boolean {
return !roundID.isZero() && roundID.lte(this.reportedRound)
}
private async observeAggregatorState() {
await this.updateStates()
conn.onAccountChange(this.aggregatorPK, async (info) => {
this.aggregator = Aggregator.deserialize(info.data)
await this.reloadState(false)
this.logger.debug("state updated", {
aggregator: this.aggregator,
submissions: this.roundSubmissions,
answerSubmissions: this.answerSubmissions,
})
if (this.isRoundReported(this.aggregator.round.id)) {
return
}
// only update states if actually reporting to save RPC calls
await this.updateStates()
this.onAggregatorStateUpdate()
})
}
@ -131,8 +143,6 @@ export class Submitter {
private async trySubmit() {
// TODO: make it possible to be triggered by chainlink task
// TODO: If from chainlink node, update state before running
this.oracle = await Oracle.load(this.oraclePK)
this.logger.debug("oracle", { oracle: this.oracle })
const { round } = this.aggregator
@ -144,10 +154,7 @@ export class Submitter {
}
// or, see if oracle can start a new round
const epoch = await conn.getEpochInfo()
const sinceLastUpdate = new BN(epoch.absoluteSlot).sub(round.updatedAt)
// console.log("slot", epoch.absoluteSlot, sinceLastUpdate.toString())
const sinceLastUpdate = new BN(this.getSlot()).sub(round.updatedAt)
if (sinceLastUpdate.ltn(MAX_ROUND_STALENESS)) {
// round is not stale yet. don't submit new round
return
@ -155,8 +162,7 @@ export class Submitter {
// The round is stale. start a new round if possible, or wait for another
// oracle to start
const oracle = await Oracle.load(this.oraclePK)
if (oracle.canStartNewRound(round.id)) {
if (this.oracle.canStartNewRound(round.id)) {
let newRoundID = round.id.addn(1)
this.logger.info("Starting a new round", {
round: newRoundID.toString(),
@ -166,6 +172,12 @@ export class Submitter {
}
private async onAggregatorStateUpdate() {
this.logger.debug("state updated", {
aggregator: this.aggregator,
submissions: this.roundSubmissions,
answerSubmissions: this.answerSubmissions,
})
if (!this.canSubmitToCurrentRound) {
return
}
@ -217,19 +229,12 @@ export class Submitter {
value,
})
await this.reloadState()
this.logger.info("Submit OK", {
withdrawable: this.oracle.withdrawable.toString(),
rewardToken: this.aggregator.config.rewardTokenAccount.toString(),
})
this.logger.info("Submit OK")
} catch (err) {
console.log(err)
this.logger.error("Submit error", {
err: err.toString(),
})
}
}
}

View File

@ -1,4 +1,4 @@
import { Connection, PublicKey } from "@solana/web3.js"
import { AccountInfo, Connection, PublicKey } from "@solana/web3.js"
import EventEmitter from "events"
import { solana, Wallet, NetworkName, Deployer } from "solray"
@ -67,3 +67,96 @@ export function eventsIter<T>(emitter: EventEmitter, key: string) {
},
}
}
export function chunks<T>(array: T[], size: number): T[][] {
return Array.apply<number, T[], T[][]>(
0,
new Array(Math.ceil(array.length / size))
).map((_, index) => array.slice(index * size, (index + 1) * size))
}
export const getAccounts = async (
connection: Connection,
keys: PublicKey[],
commitment: string = "single"
) => {
// const [a, b, c] = getAccounts(pk1, pk2, pk3)
const result = await Promise.all(
chunks(keys, 99).map((chunk) =>
getMultipleAccountsCore(
connection,
chunk.map((key) => key.toBase58()),
commitment
)
)
)
const array = result
.map(
(a) =>
a.array
.filter((acc) => !!acc)
.map((acc) => {
const { data, ...rest } = acc
const obj = {
...rest,
data: Buffer.from(data[0], "base64"),
} as AccountInfo<Buffer>
return obj
}) as AccountInfo<Buffer>[]
)
.flat()
return array
}
export const getMultipleAccounts = async (
connection: Connection,
keys: string[],
commitment: string = "single"
) => {
const result = await Promise.all(
chunks(keys, 99).map((chunk) =>
getMultipleAccountsCore(connection, chunk, commitment)
)
)
const array = result
.map(
(a) =>
a.array
.filter((acc) => !!acc)
.map((acc) => {
const { data, ...rest } = acc
const obj = {
...rest,
data: Buffer.from(data[0], "base64"),
} as AccountInfo<Buffer>
return obj
}) as AccountInfo<Buffer>[]
)
.flat()
return { keys, array }
}
const getMultipleAccountsCore = async (
connection: any,
keys: string[],
commitment: string
) => {
const args = connection._buildArgs([keys], commitment, "base64")
const unsafeRes = await connection._rpcRequest("getMultipleAccounts", args)
if (unsafeRes.error) {
throw new Error(
"failed to get info about account " + unsafeRes.error.message
)
}
if (unsafeRes.result.value) {
const array = unsafeRes.result.value as AccountInfo<string[]>[]
return { keys, array }
}
throw new Error("Unable to get account")
}