Repository: decred/dcrd
Pull request #3235 (2024-04-25 20:19:11)

wire: Add epoch field to mix key exchange message
The epoch will be included in the derivation of the session hash, instead of only deriving a session from previous pair requests.  This allows new messages in following epochs to not conflict with previous session attempts of the same pair request set.

This also provides a decent heuristic to discover and exclude any peers that were unresponsive for a full epoch or more.

This is a breaking API change but the wire module has not seen a release since mixing messages were added.


Patch:
diff --git a/connmgr/connmanager.go b/connmgr/connmanager.go
index 1e66da86..6bcc35fd 100644
--- a/connmgr/connmanager.go
+++ b/connmgr/connmanager.go
@@ -379,19 +379,21 @@ out:
 					continue
 				}
 
-				// Otherwise, we will attempt a reconnection if
-				// we do not have enough peers, or if this is a
-				// persistent peer. The connection request is
-				// re added to the pending map, so that
-				// subsequent processing of connections and
-				// failures do not ignore the request.
-				if uint32(len(conns)) < cm.cfg.TargetOutbound ||
-					connReq.Permanent {
-
-					connReq.updateState(ConnPending)
-					log.Debugf("Reconnecting to %v",
-						connReq)
-					pending[msg.id] = connReq
+				// Otherwise, attempt a reconnection when there are not already
+				// enough outbound peers to satisfy the target number of
+				// outbound peers or this is a persistent peer.
+				numConns := uint32(len(conns))
+				if numConns < cm.cfg.TargetOutbound || connReq.Permanent {
+					// The connection request is reused for persistent peers, so
+					// add it back to the pending map in that case so that
+					// subsequent processing of connections and failures do not
+					// ignore the request.
+					if connReq.Permanent {
+						connReq.updateState(ConnPending)
+						log.Debugf("Reconnecting to %v", connReq)
+						pending[msg.id] = connReq
+					}
+
 					cm.handleFailedConn(ctx, connReq)
 				}
 
diff --git a/internal/mining/mining.go b/internal/mining/mining.go
index d2b2be20..ef51e961 100644
--- a/internal/mining/mining.go
+++ b/internal/mining/mining.go
@@ -897,7 +897,15 @@ func (g *BlkTmplGenerator) handleTooFewVoters(nextHeight int64,
 			return nil, err
 		}
 		header := &block.Header
-		header.MerkleRoot = calcBlockMerkleRoot(block.Transactions, block.STransactions, hdrCmtActive)
+		header.MerkleRoot = calcBlockMerkleRoot(block.Transactions,
+			block.STransactions, hdrCmtActive)
+
+		// Calculate the required difficulty for the block.
+		reqDifficulty, err := g.cfg.CalcNextRequiredDifficulty(prevHash, ts)
+		if err != nil {
+			return nil, makeError(ErrGettingDifficulty, err.Error())
+		}
+		header.Bits = reqDifficulty
 
 		// Calculate the stake root or commitment root depending on the result
 		// of the header commitments agenda vote.
diff --git a/rpcadaptors.go b/rpcadaptors.go
index 643f9f80..ea22f783 100644
--- a/rpcadaptors.go
+++ b/rpcadaptors.go
@@ -94,7 +94,7 @@ func (p *rpcPeer) LastPingNonce() uint64 {
 // This function is safe for concurrent access and is part of the rpcserver.Peer
 // interface implementation.
 func (p *rpcPeer) IsTxRelayDisabled() bool {
-	return (*serverPeer)(p).relayTxDisabled()
+	return (*serverPeer)(p).disableRelayTx.Load()
 }
 
 // BanScore returns the current integer value that represents how close the peer
diff --git a/server.go b/server.go
index 0f4d6c16..a6692915 100644
--- a/server.go
+++ b/server.go
@@ -551,8 +551,7 @@ type serverPeer struct {
 	server         *server
 	persistent     bool
 	continueHash   atomic.Pointer[chainhash.Hash]
-	relayMtx       sync.Mutex
-	disableRelayTx bool
+	disableRelayTx atomic.Bool
 	isWhitelisted  bool
 	knownAddresses *apbf.Filter
 	banScore       connmgr.DynamicBanScore
@@ -575,8 +574,7 @@ type serverPeer struct {
 	blockProcessed chan struct{}
 
 	// peerNa is network address of the peer connected to.
-	peerNa    *wire.NetAddress
-	peerNaMtx sync.Mutex
+	peerNa atomic.Pointer[wire.NetAddress]
 
 	// announcedBlock tracks the most recent block announced to this peer and is
 	// used to filter duplicates.
@@ -824,25 +822,6 @@ func (sp *serverPeer) addressKnown(na *addrmgr.NetAddress) bool {
 	return sp.knownAddresses.Contains([]byte(na.Key()))
 }
 
-// setDisableRelayTx toggles relaying of transactions for the given peer.
-// It is safe for concurrent access.
-func (sp *serverPeer) setDisableRelayTx(disable bool) {
-	sp.relayMtx.Lock()
-	sp.disableRelayTx = disable
-	sp.relayMtx.Unlock()
-}
-
-// relayTxDisabled returns whether or not relaying of transactions for the given
-// peer is disabled.
-// It is safe for concurrent access.
-func (sp *serverPeer) relayTxDisabled() bool {
-	sp.relayMtx.Lock()
-	isDisabled := sp.disableRelayTx
-	sp.relayMtx.Unlock()
-
-	return isDisabled
-}
-
 // wireToAddrmgrNetAddress converts a wire NetAddress to an address manager
 // NetAddress.
 func wireToAddrmgrNetAddress(netAddr *wire.NetAddress) *addrmgr.NetAddress {
@@ -1022,12 +1001,10 @@ func (sp *serverPeer) OnVersion(_ *peer.Peer, msg *wire.MsgVersion) {
 		}
 	}
 
-	sp.peerNaMtx.Lock()
-	sp.peerNa = &msg.AddrYou
-	sp.peerNaMtx.Unlock()
+	sp.peerNa.Store(&msg.AddrYou)
 
 	// Choose whether or not to relay transactions.
-	sp.setDisableRelayTx(msg.DisableRelayTx)
+	sp.disableRelayTx.Store(msg.DisableRelayTx)
 
 	// Add the remote peer time as a sample for creating an offset against
 	// the local clock to keep the network time in sync.
@@ -1861,9 +1838,7 @@ func (s *server) handleAddPeerMsg(state *peerState, sp *serverPeer) bool {
 		return false
 	}
 
-	sp.peerNaMtx.Lock()
-	na := sp.peerNa
-	sp.peerNaMtx.Unlock()
+	na := sp.peerNa.Load()
 
 	// Add the new peer and start it.
 	srvrLog.Debugf("New peer %s", sp)
@@ -1985,9 +1960,14 @@ func (s *server) handleDonePeerMsg(state *peerState, sp *serverPeer) {
 		s.connManager.Disconnect(sp.connReq.ID())
 	}
 
-	// Update the address' last seen time if the peer has acknowledged
-	// our version and has sent us its version as well.
-	if sp.VerAckReceived() && sp.VersionKnown() && sp.NA() != nil {
+	// Update the address manager with the last seen time when the peer has
+	// acknowledged our version and has sent us its version as well.  This is
+	// skipped when running on the simulation and regression test networks since
+	// they are only intended to connect to specified peers and actively avoid
+	// advertising and connecting to discovered peers.
+	if !cfg.SimNet && !cfg.RegNet && sp.VerAckReceived() && sp.VersionKnown() &&
+		sp.NA() != nil {
+
 		remoteAddr := wireToAddrmgrNetAddress(sp.NA())
 		err := s.addrManager.Connected(remoteAddr)
 		if err != nil {
@@ -2057,9 +2037,9 @@ func (s *server) handleRelayInvMsg(state *peerState, msg relayMsg) {
 		}
 
 		if iv.Type == wire.InvTypeTx {
-			// Don't relay the transaction to the peer when it has
-			// transaction relaying disabled.
-			if sp.relayTxDisabled() {
+			// Don't relay the transaction to the peer when it has transaction
+			// relaying disabled.
+			if sp.disableRelayTx.Load() {
 				return
 			}
 		}
diff --git a/wire/msgmixkeyexchange.go b/wire/msgmixkeyexchange.go
index e7bfa4ab..62c4553f 100644
--- a/wire/msgmixkeyexchange.go
+++ b/wire/msgmixkeyexchange.go
@@ -30,6 +30,7 @@ type MsgMixKeyExchange struct {
 	Signature  [64]byte
 	Identity   [33]byte
 	SessionID  [32]byte
+	Epoch      uint64
 	Run        uint32
 	ECDH       [33]byte   // Secp256k1 public key
 	PQPK       [1218]byte // Sntrup4591761 public key
@@ -53,7 +54,7 @@ func (msg *MsgMixKeyExchange) BtcDecode(r io.Reader, pver uint32) error {
 	}
 
 	err := readElements(r, &msg.Signature, &msg.Identity, &msg.SessionID,
-		&msg.Run, &msg.ECDH, &msg.PQPK, &msg.Commitment)
+		&msg.Epoch, &msg.Run, &msg.ECDH, &msg.PQPK, &msg.Commitment)
 	if err != nil {
 		return err
 	}
@@ -146,8 +147,8 @@ func (msg *MsgMixKeyExchange) writeMessageNoSignature(op string, w io.Writer, pv
 		return messageError(op, ErrTooManyPrevMixMsgs, msg)
 	}
 
-	err := writeElements(w, &msg.Identity, &msg.SessionID, msg.Run,
-		&msg.ECDH, &msg.PQPK, &msg.Commitment)
+	err := writeElements(w, &msg.Identity, &msg.SessionID, msg.Epoch,
+		msg.Run, &msg.ECDH, &msg.PQPK, &msg.Commitment)
 	if err != nil {
 		return err
 	}
@@ -188,7 +189,7 @@ func (msg *MsgMixKeyExchange) MaxPayloadLength(pver uint32) uint32 {
 	}
 
 	// See tests for this calculation.
-	return 17803
+	return 17811
 }
 
 // Pub returns the message sender's public key identity.
@@ -219,12 +220,14 @@ func (msg *MsgMixKeyExchange) GetRun() uint32 {
 // NewMsgMixKeyExchange returns a new mixkeyxchg message that conforms to the
 // Message interface using the passed parameters and defaults for the
 // remaining fields.
-func NewMsgMixKeyExchange(identity [33]byte, sid [32]byte, run uint32,
-	ecdh [33]byte, pqpk [1218]byte, commitment [32]byte, seenPRs []chainhash.Hash) *MsgMixKeyExchange {
+func NewMsgMixKeyExchange(identity [33]byte, sid [32]byte, epoch uint64,
+	run uint32, ecdh [33]byte, pqpk [1218]byte, commitment [32]byte,
+	seenPRs []chainhash.Hash) *MsgMixKeyExchange {
 
 	return &MsgMixKeyExchange{
 		Identity:   identity,
 		SessionID:  sid,
+		Epoch:      epoch,
 		Run:        run,
 		ECDH:       ecdh,
 		PQPK:       pqpk,


Test patch:
diff --git a/wire/message_test.go b/wire/message_test.go
index fee36296..e7f899ab 100644
--- a/wire/message_test.go
+++ b/wire/message_test.go
@@ -84,7 +84,7 @@ func TestMessage(t *testing.T) {
 	if err != nil {
 		t.Errorf("NewMsgMixPairReq: %v", err)
 	}
-	msgMixKE := NewMsgMixKeyExchange([33]byte{}, [32]byte{}, 1, [33]byte{}, [1218]byte{}, [32]byte{}, []chainhash.Hash{})
+	msgMixKE := NewMsgMixKeyExchange([33]byte{}, [32]byte{}, 1, 1, [33]byte{}, [1218]byte{}, [32]byte{}, []chainhash.Hash{})
 	msgMixCT := NewMsgMixCiphertexts([33]byte{}, [32]byte{}, 1, [][1047]byte{}, []chainhash.Hash{})
 	msgMixSR := NewMsgMixSlotReserve([33]byte{}, [32]byte{}, 1, [][][]byte{{{}}}, []chainhash.Hash{})
 	msgMixDC := NewMsgMixDCNet([33]byte{}, [32]byte{}, 1, []MixVect{make(MixVect, 1)}, []chainhash.Hash{})
@@ -123,7 +123,7 @@ func TestMessage(t *testing.T) {
 		{msgGetInitState, msgGetInitState, pver, MainNet, 25},
 		{msgInitState, msgInitState, pver, MainNet, 28},
 		{msgMixPR, msgMixPR, pver, MainNet, 165},
-		{msgMixKE, msgMixKE, pver, MainNet, 1441},
+		{msgMixKE, msgMixKE, pver, MainNet, 1449},
 		{msgMixCT, msgMixCT, pver, MainNet, 158},
 		{msgMixSR, msgMixSR, pver, MainNet, 161},
 		{msgMixDC, msgMixDC, pver, MainNet, 181},
diff --git a/wire/msgmixkeyexchange_test.go b/wire/msgmixkeyexchange_test.go
index e5f83a32..94118c97 100644
--- a/wire/msgmixkeyexchange_test.go
+++ b/wire/msgmixkeyexchange_test.go
@@ -21,18 +21,19 @@ func newTestMixKeyExchange() *MsgMixKeyExchange {
 	id := *(*[33]byte)(repeat(0x81, 33))
 	sid := *(*[32]byte)(repeat(0x82, 32))
 
-	const run = uint32(0x83838383)
+	const epoch = uint64(0x8383838383838383)
+	const run = uint32(0x84848484)
 
-	ecdh := *(*[33]byte)(repeat(0x84, 33))
-	pqpk := *(*[1218]byte)(repeat(0x85, 1218))
-	commitment := *(*[32]byte)(repeat(0x86, 32))
+	ecdh := *(*[33]byte)(repeat(0x85, 33))
+	pqpk := *(*[1218]byte)(repeat(0x86, 1218))
+	commitment := *(*[32]byte)(repeat(0x87, 32))
 
 	seenPRs := make([]chainhash.Hash, 4)
-	for b := byte(0x87); b < 0x8B; b++ {
-		copy(seenPRs[b-0x87][:], repeat(b, 32))
+	for b := byte(0x88); b < 0x8C; b++ {
+		copy(seenPRs[b-0x88][:], repeat(b, 32))
 	}
 
-	ke := NewMsgMixKeyExchange(id, sid, run, ecdh, pqpk, commitment, seenPRs)
+	ke := NewMsgMixKeyExchange(id, sid, epoch, run, ecdh, pqpk, commitment, seenPRs)
 	ke.Signature = sig
 
 	return ke
@@ -53,16 +54,17 @@ func TestMsgMixKeyExchangeWire(t *testing.T) {
 	expected = append(expected, repeat(0x80, 64)...)   // Signature
 	expected = append(expected, repeat(0x81, 33)...)   // Identity
 	expected = append(expected, repeat(0x82, 32)...)   // Session ID
-	expected = append(expected, repeat(0x83, 4)...)    // Run
-	expected = append(expected, repeat(0x84, 33)...)   // ECDH public key
-	expected = append(expected, repeat(0x85, 1218)...) // PQ public key
-	expected = append(expected, repeat(0x86, 32)...)   // Secrets commitment
-	// Four seen PRs (repeating 32 bytes of 0x87, 0x88, 0x89, 0x8a)
+	expected = append(expected, repeat(0x83, 8)...)    // Epoch
+	expected = append(expected, repeat(0x84, 4)...)    // Run
+	expected = append(expected, repeat(0x85, 33)...)   // ECDH public key
+	expected = append(expected, repeat(0x86, 1218)...) // PQ public key
+	expected = append(expected, repeat(0x87, 32)...)   // Secrets commitment
+	// Four seen PRs (repeating 32 bytes of 0x88, 0x89, 0x8a, 0x8b)
 	expected = append(expected, 0x04)
-	expected = append(expected, repeat(0x87, 32)...)
 	expected = append(expected, repeat(0x88, 32)...)
 	expected = append(expected, repeat(0x89, 32)...)
 	expected = append(expected, repeat(0x8a, 32)...)
+	expected = append(expected, repeat(0x8b, 32)...)
 
 	expectedSerializationEqual(t, buf.Bytes(), expected)
 
@@ -155,6 +157,7 @@ func TestMsgMixKeyExchangeMaxPayloadLength(t *testing.T) {
 	var expectedLen uint32 = 64 + // Signature
 		33 + // Identity
 		32 + // Session ID
+		8 + // Epoch
 		4 + // Run
 		33 + // ECDH public key
 		1218 + // sntrup4591761 public key