s3api: recheck suspended-version condition under the lock when the filer ignores condition_alt_key (#11806)

* s3api: recheck condition under lock when routed filer ignores condition_alt_key

A filer that predates condition_alt_key silently drops the field and
evaluates a suspended-versioning condition against the empty null slot
alone, so a matching If-Match still gets a false 412. Recheck under the
object lock instead of trusting that response: the lock path resolves the
same entry the read path serves. Genuine precondition failures still
return 412 from the recheck.

* s3api: cover suspended If-Match against a filer ignoring condition_alt_key

* s3api: resend against the resolved entry instead of the lock-path recheck

Rechecking under the gateway object lock does not serialize against
routed writers, which commit under the filer's path lock; a concurrent
routed PUT could land between the recheck and the create and be
overwritten with a stale precondition. Resend the same transaction with
condition_key set to the resolved version file so the condition and the
write stay under the same filer lock. When the resolved entry is the
null slot, a delete marker, or unresolvable, the resend targets the
object path and fails closed.

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

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)
+	}
+}