polygon/sync: blockdownloader to use common.Sleep (#10467)

Original intent of this PR was to replace 1 usage of `time.Sleep` in
`polygon/sync/block_downloader.go` with `common.Sleep` which is
context-aware.

However when doing this we discussed that common.Sleep can be improved
to return an `err` to the caller when the parent ctx has been cancelled
so that the caller is made aware of that without having to make
subsequent checks.

cmd/observer/observer/crawler.go

@@ -185,7 +185,9 @@ func (crawler *Crawler) selectCandidates(ctx context.Context, nodes chan<- candi
 		}
 
 		if len(candidates) == 0 {
-			libcommon.Sleep(ctx, 1*time.Second)
+			if err := libcommon.Sleep(ctx, 1*time.Second); err != nil {
+				return err
+			}
 		}
 
 		for _, id := range candidates {

cmd/observer/observer/diplomacy.go

@@ -82,7 +82,9 @@ func (diplomacy *Diplomacy) selectCandidates(ctx context.Context, candidatesChan
 		}
 
 		if len(candidates) == 0 {
-			libcommon.Sleep(ctx, 1*time.Second)
+			if err := libcommon.Sleep(ctx, 1*time.Second); err != nil {
+				return err
+			}
 		}
 
 		for _, id := range candidates {

cmd/observer/observer/interrogation_error.go

@@ -12,6 +12,7 @@ const (
 	InterrogationErrorKeygen
 	InterrogationErrorFindNode
 	InterrogationErrorFindNodeTimeout
+	InterrogationErrorCtxCancelled
 )
 
 type InterrogationError struct {

cmd/observer/observer/interrogator.go

@@ -88,7 +88,9 @@ func (interrogator *Interrogator) Run(ctx context.Context) (*InterrogationResult
 	// We need to wait until Server sends a Pong reply to that.
 	// The remote side is waiting for this Pong no longer than v4_udp.respTimeout.
 	// If we don't wait, the ENRRequest/FindNode might fail due to errUnknownNode.
-	libcommon.Sleep(ctx, 500*time.Millisecond)
+	if err := libcommon.Sleep(ctx, 500*time.Millisecond); err != nil {
+		return nil, NewInterrogationError(InterrogationErrorCtxCancelled, err)
+	}
 
 	// request client ID
 	var handshakeResult *DiplomatResult
@@ -158,7 +160,9 @@ func (interrogator *Interrogator) Run(ctx context.Context) (*InterrogationResult
 			peersByID[node.ID()] = node
 		}
 
-		libcommon.Sleep(ctx, 1*time.Second)
+		if err := libcommon.Sleep(ctx, 1*time.Second); err != nil {
+			return nil, NewInterrogationError(InterrogationErrorCtxCancelled, err)
+		}
 	}
 
 	peers := valuesOfIDToNodeMap(peersByID)

cmd/observer/observer/status_logger.go

@@ -17,7 +17,9 @@ func StatusLoggerLoop(ctx context.Context, db database.DB, networkID uint, perio
 	var prevDistinctIPCount uint
 
 	for ctx.Err() == nil {
-		libcommon.Sleep(ctx, period)
+		if err := libcommon.Sleep(ctx, period); err != nil {
+			break
+		}
 
 		totalCount, err := db.CountNodes(ctx, maxPingTries, networkID)
 		if err != nil {

cmd/observer/utils/retry.go

@@ -24,7 +24,9 @@ func Retry(
 	for i := 0; i <= retryCount; i++ {
 		if i > 0 {
 			logger.Trace("retrying", "op", opName, "attempt", i, "err", err)
-			libcommon.Sleep(ctx, delayForAttempt(i))
+			if err := libcommon.Sleep(ctx, delayForAttempt(i)); err != nil {
+				return nil, err
+			}
 		}
 		result, err = op(ctx)
 		if (err == nil) || !isRecoverableError(err) {

erigon-lib/common/sleep.go

@@ -5,11 +5,16 @@ import (
 	"time"
 )
 
-func Sleep(parentContext context.Context, timeout time.Duration) {
-	if timeout <= 0 {
-		return
+// Sleep which is context aware. Blocks for d duration and returns no error, unless the context
+// has been cancelled, timed out, deadline reached, etc. in which case it returns ctx.Err().
+func Sleep(ctx context.Context, d time.Duration) error {
+	timer := time.NewTimer(d)
+	defer timer.Stop()
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return nil
 	}
-	ctx, cancel := context.WithTimeout(parentContext, timeout)
-	defer cancel()
-	<-ctx.Done()
 }

erigon-lib/common/sleep_test.go

@@ -0,0 +1,27 @@
+package common
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/errgroup"
+)
+
+func TestSleep(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	start := time.Now()
+	err := Sleep(ctx, 100*time.Millisecond)
+	require.NoError(t, err)
+	require.GreaterOrEqual(t, time.Since(start), 100*time.Millisecond)
+
+	eg := errgroup.Group{}
+	eg.Go(func() error {
+		return Sleep(ctx, time.Minute)
+	})
+	cancel()
+	err = eg.Wait()
+	require.ErrorIs(t, err, context.Canceled)
+	require.Less(t, time.Since(start), time.Minute)
+}

polygon/heimdall/heimdall.go

@@ -424,13 +424,19 @@ func (h *heimdall) pollSpans(ctx context.Context, tip SpanId, cb func(*Span)) {
 				"err", err,
 			)
 
-			h.waitPollingDelay(ctx)
+			if err := common.Sleep(ctx, h.pollDelay); err != nil {
+				h.logPollerSleepCancelled(err)
+				return
+			}
 			// keep background goroutine alive in case of heimdall errors
 			continue
 		}
 
 		if latestSpan.Id <= tip {
-			h.waitPollingDelay(ctx)
+			if err := common.Sleep(ctx, h.pollDelay); err != nil {
+				h.logPollerSleepCancelled(err)
+				return
+			}
 			continue
 		}
 
@@ -441,7 +447,10 @@ func (h *heimdall) pollSpans(ctx context.Context, tip SpanId, cb func(*Span)) {
 				"err", err,
 			)
 
-			h.waitPollingDelay(ctx)
+			if err := common.Sleep(ctx, h.pollDelay); err != nil {
+				h.logPollerSleepCancelled(err)
+				return
+			}
 			// keep background goroutine alive in case of heimdall errors
 			continue
 		}
@@ -489,13 +498,19 @@ func (h *heimdall) pollMilestones(ctx context.Context, tip MilestoneId, cb func(
 				"err", err,
 			)
 
-			h.waitPollingDelay(ctx)
+			if err := common.Sleep(ctx, h.pollDelay); err != nil {
+				h.logPollerSleepCancelled(err)
+				return
+			}
 			// keep background goroutine alive in case of heimdall errors
 			continue
 		}
 
 		if count <= int64(tip) {
-			h.waitPollingDelay(ctx)
+			if err := common.Sleep(ctx, h.pollDelay); err != nil {
+				h.logPollerSleepCancelled(err)
+				return
+			}
 			continue
 		}
 
@@ -506,7 +521,10 @@ func (h *heimdall) pollMilestones(ctx context.Context, tip MilestoneId, cb func(
 				"err", err,
 			)
 
-			h.waitPollingDelay(ctx)
+			if err := common.Sleep(ctx, h.pollDelay); err != nil {
+				h.logPollerSleepCancelled(err)
+				return
+			}
 			// keep background goroutine alive in case of heimdall errors
 			continue
 		}
@@ -576,6 +594,6 @@ func (h *heimdall) batchFetchCheckpoints(
 	return checkpoints, nil
 }
 
-func (h *heimdall) waitPollingDelay(ctx context.Context) {
-	common.Sleep(ctx, h.pollDelay)
+func (h *heimdall) logPollerSleepCancelled(err error) {
+	h.logger.Info("poller sleep cancelled", "err", err)
 }

polygon/heimdall/scraper.go

@@ -64,9 +64,9 @@ func (s *Scraper[TEntity]) Run(ctx context.Context) error {
 
 		if idRange.Start > idRange.End {
 			s.syncEvent.SetAndBroadcast()
-			libcommon.Sleep(ctx, s.pollDelay)
-			if ctx.Err() != nil {
+			if err := libcommon.Sleep(ctx, s.pollDelay); err != nil {
 				s.syncEvent.Reset()
+				return err
 			}
 		} else {
 			entities, err := s.fetcher.FetchEntitiesRange(ctx, idRange)

polygon/sync/block_downloader.go

@@ -158,7 +158,10 @@ func (d *blockDownloader) downloadBlocksUsingWaypoints(
 				"sleepSeconds", d.notEnoughPeersBackOffDuration.Seconds(),
 			)
 
-			time.Sleep(d.notEnoughPeersBackOffDuration)
+			if err := common.Sleep(ctx, d.notEnoughPeersBackOffDuration); err != nil {
+				return nil, err
+			}
+
 			continue
 		}