on-demand delta filtering for streams

2023-05-20 08:56:46 +00:00
parent b5f17b0133
commit 65698c344d
4 changed files with 183 additions and 62 deletions
@@ -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});
@@ -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<number>((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<number>((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);
                    }
@@ -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);
@@ -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++;