ObjectTransaction: filer-side forwarding via route_key (#9659)

A non-owner filer forwards the whole transaction to the ring owner of route_key, so the owner's per-path lock stays the single serialization point even when the caller's ring view is stale. is_moved bounds forwarding to one hop. The gateway stamps route_key on every routed builder via the shared objectRouteKey helper. Completes taking S3 object mutations off the distributed lock.

other/java/client/src/main/proto/filer.proto

@@ -342,8 +342,9 @@ message Recompute {
 // ObjectTransactionRequest applies an ordered list of mutations atomically with
 // respect to other writers of the same object, by holding the filer's per-path
 // lock on lock_key for the whole transaction. The optional condition is checked
-// first, against the entry at lock_key. Callers must route the object's writes
-// to its owner filer for the lock to be authoritative.
+// first, against condition_key when set, else lock_key. Callers set route_key to
+// the object's stable owner ring key; a filer that is not the owner forwards the
+// transaction one hop to the owner, so a stale ring view is tolerated.
 message ObjectTransactionRequest {
     string lock_key = 1;               // object path to lock and to evaluate the condition against
     WriteCondition condition = 2;      // optional precondition, checked under the lock
@@ -351,6 +352,8 @@ message ObjectTransactionRequest {
     bool is_from_other_cluster = 4;
     repeated int32 signatures = 5;
     string condition_key = 6;
+    string route_key = 7;              // ring key identifying the owner filer; a non-owner forwards the whole transaction to it
+    bool is_moved = 8;                 // set on a forwarded transaction so the receiver applies it locally instead of forwarding again
 }
 
 message ObjectTransactionResponse {

weed/pb/filer.proto

@@ -342,8 +342,9 @@ message Recompute {
 // ObjectTransactionRequest applies an ordered list of mutations atomically with
 // respect to other writers of the same object, by holding the filer's per-path
 // lock on lock_key for the whole transaction. The optional condition is checked
-// first, against the entry at lock_key. Callers must route the object's writes
-// to its owner filer for the lock to be authoritative.
+// first, against condition_key when set, else lock_key. Callers set route_key to
+// the object's stable owner ring key; a filer that is not the owner forwards the
+// transaction one hop to the owner, so a stale ring view is tolerated.
 message ObjectTransactionRequest {
     string lock_key = 1;               // object path to lock and to evaluate the condition against
     WriteCondition condition = 2;      // optional precondition, checked under the lock
@@ -351,6 +352,8 @@ message ObjectTransactionRequest {
     bool is_from_other_cluster = 4;
     repeated int32 signatures = 5;
     string condition_key = 6;          // if set, evaluate the condition against this entry instead of lock_key (still locking lock_key)
+    string route_key = 7;              // ring key identifying the owner filer; a non-owner forwards the whole transaction to it
+    bool is_moved = 8;                 // set on a forwarded transaction so the receiver applies it locally instead of forwarding again
 }
 
 message ObjectTransactionResponse {

weed/pb/filer_pb/filer.pb.go

@@ -1668,8 +1668,9 @@ func (x *Recompute) GetExcludeName() string {
 // ObjectTransactionRequest applies an ordered list of mutations atomically with
 // respect to other writers of the same object, by holding the filer's per-path
 // lock on lock_key for the whole transaction. The optional condition is checked
-// first, against the entry at lock_key. Callers must route the object's writes
-// to its owner filer for the lock to be authoritative.
+// first, against condition_key when set, else lock_key. Callers set route_key to
+// the object's stable owner ring key; a filer that is not the owner forwards the
+// transaction one hop to the owner, so a stale ring view is tolerated.
 type ObjectTransactionRequest struct {
 	state              protoimpl.MessageState `protogen:"open.v1"`
 	LockKey            string                 `protobuf:"bytes,1,opt,name=lock_key,json=lockKey,proto3" json:"lock_key,omitempty"` // object path to lock and to evaluate the condition against
@@ -1678,6 +1679,8 @@ type ObjectTransactionRequest struct {
 	IsFromOtherCluster bool                   `protobuf:"varint,4,opt,name=is_from_other_cluster,json=isFromOtherCluster,proto3" json:"is_from_other_cluster,omitempty"`
 	Signatures         []int32                `protobuf:"varint,5,rep,packed,name=signatures,proto3" json:"signatures,omitempty"`
 	ConditionKey       string                 `protobuf:"bytes,6,opt,name=condition_key,json=conditionKey,proto3" json:"condition_key,omitempty"` // if set, evaluate the condition against this entry instead of lock_key (still locking lock_key)
+	RouteKey           string                 `protobuf:"bytes,7,opt,name=route_key,json=routeKey,proto3" json:"route_key,omitempty"`             // ring key identifying the owner filer; a non-owner forwards the whole transaction to it
+	IsMoved            bool                   `protobuf:"varint,8,opt,name=is_moved,json=isMoved,proto3" json:"is_moved,omitempty"`               // set on a forwarded transaction so the receiver applies it locally instead of forwarding again
 	unknownFields      protoimpl.UnknownFields
 	sizeCache          protoimpl.SizeCache
 }
@@ -1754,6 +1757,20 @@ func (x *ObjectTransactionRequest) GetConditionKey() string {
 	return ""
 }
 
+func (x *ObjectTransactionRequest) GetRouteKey() string {
+	if x != nil {
+		return x.RouteKey
+	}
+	return ""
+}
+
+func (x *ObjectTransactionRequest) GetIsMoved() bool {
+	if x != nil {
+		return x.IsMoved
+	}
+	return false
+}
+
 type ObjectTransactionResponse struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	Error         string                 `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"`
@@ -6448,7 +6465,7 @@ const file_filer_proto_rawDesc = "" +
 	"\fexclude_name\x18\t \x01(\tR\vexcludeName\x1a?\n" +
 	"\x11CopyExtendedEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
-	"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\x9d\x02\n" +
+	"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xd5\x02\n" +
 	"\x18ObjectTransactionRequest\x12\x19\n" +
 	"\block_key\x18\x01 \x01(\tR\alockKey\x126\n" +
 	"\tcondition\x18\x02 \x01(\v2\x18.filer_pb.WriteConditionR\tcondition\x126\n" +
@@ -6457,7 +6474,9 @@ const file_filer_proto_rawDesc = "" +
 	"\n" +
 	"signatures\x18\x05 \x03(\x05R\n" +
 	"signatures\x12#\n" +
-	"\rcondition_key\x18\x06 \x01(\tR\fconditionKey\"f\n" +
+	"\rcondition_key\x18\x06 \x01(\tR\fconditionKey\x12\x1b\n" +
+	"\troute_key\x18\a \x01(\tR\brouteKey\x12\x19\n" +
+	"\bis_moved\x18\b \x01(\bR\aisMoved\"f\n" +
 	"\x19ObjectTransactionResponse\x12\x14\n" +
 	"\x05error\x18\x01 \x01(\tR\x05error\x123\n" +
 	"\n" +

weed/s3api/filer_multipart.go

@@ -559,6 +559,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
 	// object already carries this UploadId), so the lock is not needed to dedupe
 	// retries. With no owner yet (no ring), keep the lock as the bootstrap path.
 	owner := s3a.objectWriteOwner(*input.Bucket, *input.Key)
+	routeKey := s3a.objectRouteKey(*input.Bucket, *input.Key)
 	completionBody := func() s3err.ErrorCode {
 		var prepCode s3err.ErrorCode
 		completionState, output, prepCode = s3a.prepareMultipartCompletionState(r, input, uploadDirectory, entryName, dirName, completedPartNumbers, completedPartMap, maxPartNo)
@@ -688,7 +689,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
 
 		if versioningState == s3_constants.VersioningSuspended {
 			// For suspended versioning, add "null" version ID metadata and return "null" version ID
-			if err := s3a.writeMultipartObject(owner, dirName, entryName, completionState.finalParts, func(entry *filer_pb.Entry) {
+			if err := s3a.writeMultipartObject(owner, routeKey, dirName, entryName, completionState.finalParts, func(entry *filer_pb.Entry) {
 				if entry.Extended == nil {
 					entry.Extended = make(map[string][]byte)
 				}
@@ -752,7 +753,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
 		}
 
 		// For non-versioned buckets, create main object file
-		if err := s3a.writeMultipartObject(owner, dirName, entryName, completionState.finalParts, func(entry *filer_pb.Entry) {
+		if err := s3a.writeMultipartObject(owner, routeKey, dirName, entryName, completionState.finalParts, func(entry *filer_pb.Entry) {
 			if entry.Extended == nil {
 				entry.Extended = make(map[string][]byte)
 			}

weed/s3api/s3api_bucket_config.go

@@ -573,6 +573,7 @@ func (s3a *S3ApiServer) patchBucketEntry(bucket string, m *filer_pb.ObjectMutati
 	m.Name = bucket
 	req := &filer_pb.ObjectTransactionRequest{
 		LockKey:   bucketPath,
+		RouteKey:  objectWriteRouteKeyPrefix + bucketPath,
 		Mutations: []*filer_pb.ObjectMutation{m},
 	}
 	txn := func(client filer_pb.SeaweedFilerClient) error {
@@ -586,7 +587,7 @@ func (s3a *S3ApiServer) patchBucketEntry(bucket string, m *filer_pb.ObjectMutati
 		return nil
 	}
 	if s3a.objectWriteLockClient != nil {
-		if owner := s3a.objectWriteLockClient.PrimaryForKey("s3.object.write:" + bucketPath); owner != "" {
+		if owner := s3a.objectWriteLockClient.PrimaryForKey(objectWriteRouteKeyPrefix + bucketPath); owner != "" {
 			return pb.WithFilerClient(false, 0, owner, s3a.option.GrpcDialOption, txn)
 		}
 	}

weed/s3api/s3api_object_handlers_put.go

@@ -841,7 +841,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 	routed := false
 	if owner := s3a.routableWriteOwner(bucket, object); owner != "" {
 		if cond, ok := routeWriteCondition(r, uniqueWritePath); ok {
-			resp, err := s3a.routedPut(owner, filePath, entry, cond)
+			resp, err := s3a.routedPut(owner, s3a.objectRouteKey(bucket, object), filePath, entry, cond)
 			switch {
 			case err != nil:
 				glog.Warningf("putToFiler: routed PUT to %s failed for %s, falling back to lock: %v", owner, filePath, err)

weed/s3api/s3api_object_routed_write.go

@@ -15,6 +15,19 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
+// objectWriteRouteKeyPrefix namespaces an object's full path into the ring key
+// used to resolve and forward its writes. Shared by every routed builder so the
+// gateway and filer hash the same key.
+const objectWriteRouteKeyPrefix = "s3.object.write:"
+
+// objectRouteKey is the ring key the gateway hashes to resolve an object's owner
+// filer. It is also sent as route_key on each routed transaction, so a non-owner
+// filer (reached because the gateway's ring view was stale) forwards the
+// transaction to the owner. All of an object's writes share this key.
+func (s3a *S3ApiServer) objectRouteKey(bucket, object string) string {
+	return objectWriteRouteKeyPrefix + s3a.toFilerPath(bucket, object)
+}
+
 // routableWriteOwner returns the owner filer for an object's writes, or "" to
 // keep them on the distributed lock. All writes to one object (versioned,
 // suspended, non-versioned) share the owner. Any lookup error falls back.
@@ -25,7 +38,7 @@ func (s3a *S3ApiServer) routableWriteOwner(bucket, object string) pb.ServerAddre
 	// Object-lock PUTs route: a versioned PUT creates a new version (never an
 	// overwrite of a locked one), and a non-versioned overwrite is WORM-checked
 	// gateway-side before dispatch. WORM-checked deletes use routedObjectOwner.
-	return s3a.objectWriteLockClient.PrimaryForKey(fmt.Sprintf("s3.object.write:%s", s3a.toFilerPath(bucket, object)))
+	return s3a.objectWriteLockClient.PrimaryForKey(s3a.objectRouteKey(bucket, object))
 }
 
 // routedObjectOwner is routableWriteOwner restricted to non-versioned,
@@ -148,9 +161,10 @@ func (s3a *S3ApiServer) objectTxnOnFiler(owner pb.ServerAddress, req *filer_pb.O
 // routedPut writes an object entry as a one-mutation ObjectTransaction on the
 // owner filer. lock_key is the object's full path so the transaction shares the
 // per-path lock with a concurrent create or delete of the same key.
-func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, filePath string, entry *filer_pb.Entry, cond *filer_pb.WriteCondition) (*filer_pb.ObjectTransactionResponse, error) {
+func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, routeKey, filePath string, entry *filer_pb.Entry, cond *filer_pb.WriteCondition) (*filer_pb.ObjectTransactionResponse, error) {
 	return s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{
 		LockKey:   filePath,
+		RouteKey:  routeKey,
 		Condition: cond,
 		Mutations: []*filer_pb.ObjectMutation{{
 			Type:      filer_pb.ObjectMutation_PUT,
@@ -163,7 +177,7 @@ func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, filePath string, entry
 // routedMkFile builds an entry like filer_pb.MkFile and writes it through a
 // routed PUT on the owner filer, for callers that would otherwise mkFile to the
 // default filer (e.g. multipart completion of a non-versioned object).
-func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, parentDir, name string, chunks []*filer_pb.FileChunk, fn func(*filer_pb.Entry)) error {
+func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, routeKey, parentDir, name string, chunks []*filer_pb.FileChunk, fn func(*filer_pb.Entry)) error {
 	now := time.Now().Unix()
 	entry := &filer_pb.Entry{
 		Name: name,
@@ -179,7 +193,7 @@ func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, parentDir, name str
 	if fn != nil {
 		fn(entry)
 	}
-	resp, err := s3a.routedPut(owner, parentDir+"/"+name, entry, nil)
+	resp, err := s3a.routedPut(owner, routeKey, parentDir+"/"+name, entry, nil)
 	if err != nil {
 		return err
 	}
@@ -191,10 +205,11 @@ func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, parentDir, name str
 
 // writeMultipartObject writes a completed multipart object entry, routed to the
 // owner when known (so it serializes with concurrent writes to the same key)
-// and falling back to a plain mkFile otherwise.
-func (s3a *S3ApiServer) writeMultipartObject(owner pb.ServerAddress, dir, name string, chunks []*filer_pb.FileChunk, fn func(*filer_pb.Entry)) error {
+// and falling back to a plain mkFile otherwise. routeKey must be the same key the
+// caller used to resolve owner, so owner selection and forwarding stay consistent.
+func (s3a *S3ApiServer) writeMultipartObject(owner pb.ServerAddress, routeKey, dir, name string, chunks []*filer_pb.FileChunk, fn func(*filer_pb.Entry)) error {
 	if owner != "" {
-		return s3a.routedMkFile(owner, dir, name, chunks, fn)
+		return s3a.routedMkFile(owner, routeKey, dir, name, chunks, fn)
 	}
 	return s3a.mkFile(dir, name, chunks, fn)
 }
@@ -206,6 +221,7 @@ func (s3a *S3ApiServer) routedDelete(owner pb.ServerAddress, bucket, object stri
 	dir, name := fullpath.DirAndName()
 	return s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{
 		LockKey:   string(fullpath),
+		RouteKey:  s3a.objectRouteKey(bucket, object),
 		Condition: cond,
 		Mutations: []*filer_pb.ObjectMutation{{
 			Type:         filer_pb.ObjectMutation_DELETE,
@@ -235,7 +251,8 @@ func (s3a *S3ApiServer) routedMetadataReplace(owner pb.ServerAddress, bucket, ob
 		}
 	}
 	resp, err := s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{
-		LockKey: string(fullpath),
+		LockKey:  string(fullpath),
+		RouteKey: s3a.objectRouteKey(bucket, object),
 		Mutations: []*filer_pb.ObjectMutation{{
 			Type:           filer_pb.ObjectMutation_PATCH_EXTENDED,
 			Directory:      dir,

weed/s3api/s3api_object_versioned_finalize.go

@@ -20,7 +20,7 @@ func (s3a *S3ApiServer) objectWriteOwner(bucket, object string) pb.ServerAddress
 	if s3a.objectWriteLockClient == nil {
 		return ""
 	}
-	return s3a.objectWriteLockClient.PrimaryForKey("s3.object.write:" + s3a.toFilerPath(bucket, object))
+	return s3a.objectWriteLockClient.PrimaryForKey(s3a.objectRouteKey(bucket, object))
 }
 
 // latestPointerRecompute builds the RECOMPUTE_LATEST mutation that re-derives an
@@ -65,6 +65,7 @@ func (s3a *S3ApiServer) latestPointerRecompute(bucket, object string, useInverte
 func (s3a *S3ApiServer) routedVersionedFinalize(owner pb.ServerAddress, bucket, object string, useInvertedFormat bool) s3err.ErrorCode {
 	req := &filer_pb.ObjectTransactionRequest{
 		LockKey:   s3a.toFilerPath(bucket, object),
+		RouteKey:  s3a.objectRouteKey(bucket, object),
 		Mutations: []*filer_pb.ObjectMutation{s3a.latestPointerRecompute(bucket, object, useInvertedFormat, "", true)},
 	}
 	resp, err := s3a.objectTxnOnFiler(owner, req)
@@ -116,6 +117,7 @@ func (s3a *S3ApiServer) routedDeleteSpecificVersion(owner pb.ServerAddress, buck
 	cond := wormDeleteCondition(worm, bypass)
 	req := &filer_pb.ObjectTransactionRequest{
 		LockKey:      s3a.toFilerPath(bucket, object),
+		RouteKey:     s3a.objectRouteKey(bucket, object),
 		ConditionKey: versionsPath + "/" + versionFileName,
 		Condition:    cond,
 		Mutations: []*filer_pb.ObjectMutation{
@@ -148,6 +150,7 @@ func (s3a *S3ApiServer) routedDeleteNullVersion(owner pb.ServerAddress, bucket,
 	dir, name := fullpath.DirAndName()
 	resp, err := s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{
 		LockKey:   string(fullpath),
+		RouteKey:  s3a.objectRouteKey(bucket, object),
 		Condition: wormDeleteCondition(worm, bypass),
 		Mutations: []*filer_pb.ObjectMutation{
 			{Type: filer_pb.ObjectMutation_DELETE, Directory: dir, Name: name, IsDeleteData: true},

weed/s3api/s3api_server.go

@@ -290,7 +290,7 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl
 		}
 		s3ApiServer.objectWriteLockClient = objectWriteLockClient
 		s3ApiServer.newObjectWriteLock = func(bucket, object string) objectWriteLock {
-			lockKey := fmt.Sprintf("s3.object.write:%s", s3ApiServer.toFilerPath(bucket, object))
+			lockKey := objectWriteRouteKeyPrefix + s3ApiServer.toFilerPath(bucket, object)
 			owner := fmt.Sprintf("s3api-%d", s3ApiServer.randomClientId)
 			lock := objectWriteLockClient.NewShortLivedLock(lockKey, owner)
 			if err := lock.AttemptToLock(objectWriteLockTTL); err != nil {

weed/server/filer_grpc_server.go

@@ -15,6 +15,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/operation"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
@@ -256,6 +257,39 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje
 		return &filer_pb.ObjectTransactionResponse{Error: "lock_key is required"}, nil
 	}
 
+	// Route-by-key: if this filer is not the ring owner of route_key, forward the
+	// whole transaction to the owner so its per-path lock is the single
+	// 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,
+			}
+			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 {
+				var e error
+				resp, e = client.ObjectTransaction(ctx, forwarded)
+				return e
+			})
+			if err != nil {
+				return &filer_pb.ObjectTransactionResponse{}, err
+			}
+			return resp, nil
+		}
+	}
+
 	lockPath := util.FullPath(req.LockKey)
 	pathLock := fs.entryLockTable.AcquireLock("ObjectTransaction", lockPath, util.ExclusiveLock)
 	defer fs.entryLockTable.ReleaseLock(lockPath, pathLock)

weed/server/filer_grpc_server_object_txn_test.go

@@ -2,14 +2,19 @@ package weed_server
 
 import (
 	"context"
+	"net"
 	"strconv"
 	"testing"
 	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager"
 	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"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/util"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 )
 
 func newTxnTestServer(seed map[string]*filer.Entry) (*FilerServer, *renameTestStore) {
@@ -595,3 +600,121 @@ func TestObjectTransactionPatchTouchMtime(t *testing.T) {
 		t.Errorf("new meta not set: %v", e.Extended)
 	}
 }
+
+// withRing attaches a Dlm whose ring contains exactly the given servers and sets
+// the filer's own host, so route_key resolution in ObjectTransaction is decided
+// by who owns the single-server ring.
+func withRing(fs *FilerServer, self pb.ServerAddress, servers ...pb.ServerAddress) {
+	dlm := lock_manager.NewDistributedLockManager(self)
+	dlm.LockRing.SetSnapshot(servers, 1)
+	fs.filer.Dlm = dlm
+	fs.option.Host = self
+}
+
+// When this filer owns route_key, the transaction applies locally rather than
+// forwarding to itself.
+func TestObjectTransactionRouteKeyOwnerAppliesLocally(t *testing.T) {
+	self := pb.ServerAddress("localhost:1")
+	fs, store := newTxnTestServer(map[string]*filer.Entry{
+		"/buckets/b/obj": {FullPath: "/buckets/b/obj", Attr: filer.Attr{Inode: 1, Mode: 0644}},
+	})
+	withRing(fs, self, self)
+
+	resp, err := fs.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{
+		LockKey:  "/buckets/b/obj",
+		RouteKey: "s3.object.write:/buckets/b/obj",
+		Mutations: []*filer_pb.ObjectMutation{{
+			Type: filer_pb.ObjectMutation_PATCH_EXTENDED, Directory: "/buckets/b", Name: "obj",
+			SetExtended: map[string][]byte{"X-Amz-Meta-k": []byte("v")},
+		}},
+	})
+	if err != nil || resp.Error != "" {
+		t.Fatalf("txn failed: err=%v resp=%q", err, resp.Error)
+	}
+	if string(store.entries["/buckets/b/obj"].Extended["X-Amz-Meta-k"]) != "v" {
+		t.Errorf("mutation should have applied locally: %v", store.entries["/buckets/b/obj"].Extended)
+	}
+}
+
+// A forwarded transaction (is_moved) applies locally even when the ring names a
+// different owner: is_moved bounds forwarding to a single hop, so two filers that
+// disagree on the owner during a ring change cannot loop. If is_moved were
+// ignored, this would attempt to dial the bogus owner instead of applying.
+func TestObjectTransactionIsMovedSkipsForward(t *testing.T) {
+	self := pb.ServerAddress("localhost:1")
+	other := pb.ServerAddress("localhost:2")
+	fs, store := newTxnTestServer(map[string]*filer.Entry{
+		"/buckets/b/obj": {FullPath: "/buckets/b/obj", Attr: filer.Attr{Inode: 1, Mode: 0644}},
+	})
+	withRing(fs, self, other) // ring owner is "other", not self
+
+	resp, err := fs.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{
+		LockKey:  "/buckets/b/obj",
+		RouteKey: "s3.object.write:/buckets/b/obj",
+		IsMoved:  true,
+		Mutations: []*filer_pb.ObjectMutation{{
+			Type: filer_pb.ObjectMutation_PATCH_EXTENDED, Directory: "/buckets/b", Name: "obj",
+			SetExtended: map[string][]byte{"X-Amz-Meta-k": []byte("v")},
+		}},
+	})
+	if err != nil || resp.Error != "" {
+		t.Fatalf("txn failed: err=%v resp=%q", err, resp.Error)
+	}
+	if string(store.entries["/buckets/b/obj"].Extended["X-Amz-Meta-k"]) != "v" {
+		t.Errorf("forwarded txn should apply locally: %v", store.entries["/buckets/b/obj"].Extended)
+	}
+}
+
+// End-to-end forward hop: a non-owner filer dials the ring owner and the owner
+// applies the transaction. The owner's own ring points back at the (bogus)
+// sender, so it would re-forward and fail to dial unless is_moved is set on the
+// forwarded request — making this also assert that one-hop bound over the wire.
+func TestObjectTransactionForwardsToOwner(t *testing.T) {
+	owner, ownerStore := newTxnTestServer(map[string]*filer.Entry{
+		"/buckets/b/obj": {FullPath: "/buckets/b/obj", Attr: filer.Attr{Inode: 1, Mode: 0644}},
+	})
+
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	// Pin the grpc port to the real listener (ToGrpcAddress otherwise adds the
+	// +10000 convention, which dials nothing).
+	port := lis.Addr().(*net.TCPAddr).Port
+	ownerAddr := pb.NewServerAddressWithGrpcPort(lis.Addr().String(), port)
+	sender := pb.ServerAddress("127.0.0.1:1") // bogus: nothing listens here
+
+	// owner's ring points back at the sender; only is_moved keeps it from
+	// re-forwarding to (and failing to dial) that bogus address.
+	withRing(owner, ownerAddr, sender)
+	owner.grpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials())
+
+	srv := grpc.NewServer()
+	filer_pb.RegisterSeaweedFilerServer(srv, owner)
+	go srv.Serve(lis)
+	t.Cleanup(srv.Stop)
+
+	self, selfStore := newTxnTestServer(nil)
+	withRing(self, sender, ownerAddr) // ring owner is the real owner; self forwards
+	self.grpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials())
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	resp, err := self.ObjectTransaction(ctx, &filer_pb.ObjectTransactionRequest{
+		LockKey:  "/buckets/b/obj",
+		RouteKey: "s3.object.write:/buckets/b/obj",
+		Mutations: []*filer_pb.ObjectMutation{{
+			Type: filer_pb.ObjectMutation_PATCH_EXTENDED, Directory: "/buckets/b", Name: "obj",
+			SetExtended: map[string][]byte{"X-Amz-Meta-k": []byte("v")},
+		}},
+	})
+	if err != nil || resp.Error != "" {
+		t.Fatalf("forwarded txn failed: err=%v resp=%q", err, resp.Error)
+	}
+	if string(ownerStore.entries["/buckets/b/obj"].Extended["X-Amz-Meta-k"]) != "v" {
+		t.Errorf("owner should have applied the forwarded mutation: %v", ownerStore.entries["/buckets/b/obj"].Extended)
+	}
+	if _, ok := selfStore.entries["/buckets/b/obj"]; ok {
+		t.Errorf("non-owner must forward, not apply locally")
+	}
+}