From b289036b0c96389222ecf594739a84fffe780a37 Mon Sep 17 00:00:00 2001 From: Igor Lins e Silva Date: Sat, 20 May 2023 07:20:46 +0000 Subject: [PATCH] - multiple streaming improvements - stream-client v2.0.0 compatibility - improved transitions on high volume streams - added api.stream_scroll_limit (default -1 -> unlimited) specify the maximum number of documents that can be requested during a history request in the streaming api - added api.stream_scroll_batch (default 500) specify the number of documents per streamed history request --- api/helpers/functions.ts | 106 ++++++++++++++++++++++++++++++----- api/socketManager.ts | 62 ++++++++++++++++++-- chains/example.config.json | 2 + interfaces/hyperionConfig.ts | 2 + package.json | 2 +- workers/deserializer.ts | 11 +++- workers/ws-router.ts | 37 +++++++++--- 7 files changed, 194 insertions(+), 28 deletions(-) diff --git a/api/helpers/functions.ts b/api/helpers/functions.ts index e23be51..864c2db 100644 --- a/api/helpers/functions.ts +++ b/api/helpers/functions.ts @@ -16,14 +16,36 @@ export async function streamPastDeltas(fastify: FastifyInstance, socket, data) { addTermMatch(data, search_body, f); }); const responseQueue = []; + let counter = 0; + let total = 0; + let longScroll = false; + + const init_response = await fastify.elastic.search({ index: fastify.manager.chain + '-delta-*', - scroll: '30s', - size: 20, + scroll: '600s', + size: fastify.manager.config.api.stream_scroll_batch || 500, body: search_body, }); + responseQueue.push(init_response); + + const scrollLimit = fastify.manager.config.api.stream_scroll_limit; + if (scrollLimit && scrollLimit !== -1 && init_response.body.hits.total.value > scrollLimit) { + socket.emit('message', { + type: 'delta_trace', mode: 'history', messages: [], + error: `Requested ${init_response.body.hits.total.value} deltas, limit is ${scrollLimit}` + }); + return false; + } + + if (init_response.body.hits.total.value > 10000) { + total = init_response.body.hits.total.value; + longScroll = true; + hLog(`Attention! Long scroll is running!`); + } + while (responseQueue.length) { const {body} = responseQueue.shift(); counter += body['hits']['hits'].length; @@ -37,6 +59,11 @@ export async function streamPastDeltas(fastify: FastifyInstance, socket, data) { hLog('LOST CLIENT'); break; } + + if (longScroll) { + hLog(`Progress: ${counter}/${total}`); + } + if (body['hits'].total.value === counter) { hLog(`${counter} past deltas streamed to ${socket.id}`); break; @@ -45,14 +72,16 @@ export async function streamPastDeltas(fastify: FastifyInstance, socket, data) { const next_response = await fastify.elastic.scroll({ body: { scroll_id: body['_scroll_id'], - scroll: '30s' + scroll: '600s' } }); + responseQueue.push(next_response); } } export async function streamPastActions(fastify: FastifyInstance, socket, data) { + const search_body = { query: {bool: {must: []}}, sort: {global_sequence: 'asc'}, @@ -96,16 +125,53 @@ export async function streamPastActions(fastify: FastifyInstance, socket, data) const responseQueue = []; let counter = 0; + let total = 0; + let longScroll = false; + + // console.log(JSON.stringify(search_body.query.bool.must[0], null, 2)); const init_response = await fastify.elastic.search({ index: fastify.manager.chain + '-action-*', scroll: '30s', - size: 20, + size: fastify.manager.config.api.stream_scroll_batch || 500, body: search_body, }); + + const scrollLimit = fastify.manager.config.api.stream_scroll_limit; + if (scrollLimit && scrollLimit !== -1 && init_response.body.hits.total.value > scrollLimit) { + const errorMsg = `Requested ${init_response.body.hits.total.value} actions, limit is ${scrollLimit}.`; + socket.emit('message', { + reqUUID: socket.data.reqUUID, + type: 'action_trace', mode: 'history', messages: [], + error: `Requested ${init_response.body.hits.total.value} actions, limit is ${scrollLimit}.` + }); + return {status: false, error: errorMsg}; + } + + if (init_response.body.hits.total.value > 10000) { + total = init_response.body.hits.total.value; + longScroll = true; + hLog(`Attention! Long scroll is running!`); + } + + // emit first block + if (init_response.body.hits.hits.length > 0) { + socket.emit('message', { + reqUUID: socket.data.reqUUID, + type: 'action_trace_init', + mode: 'history', + first_block: init_response.body.hits.hits[0]._source.block_num, + results: init_response.body.hits.total.value + }); + } + responseQueue.push(init_response); + + let lastTransmittedBlock = 0; + let pendingScrollId = ''; while (responseQueue.length) { const {body} = responseQueue.shift(); + pendingScrollId = body['_scroll_id']; const enqueuedMessages = []; counter += body['hits']['hits'].length; for (const doc of body['hits']['hits']) { @@ -119,28 +185,40 @@ export async function streamPastActions(fastify: FastifyInstance, socket, data) } if (allow) { enqueuedMessages.push(doc._source); + if (doc._source.block_num > lastTransmittedBlock) { + lastTransmittedBlock = doc._source.block_num; + } } } if (socket.connected) { - socket.emit('message', {type: 'action_trace', mode: 'history', messages: enqueuedMessages}); + socket.emit('message', { + reqUUID: socket.data.reqUUID, + type: 'action_trace', + mode: 'history', messages: enqueuedMessages + }); } else { hLog('LOST CLIENT'); break; } + + if (longScroll) { + hLog(`Progress: ${counter}/${total}`); + } + if (body['hits'].total.value === counter) { hLog(`${counter} past actions streamed to ${socket.id}`); break; } - if (init_response.body.hits.total.value < 1000) { - const next_response = await fastify.elastic.scroll({ - body: {scroll_id: body['_scroll_id'], scroll: '30s'} - }); - responseQueue.push(next_response); - } else { - hLog('Request too large!'); - socket.emit('message', {type: 'action_trace', mode: 'history', messages: []}); - } + const next_response = await fastify.elastic.scroll({ + body: {scroll_id: body['_scroll_id'], scroll: '30s'} + }); + responseQueue.push(next_response); } + + // destroy scroll context + await fastify.elastic.clearScroll({scroll_id: pendingScrollId}); + return {status: true, lastTransmittedBlock}; + } export function addTermMatch(data, search_body, field) { diff --git a/api/socketManager.ts b/api/socketManager.ts index 484be17..8b2e79b 100644 --- a/api/socketManager.ts +++ b/api/socketManager.ts @@ -1,4 +1,4 @@ -import {hLog} from '../helpers/common_functions'; +import {hLog, sleep} from '../helpers/common_functions'; import {Server, Socket} from 'socket.io'; import {createAdapter} from 'socket.io-redis'; import {io} from 'socket.io-client'; @@ -6,6 +6,7 @@ import {FastifyInstance} from "fastify"; import IORedis from "ioredis"; import {App, TemplatedApp} from 'uWebSockets.js'; import {streamPastActions, streamPastDeltas} from "./helpers/functions"; +import {randomUUID} from "crypto"; export interface StreamDeltasRequest { code: string; @@ -41,6 +42,7 @@ export class SocketManager { private readonly server: FastifyInstance; private readonly uwsApp: TemplatedApp; private chainId: string; + private currentBlockNum: number; constructor(fastify: FastifyInstance, url, redisOptions) { this.server = fastify; @@ -83,7 +85,10 @@ export class SocketManager { if (typeof callback === 'function' && data) { try { if (data.start_from) { - await streamPastDeltas(this.server, socket, data); + const hStreamResult = await streamPastDeltas(this.server, socket, data); + if (hStreamResult === false) { + return; + } } this.emitToRelay(data, 'delta_request', socket, callback); } catch (e) { @@ -95,10 +100,45 @@ export class SocketManager { socket.on('action_stream_request', async (data: StreamActionsRequest, callback) => { if (typeof callback === 'function' && data) { try { + // generate random uuid + socket.data.reqUUID = randomUUID(); + + const lastHistoryBlock = await new Promise((resolve) => { + // start sending realtime data + this.emitToRelay(data, 'action_request', socket, (emissionResult) => { + callback(emissionResult); + resolve(emissionResult.currentBlockNum); + }); + }); + + // push history data if (data.start_from) { - await streamPastActions(this.server, socket, data); + data.read_until = lastHistoryBlock; + console.log('Performing primary scroll request...'); + let ltb; + const hStreamResult = await streamPastActions(this.server, socket, data); + if (hStreamResult.status === false) { + return; + } else { + ltb = hStreamResult.lastTransmittedBlock; + let attempts = 0; + await sleep(500); + while (lastHistoryBlock > ltb && attempts < 3) { + attempts++; + console.log(`Performing fill request from ${hStreamResult.lastTransmittedBlock}...`); + data.start_from = hStreamResult.lastTransmittedBlock + 1; + data.read_until = lastHistoryBlock; + const r = await streamPastActions(this.server, socket, data); + if (r.status === false) { + console.log(r); + return; + } else { + ltb = r.lastTransmittedBlock; + } + } + } } - this.emitToRelay(data, 'action_request', socket, callback); + } catch (e) { console.log(e); } @@ -158,8 +198,18 @@ export class SocketManager { this.emitToClient(traceData, 'action_trace'); }); + this.relay.on('block', (blockData) => { + try { + // const decodedBlock = JSON.parse(blockData.content.toString()); + // console.log(blockData.serverTime, blockData.blockNum, decodedBlock); + this.currentBlockNum = blockData.blockNum; + } catch (e) { + hLog(`Failed to decode incoming live block ${blockData.blockNum}: ${e.message}`); + } + }); this.addRelayForwarding('lib_update'); + this.addRelayForwarding('fork_event'); // // Relay LIB info to clients; @@ -190,6 +240,7 @@ export class SocketManager { if (this.io.sockets.sockets.has(traceData.client)) { this.io.sockets.sockets.get(traceData.client).emit('message', { type: type, + reqUUID: traceData.req, mode: 'live', message: traceData.message, }); @@ -199,10 +250,13 @@ export class SocketManager { emitToRelay(data, type, socket, callback) { if (this.relay.connected) { this.relay.emit('event', { + reqUUID: socket.data.reqUUID, type: type, client_socket: socket.id, request: data, }, (response) => { + response['reqUUID'] = socket.data.reqUUID; + response['currentBlockNum'] = this.currentBlockNum; callback(response); }); } else { diff --git a/chains/example.config.json b/chains/example.config.json index 035ebd1..2a6ef32 100644 --- a/chains/example.config.json +++ b/chains/example.config.json @@ -7,6 +7,8 @@ "server_addr": "127.0.0.1", "server_port": 7000, "stream_port": 1234, + "stream_scroll_limit": -1, + "stream_scroll_batch": 500, "server_name": "127.0.0.1:7000", "provider_name": "Example Provider", "provider_url": "https://example.com", diff --git a/interfaces/hyperionConfig.ts b/interfaces/hyperionConfig.ts index 032e025..2e27b7b 100644 --- a/interfaces/hyperionConfig.ts +++ b/interfaces/hyperionConfig.ts @@ -89,6 +89,8 @@ interface CachedRouteConfig { } interface ApiConfigs { + stream_scroll_batch?: number; + stream_scroll_limit?: number; enabled?: boolean; pm2_scaling?: number; node_max_old_space_size?: number; diff --git a/package.json b/package.json index 21ba032..c9b3fbe 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "hyperion-history", - "version": "3.3.9-5", + "version": "3.3.9-6", "description": "Scalable Full History API Solution for EOSIO based blockchains", "main": "launcher.js", "scripts": { diff --git a/workers/deserializer.ts b/workers/deserializer.ts index 0a6dc80..61c473a 100644 --- a/workers/deserializer.ts +++ b/workers/deserializer.ts @@ -403,7 +403,6 @@ export default class MainDSWorker extends HyperionWorker { } if (light_block.new_producers) { - process.send({ event: 'new_schedule', block_num: light_block.block_num, @@ -411,6 +410,16 @@ export default class MainDSWorker extends HyperionWorker { live: process.env.live_mode }); } + + // stream light block + if (this.allowStreaming) { + this.ch.publish('', this.chain + ':stream', Buffer.from(JSON.stringify(light_block)), { + headers: { + event: 'block', + blockNum: light_block.block_num + } + }); + } } // Process Delta Traces (must be done first to catch ABI updates) diff --git a/workers/ws-router.ts b/workers/ws-router.ts index c28eea6..c206c67 100644 --- a/workers/ws-router.ts +++ b/workers/ws-router.ts @@ -68,7 +68,17 @@ export default class WSRouter extends HyperionWorker { // push to plugin handlers this.mLoader.processStreamEvent(msg); + // process incoming live messages switch (msg.properties.headers.event) { + case 'block': { + // forward block events to all APIs + this.io.of('/').emit('block', { + serverTime: Date.now(), + blockNum: msg.properties.headers.blockNum, + content: msg.content + }); + break; + } case 'trace': { const actHeader = msg.properties.headers; const code = actHeader.account; @@ -146,7 +156,7 @@ export default class WSRouter extends HyperionWorker { } default: { - console.log('Unindentified message!'); + console.log('Unidentified message!'); console.log(msg); } } @@ -206,7 +216,10 @@ export default class WSRouter extends HyperionWorker { addActionRequest(data, id) { const req = data.request; if (typeof req.account !== 'string') { - return {status: 'FAIL', reason: 'invalid request'}; + return { + status: 'FAIL', + reason: 'invalid request' + }; } if (greylist.indexOf(req.contract) !== -1) { if (req.account === '' || req.account === req.contract) { @@ -219,6 +232,7 @@ export default class WSRouter extends HyperionWorker { const link = { type: 'action', relay: id, + reqUUID: data.reqUUID, client: data.client_socket, filters: req.filters, account: req.account, @@ -230,7 +244,10 @@ export default class WSRouter extends HyperionWorker { if (req.account !== '') { this.appendToL1Map(this.notifiedMap, req.account, link); } else { - return {status: 'FAIL', reason: 'invalid request'}; + return { + status: 'FAIL', + reason: 'invalid request' + }; } } this.addToClientIndex(data, id, [req.contract, req.action, req.account]); @@ -257,6 +274,7 @@ export default class WSRouter extends HyperionWorker { const link = { type: 'delta', relay: id, + reqUUID: data.reqUUID, client: data.client_socket, filters: data.request.filters, payer: data.request.payer, @@ -268,7 +286,10 @@ export default class WSRouter extends HyperionWorker { if (req.payer !== '' && req.payer !== '*') { this.appendToL1Map(this.payerMap, req.payer, link); } else { - return {status: 'FAIL', reason: 'invalid request'}; + return { + status: 'FAIL', + reason: 'invalid request' + }; } } this.addToClientIndex(data, id, [req.code, req.table, req.payer]); @@ -340,7 +361,7 @@ export default class WSRouter extends HyperionWorker { if (result.status === 'OK') { callback(result); } else { - callback(result.reason); + callback(result); } break; } @@ -349,7 +370,7 @@ export default class WSRouter extends HyperionWorker { if (result.status === 'OK') { callback(result); } else { - callback(result.reason); + callback(result); } break; } @@ -412,7 +433,7 @@ export default class WSRouter extends HyperionWorker { }); } if (allow) { - relay.emit('trace', {client: link.client, message: msg}); + relay.emit('trace', {client: link.client, req: link.reqUUID, message: msg}); this.totalRoutedMessages++; } } @@ -435,7 +456,7 @@ export default class WSRouter extends HyperionWorker { // }); // } if (allow) { - relay.emit('delta', {client: link.client, message: msg}); + relay.emit('delta', {client: link.client, req: link.reqUUID, message: msg}); this.totalRoutedMessages++; } }