Merge remote-tracking branch 'origin/v0.4.1-dev' into v0.5.0-dev

blockdag/utxodiffstore.go

@@ -156,7 +156,9 @@ func (diffStore *utxoDiffStore) clearDirtyEntries() {
 var maxBlueScoreDifferenceToKeepLoaded uint64 = 100
 
 // clearOldEntries removes entries whose blue score is lower than
-// virtual.blueScore - maxBlueScoreDifferenceToKeepLoaded.
+// virtual.blueScore - maxBlueScoreDifferenceToKeepLoaded. Note
+// that tips are not removed either even if their blue score is
+// lower than the above.
 func (diffStore *utxoDiffStore) clearOldEntries() {
 	virtualBlueScore := diffStore.dag.VirtualBlueScore()
 	minBlueScore := virtualBlueScore - maxBlueScoreDifferenceToKeepLoaded
@@ -164,9 +166,11 @@ func (diffStore *utxoDiffStore) clearOldEntries() {
 		minBlueScore = 0
 	}
 
+	tips := diffStore.dag.virtual.tips()
+
 	toRemove := make(map[*blockNode]struct{})
 	for node := range diffStore.loaded {
-		if node.blueScore < minBlueScore {
+		if node.blueScore < minBlueScore && !tips.contains(node) {
 			toRemove[node] = struct{}{}
 		}
 	}

blockdag/utxodiffstore_test.go

@@ -1,12 +1,13 @@
 package blockdag
 
 import (
+	"reflect"
+	"testing"
+
 	"github.com/kaspanet/kaspad/dagconfig"
 	"github.com/kaspanet/kaspad/dbaccess"
 	"github.com/kaspanet/kaspad/util/daghash"
 	"github.com/kaspanet/kaspad/wire"
-	"reflect"
-	"testing"
 )
 
 func TestUTXODiffStore(t *testing.T) {
@@ -149,10 +150,11 @@ func TestClearOldEntries(t *testing.T) {
 		t.Fatalf("TestClearOldEntries: missing blockNode for hash %s", processedBlock.BlockHash())
 	}
 
-	// Make sure that the child-of-genesis node isn't in the loaded set
+	// Make sure that the child-of-genesis node is in the loaded set, since it
+	// is a tip.
 	_, ok = dag.utxoDiffStore.loaded[node]
-	if ok {
-		t.Fatalf("TestClearOldEntries: diffData for node %s is in the loaded set", node.hash)
+	if !ok {
+		t.Fatalf("TestClearOldEntries: diffData for node %s is not in the loaded set", node.hash)
 	}
 
 	// Make sure that all the old nodes still do not exist in the loaded set

netsync/manager.go

@@ -132,13 +132,13 @@ type requestQueueAndSet struct {
 // peerSyncState stores additional information that the SyncManager tracks
 // about a peer.
 type peerSyncState struct {
-	syncCandidate           bool
-	lastSelectedTipRequest  time.Time
-	isPendingForSelectedTip bool
-	requestQueueMtx         sync.Mutex
-	requestQueues           map[wire.InvType]*requestQueueAndSet
-	requestedTxns           map[daghash.TxID]struct{}
-	requestedBlocks         map[daghash.Hash]struct{}
+	syncCandidate             bool
+	lastSelectedTipRequest    time.Time
+	peerShouldSendSelectedTip bool
+	requestQueueMtx           sync.Mutex
+	requestQueues             map[wire.InvType]*requestQueueAndSet
+	requestedTxns             map[daghash.TxID]struct{}
+	requestedBlocks           map[daghash.Hash]struct{}
 }
 
 // SyncManager is used to communicate block related messages with peers. The
@@ -158,6 +158,7 @@ type SyncManager struct {
 	wg             sync.WaitGroup
 	quit           chan struct{}
 	syncPeerLock   sync.Mutex
+	isSyncing      bool
 
 	// These fields should only be accessed from the messageHandler thread
 	rejectedTxns    map[daghash.TxID]struct{}
@@ -206,13 +207,21 @@ func (sm *SyncManager) startSync() {
 			syncPeer.SelectedTipHash(), syncPeer.Addr())
 
 		syncPeer.PushGetBlockLocatorMsg(syncPeer.SelectedTipHash(), sm.dagParams.GenesisHash)
+		sm.isSyncing = true
 		sm.syncPeer = syncPeer
 		return
 	}
 
+	pendingForSelectedTips := false
+
 	if sm.shouldQueryPeerSelectedTips() {
+		sm.isSyncing = true
 		hasSyncCandidates := false
 		for peer, state := range sm.peerStates {
+			if state.peerShouldSendSelectedTip {
+				pendingForSelectedTips = true
+				continue
+			}
 			if !state.syncCandidate {
 				continue
 			}
@@ -222,21 +231,26 @@ func (sm *SyncManager) startSync() {
 				continue
 			}
 
-			queueMsgGetSelectedTip(peer, state)
+			sm.queueMsgGetSelectedTip(peer, state)
+			pendingForSelectedTips = true
 		}
 		if !hasSyncCandidates {
 			log.Warnf("No sync peer candidates available")
 		}
 	}
+
+	if !pendingForSelectedTips {
+		sm.isSyncing = false
+	}
 }
 
 func (sm *SyncManager) shouldQueryPeerSelectedTips() bool {
 	return sm.dag.Now().Sub(sm.dag.CalcPastMedianTime()) > minDAGTimeDelay
 }
 
-func queueMsgGetSelectedTip(peer *peerpkg.Peer, state *peerSyncState) {
+func (sm *SyncManager) queueMsgGetSelectedTip(peer *peerpkg.Peer, state *peerSyncState) {
 	state.lastSelectedTipRequest = time.Now()
-	state.isPendingForSelectedTip = true
+	state.peerShouldSendSelectedTip = true
 	peer.QueueMessage(wire.NewMsgGetSelectedTip(), nil)
 }
 
@@ -417,17 +431,6 @@ func (sm *SyncManager) handleTxMsg(tmsg *txMsg) {
 	sm.peerNotifier.AnnounceNewTransactions(acceptedTxs)
 }
 
-// synced returns true if we believe we are synced with our peers, false if we
-// still have blocks to check
-//
-// We consider ourselves synced iff both of the following are true:
-// 1. there's no syncPeer, a.k.a. all connected peers are at the same tip
-// 2. the DAG considers itself synced - to prevent attacks where a peer sends an
-//    unknown tip but never lets us sync to it.
-func (sm *SyncManager) synced() bool {
-	return sm.syncPeer == nil && sm.dag.IsSynced()
-}
-
 // restartSyncIfNeeded finds a new sync candidate if we're not expecting any
 // blocks from the current one.
 func (sm *SyncManager) restartSyncIfNeeded() {
@@ -763,7 +766,7 @@ func (sm *SyncManager) handleInvMsg(imsg *invMsg) {
 		log.Errorf("Failed to send invs from queue: %s", err)
 	}
 
-	if haveUnknownInvBlock && !sm.synced() {
+	if haveUnknownInvBlock && !sm.isSyncing {
 		// If one of the inv messages is an unknown block
 		// it is an indication that one of our peers has more
 		// up-to-date data than us.
@@ -848,7 +851,7 @@ func (sm *SyncManager) sendInvsFromRequestQueue(peer *peerpkg.Peer, state *peerS
 	if err != nil {
 		return err
 	}
-	if sm.syncPeer == nil || sm.isSynced() {
+	if !sm.isSyncing || sm.isSynced() {
 		err := sm.addInvsToGetDataMessageFromQueue(gdmsg, state, wire.InvTypeBlock, wire.MaxInvPerGetDataMsg)
 		if err != nil {
 			return err
@@ -918,12 +921,12 @@ func (sm *SyncManager) handleSelectedTipMsg(msg *selectedTipMsg) {
 	peer := msg.peer
 	selectedTipHash := msg.selectedTipHash
 	state := sm.peerStates[peer]
-	if !state.isPendingForSelectedTip {
+	if !state.peerShouldSendSelectedTip {
 		log.Warnf("Got unrequested selected tip message from %s -- "+
 			"disconnecting", peer.Addr())
 		peer.Disconnect()
 	}
-	state.isPendingForSelectedTip = false
+	state.peerShouldSendSelectedTip = false
 	if selectedTipHash.IsEqual(peer.SelectedTipHash()) {
 		return
 	}
@@ -1026,9 +1029,9 @@ func (sm *SyncManager) handleBlockDAGNotification(notification *blockdag.Notific
 			}
 		})
 
-		// Relay if we are synced and the block was not just now unorphaned.
-		// Otherwise peers that are synced should already know about it
-		if sm.synced() && !data.WasUnorphaned {
+		// Relay if we are current and the block was not just now unorphaned.
+		// Otherwise peers that are current should already know about it
+		if sm.isSynced() && !data.WasUnorphaned {
 			iv := wire.NewInvVect(wire.InvTypeBlock, block.Hash())
 			sm.peerNotifier.RelayInventory(iv, block.MsgBlock().Header)
 		}

server/p2p/p2p.go

@@ -109,15 +109,6 @@ type relayMsg struct {
 	data    interface{}
 }
 
-type outboundPeerConnectedMsg struct {
-	connReq *connmgr.ConnReq
-	conn    net.Conn
-}
-
-type outboundPeerConnectionFailedMsg struct {
-	connReq *connmgr.ConnReq
-}
-
 // Peer extends the peer to maintain state shared by the server and
 // the blockmanager.
 type Peer struct {
@@ -229,19 +220,17 @@ type Server struct {
 	DAG         *blockdag.BlockDAG
 	TxMemPool   *mempool.TxPool
 
-	modifyRebroadcastInv        chan interface{}
-	newPeers                    chan *Peer
-	donePeers                   chan *Peer
-	banPeers                    chan *Peer
-	newOutboundConnection       chan *outboundPeerConnectedMsg
-	newOutboundConnectionFailed chan *outboundPeerConnectionFailedMsg
-	Query                       chan interface{}
-	relayInv                    chan relayMsg
-	broadcast                   chan broadcastMsg
-	wg                          sync.WaitGroup
-	nat                         serverutils.NAT
-	TimeSource                  blockdag.TimeSource
-	services                    wire.ServiceFlag
+	modifyRebroadcastInv chan interface{}
+	newPeers             chan *Peer
+	donePeers            chan *Peer
+	banPeers             chan *Peer
+	Query                chan interface{}
+	relayInv             chan relayMsg
+	broadcast            chan broadcastMsg
+	wg                   sync.WaitGroup
+	nat                  serverutils.NAT
+	TimeSource           blockdag.TimeSource
+	services             wire.ServiceFlag
 
 	// We add to quitWaitGroup before every instance in which we wait for
 	// the quit channel so that all those instances finish before we shut
@@ -977,17 +966,17 @@ func (s *Server) inboundPeerConnected(conn net.Conn) {
 // peer instance, associates it with the relevant state such as the connection
 // request instance and the connection itself, and finally notifies the address
 // manager of the attempt.
-func (s *Server) outboundPeerConnected(state *peerState, msg *outboundPeerConnectedMsg) {
-	sp := newServerPeer(s, msg.connReq.Permanent)
-	outboundPeer, err := peer.NewOutboundPeer(newPeerConfig(sp), msg.connReq.Addr.String())
+func (s *Server) outboundPeerConnected(connReq *connmgr.ConnReq, conn net.Conn) {
+	sp := newServerPeer(s, connReq.Permanent)
+	outboundPeer, err := peer.NewOutboundPeer(newPeerConfig(sp), connReq.Addr.String())
 	if err != nil {
-		srvrLog.Debugf("Cannot create outbound peer %s: %s", msg.connReq.Addr, err)
-		s.connManager.Disconnect(msg.connReq.ID())
+		srvrLog.Debugf("Cannot create outbound peer %s: %s", connReq.Addr, err)
+		s.connManager.Disconnect(connReq.ID())
 	}
 	sp.Peer = outboundPeer
-	sp.connReq = msg.connReq
+	sp.connReq = connReq
 
-	s.peerConnected(sp, msg.conn)
+	s.peerConnected(sp, conn)
 
 	s.addrManager.Attempt(sp.NA())
 }
@@ -1012,20 +1001,20 @@ func (s *Server) peerConnected(sp *Peer, conn net.Conn) {
 
 // outboundPeerConnected is invoked by the connection manager when a new
 // outbound connection failed to be established.
-func (s *Server) outboundPeerConnectionFailed(msg *outboundPeerConnectionFailedMsg) {
+func (s *Server) outboundPeerConnectionFailed(connReq *connmgr.ConnReq) {
 	// If the connection request has no address
 	// associated to it, do nothing.
-	if msg.connReq.Addr == nil {
+	if connReq.Addr == nil {
 		return
 	}
 
-	host, portStr, err := net.SplitHostPort(msg.connReq.Addr.String())
+	host, portStr, err := net.SplitHostPort(connReq.Addr.String())
 	if err != nil {
-		srvrLog.Debugf("Cannot extract address host and port %s: %s", msg.connReq.Addr, err)
+		srvrLog.Debugf("Cannot extract address host and port %s: %s", connReq.Addr, err)
 	}
 	port, err := strconv.ParseUint(portStr, 10, 16)
 	if err != nil {
-		srvrLog.Debugf("Cannot parse port %s: %s", msg.connReq.Addr, err)
+		srvrLog.Debugf("Cannot parse port %s: %s", connReq.Addr, err)
 	}
 
 	// defaultServices is used here because Attempt makes no use
@@ -1137,12 +1126,6 @@ out:
 			})
 			s.quitWaitGroup.Done()
 			break out
-
-		case opcMsg := <-s.newOutboundConnection:
-			s.outboundPeerConnected(state, opcMsg)
-
-		case opcfMsg := <-s.newOutboundConnectionFailed:
-			s.outboundPeerConnectionFailed(opcfMsg)
 		}
 	}
 
@@ -1497,23 +1480,21 @@ func NewServer(listenAddrs []string, dagParams *dagconfig.Params, interrupt <-ch
 	maxPeers := config.ActiveConfig().TargetOutboundPeers + config.ActiveConfig().MaxInboundPeers
 
 	s := Server{
-		DAGParams:                   dagParams,
-		addrManager:                 amgr,
-		newPeers:                    make(chan *Peer, maxPeers),
-		donePeers:                   make(chan *Peer, maxPeers),
-		banPeers:                    make(chan *Peer, maxPeers),
-		Query:                       make(chan interface{}),
-		relayInv:                    make(chan relayMsg, maxPeers),
-		broadcast:                   make(chan broadcastMsg, maxPeers),
-		quit:                        make(chan struct{}),
-		modifyRebroadcastInv:        make(chan interface{}),
-		newOutboundConnection:       make(chan *outboundPeerConnectedMsg, config.ActiveConfig().TargetOutboundPeers),
-		newOutboundConnectionFailed: make(chan *outboundPeerConnectionFailedMsg, config.ActiveConfig().TargetOutboundPeers),
-		nat:                         nat,
-		TimeSource:                  blockdag.NewTimeSource(),
-		services:                    services,
-		SigCache:                    txscript.NewSigCache(config.ActiveConfig().SigCacheMaxSize),
-		notifyNewTransactions:       notifyNewTransactions,
+		DAGParams:             dagParams,
+		addrManager:           amgr,
+		newPeers:              make(chan *Peer, maxPeers),
+		donePeers:             make(chan *Peer, maxPeers),
+		banPeers:              make(chan *Peer, maxPeers),
+		Query:                 make(chan interface{}),
+		relayInv:              make(chan relayMsg, maxPeers),
+		broadcast:             make(chan broadcastMsg, maxPeers),
+		quit:                  make(chan struct{}),
+		modifyRebroadcastInv:  make(chan interface{}),
+		nat:                   nat,
+		TimeSource:            blockdag.NewTimeSource(),
+		services:              services,
+		SigCache:              txscript.NewSigCache(config.ActiveConfig().SigCacheMaxSize),
+		notifyNewTransactions: notifyNewTransactions,
 	}
 
 	// Create indexes if needed.
@@ -1576,23 +1557,14 @@ func NewServer(listenAddrs []string, dagParams *dagconfig.Params, interrupt <-ch
 
 	// Create a connection manager.
 	cmgr, err := connmgr.New(&connmgr.Config{
-		Listeners:      listeners,
-		OnAccept:       s.inboundPeerConnected,
-		RetryDuration:  connectionRetryInterval,
-		TargetOutbound: uint32(config.ActiveConfig().TargetOutboundPeers),
-		Dial:           serverutils.KaspadDial,
-		OnConnection: func(c *connmgr.ConnReq, conn net.Conn) {
-			s.newOutboundConnection <- &outboundPeerConnectedMsg{
-				connReq: c,
-				conn:    conn,
-			}
-		},
-		OnConnectionFailed: func(c *connmgr.ConnReq) {
-			s.newOutboundConnectionFailed <- &outboundPeerConnectionFailedMsg{
-				connReq: c,
-			}
-		},
-		AddrManager: s.addrManager,
+		Listeners:          listeners,
+		OnAccept:           s.inboundPeerConnected,
+		RetryDuration:      connectionRetryInterval,
+		TargetOutbound:     uint32(config.ActiveConfig().TargetOutboundPeers),
+		Dial:               serverutils.KaspadDial,
+		OnConnection:       s.outboundPeerConnected,
+		OnConnectionFailed: s.outboundPeerConnectionFailed,
+		AddrManager:        s.addrManager,
 	})
 	if err != nil {
 		return nil, err

version/version.go

@@ -11,7 +11,7 @@ const validCharacters = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrs
 const (
 	appMajor uint = 0
 	appMinor uint = 4
-	appPatch uint = 0
+	appPatch uint = 1
 )
 
 // appBuild is defined as a variable so it can be overridden during the build