diff --git a/weed/s3api/s3api_handlers.go b/weed/s3api/s3api_handlers.go index 52c2d0cf3..59ac9f996 100644 --- a/weed/s3api/s3api_handlers.go +++ b/weed/s3api/s3api_handlers.go @@ -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 } diff --git a/weed/s3api/s3api_handlers_failover_test.go b/weed/s3api/s3api_handlers_failover_test.go new file mode 100644 index 000000000..7c10ace09 --- /dev/null +++ b/weed/s3api/s3api_handlers_failover_test.go @@ -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") + } +} diff --git a/weed/s3api/s3api_object_routed_write_test.go b/weed/s3api/s3api_object_routed_write_test.go index 0e47b1559..2dc052566 100644 --- a/weed/s3api/s3api_object_routed_write_test.go +++ b/weed/s3api/s3api_object_routed_write_test.go @@ -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())