diff --git a/weed/admin/dash/admin_server.go b/weed/admin/dash/admin_server.go index d4485bfcf..c77ab3c39 100644 --- a/weed/admin/dash/admin_server.go +++ b/weed/admin/dash/admin_server.go @@ -14,6 +14,7 @@ import ( "sync" "time" + "github.com/google/uuid" "github.com/seaweedfs/seaweedfs/weed/admin/maintenance" adminplugin "github.com/seaweedfs/seaweedfs/weed/admin/plugin" "github.com/seaweedfs/seaweedfs/weed/cluster" @@ -1155,6 +1156,7 @@ func (s *AdminServer) SetBucketLifecycle(bucketName string, rules []BucketLifecy bucketPath := filerConfig.BucketsPath + "/" + bucketName resp, err := client.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{ LockKey: bucketPath, + RequestId: uuid.NewString(), RouteKey: s3_constants.ObjectWriteRouteKeyPrefix + bucketPath, Mutations: []*filer_pb.ObjectMutation{bucketLifecycleMutation(filerConfig.BucketsPath, bucketName, lifecycleXML)}, }) diff --git a/weed/admin/dash/bucket_policy.go b/weed/admin/dash/bucket_policy.go index 31a14f614..ebc1b397a 100644 --- a/weed/admin/dash/bucket_policy.go +++ b/weed/admin/dash/bucket_policy.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" + "github.com/google/uuid" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api" "github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine" @@ -126,6 +127,7 @@ func (s *AdminServer) writeBucketPolicy(bucketName string, policyJSON []byte) er bucketPath := filerConfig.BucketsPath + "/" + bucketName resp, err := client.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{ LockKey: bucketPath, + RequestId: uuid.NewString(), RouteKey: s3_constants.ObjectWriteRouteKeyPrefix + bucketPath, Mutations: []*filer_pb.ObjectMutation{bucketPolicyMutation(filerConfig.BucketsPath, bucketName, policyJSON)}, }) diff --git a/weed/iam/integration/oidc_provider_store.go b/weed/iam/integration/oidc_provider_store.go index 0eaea1f4c..76ac002d0 100644 --- a/weed/iam/integration/oidc_provider_store.go +++ b/weed/iam/integration/oidc_provider_store.go @@ -14,6 +14,7 @@ import ( "sync" "time" + "github.com/google/uuid" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -563,8 +564,9 @@ func (f *FilerOIDCProviderStore) UpdateProvider(ctx context.Context, filerAddres func (f *FilerOIDCProviderStore) deleteProviderEntryIfUnchanged(ctx context.Context, client filer_pb.SeaweedFilerClient, entry *filer_pb.Entry) (bool, error) { fullPath := f.basePath + "/" + entry.Name resp, err := client.ObjectTransaction(ctx, &filer_pb.ObjectTransactionRequest{ - LockKey: fullPath, - RouteKey: s3_constants.ObjectWriteRouteKeyPrefix + fullPath, + LockKey: fullPath, + RequestId: uuid.NewString(), + RouteKey: s3_constants.ObjectWriteRouteKeyPrefix + fullPath, Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{ Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: entry, }}}, diff --git a/weed/iam/integration/role_store.go b/weed/iam/integration/role_store.go index 27ff79f4d..f418b46dc 100644 --- a/weed/iam/integration/role_store.go +++ b/weed/iam/integration/role_store.go @@ -11,6 +11,7 @@ import ( "sync" "time" + "github.com/google/uuid" "github.com/karlseguin/ccache/v2" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/iam/policy" @@ -380,8 +381,9 @@ func (f *FilerRoleStore) UpdateRole(ctx context.Context, filerAddress string, ro func (f *FilerRoleStore) deleteRoleEntryIfUnchanged(ctx context.Context, client filer_pb.SeaweedFilerClient, entry *filer_pb.Entry) (bool, error) { fullPath := f.basePath + "/" + entry.Name resp, err := client.ObjectTransaction(ctx, &filer_pb.ObjectTransactionRequest{ - LockKey: fullPath, - RouteKey: s3_constants.ObjectWriteRouteKeyPrefix + fullPath, + LockKey: fullPath, + RequestId: uuid.NewString(), + RouteKey: s3_constants.ObjectWriteRouteKeyPrefix + fullPath, Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{ Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: entry, }}}, diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index bfa1de291..0abe60c55 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -392,6 +392,7 @@ message ObjectTransactionRequest { 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 + string request_id = 9; // idempotency token: the owner records the outcome and replays it on a retried request instead of re-applying string condition_alt_key = 10; // if the condition's entry is absent, evaluate it against this entry instead (e.g. null slot, else latest version) } diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 1d82746e2..0c9190f81 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.6 -// protoc v3.21.12 +// protoc v7.35.0 // source: filer.proto package filer_pb @@ -1814,6 +1814,7 @@ type ObjectTransactionRequest struct { 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 + RequestId string `protobuf:"bytes,9,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` // idempotency token: the owner records the outcome and replays it on a retried request instead of re-applying ConditionAltKey string `protobuf:"bytes,10,opt,name=condition_alt_key,json=conditionAltKey,proto3" json:"condition_alt_key,omitempty"` // if the condition's entry is absent, evaluate it against this entry instead (e.g. null slot, else latest version) unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -1905,6 +1906,13 @@ func (x *ObjectTransactionRequest) GetIsMoved() bool { return false } +func (x *ObjectTransactionRequest) GetRequestId() string { + if x != nil { + return x.RequestId + } + return "" +} + func (x *ObjectTransactionRequest) GetConditionAltKey() string { if x != nil { return x.ConditionAltKey @@ -7441,7 +7449,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\"\x81\x03\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xa0\x03\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" + @@ -7452,7 +7460,9 @@ const file_filer_proto_rawDesc = "" + "signatures\x12#\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\x12*\n" + + "\bis_moved\x18\b \x01(\bR\aisMoved\x12\x1d\n" + + "\n" + + "request_id\x18\t \x01(\tR\trequestId\x12*\n" + "\x11condition_alt_key\x18\n" + " \x01(\tR\x0fconditionAltKey\"f\n" + "\x19ObjectTransactionResponse\x12\x14\n" + diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index e11dd0045..cac2d01f9 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/weed/pb/filer_pb/filer_vtproto.pb.go @@ -1574,6 +1574,13 @@ func (m *ObjectTransactionRequest) MarshalToSizedBufferVT(dAtA []byte) (int, err i-- dAtA[i] = 0x52 } + if len(m.RequestId) > 0 { + i -= len(m.RequestId) + copy(dAtA[i:], m.RequestId) + i = protohelpers.EncodeVarint(dAtA, i, uint64(len(m.RequestId))) + i-- + dAtA[i] = 0x4a + } if m.IsMoved { i-- if m.IsMoved { @@ -7195,6 +7202,10 @@ func (m *ObjectTransactionRequest) SizeVT() (n int) { if m.IsMoved { n += 2 } + l = len(m.RequestId) + if l > 0 { + n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) + } l = len(m.ConditionAltKey) if l > 0 { n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) @@ -13852,6 +13863,38 @@ func (m *ObjectTransactionRequest) UnmarshalVT(dAtA []byte) error { } } m.IsMoved = bool(v != 0) + case 9: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field RequestId", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return protohelpers.ErrInvalidLength + } + postIndex := iNdEx + intStringLen + if postIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.RequestId = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex case 10: if wireType != 2 { return fmt.Errorf("proto: wrong wireType = %d for field ConditionAltKey", wireType) diff --git a/weed/s3api/s3api_object_routed_write.go b/weed/s3api/s3api_object_routed_write.go index 7a323f1f5..5e315aa87 100644 --- a/weed/s3api/s3api_object_routed_write.go +++ b/weed/s3api/s3api_object_routed_write.go @@ -9,6 +9,7 @@ import ( "strings" "time" + "github.com/google/uuid" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" @@ -157,6 +158,11 @@ func singleStrongETag(v string) (string, bool) { // 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) { + // One id per logical transaction: failover retries reuse it so the owner + // replays the recorded outcome instead of re-applying the mutations. + if req.RequestId == "" { + req.RequestId = uuid.NewString() + } var resp *filer_pb.ObjectTransactionResponse txn := func(client filer_pb.SeaweedFilerClient) error { var e error diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index a5dfdd7fc..c338e051c 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -344,6 +344,7 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje ConditionKey: req.ConditionKey, RouteKey: req.RouteKey, IsMoved: true, + RequestId: req.RequestId, ConditionAltKey: req.ConditionAltKey, } var resp *filer_pb.ObjectTransactionResponse @@ -367,6 +368,28 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje pathLock := fs.entryLockTable.AcquireLock("ObjectTransaction", lockPath, util.ExclusiveLock) defer fs.entryLockTable.ReleaseLock(lockPath, pathLock) + // Under the same lock the first application committed with, a failover + // replay of the same request replays its recorded outcome instead of + // re-applying mutations over whatever superseded it. + if resp, found := fs.recentObjectTxn(req.RequestId, lockPath); found { + glog.V(2).InfofCtx(ctx, "ObjectTransaction %s: replaying recorded outcome for request %s", req.LockKey, req.RequestId) + return resp, nil + } + + resp := &filer_pb.ObjectTransactionResponse{} + // Outcomes are replayed rather than re-run: a mutation error may follow a + // partial apply, and a retry that re-runs could write the stale mutation + // over whatever superseded it, while replaying the recorded failure at + // least matches what the first attempt reported. The one exception is the + // condition lookup below: it fails before any mutation runs, so the retry + // must re-evaluate rather than replay an empty outcome. + record := true + defer func() { + if record { + fs.rememberObjectTxn(req.RequestId, lockPath, resp) + } + }() + if conditionIsSet(req.Condition) { // The condition is evaluated against condition_key when set (e.g. a // version entry whose WORM guards gate the delete), while the lock stays @@ -393,28 +416,102 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje current, conditionPath, findErr = fs.latestVersionedConditionEntry(ctx, util.FullPath(req.ConditionAltKey)) } if findErr != nil && findErr != filer_pb.ErrNotFound { - return &filer_pb.ObjectTransactionResponse{}, fmt.Errorf("ObjectTransaction condition %s: %w", conditionPath, findErr) + record = false // a lookup error decides nothing; a retry must re-run, not replay + return nil, fmt.Errorf("ObjectTransaction condition %s: %w", conditionPath, findErr) } if findErr == filer_pb.ErrNotFound { current = nil } if !writeConditionSatisfied(req.Condition, current) { glog.V(3).InfofCtx(ctx, "ObjectTransaction %s: precondition failed", conditionPath) - return &filer_pb.ObjectTransactionResponse{ - Error: "precondition failed", - ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED, - }, nil + resp.Error = "precondition failed" + resp.ErrorCode = filer_pb.FilerError_PRECONDITION_FAILED + return resp, nil } } for i, m := range req.Mutations { if err := fs.applyObjectMutation(ctx, m, req.IsFromOtherCluster, req.Signatures); err != nil { glog.V(2).InfofCtx(ctx, "ObjectTransaction %s mutation %d (%v): %v", lockPath, i, m.Type, err) - return &filer_pb.ObjectTransactionResponse{Error: fmt.Sprintf("mutation %d: %v", i, err)}, nil + resp.Error = fmt.Sprintf("mutation %d: %v", i, err) + return resp, nil } } - return &filer_pb.ObjectTransactionResponse{}, nil + return resp, nil +} + +const ( + recentObjectTxnTTL = 2 * time.Minute + recentObjectTxnSweepCap = 10000 + recentObjectTxnSweepPeriod = 30 * time.Second +) + +type recentObjectTxn struct { + resp *filer_pb.ObjectTransactionResponse + expiresAt time.Time +} + +// recentObjectTxnKey scopes an idempotency token to the transaction's lock +// path, so a token reused on a different object cannot replay this outcome. +func recentObjectTxnKey(requestID string, lockPath util.FullPath) string { + return requestID + "\x00" + string(lockPath) +} + +// recentObjectTxn returns the recorded outcome of an already-applied request, +// so a failover replay of the same ObjectTransaction does not re-apply it. +func (fs *FilerServer) recentObjectTxn(requestID string, lockPath util.FullPath) (*filer_pb.ObjectTransactionResponse, bool) { + if requestID == "" { + return nil, false + } + fs.recentObjectTxnsMu.Lock() + defer fs.recentObjectTxnsMu.Unlock() + + key := recentObjectTxnKey(requestID, lockPath) + txn, found := fs.recentObjectTxns[key] + if !found { + return nil, false + } + if time.Now().After(txn.expiresAt) { + delete(fs.recentObjectTxns, key) + return nil, false + } + return txn.resp, true +} + +func (fs *FilerServer) rememberObjectTxn(requestID string, lockPath util.FullPath, resp *filer_pb.ObjectTransactionResponse) { + if requestID == "" || resp == nil { + return + } + fs.recentObjectTxnsMu.Lock() + defer fs.recentObjectTxnsMu.Unlock() + if fs.recentObjectTxns == nil { + fs.recentObjectTxns = make(map[string]recentObjectTxn) + } + // A live record is never evicted early: a replay that misses re-applies the + // transaction. The sweep is amortized so a hot cache does not scan on every + // write; between sweeps the map may exceed the cap, still bounded by the + // insert rate over the TTL. + now := time.Now() + if len(fs.recentObjectTxns) >= recentObjectTxnSweepCap { + if now.After(fs.recentObjectTxnsSweepAt) { + fs.recentObjectTxnsSweepAt = now.Add(recentObjectTxnSweepPeriod) + for id, txn := range fs.recentObjectTxns { + if now.After(txn.expiresAt) { + delete(fs.recentObjectTxns, id) + } + } + } + if len(fs.recentObjectTxns) >= recentObjectTxnSweepCap { + // Full of unexpired records: drop the new one rather than evict + // an outcome a replay may still need. + return + } + } + fs.recentObjectTxns[recentObjectTxnKey(requestID, lockPath)] = recentObjectTxn{ + resp: resp, + expiresAt: now.Add(recentObjectTxnTTL), + } } // latestVersionedConditionEntry resolves the object a conditional write should diff --git a/weed/server/filer_grpc_server_object_txn_test.go b/weed/server/filer_grpc_server_object_txn_test.go index 51b1442fb..c73badd7a 100644 --- a/weed/server/filer_grpc_server_object_txn_test.go +++ b/weed/server/filer_grpc_server_object_txn_test.go @@ -2,6 +2,7 @@ package weed_server import ( "context" + "fmt" "net" "os" "strconv" @@ -855,6 +856,139 @@ func TestObjectTransactionForwardsToOwner(t *testing.T) { } } +// A failover replay carries the same request_id. Under the path lock the owner +// replays the recorded outcome instead of re-applying the mutations over a +// newer entry. +func TestObjectTransactionRequestIdDedupesReplay(t *testing.T) { + now := time.Unix(1700000000, 0) + fs, store := newTxnTestServer(t, nil) + + put := func(etag string) *filer_pb.ObjectMutation { + return &filer_pb.ObjectMutation{ + Type: filer_pb.ObjectMutation_PUT, + Directory: "/buckets/bkt", + Entry: &filer_pb.Entry{ + Name: "obj", + Attributes: &filer_pb.FuseAttributes{Mtime: now.Unix(), FileMode: 0644, Inode: 1}, + Extended: map[string][]byte{"etag": []byte(etag)}, + }, + } + } + txn := func(requestID, etag string) *filer_pb.ObjectTransactionRequest { + return &filer_pb.ObjectTransactionRequest{ + LockKey: "/buckets/bkt/obj", + RequestId: requestID, + Mutations: []*filer_pb.ObjectMutation{put(etag)}, + } + } + + respA, err := fs.ObjectTransaction(context.Background(), txn("req-a", "A")) + if err != nil || respA.Error != "" { + t.Fatalf("first put: err=%v resp=%q", err, respA.Error) + } + + // a newer writer lands the same path + respB, err := fs.ObjectTransaction(context.Background(), txn("req-b", "B")) + if err != nil || respB.Error != "" { + t.Fatalf("second put: err=%v resp=%q", err, respB.Error) + } + + // the replayed request replays A's outcome; the stored entry stays B + respReplay, err := fs.ObjectTransaction(context.Background(), txn("req-a", "A")) + if err != nil || respReplay.Error != "" { + t.Fatalf("replayed put: err=%v resp=%q", err, respReplay.Error) + } + entry := store.entries["/buckets/bkt/obj"] + if entry == nil { + t.Fatal("entry missing after replay") + } + if got := string(entry.Extended["etag"]); got != "B" { + t.Fatalf("replayed transaction re-applied the stale entry: etag = %q, want B", got) + } +} + +// A recorded application-level failure replays to the retried request even +// after the world changes enough for the condition to pass, and its mutation +// still does not run. +func TestObjectTransactionRequestIdReplaysRecordedError(t *testing.T) { + now := time.Unix(1700000000, 0) + fs, store := newTxnTestServer(t, nil) + + req := &filer_pb.ObjectTransactionRequest{ + LockKey: "/buckets/bkt/obj", + RequestId: "req-fail", + Condition: &filer_pb.WriteCondition{ + Clauses: []*filer_pb.WriteCondition_Clause{{Kind: filer_pb.WriteCondition_IF_EXISTS}}, + }, + Mutations: []*filer_pb.ObjectMutation{{ + Type: filer_pb.ObjectMutation_PUT, + Directory: "/buckets/bkt", + Entry: &filer_pb.Entry{ + Name: "obj", + Attributes: &filer_pb.FuseAttributes{Mtime: now.Unix(), FileMode: 0644, Inode: 1}, + Extended: map[string][]byte{"etag": []byte("stale")}, + }, + }}, + } + + resp, err := fs.ObjectTransaction(context.Background(), req) + if err != nil || resp.ErrorCode != filer_pb.FilerError_PRECONDITION_FAILED { + t.Fatalf("precondition: err=%v code=%v", err, resp.GetErrorCode()) + } + + // another writer lands the object the precondition wanted + other := &filer_pb.ObjectTransactionRequest{ + LockKey: "/buckets/bkt/obj", + RequestId: "req-other", + Mutations: []*filer_pb.ObjectMutation{{ + Type: filer_pb.ObjectMutation_PUT, + Directory: "/buckets/bkt", + Entry: &filer_pb.Entry{ + Name: "obj", + Attributes: &filer_pb.FuseAttributes{Mtime: now.Unix(), FileMode: 0644, Inode: 2}, + Extended: map[string][]byte{"etag": []byte("live")}, + }, + }}, + } + respOther, err := fs.ObjectTransaction(context.Background(), other) + if err != nil || respOther.Error != "" { + t.Fatalf("other put: err=%v resp=%q", err, respOther.Error) + } + + // the retry replays the recorded failure instead of re-evaluating and + // overwriting the live entry + resp2, err := fs.ObjectTransaction(context.Background(), req) + if err != nil || resp2.ErrorCode != filer_pb.FilerError_PRECONDITION_FAILED { + t.Fatalf("replayed precondition: err=%v code=%v", err, resp2.GetErrorCode()) + } + if got := string(store.entries["/buckets/bkt/obj"].Extended["etag"]); got != "live" { + t.Fatalf("replayed transaction overwrote the live entry: etag = %q", got) + } +} + +// At capacity the cache drops new recordings rather than evicting a live +// outcome a replay may still need. +func TestObjectTransactionReplayCacheDropsNewAtCap(t *testing.T) { + fs, _ := newTxnTestServer(t, nil) + path := util.FullPath("/buckets/bkt/obj") + resp := &filer_pb.ObjectTransactionResponse{} + + for i := 0; i < recentObjectTxnSweepCap; i++ { + fs.rememberObjectTxn(fmt.Sprintf("req-%d", i), path, resp) + } + fs.rememberObjectTxn("req-overflow", path, resp) + + if _, found := fs.recentObjectTxn("req-0", path); !found { + t.Fatal("a live record was evicted to admit a new one") + } + if _, found := fs.recentObjectTxn("req-overflow", path); found { + t.Fatal("overflow record was recorded instead of dropped") + } + if len(fs.recentObjectTxns) > recentObjectTxnSweepCap { + t.Fatalf("cache exceeded cap: %d", len(fs.recentObjectTxns)) + } +} + // A suspended-versioning conditional write falls back to the latest version // under the transaction lock: a real latest satisfies existence checks, and a // delete marker reads as absent. diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index 58748422c..41c725617 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -142,6 +142,13 @@ type FilerServer struct { recentCopyRequestsMu sync.Mutex recentCopyRequests map[string]recentCopyRequest + // outcomes of recently applied object transactions keyed by request_id, so + // a failover replay of the same transaction replays its outcome instead of + // re-applying its mutations over a newer entry + recentObjectTxnsMu sync.Mutex + recentObjectTxns map[string]recentObjectTxn + recentObjectTxnsSweepAt time.Time + // credential manager for IAM operations CredentialManager *credential.CredentialManager @@ -236,6 +243,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) subscribers: make(map[int32]*metadataSubscriber), inFlightDataLimitCond: sync.NewCond(new(sync.Mutex)), recentCopyRequests: make(map[string]recentCopyRequest), + recentObjectTxns: make(map[string]recentObjectTxn), CredentialManager: option.CredentialManager, entryLockTable: util.NewLockTable[util.FullPath](), posixLocks: posixlock.NewManager(),