generated from hirosystems/.github
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: detect and fetch missing pox cycle signer sets (#16)
* feat: stacker-set updater background service * feat: detect and fetch missing pox cycle signer sets * fix: re-queue stacker-set fetch jobs on failure
- Loading branch information
Showing
6 changed files
with
203 additions
and
21 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,97 @@ | ||
| import { PgStore } from '../pg/pg-store'; | ||
| import PQueue from 'p-queue'; | ||
| import { fetchStackerSet, getStacksNodeUrl } from './stacks-core-rpc-client'; | ||
| import { sleep } from '../helpers'; | ||
| import { logger } from '@hirosystems/api-toolkit'; | ||
| import { DbRewardSetSigner } from '../pg/types'; | ||
|
|
||
| // TODO: make this configurable | ||
| // How long to wait between retries when fetching fails | ||
| const FETCH_STACKER_SET_RETRY_INTERVAL_MS = 3000; | ||
|
|
||
| // TODO: make this configurable | ||
| const FETCH_STACKER_SET_CONCURRENCY_LIMIT = 2; | ||
|
|
||
| export class StackerSetUpdator { | ||
| private readonly queue: PQueue; | ||
| private readonly db: PgStore; | ||
| private readonly abortController: AbortController; | ||
| private readonly queuedCycleNumbers = new Set<number>(); | ||
|
|
||
| constructor(args: { db: PgStore }) { | ||
| this.db = args.db; | ||
| this.abortController = new AbortController(); | ||
| this.queue = new PQueue({ | ||
| concurrency: FETCH_STACKER_SET_CONCURRENCY_LIMIT, | ||
| autoStart: true, | ||
| }); | ||
| this.db.chainhook.events.on('missingStackerSet', ({ cycleNumber }) => { | ||
| this.add({ cycleNumber }); | ||
| }); | ||
| } | ||
|
|
||
| async stop() { | ||
| this.abortController.abort(); | ||
| await this.queue.onIdle(); | ||
| this.queue.pause(); | ||
| } | ||
|
|
||
| add({ cycleNumber }: { cycleNumber: number }): void { | ||
| if (this.queuedCycleNumbers.has(cycleNumber) || this.abortController.signal.aborted) { | ||
| return; | ||
| } | ||
| this.queuedCycleNumbers.add(cycleNumber); | ||
| void this.queue | ||
| .add(() => this.fetchStackerSet(cycleNumber)) | ||
| .catch(error => { | ||
| if (!this.abortController.signal.aborted) { | ||
| logger.error(error, `Unexpected stacker-set fetch queue error for cycle ${cycleNumber}`); | ||
| this.queuedCycleNumbers.delete(cycleNumber); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| private async fetchStackerSet(cycleNumber: number) { | ||
| try { | ||
| logger.info(`Fetching stacker set for cycle ${cycleNumber} from stacks-core RPC ...`); | ||
| const stackerSet = await fetchStackerSet(cycleNumber, this.abortController.signal); | ||
| if (stackerSet.prePox4) { | ||
| logger.info(`Skipping stacker set update for cycle ${cycleNumber}, PoX-4 not yet active`); | ||
| this.queuedCycleNumbers.delete(cycleNumber); | ||
| return; // Exit job successful fetch | ||
| } | ||
| logger.info(`Fetched stacker set for cycle ${cycleNumber}, updating database ...`); | ||
| const dbRewardSetSigners = stackerSet.response.stacker_set.signers.map(entry => { | ||
| const rewardSetSigner: DbRewardSetSigner = { | ||
| cycle_number: cycleNumber, | ||
| block_height: 0, | ||
| burn_block_height: 0, | ||
| signer_key: Buffer.from(entry.signing_key.replace(/^0x/, ''), 'hex'), | ||
| signer_weight: entry.weight, | ||
| signer_stacked_amount: entry.stacked_amt.toString(), | ||
| }; | ||
| return rewardSetSigner; | ||
| }); | ||
| await this.db.chainhook.sqlWriteTransaction(async sql => { | ||
| await this.db.chainhook.insertRewardSetSigners(sql, dbRewardSetSigners); | ||
| }); | ||
| logger.info( | ||
| `Updated database with stacker set for cycle ${cycleNumber}, ${dbRewardSetSigners.length} signers` | ||
| ); | ||
| this.queuedCycleNumbers.delete(cycleNumber); | ||
| } catch (error) { | ||
| if (this.abortController.signal.aborted) { | ||
| return; // Updater service was stopped, ignore error and exit loop | ||
| } | ||
| logger.warn( | ||
| error, | ||
| `Failed to fetch stacker set for cycle ${cycleNumber}, retrying in ${FETCH_STACKER_SET_RETRY_INTERVAL_MS}ms ...` | ||
| ); | ||
| await sleep(FETCH_STACKER_SET_RETRY_INTERVAL_MS, this.abortController.signal); | ||
| setImmediate(() => { | ||
| this.queuedCycleNumbers.delete(cycleNumber); | ||
| this.add({ cycleNumber }); | ||
| }); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| import { ENV } from '../env'; | ||
|
|
||
| export function getStacksNodeUrl(): string { | ||
| return `http://${ENV.STACKS_NODE_RPC_HOST}:${ENV.STACKS_NODE_RPC_PORT}`; | ||
| } | ||
|
|
||
| export interface PoxInfo { | ||
| first_burnchain_block_height: number; | ||
| reward_cycle_length: number; | ||
| } | ||
|
|
||
| export async function fetchRpcPoxInfo(abortSignal: AbortSignal) { | ||
| const url = `${getStacksNodeUrl()}/v2/pox`; | ||
| const res = await fetch(url, { signal: abortSignal }); | ||
| const json = await res.json(); | ||
| return json as PoxInfo; | ||
| } | ||
|
|
||
| export interface RpcStackerSetResponse { | ||
| stacker_set: { | ||
| rewarded_addresses: any[]; | ||
| start_cycle_state: { | ||
| missed_reward_slots: any[]; | ||
| }; | ||
| pox_ustx_threshold: number; | ||
| signers: { | ||
| signing_key: string; | ||
| stacked_amt: number; | ||
| weight: number; | ||
| }[]; | ||
| }; | ||
| } | ||
|
|
||
| export async function fetchStackerSet( | ||
| cycleNumber: number, | ||
| abortSignal: AbortSignal | ||
| ): Promise<{ prePox4: true } | { prePox4: false; response: RpcStackerSetResponse }> { | ||
| const url = `${getStacksNodeUrl()}/v3/stacker_set/${cycleNumber}`; | ||
| const res = await fetch(url, { signal: abortSignal }); | ||
| const json = await res.json(); | ||
| if (!res.ok) { | ||
| const err = JSON.stringify(json); | ||
| if (/Pre-PoX-4/i.test(err)) { | ||
| return { prePox4: true }; | ||
| } | ||
| throw new Error(`Failed to fetch stacker set for cycle ${cycleNumber}: ${err}`); | ||
| } | ||
| return { prePox4: false, response: json as RpcStackerSetResponse }; | ||
| } |