feat: add send epoch, time, elapsed epoch and elapsed time for each message in mpool to UI

gql/resolver_mpool.go

@@ -8,6 +8,7 @@ import (
 	"reflect"
 
 	gqltypes "github.com/filecoin-project/boost/gql/types"
+	"github.com/filecoin-project/boost/lib/mpoolmonitor"
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/lotus/chain/consensus"
@@ -16,6 +17,7 @@ import (
 )
 
 type msg struct {
+	SentEpoch  gqltypes.Uint64
 	To         string
 	From       string
 	Nonce      gqltypes.Uint64
@@ -31,12 +33,13 @@ type msg struct {
 // query: mpool(local): [Message]
 func (r *resolver) Mpool(ctx context.Context, args struct{ Local bool }) ([]*msg, error) {
 	var ret []*msg
-	var msgs []*types.SignedMessage
+	var msgs []*mpoolmonitor.TimeStampedMsg
 
 	ts, err := r.fullNode.ChainHead(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get chain head: %w", err)
 	}
+
 	baseFee := ts.Blocks()[0].ParentBaseFee
 
 	if args.Local {
@@ -54,17 +57,17 @@ func (r *resolver) Mpool(ctx context.Context, args struct{ Local bool }) ([]*msg
 	// Convert params to human-readable and get method name
 	for _, m := range msgs {
 		var params string
-		methodName := m.Message.Method.String()
-		toact, err := r.fullNode.StateGetActor(ctx, m.Message.To, types.EmptyTSK)
+		methodName := m.SignedMessage.Message.Method.String()
+		toact, err := r.fullNode.StateGetActor(ctx, m.SignedMessage.Message.To, types.EmptyTSK)
 		if err == nil {
-			method, ok := consensus.NewActorRegistry().Methods[toact.Code][m.Message.Method]
+			method, ok := consensus.NewActorRegistry().Methods[toact.Code][m.SignedMessage.Message.Method]
 			if ok {
 				methodName = method.Name
 
-				params = string(m.Message.Params)
+				params = string(m.SignedMessage.Message.Params)
 				p, ok := reflect.New(method.Params.Elem()).Interface().(cbg.CBORUnmarshaler)
 				if ok {
-					if err := p.UnmarshalCBOR(bytes.NewReader(m.Message.Params)); err == nil {
+					if err := p.UnmarshalCBOR(bytes.NewReader(m.SignedMessage.Message.Params)); err == nil {
 						b, err := json.MarshalIndent(p, "", "  ")
 						if err == nil {
 							params = string(b)
@@ -75,13 +78,14 @@ func (r *resolver) Mpool(ctx context.Context, args struct{ Local bool }) ([]*msg
 		}
 
 		ret = append(ret, &msg{
-			To:         m.Message.To.String(),
-			From:       m.Message.From.String(),
-			Nonce:      gqltypes.Uint64(m.Message.Nonce),
-			Value:      gqltypes.BigInt{Int: m.Message.Value},
-			GasFeeCap:  gqltypes.BigInt{Int: m.Message.GasFeeCap},
-			GasLimit:   gqltypes.Uint64(uint64(m.Message.GasLimit)),
-			GasPremium: gqltypes.BigInt{Int: m.Message.GasPremium},
+			SentEpoch:  gqltypes.Uint64(m.Added),
+			To:         m.SignedMessage.Message.To.String(),
+			From:       m.SignedMessage.Message.From.String(),
+			Nonce:      gqltypes.Uint64(m.SignedMessage.Message.Nonce),
+			Value:      gqltypes.BigInt{Int: m.SignedMessage.Message.Value},
+			GasFeeCap:  gqltypes.BigInt{Int: m.SignedMessage.Message.GasFeeCap},
+			GasLimit:   gqltypes.Uint64(uint64(m.SignedMessage.Message.GasLimit)),
+			GasPremium: gqltypes.BigInt{Int: m.SignedMessage.Message.GasPremium},
 			Method:     methodName,
 			Params:     params,
 			BaseFee:    gqltypes.BigInt{Int: baseFee},

gql/schema.graphql

@@ -363,6 +363,7 @@ type TransferStats {
 }
 
 type MpoolMessage {
+  SentEpoch: Uint64!
   From: String!
   To: String!
   Nonce: Uint64!

lib/mpoolmonitor/mpoolmonitor.go

@@ -16,10 +16,11 @@ import (
 
 var log = logging.Logger("mpoolmonitor")
 
-// timeStampedMsg wraps the pending msg with a chainEpoch
-type timeStampedMsg struct {
-	m     types.SignedMessage
-	added abi.ChainEpoch // Epoch when message was first noticed in mpool
+// TimeStampedMsg wraps the pending msg with a chainEpoch
+type TimeStampedMsg struct {
+	SignedMessage types.SignedMessage
+	Added         abi.ChainEpoch // Epoch when message was first noticed in mpool
+	AddedTime     time.Time
 }
 
 type MpoolMonitor struct {
@@ -28,14 +29,14 @@ type MpoolMonitor struct {
 	fullNode         v1api.FullNode
 	lk               sync.Mutex
 	mpoolAlertEpochs abi.ChainEpoch
-	msgs             map[cid.Cid]*timeStampedMsg
+	msgs             map[cid.Cid]*TimeStampedMsg
 }
 
 func NewMonitor(fullNode v1api.FullNode, mpoolAlertEpochs int64) *MpoolMonitor {
 	return &MpoolMonitor{
 		fullNode:         fullNode,
 		mpoolAlertEpochs: abi.ChainEpoch(mpoolAlertEpochs),
-		msgs:             make(map[cid.Cid]*timeStampedMsg),
+		msgs:             make(map[cid.Cid]*TimeStampedMsg),
 	}
 }
 
@@ -71,10 +72,11 @@ func (mm *MpoolMonitor) startMonitoring(ctx context.Context) {
 }
 
 // update gets the current pending messages from mpool. It updated the local
-// copy of a message(not timeStampedMsg.added) if CID is already present in MpoolMonitor.msgs
+// copy of a message(not TimeStampedMsg.added) if CID is already present in MpoolMonitor.msgs
 // Otherwise, it inserts a new key value pair for the message. It removed any msgs not found in the
 // latest output of MpoolPending
 func (mm *MpoolMonitor) update(ctx context.Context) error {
+	t := time.Now()
 	ts, err := mm.fullNode.ChainHead(ctx)
 	if err != nil {
 		return fmt.Errorf("failed to get chain head: %w", err)
@@ -84,12 +86,13 @@ func (mm *MpoolMonitor) update(ctx context.Context) error {
 		return fmt.Errorf("getting mpool messages: %w", err)
 	}
 
-	newMsgs := make(map[cid.Cid]*timeStampedMsg)
+	newMsgs := make(map[cid.Cid]*TimeStampedMsg)
 
 	for _, pm := range msgs {
-		newMsgs[pm.Cid()] = &timeStampedMsg{
-			m:     *pm,
-			added: ts.Height(),
+		newMsgs[pm.Cid()] = &TimeStampedMsg{
+			SignedMessage: *pm,
+			Added:         ts.Height(),
+			AddedTime:     t,
 		}
 	}
 
@@ -100,7 +103,7 @@ func (mm *MpoolMonitor) update(ctx context.Context) error {
 	for k, v := range newMsgs {
 		_, ok := mm.msgs[k]
 		if ok {
-			mm.msgs[k].m = v.m
+			mm.msgs[k].SignedMessage = v.SignedMessage
 		} else {
 			mm.msgs[k] = v
 		}
@@ -117,9 +120,9 @@ func (mm *MpoolMonitor) update(ctx context.Context) error {
 	return nil
 }
 
-func (mm *MpoolMonitor) PendingLocal(ctx context.Context) ([]*types.SignedMessage, error) {
+func (mm *MpoolMonitor) PendingLocal(ctx context.Context) ([]*TimeStampedMsg, error) {
 	localAddr := make(map[string]struct{})
-	var ret []*types.SignedMessage
+	var ret []*TimeStampedMsg
 
 	addrs, err := mm.fullNode.WalletList(ctx)
 	if err != nil {
@@ -134,23 +137,23 @@ func (mm *MpoolMonitor) PendingLocal(ctx context.Context) ([]*types.SignedMessag
 	defer mm.lk.Unlock()
 
 	for _, msg := range mm.msgs {
-		if _, has := localAddr[msg.m.Message.From.String()]; !has {
+		if _, has := localAddr[msg.SignedMessage.Message.From.String()]; !has {
 			continue
 		}
-		ret = append(ret, &msg.m)
+		ret = append(ret, msg)
 	}
 
 	return ret, nil
 }
 
-func (mm *MpoolMonitor) PendingAll() ([]*types.SignedMessage, error) {
-	var ret []*types.SignedMessage
+func (mm *MpoolMonitor) PendingAll() ([]*TimeStampedMsg, error) {
+	var ret []*TimeStampedMsg
 
 	mm.lk.Lock()
 	defer mm.lk.Unlock()
 
 	for _, msg := range mm.msgs {
-		ret = append(ret, &msg.m)
+		ret = append(ret, msg)
 	}
 
 	return ret, nil
@@ -163,12 +166,14 @@ func (mm *MpoolMonitor) Alerts(ctx context.Context) ([]cid.Cid, error) {
 		return nil, fmt.Errorf("failed to get chain head: %w", err)
 	}
 
-	mm.lk.Lock()
-	defer mm.lk.Unlock()
+	msgs, err := mm.PendingLocal(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get local messages: %w", err)
+	}
 
-	for mcid := range mm.msgs {
-		if mm.msgs[mcid].added+mm.mpoolAlertEpochs <= ts.Height() {
-			ret = append(ret, mcid)
+	for _, msg := range msgs {
+		if msg.Added+mm.mpoolAlertEpochs <= ts.Height() {
+			ret = append(ret, msg.SignedMessage.Cid())
 		}
 	}
 

react/src/Mpool.js

@@ -4,6 +4,8 @@ import {React, useState} from "react";
 import {humanFIL} from "./util";
 import './Mpool.css'
 import {PageContainer} from "./Components";
+import {EpochQuery} from "./gql";
+import moment from "moment";
 
 export function MpoolPage(props) {
     return <PageContainer pageType="mpool" title="Message Pool">
@@ -13,7 +15,7 @@ export function MpoolPage(props) {
 
 function MpoolContent(props) {
     const [local, setLocal] = useState(true)
-    const {loading, error, data} = useQuery(MpoolQuery, { variables: { local } })
+    const {loading, error, data} = useQuery(MpoolQuery, { variables: { local } , pollInterval: 10000, fetchPolicy: "network-only"})
 
     if (loading) {
         return <div>Loading...</div>
@@ -44,11 +46,20 @@ function MpoolMessage(props) {
     const i = props.i
     const msg = props.msg
 
+    const {data} = useQuery(EpochQuery)
+
     const handleParamsClick = (el) => {
         el.target.classList.toggle('expanded')
     }
 
+    let elapsed = (data.epoch.Epoch+'') - (msg.SentEpoch+'')
+    let x = moment().add(-(elapsed)*(data.epoch.SecondsPerEpoch+''), 'seconds').fromNow()
+
     return <>
+        <tr key={"sentepoch"}>
+            <td>Sent At Epoch</td>
+            <td>{msg.SentEpoch+''} ({elapsed} epochs / {x} ago)</td>
+        </tr>
         <tr key={"to"}>
             <td>To</td>
             <td className="address">{msg.To}</td>

react/src/gql.js

@@ -620,6 +620,7 @@ const StorageAskUpdate = gql`
 const MpoolQuery = gql`
     query AppMpoolQuery($local: Boolean!) {
         mpool(local: $local) {
+            SentEpoch
             From
             To
             Nonce