s3: route versioned COPY and delete-marker off the DLM (#9633)

s3: route versioned/suspended delete markers and versioned COPY off the lock

createDeleteMarker flips the .versions pointer via routedVersionedFinalize
(RECOMPUTE_LATEST on the owner filer) when an owner is known, so an Enabled or
Suspended DeleteObject takes its pointer flip off the distributed lock; the
delete marker file is written first and the owner re-derives the pointer.

DeleteObjectHandler routes a versioned/suspended delete with no specific version
straight to the owner, off the lock. A specific-version delete and object-lock
buckets keep the lock (the former needs a recompute-after-delete handled
separately; the latter needs gateway-side enforcement).

CopyObject into a versioned bucket finalizes the new version through the same
routed pointer flip.

weed/s3api/s3api_object_handlers_copy.go

@@ -444,7 +444,16 @@ func (s3a *S3ApiServer) finalizeCopyDestination(dstBucket, dstObject, dstVersion
 			return "", "", err
 		}
 
-		if err = s3a.updateLatestVersionInDirectory(dstBucket, normalizedObject, versionId, versionFileName, dstEntry); err != nil {
+		// Route the pointer flip to the owner filer when known (off the
+		// distributed lock); RECOMPUTE_LATEST picks the just-written version.
+		if owner := s3a.objectWriteOwner(dstBucket, normalizedObject); owner != "" {
+			if code := s3a.routedVersionedFinalize(owner, dstBucket, normalizedObject, isNewFormatVersionId(versionId)); code != s3err.ErrNone {
+				if rollbackErr := s3a.rollbackCopyVersion(bucketDir, versionObjectPath); rollbackErr != nil {
+					glog.Errorf("CopyObjectHandler: failed to rollback version %s for %s/%s after routed finalize error: %v", versionId, dstBucket, normalizedObject, rollbackErr)
+				}
+				return "", "", fmt.Errorf("routed finalize for %s/%s: code %d", dstBucket, normalizedObject, code)
+			}
+		} else if err = s3a.updateLatestVersionInDirectory(dstBucket, normalizedObject, versionId, versionFileName, dstEntry); err != nil {
 			if rollbackErr := s3a.rollbackCopyVersion(bucketDir, versionObjectPath); rollbackErr != nil {
 				glog.Errorf("CopyObjectHandler: failed to rollback version %s for %s/%s after latest pointer update error: %v", versionId, dstBucket, normalizedObject, rollbackErr)
 			}

weed/s3api/s3api_object_handlers_delete.go

@@ -238,6 +238,17 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque
 			}
 		}
 	}
+	// Versioned/suspended delete with no specific version: route off the lock when
+	// the bucket has an owner. createDeleteMarker routes its own pointer flip;
+	// object-lock buckets are excluded by routableWriteOwner and stay on the lock,
+	// and the If-Match precondition was already checked above. A specific-version
+	// delete keeps the lock (its recompute-after-delete is a separate change).
+	if !deleteHandled && versioningConfigured && versionId == "" {
+		if owner := s3a.routableWriteOwner(bucket, object); owner != "" {
+			deleteResult, deleteCode = s3a.deleteVersionedObject(r, bucket, object, versionId, versioningState)
+			deleteHandled = true
+		}
+	}
 	if !deleteHandled {
 		deleteCode = s3a.withObjectWriteLock(bucket, object, func() s3err.ErrorCode {
 			return s3a.checkDeleteIfMatch(bucket, object, versionId, versioningState, r.Header.Get(s3_constants.IfMatch), s3err.ErrPreconditionFailed)

weed/s3api/s3api_object_versioning.go

@@ -242,8 +242,13 @@ func (s3a *S3ApiServer) createDeleteMarker(bucket, object string) (string, error
 		},
 		Extended: deleteMarkerExtended,
 	}
-	err = s3a.updateLatestVersionInDirectory(bucket, cleanObject, versionId, versionFileName, deleteMarkerEntry)
-	if err != nil {
+	// Route the pointer flip to the owner filer when known (off the distributed
+	// lock); RECOMPUTE_LATEST picks the just-written marker as the new latest.
+	if owner := s3a.objectWriteOwner(bucket, cleanObject); owner != "" {
+		if code := s3a.routedVersionedFinalize(owner, bucket, cleanObject, useInvertedFormat); code != s3err.ErrNone {
+			return "", fmt.Errorf("createDeleteMarker: routed finalize failed for %s/%s: code %d", bucket, object, code)
+		}
+	} else if err = s3a.updateLatestVersionInDirectory(bucket, cleanObject, versionId, versionFileName, deleteMarkerEntry); err != nil {
 		glog.Errorf("createDeleteMarker: failed to update latest version in directory: %v", err)
 		return "", fmt.Errorf("failed to update latest version in directory: %w", err)
 	}