fix: speed up node shutdown

pkg/api/api.go

@@ -174,8 +174,9 @@ type Service struct {
 
 	metrics metrics
 
-	wsWg sync.WaitGroup // wait for all websockets to close on exit
-	quit chan struct{}
+	wsWg      sync.WaitGroup // wait for all websockets to close on exit
+	closeOnce sync.Once
+	quit      chan struct{}
 
 	overlay           *swarm.Address
 	publicKey         ecdsa.PublicKey
@@ -398,7 +399,9 @@ func (s *Service) SetIsWarmingUp(v bool) {
 // Close hangs up running websockets on shutdown.
 func (s *Service) Close() error {
 	s.logger.Info("api shutting down")
-	close(s.quit)
+	s.closeOnce.Do(func() {
+		close(s.quit)
+	})
 
 	done := make(chan struct{})
 	go func() {

pkg/blocker/blocker.go

@@ -37,6 +37,7 @@ type Blocker struct {
 	logger            log.Logger
 	wakeupCh          chan struct{}
 	quit              chan struct{}
+	closeOnce         sync.Once
 	closeWg           sync.WaitGroup
 	blocklistCallback func(swarm.Address)
 }
@@ -154,9 +155,10 @@ func (b *Blocker) PruneUnseen(seen []swarm.Address) {
 }
 
 // Close will exit the worker loop.
-// must be called only once.
 func (b *Blocker) Close() error {
-	close(b.quit)
+	b.closeOnce.Do(func() {
+		close(b.quit)
+	})
 	b.closeWg.Wait()
 	return nil
 }

pkg/gsoc/gsoc.go

@@ -25,6 +25,7 @@ type Listener interface {
 type listener struct {
 	handlers   map[string][]*Handler
 	handlersMu sync.Mutex
+	closeOnce  sync.Once
 	quit       chan struct{}
 	logger     log.Logger
 }
@@ -86,7 +87,9 @@ func (p *listener) getHandlers(address swarm.Address) []*Handler {
 }
 
 func (l *listener) Close() error {
-	close(l.quit)
+	l.closeOnce.Do(func() {
+		close(l.quit)
+	})
 	l.handlersMu.Lock()
 	defer l.handlersMu.Unlock()
 

pkg/hive/hive.go

@@ -58,6 +58,7 @@ type Service struct {
 	metrics           metrics
 	inLimiter         *ratelimit.Limiter
 	outLimiter        *ratelimit.Limiter
+	closeOnce         sync.Once
 	quit              chan struct{}
 	wg                sync.WaitGroup
 	peersChan         chan pb.Peers
@@ -144,7 +145,9 @@ func (s *Service) SetAddPeersHandler(h func(addr ...swarm.Address)) {
 }
 
 func (s *Service) Close() error {
-	close(s.quit)
+	s.closeOnce.Do(func() {
+		close(s.quit)
+	})
 
 	stopped := make(chan struct{})
 	go func() {

pkg/node/node.go

@@ -90,6 +90,7 @@ import (
 const LoggerName = "node"
 
 type Bee struct {
+	logger                   log.Logger
 	p2pService               io.Closer
 	p2pHalter                p2p.Halter
 	ctxCancel                context.CancelFunc
@@ -260,6 +261,7 @@ func NewBee(
 	})
 
 	b = &Bee{
+		logger:         logger,
 		ctxCancel:      ctxCancel,
 		errorLogWriter: sink,
 		tracerCloser:   tracerCloser,
@@ -1350,12 +1352,15 @@ func (b *Bee) Shutdown() error {
 	}
 	// tryClose is a convenient closure which decrease
 	// repetitive io.Closer tryClose procedure.
-	tryClose := func(c io.Closer, errMsg string) {
+	tryClose := func(c io.Closer, component string) {
 		if c == nil {
 			return
 		}
+
+		b.logger.Debug("starting shutdown", "component", component)
+		defer b.logger.Debug("finished shutdown", "component", component)
 		if err := c.Close(); err != nil {
-			mErr = multierror.Append(mErr, fmt.Errorf("%s: %w", errMsg, err))
+			mErr = multierror.Append(mErr, fmt.Errorf("%s: %w", component, err))
 		}
 	}
 
@@ -1429,9 +1434,10 @@ func (b *Bee) Shutdown() error {
 	tryClose(b.tracerCloser, "tracer")
 	tryClose(b.topologyCloser, "topology driver")
 	tryClose(b.storageIncetivesCloser, "storage incentives agent")
+	// close localstore before StateStore to avoid ErrClosed / incomplete flush.
+	tryClose(b.localstoreCloser, "localstore")
 	tryClose(b.stateStoreCloser, "statestore")
 	tryClose(b.stamperStoreCloser, "stamperstore")
-	tryClose(b.localstoreCloser, "localstore")
 	tryClose(b.resolverCloser, "resolver service")
 
 	return mErr

pkg/p2p/libp2p/internal/reacher/reacher.go

@@ -43,8 +43,9 @@ type reacher struct {
 	peerHeap  peerHeap         // min-heap ordered by retryAfter
 	peerIndex map[string]*peer // lookup by overlay for O(1) access
 
-	newPeer chan struct{}
-	quit    chan struct{}
+	newPeer   chan struct{}
+	quit      chan struct{}
+	closeOnce sync.Once
 
 	pinger   p2p.Pinger
 	notifier p2p.ReachableNotifier
@@ -290,15 +291,11 @@ func (r *reacher) jitter(d time.Duration) time.Duration {
 	return time.Duration(float64(d) * j)
 }
 
-// Close stops the worker. Must be called once.
+// Close stops the worker.
 func (r *reacher) Close() error {
-	select {
-	case <-r.quit:
-		return nil
-	default:
-	}
-
-	close(r.quit)
+	r.closeOnce.Do(func() {
+		close(r.quit)
+	})
 	r.wg.Wait()
 	return nil
 }

pkg/p2p/libp2p/libp2p.go

@@ -1019,7 +1019,14 @@ func (s *Service) Connect(ctx context.Context, addrs []ma.Multiaddr) (address *b
 				s.metrics.ConnectBreakerCount.Inc()
 				return nil, p2p.NewConnectionBackoffError(err, s.connectionBreaker.ClosedUntil())
 			}
-			s.logger.Warning("libp2p connect", "peer_id", peerID, "underlay", info.Addrs, "error", err)
+			if !errors.Is(err, context.Canceled) {
+				select {
+				case <-s.halt:
+					s.logger.Debug("libp2p connect", "peer_id", peerID, "underlay", info.Addrs, "error", err)
+				default:
+					s.logger.Warning("libp2p connect", "peer_id", peerID, "underlay", info.Addrs, "error", err)
+				}
+			}
 			connectErr = err
 			continue
 		}

pkg/postage/listener/listener.go

@@ -57,6 +57,7 @@ type listener struct {
 
 	postageStampContractAddress common.Address
 	postageStampContractABI     abi.ABI
+	closeOnce                   sync.Once
 	quit                        chan struct{}
 	wg                          sync.WaitGroup
 	metrics                     metrics
@@ -374,7 +375,9 @@ func (l *listener) Listen(ctx context.Context, from uint64, updater postage.Even
 }
 
 func (l *listener) Close() error {
-	close(l.quit)
+	l.closeOnce.Do(func() {
+		close(l.quit)
+	})
 
 	done := make(chan struct{})
 	go func() {

pkg/pss/pss.go

@@ -55,6 +55,7 @@ type pss struct {
 	handlersMu sync.Mutex
 	metrics    metrics
 	logger     log.Logger
+	closeOnce  sync.Once
 	quit       chan struct{}
 }
 
@@ -70,7 +71,9 @@ func New(key *ecdsa.PrivateKey, logger log.Logger) Interface {
 }
 
 func (ps *pss) Close() error {
-	close(ps.quit)
+	ps.closeOnce.Do(func() {
+		close(ps.quit)
+	})
 	ps.handlersMu.Lock()
 	defer ps.handlersMu.Unlock()
 

pkg/pullsync/pullsync.go

@@ -13,6 +13,7 @@ import (
 	"fmt"
 	"io"
 	"math"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -66,6 +67,7 @@ type Syncer struct {
 	metrics        metrics
 	logger         log.Logger
 	store          storer.Reserve
+	closeOnce      sync.Once
 	quit           chan struct{}
 	unwrap         func(swarm.Chunk)
 	gsocHandler    func(*soc.SOC)
@@ -193,7 +195,7 @@ func (s *Syncer) handler(streamCtx context.Context, p p2p.Peer, stream p2p.Strea
 	}
 
 	// slow down future requests
-	waitDur, err := s.limiter.Wait(streamCtx, p.Address.ByteString(), max(1, len(chs)))
+	waitDur, err := s.limiter.Wait(ctx, p.Address.ByteString(), max(1, len(chs)))
 	if err != nil {
 		return fmt.Errorf("rate limiter: %w", err)
 	}
@@ -568,7 +570,9 @@ func (s *Syncer) disconnect(peer p2p.Peer) error {
 
 func (s *Syncer) Close() error {
 	s.logger.Info("pull syncer shutting down")
-	close(s.quit)
+	s.closeOnce.Do(func() {
+		close(s.quit)
+	})
 	cc := make(chan struct{})
 	go func() {
 		defer close(cc)

pkg/pusher/pusher.go

@@ -55,6 +55,7 @@ type Service struct {
 	batchExist        postage.BatchExist
 	logger            log.Logger
 	metrics           metrics
+	closeOnce         sync.Once
 	quit              chan struct{}
 	chunksWorkerQuitC chan struct{}
 	inflight          *inflight
@@ -363,7 +364,9 @@ func (s *Service) AddFeed(c <-chan *Op) {
 
 func (s *Service) Close() error {
 	s.logger.Info("pusher shutting down")
-	close(s.quit)
+	s.closeOnce.Do(func() {
+		close(s.quit)
+	})
 
 	// Wait for chunks worker to finish
 	select {

pkg/salud/salud.go

@@ -44,6 +44,7 @@ type peerStatus interface {
 
 type service struct {
 	wg            sync.WaitGroup
+	closeOnce     sync.Once
 	quit          chan struct{}
 	logger        log.Logger
 	topology      topologyDriver
@@ -116,7 +117,9 @@ func (s *service) worker(startupStabilizer stabilization.Subscriber, mode string
 }
 
 func (s *service) Close() error {
-	close(s.quit)
+	s.closeOnce.Do(func() {
+		close(s.quit)
+	})
 	s.wg.Wait()
 	return nil
 }

pkg/settlement/swap/priceoracle/priceoracle.go

@@ -9,6 +9,7 @@ import (
 	"errors"
 	"io"
 	"math/big"
+	"sync"
 	"time"
 
 	"github.com/ethereum/go-ethereum/accounts/abi"
@@ -33,6 +34,7 @@ type service struct {
 	exchangeRate       *big.Int
 	deduction          *big.Int
 	timeDivisor        int64
+	closeOnce          sync.Once
 	quitC              chan struct{}
 }
 
@@ -148,6 +150,8 @@ func (s *service) CurrentRates() (exchangeRate, deduction *big.Int, err error) {
 }
 
 func (s *service) Close() error {
-	close(s.quitC)
+	s.closeOnce.Do(func() {
+		close(s.quitC)
+	})
 	return nil
 }

pkg/sharky/store.go

@@ -32,7 +32,8 @@ type Store struct {
 	writes      chan write      // shared write operations channel
 	shards      []*shard        // shards
 	wg          *sync.WaitGroup // count started operations
-	quit        chan struct{}   // quit channel
+	closeOnce   sync.Once
+	quit        chan struct{} // quit channel
 	metrics     metrics
 }
 
@@ -65,7 +66,9 @@ func New(basedir fs.FS, shardCnt int, maxDataSize int) (*Store, error) {
 
 // Close closes each shard and return incidental errors from each shard
 func (s *Store) Close() error {
-	close(s.quit)
+	s.closeOnce.Do(func() {
+		close(s.quit)
+	})
 	err := new(multierror.Error)
 	for _, sh := range s.shards {
 		err = multierror.Append(err, sh.close())

pkg/shed/db.go

@@ -24,6 +24,7 @@ package shed
 
 import (
 	"errors"
+	"sync"
 
 	"github.com/syndtr/goleveldb/leveldb"
 	"github.com/syndtr/goleveldb/leveldb/iterator"
@@ -51,9 +52,10 @@ type Options struct {
 // It provides a schema functionality to store fields and indexes
 // information about naming and types.
 type DB struct {
-	ldb     *leveldb.DB
-	metrics metrics
-	quit    chan struct{} // Quit channel to stop the metrics collection before closing the database
+	ldb       *leveldb.DB
+	metrics   metrics
+	closeOnce sync.Once
+	quit      chan struct{} // Quit channel to stop the metrics collection before closing the database
 }
 
 // NewDB constructs a new DB and validates the schema
@@ -193,6 +195,8 @@ func (db *DB) Compact(start, end []byte) error {
 
 // Close closes LevelDB database.
 func (db *DB) Close() (err error) {
-	close(db.quit)
+	db.closeOnce.Do(func() {
+		close(db.quit)
+	})
 	return db.ldb.Close()
 }

pkg/skippeers/skippeers.go

@@ -17,8 +17,9 @@ const maxDuration time.Duration = math.MaxInt64
 type List struct {
 	mtx sync.Mutex
 
-	durC chan time.Duration
-	quit chan struct{}
+	closeOnce sync.Once
+	durC      chan time.Duration
+	quit      chan struct{}
 	// key is chunk address, value is map of peer address to expiration
 	skip map[string]map[string]int64
 
@@ -133,7 +134,9 @@ func (l *List) pruneChunk(ch string, now int64) int {
 }
 
 func (l *List) Close() error {
-	close(l.quit)
+	l.closeOnce.Do(func() {
+		close(l.quit)
+	})
 	l.wg.Wait()
 
 	l.mtx.Lock()