Repository: minio/minio (language: go)

Issue:
Server-side replication of ~3 GB files retries endlessly over a slower network connection
<!--- Provide a general summary of the issue in the Title above -->
When server-side replication of buckets is configured and there are objects to be sent to the destination, "larger" objects never finish transferring and instead are retried endlessly. (For the purposes of this bug, "larger" means somewhere between 1 and 4 GB. I was able to reproduce the problem with bandwidth (limited using the `tc` utility) of 10 mbit--50 mbit.)

For test purposes, `Replication Workers` `Replication Failed Workers` in the 'API' section of 'Settings' were each set to 1.

#### Log messages (repeated in a loop, every 10--12 minutes):
```API: SYSTEM()
Time: 04:33:18 UTC 02/25/2023
DeploymentID: ...
Error: Unable to replicate for object testing-site/debian-live-11.4.0-amd64-cinnamon+nonfree.iso(517342eb-eeb5-4299-9dec-447ef53d3eee): Put "http://garrett0.lan:9000/testing/debian-live-11.4.0-amd64-cinnamon%2Bnonfree.iso?versionId=517342eb-eeb5-4299-9dec-447ef53d3eee": write tcp 172.17.0.2:58300->10.NN.NN.NN:9000: use of closed network connection (*errors.errorString)
       3: internal/logger/logger.go:258:logger.LogIf()
       2: cmd/bucket-replication.go:1377:cmd.ReplicateObjectInfo.replicateAll()
       1: cmd/bucket-replication.go:969:cmd.replicateObject.func2()
```
```API: SYSTEM()
Time: 04:43:40 UTC 02/25/2023
DeploymentID: ...
Error: Unable to replicate for object testing-site/debian-live-11.4.0-amd64-cinnamon+nonfree.iso(517342eb-eeb5-4299-9dec-447ef53d3eee): Put "http://garrett0.lan:9000/testing/debian-live-11.4.0-amd64-cinnamon%2Bnonfree.iso?versionId=517342eb-eeb5-4299-9dec-447ef53d3eee": write tcp 172.17.0.2:60684->10.NN.NN.NN:9000: use of closed network connection (*errors.errorString)
       3: internal/logger/logger.go:258:logger.LogIf()
       2: cmd/bucket-replication.go:1377:cmd.ReplicateObjectInfo.replicateAll()
       1: cmd/bucket-replication.go:969:cmd.replicateObject.func2()
```
#### Disk and Network I/O
The network connection from the source to the destination cluster is roughly saturated, as measured by `iftop`.
During one of my tests at 10 mbit, I watched how the file sizes on the destination drive (Erasure backend) changed over time using:
```du -h --summarize /mnt/DISK0/minio/.minio.sys/tmp /mnt/DISK0/minio/.minio.sys/multipart```

`.minio.sys/tmp` would grow up to about 145 GB, then roll over to zero as a new `part.NNN` file was added to `.minio.sys/multipart`. Then eventually, when `.minio.sys/multipart` was 19--50 GB in size, both locations would go down close to zero size, and the data was apparently deleted, since I couldn't account for anywhere it was taking up space on disk using `du`.
Each `part.NNN` in `.minio.sys/multipart/<hash>/<UUID>/<UUID>` is 145 MB.

## Expected Behavior
<!--- If you're describing a bug, tell us what should happen -->
<!--- If you're suggesting a change/improvement, tell us how it should work -->
After configuring server-side bucket replication and adding objects to the source bucket, disk usage in the subdirectories of the destination cluster/drive responsible for receiving incoming objects from server-side replication should monotonically increase until an entire file is transferred and is visible in the Object Browser of the Web UI.

## Current Behavior
<!--- If describing a bug, tell us what happens instead of the expected behavior -->
<!--- If suggesting a change/improvement, explain the difference from current behavior -->
Network bandwidth from the source cluster to the destination cluster is nearly saturated, the sizes of `.minio.sys/tmp` and `.minio.sys/multipart` keep changing and getting reset back to low numbers, and log messages appear every 10--15 minutes, however the objects are never transferred successfully and never show up in the destination bucket as visible from the Web UI.

## Possible Solution
<!--- Not obligatory, but suggest a fix/reason for the bug, -->
<!--- or ideas how to implement the addition or change -->
Perhaps the log error messages,
`Error: Unable to replicate for object ...` and
`use of closed network connection (*errors.errorString)` cited above hint at a root cause?

## Steps to Reproduce (for bugs)
<!--- Provide a link to a live example, or an unambiguous set of steps to -->
<!--- reproduce this bug. Include code to reproduce, if relevant -->
<!--- and make sure you have followed https://github.com/minio/minio/tree/release/docs/debugging to capture relevant logs -->

1. Run Minio in docker on the source and destination servers
2. Install `iftop` on your destination server and `iproute2` on your source server, if necessary.
3. Set up credentials. Source must have credentials to replicate the bucket in question to the destination.
4. In the Settings --> API screen of the source system, set Replication Workers and Replication Failed Workers each to 1.
5. Create buckets on the source and destination systems, (e.g. "testing").
6. Set up `mc` to control the source system (alias/credentials). Run `mc replicate add <source-server-alias>/testing --remote-bucket 'http://<access-key>:<secret-key>@<destination-server>:9000/testing' --replicate 'existing-objects'`
7. Verify the replication is configured correct by uploading a < 1MB file to the source bucket and verifying this file/object is replicated to the source bucket.
8. On the source system run `tc qdisc add dev <iface-name> root handle 1: tbf rate 20mbit latency 30ms burst 8192
9. The test files I used are available at [https://cdimage.debian.org/cdimage/unofficial/non-free/cd-including-firmware/current-live/amd64/iso-hybrid/](url): `debian-live-11.4.0-amd64-cinnamon+nonfree.iso` and `debian-live-11.4.0-amd64-gnome+nonfree.iso`. Upload a test file of 2--4 GB to your source bucket using web UI.
10. Run `iftop -i <iface-name>` on your destination server to verify that data is being received (18 mbps or more).
11. On the source server, do `docker logs --follow minio` to watch the log messages.
12. Wait 15--20 minutes for one or both of the log messages cited above to appear.
13. Wait an hour or so if desired, to convince yourself the file should have transferred by now. The network bandwidth usage will continue, as evidenced by `iftop`.

## Context
<!--- How has this issue affected you? What are you trying to accomplish? -->
<!--- Providing context helps us come up with a solution that is most useful in the real world -->
I want to set up a server for offsite replication (personal and home office data over residential internet connections).

## Regression
<!-- Is this issue a regression? (Yes / No) -->
No
<!-- If Yes, optionally please include minio version or commit id or PR# that caused this regression, if you have these details. -->

## Your Environment
<!--- Include as many relevant details about the environment you experienced the bug in -->
* Version used (`minio --version`): RELEASE.2023-02-22T18-23-45Z (go1.19.6 linux/amd64)
* Server setup and configuration: Ubuntu 22.04.1 LTS (GNU/Linux 5.15.0-60-generic x86_64)
* Both source and destination have Erasure backend. Source has one disk; destination has two servers with three disks each.
* Docker version 23.0.1, build a5ee5b1
* Operating System and version (`uname -a`): Linux hdd-utility 5.15.0-60-generic #66-Ubuntu SMP Fri Jan 20 14:29:49 UTC 2023 x86_64 x86_64 x86_64 GNU/Linux



Discussion / hints:
@sebastiandanconia will check.
I left something out of the log messages I quoted in my original post. I meant to post two different messages instead of two of the same. Here's an example of the one I missed:
```API: SYSTEM()
Time: 05:32:36 UTC 02/25/2023
DeploymentID: ...
Error: failed to get lock for object: debian-live-11.4.0-amd64-cinnamon+nonfree.iso bucket:testing-site arn: (*errors.errorString)
       3: internal/logger/logger.go:258:logger.LogIf()
       2: cmd/bucket-replication.go:941:cmd.replicateObject()
       1: cmd/bucket-replication.go:1630:cmd.(*ReplicationPool).AddMRFWorker()
```
The two types of log message, `Unable to replicate for object` and `use of closed network connection (*errors.errorString)`, are interspersed.
> 

This is normal @sebastiandanconia  because of the previous attempt. There are two concurrent replication in progress.

Should be fairly simply to understand and reproduce.

Fix patch:
diff --git a/cmd/bucket-replication.go b/cmd/bucket-replication.go
index 331c2f430b426..24c7547fd58c3 100644
--- a/cmd/bucket-replication.go
+++ b/cmd/bucket-replication.go
@@ -802,10 +802,12 @@ func getReplicationAction(oi1 ObjectInfo, oi2 minio.ObjectInfo, opType replicati
 		oi1.ModTime.Unix() > oi2.LastModified.Unix() && oi1.VersionID == nullVersionID {
 		return replicateNone
 	}
+	sz, _ := oi1.GetActualSize()
+
 	// needs full replication
 	if oi1.ETag != oi2.ETag ||
 		oi1.VersionID != oi2.VersionID ||
-		oi1.Size != oi2.Size ||
+		sz != oi2.Size ||
 		oi1.DeleteMarker != oi2.IsDeleteMarker ||
 		oi1.ModTime.Unix() != oi2.LastModified.Unix() {
 		return replicateAll
@@ -1202,12 +1204,6 @@ func (ri ReplicateObjectInfo) replicateAll(ctx context.Context, objectAPI Object
 		ReplicationAction:     rAction,
 	}
 
-	if ri.ObjectInfo.TargetReplicationStatus(tgt.ARN) == replication.Completed && !ri.ExistingObjResync.Empty() && !ri.ExistingObjResync.mustResyncTarget(tgt.ARN) {
-		rinfo.ReplicationStatus = replication.Completed
-		rinfo.ReplicationResynced = true
-		return
-	}
-
 	if globalBucketTargetSys.isOffline(tgt.EndpointURL()) {
 		logger.LogIf(ctx, fmt.Errorf("remote target is offline for bucket:%s arn:%s", bucket, tgt.ARN))
 		sendEvent(eventArgs{
@@ -1243,6 +1239,13 @@ func (ri ReplicateObjectInfo) replicateAll(ctx context.Context, objectAPI Object
 
 	objInfo = gr.ObjInfo
 
+	// use latest ObjectInfo to check if previous replication attempt succeeded
+	if objInfo.TargetReplicationStatus(tgt.ARN) == replication.Completed && !ri.ExistingObjResync.Empty() && !ri.ExistingObjResync.mustResyncTarget(tgt.ARN) {
+		rinfo.ReplicationStatus = replication.Completed
+		rinfo.ReplicationResynced = true
+		return
+	}
+
 	size, err := objInfo.GetActualSize()
 	if err != nil {
 		logger.LogIf(ctx, err)
@@ -1256,7 +1259,7 @@ func (ri ReplicateObjectInfo) replicateAll(ctx context.Context, objectAPI Object
 	}
 
 	if tgt.Bucket == "" {
-		logger.LogIf(ctx, fmt.Errorf("Unable to replicate object %s(%s), bucket is empty", objInfo.Name, objInfo.VersionID))
+		logger.LogIf(ctx, fmt.Errorf("unable to replicate object %s(%s), bucket is empty", objInfo.Name, objInfo.VersionID))
 		sendEvent(eventArgs{
 			EventName:  event.ObjectReplicationNotTracked,
 			BucketName: bucket,
@@ -1286,7 +1289,7 @@ func (ri ReplicateObjectInfo) replicateAll(ctx context.Context, objectAPI Object
 		if rAction == replicateNone {
 			if ri.OpType == replication.ExistingObjectReplicationType &&
 				objInfo.ModTime.Unix() > oi.LastModified.Unix() && objInfo.VersionID == nullVersionID {
-				logger.LogIf(ctx, fmt.Errorf("Unable to replicate %s/%s (null). Newer version exists on target", bucket, object))
+				logger.LogIf(ctx, fmt.Errorf("unable to replicate %s/%s (null). Newer version exists on target", bucket, object))
 				sendEvent(eventArgs{
 					EventName:  event.ObjectReplicationNotTracked,
 					BucketName: bucket,
@@ -1328,7 +1331,7 @@ func (ri ReplicateObjectInfo) replicateAll(ctx context.Context, objectAPI Object
 		}
 		if _, err = c.CopyObject(ctx, tgt.Bucket, object, tgt.Bucket, object, getCopyObjMetadata(objInfo, tgt.StorageClass), srcOpts, dstOpts); err != nil {
 			rinfo.ReplicationStatus = replication.Failed
-			logger.LogIf(ctx, fmt.Errorf("Unable to replicate metadata for object %s/%s(%s): %s", bucket, objInfo.Name, objInfo.VersionID, err))
+			logger.LogIf(ctx, fmt.Errorf("unable to replicate metadata for object %s/%s(%s): %s", bucket, objInfo.Name, objInfo.VersionID, err))
 		}
 	} else {
 		var putOpts minio.PutObjectOptions
@@ -1365,7 +1368,7 @@ func (ri ReplicateObjectInfo) replicateAll(ctx context.Context, objectAPI Object
 				r, objInfo, putOpts); err != nil {
 				if minio.ToErrorResponse(err).Code != "PreconditionFailed" {
 					rinfo.ReplicationStatus = replication.Failed
-					logger.LogIf(ctx, fmt.Errorf("Unable to replicate for object %s/%s(%s): %s", bucket, objInfo.Name, objInfo.VersionID, err))
+					logger.LogIf(ctx, fmt.Errorf("unable to replicate for object %s/%s(%s): %s", bucket, objInfo.Name, objInfo.VersionID, err))
 				} else {
 					rinfo.ReplicationStatus = replication.Completed
 				}
@@ -1374,7 +1377,7 @@ func (ri ReplicateObjectInfo) replicateAll(ctx context.Context, objectAPI Object
 			if _, err = c.PutObject(ctx, tgt.Bucket, object, r, size, "", "", putOpts); err != nil {
 				if minio.ToErrorResponse(err).Code != "PreconditionFailed" {
 					rinfo.ReplicationStatus = replication.Failed
-					logger.LogIf(ctx, fmt.Errorf("Unable to replicate for object %s/%s(%s): %s", bucket, objInfo.Name, objInfo.VersionID, err))
+					logger.LogIf(ctx, fmt.Errorf("unable to replicate for object %s/%s(%s): %s", bucket, objInfo.Name, objInfo.VersionID, err))
 				} else {
 					rinfo.ReplicationStatus = replication.Completed
 				}


Test patch: