fix(pullsync): swallow and log process want reserve get errs

pkg/pullsync/metrics.go

@@ -12,6 +12,7 @@ import (
 type metrics struct {
 	Offered       prometheus.Counter     // number of chunks offered
 	Wanted        prometheus.Counter     // number of chunks wanted
+	MissingChunks prometheus.Counter     // number of reserve get errs
 	Delivered     prometheus.Counter     // number of chunk deliveries
 	Sent          prometheus.Counter     // number of chunks sent
 	DuplicateRuid prometheus.Counter     // number of duplicate RUID requests we got
@@ -34,6 +35,12 @@ func newMetrics() metrics {
 			Name:      "chunks_wanted",
 			Help:      "Total chunks wanted.",
 		}),
+		MissingChunks: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "missing_chunks",
+			Help:      "Total reserve get errors.",
+		}),
 		Delivered: prometheus.NewCounter(prometheus.CounterOpts{
 			Namespace: m.Namespace,
 			Subsystem: subsystem,

pkg/pullsync/pullsync.go

@@ -215,6 +215,11 @@ func (s *Syncer) Sync(ctx context.Context, peer swarm.Address, bin uint8, start
 		}
 
 		addr := swarm.NewAddress(delivery.Address)
+		if addr.Equal(swarm.ZeroAddress) {
+			s.logger.Debug("received zero address chunk", "peer_address", peer)
+			continue
+		}
+
 		newChunk := swarm.NewChunk(addr, delivery.Data)
 
 		stamp := new(postage.Stamp)
@@ -343,18 +348,19 @@ func (s *Syncer) handler(streamCtx context.Context, p p2p.Peer, stream p2p.Strea
 
 	chs, err := s.processWant(ctx, offer, &want)
 	if err != nil {
-		if errors.Is(err, storage.ErrNotFound) {
-			s.intervalsSF.Forget(sfKey(uint8(rn.Bin), rn.Start))
-		}
 		return fmt.Errorf("process want: %w", err)
 	}
 
-	for _, v := range chs {
-		stamp, err := v.Stamp().MarshalBinary()
-		if err != nil {
-			return fmt.Errorf("serialise stamp: %w", err)
+	for _, c := range chs {
+		var stamp []byte
+		if c.Stamp() != nil {
+			stamp, err = c.Stamp().MarshalBinary()
+			if err != nil {
+				return fmt.Errorf("serialise stamp: %w", err)
+			}
 		}
-		deliver := pb.Delivery{Address: v.Address().Bytes(), Data: v.Data(), Stamp: stamp}
+
+		deliver := pb.Delivery{Address: c.Address().Bytes(), Data: c.Data(), Stamp: stamp}
 		if err := w.WriteMsgWithContext(ctx, &deliver); err != nil {
 			return fmt.Errorf("write delivery: %w", err)
 		}
@@ -467,9 +473,13 @@ func (s *Syncer) processWant(ctx context.Context, o *pb.Offer, w *pb.Want) ([]sw
 	for i := 0; i < len(o.Chunks); i++ {
 		if bv.Get(i) {
 			ch := o.Chunks[i]
-			c, err := s.store.ReserveGet(ctx, swarm.NewAddress(ch.Address), ch.BatchID)
+			addr := swarm.NewAddress(ch.Address)
+			c, err := s.store.ReserveGet(ctx, addr, ch.BatchID)
 			if err != nil {
-				return nil, err
+				s.logger.Error(err, "processing want: unable to find chunk", "chunk_address", addr, "batch_id", ch.BatchID)
+				chunks = append(chunks, swarm.NewChunk(swarm.ZeroAddress, nil))
+				s.metrics.MissingChunks.Inc()
+				continue
 			}
 			chunks = append(chunks, c)
 		}

pkg/pullsync/pullsync_test.go

@@ -228,6 +228,27 @@ func TestIncoming_UnsolicitedChunk(t *testing.T) {
 	}
 }
 
+func TestMissingChunk(t *testing.T) {
+	t.Parallel()
+
+	var (
+		zeroChunk   = swarm.NewChunk(swarm.ZeroAddress, nil)
+		topMost     = uint64(4)
+		ps, _       = newPullSync(t, nil, 5, mock.WithSubscribeResp(results, nil), mock.WithChunks([]swarm.Chunk{zeroChunk}...))
+		recorder    = streamtest.New(streamtest.WithProtocols(ps.Protocol()))
+		psClient, _ = newPullSync(t, recorder, 0)
+	)
+
+	topmost, _, err := psClient.Sync(context.Background(), swarm.ZeroAddress, 0, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if topmost != topMost {
+		t.Fatalf("got offer topmost %d but want %d", topmost, topMost)
+	}
+}
+
 func TestGetCursors(t *testing.T) {
 	t.Parallel()
 

pkg/storer/mock/mockreserve.go

@@ -34,7 +34,11 @@ func WithChunks(chs ...swarm.Chunk) Option {
 	return optionFunc(func(p *ReserveStore) {
 		for _, c := range chs {
 			c := c
-			p.chunks[c.Address().String()+string(c.Stamp().BatchID())] = c
+			if c.Stamp() != nil {
+				p.chunks[c.Address().String()+string(c.Stamp().BatchID())] = c
+			} else {
+				p.chunks[c.Address().String()] = c
+			}
 		}
 	})
 }