filer: keep a moved key on its prior owner while the ring settles (#11108)

ObjectTransaction forwards to the ring owner so one filer's per-path lock
arbitrates every writer of a key. But a ring change hands the key over before
the new owner has rebuilt the locks the prior owner still holds, so for the
cooling-off window both can grant it. LockRing.PriorOwner exists for exactly
this and nothing consulted it.

Route to the prior owner while that window is open. LockRing.WriteOwner
resolves prior-else-current under one read lock, so the pair cannot come from
different rings and name the same filer twice.

An unreachable owner fails the request rather than falling back to the current
one. gRPC reports a response lost in transit as Unavailable, indistinguishable
from a request the owner never saw, so re-sending elsewhere could re-apply what
the owner already committed; and an owner unreachable from here may be
partitioned rather than down, still serving the key to everyone else — which is
the split brain the routing exists to prevent. The window is bounded: once it
closes the ring hands the key to its new owner.

The owner resolution and the forward move into writeOwner/forwardToWriteOwner
so the next routed RPC reuses them rather than copying the block.

Claude-Session: https://claude.ai/code/session_01Fx1Hx8RqsJqHpbfbgTf4WJ

weed/cluster/lock_manager/lock_ring.go

@@ -163,6 +163,24 @@ func (r *LockRing) GetPrimary(key string) pb.ServerAddress {
 func (r *LockRing) PriorOwner(key string) pb.ServerAddress {
 	r.RLock()
 	defer r.RUnlock()
+	return r.priorOwnerLocked(key)
+}
+
+// WriteOwner returns the filer that should serialize writes to key: the prior
+// owner while a ring change is still within the cooling-off window, otherwise
+// the current primary. Both are read under one lock so the pair cannot come
+// from different rings, which could otherwise name the same filer twice.
+func (r *LockRing) WriteOwner(key string) pb.ServerAddress {
+	r.RLock()
+	defer r.RUnlock()
+	if prior := r.priorOwnerLocked(key); prior != "" {
+		return prior
+	}
+	return r.Ring.GetPrimary(key)
+}
+
+// priorOwnerLocked is PriorOwner's body; the caller holds at least RLock.
+func (r *LockRing) priorOwnerLocked(key string) pb.ServerAddress {
 	if len(r.snapshots) < 2 {
 		return ""
 	}

weed/cluster/lock_manager/lock_ring_prior_test.go

@@ -61,3 +61,42 @@ func TestLockRing_PriorOwnerExpires(t *testing.T) {
 		}
 	}
 }
+
+// WriteOwner resolves prior-else-current under one read lock, so the two views
+// can never come from different rings.
+func TestLockRing_WriteOwner(t *testing.T) {
+	r := NewLockRing(5 * time.Second)
+	t.Cleanup(r.WaitForCleanup)
+
+	setA := []pb.ServerAddress{"s1:1", "s2:1", "s3:1"}
+	r.SetSnapshot(setA, 1)
+
+	// One snapshot: no prior owner, so the primary owns every key.
+	if got, want := r.WriteOwner("any"), r.GetPrimary("any"); got != want {
+		t.Fatalf("WriteOwner=%q, want the primary %q", got, want)
+	}
+
+	r.SetSnapshot([]pb.ServerAddress{"s1:1", "s2:1", "s3:1", "s4:1"}, 2)
+
+	var moved, stable string
+	for i := 0; i < 2000 && (moved == "" || stable == ""); i++ {
+		key := fmt.Sprintf("key-%d", i)
+		if r.PriorOwner(key) != "" {
+			if moved == "" {
+				moved = key
+			}
+		} else if stable == "" {
+			stable = key
+		}
+	}
+	if moved == "" || stable == "" {
+		t.Skip("could not find both a moved and a stable key")
+	}
+
+	if got, want := r.WriteOwner(moved), r.PriorOwner(moved); got != want {
+		t.Fatalf("moved key: WriteOwner=%q, want the prior owner %q", got, want)
+	}
+	if got, want := r.WriteOwner(stable), r.GetPrimary(stable); got != want {
+		t.Fatalf("stable key: WriteOwner=%q, want the primary %q", got, want)
+	}
+}

weed/server/filer_grpc_server.go

@@ -291,27 +291,29 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje
 	// serialization point — even when the caller's ring view was stale. is_moved
 	// bounds this to one hop: a forwarded transaction is applied locally, so two
 	// filers that disagree on the owner during a ring change cannot loop.
-	if req.RouteKey != "" && !req.IsMoved && fs.filer.Dlm != nil {
-		if owner := fs.filer.Dlm.LockRing.GetPrimary(req.RouteKey); owner != "" && owner != fs.option.Host {
-			// Rebuild rather than copy the request struct (it carries a mutex);
-			// the pointer/slice fields are shared since the original is not mutated.
-			forwarded := &filer_pb.ObjectTransactionRequest{
-				LockKey:            req.LockKey,
-				Condition:          req.Condition,
-				Mutations:          req.Mutations,
-				IsFromOtherCluster: req.IsFromOtherCluster,
-				Signatures:         req.Signatures,
-				ConditionKey:       req.ConditionKey,
-				RouteKey:           req.RouteKey,
-				IsMoved:            true,
-			}
+	if req.RouteKey != "" && !req.IsMoved {
+		// Rebuild rather than copy the request struct (it carries a mutex); the
+		// pointer/slice fields are shared since the original is not mutated.
+		forwarded := &filer_pb.ObjectTransactionRequest{
+			LockKey:            req.LockKey,
+			Condition:          req.Condition,
+			Mutations:          req.Mutations,
+			IsFromOtherCluster: req.IsFromOtherCluster,
+			Signatures:         req.Signatures,
+			ConditionKey:       req.ConditionKey,
+			RouteKey:           req.RouteKey,
+			IsMoved:            true,
+		}
+		var resp *filer_pb.ObjectTransactionResponse
+		handled, err := fs.forwardToWriteOwner(ctx, req.RouteKey, func(owner pb.ServerAddress) error {
 			glog.V(2).InfofCtx(ctx, "ObjectTransaction %s: forwarding to owner %s", req.LockKey, owner)
-			var resp *filer_pb.ObjectTransactionResponse
-			err := pb.WithFilerClient(false, 0, owner, fs.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+			return pb.WithFilerClient(false, 0, owner, fs.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
 				var e error
 				resp, e = client.ObjectTransaction(ctx, forwarded)
 				return e
 			})
+		})
+		if handled {
 			if err != nil {
 				return &filer_pb.ObjectTransactionResponse{}, err
 			}

weed/server/filer_grpc_server_route.go

@@ -0,0 +1,48 @@
+package weed_server
+
+import (
+	"context"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+)
+
+// writeOwner returns the filer that serializes writes to key, or "" when this
+// filer is the serialization point — because it owns the key, or because there
+// is no ring and every filer applies locally.
+//
+// A ring change hands a key to its new owner before that owner has rebuilt the
+// locks the prior owner still holds, so the prior owner keeps the key until the
+// cooling-off window closes.
+func (fs *FilerServer) writeOwner(key string) pb.ServerAddress {
+	if fs.filer.Dlm == nil {
+		return ""
+	}
+	owner := fs.filer.Dlm.LockRing.WriteOwner(key)
+	if owner == fs.option.Host {
+		return ""
+	}
+	return owner
+}
+
+// forwardToWriteOwner sends the request to key's write owner so a single filer's
+// per-path lock arbitrates every writer of that key. handled=false means this
+// filer is the owner and the caller should apply the request locally.
+//
+// An unreachable owner fails the request; it is never re-sent to another filer.
+// gRPC reports a response lost in transit as Unavailable, indistinguishable from
+// one the owner never saw, so a retry elsewhere could re-apply what the owner
+// already committed — and an owner unreachable from here may be partitioned
+// rather than down, still serving the key to everyone else. The ring hands the
+// key on when the cooling-off window closes, so the outage is bounded.
+func (fs *FilerServer) forwardToWriteOwner(ctx context.Context, key string, send func(owner pb.ServerAddress) error) (handled bool, err error) {
+	owner := fs.writeOwner(key)
+	if owner == "" {
+		return false, nil
+	}
+	if err := send(owner); err != nil {
+		glog.V(1).InfofCtx(ctx, "route %s to owner %s: %v", key, owner, err)
+		return true, err
+	}
+	return true, nil
+}

weed/server/filer_grpc_server_route_test.go

@@ -0,0 +1,108 @@
+package weed_server
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager"
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+)
+
+// routeTestServer builds a filer whose ring holds the given snapshots, newest
+// last. Host is deliberately outside the ring in the forwarding tests so no key
+// is ever owned locally.
+func routeTestServer(host pb.ServerAddress, snapshots ...[]pb.ServerAddress) *FilerServer {
+	dlm := lock_manager.NewDistributedLockManager(host)
+	for i, servers := range snapshots {
+		dlm.LockRing.SetSnapshot(servers, int64(i+1))
+	}
+	return &FilerServer{
+		filer:  &filer.Filer{Dlm: dlm},
+		option: &FilerOption{Host: host},
+	}
+}
+
+var (
+	routeSetA = []pb.ServerAddress{"f1:8888", "f2:8888", "f3:8888"}
+	routeSetB = []pb.ServerAddress{"f1:8888", "f2:8888", "f3:8888", "f4:8888"}
+)
+
+// firstKey returns the first synthetic path the ring answers match for.
+func firstKey(t *testing.T, match func(key string) bool) string {
+	t.Helper()
+	for i := 0; i < 4000; i++ {
+		key := fmt.Sprintf("/buckets/b/key-%d", i)
+		if match(key) {
+			return key
+		}
+	}
+	t.Skip("no key satisfying the ring condition")
+	return ""
+}
+
+func TestWriteOwnerNoRing(t *testing.T) {
+	fs := routeTestServer("f9:8888")
+	if got := fs.writeOwner("/any"); got != "" {
+		t.Fatalf("no ring snapshot must leave the write local, got %v", got)
+	}
+}
+
+func TestWriteOwnerStableKeyUsesPrimary(t *testing.T) {
+	fs := routeTestServer("f9:8888", routeSetA, routeSetB)
+	ring := fs.filer.Dlm.LockRing
+
+	stable := firstKey(t, func(key string) bool { return ring.PriorOwner(key) == "" })
+	if got := fs.writeOwner(stable); got != ring.GetPrimary(stable) {
+		t.Fatalf("stable key must route to the primary, got %v want %v", got, ring.GetPrimary(stable))
+	}
+}
+
+// A key whose ownership just moved must keep going to the prior owner: the new
+// owner has not rebuilt the locks the prior one still holds.
+func TestWriteOwnerMovedKeyUsesPriorOwner(t *testing.T) {
+	fs := routeTestServer("f9:8888", routeSetA, routeSetB)
+	ring := fs.filer.Dlm.LockRing
+
+	moved := firstKey(t, func(key string) bool { return ring.PriorOwner(key) != "" })
+	if got := fs.writeOwner(moved); got != ring.PriorOwner(moved) {
+		t.Fatalf("moved key must route to the prior owner, got %v want %v", got, ring.PriorOwner(moved))
+	}
+}
+
+func TestForwardToWriteOwnerAppliesLocallyWhenOwned(t *testing.T) {
+	fs := routeTestServer("f2:8888", routeSetA)
+	ring := fs.filer.Dlm.LockRing
+
+	key := firstKey(t, func(key string) bool { return ring.GetPrimary(key) == "f2:8888" })
+	handled, err := fs.forwardToWriteOwner(context.Background(), key, func(pb.ServerAddress) error {
+		t.Fatal("must not forward a key this filer owns")
+		return nil
+	})
+	if handled || err != nil {
+		t.Fatalf("owned key must be applied locally, got handled=%v err=%v", handled, err)
+	}
+}
+
+// A failed forward must surface, never re-send to a second filer: gRPC cannot
+// tell a lost response from an unsent request, and an owner unreachable from
+// here may be partitioned rather than down.
+func TestForwardToWriteOwnerNeverTriesASecondFiler(t *testing.T) {
+	fs := routeTestServer("f9:8888", routeSetA, routeSetB)
+	ring := fs.filer.Dlm.LockRing
+
+	moved := firstKey(t, func(key string) bool { return ring.PriorOwner(key) != "" })
+	var tried []pb.ServerAddress
+	handled, err := fs.forwardToWriteOwner(context.Background(), moved, func(owner pb.ServerAddress) error {
+		tried = append(tried, owner)
+		return errors.New("dial tcp: connection refused")
+	})
+	if !handled || err == nil {
+		t.Fatalf("unreachable owner must surface an error, got handled=%v err=%v", handled, err)
+	}
+	if len(tried) != 1 || tried[0] != ring.PriorOwner(moved) {
+		t.Fatalf("expected exactly the prior owner, tried %v", tried)
+	}
+}