diff --git a/example-connections.json b/example-connections.json index ee3489a..a3c8c19 100644 --- a/example-connections.json +++ b/example-connections.json @@ -27,8 +27,9 @@ "chain_id": "aca376f206b8fc25a6ed44dbdc66547c36c6c33e3a119ffbeaef943642f0e906", "http": "http://127.0.0.1:8888", "ship": "ws://127.0.0.1:8080", + "WS_ROUTER_HOST": "127.0.0.1", "WS_ROUTER_PORT": 7001, - "WS_ROUTER_HOST": "127.0.0.1" + "control_port": 7002 } } } diff --git a/interfaces/hyperionConnections.ts b/interfaces/hyperionConnections.ts index 9aa3f90..a7c8742 100644 --- a/interfaces/hyperionConnections.ts +++ b/interfaces/hyperionConnections.ts @@ -23,6 +23,7 @@ interface HyperionChainData { ship: string; WS_ROUTER_PORT: number; WS_ROUTER_HOST: string; + control_port: number; } interface RedisConfig { diff --git a/modules/master.ts b/modules/master.ts index ffb25b1..49bcf1c 100644 --- a/modules/master.ts +++ b/modules/master.ts @@ -177,9 +177,11 @@ export class HyperionMaster { private repairReader?: HyperionWorkerDef; private pendingRepairRanges: { start: number, - end: number + end: number, + size?: number }[] = []; private connectedController?: WebSocket; + private lastIrreversibleBlock: number = 0; constructor() { this.cm = new ConfigurationModule(); @@ -336,19 +338,19 @@ export class HyperionMaster { // Assign next range const start = this.lastAssignedBlock; let end = this.lastAssignedBlock + this.maxBatchSize; + // Check if we are not exceeding the head block if (end > this.head) { end = this.head; } - const def = { - first_block: start, - last_block: end - }; - this.lastAssignedBlock = def.last_block; + this.lastAssignedBlock = end; this.activeReadersCount++; messageAllWorkers(cluster, { event: 'new_range', target: msg.id, - data: def + data: { + first_block: start, + last_block: end + } }); } else { if (this.lastAssignedBlock >= this.head) { @@ -416,6 +418,10 @@ export class HyperionMaster { 'lib_update': (msg: any) => { // publish LIB to hub if (msg.data) { + + // save local lib + this.lastIrreversibleBlock = msg.data.block_num; + this.revBlockArray = this.revBlockArray.filter(item => item.num > msg.data.block_num); if (this.conf.hub && this.conf.hub.inform_url) { this.hub.emit('hyp_ev', {e: 'lib', d: msg.data}); @@ -1498,8 +1504,8 @@ export class HyperionMaster { const log_msg = []; // print current head for live reading - if (this.lastProducedBlockNum > 0) { - log_msg.push(`#${this.lastProducedBlockNum}`); + if (this.lastProducedBlockNum > 0 && this.lastIrreversibleBlock > 0) { + log_msg.push(`H:${this.lastProducedBlockNum} L:${this.lastIrreversibleBlock}`); } log_msg.push(`W:${_workers}`); @@ -1744,12 +1750,18 @@ export class HyperionMaster { return range.end - range.start >= 0; }); this.pendingRepairRanges.reverse(); - console.log(this.pendingRepairRanges); + let totalBlocks = 0; + this.pendingRepairRanges.forEach(value => { + value["size"] = value.end - value.start + 1; + totalBlocks += value.size; + }); + hLog(`Filling ${totalBlocks} missing blocks...`); this.repairReader = this.addWorker({ worker_role: 'repair_reader' }); this.launchWorkers(); this.connectedController = ws; + // for (const blockRange of data) { // console.log(`Filling missing blocks ${blockRange.start} - ${blockRange.end}`); // ws.send(JSON.stringify({ @@ -1759,7 +1771,7 @@ export class HyperionMaster { // } } - private createLocalController() { + private createLocalController(controlPort: number) { this.localController = App().ws('/local', { open: (ws) => { hLog(`Local controller connected!`); @@ -1775,8 +1787,10 @@ export class HyperionMaster { hLog(`Local controller disconnected!`); } }); - this.localController.listen(4321, (listenSocket) => { - hLog(`Local controller listening on port 4321`); + this.localController.listen(controlPort, (token) => { + if (token) { + hLog(`Local controller listening on port ${controlPort}`); + } }); } @@ -1804,8 +1818,7 @@ export class HyperionMaster { } this.printMode(); - - // this.createLocalController(); + this.createLocalController(this.manager.conn.chains[this.conf.settings.chain].control_port); // Preview mode - prints only the proposed worker map let preview = this.conf.settings.preview; diff --git a/scripts/hyp-config.ts b/scripts/hyp-config.ts index 433e7a7..bb93eb6 100644 --- a/scripts/hyp-config.ts +++ b/scripts/hyp-config.ts @@ -146,7 +146,8 @@ async function newChain(shortName, options) { http: '', chain_id: '', WS_ROUTER_HOST: '127.0.0.1', - WS_ROUTER_PORT: 7001 + WS_ROUTER_PORT: 7001, + control_port: 7002 }; } diff --git a/scripts/hyp-repair.ts b/scripts/hyp-repair.ts index f454e36..2945a18 100644 --- a/scripts/hyp-repair.ts +++ b/scripts/hyp-repair.ts @@ -48,7 +48,7 @@ async function run(client: Client, rpc: JsonRpc, indexName: string, blockInit: n blockFinal = blockInitial - qtdTotal; progressBar.update(i); } catch (e: any) { - console.log("Error: ", e.message); + console.log("Error: ", e); process.exit(1); } } @@ -66,7 +66,7 @@ async function findForksOnRange(blocks: HyperionBlock[], rpc: JsonRpc) { const removals: Set = new Set(); let start = null; let end = null; - if (pendingBlock && blocks[0].block_num !== pendingBlock.block_num) { + if (blocks.length > 0 && pendingBlock && blocks[0].block_num !== pendingBlock.block_num) { blocks.unshift(pendingBlock); } @@ -197,9 +197,7 @@ async function scanChain(chain: string, args: any) { } async function repairMissing(chain: string, file: string, args: any) { - console.log(chain, file, args); const chainConfig = readChainConfig(chain); - console.log(chainConfig.settings.index_version); const config = readConnectionConfig(); const client = initESClient(config); const ping = await client.ping(); @@ -208,9 +206,11 @@ async function repairMissing(chain: string, file: string, args: any) { process.exit(); } - let hyperionIndexer = 'ws://localhost:4321'; + const controlPort = config.chains[chain].control_port; + + let hyperionIndexer = `ws://localhost:${controlPort}`; if (args.host) { - hyperionIndexer = args.host; + hyperionIndexer = 'ws://' + args.host + ':' + controlPort; } const controller = new WebSocket(hyperionIndexer + '/local'); @@ -218,11 +218,16 @@ async function repairMissing(chain: string, file: string, args: any) { console.log('Connected to Hyperion Controller'); const missingBlocks = JSON.parse(readFileSync(file).toString()); console.log(missingBlocks); - const payload = { - "event": "fill_missing_blocks", - "data": missingBlocks - }; - controller.send(JSON.stringify(payload)); + if (!args.dry) { + const payload = { + "event": "fill_missing_blocks", + "data": missingBlocks + }; + controller.send(JSON.stringify(payload)); + } else { + console.log('Dry run, skipping repair'); + controller.close(); + } }); controller.on('message', (data) => { const parsed = JSON.parse(data.toString()); @@ -499,6 +504,7 @@ program.command('repair ') program.command('fill-missing ') .description('write missing blocks') + .option('-h, --host ', 'Hyperion local control api') .option('-d, --dry', 'dry-run, do not delete or repair blocks') .action(repairMissing); @@ -530,7 +536,7 @@ program.command('connect') console.log(`Failed to connect on Hyperion Indexer at ${hyperionIndexer}, please use "--host ws://ADDRESS:PORT" to specify a remote indexer connection`); }); } catch (e) { - console.log(e.message); + console.log(e); } }); diff --git a/workers/state-reader.ts b/workers/state-reader.ts index 4d9b3a9..198ac1f 100644 --- a/workers/state-reader.ts +++ b/workers/state-reader.ts @@ -43,6 +43,7 @@ export default class StateReader extends HyperionWorker { private forkedBlocks = new Map(); private idle = true; private repairMode = false; + private internalPendingRanges = []; constructor() { @@ -193,30 +194,64 @@ export default class StateReader extends HyperionWorker { newRange(data: any) { debugLog(`new_range [${data.first_block},${data.last_block}]`); + if (this.repairMode) { hLog(`Received Repair request [${data.first_block},${data.last_block}]`); } + this.local_distributed_count = 0; clearInterval(this.completionMonitoring); this.completionMonitoring = null; this.completionSignaled = false; this.local_last_block = data.last_block; + + // make sure no data is requested from before the first indexed block on ship + + console.log(this.shipInitStatus); + + if (parseInt(data.first_block) < this.shipInitStatus['trace_begin_block']) { + data.first_block = this.shipInitStatus['trace_begin_block']; + hLog("Impossible to repair requested range, first block is before the first indexed block on ship, trimming to " + data.first_block); + } + this.range_size = parseInt(data.last_block) - parseInt(data.first_block); - if (this.allowRequests) { - this.requestBlockRange(data.first_block, data.last_block); - this.pendingRequest = null; + + // if range size is not positive we need to immediately signal completion to master + if (this.range_size <= 0) { + this.emitCompletedSignal(); + } + + // make sure the range doesn't exceed the batch size + if (this.range_size > this.conf.scaling.batch_size) { + // split the range into multiple ranges + const ranges = []; + let start = parseInt(data.first_block); + let end = parseInt(data.last_block); + while (start < end) { + const s = start; + const e = start + this.conf.scaling.batch_size; + ranges.push({ + first_block: s, + last_block: e > end ? end : e + }); + start = e + 1; + } + console.log(`Splitting range [${data.first_block},${data.last_block}] into ${ranges.length} ranges`); + this.internalPendingRanges = ranges; + const next = this.internalPendingRanges.shift(); + this.newRange(next); } else { - this.pendingRequest = [data.first_block, data.last_block]; + if (this.allowRequests) { + this.requestBlockRange(data.first_block, data.last_block); + this.pendingRequest = null; + } else { + this.pendingRequest = [data.first_block, data.last_block]; + } } } onIpcMessage(msg: any): void { switch (msg.event) { - case 'pull_range': { - console.log(msg.data); - this.requestBlockRange(msg.data.start, msg.data.end); - break; - } case 'new_range': { if (msg.target === process.env.worker_id) { this.newRange(msg.data); @@ -281,10 +316,18 @@ export default class StateReader extends HyperionWorker { if (pending === 0) { debugLog(`Reader completed - ${this.range_size} - ${this.local_distributed_count}`); clearInterval(this.completionMonitoring); - process.send({ - event: 'completed', - id: process.env['worker_id'] - }); + + // check if there are any pending ranges, then signal completion to master + if (this.internalPendingRanges.length === 0) { + process.send({ + event: 'completed', + id: process.env['worker_id'] + }); + } else { + // process next range in queue + const next = this.internalPendingRanges.shift(); + this.newRange(next); + } } }, 200); } @@ -422,7 +465,17 @@ export default class StateReader extends HyperionWorker { prev_id = res['prev_block']['block_id']; } + // // 2% chance to modify the previous block id to simulate a fork + // if (Math.random() < 0.02) { + // // prev_id = prev_id + 'a'; + // this.local_block_num = this.local_block_num - 1; + // } + if ((this.local_block_id && prev_id && this.local_block_id !== prev_id) || (blk_num !== this.local_block_num + 1)) { + + console.log((this.local_block_id && prev_id && this.local_block_id !== prev_id), (blk_num !== this.local_block_num + 1)); + console.log(blk_num, this.local_block_num + 1); + hLog(`Unlinked block at ${prev_id} with previous block ${this.local_block_id}`); hLog(`Forked block: ${blk_num}`); try { @@ -431,24 +484,10 @@ export default class StateReader extends HyperionWorker { } catch (e) { hLog(`Failed to handle fork during live reading! - Error: ${e.message}`); } - this.local_block_id = ''; - } else { - this.local_block_num = blk_num; - this.local_block_id = blk_id; } - // if (blk_num !== this.local_block_num + 1) { - // hLog(`Expected: ${this.local_block_num + 1}, received: ${blk_num}`); - // try { - // // delete all previously stored data for the forked blocks - // await this.handleFork(res); - // } catch (e) { - // hLog(`Failed to handle fork during live reading! - Error: ${e.message}`); - // } - // } else { - // this.local_block_num = blk_num; - // this.local_block_id = blk_id; - // } + this.local_block_num = blk_num; + this.local_block_id = blk_id; if (lib.block_num > this.local_lib) { this.local_lib = lib.block_num; @@ -457,7 +496,9 @@ export default class StateReader extends HyperionWorker { } await this.stageOneDistQueue.push(task_payload); + return 1; + } else { // Detect skipped first block @@ -505,6 +546,9 @@ export default class StateReader extends HyperionWorker { } } } else { + if (this.repairMode) { + hLog("No data on requested block!"); + } this.idle = true; // hLog(`Reader is idle! - Head at: ${result[1].head.block_num}`); // this.ship.close(true); @@ -591,6 +635,7 @@ export default class StateReader extends HyperionWorker { this.local_block_num = request.start_block_num - 1; request.end_block_num = parseInt(last, 10); const reqType = 'get_blocks_request_' + this.shipRev; + console.log(request); if (this.ship.connected) { debugLog(`Reader ${process.env.worker_id} sending ${reqType} from: ${request.start_block_num} to: ${request.end_block_num}`); this.send([reqType, request]); @@ -740,4 +785,11 @@ export default class StateReader extends HyperionWorker { this.startWS(); }, 5000); } + + private emitCompletedSignal() { + process.send({ + event: 'completed', + id: process.env['worker_id'] + }); + } }