diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 527f46a805..e8530950e0 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -1058,7 +1058,22 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader 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, "", condAltKey, finalizeMutations) + routeKey := s3a.objectRouteKey(bucket, object) + resp, err := s3a.routedPut(owner, routeKey, lockKey, filePath, entry, cond, "", condAltKey, finalizeMutations) + if err == nil && resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED && condAltKey != "" { + // A filer that ignores condition_alt_key evaluated the empty + // null slot alone; resend the transaction against the entry the + // read path serves, still under the object's path lock. + conditionKey := filePath + if resolved, resolveErr := s3a.resolveObjectEntry(bucket, object, ""); resolveErr == nil && + resolved != nil && + string(resolved.Extended[s3_constants.ExtDeleteMarkerKey]) != "true" && + resolved.Name != path.Base(filePath) { + conditionKey = condAltKey + "/" + resolved.Name + } + glog.Warningf("putToFiler: routed PUT to %s precondition-failed for %s; resending against %s", owner, filePath, conditionKey) + resp, err = s3a.routedPut(owner, routeKey, lockKey, filePath, entry, cond, conditionKey, condAltKey, finalizeMutations) + } switch { case err != nil: routedAmbiguous = true diff --git a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go index 7c2b9ae7cd..01d45bc701 100644 --- a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go +++ b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go @@ -135,6 +135,7 @@ type ambiguousPutFiler struct { txnCount int txnCheck func(req *filer_pb.ObjectTransactionRequest) apply bool + ignoreAltKey bool createErr error respError string lookupErr error @@ -257,7 +258,7 @@ func (f *ambiguousPutFiler) ObjectTransaction(_ context.Context, req *filer_pb.O key = req.ConditionKey } current := f.entries[key] - if current == nil && req.ConditionAltKey != "" { + if current == nil && req.ConditionAltKey != "" && !f.ignoreAltKey { // 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. @@ -1057,3 +1058,67 @@ func TestSuspendedPutIfNoneMatchWithDeleteMarkerLatest(t *testing.T) { t.Fatal("null-version entry was not created") } } + +// A filer that predates condition_alt_key ignores the field and checks the +// empty null slot alone, so the transaction precondition-fails even though +// If-Match matches the served version. The gateway resends the transaction +// against the resolved version instead of surfacing the false 412. +func TestSuspendedPutIfMatchRecheckedWhenAltKeyIgnored(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, + ignoreAltKey: 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) + + var conditionKeys []string + filerImpl.txnCheck = func(req *filer_pb.ObjectTransactionRequest) { + conditionKeys = append(conditionKeys, req.ConditionKey) + } + + 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 against a filer ignoring condition_alt_key = %v", code) + } + if filerImpl.entries["/buckets/b/k"] == nil { + t.Fatal("null-version entry was not created") + } + if filerImpl.txnCount != 2 { + t.Fatalf("got %d transactions, want the routed attempt plus a resend against the resolved version", filerImpl.txnCount) + } + if len(conditionKeys) != 2 || conditionKeys[0] != "" || conditionKeys[1] != "/buckets/b/k.versions/v_v1" { + t.Fatalf("condition keys = %v, want [\"\" \"/buckets/b/k.versions/v_v1\"]", conditionKeys) + } +}