mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-22 06:22:38 +00:00
filer: resolve the condition fallback under the transaction lock
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.
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user