Display state of Grandpa

packages/backend/src/Aggregator.ts

@@ -46,6 +46,24 @@ export default class Aggregator {
         feed.sendMessage(Feed.unsubscribedFrom(label));
       }
     });
+
+    feed.events.on('subscribe-consensus-info', (label: Types.ChainLabel) => {
+      const chain = this.chains.get(label);
+
+      if (chain) {
+        feed.sendMessage(Feed.subscribedTo(label));
+        chain.addFeed(feed);
+      }
+    });
+
+    feed.events.on('unsubscribe-consensus-info', (label: Types.ChainLabel) => {
+      const chain = this.chains.get(label);
+
+      if (chain) {
+        chain.removeFeed(feed);
+        feed.sendMessage(Feed.unsubscribedFrom(label));
+      }
+    });
   }
 
   public getExistingChain(label: Types.ChainLabel) : Maybe<Chain> {

packages/backend/src/Chain.ts

@@ -3,7 +3,7 @@ import Node from './Node';
 import Feed from './Feed';
 import FeedSet from './FeedSet';
 import Block from './Block';
-import { Maybe, Types, FeedMessage, NumStats } from '@dotstats/common';
+import { Maybe, Types, NumStats } from '@dotstats/common';
 
 const BLOCK_TIME_HISTORY = 10;
 
@@ -21,6 +21,8 @@ export default class Chain {
   private blockTimes = new NumStats<Types.Milliseconds>(BLOCK_TIME_HISTORY);
   private averageBlockTime: Maybe<Types.Milliseconds> = null;
 
+  public lastBroadcastedAuthoritySetInfo: Maybe<Types.AuthoritySetInfo> = null;
+
   constructor(label: Types.ChainLabel) {
     this.label = label;
   }
@@ -39,6 +41,33 @@ export default class Chain {
 
     node.events.on('block', () => this.updateBlock(node));
     node.events.on('finalized', () => this.updateFinalized(node));
+
+    node.events.on('afg-finalized', (finalizedNumber, finalizedHash) => this.feeds.each(
+      f => f.sendConsensusMessage(Feed.afgFinalized(node, finalizedNumber, finalizedHash))
+    ));
+    node.events.on('afg-received-prevote', (finalizedNumber, finalizedHash, voter) => this.feeds.each(
+      f => f.sendConsensusMessage(Feed.afgReceivedPrevote(node, finalizedNumber, finalizedHash, voter))
+    ));
+    node.events.on('afg-received-precommit', (finalizedNumber, finalizedHash, voter) => this.feeds.each(
+      f => f.sendConsensusMessage(Feed.afgReceivedPrecommit(node, finalizedNumber, finalizedHash, voter))
+    ));
+    node.events.on('authority-set-changed', (authorities, authoritySetId, blockNumber, blockHash) => {
+      let newSet;
+      if (this.lastBroadcastedAuthoritySetInfo == null) {
+        newSet = true;
+      } else {
+        const [lastBroadcastedAuthoritySetId] = this.lastBroadcastedAuthoritySetInfo;
+        newSet = authoritySetId !== lastBroadcastedAuthoritySetId;
+      }
+
+      if (node.isAuthority() && newSet) {
+        const addr = node.address != null ? node.address : "" as Types.Address;
+        const set = [authoritySetId, authorities, addr, blockNumber, blockHash] as Types.AuthoritySetInfo;
+        this.feeds.broadcast(Feed.afgAuthoritySet(set));
+        this.lastBroadcastedAuthoritySetInfo = set;
+      }
+    });
+
     node.events.on('stats', () => this.feeds.broadcast(Feed.stats(node)));
     node.events.on('hardware', () => this.feeds.broadcast(Feed.hardware(node)));
     node.events.on('location', (location) => this.feeds.broadcast(Feed.locatedNode(node, location)));
@@ -70,6 +99,10 @@ export default class Chain {
     feed.sendMessage(Feed.bestBlock(this.height, this.blockTimestamp, this.averageBlockTime));
     feed.sendMessage(Feed.bestFinalizedBlock(this.finalized));
 
+    if (this.lastBroadcastedAuthoritySetInfo != null) {
+      feed.sendMessage(Feed.afgAuthoritySet(this.lastBroadcastedAuthoritySetInfo));
+    }
+
     for (const node of this.nodes.values()) {
       feed.sendMessage(Feed.addedNode(node));
       feed.sendMessage(Feed.finalized(node));

packages/backend/src/Feed.ts

@@ -18,6 +18,7 @@ export default class Feed {
   private socket: WebSocket;
   private messages: Array<FeedMessage.Message> = [];
   private waitingForPong = false;
+  private sendFinality = false;
 
   constructor(socket: WebSocket) {
     this.id = nextId();
@@ -92,6 +93,49 @@ export default class Feed {
     };
   }
 
+  public static afgFinalized(node: Node, finalizedNumber: Types.BlockNumber, finalizedHash: Types.BlockHash): FeedMessage.Message {
+    const addr = node.address != null ? node.address : "" as Types.Address;
+    return {
+      action: Actions.AfgFinalized,
+      payload: [addr, finalizedNumber, finalizedHash]
+    };
+  }
+
+  public static afgReceivedPrevote(
+    node: Node,
+    targetNumber: Types.BlockNumber,
+    targetHash: Types.BlockHash,
+    voter: Types.Address
+  ): FeedMessage.Message {
+    const addr = node.address != null ? node.address : "" as Types.Address;
+    return {
+      action: Actions.AfgReceivedPrevote,
+      payload: [addr, targetNumber, targetHash, voter]
+    };
+  }
+
+  public static afgReceivedPrecommit(
+    node: Node,
+    targetNumber: Types.BlockNumber,
+    targetHash: Types.BlockHash,
+    voter: Types.Address
+  ): FeedMessage.Message {
+    const addr = node.address != null ? node.address : "" as Types.Address;
+    return {
+      action: Actions.AfgReceivedPrecommit,
+      payload: [addr, targetNumber, targetHash, voter]
+    };
+  }
+
+  public static afgAuthoritySet(
+    authoritySetInfo: Types.AuthoritySetInfo,
+  ): FeedMessage.Message {
+    return {
+      action: Actions.AfgAuthoritySet,
+      payload: authoritySetInfo,
+    };
+  }
+
   public static hardware(node: Node): FeedMessage.Message {
     return {
       action: Actions.NodeHardware,
@@ -155,6 +199,14 @@ export default class Feed {
     }
   }
 
+  public sendConsensusMessage(message: FeedMessage.Message) {
+    if (!this.sendFinality) {
+      return;
+    }
+
+    this.sendMessage(message);
+  }
+
   public ping() {
     if (this.waitingForPong) {
       this.disconnect();
@@ -188,6 +240,14 @@ export default class Feed {
         this.events.emit('subscribe', payload as Types.ChainLabel);
         break;
 
+      case 'send-finality':
+        this.sendFinality = true;
+        break;
+
+      case 'no-more-finality':
+        this.sendFinality = false;
+        break;
+
       case 'ping':
         this.sendMessage(Feed.pong(payload));
         break;

packages/backend/src/Node.ts

@@ -2,7 +2,18 @@ import * as WebSocket from 'ws';
 import * as EventEmitter from 'events';
 
 import { noop, timestamp, idGenerator, Maybe, Types, NumStats } from '@dotstats/common';
-import { parseMessage, getBestBlock, Message, BestBlock, SystemInterval } from './message';
+import { BlockHash, BlockNumber, ConsensusView } from "@dotstats/common/build/types";
+import {
+  parseMessage,
+  getBestBlock,
+  Message,
+  BestBlock,
+  SystemInterval,
+  AfgFinalized,
+  AfgReceivedPrecommit,
+  AfgReceivedPrevote,
+  AfgAuthoritySet,
+} from './message';
 import { locate, Location } from './location';
 import MeanList from './MeanList';
 import Block from './Block';
@@ -57,6 +68,9 @@ export default class Node {
   private pingStart = 0 as Types.Timestamp;
   private throttle = false;
 
+  private authorities: Types.Authorities = [] as Types.Authorities;
+  private authoritySetId: Types.AuthoritySetId = 0 as Types.AuthoritySetId;
+
   constructor(
     ip: string,
     socket: WebSocket,
@@ -182,8 +196,9 @@ export default class Node {
 
   public nodeDetails(): Types.NodeDetails {
     const authority = this.authority ? this.address : null;
+    const addr = this.address ? this.address : '' as Types.Address;
 
-    return [this.name, this.implementation, this.version, authority, this.networkId];
+    return [this.name, addr, this.implementation, this.version, authority, this.networkId];
   }
 
   public nodeStats(): Types.NodeStats {
@@ -236,6 +251,19 @@ export default class Node {
     if (message.msg === 'system.interval') {
       this.onSystemInterval(message);
     }
+
+    if (message.msg === 'afg.finalized') {
+      this.onAfgFinalized(message);
+    }
+    if (message.msg === 'afg.received_precommit') {
+      this.onAfgReceivedPrecommit(message);
+    }
+    if (message.msg === 'afg.received_prevote') {
+      this.onAfgReceivedPrevote(message);
+    }
+    if (message.msg === 'afg.authority_set') {
+      this.onAfgAuthoritySet(message);
+    }
   }
 
   private onSystemInterval(message: SystemInterval) {
@@ -283,6 +311,57 @@ export default class Node {
     }
   }
 
+  public isAuthority(): boolean {
+    return this.authority;
+  }
+
+  private onAfgReceivedPrecommit(message: AfgReceivedPrecommit) {
+    const {
+      target_number: targetNumber,
+      target_hash: targetHash,
+    } = message;
+    const voter = this.extractVoter(message.voter);
+    this.events.emit('afg-received-precommit', targetNumber, targetHash, voter);
+  }
+
+  private onAfgReceivedPrevote(message: AfgReceivedPrevote) {
+    const {
+      target_number: targetNumber,
+      target_hash: targetHash,
+    } = message;
+    const voter = this.extractVoter(message.voter);
+    this.events.emit('afg-received-prevote', targetNumber, targetHash, voter);
+  }
+
+  private onAfgAuthoritySet(message: AfgAuthoritySet) {
+    const {
+      authority_set_id: authoritySetId,
+      hash,
+      number,
+    } = message;
+
+    // we manually parse the authorities message, because the array was formatted as a
+    // string by substrate before sending it.
+    const authorities = JSON.parse(String(message.authorities)) as Types.Authorities;
+
+    if (JSON.stringify(this.authorities) !== String(message.authorities) ||
+        this.authoritySetId !== authoritySetId) {
+      this.events.emit('authority-set-changed', authorities, authoritySetId, number, hash);
+    }
+  }
+
+  private onAfgFinalized(message: AfgFinalized) {
+    const {
+      finalized_number: finalizedNumber,
+      finalized_hash: finalizedHash,
+    } = message;
+    this.events.emit('afg-finalized', finalizedNumber, finalizedHash);
+  }
+
+  private extractVoter(message_voter: String): Types.Address {
+    return String(message_voter.replace(/"/g, '')) as Types.Address;
+  }
+
   private updateLatency(now: Types.Timestamp) {
     // if (this.pingStart) {
     //   console.error(`${this.name} timed out on ping message.`);

packages/backend/src/message.ts

@@ -39,6 +39,41 @@ export interface BestBlock {
   ts: Date;
 }
 
+export interface AfgFinalized {
+  ts: Date;
+  finalized_number: Types.BlockNumber;
+  finalized_hash: Types.BlockHash;
+  msg: 'afg.finalized';
+}
+
+export interface AfgReceived {
+  ts: Date;
+  target_number: Maybe<Types.BlockNumber>;
+  target_hash: Maybe<Types.BlockHash>;
+  voter: Types.Address;
+}
+
+export interface AfgReceivedPrecommit extends AfgReceived {
+  msg: 'afg.received_precommit';
+}
+
+export interface AfgReceivedPrevote extends AfgReceived {
+  msg: 'afg.received_prevote';
+}
+
+export interface AfgReceivedCommit extends AfgReceived {
+  msg: 'afg.received_commit';
+}
+
+export interface AfgAuthoritySet {
+  msg: 'afg.authority_set';
+  ts: Date;
+  authorities: Types.Authorities;
+  authority_set_id: Types.AuthoritySetId;
+  number: Types.BlockNumber;
+  hash: Types.BlockHash;
+}
+
 export interface SystemConnected {
   msg: 'system.connected';
   name: Types.NodeName;
@@ -79,6 +114,11 @@ export type Message = MessageBase & (
   | SystemInterval
   | NodeStart
   | BlockImport
+  | AfgFinalized
+  | AfgReceivedPrecommit
+  | AfgReceivedPrevote
+  | AfgReceivedCommit
+  | AfgAuthoritySet
 );