support putifabsent

pkg/gateway/errors/errors.go

@@ -169,6 +169,7 @@ const (
 
 	// S3 extended errors.
 	ErrContentSHA256Mismatch
+	ErrObjectExists
 
 	// Lakefs errors
 	ERRLakeFSNotSupported
@@ -756,7 +757,11 @@ var Codes = errorCodeMap{
 		Description:    "Invalid version found in the request",
 		HTTPStatusCode: http.StatusNotFound,
 	},
-
+	ErrObjectExists: {
+		Code:           "ErrObjectExists",
+		Description:    "Object path already exists in DB",
+		HTTPStatusCode: http.StatusNotModified,
+	},
 	// LakeFS errors
 	ERRLakeFSNotSupported: {
 		Code:           "ERRLakeFSNotSupported",

pkg/gateway/operations/postobject.go

@@ -94,6 +94,11 @@ func (controller *PostObject) HandleCompleteMultipartUpload(w http.ResponseWrite
 		_ = o.EncodeError(w, req, err, gatewayErrors.Codes.ToAPIErr(gatewayErrors.ErrInternalError))
 		return
 	}
+	err = o.checkIfAbsent(req)
+	if err != nil {
+		_ = o.EncodeError(w, req, err, gatewayErrors.Codes.ToAPIErr(gatewayErrors.ErrObjectExists))
+		return
+	}
 	objName := multiPart.PhysicalAddress
 	req = req.WithContext(logging.AddFields(req.Context(), logging.Fields{logging.PhysicalAddressFieldKey: objName}))
 	xmlMultipartComplete, err := io.ReadAll(req.Body)

pkg/gateway/operations/putobject.go

@@ -20,6 +20,7 @@ import (
 )
 
 const (
+	IfNoneMatchHeader     = "If-None-Match"
 	CopySourceHeader      = "x-amz-copy-source"
 	CopySourceRangeHeader = "x-amz-copy-source-range"
 	QueryParamUploadID    = "uploadId"
@@ -30,7 +31,6 @@ type PutObject struct{}
 
 func (controller *PutObject) RequiredPermissions(req *http.Request, repoID, _, destPath string) (permissions.Node, error) {
 	copySource := req.Header.Get(CopySourceHeader)
-
 	if len(copySource) == 0 {
 		return permissions.Node{
 			Permission: permissions.Permission{
@@ -298,6 +298,11 @@ func handlePut(w http.ResponseWriter, req *http.Request, o *PathOperation) {
 	o.Incr("put_object", o.Principal, o.Repository.Name, o.Reference)
 	storageClass := StorageClassFromHeader(req.Header)
 	opts := block.PutOpts{StorageClass: storageClass}
+	err := o.checkIfAbsent(req)
+	if err != nil {
+		_ = o.EncodeError(w, req, err, gatewayErrors.Codes.ToAPIErr(gatewayErrors.ErrObjectExists))
+		return
+	}
 	address := o.PathProvider.NewPath()
 	blob, err := upload.WriteBlob(req.Context(), o.BlockStore, o.Repository.StorageNamespace, address, req.Body, req.ContentLength, opts)
 	if err != nil {
@@ -325,3 +330,22 @@ func handlePut(w http.ResponseWriter, req *http.Request, o *PathOperation) {
 	o.SetHeader(w, "ETag", httputil.ETag(blob.Checksum))
 	w.WriteHeader(http.StatusOK)
 }
+
+func (o *PathOperation) checkIfAbsent(req *http.Request) error {
+	Header := req.Header.Get(IfNoneMatchHeader)
+	switch Header {
+	case "":
+		return nil
+	case "*":
+		_, err := o.Catalog.GetEntry(req.Context(), o.Repository.Name, o.Reference, o.Path, catalog.GetEntryParams{})
+		if err == nil {
+			return gatewayErrors.ErrObjectExists
+		}
+	default:
+		_, err := o.Catalog.GetEntry(req.Context(), o.Repository.Name, o.Reference, Header, catalog.GetEntryParams{})
+		if err == nil {
+			return gatewayErrors.ErrObjectExists
+		}
+	}
+	return nil
+}