Fix error status when the user share the locked file

2023-05-22 15:59:48 +02:00
parent 2dfb20b354
commit 55b3bd2db7
21 changed files with 660 additions and 496 deletions
@@ -156,7 +156,7 @@ func (s *service) UnprotectedEndpoints() []string {
 	return []string{}
 }

-func (s *service) PullTransfer(ctx context.Context, req *datatx.PullTransferRequest) (*datatx.PullTransferResponse, error) {
+func (s *service) CreateTransfer(ctx context.Context, req *datatx.CreateTransferRequest) (*datatx.CreateTransferResponse, error) {
 	srcEp, err := s.extractEndpointInfo(ctx, req.SrcTargetUri)
 	if err != nil {
 		return nil, err
@@ -189,7 +189,7 @@ func (s *service) PullTransfer(ctx context.Context, req *datatx.PullTransferRequ
 	s.txShareDriver.model.TxShares[txInfo.GetId().OpaqueId] = txShare
 	if err := s.txShareDriver.model.saveTxShare(); err != nil {
 		err = errors.Wrap(err, "datatx service: error saving transfer share: "+datatx.Status_STATUS_INVALID.String())
-		return &datatx.PullTransferResponse{
+		return &datatx.CreateTransferResponse{
 			Status: status.NewInvalid(ctx, "error pulling transfer"),
 		}, err
 	}
@@ -197,13 +197,13 @@ func (s *service) PullTransfer(ctx context.Context, req *datatx.PullTransferRequ
 	// now check start transfer outcome
 	if startTransferErr != nil {
 		startTransferErr = errors.Wrap(startTransferErr, "datatx service: error starting transfer job")
-		return &datatx.PullTransferResponse{
+		return &datatx.CreateTransferResponse{
 			Status: status.NewInvalid(ctx, "datatx service: error pulling transfer"),
 			TxInfo: txInfo,
 		}, startTransferErr
 	}

-	return &datatx.PullTransferResponse{
+	return &datatx.CreateTransferResponse{
 		Status: status.NewOK(ctx),
 		TxInfo: txInfo,
 	}, err
@@ -27,15 +27,15 @@ import (
 	"github.com/pkg/errors"
 )

-func (s *svc) PullTransfer(ctx context.Context, req *datatx.PullTransferRequest) (*datatx.PullTransferResponse, error) {
+func (s *svc) CreateTransfer(ctx context.Context, req *datatx.CreateTransferRequest) (*datatx.CreateTransferResponse, error) {
 	c, err := pool.GetDataTxClient(s.c.DataTxEndpoint)
 	if err != nil {
-		return &datatx.PullTransferResponse{
+		return &datatx.CreateTransferResponse{
 			Status: status.NewInternal(ctx, "error getting data transfer client"),
 		}, nil
 	}

-	res, err := c.PullTransfer(ctx, req)
+	res, err := c.CreateTransfer(ctx, req)
 	if err != nil {
 		return nil, errors.Wrap(err, "gateway: error calling PullTransfer")
 	}
@@ -345,12 +345,12 @@ func (s *svc) UpdateReceivedOCMShare(ctx context.Context, req *ocm.UpdateReceive
 							},
 						},
 					}
-					req := &datatx.PullTransferRequest{
+					req := &datatx.CreateTransferRequest{
 						SrcTargetUri:  srcTargetURI,
 						DestTargetUri: destTargetURI,
 						Opaque:        opaqueObj,
 					}
-					res, err := s.PullTransfer(ctx, req)
+					res, err := s.CreateTransfer(ctx, req)
 					if err != nil {
 						log.Err(err).Msg("gateway: error calling PullTransfer")
 						return &ocm.UpdateReceivedOCMShareResponse{
@@ -468,6 +468,23 @@ func (s *svc) addShare(ctx context.Context, req *collaboration.CreateShareReques
 		return res, nil
 	}

+	rollBackFn := func(status *rpc.Status) {
+		rmvReq := &collaboration.RemoveShareRequest{
+			Ref: &collaboration.ShareReference{
+				Spec: &collaboration.ShareReference_Key{
+					Key: &collaboration.ShareKey{
+						ResourceId: req.ResourceInfo.Id,
+						Grantee:    req.Grant.Grantee,
+					},
+				},
+			},
+		}
+		appctx.GetLogger(ctx).Debug().Interface("status", status).Interface("req", req).Msg("rollback the CreateShare attempt")
+		if resp, err := s.removeShare(ctx, rmvReq); err != nil {
+			appctx.GetLogger(ctx).Debug().Interface("status", resp.GetStatus()).Interface("req", req).Msg(err.Error())
+		}
+	}
+
 	if s.c.CommitShareToStorageGrant {
 		// If the share is a denial we call  denyGrant instead.
 		var status *rpc.Status
@@ -479,6 +496,8 @@ func (s *svc) addShare(ctx context.Context, req *collaboration.CreateShareReques
 		} else {
 			status, err = s.addGrant(ctx, req.ResourceInfo.Id, req.Grant.Grantee, req.Grant.Permissions.Permissions, req.Grant.Expiration, nil)
 			if err != nil {
+				appctx.GetLogger(ctx).Debug().Interface("status", status).Interface("req", req).Msg(err.Error())
+				rollBackFn(status)
 				return nil, errors.Wrap(err, "gateway: error adding grant to storage")
 			}
 		}
@@ -488,7 +507,10 @@ func (s *svc) addShare(ctx context.Context, req *collaboration.CreateShareReques
 			s.statCache.RemoveStat(ctxpkg.ContextMustGetUser(ctx).GetId(), req.ResourceInfo.Id)
 		case rpc.Code_CODE_UNIMPLEMENTED:
 			appctx.GetLogger(ctx).Debug().Interface("status", status).Interface("req", req).Msg("storing grants not supported, ignoring")
+			rollBackFn(status)
 		default:
+			appctx.GetLogger(ctx).Debug().Interface("status", status).Interface("req", req).Msg("storing grants is not successful")
+			rollBackFn(status)
 			return &collaboration.CreateShareResponse{
 				Status: status,
 			}, err
@@ -112,7 +112,7 @@ func New(m map[string]interface{}) (global.Middleware, int, error) {
 	}

 	// TODO(jfd): use log from request context, otherwise fmt will be used to log,
-	// preventing us from pinging the log to eg jq
+	// preventing us from piping the log to eg jq
 	c := cors.New(cors.Options{
 		AllowCredentials:   conf.AllowCredentials,
 		AllowedHeaders:     conf.AllowedHeaders,
@@ -393,7 +393,7 @@ func (s *svc) doPatch(w http.ResponseWriter, r *http.Request) {
 func copyHeader(dst, src http.Header) {
 	for key, values := range src {
 		for i := range values {
-			dst.Add(key, values[i])
+			dst.Set(key, values[i])
 		}
 	}
 }
@@ -92,7 +92,7 @@ func (s *svc) handleSpacesTusPost(w http.ResponseWriter, r *http.Request, spaceI

 func (s *svc) handleTusPost(ctx context.Context, w http.ResponseWriter, r *http.Request, meta map[string]string, ref *provider.Reference, log zerolog.Logger) {
 	w.Header().Add(net.HeaderAccessControlAllowHeaders, strings.Join([]string{net.HeaderTusResumable, net.HeaderUploadLength, net.HeaderUploadMetadata, net.HeaderIfMatch}, ", "))
-	w.Header().Add(net.HeaderAccessControlExposeHeaders, strings.Join([]string{net.HeaderTusResumable, net.HeaderLocation}, ", "))
+	w.Header().Add(net.HeaderAccessControlExposeHeaders, strings.Join([]string{net.HeaderTusResumable, net.HeaderUploadOffset, net.HeaderLocation}, ", "))
 	w.Header().Set(net.HeaderTusExtension, "creation,creation-with-upload,checksum,expiration")

 	w.Header().Set(net.HeaderTusResumable, "1.0.0")
@@ -1453,6 +1453,12 @@ func (h *Handler) createCs3Share(ctx context.Context, w http.ResponseWriter, r *
 				Message: createShareResponse.Status.Message,
 				Error:   nil,
 			}
+		case rpc.Code_CODE_LOCKED:
+			return nil, &ocsError{
+				Code:    response.MetaLocked.StatusCode,
+				Message: response.MessageLockedForSharing,
+				Error:   nil,
+			}
 		default:
 			return nil, &ocsError{
 				Code:    response.MetaServerError.StatusCode,
@@ -127,6 +127,9 @@ var MetaBadRequest = Meta{Status: "error", StatusCode: 400, Message: "Bad Reques
 // MetaPathNotFound is returned when trying to share not existing resources
 var MetaPathNotFound = Meta{Status: "failure", StatusCode: 404, Message: MessagePathNotFound}

+// MetaLocked is returned when trying to share not existing resources
+var MetaLocked = Meta{Status: "failure", StatusCode: 423, Message: "The file is locked"}
+
 // MetaServerError is returned on server errors
 var MetaServerError = Meta{Status: "error", StatusCode: 996, Message: "Server Error"}

@@ -151,6 +154,9 @@ var MessagePathNotFound = "Wrong path, file/folder doesn't exist"
 // MessageShareExists is used when a user tries to create a new share for the same user
 var MessageShareExists = "A share for the recipient already exists"

+// MessageLockedForSharing is used when a user tries to create a new share until the file is in use by at least one user
+var MessageLockedForSharing = "The file is locked until the file is in use by at least one user"
+
 // WriteOCSSuccess handles writing successful ocs response data
 func WriteOCSSuccess(w http.ResponseWriter, r *http.Request, d interface{}) {
 	WriteOCSData(w, r, MetaOK, d, nil)
@@ -286,8 +286,8 @@ func NewErrtypeFromStatus(status *rpc.Status) error {
 			return Locked(strings.TrimPrefix(status.Message, "set lock: error: locked by "))
 		}
 		return PermissionDenied(status.Message)
-	// case rpc.Code_CODE_LOCKED:
-	//	return Locked(status.Message)
+	case rpc.Code_CODE_LOCKED:
+		return Locked(status.Message)
 	// case rpc.Code_CODE_DATA_LOSS: ?
 	//	IsPartialContent
 	case rpc.Code_CODE_ABORTED:
@@ -144,6 +144,15 @@ func NewConflict(ctx context.Context, err error, msg string) *rpc.Status {
 	}
 }

+// NewLocked returns a status Code_CODE_LOCKED
+func NewLocked(ctx context.Context, msg string) *rpc.Status {
+	return &rpc.Status{
+		Code:    rpc.Code_CODE_LOCKED,
+		Message: msg,
+		Trace:   getTrace(ctx),
+	}
+}
+
 // NewStatusFromErrType returns a status that corresponds to the given errtype
 func NewStatusFromErrType(ctx context.Context, msg string, err error) *rpc.Status {
 	switch e := err.(type) {
@@ -161,7 +170,7 @@ func NewStatusFromErrType(ctx context.Context, msg string, err error) *rpc.Statu
 	case errtypes.Locked:
 		// FIXME a locked error returns the current lockid
 		// FIXME use NewAborted as per the rpc code docs
-		return NewPermissionDenied(ctx, e, msg+": "+err.Error())
+		return NewLocked(ctx, msg+": "+err.Error())
 	case errtypes.Aborted:
 		return NewAborted(ctx, e, msg+": "+err.Error())
 	case errtypes.PreconditionFailed:
@@ -231,6 +240,7 @@ var httpStatusCode = map[rpc.Code]int{
 	rpc.Code_CODE_UNAVAILABLE:          http.StatusServiceUnavailable,
 	rpc.Code_CODE_UNIMPLEMENTED:        http.StatusNotImplemented,
 	rpc.Code_CODE_UNKNOWN:              http.StatusInternalServerError,
+	rpc.Code_CODE_LOCKED:               http.StatusLocked,
 }

 // HTTPStatusFromCode returns an HTTP status code for the rpc code. It returns
@@ -495,6 +495,39 @@ func (_m *GatewayAPIClient) CreateSymlink(ctx context.Context, in *providerv1bet
 	return r0, r1
 }

+// CreateTransfer provides a mock function with given fields: ctx, in, opts
+func (_m *GatewayAPIClient) CreateTransfer(ctx context.Context, in *txv1beta1.CreateTransferRequest, opts ...grpc.CallOption) (*txv1beta1.CreateTransferResponse, error) {
+	_va := make([]interface{}, len(opts))
+	for _i := range opts {
+		_va[_i] = opts[_i]
+	}
+	var _ca []interface{}
+	_ca = append(_ca, ctx, in)
+	_ca = append(_ca, _va...)
+	ret := _m.Called(_ca...)
+
+	var r0 *txv1beta1.CreateTransferResponse
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context, *txv1beta1.CreateTransferRequest, ...grpc.CallOption) (*txv1beta1.CreateTransferResponse, error)); ok {
+		return rf(ctx, in, opts...)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, *txv1beta1.CreateTransferRequest, ...grpc.CallOption) *txv1beta1.CreateTransferResponse); ok {
+		r0 = rf(ctx, in, opts...)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*txv1beta1.CreateTransferResponse)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, *txv1beta1.CreateTransferRequest, ...grpc.CallOption) error); ok {
+		r1 = rf(ctx, in, opts...)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // Delete provides a mock function with given fields: ctx, in, opts
 func (_m *GatewayAPIClient) Delete(ctx context.Context, in *providerv1beta1.DeleteRequest, opts ...grpc.CallOption) (*providerv1beta1.DeleteResponse, error) {
 	_va := make([]interface{}, len(opts))
@@ -2310,39 +2343,6 @@ func (_m *GatewayAPIClient) OpenInApp(ctx context.Context, in *gatewayv1beta1.Op
 	return r0, r1
 }

-// PullTransfer provides a mock function with given fields: ctx, in, opts
-func (_m *GatewayAPIClient) PullTransfer(ctx context.Context, in *txv1beta1.PullTransferRequest, opts ...grpc.CallOption) (*txv1beta1.PullTransferResponse, error) {
-	_va := make([]interface{}, len(opts))
-	for _i := range opts {
-		_va[_i] = opts[_i]
-	}
-	var _ca []interface{}
-	_ca = append(_ca, ctx, in)
-	_ca = append(_ca, _va...)
-	ret := _m.Called(_ca...)
-
-	var r0 *txv1beta1.PullTransferResponse
-	var r1 error
-	if rf, ok := ret.Get(0).(func(context.Context, *txv1beta1.PullTransferRequest, ...grpc.CallOption) (*txv1beta1.PullTransferResponse, error)); ok {
-		return rf(ctx, in, opts...)
-	}
-	if rf, ok := ret.Get(0).(func(context.Context, *txv1beta1.PullTransferRequest, ...grpc.CallOption) *txv1beta1.PullTransferResponse); ok {
-		r0 = rf(ctx, in, opts...)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(*txv1beta1.PullTransferResponse)
-		}
-	}
-
-	if rf, ok := ret.Get(1).(func(context.Context, *txv1beta1.PullTransferRequest, ...grpc.CallOption) error); ok {
-		r1 = rf(ctx, in, opts...)
-	} else {
-		r1 = ret.Error(1)
-	}
-
-	return r0, r1
-}
-
 // PurgeRecycle provides a mock function with given fields: ctx, in, opts
 func (_m *GatewayAPIClient) PurgeRecycle(ctx context.Context, in *providerv1beta1.PurgeRecycleRequest, opts ...grpc.CallOption) (*providerv1beta1.PurgeRecycleResponse, error) {
 	_va := make([]interface{}, len(opts))