From e724f8f470bcf4fcfe65876763cdcce74f181522 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 16 Oct 2026 02:00:27 -0700 Subject: [PATCH] filer: resolve the condition fallback under the transaction lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The gateway named the pointed-at version file as condition_alt_key, resolved before the transaction ran: a concurrent version delete could move the latest pointer between the gateway read and the locked check, and a delete-marker latest read as an existing object. The alt key is now the .versions directory and the filer resolves the latest-version pointer under the same lock that serializes the write — a delete marker resolves to absent, and unresolvable pointer state fails closed. --- weed/s3api/s3api_object_handlers_put.go | 10 ++- ...3api_object_handlers_put_ambiguous_test.go | 68 +++++++++++++++++-- weed/server/filer_grpc_server.go | 45 ++++++++++-- .../filer_grpc_server_object_txn_test.go | 53 +++++++++++++++ 4 files changed, 161 insertions(+), 15 deletions(-) diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 850a9f5059..9a6f900a63 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -1838,12 +1838,10 @@ func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, ob // splitting check and write across a gateway lock that routed writes skip. conditionAltKey := "" if headers, headersErr := parseConditionalHeaders(r); headersErr == s3err.ErrNone && headers.isSet { - versionsPath := normalizedObject + s3_constants.VersionsFolder - if versionsEntry, verErr := s3a.getEntry(s3a.bucketDir(bucket), versionsPath); verErr == nil { - if name := string(versionsEntry.Extended[s3_constants.ExtLatestVersionFileNameKey]); name != "" { - conditionAltKey = s3a.toFilerPath(bucket, versionsPath+"/"+name) - } - } + // The filer resolves the latest pointer under the same lock that + // serializes the write, so the condition cannot read a version a + // concurrent delete has already displaced. + conditionAltKey = s3a.toFilerPath(bucket, normalizedObject+s3_constants.VersionsFolder) } etag, errCode, sseMetadata = s3a.putToFiler(r, filePath, body, bucket, normalizedObject, 1, 0, &putFinalize{ conditionAltKey: conditionAltKey, diff --git a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go index a39985b8a2..b2c236f9a9 100644 --- a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go +++ b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go @@ -187,8 +187,18 @@ func (f *ambiguousPutFiler) ObjectTransaction(_ context.Context, req *filer_pb.O } current := f.entries[key] if current == nil && req.ConditionAltKey != "" { - key = req.ConditionAltKey - current = f.entries[key] + // Mirrors the filer's fallback: the alt key is the object's .versions + // directory and the entry under test is the one its latest pointer + // names; a delete marker resolves to absent. + if versionsEntry := f.entries[req.ConditionAltKey]; versionsEntry != nil { + if latest := string(versionsEntry.Extended[s3_constants.ExtLatestVersionFileNameKey]); latest != "" { + key = req.ConditionAltKey + "/" + latest + current = f.entries[key] + if current != nil && string(current.Extended[s3_constants.ExtDeleteMarkerKey]) == "true" { + current = nil + } + } + } } for _, clause := range req.GetCondition().GetClauses() { var md5Hex string @@ -534,8 +544,8 @@ func TestSuspendedPutIfMatchRoutesWithConditionAltKey(t *testing.T) { s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{filerAddr}, 1) filerImpl.txnCheck = func(req *filer_pb.ObjectTransactionRequest) { - if req.ConditionAltKey != "/buckets/b/k.versions/v_v1" { - t.Fatalf("condition alt key = %q, want /buckets/b/k.versions/v_v1", req.ConditionAltKey) + if req.ConditionAltKey != "/buckets/b/k.versions" { + t.Fatalf("condition alt key = %q, want /buckets/b/k.versions", req.ConditionAltKey) } } @@ -552,3 +562,53 @@ func TestSuspendedPutIfMatchRoutesWithConditionAltKey(t *testing.T) { t.Fatalf("conditional suspended write should route one transaction, got %d", filerImpl.txnCount) } } + +// When the latest version is a delete marker, a suspended If-None-Match: * +// write must treat the object as absent — the fallback resolves a tombstone +// to "no current version", not an existing one. +func TestSuspendedPutIfNoneMatchWithDeleteMarkerLatest(t *testing.T) { + volume := startFakeVolumeServer(t) + filerImpl := &ambiguousPutFiler{ + volume: volume, + entries: map[string]*filer_pb.Entry{ + "/buckets/b": { + Name: "b", + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{}, + Extended: map[string][]byte{s3_constants.ExtVersioningKey: []byte(s3_constants.VersioningSuspended)}, + }, + "/buckets/b/k.versions": { + Name: "k.versions", + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{}, + Extended: map[string][]byte{ + s3_constants.ExtLatestVersionFileNameKey: []byte("v_v1"), + }, + }, + "/buckets/b/k.versions/v_v1": { + Name: "v_v1", + Attributes: &filer_pb.FuseAttributes{FileSize: 0}, + Extended: map[string][]byte{ + s3_constants.ExtVersionIdKey: []byte("v1"), + s3_constants.ExtDeleteMarkerKey: []byte("true"), + }, + }, + }, + apply: true, + } + filerAddr := startFakeFiler(t, filerImpl) + s3a := newPutTestServer(t, filerAddr) + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, filerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{filerAddr}, 1) + + r := httptest.NewRequest(http.MethodPut, "/b/k", strings.NewReader("new")) + r.Header.Set(s3_constants.IfNoneMatch, "*") + _, code, _ := s3a.putSuspendedVersioningObject(r, "b", "k", r.Body, "") + if code != s3err.ErrNone { + t.Fatalf("suspended If-None-Match PUT over a delete-marker latest = %v, want success", code) + } + if filerImpl.entries["/buckets/b/k"] == nil { + t.Fatal("null-version entry was not created") + } +} diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index dceb6559f4..e098e119d6 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -18,6 +18,7 @@ import ( "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/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/util" "github.com/seaweedfs/seaweedfs/weed/wdclient" @@ -375,12 +376,14 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje conditionPath = util.FullPath(req.ConditionKey) } current, findErr := fs.filer.FindEntry(ctx, conditionPath) - // The caller can name a fallback entry evaluated when the primary is - // absent — a suspended-versioning write checks "the null slot if - // present, else the latest version" atomically under this lock. + // The caller can name a fallback evaluated when the primary is absent — + // a suspended-versioning write checks "the null slot if present, else + // the latest version" atomically under this lock. condition_alt_key is + // the object's .versions directory, and the latest it resolves to is + // read from the pointer maintained by the same lock, so a concurrent + // version delete cannot move which entry the condition sees. if findErr == filer_pb.ErrNotFound && req.ConditionAltKey != "" { - conditionPath = util.FullPath(req.ConditionAltKey) - current, findErr = fs.filer.FindEntry(ctx, conditionPath) + 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) @@ -407,6 +410,38 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje return &filer_pb.ObjectTransactionResponse{}, nil } +// latestVersionedConditionEntry resolves the object a conditional write should +// be checked against inside a .versions directory: the entry the latest-version +// pointer names, or nothing when the pointer names a delete marker — a +// tombstone is not a current version. A missing pointer or a pointer to a file +// that is gone means the metadata cannot answer the question, which reports an +// error rather than guessing at either branch. +func (fs *FilerServer) latestVersionedConditionEntry(ctx context.Context, versionsDir util.FullPath) (*filer.Entry, util.FullPath, error) { + versionsEntry, err := fs.filer.FindEntry(ctx, versionsDir) + if err == filer_pb.ErrNotFound || versionsEntry == nil { + return nil, versionsDir, filer_pb.ErrNotFound + } + if err != nil { + return nil, versionsDir, err + } + latestName := string(versionsEntry.Extended[s3_constants.ExtLatestVersionFileNameKey]) + if latestName == "" { + return nil, versionsDir, fmt.Errorf("%s has no latest-version pointer", versionsDir) + } + latestPath := util.NewFullPath(string(versionsDir), latestName) + entry, err := fs.filer.FindEntry(ctx, latestPath) + if err == filer_pb.ErrNotFound { + return nil, latestPath, fmt.Errorf("latest-version pointer %s names a missing entry", latestPath) + } + if err != nil { + return nil, latestPath, err + } + if string(entry.Extended[s3_constants.ExtDeleteMarkerKey]) == "true" { + return nil, latestPath, filer_pb.ErrNotFound + } + return entry, latestPath, nil +} + // ObjectTransactionBatch applies several object transactions in one round trip, // each under its own per-path lock and independent of the others. A failed // transaction (precondition or mutation error) is reported in its own response diff --git a/weed/server/filer_grpc_server_object_txn_test.go b/weed/server/filer_grpc_server_object_txn_test.go index ad2c561297..bdb654fd28 100644 --- a/weed/server/filer_grpc_server_object_txn_test.go +++ b/weed/server/filer_grpc_server_object_txn_test.go @@ -852,3 +852,56 @@ func TestObjectTransactionForwardsToOwner(t *testing.T) { t.Errorf("non-owner must forward, not apply locally") } } + +// 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. +func TestObjectTransactionConditionFallbackResolvesLatestPointer(t *testing.T) { + now := time.Unix(1700000000, 0) + seed := func(latestName, marker string) map[string]*filer.Entry { + entries := map[string]*filer.Entry{ + "/buckets/bkt/obj/.versions": { + Attr: filer.Attr{Inode: 2, Mtime: now, Crtime: now, Mode: 0755}, + Extended: map[string][]byte{s3_constants.ExtLatestVersionFileNameKey: []byte(latestName)}, + }, + } + ext := map[string][]byte{s3_constants.ExtETagKey: []byte("abc")} + if marker != "" { + ext[s3_constants.ExtDeleteMarkerKey] = []byte(marker) + } + entries["/buckets/bkt/obj/.versions/"+latestName] = &filer.Entry{ + Attr: filer.Attr{Inode: 3, Mtime: now, Crtime: now, Mode: 0644}, + Extended: ext, + } + return entries + } + req := func() *filer_pb.ObjectTransactionRequest { + return &filer_pb.ObjectTransactionRequest{ + LockKey: "/buckets/bkt/obj", + ConditionAltKey: "/buckets/bkt/obj/.versions", + Condition: &filer_pb.WriteCondition{ + Clauses: []*filer_pb.WriteCondition_Clause{{Kind: filer_pb.WriteCondition_IF_NOT_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: 9}}, + }}, + } + } + + fs, store := newTxnTestServer(t, seed("v_live", "")) + resp, err := fs.ObjectTransaction(context.Background(), req()) + if err != nil || resp.ErrorCode != filer_pb.FilerError_PRECONDITION_FAILED { + t.Fatalf("latest version present: err=%v code=%v, want precondition failed", err, resp.GetErrorCode()) + } + + fs, store = newTxnTestServer(t, seed("v_gone", "true")) + resp, err = fs.ObjectTransaction(context.Background(), req()) + if err != nil || resp.Error != "" { + t.Fatalf("delete-marker latest: err=%v resp=%q, want applied", err, resp.Error) + } + if store.entries["/buckets/bkt/obj"] == nil { + t.Fatal("mutation did not apply over a delete-marker latest") + } +}