diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto index c7fa357fe..f5b91e1a4 100644 --- a/other/java/client/src/main/proto/filer.proto +++ b/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 { diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index 5f3d2f407..5203bf619 100644 --- a/weed/pb/filer.proto +++ b/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 { diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 555958632..b6f8b29a9 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/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" + diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index 48b26770b..dc2c8c81f 100644 --- a/weed/s3api/filer_multipart.go +++ b/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) } diff --git a/weed/s3api/s3api_bucket_config.go b/weed/s3api/s3api_bucket_config.go index f12f54200..363903431 100644 --- a/weed/s3api/s3api_bucket_config.go +++ b/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) } } diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 8605450d1..b941f6af9 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/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) diff --git a/weed/s3api/s3api_object_routed_write.go b/weed/s3api/s3api_object_routed_write.go index 4febcc4a5..1a4b072ff 100644 --- a/weed/s3api/s3api_object_routed_write.go +++ b/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, diff --git a/weed/s3api/s3api_object_versioned_finalize.go b/weed/s3api/s3api_object_versioned_finalize.go index 41627b965..7844e6509 100644 --- a/weed/s3api/s3api_object_versioned_finalize.go +++ b/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}, diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go index 2d18f76d5..af4ed6785 100644 --- a/weed/s3api/s3api_server.go +++ b/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 { diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index 5d9e4f5ef..88ca80e20 100644 --- a/weed/server/filer_grpc_server.go +++ b/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) diff --git a/weed/server/filer_grpc_server_object_txn_test.go b/weed/server/filer_grpc_server_object_txn_test.go index 05c03cacc..11502be64 100644 --- a/weed/server/filer_grpc_server_object_txn_test.go +++ b/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") + } +}