s3api: NoSuchKey for object lock calls on missing or deleted keys (#12592)

* s3api: answer PutObjectLegalHold on a missing key with NoSuchKey

In a versioned bucket a key with no latest version resolves to
ErrLatestVersionNotFound, which the legal hold handler did not map, so
the request failed with 500 InternalError. PutObjectRetention already
maps it to NoSuchKey.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3api: keep failed reads distinct when resolving an object lock target

getObjectEntryForUpdate turned every lookup error into a not-found
sentinel, so a failed filer read made PutObjectRetention and
PutObjectLegalHold answer NoSuchKey for an object that exists. Only a
genuine miss becomes the sentinel now, matching getObjectEntry.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3api: answer object lock calls on a delete marker with NoSuchKey

PutObjectLegalHold and PutObjectRetention on a key whose current
version is a delete marker returned 200 and wrote the lock onto the
marker, and the matching GETs read it back. A key whose current version
is a delete marker does not exist, and delete markers take no lock, so
these calls now answer NoSuchKey like the tagging and ACL handlers.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3api: let a lock recorded on a delete marker not block its removal

Delete markers are not WORM-protected, but a legal hold or retention
written onto one (possible before the previous change) made both the
gateway check and the routed delete's filer-side WORM guard refuse to
remove the marker, so the object could not be restored and lifecycle
could not clean the expired marker up.

Lock enforcement now ignores a lock on a delete marker, and the routed
version deletes in DeleteObject and lifecycle drop the WORM guard when
the version they read is a marker. A version id never changes kind, so
the marker read is the one deleted; the null version keeps its guard
because a PUT can replace it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/s3api/s3api_internal_lifecycle.go

@@ -186,7 +186,7 @@ func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle
 						versionDeleted = true
 					}
 				} else {
-					switch code := s3a.routedDeleteSpecificVersion(owner, req.Bucket, req.ObjectPath, req.VersionId, worm, false, metadataOnly); code {
+					switch code := s3a.routedDeleteSpecificVersion(owner, req.Bucket, req.ObjectPath, req.VersionId, worm && !isDeleteMarkerEntry(entry), false, metadataOnly); code {
 					case s3err.ErrNone:
 						versionDeleted = true
 					case s3err.ErrAccessDenied:

weed/s3api/s3api_internal_lifecycle_test.go

@@ -8,6 +8,8 @@ import (
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/seaweedfs/seaweedfs/weed/cluster"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
@@ -468,3 +470,53 @@ func TestLifecycleDelete_CurrentExpirationRequiresLatestVersion(t *testing.T) {
 		}
 	}
 }
+
+// Expired-marker cleanup removes a marker that a lock was recorded on: markers
+// are never WORM-protected, so the routed delete carries no WORM guard.
+func TestLifecycleDelete_LockedExpiredDeleteMarkerSkipsWormGuard(t *testing.T) {
+	marker := &filer_pb.Entry{
+		Name:       "v_m1",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 1_700_000_000},
+		Extended: map[string][]byte{
+			s3_constants.ExtVersionIdKey:    []byte("m1"),
+			s3_constants.ExtDeleteMarkerKey: []byte("true"),
+			s3_constants.ExtLegalHoldKey:    []byte(s3_constants.LegalHoldOn),
+		},
+	}
+	fake := &ambiguousPutFiler{apply: true, entries: map[string]*filer_pb.Entry{
+		"/buckets/b/obj.versions": {
+			Name:        "obj" + s3_constants.VersionsFolder,
+			IsDirectory: true,
+			Attributes:  &filer_pb.FuseAttributes{},
+			Extended: map[string][]byte{
+				s3_constants.ExtLatestVersionIdKey:       []byte("m1"),
+				s3_constants.ExtLatestVersionFileNameKey: []byte("v_m1"),
+			},
+		},
+		"/buckets/b/obj.versions/v_m1": proto.Clone(marker).(*filer_pb.Entry),
+	}}
+	var conditions []*filer_pb.WriteCondition
+	fake.txnCheck = func(req *filer_pb.ObjectTransactionRequest) { conditions = append(conditions, req.GetCondition()) }
+	addr := startFakeFiler(t, fake)
+	s3a := newPutTestServer(t, addr)
+	s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
+	s3a.bucketConfigCache.Set("b", &BucketConfig{
+		Name:             "b",
+		Versioning:       s3_constants.VersioningEnabled,
+		ObjectLockConfig: &ObjectLockConfiguration{ObjectLockEnabled: s3_constants.ObjectLockEnabled},
+	})
+	s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, addr)
+	s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{addr}, 1)
+
+	resp, err := s3a.LifecycleDelete(s3LocalPeerCtx(), &s3_lifecycle_pb.LifecycleDeleteRequest{
+		Bucket:           "b",
+		ObjectPath:       "obj",
+		VersionId:        "m1",
+		ActionKind:       s3_lifecycle_pb.ActionKind_EXPIRED_DELETE_MARKER,
+		ExpectedIdentity: computeEntryIdentity(marker),
+	})
+	require.NoError(t, err)
+	require.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_DONE, resp.Outcome, resp.Reason)
+	require.Len(t, conditions, 1)
+	require.Nil(t, conditions[0], "marker delete must not carry WORM guards")
+}

weed/s3api/s3api_object_handlers_delete.go

@@ -337,7 +337,9 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque
 				if versionId == "null" {
 					deleteCode, deleteHandled = s3a.routedDeleteNullVersion(owner, bucket, object, worm, bypass, false)
 				} else {
-					deleteCode, deleteHandled = s3a.routedDeleteSpecificVersion(owner, bucket, object, versionId, worm, bypass, false), true
+					// A version id never changes kind, so a marker read here is
+					// the one deleted; markers carry no WORM guard.
+					deleteCode, deleteHandled = s3a.routedDeleteSpecificVersion(owner, bucket, object, versionId, worm && !deleteResult.deleteMarker, bypass, false), true
 				}
 			}
 		}

weed/s3api/s3api_object_handlers_legal_hold.go

@@ -50,7 +50,7 @@ func (s3a *S3ApiServer) PutObjectLegalHoldHandler(w http.ResponseWriter, r *http
 		glog.Errorf("PutObjectLegalHoldHandler: failed to set legal hold: %v", err)
 
 		// Handle specific error cases
-		if errors.Is(err, ErrObjectNotFound) || errors.Is(err, ErrVersionNotFound) {
+		if errors.Is(err, ErrObjectNotFound) || errors.Is(err, ErrVersionNotFound) || errors.Is(err, ErrLatestVersionNotFound) {
 			s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
 			return
 		}

weed/s3api/s3api_object_retention.go

@@ -238,6 +238,14 @@ func isObjectEntryMiss(err error) bool {
 		errors.Is(err, errInvalidVersionID)
 }
 
+// missAs reports a genuine miss as sentinel and keeps any other failure as is.
+func missAs(err, sentinel error) error {
+	if isObjectEntryMiss(err) {
+		return sentinel
+	}
+	return err
+}
+
 // getObjectEntryForUpdate resolves the entry a metadata update targets and the
 // filer path it is written under. It applies the same version resolution as
 // getSpecificObjectVersion - "null" names the plain pre-versioning entry, not a
@@ -247,7 +255,10 @@ func (s3a *S3ApiServer) getObjectEntryForUpdate(bucket, object, versionId string
 	if versionId != "" {
 		entry, err = s3a.getSpecificObjectVersion(bucket, object, versionId)
 		if err != nil {
-			return nil, "", fmt.Errorf("failed to get version %s for object %s/%s: %w", versionId, bucket, object, ErrVersionNotFound)
+			return nil, "", fmt.Errorf("failed to get version %s for object %s/%s: %w", versionId, bucket, object, missAs(err, ErrVersionNotFound))
+		}
+		if isDeleteMarkerEntry(entry) {
+			return nil, "", fmt.Errorf("version %s of object %s/%s is a delete marker: %w", versionId, bucket, object, ErrVersionNotFound)
 		}
 		if versionId != "null" {
 			entryPath = object + ".versions/" + s3a.getVersionFileName(versionId)
@@ -262,14 +273,17 @@ func (s3a *S3ApiServer) getObjectEntryForUpdate(bucket, object, versionId string
 	if !versioningEnabled {
 		entry, err = s3a.fetchObjectEntryRequired(bucket, object)
 		if err != nil {
-			return nil, "", fmt.Errorf("failed to get object %s/%s: %w", bucket, object, ErrObjectNotFound)
+			return nil, "", fmt.Errorf("failed to get object %s/%s: %w", bucket, object, missAs(err, ErrObjectNotFound))
 		}
 		return entry, entryPath, nil
 	}
 
 	entry, err = s3a.getLatestObjectVersion(bucket, object)
 	if err != nil {
-		return nil, "", fmt.Errorf("failed to get latest version for object %s/%s: %w", bucket, object, ErrLatestVersionNotFound)
+		return nil, "", fmt.Errorf("failed to get latest version for object %s/%s: %w", bucket, object, missAs(err, ErrLatestVersionNotFound))
+	}
+	if isDeleteMarkerEntry(entry) {
+		return nil, "", fmt.Errorf("latest version of object %s/%s is a delete marker: %w", bucket, object, ErrLatestVersionNotFound)
 	}
 	if entry.Extended != nil {
 		if versionIdBytes, exists := entry.Extended[s3_constants.ExtVersionIdKey]; exists {
@@ -291,6 +305,9 @@ func (s3a *S3ApiServer) getObjectRetention(bucket, object, versionId string) (*O
 	if err != nil {
 		return nil, err
 	}
+	if isDeleteMarkerEntry(entry) {
+		return nil, fmt.Errorf("object %s/%s is a delete marker: %w", bucket, object, ErrObjectNotFound)
+	}
 
 	if entry.Extended == nil {
 		return nil, ErrNoRetentionConfiguration
@@ -406,6 +423,9 @@ func (s3a *S3ApiServer) getObjectLegalHold(bucket, object, versionId string) (*O
 	if err != nil {
 		return nil, err
 	}
+	if isDeleteMarkerEntry(entry) {
+		return nil, fmt.Errorf("object %s/%s is a delete marker: %w", bucket, object, ErrObjectNotFound)
+	}
 
 	if entry.Extended == nil {
 		return nil, ErrNoLegalHoldConfiguration
@@ -593,6 +613,11 @@ func (s3a *S3ApiServer) enforceObjectLockProtections(request *http.Request, buck
 // of its own, or the one a write is about to replace -- use it directly, so that
 // the decision lives in one place.
 func (s3a *S3ApiServer) enforceObjectLockOnEntry(entry *filer_pb.Entry, bucket, object, versionId string, governanceBypassAllowed bool) error {
+	// Delete markers are never WORM-protected; a lock recorded on one must
+	// not keep it from being removed.
+	if isDeleteMarkerEntry(entry) {
+		return nil
+	}
 	retention, retentionActive, err := s3a.getRetentionFromEntry(entry)
 	if err != nil {
 		glog.Warningf("Error parsing retention for %s/%s (versionId: %s): %v", bucket, object, versionId, err)

weed/s3api/s3api_object_retention_test.go

@@ -6,17 +6,21 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"net/http/httptest"
+	"strconv"
 	"strings"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/gorilla/mux"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 
+	"github.com/seaweedfs/seaweedfs/weed/cluster"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
@@ -1102,3 +1106,190 @@ func TestGetObjectEntryMissStillNotFound(t *testing.T) {
 		t.Fatalf("missing object must report ErrObjectNotFound, got: %v", err)
 	}
 }
+
+func putLegalHoldStatus(t *testing.T, s3a *S3ApiServer, object, versionId string) int {
+	t.Helper()
+	target := "http://localhost:8333/b/" + object + "?legal-hold"
+	if versionId != "" {
+		target += "&versionId=" + versionId
+	}
+	body := `<LegalHold xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Status>ON</Status></LegalHold>`
+	req := httptest.NewRequest(http.MethodPut, target, strings.NewReader(body))
+	req = mux.SetURLVars(req, map[string]string{"bucket": "b", "object": object})
+	rec := httptest.NewRecorder()
+	s3a.PutObjectLegalHoldHandler(rec, req)
+	return rec.Code
+}
+
+// PutObjectLegalHold on a key with no version in a versioned bucket is a miss,
+// answered like PutObjectRetention with NoSuchKey rather than InternalError.
+func TestPutObjectLegalHoldMissingObjectIsNoSuchKey(t *testing.T) {
+	f := &retentionTestFiler{entries: map[string]*filer_pb.Entry{
+		"/buckets/b": {
+			Name:        "b",
+			IsDirectory: true,
+			Extended:    map[string][]byte{s3_constants.ExtVersioningKey: []byte(s3_constants.VersioningEnabled)},
+		},
+	}}
+	s3a := newRetentionTestServer(t, f)
+
+	if got := putLegalHoldStatus(t, s3a, "missing", ""); got != http.StatusNotFound {
+		t.Fatalf("PutObjectLegalHold on a missing key = %d, want 404", got)
+	}
+	if len(f.updates) != 0 {
+		t.Fatalf("missing key must not be written, got %v", f.updates)
+	}
+}
+
+// A failed read of the target version is not a miss: the lock write must
+// surface it as a server error the client can retry, not NoSuchKey.
+func TestPutObjectLegalHoldFailedReadIsNotNoSuchKey(t *testing.T) {
+	f := &retentionTestFiler{
+		entries: map[string]*filer_pb.Entry{
+			"/buckets/b": {
+				Name:        "b",
+				IsDirectory: true,
+				Extended:    map[string][]byte{s3_constants.ExtVersioningKey: []byte(s3_constants.VersioningEnabled)},
+			},
+		},
+		lookupErrs: map[string]error{
+			"/buckets/b/key.versions/v_v1": status.Error(codes.Internal, "disk read failed"),
+		},
+	}
+	s3a := newRetentionTestServer(t, f)
+
+	err := s3a.setObjectLegalHold("b", "key", "v1", &ObjectLegalHold{Status: "ON"})
+	if err == nil || errors.Is(err, ErrVersionNotFound) {
+		t.Fatalf("failed read must not report ErrVersionNotFound, got: %v", err)
+	}
+	if got := putLegalHoldStatus(t, s3a, "key", "v1"); got != http.StatusInternalServerError {
+		t.Fatalf("PutObjectLegalHold on a failed read = %d, want 500", got)
+	}
+}
+
+func deleteMarkerTestFiler(markerExtended map[string][]byte) *retentionTestFiler {
+	marker := map[string][]byte{
+		s3_constants.ExtDeleteMarkerKey: []byte("true"),
+		s3_constants.ExtVersionIdKey:    []byte("m1"),
+	}
+	for k, v := range markerExtended {
+		marker[k] = v
+	}
+	return &retentionTestFiler{entries: map[string]*filer_pb.Entry{
+		"/buckets/b": {
+			Name:        "b",
+			IsDirectory: true,
+			Extended:    map[string][]byte{s3_constants.ExtVersioningKey: []byte(s3_constants.VersioningEnabled)},
+		},
+		"/buckets/b/key.versions": {
+			Name:        "key.versions",
+			IsDirectory: true,
+			Extended: map[string][]byte{
+				s3_constants.ExtLatestVersionIdKey:       []byte("m1"),
+				s3_constants.ExtLatestVersionFileNameKey: []byte("v_m1"),
+			},
+		},
+		"/buckets/b/key.versions/v_m1": {Name: "v_m1", Extended: marker},
+	}}
+}
+
+// A key whose current version is a delete marker does not exist, and a marker
+// takes no lock of its own: reading or writing its legal hold is NoSuchKey.
+func TestLegalHoldOnDeleteMarkerIsNoSuchKey(t *testing.T) {
+	f := deleteMarkerTestFiler(nil)
+	s3a := newRetentionTestServer(t, f)
+
+	if got := putLegalHoldStatus(t, s3a, "key", ""); got != http.StatusNotFound {
+		t.Fatalf("PutObjectLegalHold on a deleted key = %d, want 404", got)
+	}
+	if got := putLegalHoldStatus(t, s3a, "key", "m1"); got != http.StatusNotFound {
+		t.Fatalf("PutObjectLegalHold on a delete marker version = %d, want 404", got)
+	}
+	if len(f.updates) != 0 {
+		t.Fatalf("delete marker must not be written, got %v", f.updates)
+	}
+
+	req := httptest.NewRequest(http.MethodGet, "http://localhost:8333/b/key?legal-hold", nil)
+	req = mux.SetURLVars(req, map[string]string{"bucket": "b", "object": "key"})
+	rec := httptest.NewRecorder()
+	s3a.GetObjectLegalHoldHandler(rec, req)
+	if rec.Code != http.StatusNotFound {
+		t.Fatalf("GetObjectLegalHold on a deleted key = %d, want 404", rec.Code)
+	}
+
+	retainUntil := time.Now().Add(time.Hour)
+	err := s3a.setObjectRetention("b", "key", "", &ObjectRetention{Mode: s3_constants.RetentionModeGovernance, RetainUntilDate: &retainUntil}, false)
+	if !errors.Is(err, ErrLatestVersionNotFound) {
+		t.Fatalf("setObjectRetention on a deleted key = %v, want ErrLatestVersionNotFound", err)
+	}
+	if _, err := s3a.getObjectRetention("b", "key", "m1"); !errors.Is(err, ErrObjectNotFound) {
+		t.Fatalf("getObjectRetention on a delete marker = %v, want ErrObjectNotFound", err)
+	}
+}
+
+// A lock left on a delete marker by an older server must not keep the marker
+// from being removed: markers are never WORM-protected.
+func TestLockOnDeleteMarkerDoesNotBlockRemoval(t *testing.T) {
+	retainUntil := strconv.FormatInt(time.Now().Add(time.Hour).Unix(), 10)
+	f := deleteMarkerTestFiler(map[string][]byte{
+		s3_constants.ExtLegalHoldKey:          []byte(s3_constants.LegalHoldOn),
+		s3_constants.ExtObjectLockModeKey:     []byte(s3_constants.RetentionModeCompliance),
+		s3_constants.ExtRetentionUntilDateKey: []byte(retainUntil),
+	})
+	f.entries["/buckets/b"].Extended[s3_constants.ExtObjectLockEnabledKey] = []byte(s3_constants.ObjectLockEnabled)
+	s3a := newRetentionTestServer(t, f)
+
+	for _, versionId := range []string{"m1", ""} {
+		if err := s3a.enforceObjectLockProtections(nil, "b", "key", versionId, false); err != nil {
+			t.Fatalf("enforceObjectLockProtections(versionId=%q) on a locked delete marker = %v, want nil", versionId, err)
+		}
+	}
+}
+
+type deleteMarkerTxnFiler struct {
+	*retentionTestFiler
+	txns []*filer_pb.ObjectTransactionRequest
+}
+
+func (f *deleteMarkerTxnFiler) ObjectTransaction(_ context.Context, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.txns = append(f.txns, req)
+	return &filer_pb.ObjectTransactionResponse{}, nil
+}
+
+// The routed version delete carries the WORM guards to the owner filer, which
+// would refuse a marker that holds a lock; a marker is deleted without them.
+func TestRoutedDeleteOfLockedDeleteMarkerSkipsWormGuard(t *testing.T) {
+	retainUntil := strconv.FormatInt(time.Now().Add(time.Hour).Unix(), 10)
+	base := deleteMarkerTestFiler(map[string][]byte{
+		s3_constants.ExtLegalHoldKey:          []byte(s3_constants.LegalHoldOn),
+		s3_constants.ExtObjectLockModeKey:     []byte(s3_constants.RetentionModeCompliance),
+		s3_constants.ExtRetentionUntilDateKey: []byte(retainUntil),
+	})
+	base.entries["/buckets/b"].Extended[s3_constants.ExtObjectLockEnabledKey] = []byte(s3_constants.ObjectLockEnabled)
+	f := &deleteMarkerTxnFiler{retentionTestFiler: base}
+	addr := startFakeFiler(t, f)
+	dialOption := grpc.WithTransportCredentials(insecure.NewCredentials())
+	s3a := &S3ApiServer{
+		option:            &S3ApiServerOption{Filers: []pb.ServerAddress{addr}, GrpcDialOption: dialOption, BucketsPath: "/buckets"},
+		filerClient:       wdclient.NewFilerClient([]pb.ServerAddress{addr}, dialOption, ""),
+		bucketConfigCache: NewBucketConfigCache(time.Minute),
+	}
+	s3a.objectWriteLockClient = cluster.NewLockClient(dialOption, addr)
+	s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{addr}, 1)
+
+	req := httptest.NewRequest(http.MethodDelete, "http://localhost:8333/b/key?versionId=m1", nil)
+	req = mux.SetURLVars(req, map[string]string{"bucket": "b", "object": "key"})
+	rec := httptest.NewRecorder()
+	s3a.DeleteObjectHandler(rec, req)
+	if rec.Code != http.StatusNoContent {
+		t.Fatalf("DeleteObject of a locked delete marker = %d, want 204: %s", rec.Code, rec.Body.String())
+	}
+	if len(f.txns) != 1 {
+		t.Fatalf("expected one routed delete, got %d", len(f.txns))
+	}
+	if cond := f.txns[0].GetCondition(); cond != nil {
+		t.Fatalf("delete marker removal must not carry WORM guards, got %+v", cond)
+	}
+}

weed/s3api/s3api_object_versioning.go

@@ -32,6 +32,10 @@ import (
 // ErrDeleteMarker is returned when the latest version is a delete marker (expected condition)
 var ErrDeleteMarker = errors.New("latest version is a delete marker")
 
+func isDeleteMarkerEntry(entry *filer_pb.Entry) bool {
+	return entry != nil && string(entry.Extended[s3_constants.ExtDeleteMarkerKey]) == "true"
+}
+
 // clearCachedVersionMetadata clears only the version metadata fields (not ID/filename).
 // Used by setCachedListMetadata to prevent stale values when updating.
 func clearCachedVersionMetadata(extended map[string][]byte) {