swift: reserve segments of dynamic large object when delete objects in container what was enabled versioning.

add code handle move object when moving the object is contained by the container what was enabled versioning with "X-History-Location".

backend/swift/swift.go

@@ -530,10 +530,10 @@ type listFn func(remote string, object *swift.Object, isDirectory bool) error
 //
 // Set recurse to read sub directories
 func (f *Fs) listContainerRoot(container, directory, prefix string, addContainer bool, recurse bool, fn listFn) error {
-	if prefix != "" {
+	if prefix != "" && !strings.HasSuffix(prefix, "/") {
 		prefix += "/"
 	}
-	if directory != "" {
+	if directory != "" && !strings.HasSuffix(directory, "/") {
 		directory += "/"
 	}
 	// Options for ObjectsWalk
@@ -952,6 +952,18 @@ func (o *Object) isStaticLargeObject() (bool, error) {
 	return o.hasHeader("X-Static-Large-Object")
 }
 
+func (o *Object) isInContainerVersioning() (bool, error) {
+	_, headers, err := o.fs.c.Container(o.fs.root)
+	if err != nil {
+		return false, err
+	}
+	xHistoryLocation := headers["X-History-Location"]
+	if len(xHistoryLocation) > 0 {
+		return true, nil
+	}
+	return false, nil
+}
+
 // Size returns the size of an object in bytes
 func (o *Object) Size() int64 {
 	return o.size
@@ -1083,9 +1095,8 @@ func min(x, y int64) int64 {
 //
 // if except is passed in then segments with that prefix won't be deleted
 func (o *Object) removeSegments(except string) error {
-	container, containerPath := o.split()
-	segmentsContainer := container + "_segments"
-	err := o.fs.listContainerRoot(segmentsContainer, containerPath, "", false, true, func(remote string, object *swift.Object, isDirectory bool) error {
+	segmentsContainer, prefix, err := o.getSegmentsDlo()
+	err = o.fs.listContainerRoot(segmentsContainer, prefix, "", false, true, func(remote string, object *swift.Object, isDirectory bool) error {
 		if isDirectory {
 			return nil
 		}
@@ -1114,6 +1125,19 @@ func (o *Object) removeSegments(except string) error {
 	return nil
 }
 
+func (o *Object) getSegmentsDlo() (segmentsContainer string, prefix string, err error) {
+	if err = o.readMetaData(); err != nil {
+		return
+	}
+	dirManifest := o.headers["X-Object-Manifest"]
+	delimiter := strings.Index(dirManifest, "/")
+	if len(dirManifest) == 0 || delimiter < 0 {
+		err = errors.New("Missing or wrong structure of manifest of Dynamic large object")
+		return
+	}
+	return dirManifest[:delimiter], dirManifest[delimiter+1:], nil
+}
+
 // urlEncode encodes a string so that it is a valid URL
 //
 // We don't use any of Go's standard methods as we need `/` not
@@ -1300,12 +1324,9 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 }
 
 // Remove an object
-func (o *Object) Remove(ctx context.Context) error {
+func (o *Object) Remove(ctx context.Context) (err error) {
 	container, containerPath := o.split()
-	isDynamicLargeObject, err := o.isDynamicLargeObject()
-	if err != nil {
-		return err
-	}
+
 	// Remove file/manifest first
 	err = o.fs.pacer.Call(func() (bool, error) {
 		err = o.fs.c.ObjectDelete(container, containerPath)
@@ -1314,13 +1335,23 @@ func (o *Object) Remove(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
+	isDynamicLargeObject, err := o.isDynamicLargeObject()
+	if err != nil {
+		return err
+	}
 	// ...then segments if required
 	if isDynamicLargeObject {
+		isInContainerVersioning, err := o.isInContainerVersioning()
+		if err != nil {
+			return err
+		}
+		if !isInContainerVersioning {
 			err = o.removeSegments("")
 			if err != nil {
 				return err
 			}
 		}
+	}
 	return nil
 }