From 65698c344d977063c46cf45cdc7862c8f1c954dc Mon Sep 17 00:00:00 2001 From: Igor Lins e Silva Date: Sat, 20 May 2023 08:56:46 +0000 Subject: [PATCH] on-demand delta filtering for streams --- api/helpers/functions.ts | 160 +++++++++++++++++++++++++----------- api/socketManager.ts | 42 ++++++++-- helpers/common_functions.ts | 27 ++++++ workers/ws-router.ts | 16 ++-- 4 files changed, 183 insertions(+), 62 deletions(-) diff --git a/api/helpers/functions.ts b/api/helpers/functions.ts index fa61242..9e45a99 100644 --- a/api/helpers/functions.ts +++ b/api/helpers/functions.ts @@ -2,7 +2,8 @@ import {createHash} from "crypto"; import * as _ from "lodash"; import {FastifyInstance, FastifyReply, FastifyRequest, FastifySchema, HTTPMethods} from "fastify"; import got from "got"; -import {checkFilter, hLog} from "../../helpers/common_functions"; +import {checkDeltaFilter, checkFilter, hLog} from "../../helpers/common_functions"; +import {Socket} from "socket.io"; const deltaQueryFields = ['code', 'table', 'scope', 'payer']; @@ -15,12 +16,30 @@ export async function streamPastDeltas(fastify: FastifyInstance, socket, data) { deltaQueryFields.forEach(f => { addTermMatch(data, search_body, f); }); + + const onDemandDeltaFilters = []; + if (data.filters.length > 0) { + data.filters.forEach(f => { + if (f.field && f.value) { + if (f.field.startsWith('@') && !f.field.startsWith('data')) { + const _q = {}; + _q[f.field] = f.value; + search_body.query.bool.must.push({'term': _q}); + } else { + onDemandDeltaFilters.push(f); + } + } + }); + } + const responseQueue = []; let counter = 0; let total = 0; + let totalFiltered = 0; let longScroll = false; + // console.log(JSON.stringify(search_body, null, 2)); const init_response = await fastify.elastic.search({ index: fastify.manager.chain + '-delta-*', @@ -29,66 +48,109 @@ export async function streamPastDeltas(fastify: FastifyInstance, socket, data) { 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) { + const errorMsg = `Requested ${init_response.body.hits.total.value} deltas, limit is ${scrollLimit}.`; socket.emit('message', { + reqUUID: socket.data.reqUUID, type: 'delta_trace', mode: 'history', messages: [], - error: `Requested ${init_response.body.hits.total.value} deltas, limit is ${scrollLimit}` + error: errorMsg }); - return false; + 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!`); + hLog(`Attention! Long scroll (deltas) is running!`); } + // emit first block + if (init_response.body.hits.hits.length > 0) { + emitTraceInit(socket, init_response.body.hits.hits[0]._source.block_num, init_response.body.hits.total.value); + } + + responseQueue.push(init_response); + + let lastTransmittedBlock = 0; + let pendingScrollId = ''; while (responseQueue.length) { + let filterCount = 0; const {body} = responseQueue.shift(); + pendingScrollId = body['_scroll_id']; + const enqueuedMessages = []; counter += body['hits']['hits'].length; + + for (const doc of body['hits']['hits']) { + let allow = false; + if (onDemandDeltaFilters.length > 0) { + allow = onDemandDeltaFilters.every(filter => { + return checkDeltaFilter(filter, doc._source); + }); + } else { + allow = true; + } + if (allow) { + enqueuedMessages.push(doc._source); + } else { + filterCount++; + } + // set last block + if (doc._source.block_num > lastTransmittedBlock) { + lastTransmittedBlock = doc._source.block_num; + } + } + + totalFiltered += filterCount; + if (socket.connected) { - socket.emit('message', { - type: 'delta_trace', - mode: 'history', - messages: body['hits']['hits'].map(doc => doc._source), - }); + if (enqueuedMessages.length > 0) { + socket.emit('message', { + reqUUID: socket.data.reqUUID, + type: 'delta_trace', + mode: 'history', + messages: enqueuedMessages, + filtered: filterCount + }); + } } else { hLog('LOST CLIENT'); break; } if (longScroll) { - hLog(`Progress: ${counter}/${total}`); + hLog(`Progress: ${counter + totalFiltered}/${total}`); } if (body['hits'].total.value === counter) { - hLog(`${counter} past deltas streamed to ${socket.id}`); + hLog(`${counter} past deltas streamed to ${socket.id} (${totalFiltered} filtered)`); break; } const next_response = await fastify.elastic.scroll({ - body: { - scroll_id: body['_scroll_id'], - scroll: '30s' - } + 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 emitTraceInit(socket: Socket, firstBlock: number, totalResults: number) { + socket.emit('message', { + reqUUID: socket.data.reqUUID, + type: 'trace_init', + mode: 'history', + first_block: firstBlock, + results: totalResults + }); } export async function streamPastActions(fastify: FastifyInstance, socket, data) { - - const search_body = { - query: {bool: {must: []}}, - sort: {global_sequence: 'asc'}, - }; - + const search_body = {query: {bool: {must: []}}, sort: {global_sequence: 'asc'}}; await addBlockRangeOpts(data, search_body, fastify); - if (data.account !== '') { search_body.query.bool.must.push({ bool: { @@ -126,10 +188,9 @@ export async function streamPastActions(fastify: FastifyInstance, socket, data) const responseQueue = []; let counter = 0; let total = 0; + let totalFiltered = 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', @@ -151,18 +212,12 @@ export async function streamPastActions(fastify: FastifyInstance, socket, data) if (init_response.body.hits.total.value > 10000) { total = init_response.body.hits.total.value; longScroll = true; - hLog(`Attention! Long scroll is running!`); + hLog(`Attention! Long scroll (actions) 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 - }); + emitTraceInit(socket, init_response.body.hits.hits[0]._source.block_num, init_response.body.hits.total.value); } responseQueue.push(init_response); @@ -170,10 +225,12 @@ export async function streamPastActions(fastify: FastifyInstance, socket, data) let lastTransmittedBlock = 0; let pendingScrollId = ''; while (responseQueue.length) { + let filterCount = 0; const {body} = responseQueue.shift(); pendingScrollId = body['_scroll_id']; const enqueuedMessages = []; counter += body['hits']['hits'].length; + for (const doc of body['hits']['hits']) { let allow = false; if (onDemandFilters.length > 0) { @@ -185,30 +242,41 @@ 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; - } + } else { + filterCount++; + } + // set last block + if (doc._source.block_num > lastTransmittedBlock) { + lastTransmittedBlock = doc._source.block_num; } } + + totalFiltered += filterCount; + if (socket.connected) { - socket.emit('message', { - reqUUID: socket.data.reqUUID, - type: 'action_trace', - mode: 'history', messages: enqueuedMessages - }); + if (enqueuedMessages.length > 0) { + socket.emit('message', { + reqUUID: socket.data.reqUUID, + type: 'action_trace', + mode: 'history', + messages: enqueuedMessages, + filtered: filterCount + }); + } } else { hLog('LOST CLIENT'); break; } if (longScroll) { - hLog(`Progress: ${counter}/${total}`); + hLog(`Progress: ${counter + totalFiltered}/${total}`); } if (body['hits'].total.value === counter) { - hLog(`${counter} past actions streamed to ${socket.id}`); + hLog(`${counter} past actions streamed to ${socket.id} (${totalFiltered} filtered)`); break; } + const next_response = await fastify.elastic.scroll({ body: {scroll_id: body['_scroll_id'], scroll: '30s'} }); @@ -222,7 +290,7 @@ export async function streamPastActions(fastify: FastifyInstance, socket, data) } export function addTermMatch(data, search_body, field) { - if (data[field] !== '*' && data[field] !== '') { + if (data[field] && data[field] !== '*' && data[field] !== '') { const termQuery = {}; termQuery[field] = data[field]; search_body.query.bool.must.push({'term': termQuery}); diff --git a/api/socketManager.ts b/api/socketManager.ts index 8b2e79b..785bab7 100644 --- a/api/socketManager.ts +++ b/api/socketManager.ts @@ -41,7 +41,7 @@ export class SocketManager { private readonly url; private readonly server: FastifyInstance; private readonly uwsApp: TemplatedApp; - private chainId: string; + private readonly chainId: string; private currentBlockNum: number; constructor(fastify: FastifyInstance, url, redisOptions) { @@ -84,13 +84,42 @@ export class SocketManager { socket.on('delta_stream_request', async (data: StreamDeltasRequest, 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, 'delta_request', socket, (emissionResult) => { + callback(emissionResult); + resolve(emissionResult.currentBlockNum); + }); + }); + // push history data if (data.start_from) { + data.read_until = lastHistoryBlock; + console.log('Performing primary scroll request...'); + let ltb = 0; const hStreamResult = await streamPastDeltas(this.server, socket, data); - if (hStreamResult === false) { + if (hStreamResult.status === false) { return; + } else { + ltb = hStreamResult.lastTransmittedBlock; + let attempts = 0; + await sleep(500); + while (ltb > 0 && lastHistoryBlock > ltb && attempts < 3) { + attempts++; + console.log(`Performing fill request from ${ltb}...`); + data.start_from = hStreamResult.lastTransmittedBlock + 1; + data.read_until = lastHistoryBlock; + const r = await streamPastDeltas(this.server, socket, data); + if (r.status === false) { + console.log(r); + return; + } else { + ltb = r.lastTransmittedBlock; + } + } } } - this.emitToRelay(data, 'delta_request', socket, callback); } catch (e) { console.log(e); } @@ -102,7 +131,6 @@ export class SocketManager { 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) => { @@ -110,12 +138,11 @@ export class SocketManager { resolve(emissionResult.currentBlockNum); }); }); - // push history data if (data.start_from) { data.read_until = lastHistoryBlock; console.log('Performing primary scroll request...'); - let ltb; + let ltb = 0; const hStreamResult = await streamPastActions(this.server, socket, data); if (hStreamResult.status === false) { return; @@ -123,7 +150,7 @@ export class SocketManager { ltb = hStreamResult.lastTransmittedBlock; let attempts = 0; await sleep(500); - while (lastHistoryBlock > ltb && attempts < 3) { + while (ltb > 0 && lastHistoryBlock > ltb && attempts < 3) { attempts++; console.log(`Performing fill request from ${hStreamResult.lastTransmittedBlock}...`); data.start_from = hStreamResult.lastTransmittedBlock + 1; @@ -138,7 +165,6 @@ export class SocketManager { } } } - } catch (e) { console.log(e); } diff --git a/helpers/common_functions.ts b/helpers/common_functions.ts index 9644323..54dd745 100644 --- a/helpers/common_functions.ts +++ b/helpers/common_functions.ts @@ -191,6 +191,33 @@ function getNested(path_array, jsonObj) { } } +export function checkDeltaFilter(filter, _source) { + if (filter.field && filter.value) { + let fieldValue = getNested(filter.field.split("."), _source); + if (!fieldValue) { + const fArray = filter.field.split("."); + if (fArray[0].startsWith('@')) { + const tableName = fArray[0].replace('@', ''); + if (_source.table === tableName) { + fArray[0] = 'data'; + fieldValue = getNested(fArray, _source); + } + } + } + if (fieldValue) { + if (Array.isArray(fieldValue)) { + return fieldValue.indexOf(filter.value) !== -1; + } else { + return fieldValue === filter.value; + } + } else { + return !filter.value; + } + } else { + return false; + } +} + export function checkFilter(filter, _source) { if (filter.field && filter.value) { let fieldValue = getNested(filter.field.split("."), _source); diff --git a/workers/ws-router.ts b/workers/ws-router.ts index c206c67..e297e12 100644 --- a/workers/ws-router.ts +++ b/workers/ws-router.ts @@ -1,7 +1,7 @@ import {HyperionWorker} from "./hyperionWorker"; import {Server, Socket} from "socket.io"; -import {checkFilter, hLog} from "../helpers/common_functions"; +import {checkDeltaFilter, checkFilter, hLog} from "../helpers/common_functions"; import {createServer} from "http"; const greylist = ['eosio.token']; @@ -448,13 +448,13 @@ export default class WSRouter extends HyperionWorker { } else { allow = true; } - // if (link.filters?.length > 0) { - // // check filters - // const _parsedMsg = JSON.parse(msg); - // allow = link.filters.every(filter => { - // return checkDeltaFilter(filter, _parsedMsg); - // }); - // } + if (link.filters?.length > 0) { + // check filters + const _parsedMsg = JSON.parse(msg); + allow = link.filters.every(filter => { + return checkDeltaFilter(filter, _parsedMsg); + }); + } if (allow) { relay.emit('delta', {client: link.client, req: link.reqUUID, message: msg}); this.totalRoutedMessages++;