fix lib event forwarding on shared api hosts

api/server.ts

@@ -37,7 +37,7 @@ class HyperionApiServer {
         const cm = new ConfigurationModule();
         this.conf = cm.config;
 
-        if(this.conf.settings.use_global_agent) {
+        if (this.conf.settings.use_global_agent) {
             bootstrap();
         }
 
@@ -150,20 +150,21 @@ class HyperionApiServer {
     }
 
     activateStreaming() {
-        console.log('Importing stream module');
+        console.log('Importing stream module...');
         import('./socketManager').then((mod) => {
             const connOpts = this.manager.conn.chains[this.chain];
-
             let _port = 57200;
             if (connOpts.WS_ROUTER_PORT) {
                 _port = connOpts.WS_ROUTER_PORT;
             }
-
             let _host = "127.0.0.1";
             if (connOpts.WS_ROUTER_HOST) {
                 _host = connOpts.WS_ROUTER_HOST;
             }
-
+            if (_host === "0.0.0.0") {
+                hLog(`[ERROR] WS Router Host is set to 0.0.0.0, please use a fixed IP address instead. Can't start streaming.`);
+                return;
+            }
             this.socketManager = new mod.SocketManager(
                 this.fastify,
                 `http://${_host}:${_port}`,

api/socketManager.ts

@@ -40,17 +40,13 @@ export class SocketManager {
     private readonly url;
     private readonly server: FastifyInstance;
     private readonly uwsApp: TemplatedApp;
+    private chainId: string;
 
     constructor(fastify: FastifyInstance, url, redisOptions) {
         this.server = fastify;
         this.url = url;
         this.uwsApp = App({});
 
-        // this.io = new Server(fastify.server, {
-        // 	allowEIO3: true,
-        // 	transports: ['websocket', 'polling'],
-        // });
-
         // WS Server for public access
         this.io = new Server({
             transports: ['websocket'],
@@ -59,9 +55,11 @@ export class SocketManager {
 
         this.io.attachApp(this.uwsApp);
 
+        this.chainId = this.server.manager.conn.chains[this.server.manager.chain].chain_id
+        hLog(`[SocketManager] chain_id: ${this.chainId}`);
         const pubClient = new IORedis(redisOptions);
         const subClient = pubClient.duplicate();
-        this.io.adapter(createAdapter({pubClient, subClient}));
+        this.io.adapter(createAdapter({pubClient, subClient, key: this.chainId}));
 
         this.io.on('connection', (socket: Socket) => {
 
@@ -160,18 +158,30 @@ export class SocketManager {
             this.emitToClient(traceData, 'action_trace');
         });
 
-        // Relay LIB info to clients;
-        this.relay.on('lib_update', (data) => {
-            if (this.server.manager.conn.chains[this.server.manager.chain].chain_id === data.chain_id) {
-                this.io.emit('lib_update', data);
-            }
-        });
 
-        // Relay LIB info to clients;
-        this.relay.on('fork_event', (data) => {
-            hLog(data);
-            if (this.server.manager.conn.chains[this.server.manager.chain].chain_id === data.chain_id) {
-                this.io.emit('fork_event', data);
+        this.addRelayForwarding('lib_update');
+        this.addRelayForwarding('fork_event');
+
+        // // Relay LIB info to clients;
+        // this.relay.on('lib_update', (data) => {
+        //     if (this.server.manager.conn.chains[this.server.manager.chain].chain_id === data.chain_id) {
+        //         this.io.emit('lib_update', data);
+        //     }
+        // });
+        //
+        // // Relay fork info to clients;
+        // this.relay.on('fork_event', (data) => {
+        //     if (this.server.manager.conn.chains[this.server.manager.chain].chain_id === data.chain_id) {
+        //         this.io.emit('fork_event', data);
+        //     }
+        // });
+    }
+
+    // Relay events to clients
+    addRelayForwarding(event: string) {
+        this.relay.on(event, (data: any) => {
+            if (data.chain_id && this.chainId === data.chain_id) {
+                this.io.emit(event, data);
             }
         });
     }

modules/master.ts

@@ -420,6 +420,7 @@ export class HyperionMaster {
                     if (this.conf.hub && this.conf.hub.inform_url) {
                         this.hub.emit('hyp_ev', {e: 'lib', d: msg.data});
                     }
+                    // forward LIB to streaming router
                     if (this.conf.features.streaming.enable) {
                         debugLog(`Live Reader reported LIB update: ${msg.data.block_num} | ${msg.data.block_id}`);
                         this.wsRouterWorker.send(msg);

workers/state-reader.ts

@@ -431,6 +431,7 @@ 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;

workers/ws-router.ts

@@ -34,21 +34,21 @@ export default class WSRouter extends HyperionWorker {
         this.ch.consume(this.q, this.onConsume.bind(this));
     }
 
+    appendIdAndEmit(event, data) {
+        this.io.emit(event, {
+            chain_id: this.manager.conn.chains[this.chain]?.chain_id,
+            ...data
+        });
+    };
 
     onIpcMessage(msg: any): void {
         switch (msg.event) {
             case 'lib_update': {
-                this.io.emit('lib_update', {
-                    chain_id: this.manager.conn.chains[this.chain]?.chain_id,
-                    ...msg.data
-                });
+                this.appendIdAndEmit('lib_update', msg.data);
                 break;
             }
             case 'fork_event': {
-                this.io.emit('fork_event', {
-                    chain_id: this.manager.conn.chains[this.chain]?.chain_id,
-                    ...msg.data
-                });
+                this.appendIdAndEmit('fork_event', msg.data);
                 break;
             }
         }