s3api: skip object lock check for buckets without object lockorigin/optimize-delete-lookups

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.

3 files changed, 47 insertions, 29 deletions

diff --git a/weed/s3api/s3api_bucket_config.go b/weed/s3api/s3api_bucket_config.go
index a10374339..2df84179a 100644
--- a/weed/s3api/s3api_bucket_config.go
+++ b/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 {
diff --git a/weed/s3api/s3api_object_handlers_delete.go b/weed/s3api/s3api_object_handlers_delete.go
index 1cc11da70..1040f5372 100644
--- a/weed/s3api/s3api_object_handlers_delete.go
+++ b/weed/s3api/s3api_object_handlers_delete.go
@@ -53,17 +53,20 @@ 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
-			// The returned entry is reused to avoid duplicate filer lookups
-			governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
-			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
+			// 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, passing prefetched entry to avoid duplicate lookup
+			// 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)
@@ -93,17 +96,20 @@ 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
-				// The returned entry is reused to avoid duplicate filer lookups
-				governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
-				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
+				// 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), passing prefetched entry
+				// Delete the "null" version (the regular file)
 				err := s3a.deleteSpecificObjectVersion(bucket, object, "null", prefetchedEntry)
 				if err != nil {
 					glog.Errorf("Failed to delete null version: %v", err)
@@ -117,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 _, 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
+		// 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))
@@ -238,10 +247,9 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 				continue
 			}
 
-			// Check object lock permissions before deletion (only for versioned buckets)
-			// The returned entry is reused to avoid duplicate filer lookups
+			// Only check object lock if enabled for this bucket (avoids expensive entry lookup)
 			var prefetchedEntry *filer_pb.Entry
-			if versioningConfigured {
+			if versioningConfigured && s3a.isObjectLockEnabled(bucket) {
 				// Validate governance bypass for this specific object
 				governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object.Key)
 				var lockErr error
diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go
index 88166c4a5..31c33a225 100644
--- a/weed/s3api/s3api_object_handlers_put.go
+++ b/weed/s3api/s3api_object_handlers_put.go
@@ -166,9 +166,10 @@ 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)
 			// Note: We ignore the returned entry since PUT creates a new entry anyway
 			if _, lockErr := s3a.enforceObjectLockProtections(r, bucket, object, "", governanceBypassAllowed); lockErr != nil {