s3api: skip object lock check for buckets without object lock

Optimize DELETE and PUT operations by skipping the object lock
protection check for buckets that don't have object lock enabled.

Previously, enforceObjectLockProtections was always called, which
performed an expensive entry lookup even for buckets without object lock.

This optimization:
- Adds isObjectLockEnabled() helper that uses cached bucket config
- Skips enforceObjectLockProtections for buckets without object lock
- Applies to both versioned and non-versioned DELETE paths
- Applies to PUT operations on non-versioned buckets

Performance impact: For buckets without object lock (the common case),
this eliminates one filer lookup per DELETE/PUT operation.

weed/s3api/s3api_bucket_config.go

@@ -548,6 +548,15 @@ func (s3a *S3ApiServer) getBucketVersioningStatus(bucket string) (string, s3err.
 	return config.Versioning, s3err.ErrNone
 }
 
+// isObjectLockEnabled checks if object lock is enabled for a bucket (cached check)
+func (s3a *S3ApiServer) isObjectLockEnabled(bucket string) bool {
+	config, errCode := s3a.getBucketConfig(bucket)
+	if errCode != s3err.ErrNone {
+		return false
+	}
+	return config.ObjectLockConfig != nil
+}
+
 // setBucketVersioningStatus sets the versioning status for a bucket
 func (s3a *S3ApiServer) setBucketVersioningStatus(bucket, status string) s3err.ErrorCode {
 	errCode := s3a.updateBucketConfig(bucket, func(config *BucketConfig) error {

weed/s3api/s3api_object_handlers_delete.go

@@ -53,16 +53,21 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque
 		// Handle versioned delete based on specific versioning state
 		if versionId != "" {
 			// Delete specific version (same for both enabled and suspended)
-			// Check object lock permissions before deleting specific version
-			governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
-			if err := s3a.enforceObjectLockProtections(r, bucket, object, versionId, governanceBypassAllowed); err != nil {
-				glog.V(2).Infof("DeleteObjectHandler: object lock check failed for %s/%s: %v", bucket, object, err)
-				s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
-				return
+			// Only check object lock if enabled for this bucket (avoids expensive entry lookup)
+			var prefetchedEntry *filer_pb.Entry
+			if s3a.isObjectLockEnabled(bucket) {
+				governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
+				var lockErr error
+				prefetchedEntry, lockErr = s3a.enforceObjectLockProtections(r, bucket, object, versionId, governanceBypassAllowed)
+				if lockErr != nil {
+					glog.V(2).Infof("DeleteObjectHandler: object lock check failed for %s/%s: %v", bucket, object, lockErr)
+					s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
+					return
+				}
 			}
 
-			// Delete specific version
-			err := s3a.deleteSpecificObjectVersion(bucket, object, versionId)
+			// Delete specific version, passing prefetched entry if available
+			err := s3a.deleteSpecificObjectVersion(bucket, object, versionId, prefetchedEntry)
 			if err != nil {
 				glog.Errorf("Failed to delete specific version %s: %v", versionId, err)
 				s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
@@ -91,16 +96,21 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque
 				// Suspended versioning: Actually delete the "null" version object
 				glog.V(2).Infof("DeleteObjectHandler: deleting null version for suspended versioning %s/%s", bucket, object)
 
-				// Check object lock permissions before deleting "null" version
-				governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
-				if err := s3a.enforceObjectLockProtections(r, bucket, object, "null", governanceBypassAllowed); err != nil {
-					glog.V(2).Infof("DeleteObjectHandler: object lock check failed for %s/%s: %v", bucket, object, err)
-					s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
-					return
+				// Only check object lock if enabled for this bucket (avoids expensive entry lookup)
+				var prefetchedEntry *filer_pb.Entry
+				if s3a.isObjectLockEnabled(bucket) {
+					governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
+					var lockErr error
+					prefetchedEntry, lockErr = s3a.enforceObjectLockProtections(r, bucket, object, "null", governanceBypassAllowed)
+					if lockErr != nil {
+						glog.V(2).Infof("DeleteObjectHandler: object lock check failed for %s/%s: %v", bucket, object, lockErr)
+						s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
+						return
+					}
 				}
 
 				// Delete the "null" version (the regular file)
-				err := s3a.deleteSpecificObjectVersion(bucket, object, "null")
+				err := s3a.deleteSpecificObjectVersion(bucket, object, "null", prefetchedEntry)
 				if err != nil {
 					glog.Errorf("Failed to delete null version: %v", err)
 					s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
@@ -113,12 +123,15 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque
 		}
 	} else {
 		// Handle regular delete (non-versioned)
-		// Check object lock permissions before deleting object
-		governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
-		if err := s3a.enforceObjectLockProtections(r, bucket, object, "", governanceBypassAllowed); err != nil {
-			glog.V(2).Infof("DeleteObjectHandler: object lock check failed for %s/%s: %v", bucket, object, err)
-			s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
-			return
+		// Only check object lock permissions if object lock is enabled for the bucket
+		// This avoids an expensive entry lookup for buckets without object lock
+		if s3a.isObjectLockEnabled(bucket) {
+			governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
+			if _, lockErr := s3a.enforceObjectLockProtections(r, bucket, object, "", governanceBypassAllowed); lockErr != nil {
+				glog.V(2).Infof("DeleteObjectHandler: object lock check failed for %s/%s: %v", bucket, object, lockErr)
+				s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
+				return
+			}
 		}
 
 		target := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object))
@@ -234,12 +247,15 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 				continue
 			}
 
-			// Check object lock permissions before deletion (only for versioned buckets)
-			if versioningConfigured {
+			// Only check object lock if enabled for this bucket (avoids expensive entry lookup)
+			var prefetchedEntry *filer_pb.Entry
+			if versioningConfigured && s3a.isObjectLockEnabled(bucket) {
 				// Validate governance bypass for this specific object
 				governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object.Key)
-				if err := s3a.enforceObjectLockProtections(r, bucket, object.Key, object.VersionId, governanceBypassAllowed); err != nil {
-					glog.V(2).Infof("DeleteMultipleObjectsHandler: object lock check failed for %s/%s (version: %s): %v", bucket, object.Key, object.VersionId, err)
+				var lockErr error
+				prefetchedEntry, lockErr = s3a.enforceObjectLockProtections(r, bucket, object.Key, object.VersionId, governanceBypassAllowed)
+				if lockErr != nil {
+					glog.V(2).Infof("DeleteMultipleObjectsHandler: object lock check failed for %s/%s (version: %s): %v", bucket, object.Key, object.VersionId, lockErr)
 					deleteErrors = append(deleteErrors, DeleteError{
 						Code:      s3err.GetAPIError(s3err.ErrAccessDenied).Code,
 						Message:   s3err.GetAPIError(s3err.ErrAccessDenied).Description,
@@ -257,7 +273,8 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 				// Handle versioned delete based on specific versioning state
 				if object.VersionId != "" {
 					// Delete specific version (same for both enabled and suspended)
-					err := s3a.deleteSpecificObjectVersion(bucket, object.Key, object.VersionId)
+					// Pass prefetched entry to avoid duplicate lookup
+					err := s3a.deleteSpecificObjectVersion(bucket, object.Key, object.VersionId, prefetchedEntry)
 					if err != nil {
 						deleteErrors = append(deleteErrors, DeleteError{
 							Code:      "",
@@ -288,7 +305,8 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 						// Suspended versioning: Actually delete the "null" version object
 						glog.V(2).Infof("DeleteMultipleObjectsHandler: deleting null version for suspended versioning %s/%s", bucket, object.Key)
 
-						err := s3a.deleteSpecificObjectVersion(bucket, object.Key, "null")
+						// Pass prefetched entry to avoid duplicate lookup
+						err := s3a.deleteSpecificObjectVersion(bucket, object.Key, "null", prefetchedEntry)
 						if err != nil {
 							deleteErrors = append(deleteErrors, DeleteError{
 								Code:      "",

weed/s3api/s3api_object_handlers_put.go

@@ -166,12 +166,14 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 			return
 		}
 
-		// For non-versioned buckets, check if existing object has object lock protections
+		// For non-versioned buckets with object lock enabled, check if existing object has protections
 		// that would prevent overwrite (PUT operations overwrite existing objects in non-versioned buckets)
-		if !versioningConfigured {
+		// Skip this check for buckets without object lock to avoid expensive entry lookup
+		if !versioningConfigured && s3a.isObjectLockEnabled(bucket) {
 			governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
-			if err := s3a.enforceObjectLockProtections(r, bucket, object, "", governanceBypassAllowed); err != nil {
-				glog.V(2).Infof("PutObjectHandler: object lock permissions check failed for %s/%s: %v", bucket, object, err)
+			// Note: We ignore the returned entry since PUT creates a new entry anyway
+			if _, lockErr := s3a.enforceObjectLockProtections(r, bucket, object, "", governanceBypassAllowed); lockErr != nil {
+				glog.V(2).Infof("PutObjectHandler: object lock permissions check failed for %s/%s: %v", bucket, object, lockErr)
 				s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
 				return
 			}

weed/s3api/s3api_object_retention.go

@@ -584,8 +584,10 @@ func (s3a *S3ApiServer) evaluateGovernanceBypassRequest(r *http.Request, bucket,
 	return true
 }
 
-// enforceObjectLockProtections enforces object lock protections for operations
-func (s3a *S3ApiServer) enforceObjectLockProtections(request *http.Request, bucket, object, versionId string, governanceBypassAllowed bool) error {
+// enforceObjectLockProtections enforces object lock protections for operations.
+// Returns the fetched entry (if found) to allow callers to reuse it, avoiding duplicate lookups.
+// Returns (nil, nil) if object doesn't exist (not an error for delete operations).
+func (s3a *S3ApiServer) enforceObjectLockProtections(request *http.Request, bucket, object, versionId string, governanceBypassAllowed bool) (*filer_pb.Entry, error) {
 	// Get the object entry to check both retention and legal hold
 	// For delete operations without versionId, we need to check the latest version
 	var entry *filer_pb.Entry
@@ -604,10 +606,10 @@ func (s3a *S3ApiServer) enforceObjectLockProtections(request *http.Request, buck
 		if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrObjectNotFound) || errors.Is(err, ErrVersionNotFound) || errors.Is(err, ErrLatestVersionNotFound) {
 			// Object doesn't exist, so it can't be under retention or legal hold - this is normal
 			glog.V(4).Infof("Object %s/%s (versionId: %s) not found during object lock check (expected during delete operations)", bucket, object, versionId)
-			return nil
+			return nil, nil
 		}
 		glog.Warningf("Error retrieving object %s/%s (versionId: %s) for lock check: %v", bucket, object, versionId, err)
-		return err
+		return nil, err
 	}
 
 	// Extract retention information from the entry
@@ -626,25 +628,25 @@ func (s3a *S3ApiServer) enforceObjectLockProtections(request *http.Request, buck
 
 	// If object is under legal hold, it cannot be deleted or modified (including delete marker creation)
 	if legalHoldActive {
-		return ErrObjectUnderLegalHold
+		return entry, ErrObjectUnderLegalHold
 	}
 
 	// If object is under retention, check the mode
 	if retentionActive && retention != nil {
 		if retention.Mode == s3_constants.RetentionModeCompliance {
-			return ErrComplianceModeActive
+			return entry, ErrComplianceModeActive
 		}
 
 		if retention.Mode == s3_constants.RetentionModeGovernance {
 			if !governanceBypassAllowed {
-				return ErrGovernanceModeActive
+				return entry, ErrGovernanceModeActive
 			}
 			// Note: governanceBypassAllowed parameter is already validated by evaluateGovernanceBypassRequest()
 			// which checks both header presence and IAM permissions, so we trust it here
 		}
 	}
 
-	return nil
+	return entry, nil
 }
 
 // ====================================================================

weed/s3api/s3api_object_versioning.go

@@ -637,8 +637,10 @@ func (s3a *S3ApiServer) getSpecificObjectVersion(bucket, object, versionId strin
 	return entry, nil
 }
 
-// deleteSpecificObjectVersion deletes a specific version of an object
-func (s3a *S3ApiServer) deleteSpecificObjectVersion(bucket, object, versionId string) error {
+// deleteSpecificObjectVersion deletes a specific version of an object.
+// The optional prefetchedEntry parameter allows callers to pass an already-fetched entry
+// to avoid duplicate filer lookups (optimization for DELETE operations).
+func (s3a *S3ApiServer) deleteSpecificObjectVersion(bucket, object, versionId string, prefetchedEntry *filer_pb.Entry) error {
 	// Normalize object path to ensure consistency with toFilerPath behavior
 	normalizedObject := removeDuplicateSlashes(object)
 
@@ -651,12 +653,14 @@ func (s3a *S3ApiServer) deleteSpecificObjectVersion(bucket, object, versionId st
 		bucketDir := s3a.option.BucketsPath + "/" + bucket
 		cleanObject := strings.TrimPrefix(normalizedObject, "/")
 
-		// Check if the object exists
-		_, err := s3a.getEntry(bucketDir, cleanObject)
-		if err != nil {
-			// Object doesn't exist - this is OK for delete operations (idempotent)
-			glog.V(2).Infof("deleteSpecificObjectVersion: null version object %s already deleted or doesn't exist", cleanObject)
-			return nil
+		// Check if the object exists (skip if we have a prefetched entry)
+		if prefetchedEntry == nil {
+			_, err := s3a.getEntry(bucketDir, cleanObject)
+			if err != nil {
+				// Object doesn't exist - this is OK for delete operations (idempotent)
+				glog.V(2).Infof("deleteSpecificObjectVersion: null version object %s already deleted or doesn't exist", cleanObject)
+				return nil
+			}
 		}
 
 		// Delete the regular file