s3: optimize DELETE by skipping lock check for buckets without Object Lock (#7642)

This optimization avoids an expensive filer gRPC call for every DELETE
operation on buckets that don't have Object Lock enabled.

Before this change, enforceObjectLockProtections() would always call
getObjectEntry() to fetch object metadata to check for retention/legal
hold, even for buckets that never had Object Lock configured.

Changes:
1. Add early return in enforceObjectLockProtections() if bucket has no
   Object Lock config or bucket doesn't exist
2. Add isObjectLockEnabled() helper function to check if a bucket has
   Object Lock configured
3. Fix validateObjectLockHeaders() to check ObjectLockConfig instead of
   just versioningEnabled - this ensures object-lock headers are properly
   rejected on buckets without Object Lock enabled, which aligns with
   AWS S3 semantics
4. Make bucket creation with Object Lock atomic - set Object Lock config
   in the same CreateEntry call as bucket creation, preventing race
   conditions where bucket exists without Object Lock enabled
5. Properly handle Object Lock setup failures during bucket creation -
   if StoreObjectLockConfigurationInExtended fails, roll back the bucket
   creation and return an error instead of leaving a bucket without
   the requested Object Lock configuration

This significantly improves DELETE latency for non-Object-Lock buckets,
which is the common case (lockCheck time reduced from 1-10ms to ~1µs).

1 files changed, 21 insertions, 9 deletions

diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go
index f848790de..3da9047ac 100644
--- a/weed/s3api/s3api_object_handlers_put.go
+++ b/weed/s3api/s3api_object_handlers_put.go
@@ -30,14 +30,14 @@ import (
 
 // Object lock validation errors
 var (
-	ErrObjectLockVersioningRequired          = errors.New("object lock headers can only be used on versioned buckets")
+	ErrObjectLockVersioningRequired          = errors.New("object lock headers can only be used on buckets with Object Lock enabled")
 	ErrInvalidObjectLockMode                 = errors.New("invalid object lock mode")
 	ErrInvalidLegalHoldStatus                = errors.New("invalid legal hold status")
 	ErrInvalidRetentionDateFormat            = errors.New("invalid retention until date format")
 	ErrRetentionDateMustBeFuture             = errors.New("retain until date must be in the future")
 	ErrObjectLockModeRequiresDate            = errors.New("object lock mode requires retention until date")
 	ErrRetentionDateRequiresMode             = errors.New("retention until date requires object lock mode")
-	ErrGovernanceBypassVersioningRequired    = errors.New("governance bypass header can only be used on versioned buckets")
+	ErrGovernanceBypassVersioningRequired    = errors.New("governance bypass header can only be used on buckets with Object Lock enabled")
 	ErrInvalidObjectLockDuration             = errors.New("object lock duration must be greater than 0 days")
 	ErrObjectLockDurationExceeded            = errors.New("object lock duration exceeds maximum allowed days")
 	ErrObjectLockConfigurationMissingEnabled = errors.New("object lock configuration must specify ObjectLockEnabled")
@@ -159,8 +159,16 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 
 		glog.V(3).Infof("PutObjectHandler: bucket=%s, object=%s, versioningState='%s', versioningEnabled=%v, versioningConfigured=%v", bucket, object, versioningState, versioningEnabled, versioningConfigured)
 
+		// Check if Object Lock is enabled for this bucket
+		objectLockEnabled, err := s3a.isObjectLockEnabled(bucket)
+		if err != nil && !errors.Is(err, filer_pb.ErrNotFound) {
+			glog.Errorf("Error checking Object Lock status for bucket %s: %v", bucket, err)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+			return
+		}
+
 		// Validate object lock headers before processing
-		if err := s3a.validateObjectLockHeaders(r, versioningEnabled); err != nil {
+		if err := s3a.validateObjectLockHeaders(r, objectLockEnabled); err != nil {
 			glog.V(2).Infof("PutObjectHandler: object lock header validation failed for bucket %s, object %s: %v", bucket, object, err)
 			s3err.WriteErrorResponse(w, r, mapValidationErrorToS3Error(err))
 			return
@@ -1311,7 +1319,8 @@ func (s3a *S3ApiServer) applyBucketDefaultRetention(bucket string, entry *filer_
 }
 
 // validateObjectLockHeaders validates object lock headers in PUT requests
-func (s3a *S3ApiServer) validateObjectLockHeaders(r *http.Request, versioningEnabled bool) error {
+// objectLockEnabled should be true only if the bucket has Object Lock configured
+func (s3a *S3ApiServer) validateObjectLockHeaders(r *http.Request, objectLockEnabled bool) error {
 	// Extract object lock headers from request
 	mode := r.Header.Get(s3_constants.AmzObjectLockMode)
 	retainUntilDateStr := r.Header.Get(s3_constants.AmzObjectLockRetainUntilDate)
@@ -1320,8 +1329,11 @@ func (s3a *S3ApiServer) validateObjectLockHeaders(r *http.Request, versioningEna
 	// Check if any object lock headers are present
 	hasObjectLockHeaders := mode != "" || retainUntilDateStr != "" || legalHold != ""
 
-	// Object lock headers can only be used on versioned buckets
-	if hasObjectLockHeaders && !versioningEnabled {
+	// Object lock headers can only be used on buckets with Object Lock enabled
+	// Per AWS S3: Object Lock can only be enabled at bucket creation, and once enabled,
+	// objects can have retention/legal-hold metadata. Without Object Lock enabled,
+	// these headers must be rejected.
+	if hasObjectLockHeaders && !objectLockEnabled {
 		return ErrObjectLockVersioningRequired
 	}
 
@@ -1362,11 +1374,11 @@ func (s3a *S3ApiServer) validateObjectLockHeaders(r *http.Request, versioningEna
 		}
 	}
 
-	// Check for governance bypass header - only valid for versioned buckets
+	// Check for governance bypass header - only valid for buckets with Object Lock enabled
 	bypassGovernance := r.Header.Get("x-amz-bypass-governance-retention") == "true"
 
-	// Governance bypass headers are only valid for versioned buckets (like object lock headers)
-	if bypassGovernance && !versioningEnabled {
+	// Governance bypass headers are only valid for buckets with Object Lock enabled
+	if bypassGovernance && !objectLockEnabled {
 		return ErrGovernanceBypassVersioningRequired
 	}