fix(steward): swallow ErrWantSelf (#1980)

pkg/steward/steward.go

@@ -8,11 +8,13 @@ package steward
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/ethersphere/bee/pkg/pushsync"
 	"github.com/ethersphere/bee/pkg/storage"
 	"github.com/ethersphere/bee/pkg/swarm"
+	"github.com/ethersphere/bee/pkg/topology"
 	"github.com/ethersphere/bee/pkg/traversal"
 	"golang.org/x/sync/errgroup"
 )
@@ -55,7 +57,10 @@ func (s *steward) Reupload(ctx context.Context, root swarm.Address) error {
 			defer func() { <-sem }()
 			_, err := s.push.PushChunkToClosest(ctx, c)
 			if err != nil {
-				return err
+				if !errors.Is(err, topology.ErrWantSelf) {
+					return err
+				}
+				// swallow the error in case we are the closest node
 			}
 			return nil
 		})

pkg/steward/steward_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/ethersphere/bee/pkg/storage"
 	"github.com/ethersphere/bee/pkg/storage/mock"
 	"github.com/ethersphere/bee/pkg/swarm"
+	"github.com/ethersphere/bee/pkg/topology"
 	"github.com/ethersphere/bee/pkg/traversal"
 )
 
@@ -69,6 +70,40 @@ func TestSteward(t *testing.T) {
 	}
 }
 
+func TestSteward_ErrWantSelf(t *testing.T) {
+	var (
+		ctx       = context.Background()
+		chunks    = 10
+		data      = make([]byte, chunks*4096)
+		store     = mock.NewStorer()
+		traverser = traversal.New(store)
+		fn        = func(_ context.Context, ch swarm.Chunk) (*pushsync.Receipt, error) {
+			return nil, topology.ErrWantSelf
+		}
+		ps = psmock.New(fn)
+		s  = steward.New(store, traverser, ps)
+	)
+	n, err := rand.Read(data)
+	if n != cap(data) {
+		t.Fatal("short read")
+	}
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	l := &loggingStore{Storer: store}
+	pipe := builder.NewPipelineBuilder(ctx, l, storage.ModePutUpload, false)
+	addr, err := builder.FeedPipeline(ctx, pipe, bytes.NewReader(data))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = s.Reupload(ctx, addr)
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
 type loggingStore struct {
 	storage.Storer
 	addrs []swarm.Address