feat: add safe package for unified goroutine recovery

pkg/api/pin.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/ethersphere/bee/v2/pkg/file/redundancy"
 	"github.com/ethersphere/bee/v2/pkg/jsonhttp"
+	"github.com/ethersphere/bee/v2/pkg/safe"
 	"github.com/ethersphere/bee/v2/pkg/storage"
 	"github.com/ethersphere/bee/v2/pkg/storer"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
@@ -238,7 +239,9 @@ func (s *Service) pinIntegrityHandler(w http.ResponseWriter, r *http.Request) {
 
 	out := make(chan storer.PinStat)
 
-	go s.pinIntegrity.Check(r.Context(), logger, querie.Ref.String(), out)
+	safe.Go(logger, "pin-integrity-check", func() {
+		s.pinIntegrity.Check(r.Context(), logger, querie.Ref.String(), out)
+	})
 
 	flusher, ok := w.(http.Flusher)
 	if !ok {

pkg/blocker/blocker.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/ethersphere/bee/v2/pkg/log"
 	"github.com/ethersphere/bee/v2/pkg/p2p"
+	"github.com/ethersphere/bee/v2/pkg/safe"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
 	"go.uber.org/atomic"
 )
@@ -62,27 +63,31 @@ func New(blocklister p2p.Blocklister, flagTimeout, blockDuration, wakeUpTime tim
 	}
 
 	b.closeWg.Go(func() {
-		for {
-			select {
-			case <-b.quit:
-				return
-			case <-time.After(sequencerResolution):
-				if b.blocklister.NetworkStatus() == p2p.NetworkStatusAvailable {
-					b.sequence.Inc()
+		safe.Run(b.logger, "blocker-network-status-sequencer", func() {
+			for {
+				select {
+				case <-b.quit:
+					return
+				case <-time.After(sequencerResolution):
+					if b.blocklister.NetworkStatus() == p2p.NetworkStatusAvailable {
+						b.sequence.Inc()
+					}
 				}
 			}
-		}
+		})
 	})
 
 	b.closeWg.Go(func() {
-		for {
-			select {
-			case <-time.After(wakeUpTime):
-				b.block()
-			case <-b.quit:
-				return
+		safe.Run(b.logger, "blocker-blacklist-worker", func() {
+			for {
+				select {
+				case <-time.After(wakeUpTime):
+					b.block()
+				case <-b.quit:
+					return
+				}
 			}
-		}
+		})
 	})
 
 	return b

pkg/file/joiner/joiner.go

@@ -19,6 +19,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/file/redundancy"
 	"github.com/ethersphere/bee/v2/pkg/file/redundancy/getter"
 	"github.com/ethersphere/bee/v2/pkg/replicas"
+	"github.com/ethersphere/bee/v2/pkg/safe"
 	"github.com/ethersphere/bee/v2/pkg/storage"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
 	"golang.org/x/sync/errgroup"
@@ -279,7 +280,7 @@ func (j *joiner) readAtOffset(
 		currentReadSize = min(currentReadSize, subtrieSpan)
 
 		func(address swarm.Address, b []byte, cur, subTrieSize, off, bufferOffset, bytesToRead, subtrieSpanLimit int64) {
-			eg.Go(func() error {
+			eg.Go(safe.RunFunc(nil, "joiner-read-at-offset", func() error {
 				ch, err := g.Get(j.ctx, addr)
 				if err != nil {
 					return err
@@ -295,7 +296,7 @@ func (j *joiner) readAtOffset(
 
 				j.readAtOffset(b, chunkData, cur, subtrieSpan, off, bufferOffset, currentReadSize, bytesRead, subtrieParity, eg)
 				return nil
-			})
+			}))
 		}(addr, b, cur, subtrieSpan, off, bufferOffset, currentReadSize, subtrieSpanLimit)
 
 		bufferOffset += currentReadSize

pkg/hive/hive.go

@@ -25,6 +25,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/p2p"
 	"github.com/ethersphere/bee/v2/pkg/p2p/protobuf"
 	"github.com/ethersphere/bee/v2/pkg/ratelimit"
+	"github.com/ethersphere/bee/v2/pkg/safe"
 	"github.com/ethersphere/bee/v2/pkg/settlement/swap/chequebook"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
 	ma "github.com/multiformats/go-multiaddr"
@@ -313,7 +314,9 @@ func (s *Service) startCheckPeersHandler() {
 				return
 			case newPeers := <-s.peersChan:
 				s.wg.Go(func() {
-					s.checkAndAddPeers(ctx, newPeers)
+					safe.Run(s.logger, "hive-check-and-add-peers", func() {
+						s.checkAndAddPeers(ctx, newPeers)
+					})
 				})
 			}
 		}

pkg/postage/listener/listener.go

@@ -19,6 +19,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/log"
 	"github.com/ethersphere/bee/v2/pkg/postage"
 	"github.com/ethersphere/bee/v2/pkg/postage/batchservice"
+	"github.com/ethersphere/bee/v2/pkg/safe"
 	"github.com/ethersphere/bee/v2/pkg/transaction"
 	"github.com/ethersphere/bee/v2/pkg/util/syncutil"
 	"github.com/prometheus/client_golang/prometheus"
@@ -250,7 +251,7 @@ func (l *listener) Listen(ctx context.Context, from uint64, updater postage.Even
 	lastConfirmedBlock := uint64(0)
 
 	l.wg.Add(1)
-	listenf := func() error {
+	listenf := safe.RunFunc(l.logger, "postage-listener-func", func() error {
 		defer l.wg.Done()
 		for {
 			// if for whatever reason we are stuck for too long we terminate
@@ -350,9 +351,9 @@ func (l *listener) Listen(ctx context.Context, from uint64, updater postage.Even
 			totalTimeMetric(l.metrics.PageProcessDuration, start)
 			l.metrics.PagesProcessed.Inc()
 		}
-	}
+	})
 
-	go func() {
+	safe.Go(l.logger, "postage-listener", func() {
 		err := listenf()
 		if err != nil {
 			if errors.Is(err, context.Canceled) {
@@ -366,7 +367,7 @@ func (l *listener) Listen(ctx context.Context, from uint64, updater postage.Even
 		if l.syncingStopped != nil {
 			l.syncingStopped.Signal() // trigger shutdown in start.go
 		}
-	}()
+	})
 
 	return synced
 }

pkg/postage/service.go

@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"github.com/ethersphere/bee/v2/pkg/log"
+	"github.com/ethersphere/bee/v2/pkg/safe"
 	"github.com/ethersphere/bee/v2/pkg/storage"
 )
 
@@ -99,7 +100,7 @@ func NewService(logger log.Logger, store storage.Store, postageStore Storer, cha
 		}
 	}
 
-	go s.run()
+	safe.Go(s.logger, "postage-service", s.run)
 
 	return s, nil
 }

pkg/pss/pss.go

@@ -20,6 +20,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/log"
 	"github.com/ethersphere/bee/v2/pkg/postage"
 	"github.com/ethersphere/bee/v2/pkg/pushsync"
+	"github.com/ethersphere/bee/v2/pkg/safe"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
 	"github.com/ethersphere/bee/v2/pkg/topology"
 )
@@ -180,7 +181,9 @@ func (p *pss) TryUnwrap(c swarm.Chunk) {
 		wg.Add(1)
 		go func(hh Handler) {
 			defer wg.Done()
-			hh(ctx, msg)
+			safe.Run(p.logger, "pss-handler", func() {
+				hh(ctx, msg)
+			})
 		}(*hh)
 	}
 	go func() {

pkg/puller/puller.go

@@ -21,6 +21,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/puller/intervalstore"
 	"github.com/ethersphere/bee/v2/pkg/pullsync"
 	"github.com/ethersphere/bee/v2/pkg/rate"
+	"github.com/ethersphere/bee/v2/pkg/safe"
 	"github.com/ethersphere/bee/v2/pkg/storage"
 	"github.com/ethersphere/bee/v2/pkg/storer"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
@@ -149,7 +150,9 @@ func (p *Puller) Start(ctx context.Context) {
 		p.cancel = cancel
 
 		p.wg.Add(1)
-		go p.manage(cctx)
+		safe.Go(p.logger, "puller-manage", func() {
+			p.manage(cctx)
+		})
 	})
 }
 
@@ -240,13 +243,13 @@ func (p *Puller) recalcPeers(ctx context.Context, storageRadius uint8) {
 	for _, peer := range p.syncPeers {
 		wg.Add(1)
 		p.wg.Add(1)
-		go func(peer *syncPeer) {
+		safe.Go(p.logger, "puller-recalc-peers", func() {
 			defer p.wg.Done()
 			defer wg.Done()
 			if err := p.syncPeer(ctx, peer, storageRadius); err != nil {
 				p.logger.Debug("sync peer failed", "peer_address", peer.address, "error", err)
 			}
-		}(peer)
+		})
 	}
 	wg.Wait()
 }
@@ -409,12 +412,16 @@ func (p *Puller) syncPeerBin(parentCtx context.Context, peer *syncPeer, bin uint
 	if cursor > 0 {
 		peer.wg.Add(1)
 		p.wg.Add(1)
-		go sync(true, peer.address, cursor)
+		safe.Go(p.logger, "puller-sync-historical", func() {
+			sync(true, peer.address, cursor)
+		})
 	}
 
 	peer.wg.Add(1)
 	p.wg.Add(1)
-	go sync(false, peer.address, cursor+1)
+	safe.Go(p.logger, "puller-sync-live", func() {
+		sync(false, peer.address, cursor+1)
+	})
 }
 
 func (p *Puller) Close() error {

pkg/pusher/pusher.go

@@ -18,6 +18,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/log"
 	"github.com/ethersphere/bee/v2/pkg/postage"
 	"github.com/ethersphere/bee/v2/pkg/pushsync"
+	"github.com/ethersphere/bee/v2/pkg/safe"
 	"github.com/ethersphere/bee/v2/pkg/stabilization"
 	storage "github.com/ethersphere/bee/v2/pkg/storage"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
@@ -89,7 +90,9 @@ func New(
 		attempts:          &attempts{retryCount: retryCount, attempts: make(map[string]int)},
 		smuggler:          make(chan OpChan),
 	}
-	go p.chunksWorker(startupStabilizer)
+	safe.Go(p.logger, "pusher-chunks-worker", func() {
+		p.chunksWorker(startupStabilizer)
+	})
 	return p
 }
 
@@ -178,7 +181,7 @@ func (s *Service) chunksWorker(startupStabilizer stabilization.Subscriber) {
 		s.metrics.TotalSynced.Inc()
 	}
 
-	go func() {
+	safe.Go(s.logger, "pusher-chunks-multiplexer", func() {
 		for {
 			select {
 			case ch, ok := <-chunks:
@@ -192,7 +195,7 @@ func (s *Service) chunksWorker(startupStabilizer stabilization.Subscriber) {
 					return
 				}
 			case apiC := <-s.smuggler:
-				go func() {
+				safe.Go(s.logger, "pusher-smuggler-worker", func() {
 					for {
 						select {
 						case op := <-apiC:
@@ -205,12 +208,12 @@ func (s *Service) chunksWorker(startupStabilizer stabilization.Subscriber) {
 							return
 						}
 					}
-				}()
+				})
 			case <-s.quit:
 				return
 			}
 		}
-	}()
+	})
 
 	defer wg.Wait()
 
@@ -236,7 +239,9 @@ func (s *Service) chunksWorker(startupStabilizer stabilization.Subscriber) {
 			select {
 			case sem <- struct{}{}:
 				wg.Add(1)
-				go push(op)
+				safe.Go(s.logger, "pusher-push-worker", func() {
+					push(op)
+				})
 			case <-s.quit:
 				return
 			}

pkg/pushsync/pushsync.go

@@ -22,6 +22,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/postage"
 	"github.com/ethersphere/bee/v2/pkg/pricer"
 	"github.com/ethersphere/bee/v2/pkg/pushsync/pb"
+	"github.com/ethersphere/bee/v2/pkg/safe"
 	"github.com/ethersphere/bee/v2/pkg/skippeers"
 	"github.com/ethersphere/bee/v2/pkg/soc"
 	"github.com/ethersphere/bee/v2/pkg/stabilization"
@@ -242,7 +243,9 @@ func (ps *PushSync) handler(ctx context.Context, p p2p.Peer, stream p2p.Stream)
 	chunk.WithStamp(stamp)
 
 	if cac.Valid(chunk) {
-		go ps.unwrap(chunk)
+		safe.Go(ps.logger, "pushsync-unwrap-chunk", func() {
+			ps.unwrap(chunk)
+		})
 	} else if chunk, err := soc.FromChunk(chunk); err == nil {
 		addr, err := chunk.Address()
 		if err != nil {
@@ -424,7 +427,9 @@ func (ps *PushSync) pushToClosest(ctx context.Context, ch swarm.Chunk, origin bo
 					if inflight == 0 {
 						if ps.fullNode {
 							if cac.Valid(ch) {
-								go ps.unwrap(ch)
+								safe.Go(ps.logger, "pushsync-unwrap-ch", func() {
+									ps.unwrap(ch)
+								})
 							}
 							return nil, topology.ErrWantSelf
 						}
@@ -477,7 +482,9 @@ func (ps *PushSync) pushToClosest(ctx context.Context, ch swarm.Chunk, origin bo
 			ps.metrics.TotalSendAttempts.Inc()
 			inflight++
 
-			go ps.push(ctx, resultChan, peer, ch, action)
+			safe.Go(ps.logger, "pushsync-push", func() {
+				ps.push(ctx, resultChan, peer, ch, action)
+			})
 
 		case result := <-resultChan:
 			inflight--