Commit 836c8468 authored by Janoš Guljaš's avatar Janoš Guljaš Committed by Viktor Trón

swarm/network/master: protect SetNextBatch iterator after close (#19147)

parent b9808e39
...@@ -597,6 +597,16 @@ func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error { ...@@ -597,6 +597,16 @@ func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error {
// HandleMsg is the message handler that delegates incoming messages // HandleMsg is the message handler that delegates incoming messages
func (p *Peer) HandleMsg(ctx context.Context, msg interface{}) error { func (p *Peer) HandleMsg(ctx context.Context, msg interface{}) error {
select {
case <-p.streamer.quit:
log.Trace("message received after the streamer is closed", "peer", p.ID())
// return without an error since streamer is closed and
// no messages should be handled as other subcomponents like
// storage leveldb may be closed
return nil
default:
}
switch msg := msg.(type) { switch msg := msg.(type) {
case *SubscribeMsg: case *SubscribeMsg:
......
...@@ -107,6 +107,11 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6 ...@@ -107,6 +107,11 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6
metrics.GetOrRegisterCounter("syncer.setnextbatch.iterator", nil).Inc(1) metrics.GetOrRegisterCounter("syncer.setnextbatch.iterator", nil).Inc(1)
err := s.store.Iterator(from, to, s.po, func(key storage.Address, idx uint64) bool { err := s.store.Iterator(from, to, s.po, func(key storage.Address, idx uint64) bool {
select {
case <-s.quit:
return false
default:
}
batch = append(batch, key[:]...) batch = append(batch, key[:]...)
i++ i++
to = idx to = idx
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment