s3api: no filer failover after the callback has consumed part of a response (#10902)

s3api: no filer failover after fn has consumed part of a response withFilerClientFailover replays fn verbatim on the next filer, so a filer that died mid-stream followed by a healthy peer returned success with the callback's closure-captured accumulator holding the dead filer's prefix twice; the per-attempt accumulator in listWithRetry could not close this, because the replay happens inside a single attempt. Track delivery on the connection handed to fn: once a unary reply or streamed message has reached the callback, surface the transport error unwrapped instead of failing over, and let callers replay from a clean slate. A filer that fails before delivering anything fails over exactly as before.
2026-08-28 11:56:07 +00:00 · 2026-08-23 11:30:43 -07:00
parent c167af541e
commit cf0dba334c
3 changed files with 188 additions and 4 deletions
@@ -40,6 +40,10 @@ func (s3a *S3ApiServer) WithFilerClient(streamingMode bool, fn func(filer_pb.Sea
 // caller route to a key's ring owner for read-after-write; it may be a filer outside
 // the static list (the bookkeeping no-ops for untracked addresses). A failover
 // updates the current filer; a preferred read does not, as its owner is per-key.
+// Failover replays fn from scratch, so it stops once any response has reached
+// fn: a replay after that could silently duplicate state fn accumulated (a
+// listing that failed mid-stream, say), so the error surfaces instead and the
+// caller decides whether a clean-slate retry is safe.
 func (s3a *S3ApiServer) withFilerClientFailover(preferred pb.ServerAddress, streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error {
 	currentFiler := s3a.filerClient.GetCurrentFiler()

@@ -82,8 +86,9 @@ func (s3a *S3ApiServer) withFilerClientFailover(preferred pb.ServerAddress, stre

 	var lastErr error
 	for _, filer := range ordered {
+		received := false
 		err := pb.WithGrpcClient(context.Background(), streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error {
-			return fn(filer_pb.NewSeaweedFilerClient(grpcConnection))
+			return fn(filer_pb.NewSeaweedFilerClient(receiveTrackingConn{ClientConnInterface: grpcConnection, received: &received}))
 		}, filer.ToGrpcAddress(), false, s3a.option.GrpcDialOption)

 		if err == nil {
@@ -105,6 +110,11 @@ func (s3a *S3ApiServer) withFilerClientFailover(preferred pb.ServerAddress, stre
 			s3a.markOwnerUnreachable(filer)
 		}
 		glog.V(2).Infof("WithFilerClient: filer %s failed: %v", filer, err)
+		// fn consumed part of a response; a replay would stack a second copy
+		// onto whatever it accumulated, so surface the error unwrapped.
+		if received {
+			return err
+		}
 		lastErr = err
 	}

@@ -114,6 +124,43 @@ func (s3a *S3ApiServer) withFilerClientFailover(preferred pb.ServerAddress, stre
 	return fmt.Errorf("all filers failed, last error: %w", lastErr)
 }

+// receiveTrackingConn flags *received once a unary reply or a streamed message
+// has been handed to the callback, the point past which a failover replay is
+// no longer transparent.
+type receiveTrackingConn struct {
+	grpc.ClientConnInterface
+	received *bool
+}
+
+func (c receiveTrackingConn) Invoke(ctx context.Context, method string, args, reply any, opts ...grpc.CallOption) error {
+	err := c.ClientConnInterface.Invoke(ctx, method, args, reply, opts...)
+	if err == nil {
+		*c.received = true
+	}
+	return err
+}
+
+func (c receiveTrackingConn) NewStream(ctx context.Context, desc *grpc.StreamDesc, method string, opts ...grpc.CallOption) (grpc.ClientStream, error) {
+	stream, err := c.ClientConnInterface.NewStream(ctx, desc, method, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return receiveTrackingStream{ClientStream: stream, received: c.received}, nil
+}
+
+type receiveTrackingStream struct {
+	grpc.ClientStream
+	received *bool
+}
+
+func (s receiveTrackingStream) RecvMsg(m any) error {
+	err := s.ClientStream.RecvMsg(m)
+	if err == nil {
+		*s.received = true
+	}
+	return err
+}
+
 func (s3a *S3ApiServer) AdjustedUrl(location *filer_pb.Location) string {
 	return location.Url
 }
@@ -0,0 +1,137 @@
+package s3api
+
+import (
+	"context"
+	"io"
+	"math"
+	"reflect"
+	"sync/atomic"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/wdclient"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+)
+
+// fakeListFiler serves ListEntries from names; the first failCalls calls die
+// with Unavailable after streaming sendBeforeFail entries.
+type fakeListFiler struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+	names          []string
+	failCalls      int32
+	sendBeforeFail int
+	calls          int32
+}
+
+func (f *fakeListFiler) ListEntries(req *filer_pb.ListEntriesRequest, stream filer_pb.SeaweedFiler_ListEntriesServer) error {
+	names := f.names
+	failing := atomic.AddInt32(&f.calls, 1) <= f.failCalls
+	if failing {
+		names = names[:f.sendBeforeFail]
+	}
+	for _, name := range names {
+		if err := stream.Send(&filer_pb.ListEntriesResponse{Entry: &filer_pb.Entry{Name: name}}); err != nil {
+			return err
+		}
+	}
+	if failing {
+		return status.Error(codes.Unavailable, "filer restarting")
+	}
+	return nil
+}
+
+func newFailoverTestServer(t *testing.T, filers ...pb.ServerAddress) *S3ApiServer {
+	t.Helper()
+	dialOption := grpc.WithTransportCredentials(insecure.NewCredentials())
+	return &S3ApiServer{
+		option:      &S3ApiServerOption{Filers: filers, GrpcDialOption: dialOption},
+		filerClient: wdclient.NewFilerClient(filers, dialOption, ""),
+	}
+}
+
+// accumulateListing is the callback shape the failover contract has to protect:
+// entries collect into a variable that survives a replay of the callback.
+func accumulateListing(got *[]string) func(filer_pb.SeaweedFilerClient) error {
+	return func(client filer_pb.SeaweedFilerClient) error {
+		stream, err := client.ListEntries(context.Background(), &filer_pb.ListEntriesRequest{Directory: "/d"})
+		if err != nil {
+			return err
+		}
+		for {
+			resp, err := stream.Recv()
+			if err == io.EOF {
+				return nil
+			}
+			if err != nil {
+				return err
+			}
+			*got = append(*got, resp.Entry.Name)
+		}
+	}
+}
+
+// A filer that dies mid-stream must surface the error, not fail over: the
+// callback has already consumed part of the response, and replaying it against
+// the next filer would append a second copy of what it accumulated.
+func TestFailoverStopsAfterPartialResponse(t *testing.T) {
+	names := []string{"e1", "e2", "e3", "e4"}
+	flaky := &fakeListFiler{names: names, failCalls: math.MaxInt32, sendBeforeFail: 2}
+	healthy := &fakeListFiler{names: names}
+	s3a := newFailoverTestServer(t, startFakeFiler(t, flaky), startFakeFiler(t, healthy))
+
+	var got []string
+	err := s3a.WithFilerClient(true, accumulateListing(&got))
+	if err == nil {
+		t.Fatalf("want the mid-stream failure surfaced, got success with %v", got)
+	}
+	if calls := atomic.LoadInt32(&healthy.calls); calls != 0 {
+		t.Fatalf("callback was replayed on the second filer %d time(s)", calls)
+	}
+}
+
+// A filer that fails before delivering anything is still failed over, so the
+// partial-response guard does not cost the healthy-peer retry that failover exists for.
+func TestFailoverBeforeFirstResponse(t *testing.T) {
+	names := []string{"e1", "e2", "e3", "e4"}
+	flaky := &fakeListFiler{names: names, failCalls: math.MaxInt32, sendBeforeFail: 0}
+	healthy := &fakeListFiler{names: names}
+	s3a := newFailoverTestServer(t, startFakeFiler(t, flaky), startFakeFiler(t, healthy))
+
+	var got []string
+	err := s3a.WithFilerClient(true, accumulateListing(&got))
+	if err != nil {
+		t.Fatalf("want failover success, got %v", err)
+	}
+	if !reflect.DeepEqual(got, names) {
+		t.Fatalf("entries = %v, want %v", got, names)
+	}
+}
+
+// End to end through the real listing path: a mid-stream failure surfaces to
+// listWithRetry, whose replay starts from a fresh accumulator, so the caller
+// sees each entry exactly once instead of a silently duplicated prefix.
+func TestListAfterMidStreamFailureHasNoDuplicates(t *testing.T) {
+	names := []string{"e1", "e2", "e3", "e4"}
+	flaky := &fakeListFiler{names: names, failCalls: 1, sendBeforeFail: 2}
+	healthy := &fakeListFiler{names: names}
+	s3a := newFailoverTestServer(t, startFakeFiler(t, flaky), startFakeFiler(t, healthy))
+
+	entries, isLast, err := s3a.list("/d", "", "", false, 10)
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	var got []string
+	for _, entry := range entries {
+		got = append(got, entry.Name)
+	}
+	if !reflect.DeepEqual(got, names) {
+		t.Fatalf("entries = %v, want %v", got, names)
+	}
+	if !isLast {
+		t.Fatal("want isLast")
+	}
+}
@@ -262,9 +262,9 @@ func (f *fakeTxnFiler) ObjectTransaction(ctx context.Context, req *filer_pb.Obje
 	return &filer_pb.ObjectTransactionResponse{}, nil
 }

-// startFakeTxnFiler serves impl on a random localhost port and returns the S3-style
+// startFakeFiler serves impl on a random localhost port and returns the S3-style
 // filer address whose ToGrpcAddress resolves back to that port.
-func startFakeTxnFiler(t *testing.T, impl *fakeTxnFiler) pb.ServerAddress {
+func startFakeFiler(t *testing.T, impl filer_pb.SeaweedFilerServer) pb.ServerAddress {
 	t.Helper()
 	lis, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {
@@ -296,7 +296,7 @@ func closedFilerAddress(t *testing.T) pb.ServerAddress {
 // so an object write survives a filer pod IP change without an S3 gateway restart.
 func TestObjectTxnFailsOverStaleOwner(t *testing.T) {
 	live := &fakeTxnFiler{}
-	liveAddr := startFakeTxnFiler(t, live)
+	liveAddr := startFakeFiler(t, live)
 	deadOwner := closedFilerAddress(t)

 	dialOption := grpc.WithTransportCredentials(insecure.NewCredentials())