s3: fail over routed object writes when the owner filer is unreachable (#10251)

* s3: fail over routed object writes when the owner filer is unreachable

A routed object write (multipart completion, PUT, delete, versioned
finalize, metadata replace) dialed the ring-selected owner filer
directly with no failover. After a filer restarts onto a new address the
lock ring can still name the old one, so every routed write hangs on the
dead address until the gateway is restarted; CompleteMultipartUpload in
particular exceeds client timeouts.

Route the transaction through withFilerClientFailover, skipping an owner
that recently failed, so a live filer forwards it to the real owner by
route_key. Mirrors the read path's getObjectEntryRoutedByKey.

* s3: fail over bucket-config writes when the owner filer is unreachable

patchBucketEntry dialed the bucket's ring owner directly, so a restarted
filer's stale ring address hung every bucket-config write (versioning,
lifecycle, object lock, ownership, ACL, policy, CORS) - the same failure
as routed object writes. Route it through objectTxnOnFiler so it skips an
unreachable owner and a live filer forwards by route_key.

weed/s3api/s3api_bucket_config.go

@@ -17,7 +17,6 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/kms"
-	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/s3_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/cors"
@@ -587,8 +586,7 @@ func (s3a *S3ApiServer) patchBucketEntry(bucket string, m *filer_pb.ObjectMutati
 		RouteKey:  objectWriteRouteKeyPrefix + bucketPath,
 		Mutations: []*filer_pb.ObjectMutation{m},
 	}
-	txn := func(client filer_pb.SeaweedFilerClient) error {
-		resp, err := client.ObjectTransaction(context.Background(), req)
+	respErr := func(resp *filer_pb.ObjectTransactionResponse, err error) error {
 		if err != nil {
 			return err
 		}
@@ -598,11 +596,13 @@ func (s3a *S3ApiServer) patchBucketEntry(bucket string, m *filer_pb.ObjectMutati
 		return nil
 	}
 	if s3a.objectWriteLockClient != nil {
-		if owner := s3a.objectWriteLockClient.PrimaryForKey(objectWriteRouteKeyPrefix + bucketPath); owner != "" {
-			return pb.WithFilerClient(false, 0, owner, s3a.option.GrpcDialOption, txn)
+		if owner := s3a.objectWriteLockClient.PrimaryForKey(req.RouteKey); owner != "" {
+			return respErr(s3a.objectTxnOnFiler(owner, req))
 		}
 	}
-	return s3a.WithFilerClient(false, txn)
+	return s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		return respErr(client.ObjectTransaction(context.Background(), req))
+	})
 }
 
 // isVersioningEnabled checks if versioning is enabled for a bucket (with caching)

weed/s3api/s3api_object_routed_write.go

@@ -148,13 +148,29 @@ func singleStrongETag(v string) (string, bool) {
 	return strings.Trim(v, `"`), true
 }
 
+// objectTxnOnFiler applies a routed transaction, preferring the object's owner
+// filer but failing over to a live filer when the owner is unreachable, so a
+// restarted filer's stale ring address (e.g. a rolled K8s pod's old IP) can't hang
+// the write. The reached filer forwards to the real owner by route_key, keeping its
+// per-path lock authoritative. Mirrors getObjectEntryRoutedByKey on the read path.
 func (s3a *S3ApiServer) objectTxnOnFiler(owner pb.ServerAddress, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) {
 	var resp *filer_pb.ObjectTransactionResponse
-	err := pb.WithFilerClient(false, 0, owner, s3a.option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+	txn := func(client filer_pb.SeaweedFilerClient) error {
 		var e error
 		resp, e = client.ObjectTransaction(context.Background(), req)
 		return e
-	})
+	}
+	if s3a.filerClient == nil {
+		err := pb.WithFilerClient(false, 0, owner, s3a.option.GrpcDialOption, txn)
+		return resp, err
+	}
+	// Skip an owner whose recent write hit a transport error so a dead/stale
+	// address isn't re-dialed every request.
+	preferred := owner
+	if s3a.ownerRecentlyUnreachable(owner) {
+		preferred = ""
+	}
+	err := s3a.withFilerClientFailover(preferred, false, txn)
 	return resp, err
 }
 

weed/s3api/s3api_object_routed_write_test.go

@@ -1,13 +1,21 @@
 package s3api
 
 import (
+	"context"
+	"fmt"
+	"net"
 	"net/http"
+	"sync/atomic"
 	"testing"
 	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+	"github.com/seaweedfs/seaweedfs/weed/wdclient"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 )
 
 func reqWith(headers map[string]string) *http.Request {
@@ -242,6 +250,87 @@ func TestSingleStrongETag(t *testing.T) {
 	}
 }
 
+// fakeTxnFiler is a minimal SeaweedFiler gRPC server that records ObjectTransaction
+// calls, standing in for a live filer that a routed write fails over to.
+type fakeTxnFiler struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+	calls int32
+}
+
+func (f *fakeTxnFiler) ObjectTransaction(ctx context.Context, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) {
+	atomic.AddInt32(&f.calls, 1)
+	return &filer_pb.ObjectTransactionResponse{}, nil
+}
+
+// startFakeTxnFiler 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 {
+	t.Helper()
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	srv := grpc.NewServer()
+	filer_pb.RegisterSeaweedFilerServer(srv, impl)
+	go srv.Serve(lis)
+	t.Cleanup(srv.Stop)
+	port := lis.Addr().(*net.TCPAddr).Port
+	return pb.ServerAddress(fmt.Sprintf("127.0.0.1:1.%d", port))
+}
+
+// closedFilerAddress returns an address whose gRPC port has nothing listening,
+// modeling a filer whose ring address is stale after a pod restart.
+func closedFilerAddress(t *testing.T) pb.ServerAddress {
+	t.Helper()
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	port := lis.Addr().(*net.TCPAddr).Port
+	lis.Close()
+	return pb.ServerAddress(fmt.Sprintf("127.0.0.1:1.%d", port))
+}
+
+// A routed write whose owner address is dead fails over to a live filer (which
+// forwards to the real owner by route_key) instead of hanging on the dead owner,
+// 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)
+	deadOwner := closedFilerAddress(t)
+
+	dialOption := grpc.WithTransportCredentials(insecure.NewCredentials())
+	s3a := &S3ApiServer{
+		option:      &S3ApiServerOption{GrpcDialOption: dialOption},
+		filerClient: wdclient.NewFilerClient([]pb.ServerAddress{liveAddr}, dialOption, ""),
+	}
+	req := &filer_pb.ObjectTransactionRequest{LockKey: "/buckets/b/o", RouteKey: "s3.object.write:/buckets/b/o"}
+
+	// Owner not yet flagged: the first attempt dials it, fails fast, then fails
+	// over to the live filer and flags the owner unreachable.
+	resp, err := s3a.objectTxnOnFiler(deadOwner, req)
+	if err != nil {
+		t.Fatalf("expected failover success, got %v", err)
+	}
+	if resp == nil || resp.Error != "" {
+		t.Fatalf("unexpected response %+v", resp)
+	}
+	if got := atomic.LoadInt32(&live.calls); got != 1 {
+		t.Fatalf("live filer calls = %d, want 1", got)
+	}
+	if !s3a.ownerRecentlyUnreachable(deadOwner) {
+		t.Fatal("dead owner should be flagged unreachable after the failed dial")
+	}
+
+	// Once flagged, the owner is skipped entirely and the write still lands.
+	if _, err := s3a.objectTxnOnFiler(deadOwner, req); err != nil {
+		t.Fatalf("expected success while owner flagged, got %v", err)
+	}
+	if got := atomic.LoadInt32(&live.calls); got != 2 {
+		t.Fatalf("live filer calls = %d, want 2", got)
+	}
+}
+
 func TestRouteWriteCondition(t *testing.T) {
 	// Unconditional routes either way.
 	if c, ok := routeWriteCondition(reqWith(nil), false); !ok || c != nil {