From 699a24f7e57034c808360ff5950ca2d6701e6a24 Mon Sep 17 00:00:00 2001
From: Poorna <poornas@users.noreply.github.com>
Date: Tue, 4 Apr 2023 10:50:11 -0700
Subject: [PATCH] batch: validate versioning on src/tgt buckets (#16955)

---
 cmd/batch-handlers.go | 8 +++++---
 cmd/erasure-object.go | 2 +-
 2 files changed, 6 insertions(+), 4 deletions(-)

diff --git a/cmd/batch-handlers.go b/cmd/batch-handlers.go
index ca5b4147f..f7f67a707 100644
--- a/cmd/batch-handlers.go
+++ b/cmd/batch-handlers.go
@@ -1116,16 +1116,18 @@ func (r *BatchJobReplicateV1) Validate(ctx context.Context, job BatchJobRequest,
 	if r.Source.Bucket == "" {
 		return errInvalidArgument
 	}
+	var isRemoteToLocal bool
 	localBkt := r.Source.Bucket
 	if r.Source.Endpoint != "" {
 		localBkt = r.Target.Bucket
+		isRemoteToLocal = true
 	}
 	info, err := o.GetBucketInfo(ctx, localBkt, BucketOptions{})
 	if err != nil {
 		if isErrBucketNotFound(err) {
 			return batchReplicationJobError{
 				Code:           "NoSuchSourceBucket",
-				Description:    "The specified source bucket does not exist",
+				Description:    fmt.Sprintf("The specified bucket %s does not exist", localBkt),
 				HTTPStatusCode: http.StatusNotFound,
 			}
 		}
@@ -1222,8 +1224,8 @@ func (r *BatchJobReplicateV1) Validate(ctx context.Context, job BatchJobRequest,
 		}
 		return err
 	}
-
-	if info.Versioning && !vcfg.Enabled() {
+	// If source has versioning enabled, target must have versioning enabled
+	if (info.Versioning && !vcfg.Enabled() && !isRemoteToLocal) || (!info.Versioning && vcfg.Enabled() && isRemoteToLocal) {
 		return batchReplicationJobError{
 			Code: "InvalidBucketState",
 			Description: fmt.Sprintf("The source '%s' has versioning enabled, target '%s' must have versioning enabled",
diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go
index fac8be291..a9bed6534 100644
--- a/cmd/erasure-object.go
+++ b/cmd/erasure-object.go
@@ -998,7 +998,7 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 
 	if opts.CheckPrecondFn != nil {
 		obj, err := er.getObjectInfo(ctx, bucket, object, opts)
-		if err != nil && !isErrVersionNotFound(err) {
+		if err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
 			return objInfo, err
 		}
 		if opts.CheckPrecondFn(obj) {