diff --git a/eth/handler.go b/eth/handler.go index 58869a2eec..1e4dc1289e 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -59,7 +59,9 @@ type blockFetcherFn func([]common.Hash) error type ProtocolManager struct { networkId int - fastSync uint32 + fastSync uint32 // Flag whether fast sync is enabled (gets disabled if we already have blocks) + synced uint32 // Flag whether we're considered synchronised (enables transaction processing) + txpool txPool blockchain *core.BlockChain chaindb ethdb.Database @@ -161,7 +163,11 @@ func NewProtocolManager(config *core.ChainConfig, fastSync bool, networkId int, heighter := func() uint64 { return blockchain.CurrentBlock().NumberU64() } - manager.fetcher = fetcher.New(blockchain.GetBlock, validator, manager.BroadcastBlock, heighter, manager.insertChain, manager.removePeer) + inserter := func(blocks types.Blocks) (int, error) { + atomic.StoreUint32(&manager.synced, 1) // Mark initial sync done on any fetcher import + return manager.insertChain(blocks) + } + manager.fetcher = fetcher.New(blockchain.GetBlock, validator, manager.BroadcastBlock, heighter, inserter, manager.removePeer) if blockchain.Genesis().Hash().Hex() == defaultGenesisHash && networkId == 1 { glog.V(logger.Debug).Infoln("Bad Block Reporting is enabled") @@ -698,8 +704,8 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } case msg.Code == TxMsg: - // Transactions arrived, make sure we have a valid chain to handle them - if atomic.LoadUint32(&pm.fastSync) == 1 { + // Transactions arrived, make sure we have a valid and fresh chain to handle them + if atomic.LoadUint32(&pm.synced) == 0 { break } // Transactions can be processed, parse all of them and deliver to the pool diff --git a/eth/protocol_test.go b/eth/protocol_test.go index 0a82e2e795..f860d0a35a 100644 --- a/eth/protocol_test.go +++ b/eth/protocol_test.go @@ -97,6 +97,7 @@ func TestRecvTransactions63(t *testing.T) { testRecvTransactions(t, 63) } func testRecvTransactions(t *testing.T, protocol int) { txAdded := make(chan []*types.Transaction) pm := newTestProtocolManagerMust(t, false, 0, nil, txAdded) + pm.synced = 1 // mark synced to accept transactions p, _ := newTestPeer("peer", protocol, pm, true) defer pm.Stop() defer p.close() diff --git a/eth/sync.go b/eth/sync.go index 4b16c13226..52f7e90e7b 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -174,6 +174,8 @@ func (pm *ProtocolManager) synchronise(peer *peer) { if err := pm.downloader.Synchronise(peer.id, peer.Head(), peer.Td(), mode); err != nil { return } + atomic.StoreUint32(&pm.synced, 1) // Mark initial sync done + // If fast sync was enabled, and we synced up, disable it if atomic.LoadUint32(&pm.fastSync) == 1 { // Disable fast sync if we indeed have something in our chain