mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-17 19:48:35 +02:00
s3: fix PutObject If-Match under suspended versioning (#11780)
* s3: keep suspended-versioning conditional writes off the routed condition A suspended PUT writes the null-version slot at the regular path while the entry the conditional headers evaluate lives under .versions/. The routed WriteCondition checked the request's etag against that empty slot, so a correct If-Match failed with 412. Passing uniqueWritePath sends conditional requests through the lock path, whose precondition resolves the same entry the read path serves; unconditional writes still route. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer_pb: add condition_alt_key to ObjectTransactionRequest * filer: evaluate a transaction condition against a fallback entry condition_alt_key names the entry consulted when the primary is absent, so a suspended-versioning write can check 'the null slot if present, else the latest version' atomically under the lock. * s3: route suspended-versioning conditional writes with a fallback key The gateway names the latest version file as condition_alt_key, so the transaction checks the entry the read path serves — null slot first, latest version when it is absent — under the same lock that applies the write. Conditional suspended PUTs route again instead of splitting check and write across a gateway lock that routed writes skip. * 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. * filer: fail closed on alt-key lookups and skip directory primaries A transient FindEntry error returning a nil entry read as "versions dir absent" and let If-None-Match writes pass. Check the error before the entry. And a bare directory at the object path satisfied the primary lookup, so the .versions fallback never ran — directories are not the object, treat them as an absent slot. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: scope directory-as-absent conditions to alt-key callers --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
1 parent
cdc6976e3a
commit
d3a97f76bd
9 files changed
+455
-16
No files matched your search
@@ -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 condition_alt_key = 10; // if the condition's entry is absent, evaluate it against this entry instead (e.g. null slot, else latest version)
|
||||
}
|
||||
|
||||
message ObjectTransactionResponse {
|
||||
|
||||
@@ -1811,9 +1811,10 @@ type ObjectTransactionRequest struct {
|
||||
Mutations []*ObjectMutation `protobuf:"bytes,3,rep,name=mutations,proto3" json:"mutations,omitempty"`
|
||||
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
|
||||
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
|
||||
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
|
||||
}
|
||||
@@ -1904,6 +1905,13 @@ func (x *ObjectTransactionRequest) GetIsMoved() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
func (x *ObjectTransactionRequest) GetConditionAltKey() string {
|
||||
if x != nil {
|
||||
return x.ConditionAltKey
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
type ObjectTransactionResponse struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
Error string `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"`
|
||||
@@ -7433,7 +7441,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\"\xd5\x02\n" +
|
||||
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\x81\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" +
|
||||
@@ -7444,7 +7452,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\"f\n" +
|
||||
"\bis_moved\x18\b \x01(\bR\aisMoved\x12*\n" +
|
||||
"\x11condition_alt_key\x18\n" +
|
||||
" \x01(\tR\x0fconditionAltKey\"f\n" +
|
||||
"\x19ObjectTransactionResponse\x12\x14\n" +
|
||||
"\x05error\x18\x01 \x01(\tR\x05error\x123\n" +
|
||||
"\n" +
|
||||
|
||||
@@ -1567,6 +1567,13 @@ func (m *ObjectTransactionRequest) MarshalToSizedBufferVT(dAtA []byte) (int, err
|
||||
i -= len(m.unknownFields)
|
||||
copy(dAtA[i:], m.unknownFields)
|
||||
}
|
||||
if len(m.ConditionAltKey) > 0 {
|
||||
i -= len(m.ConditionAltKey)
|
||||
copy(dAtA[i:], m.ConditionAltKey)
|
||||
i = protohelpers.EncodeVarint(dAtA, i, uint64(len(m.ConditionAltKey)))
|
||||
i--
|
||||
dAtA[i] = 0x52
|
||||
}
|
||||
if m.IsMoved {
|
||||
i--
|
||||
if m.IsMoved {
|
||||
@@ -7188,6 +7195,10 @@ func (m *ObjectTransactionRequest) SizeVT() (n int) {
|
||||
if m.IsMoved {
|
||||
n += 2
|
||||
}
|
||||
l = len(m.ConditionAltKey)
|
||||
if l > 0 {
|
||||
n += 1 + l + protohelpers.SizeOfVarint(uint64(l))
|
||||
}
|
||||
n += len(m.unknownFields)
|
||||
return n
|
||||
}
|
||||
@@ -13841,6 +13852,38 @@ func (m *ObjectTransactionRequest) UnmarshalVT(dAtA []byte) error {
|
||||
}
|
||||
}
|
||||
m.IsMoved = bool(v != 0)
|
||||
case 10:
|
||||
if wireType != 2 {
|
||||
return fmt.Errorf("proto: wrong wireType = %d for field ConditionAltKey", 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.ConditionAltKey = string(dAtA[iNdEx:postIndex])
|
||||
iNdEx = postIndex
|
||||
default:
|
||||
iNdEx = preIndex
|
||||
skippy, err := protohelpers.Skip(dAtA[iNdEx:])
|
||||
|
||||
@@ -478,6 +478,10 @@ type putFinalize struct {
|
||||
lockKey string
|
||||
mutations []*filer_pb.ObjectMutation
|
||||
afterCreate func(entry *filer_pb.Entry) s3err.ErrorCode
|
||||
// conditionAltKey names the entry a routed WriteCondition falls back to
|
||||
// when the write path is absent — a suspended-versioning write points it
|
||||
// at the latest version so check and write stay under the filer lock
|
||||
conditionAltKey string
|
||||
}
|
||||
|
||||
const (
|
||||
@@ -1010,15 +1014,19 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
|
||||
routed := false
|
||||
routedAmbiguous := false
|
||||
var owner pb.ServerAddress
|
||||
condAltKey := ""
|
||||
if finalize != nil {
|
||||
condAltKey = finalize.conditionAltKey
|
||||
}
|
||||
if owner = s3a.routableWriteOwner(bucket, object); owner != "" {
|
||||
if cond, ok := routeWriteCondition(r, uniqueWritePath); ok {
|
||||
if cond, ok := routeWriteCondition(r, uniqueWritePath && condAltKey == ""); ok {
|
||||
// Routed mutations ride in the PUT's transaction (committing atomically),
|
||||
// so lockKey is the object path they carry, not the version file path.
|
||||
lockKey, finalizeMutations := filePath, []*filer_pb.ObjectMutation(nil)
|
||||
if finalize != nil && len(finalize.mutations) > 0 {
|
||||
lockKey, finalizeMutations = finalize.lockKey, finalize.mutations
|
||||
}
|
||||
resp, err := s3a.routedPut(owner, s3a.objectRouteKey(bucket, object), lockKey, filePath, entry, cond, "", finalizeMutations)
|
||||
resp, err := s3a.routedPut(owner, s3a.objectRouteKey(bucket, object), lockKey, filePath, entry, cond, "", condAltKey, finalizeMutations)
|
||||
switch {
|
||||
case err != nil:
|
||||
routedAmbiguous = true
|
||||
@@ -1824,7 +1832,19 @@ func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, ob
|
||||
// naming a marker that is gone, and the read path heals that by promoting an older
|
||||
// version, republishing the deleted key. Best-effort, as a stale flag self-heals on
|
||||
// the next list.
|
||||
// A conditional suspended write checks "the null slot if present, else the
|
||||
// latest version". Naming the pointed-at version as the fallback lets the
|
||||
// routed transaction evaluate that rule under the filer lock instead of
|
||||
// splitting check and write across a gateway lock that routed writes skip.
|
||||
conditionAltKey := ""
|
||||
if headers, headersErr := parseConditionalHeaders(r); headersErr == s3err.ErrNone && headers.isSet {
|
||||
// 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,
|
||||
afterCreate: func(entry *filer_pb.Entry) s3err.ErrorCode {
|
||||
writtenETag := string(entry.Extended[s3_constants.ExtETagKey])
|
||||
if err := s3a.finalizeSuspendedNullWrite("", bucket, normalizedObject, s3_constants.ExtETagKey, writtenETag); err != nil {
|
||||
@@ -1832,7 +1852,7 @@ func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, ob
|
||||
}
|
||||
return s3err.ErrNone
|
||||
},
|
||||
}, false, "")
|
||||
}, true, "")
|
||||
if errCode != s3err.ErrNone {
|
||||
glog.Errorf("putSuspendedVersioningObject: failed to upload object: %v", errCode)
|
||||
return "", errCode, SSEResponseMetadata{}
|
||||
|
||||
@@ -8,9 +8,12 @@ import (
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"path"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
@@ -18,10 +21,12 @@ import (
|
||||
"google.golang.org/grpc/status"
|
||||
"google.golang.org/protobuf/proto"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/cluster"
|
||||
"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/pb/volume_server_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
"github.com/seaweedfs/seaweedfs/weed/wdclient"
|
||||
@@ -112,6 +117,8 @@ type ambiguousPutFiler struct {
|
||||
|
||||
mu sync.Mutex
|
||||
entries map[string]*filer_pb.Entry
|
||||
txnCount int
|
||||
txnCheck func(req *filer_pb.ObjectTransactionRequest)
|
||||
apply bool
|
||||
createErr error
|
||||
respError string
|
||||
@@ -163,6 +170,92 @@ func (f *ambiguousPutFiler) LookupDirectoryEntry(_ context.Context, req *filer_p
|
||||
return &filer_pb.LookupDirectoryEntryResponse{}, nil
|
||||
}
|
||||
|
||||
// ObjectTransaction applies the WriteCondition against the entry the condition
|
||||
// key names, the way a real filer evaluates it under the path lock, then
|
||||
// applies PUT mutations. Routed-write tests use it to catch a condition sent
|
||||
// for an entry the gateway's conditional check never looked at.
|
||||
func (f *ambiguousPutFiler) ObjectTransaction(_ context.Context, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.txnCount++
|
||||
if f.txnCheck != nil {
|
||||
f.txnCheck(req)
|
||||
}
|
||||
key := req.LockKey
|
||||
if req.ConditionKey != "" {
|
||||
key = req.ConditionKey
|
||||
}
|
||||
current := f.entries[key]
|
||||
if current == nil && req.ConditionAltKey != "" {
|
||||
// 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
|
||||
if current != nil && current.Attributes != nil {
|
||||
md5Hex = fmt.Sprintf("%x", current.Attributes.Md5)
|
||||
}
|
||||
ok := true
|
||||
switch clause.Kind {
|
||||
case filer_pb.WriteCondition_IF_NOT_EXISTS:
|
||||
ok = current == nil
|
||||
case filer_pb.WriteCondition_IF_EXISTS:
|
||||
ok = current != nil
|
||||
case filer_pb.WriteCondition_IF_ETAG_MATCH:
|
||||
ok = current != nil && slices.Contains(clause.Etags, md5Hex)
|
||||
case filer_pb.WriteCondition_IF_ETAG_NOT_MATCH:
|
||||
ok = current == nil || !slices.Contains(clause.Etags, md5Hex)
|
||||
}
|
||||
if !ok {
|
||||
return &filer_pb.ObjectTransactionResponse{ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED}, nil
|
||||
}
|
||||
}
|
||||
for _, m := range req.Mutations {
|
||||
if m.Type == filer_pb.ObjectMutation_PUT {
|
||||
f.entries[m.Directory+"/"+m.Entry.Name] = proto.Clone(m.Entry).(*filer_pb.Entry)
|
||||
}
|
||||
}
|
||||
return &filer_pb.ObjectTransactionResponse{}, nil
|
||||
}
|
||||
|
||||
func (f *ambiguousPutFiler) ListEntries(req *filer_pb.ListEntriesRequest, stream filer_pb.SeaweedFiler_ListEntriesServer) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
for fullPath, entry := range f.entries {
|
||||
if path.Dir(fullPath) != req.Directory {
|
||||
continue
|
||||
}
|
||||
if req.Prefix != "" && !strings.HasPrefix(entry.Name, req.Prefix) {
|
||||
continue
|
||||
}
|
||||
if req.StartFromFileName != "" && (entry.Name < req.StartFromFileName ||
|
||||
(entry.Name == req.StartFromFileName && !req.InclusiveStartFrom)) {
|
||||
continue
|
||||
}
|
||||
if err := stream.Send(&filer_pb.ListEntriesResponse{Entry: proto.Clone(entry).(*filer_pb.Entry)}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *ambiguousPutFiler) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.entries[req.Directory+"/"+req.Entry.Name] = proto.Clone(req.Entry).(*filer_pb.Entry)
|
||||
return &filer_pb.UpdateEntryResponse{}, nil
|
||||
}
|
||||
|
||||
func (f *ambiguousPutFiler) LookupVolume(_ context.Context, req *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) {
|
||||
resp := &filer_pb.LookupVolumeResponse{LocationsMap: map[string]*filer_pb.Locations{}}
|
||||
for _, vid := range req.VolumeIds {
|
||||
@@ -407,3 +500,115 @@ func TestFetchChunkDataDecryptsVolumeCipher(t *testing.T) {
|
||||
t.Fatalf("ranged plain read = %q, want %q", got, plaintext[5:9])
|
||||
}
|
||||
}
|
||||
|
||||
// A suspended-versioning overwrite resolves the conditional check against the
|
||||
// entry the read path serves: the null slot when present, else the latest
|
||||
// version under .versions/. The routed transaction carries that version as the
|
||||
// condition's fallback key so check and write stay under the filer lock —
|
||||
// checking the empty null-version slot alone would reject a matching If-Match.
|
||||
func TestSuspendedPutIfMatchRoutesWithConditionAltKey(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.ExtLatestVersionIdKey: []byte("v1"),
|
||||
s3_constants.ExtLatestVersionFileNameKey: []byte("v_v1"),
|
||||
},
|
||||
},
|
||||
"/buckets/b/k.versions/v_v1": {
|
||||
Name: "v_v1",
|
||||
Attributes: &filer_pb.FuseAttributes{Md5: []byte{0xab, 0xc1, 0x23}, FileSize: 1},
|
||||
Extended: map[string][]byte{
|
||||
s3_constants.ExtETagKey: []byte("abc123"),
|
||||
s3_constants.ExtVersionIdKey: []byte("v1"),
|
||||
},
|
||||
},
|
||||
},
|
||||
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)
|
||||
|
||||
filerImpl.txnCheck = func(req *filer_pb.ObjectTransactionRequest) {
|
||||
if req.ConditionAltKey != "/buckets/b/k.versions" {
|
||||
t.Fatalf("condition alt key = %q, want /buckets/b/k.versions", req.ConditionAltKey)
|
||||
}
|
||||
}
|
||||
|
||||
r := httptest.NewRequest(http.MethodPut, "/b/k", strings.NewReader("new"))
|
||||
r.Header.Set(s3_constants.IfMatch, `"abc123"`)
|
||||
_, code, _ := s3a.putSuspendedVersioningObject(r, "b", "k", r.Body, "")
|
||||
if code != s3err.ErrNone {
|
||||
t.Fatalf("suspended If-Match PUT = %v", code)
|
||||
}
|
||||
if filerImpl.entries["/buckets/b/k"] == nil {
|
||||
t.Fatal("null-version entry was not created")
|
||||
}
|
||||
if filerImpl.txnCount != 1 {
|
||||
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")
|
||||
}
|
||||
}
|
||||
@@ -183,7 +183,7 @@ func (s3a *S3ApiServer) objectTxnOnFiler(owner pb.ServerAddress, req *filer_pb.O
|
||||
// object path plus a RECOMPUTE_LATEST finalize, so the version's PUT and its
|
||||
// .versions pointer flip commit atomically (the recompute scans .versions/ after
|
||||
// the PUT and sees the new version).
|
||||
func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, routeKey, lockKey, filePath string, entry *filer_pb.Entry, cond *filer_pb.WriteCondition, conditionKey string, finalize []*filer_pb.ObjectMutation) (*filer_pb.ObjectTransactionResponse, error) {
|
||||
func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, routeKey, lockKey, filePath string, entry *filer_pb.Entry, cond *filer_pb.WriteCondition, conditionKey, conditionAltKey string, finalize []*filer_pb.ObjectMutation) (*filer_pb.ObjectTransactionResponse, error) {
|
||||
mutations := make([]*filer_pb.ObjectMutation, 0, 1+len(finalize))
|
||||
mutations = append(mutations, &filer_pb.ObjectMutation{
|
||||
Type: filer_pb.ObjectMutation_PUT,
|
||||
@@ -192,11 +192,12 @@ func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, routeKey, lockKey, fil
|
||||
})
|
||||
mutations = append(mutations, finalize...)
|
||||
return s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{
|
||||
LockKey: lockKey,
|
||||
RouteKey: routeKey,
|
||||
Condition: cond,
|
||||
ConditionKey: conditionKey,
|
||||
Mutations: mutations,
|
||||
LockKey: lockKey,
|
||||
RouteKey: routeKey,
|
||||
Condition: cond,
|
||||
ConditionKey: conditionKey,
|
||||
ConditionAltKey: conditionAltKey,
|
||||
Mutations: mutations,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -227,7 +228,7 @@ func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, routeKey, parentDir
|
||||
cond, conditionKey = removal.condition, removal.conditionKey
|
||||
finalize = []*filer_pb.ObjectMutation{removal.mutation}
|
||||
}
|
||||
resp, err := s3a.routedPut(owner, routeKey, filePath, filePath, entry, cond, conditionKey, finalize)
|
||||
resp, err := s3a.routedPut(owner, routeKey, filePath, filePath, entry, cond, conditionKey, "", finalize)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -139,7 +139,7 @@ func (s3a *S3ApiServer) routedMultipartFinalize(owner pb.ServerAddress, bucket,
|
||||
routeKey = s3a.objectRouteKey(bucket, object)
|
||||
}
|
||||
conditionKey, condition := uploadExistsCondition(s3a.genUploadsFolder(bucket) + "/" + uploadID)
|
||||
resp, err := s3a.routedPut(owner, routeKey, s3a.toFilerPath(bucket, object), versionDir+"/"+versionFileName, versionEntry, condition, conditionKey, []*filer_pb.ObjectMutation{
|
||||
resp, err := s3a.routedPut(owner, routeKey, s3a.toFilerPath(bucket, object), versionDir+"/"+versionFileName, versionEntry, condition, conditionKey, "", []*filer_pb.ObjectMutation{
|
||||
s3a.removeUploadDirMutation(bucket, uploadID),
|
||||
s3a.latestPointerRecompute(bucket, object, useInvertedFormat, "", true),
|
||||
})
|
||||
|
||||
@@ -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"
|
||||
@@ -343,6 +344,7 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje
|
||||
ConditionKey: req.ConditionKey,
|
||||
RouteKey: req.RouteKey,
|
||||
IsMoved: true,
|
||||
ConditionAltKey: req.ConditionAltKey,
|
||||
}
|
||||
var resp *filer_pb.ObjectTransactionResponse
|
||||
handled, err := fs.forwardToWriteOwner(ctx, req.RouteKey, func(owner pb.ServerAddress) error {
|
||||
@@ -374,6 +376,22 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje
|
||||
conditionPath = util.FullPath(req.ConditionKey)
|
||||
}
|
||||
current, findErr := fs.filer.FindEntry(ctx, conditionPath)
|
||||
if findErr == nil && current != nil && current.IsDirectory() && req.ConditionAltKey != "" {
|
||||
// A directory is not the object: implicit directories can share a
|
||||
// path with an S3 key, so treat it as an absent slot. Only the
|
||||
// alt-key callers want object-slot semantics; an explicit
|
||||
// condition_key may legitimately name a directory.
|
||||
current, findErr = nil, filer_pb.ErrNotFound
|
||||
}
|
||||
// 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 != "" {
|
||||
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)
|
||||
}
|
||||
@@ -399,6 +417,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 != nil {
|
||||
return nil, versionsDir, err
|
||||
}
|
||||
if versionsEntry == nil {
|
||||
return nil, versionsDir, filer_pb.ErrNotFound
|
||||
}
|
||||
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
|
||||
|
||||
@@ -3,7 +3,9 @@ package weed_server
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -852,3 +854,110 @@ 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")
|
||||
}
|
||||
}
|
||||
|
||||
// A bare directory at the object path — left over when /bkt/obj/child was
|
||||
// created — is not the object, so the condition must fall back to the
|
||||
// .versions resolution instead of counting the directory as an existing key.
|
||||
func TestObjectTransactionConditionIgnoresDirectory(t *testing.T) {
|
||||
now := time.Unix(1700000000, 0)
|
||||
seed := func(marker string) map[string]*filer.Entry {
|
||||
return map[string]*filer.Entry{
|
||||
"/buckets/bkt/obj": {
|
||||
Attr: filer.Attr{Inode: 1, Mtime: now, Crtime: now, Mode: 0755 | os.ModeDir},
|
||||
},
|
||||
"/buckets/bkt/obj/.versions": {
|
||||
Attr: filer.Attr{Inode: 2, Mtime: now, Crtime: now, Mode: 0755},
|
||||
Extended: map[string][]byte{s3_constants.ExtLatestVersionFileNameKey: []byte("v1")},
|
||||
},
|
||||
"/buckets/bkt/obj/.versions/v1": {
|
||||
Attr: filer.Attr{Inode: 3, Mtime: now, Crtime: now, Mode: 0644},
|
||||
Extended: map[string][]byte{s3_constants.ExtETagKey: []byte("abc"), s3_constants.ExtDeleteMarkerKey: []byte(marker)},
|
||||
},
|
||||
}
|
||||
}
|
||||
req := &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, _ := newTxnTestServer(t, seed("true"))
|
||||
resp, err := fs.ObjectTransaction(context.Background(), req)
|
||||
if err != nil {
|
||||
t.Fatalf("directory at key with delete-marker latest: %v", err)
|
||||
}
|
||||
// The condition must pass so the write reaches the type-conflict path;
|
||||
// the directory itself rejects the PUT, which is the caller's business.
|
||||
if resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED {
|
||||
t.Fatalf("condition must not fail on a bare directory: resp=%q", resp.Error)
|
||||
}
|
||||
if !strings.Contains(resp.Error, "is a directory") {
|
||||
t.Fatalf("want the dir/file type-conflict error, got %q", resp.Error)
|
||||
}
|
||||
|
||||
fs, _ = newTxnTestServer(t, seed(""))
|
||||
resp, err = fs.ObjectTransaction(context.Background(), req)
|
||||
if err != nil || resp.ErrorCode != filer_pb.FilerError_PRECONDITION_FAILED {
|
||||
t.Fatalf("directory at key with live latest: err=%v code=%v, want precondition failed", err, resp.GetErrorCode())
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user