fix: add logs and metrics around pusher retries

pkg/pusher/metrics.go

@@ -17,7 +17,8 @@ type metrics struct {
 	SyncTime         prometheus.Histogram
 	ErrorTime        prometheus.Histogram
 
-	ReceiptDepth *prometheus.CounterVec
+	ReceiptDepth   *prometheus.CounterVec
+	ShallowRetries prometheus.Counter
 }
 
 func newMetrics() metrics {
@@ -72,6 +73,12 @@ func newMetrics() metrics {
 			},
 			[]string{"depth"},
 		),
+		ShallowRetries: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "shallow_retries",
+			Help:      "Chunks retried because receipts were too shallow",
+		}),
 	}
 }
 

pkg/pusher/pusher.go

@@ -34,6 +34,7 @@ type Service struct {
 	networkID         uint64
 	storer            storage.Storer
 	pushSyncer        pushsync.PushSyncer
+	depther           topology.NeighborhoodDepther
 	logger            logging.Logger
 	tag               *tags.Tags
 	tracer            *tracing.Tracer
@@ -45,15 +46,20 @@ type Service struct {
 var (
 	retryInterval  = 5 * time.Second // time interval between retries
 	concurrentJobs = 10              // how many chunks to push simultaneously
+	retryCount     = 3
 )
 
-var ErrInvalidAddress = errors.New("invalid address")
+var (
+	ErrInvalidAddress = errors.New("invalid address")
+	ErrShallowReceipt = errors.New("shallow recipt")
+)
 
-func New(networkID uint64, storer storage.Storer, peerSuggester topology.ClosestPeerer, pushSyncer pushsync.PushSyncer, tagger *tags.Tags, logger logging.Logger, tracer *tracing.Tracer) *Service {
+func New(networkID uint64, storer storage.Storer, depther topology.NeighborhoodDepther, pushSyncer pushsync.PushSyncer, tagger *tags.Tags, logger logging.Logger, tracer *tracing.Tracer) *Service {
 	service := &Service{
 		networkID:         networkID,
 		storer:            storer,
 		pushSyncer:        pushSyncer,
+		depther:           depther,
 		tag:               tagger,
 		logger:            logger,
 		tracer:            tracer,
@@ -80,6 +86,7 @@ func (s *Service) chunksWorker() {
 		mtx           sync.Mutex
 		span          opentracing.Span
 		logger        *logrus.Entry
+		retryCounter  = make(map[string]int)
 	)
 	defer timer.Stop()
 	defer close(s.chunksWorkerQuitC)
@@ -148,9 +155,10 @@ LOOP:
 						s.metrics.TotalSynced.Inc()
 						s.metrics.SyncTime.Observe(time.Since(startTime).Seconds())
 						// only print this if there was no error while sending the chunk
-						logger.Tracef("pusher: pushed chunk %s to node %s", ch.Address().String(), storerPeer.String())
 						po := swarm.Proximity(ch.Address().Bytes(), storerPeer.Bytes())
+						logger.Tracef("pusher: pushed chunk %s to node %s with receipt depth %d", ch.Address().String(), storerPeer.String(), po)
 						s.metrics.ReceiptDepth.WithLabelValues(strconv.Itoa(int(po))).Inc()
+						delete(retryCounter, ch.Address().ByteString())
 					} else {
 						s.metrics.TotalErrors.Inc()
 						s.metrics.ErrorTime.Observe(time.Since(startTime).Seconds())
@@ -190,6 +198,22 @@ LOOP:
 						err = fmt.Errorf("pusher: receipt storer address: %w", err)
 						return
 					}
+
+					po := swarm.Proximity(ch.Address().Bytes(), storerPeer.Bytes())
+					d := s.depther.NeighborhoodDepth()
+					if po < d {
+						mtx.Lock()
+						if retryCounter[ch.Address().ByteString()] < retryCount {
+							retryCounter[ch.Address().ByteString()]++
+							mtx.Unlock()
+							po := swarm.Proximity(ch.Address().Bytes(), storerPeer.Bytes())
+							// QUESTION: where is this error logged?
+							err = fmt.Errorf("pusher: reshallow receipt depth %d from chunk %s, want at least %d", po, ch.Address().String(), d)
+							s.metrics.ShallowRetries.Inc()
+							return
+						}
+						mtx.Unlock()
+					}
 				}
 
 				if err = s.storer.Set(ctx, storage.ModeSetSync, ch.Address()); err != nil {

pkg/topology/topology.go

@@ -25,7 +25,7 @@ type Driver interface {
 	ClosestPeerer
 	EachPeerer
 	EachNeighbor
-	NeighborhoodDepth() uint8
+	NeighborhoodDepther
 	SubscribePeersChange() (c <-chan struct{}, unsubscribe func())
 	io.Closer
 	Halter
@@ -137,3 +137,7 @@ type Halter interface {
 	// while allowing it to still run.
 	Halt()
 }
+
+type NeighborhoodDepther interface {
+	NeighborhoodDepth() uint8
+}