s3: dial the object lock's primary filer directly (#9626)

* s3: dial the object lock's primary filer directly

The S3 object write lock builds a fresh short-lived lock per write, each
starting at the seed filer. When the seed isn't the key's hash-ring primary
the filer forwards the request to the primary, and in multi-cluster setups
that forward crosses clusters on every write.

Give the lock client a view of the filer lock ring, fed by the master's
LockRingUpdate broadcasts the gateway already receives, so it dials the
primary directly. The view tracks filer membership by version; a stale view
stays correct because the filer still forwards as a fallback.

Also send the initial ring snapshot to S3 clients, not just filers.

* s3: subscribe to lock-ring updates before starting the master loop

The master delivers the initial LockRingUpdate once, on connect. Registering the
callback after KeepConnectedToMaster started left a window where that first
update could arrive before the handler was set and be dropped, delaying the ring
view until the next membership change. Build the lock client and register the
callback in the masters block before launching the loop; the filers block reuses
that client (or creates a plain one when no masters are configured).

* lock_manager: build the hash ring in a deterministic server order

rebuildRing ranged over the server set (a map), whose iteration order is
randomized per process. On a vnode hash collision the last writer into
vnodeToServer wins, so two nodes holding the same server set could resolve the
collision to different servers and disagree on the primary for keys near that
slot. Now that the S3 gateway also computes PrimaryForKey, such a disagreement
would route the same key to different filers and defeat per-path serialization.

Iterate the servers in sorted order so the ring is identical on every node with
the same set, regardless of discovery order.

* lock_manager: skip redundant ring rebuilds, trim comments

SetRing now ignores a non-zero version at or below the current one once a ring
exists, so repeated LockRingUpdate broadcasts on reconnect no longer rebuild the
ring.

* s3: hold the lock-ring client on the server for route-by-key

Store the object-write lock client on S3ApiServer so handlers can resolve a
key's owner filer via PrimaryForKey.

weed/cluster/lock_client.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -19,6 +20,14 @@ type LockClient struct {
 	maxLockDuration time.Duration
 	sleepDuration   time.Duration
 	seedFiler       pb.ServerAddress
+
+	// ring is an optional client-side view of the filer lock hash ring. When
+	// populated, a new lock starts at the key's primary filer instead of the
+	// seed filer, avoiding the seed->primary forward hop. A stale view stays
+	// correct: the filer forwards to the real primary as a fallback.
+	ringMu      sync.RWMutex
+	ring        *lock_manager.HashRing
+	ringVersion int64
 }
 
 func NewLockClient(grpcDialOption grpc.DialOption, seedFiler pb.ServerAddress) *LockClient {
@@ -30,6 +39,49 @@ func NewLockClient(grpcDialOption grpc.DialOption, seedFiler pb.ServerAddress) *
 	}
 }
 
+// SetRing mirrors the master's LockRingUpdate so the client computes the same
+// primary the filers do. A non-zero version at or below the current one is
+// ignored once a ring exists, dropping reordered and redundant broadcasts;
+// version 0 always applies (bootstrap).
+func (lc *LockClient) SetRing(servers []pb.ServerAddress, version int64) {
+	lc.ringMu.Lock()
+	defer lc.ringMu.Unlock()
+	if version != 0 && version <= lc.ringVersion && lc.ring != nil {
+		return
+	}
+	lc.ringVersion = version
+	if lc.ring == nil {
+		lc.ring = lock_manager.NewHashRing(lock_manager.DefaultVnodeCount)
+	}
+	lc.ring.SetServers(servers)
+}
+
+// hostForKey returns the filer that should own key per the current ring view,
+// falling back to the seed filer when no view has been received yet.
+func (lc *LockClient) hostForKey(key string) pb.ServerAddress {
+	lc.ringMu.RLock()
+	defer lc.ringMu.RUnlock()
+	if lc.ring == nil {
+		return lc.seedFiler
+	}
+	if primary := lc.ring.GetPrimary(key); primary != "" {
+		return primary
+	}
+	return lc.seedFiler
+}
+
+// PrimaryForKey returns the ring owner for key, or "" before any ring arrives.
+// Unlike hostForKey it does not fall back to the seed, so a route-by-key caller
+// stays on the distributed lock until the ring is known.
+func (lc *LockClient) PrimaryForKey(key string) pb.ServerAddress {
+	lc.ringMu.RLock()
+	defer lc.ringMu.RUnlock()
+	if lc.ring == nil {
+		return ""
+	}
+	return lc.ring.GetPrimary(key)
+}
+
 type LiveLock struct {
 	key                 string
 	renewToken          string
@@ -51,7 +103,7 @@ type LiveLock struct {
 func (lc *LockClient) NewShortLivedLock(key string, owner string) (lock *LiveLock) {
 	lock = &LiveLock{
 		key:            key,
-		hostFiler:      lc.seedFiler,
+		hostFiler:      lc.hostForKey(key),
 		cancelCh:       make(chan struct{}),
 		expireAtNs:     time.Now().Add(5 * time.Second).UnixNano(),
 		grpcDialOption: lc.grpcDialOption,
@@ -72,7 +124,7 @@ func (lc *LockClient) NewBlockingLongLivedLock(key, owner string, lockTTL time.D
 	}
 	lock := &LiveLock{
 		key:            key,
-		hostFiler:      lc.seedFiler,
+		hostFiler:      lc.hostForKey(key),
 		cancelCh:       make(chan struct{}),
 		expireAtNs:     time.Now().Add(lockTTL).UnixNano(),
 		grpcDialOption: lc.grpcDialOption,
@@ -110,7 +162,7 @@ func (lc *LockClient) NewBlockingLongLivedLock(key, owner string, lockTTL time.D
 func (lc *LockClient) StartLongLivedLock(key string, owner string, onLockOwnerChange func(newLockOwner string), lockTTL time.Duration) (lock *LiveLock) {
 	lock = &LiveLock{
 		key:            key,
-		hostFiler:      lc.seedFiler,
+		hostFiler:      lc.hostForKey(key),
 		cancelCh:       make(chan struct{}),
 		expireAtNs:     time.Now().Add(lockTTL).UnixNano(),
 		grpcDialOption: lc.grpcDialOption,

weed/cluster/lock_client_test.go

@@ -0,0 +1,92 @@
+package cluster
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+)
+
+// The gateway must resolve a lock key to the same primary the filers do,
+// otherwise it dials the wrong filer and the lock still gets forwarded. Both
+// sides use the same HashRing over the same server set, so for every key the
+// client's hostForKey must equal the filer ring's GetPrimary.
+func TestLockClientHostMatchesFilerRing(t *testing.T) {
+	servers := []pb.ServerAddress{
+		"filer-a:8888", "filer-b:8888", "filer-c:8888", "filer-d:8888",
+	}
+
+	filerRing := lock_manager.NewHashRing(lock_manager.DefaultVnodeCount)
+	filerRing.SetServers(servers)
+
+	lc := NewLockClient(nil, "seed:8888")
+	lc.SetRing(servers, 1)
+
+	for _, key := range []string{
+		"s3.object.write:/buckets/b/obj-0",
+		"s3.object.write:/buckets/b/obj-1",
+		"s3.object.write:/buckets/b/obj-2",
+		"s3.object.write:/buckets/gosbench-0/w0obj-kilo-0877",
+		"some/other/key",
+	} {
+		if got, want := lc.hostForKey(key), filerRing.GetPrimary(key); got != want {
+			t.Errorf("key %q: client host %q != filer primary %q", key, got, want)
+		}
+	}
+}
+
+// Without a ring view, the client falls back to the seed filer (which the filer
+// forwards from), preserving the pre-optimization behavior.
+func TestLockClientHostFallsBackToSeed(t *testing.T) {
+	lc := NewLockClient(nil, "seed:8888")
+	if got := lc.hostForKey("any-key"); got != "seed:8888" {
+		t.Errorf("expected seed fallback, got %q", got)
+	}
+
+	// An empty ring (no members yet) also falls back to the seed.
+	lc.SetRing(nil, 1)
+	if got := lc.hostForKey("any-key"); got != "seed:8888" {
+		t.Errorf("expected seed fallback on empty ring, got %q", got)
+	}
+}
+
+// A stale (older-version) update must not regress a newer ring view, while
+// version 0 always applies as a bootstrap.
+func TestLockClientSetRingVersionGuard(t *testing.T) {
+	lc := NewLockClient(nil, "seed:8888")
+
+	newer := []pb.ServerAddress{"filer-a:8888", "filer-b:8888"}
+	lc.SetRing(newer, 10)
+	primaryAt10 := lc.hostForKey("k")
+
+	// Older version is ignored.
+	lc.SetRing([]pb.ServerAddress{"filer-z:8888"}, 5)
+	if got := lc.hostForKey("k"); got != primaryAt10 {
+		t.Errorf("stale update applied: host changed to %q", got)
+	}
+
+	// version 0 is always accepted.
+	lc.SetRing([]pb.ServerAddress{"filer-z:8888"}, 0)
+	if got := lc.hostForKey("k"); got != "filer-z:8888" {
+		t.Errorf("bootstrap update not applied, got %q", got)
+	}
+}
+
+// PrimaryForKey returns "" before any ring is received (so a route-by-key
+// caller falls back to the distributed lock) and the ring owner afterwards,
+// unlike hostForKey which falls back to the seed.
+func TestLockClientPrimaryForKey(t *testing.T) {
+	lc := NewLockClient(nil, "seed:8888")
+	if got := lc.PrimaryForKey("k"); got != "" {
+		t.Errorf("expected empty before ring, got %q", got)
+	}
+
+	lc.SetRing([]pb.ServerAddress{"filer-a:8888", "filer-b:8888"}, 1)
+	got := lc.PrimaryForKey("k")
+	if got == "" {
+		t.Fatal("expected an owner after ring set")
+	}
+	if got != lc.hostForKey("k") {
+		t.Errorf("PrimaryForKey %q disagrees with hostForKey %q", got, lc.hostForKey("k"))
+	}
+}

weed/cluster/lock_manager/hash_ring.go

@@ -145,7 +145,15 @@ func (hr *HashRing) rebuildRing() {
 	hr.vnodeToServer = make(map[uint32]pb.ServerAddress, len(hr.servers)*hr.vnodeCount)
 	hr.sortedHashes = make([]uint32, 0, len(hr.servers)*hr.vnodeCount)
 
+	// Sort so a vnode-hash collision resolves to the same server on every node;
+	// map iteration order alone is randomized per process.
+	servers := make([]pb.ServerAddress, 0, len(hr.servers))
 	for server := range hr.servers {
+		servers = append(servers, server)
+	}
+	sort.Slice(servers, func(i, j int) bool { return servers[i] < servers[j] })
+
+	for _, server := range servers {
 		for i := 0; i < hr.vnodeCount; i++ {
 			vnodeKey := vnodeKeyFor(server, i)
 			hash := hashKey(vnodeKey)

weed/cluster/lock_manager/hash_ring_test.go

@@ -171,3 +171,38 @@ func TestHashRing_GetPrimary(t *testing.T) {
 	primary, _ := hr.GetPrimaryAndBackup("mykey")
 	assert.Equal(t, primary, hr.GetPrimary("mykey"))
 }
+
+// The ring must be identical on every node holding the same server set,
+// regardless of the order servers were added or supplied. Build rings several
+// ways and assert they agree on the primary for a wide range of keys.
+func TestHashRing_OrderIndependent(t *testing.T) {
+	servers := []pb.ServerAddress{
+		"filer-a:8888", "filer-b:8888", "filer-c:8888", "filer-d:8888", "filer-e:8888",
+	}
+
+	bySet := NewHashRing(50)
+	bySet.SetServers(servers)
+
+	byReverse := NewHashRing(50)
+	rev := append([]pb.ServerAddress(nil), servers...)
+	for i, j := 0, len(rev)-1; i < j; i, j = i+1, j-1 {
+		rev[i], rev[j] = rev[j], rev[i]
+	}
+	byReverse.SetServers(rev)
+
+	byAdd := NewHashRing(50)
+	for _, s := range []pb.ServerAddress{"filer-c:8888", "filer-e:8888", "filer-a:8888", "filer-d:8888", "filer-b:8888"} {
+		byAdd.AddServer(s)
+	}
+
+	for i := 0; i < 5000; i++ {
+		key := fmt.Sprintf("s3.object.write:/buckets/b/obj-%d", i)
+		p := bySet.GetPrimary(key)
+		if got := byReverse.GetPrimary(key); got != p {
+			t.Fatalf("reverse-order ring disagrees on %q: %s vs %s", key, got, p)
+		}
+		if got := byAdd.GetPrimary(key); got != p {
+			t.Fatalf("add-order ring disagrees on %q: %s vs %s", key, got, p)
+		}
+	}
+}

weed/s3api/s3api_server.go

@@ -25,6 +25,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/iam/policy"
 	"github.com/seaweedfs/seaweedfs/weed/iam/sts"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/s3_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
@@ -95,6 +96,8 @@ type S3ApiServer struct {
 	stsHandlers           *STSHandlers    // STS HTTP handlers for AssumeRoleWithWebIdentity
 	cipher                bool            // encrypt data on volume servers
 	newObjectWriteLock    func(bucket, object string) objectWriteLock
+	// objectWriteLockClient resolves a key's owner filer for route-by-key.
+	objectWriteLockClient *cluster.LockClient
 	// Shared ReaderCache used by the S3 GET streaming path. It lives for the
 	// lifetime of the server so that concurrent and repeat reads share a
 	// single in-flight download per chunk, and so that no per-request
@@ -152,6 +155,8 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl
 	// Uses the battle-tested vidMap with filer-based lookups
 	// Supports multiple filer addresses with automatic failover for high availability
 	var filerClient *wdclient.FilerClient
+	var masterClient *wdclient.MasterClient
+	var objectWriteLockClient *cluster.LockClient
 	if len(option.Masters) > 0 {
 		// Enable filer discovery via master
 		masterMap := make(map[string]pb.ServerAddress)
@@ -162,7 +167,22 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl
 		if clientHost == "0.0.0.0" || clientHost == "" {
 			clientHost = util.DetectedHostAddress()
 		}
-		masterClient := wdclient.NewMasterClient(option.GrpcDialOption, option.FilerGroup, cluster.S3Type, pb.ServerAddress(util.JoinHostPort(clientHost, option.GrpcPort)), "", "", *pb.NewServiceDiscoveryFromMap(masterMap))
+		masterClient = wdclient.NewMasterClient(option.GrpcDialOption, option.FilerGroup, cluster.S3Type, pb.ServerAddress(util.JoinHostPort(clientHost, option.GrpcPort)), "", "", *pb.NewServiceDiscoveryFromMap(masterMap))
+		// Build the object-write lock client and subscribe to the master's
+		// lock-ring updates BEFORE starting the master loop, so the initial
+		// LockRingUpdate sent on connect isn't dropped (the master only delivers
+		// it once per connect). The masterClient already filters updates to this
+		// server's filer group.
+		if len(option.Filers) > 0 {
+			objectWriteLockClient = cluster.NewLockClient(option.GrpcDialOption, option.Filers[0])
+			masterClient.SetOnLockRingUpdateFn(func(update *master_pb.LockRingUpdate) {
+				servers := make([]pb.ServerAddress, 0, len(update.Servers))
+				for _, s := range update.Servers {
+					servers = append(servers, pb.ServerAddress(s))
+				}
+				objectWriteLockClient.SetRing(servers, update.Version)
+			})
+		}
 		// Start the master client connection loop - required for GetMaster() to work
 		go masterClient.KeepConnectedToMaster(context.Background())
 
@@ -263,7 +283,12 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl
 	}
 
 	if len(option.Filers) > 0 {
-		objectWriteLockClient := cluster.NewLockClient(option.GrpcDialOption, option.Filers[0])
+		// Reuse the lock client built in the masters block (already subscribed to
+		// ring updates); create a plain one when no masters are configured.
+		if objectWriteLockClient == nil {
+			objectWriteLockClient = cluster.NewLockClient(option.GrpcDialOption, option.Filers[0])
+		}
+		s3ApiServer.objectWriteLockClient = objectWriteLockClient
 		s3ApiServer.newObjectWriteLock = func(bucket, object string) objectWriteLock {
 			lockKey := fmt.Sprintf("s3.object.write:%s", s3ApiServer.toFilerPath(bucket, object))
 			owner := fmt.Sprintf("s3api-%d", s3ApiServer.randomClientId)

weed/server/master_grpc_server.go

@@ -393,7 +393,13 @@ func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServ
 }
 
 func (ms *MasterServer) initialLockRingUpdate(clientType string, filerGroup string) *master_pb.KeepConnectedResponse {
-	if clientType != cluster.FilerType || ms.LockRingManager == nil {
+	if ms.LockRingManager == nil {
+		return nil
+	}
+	// Filers are ring members; S3 gateways are lock clients that need the same
+	// view to dial a key's primary directly. Both get the initial snapshot;
+	// later membership changes already broadcast to every connected client.
+	if clientType != cluster.FilerType && clientType != cluster.S3Type {
 		return nil
 	}