C9S/explorer-api
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

- multiple streaming improvements

Browse Source 
- 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
This commit is contained in:
Igor Lins e Silva
2023-05-20 07:20:46 +00:00
parent a2a040ab55
commit b289036b0c
7 changed files with 194 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/helpers/functions.ts
+92 -14
View File
@@ -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) {
api/socketManager.ts
+58 -4
View File
@@ -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<number>((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 {
chains/example.config.json
+2
View File
@@ -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",
interfaces/hyperionConfig.ts
+2
View File
@@ -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;
package.json
+1 -1
View File
@@ -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": {
workers/deserializer.ts
+10 -1
View File
@@ -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)
workers/ws-router.ts
+29 -8
View File
@@ -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++;
}
}