[NOD-1204] Add timestamp and message number to domain messages (#854)

domainmessage/base_message.go

@@ -0,0 +1,24 @@
+package domainmessage
+
+import "time"
+
+type baseMessage struct {
+	messageNumber uint64
+	receivedAt    time.Time
+}
+
+func (b *baseMessage) MessageNumber() uint64 {
+	return b.messageNumber
+}
+
+func (b *baseMessage) SetMessageNumber(messageNumber uint64) {
+	b.messageNumber = messageNumber
+}
+
+func (b *baseMessage) ReceivedAt() time.Time {
+	return b.receivedAt
+}
+
+func (b *baseMessage) SetReceivedAt(receivedAt time.Time) {
+	b.receivedAt = receivedAt
+}

domainmessage/message.go

@@ -6,6 +6,7 @@ package domainmessage
 
 import (
 	"fmt"
+	"time"
 )
 
 // MaxMessagePayload is the maximum bytes a message can be regardless of other
@@ -79,4 +80,8 @@ var MessageCommandToString = map[MessageCommand]string{
 // are used directly in the protocol encoded message.
 type Message interface {
 	Command() MessageCommand
+	MessageNumber() uint64
+	SetMessageNumber(index uint64)
+	ReceivedAt() time.Time
+	SetReceivedAt(receivedAt time.Time)
 }

domainmessage/msgaddresses.go

@@ -24,6 +24,7 @@ const MaxAddressesPerMsg = 1000
 // Use the AddAddress function to build up the list of known addresses when
 // sending an Addresses message to another peer.
 type MsgAddresses struct {
+	baseMessage
 	IncludeAllSubnetworks bool
 	SubnetworkID          *subnetworkid.SubnetworkID
 	AddrList              []*NetAddress

domainmessage/msgblock.go

@@ -42,6 +42,7 @@ type TxLoc struct {
 // block message. It is used to deliver block and transaction information in
 // response to a getdata message (MsgGetData) for a given block hash.
 type MsgBlock struct {
+	baseMessage
 	Header       BlockHeader
 	Transactions []*MsgTx
 }

domainmessage/msgblocklocator.go

@@ -12,6 +12,7 @@ const MaxBlockLocatorsPerMsg = 500
 // locator message. It is used to find the blockLocator of a peer that is
 // syncing with you.
 type MsgBlockLocator struct {
+	baseMessage
 	BlockLocatorHashes []*daghash.Hash
 }
 

domainmessage/msgdoneibdblocks.go

@@ -5,7 +5,9 @@ package domainmessage
 // syncer sent all the requested blocks.
 //
 // This message has no payload.
-type MsgDoneIBDBlocks struct{}
+type MsgDoneIBDBlocks struct {
+	baseMessage
+}
 
 // Command returns the protocol command string for the message. This is part
 // of the Message interface implementation.

domainmessage/msgibdblock.go

@@ -8,6 +8,7 @@ package domainmessage
 // ibdblock message. It is used to deliver block and transaction information in
 // response to a RequestIBDBlocks message (MsgRequestIBDBlocks).
 type MsgIBDBlock struct {
+	baseMessage
 	*MsgBlock
 }
 
@@ -26,5 +27,5 @@ func (msg *MsgIBDBlock) MaxPayloadLength(pver uint32) uint32 {
 // NewMsgIBDBlock returns a new kaspa ibdblock message that conforms to the
 // Message interface. See MsgIBDBlock for details.
 func NewMsgIBDBlock(msgBlock *MsgBlock) *MsgIBDBlock {
-	return &MsgIBDBlock{msgBlock}
+	return &MsgIBDBlock{MsgBlock: msgBlock}
 }

domainmessage/msgibdblock_test.go

@@ -77,8 +77,8 @@ func TestIBDBlockEncoding(t *testing.T) {
 	}{
 		// Latest protocol version.
 		{
-			&MsgIBDBlock{&blockOne},
-			&MsgIBDBlock{&blockOne},
+			&MsgIBDBlock{MsgBlock: &blockOne},
+			&MsgIBDBlock{MsgBlock: &blockOne},
 			blockOneBytes,
 			blockOneTxLocs,
 			ProtocolVersion,

domainmessage/msginvrelayblock.go

@@ -8,6 +8,7 @@ import (
 // block inventory message. It is used to notify the network about new block
 // by sending their hash, and let the receiving node decide if it needs it.
 type MsgInvRelayBlock struct {
+	baseMessage
 	Hash *daghash.Hash
 }
 

domainmessage/msginvtransaction.go

@@ -12,6 +12,7 @@ const MaxInvPerTxInvMsg = MaxInvPerMsg
 // TxInv message. It is used to notify the network about new transactions
 // by sending their ID, and let the receiving node decide if it needs it.
 type MsgInvTransaction struct {
+	baseMessage
 	TxIDs []*daghash.TxID
 }
 

domainmessage/msgping.go

@@ -16,6 +16,7 @@ package domainmessage
 // The payload for this message just consists of a nonce used for identifying
 // it later.
 type MsgPing struct {
+	baseMessage
 	// Unique value associated with message that is used to identify
 	// specific ping message.
 	Nonce uint64

domainmessage/msgpong.go

@@ -10,6 +10,7 @@ package domainmessage
 //
 // This message was not added until protocol versions AFTER BIP0031Version.
 type MsgPong struct {
+	baseMessage
 	// Unique value associated with message that is used to identify
 	// specific ping message.
 	Nonce uint64

domainmessage/msgrequestaddresses.go

@@ -15,6 +15,7 @@ import (
 //
 // This message has no payload.
 type MsgRequestAddresses struct {
+	baseMessage
 	IncludeAllSubnetworks bool
 	SubnetworkID          *subnetworkid.SubnetworkID
 }

domainmessage/msgrequestblocklocator.go

@@ -9,6 +9,7 @@ import (
 // and low hash.
 // The locator is returned via a locator message (MsgBlockLocator).
 type MsgRequestBlockLocator struct {
+	baseMessage
 	HighHash *daghash.Hash
 	LowHash  *daghash.Hash
 }

domainmessage/msgrequestibdblocks.go

@@ -12,6 +12,7 @@ import (
 // RequestIBDBlocks message. It is used to request a list of blocks starting after the
 // low hash and until the high hash.
 type MsgRequestIBDBlocks struct {
+	baseMessage
 	LowHash  *daghash.Hash
 	HighHash *daghash.Hash
 }

domainmessage/msgrequestnextibdblocks.go

@@ -5,7 +5,9 @@ package domainmessage
 // more blocks.
 //
 // This message has no payload.
-type MsgRequestNextIBDBlocks struct{}
+type MsgRequestNextIBDBlocks struct {
+	baseMessage
+}
 
 // Command returns the protocol command string for the message. This is part
 // of the Message interface implementation.

domainmessage/msgrequestrelayblocks.go

@@ -12,6 +12,7 @@ const MsgRequestRelayBlocksHashes = MaxInvPerMsg
 // RequestRelayBlocks message. It is used to request blocks as part of the block
 // relay protocol.
 type MsgRequestRelayBlocks struct {
+	baseMessage
 	Hashes []*daghash.Hash
 }
 

domainmessage/msgrequestselectedtip.go

@@ -4,7 +4,9 @@ package domainmessage
 // RequestSelectedTip message. It is used to request the selected tip of another peer.
 //
 // This message has no payload.
-type MsgRequestSelectedTip struct{}
+type MsgRequestSelectedTip struct {
+	baseMessage
+}
 
 // Command returns the protocol command string for the message. This is part
 // of the Message interface implementation.

domainmessage/msgrequesttransactions.go

@@ -12,6 +12,7 @@ const MaxInvPerRequestTransactionsMsg = MaxInvPerMsg
 // RequestTransactions message. It is used to request transactions as part of the
 // transactions relay protocol.
 type MsgRequestTransactions struct {
+	baseMessage
 	IDs []*daghash.TxID
 }
 

domainmessage/msgselectedtip.go

@@ -8,6 +8,7 @@ import (
 // selectedtip message. It is used to answer getseltip messages and tell
 // the asking peer what is the selected tip of this peer.
 type MsgSelectedTip struct {
+	baseMessage
 	// The selected tip hash of the generator of the message.
 	SelectedTipHash *daghash.Hash
 }

domainmessage/msgtransactionnotfound.go

@@ -11,6 +11,7 @@ import (
 // MsgTransactionNotFound defines a kaspa TransactionNotFound message which is sent in response to
 // a RequestTransactions message if any of the requested data in not available on the peer.
 type MsgTransactionNotFound struct {
+	baseMessage
 	ID *daghash.TxID
 }
 

domainmessage/msgtx.go

@@ -268,6 +268,7 @@ func NewTxOut(value uint64, scriptPubKey []byte) *TxOut {
 // Use the AddTxIn and AddTxOut functions to build up the list of transaction
 // inputs and outputs.
 type MsgTx struct {
+	baseMessage
 	Version      int32
 	TxIn         []*TxIn
 	TxOut        []*TxOut

domainmessage/msgverack.go

@@ -9,7 +9,9 @@ package domainmessage
 // to negotiate parameters. It implements the Message interface.
 //
 // This message has no payload.
-type MsgVerAck struct{}
+type MsgVerAck struct {
+	baseMessage
+}
 
 // Command returns the protocol command string for the message. This is part
 // of the Message interface implementation.

domainmessage/msgversion.go

@@ -31,6 +31,7 @@ var DefaultUserAgent = fmt.Sprintf("/kaspad:%s/", version.Version())
 // message (MsgVerAck). This exchange must take place before any further
 // communication is allowed to proceed.
 type MsgVersion struct {
+	baseMessage
 	// Version of the protocol the node is using.
 	ProtocolVersion uint32
 

netadapter/server/grpcserver/connection_loops.go

@@ -2,6 +2,7 @@ package grpcserver
 
 import (
 	"io"
+	"time"
 
 	routerpkg "github.com/kaspanet/kaspad/netadapter/router"
 	"github.com/pkg/errors"
@@ -61,6 +62,7 @@ func (c *gRPCConnection) sendLoop() error {
 }
 
 func (c *gRPCConnection) receiveLoop() error {
+	messageNumber := uint64(0)
 	for c.IsConnected() {
 		protoMessage, err := c.stream.Recv()
 		if err != nil {
@@ -75,8 +77,15 @@ func (c *gRPCConnection) receiveLoop() error {
 			return err
 		}
 
-		log.Debugf("incoming '%s' message from %s", message.Command(), c)
-		log.Tracef("incoming '%s' message from %s: %s", message.Command(), c, logger.NewLogClosure(func() string {
+		messageNumber++
+		message.SetMessageNumber(messageNumber)
+		message.SetReceivedAt(time.Now())
+
+		log.Debugf("incoming '%s' message from %s (message number %d)", message.Command(), c,
+			message.MessageNumber())
+
+		log.Tracef("incoming '%s' message from %s  (message number %d): %s", message.Command(),
+			c, message.MessageNumber(), logger.NewLogClosure(func() string {
 				return spew.Sdump(message)
 			}))