function objectPutPart(authInfo, request, streamingV4Params, log,
    cb) {
    log.debug('processing request', { method: 'objectPutPart' });
    const size = request.parsedContentLength;

    const putVersionId = request.headers['x-scal-s3-version-id'];
    const isPutVersion = putVersionId || putVersionId === '';

    if (Number.parseInt(size, 10) > constants.maximumAllowedPartSize) {
        log.debug('put part size too large', { size });
        monitoring.promMetrics('PUT', request.bucketName, 400,
            'putObjectPart');
        return cb(errors.EntityTooLarge);
    }

    const checksumHeaderErr = validateChecksumHeaders(request.headers);
    if (checksumHeaderErr) {
        return cb(checksumHeaderErr);
    }

    // Note: Part sizes cannot be less than 5MB in size except for the last.
    // However, we do not check this value here because we cannot know which
    // part will be the last until a complete MPU request is made. Thus, we let
    // the completeMultipartUpload API check that all parts except the last are
    // at least 5MB.

    const partNumber = Number.parseInt(request.query.partNumber, 10);
    // AWS caps partNumbers at 10,000
    if (partNumber > 10000) {
        monitoring.promMetrics('PUT', request.bucketName, 400,
            'putObjectPart');
        return cb(errors.TooManyParts);
    }
    if (!Number.isInteger(partNumber) || partNumber < 1) {
        monitoring.promMetrics('PUT', request.bucketName, 400,
            'putObjectPart');
        return cb(errors.InvalidArgument);
    }
    const bucketName = request.bucketName;
    assert.strictEqual(typeof bucketName, 'string');
    const canonicalID = authInfo.getCanonicalID();
    assert.strictEqual(typeof canonicalID, 'string');
    log.trace('owner canonicalid to send to data', {
        canonicalID: authInfo.getCanonicalID,
    });
    // Note that keys in the query object retain their case, so
    // `request.query.uploadId` must be called with that exact capitalization.
    const { query: { uploadId } } = request;
    const mpuBucketName = `${constants.mpuBucketPrefix}${bucketName}`;
    const { objectKey } = request;
    const originalIdentityAuthzResults = request.actionImplicitDenies;
    // For validating the request at the destinationBucket level the
    // `requestType` is the general 'objectPut'.
    const requestType = request.apiMethods || 'objectPutPart';

    return async.waterfall([
        // Get the destination bucket.
        next => metadata.getBucket(bucketName, log,
            (err, destinationBucket) => {
                if (err?.is?.NoSuchBucket) {
                    return next(errors.NoSuchBucket, destinationBucket);
                }
                if (err) {
                    log.error('error getting the destination bucket', {
                        error: err,
                        method: 'objectPutPart::metadata.getBucket',
                    });
                    return next(err, destinationBucket);
                }
                return next(null, destinationBucket);
            }),
        // Check the bucket authorization.
        (destinationBucket, next) => {
            if (!isBucketAuthorized(destinationBucket, requestType, canonicalID, authInfo,
                log, request, request.actionImplicitDenies)) {
                log.debug('access denied for user on bucket', { requestType });
                return next(errors.AccessDenied, destinationBucket);
            }
            return next(null, destinationBucket);
        },
        (destinationBucket, next) => validateQuotas(request, destinationBucket, request.accountQuotas,
            requestType, request.apiMethod, size, isPutVersion, log, err => next(err, destinationBucket)),
        // Validate that no object SSE is provided for part.
        // Part must use SSE from initiateMPU (overview in metadata)
        (destinationBucket, next) => {
            const { error, objectSSE } = parseObjectEncryptionHeaders(request.headers);
            if (error) {
                return next(error, destinationBucket);
            }
            if (objectSSE.algorithm) {
                return next(errors.InvalidArgument.customizeDescription(
                    'x-amz-server-side-encryption header is not supported for this operation.'));
            }
            return next(null, destinationBucket);
        },
        // Get the MPU shadow bucket.
        (destinationBucket, next) =>
            metadata.getBucket(mpuBucketName, log,
            (err, mpuBucket) => {
                if (err?.is?.NoSuchBucket) {
                    return next(errors.NoSuchUpload, destinationBucket);
                }
                if (err) {
                    log.error('error getting the shadow mpu bucket', {
                        error: err,
                        method: 'objectPutPart::metadata.getBucket',
                    });
                    return next(err, destinationBucket);
                }
                let splitter = constants.splitter;
                // BACKWARD: Remove to remove the old splitter
                if (mpuBucket.getMdBucketModelVersion() < 2) {
                    splitter = constants.oldSplitter;
                }
                return next(null, destinationBucket, splitter);
            }),
        // Check authorization of the MPU shadow bucket.
        (destinationBucket, splitter, next) => {
            const mpuOverviewKey = _getOverviewKey(splitter, objectKey,
                uploadId);
            return metadata.getObjectMD(mpuBucketName, mpuOverviewKey, {}, log,
                (err, res) => {
                    if (err) {
                        log.error('error getting the object from mpu bucket', {
                            error: err,
                            method: 'objectPutPart::metadata.getObjectMD',
                        });
                        return next(err, destinationBucket);
                    }
                    const initiatorID = res.initiator.ID;
                    const requesterID = authInfo.isRequesterAnIAMUser() ?
                        authInfo.getArn() : authInfo.getCanonicalID();
                    if (initiatorID !== requesterID) {
                        return next(errors.AccessDenied, destinationBucket);
                    }

                    const objectLocationConstraint =
                        res.controllingLocationConstraint;
                    const sseAlgo = res['x-amz-server-side-encryption'];
                    const sse = sseAlgo ? {
                        algorithm: sseAlgo,
                        masterKeyId: res['x-amz-server-side-encryption-aws-kms-key-id'],
                    } : null;
                    return next(null, destinationBucket,
                        objectLocationConstraint,
                        sse, splitter);
                });
        },
        // Use MPU overview SSE config
        (destinationBucket, objectLocationConstraint, encryption, splitter, next) => {
            // If MPU has server-side encryption, pass the `res` value
            if (encryption) {
                return kms.createCipherBundle(encryption, log, (err, res) => {
                    if (err) {
                        log.error('error processing the cipher bundle for ' +
                            'the destination bucket', {
                                error: err,
                            });
                        return next(err, destinationBucket);
                    }
                    return next(null, destinationBucket, objectLocationConstraint, res, splitter);
                // Allow KMS to use a key from previous provider (if sseMigration configured)
                // Because ongoing MPU started before sseMigration is no migrated
                }, { previousOk: true });
            }
            // The MPU does not have server-side encryption, so pass `null`
            return next(null, destinationBucket, objectLocationConstraint, null, splitter);
        },
        // If data backend is backend that handles mpu (like real AWS),
        // no need to store part info in metadata
        (destinationBucket, objectLocationConstraint, cipherBundle,
        splitter, next) => {
            const mpuInfo = {
                destinationBucket,
                size,
                objectKey,
                uploadId,
                partNumber,
                bucketName,
            };
            // eslint-disable-next-line no-param-reassign
            delete request.actionImplicitDenies;
            writeContinue(request, request._response);
            return data.putPart(request, mpuInfo, streamingV4Params,
            objectLocationConstraint, locationConstraintCheck, log,
            (err, partInfo, updatedObjectLC) => {
                if (err) {
                    return next(err, destinationBucket);
                }
                // if data backend handles mpu, skip to end of waterfall
                // TODO CLDSRV-640 (artesca) data backend should return SSE to include in response headers
                if (partInfo && partInfo.dataStoreType === 'aws_s3') {
                    return next(skipError, destinationBucket,
                        partInfo.dataStoreETag);
                }
                // partInfo will be null if data backend is not external
                // if the object location constraint undefined because
                // mpu was initiated in legacy version, update it
                return next(null, destinationBucket, updatedObjectLC,
                    cipherBundle, splitter, partInfo);
            });
        },
        // Get any pre-existing part.
        (destinationBucket, objectLocationConstraint, cipherBundle,
            splitter, partInfo, next) => {
            const paddedPartNumber = _getPaddedPartNumber(partNumber);
            const partKey = _getPartKey(uploadId, splitter, paddedPartNumber);
            return metadata.getObjectMD(mpuBucketName, partKey, {}, log,
                (err, res) => {
                    // If there is no object with the same key, continue.
                    if (err && !err.is.NoSuchKey) {
                        log.error('error getting current part (if any)', {
                            error: err,
                            method: 'objectPutPart::metadata.getObjectMD',
                        });
                        return next(err, destinationBucket);
                    }
                    let prevObjectSize = null;
                    let oldLocations = null;
                    // If there is a pre-existing part, update utapi metrics and
                    // get the locations of data to be overwritten.
                    if (res) {
                        prevObjectSize = res['content-length'];
                        // Pull locations to clean up any potential orphans in
                        // data if object put is an overwrite of a pre-existing
                        // object with the same key and part number.
                        oldLocations = Array.isArray(res.partLocations) ?
                            res.partLocations : [res.partLocations];
                    }
                    return next(null, destinationBucket,
                        objectLocationConstraint, cipherBundle,
                        partKey, prevObjectSize, oldLocations, partInfo, splitter);
                });
        },
        // Store in data backend.
        (destinationBucket, objectLocationConstraint, cipherBundle,
        partKey, prevObjectSize, oldLocations, partInfo, splitter, next) => {
            // NOTE: set oldLocations to null so we do not batchDelete for now
            if (partInfo &&
                constants.skipBatchDeleteBackends[partInfo.dataStoreType]) {
                // skip to storing metadata
                return next(null, destinationBucket, partInfo,
                  partInfo.dataStoreETag,
                  cipherBundle, partKey, prevObjectSize, null,
                  objectLocationConstraint, splitter);
            }
            const objectContext = {
                bucketName,
                owner: canonicalID,
                namespace: request.namespace,
                objectKey,
                partNumber: _getPaddedPartNumber(partNumber),
                uploadId,
            };
            const backendInfo = new BackendInfo(config,
                objectLocationConstraint);
            return dataStore(objectContext, cipherBundle, request,
                size, streamingV4Params, backendInfo, log,
                (err, dataGetInfo, hexDigest) => {
                    if (err) {
                        return next(err, destinationBucket);
                    }
                    return next(null, destinationBucket, dataGetInfo, hexDigest,
                        cipherBundle, partKey, prevObjectSize, oldLocations,
                        objectLocationConstraint, splitter);
                });
        },
        // Store data locations in metadata and delete any overwritten
        // data if completeMPU hasn't been initiated yet.
        (destinationBucket, dataGetInfo, hexDigest, cipherBundle, partKey,
         prevObjectSize, oldLocations, objectLocationConstraint, splitter, next) => {
            // Use an array to be consistent with objectPutCopyPart where there
            // could be multiple locations.
            const partLocations = [dataGetInfo];
            const sseHeaders = {};
            if (cipherBundle) {
                const { algorithm, masterKeyId, cryptoScheme,
                    cipheredDataKey } = cipherBundle;
                partLocations[0].sseAlgorithm = algorithm;
                partLocations[0].sseMasterKeyId = masterKeyId;
                partLocations[0].sseCryptoScheme = cryptoScheme;
                partLocations[0].sseCipheredDataKey = cipheredDataKey;
                sseHeaders.algo = algorithm;
                sseHeaders.kmsKey = masterKeyId;
            }
            const omVal = {
                // back to Version 3 since number-subparts is not needed
                'md-model-version': 3,
                partLocations,
                'key': partKey,
                'last-modified': new Date().toJSON(),
                'content-md5': hexDigest,
                'content-length': size,
                'owner-id': destinationBucket.getOwner(),
            };
            const mdParams = { overheadField: constants.overheadField };
            return metadata.putObjectMD(mpuBucketName, partKey, omVal, mdParams, log,
                err => {
                    if (err) {
                        log.error('error putting object in mpu bucket', {
                            error: err,
                            method: 'objectPutPart::metadata.putObjectMD',
                        });
                        return next(err, destinationBucket);
                    }
                    return next(null, partLocations, oldLocations, objectLocationConstraint,
                        destinationBucket, hexDigest, sseHeaders, prevObjectSize, splitter);
                });
        },
        (partLocations, oldLocations, objectLocationConstraint, destinationBucket,
        hexDigest, sseHeaders, prevObjectSize, splitter, next) => {
            if (!oldLocations) {
                return next(null, oldLocations, objectLocationConstraint,
                    destinationBucket, hexDigest, sseHeaders, prevObjectSize);
            }
            return services.isCompleteMPUInProgress({
                bucketName,
                objectKey,
                uploadId,
                splitter,
            }, log, (err, completeInProgress) => {
                if (err) {
                    return next(err, destinationBucket);
                }
                let oldLocationsToDelete = oldLocations;
                // Prevent deletion of old data if a completeMPU
                // is already in progress because then there is no
                // guarantee that the old location will not be the
                // committed one.
                if (completeInProgress) {
                    log.warn('not deleting old locations because CompleteMPU is in progress', {
                        method: 'objectPutPart::metadata.getObjectMD',
                        bucketName,
                        objectKey,
                        uploadId,
                        partLocations,
                        oldLocations,
                    });
                    oldLocationsToDelete = null;
                }
                return next(null, oldLocationsToDelete, objectLocationConstraint,
                    destinationBucket, hexDigest, sseHeaders, prevObjectSize);
            });
        },
        // Clean up any old data now that new metadata (with new
        // data locations) has been stored.
        (oldLocationsToDelete, objectLocationConstraint, destinationBucket, hexDigest,
            sseHeaders, prevObjectSize, next) => {
            if (oldLocationsToDelete) {
                log.trace('overwriting mpu part, deleting data');
                return data.batchDelete(oldLocationsToDelete, request.method,
                    objectLocationConstraint, log, err => {
                        if (err) {
                            // if error, log the error and move on as it is not
                            // relevant to the client as the client's
                            // object already succeeded putting data, metadata
                            log.error('error deleting existing data',
                                { error: err });
                        }
                        return next(null, destinationBucket, hexDigest,
                            sseHeaders, prevObjectSize);
                    });
            }
            return next(null, destinationBucket, hexDigest,
                sseHeaders, prevObjectSize);
        },
    ], (err, destinationBucket, hexDigest, sseHeaders, prevObjectSize) => {
        const corsHeaders = collectCorsHeaders(request.headers.origin,
            request.method, destinationBucket);
        // eslint-disable-next-line no-param-reassign
        request.actionImplicitDenies = originalIdentityAuthzResults;
        if (sseHeaders) {
            setSSEHeaders(corsHeaders, sseHeaders.algo, sseHeaders.kmsKey);
        }
        if (err) {
            if (err === skipError) {
                return cb(null, hexDigest, corsHeaders);
            }
            log.error('error in object put part (upload part)', {
                error: err,
                method: 'objectPutPart',
            });
            monitoring.promMetrics('PUT', bucketName, err.code,
                'putObjectPart');
            return cb(err, null, corsHeaders);
        }
        pushMetric('uploadPart', log, {
            authInfo,
            canonicalID: destinationBucket.getOwner(),
            bucket: bucketName,
            keys: [objectKey],
            newByteLength: size,
            oldByteLength: prevObjectSize,
            location: destinationBucket.getLocationConstraint(),
        });
        monitoring.promMetrics('PUT', bucketName,
            '200', 'putObjectPart', size, prevObjectSize);
        return cb(null, hexDigest, corsHeaders);
    });
}
PUT part of object during a multipart upload. Steps include:
validating metadata for authorization, bucket existence
and multipart upload initiation existence,
store object data in datastore upon successful authorization,
store object location returned by datastore in metadata and
return the result in final cb

@param {AuthInfo} authInfo - Instance of AuthInfo class with requester's info
@param {object} request - request object
@param {object | undefined } streamingV4Params - if v4 auth,
object containing accessKey, signatureFromRequest, region, scopeDate,
timestamp, and credentialScope
(to be used for streaming v4 auth if applicable)
@param {object} log - Werelogs logger
@param {function} cb - final callback to call with the result
@return {undefined}