Refactor writeObject to only use MultipartUpload when required

Signed-off-by: Bernd Rederlechner <Bernd.Rederlechner@t-systems.com>

Co-authored-by: Julius Härtl <jus@bitgrid.net>

2 files changed, 61 insertions, 20 deletions

diff --git a/lib/private/Files/ObjectStore/S3ConnectionTrait.php b/lib/private/Files/ObjectStore/S3ConnectionTrait.php
index 47c20d96d53..c99ebdbcd5c 100644
--- a/lib/private/Files/ObjectStore/S3ConnectionTrait.php
+++ b/lib/private/Files/ObjectStore/S3ConnectionTrait.php
@@ -74,9 +74,9 @@ trait S3ConnectionTrait {
 
 		$this->test = isset($params['test']);
 		$this->bucket = $params['bucket'];
-		$this->proxy = isset($params['proxy']) ? $params['proxy'] : false;
-		$this->timeout = !isset($params['timeout']) ? 15 : $params['timeout'];
-		$this->uploadPartSize = !isset($params['uploadPartSize']) ? 524288000 : $params['uploadPartSize'];
+		$this->proxy = $params['proxy'] ?? false;
+		$this->timeout = $params['timeout'] ?? 15;
+		$this->uploadPartSize = $params['uploadPartSize'] ?? 524288000;
 		$params['region'] = empty($params['region']) ? 'eu-west-1' : $params['region'];
 		$params['hostname'] = empty($params['hostname']) ? 's3.' . $params['region'] . '.amazonaws.com' : $params['hostname'];
 		if (!isset($params['port']) || $params['port'] === '') {
diff --git a/lib/private/Files/ObjectStore/S3ObjectTrait.php b/lib/private/Files/ObjectStore/S3ObjectTrait.php
index bb71306c17d..c88246094ed 100644
--- a/lib/private/Files/ObjectStore/S3ObjectTrait.php
+++ b/lib/private/Files/ObjectStore/S3ObjectTrait.php
@@ -28,10 +28,11 @@ namespace OC\Files\ObjectStore;
 
 use Aws\S3\Exception\S3MultipartUploadException;
 use Aws\S3\MultipartUploader;
-use Aws\S3\ObjectUploader;
 use Aws\S3\S3Client;
-use Icewind\Streams\CallbackWrapper;
+use GuzzleHttp\Psr7\Utils;
 use OC\Files\Stream\SeekableHttpStream;
+use GuzzleHttp\Psr7;
+use Psr\Http\Message\StreamInterface;
 
 trait S3ObjectTrait {
 	/**
@@ -80,37 +81,77 @@ trait S3ObjectTrait {
 	}
 
 	/**
+	 * Single object put helper
+	 *
 	 * @param string $urn the unified resource name used to identify the object
-	 * @param resource $stream stream with the data to write
+	 * @param StreamInterface $stream stream with the data to write
 	 * @param string|null $mimetype the mimetype to set for the remove object @since 22.0.0
 	 * @throws \Exception when something goes wrong, message will be logged
-	 * @since 7.0.0
 	 */
-	public function writeObject($urn, $stream, string $mimetype = null) {
-		$count = 0;
-		$countStream = CallbackWrapper::wrap($stream, function ($read) use (&$count) {
-			$count += $read;
-		});
+	protected function writeSingle(string $urn, StreamInterface $stream, string $mimetype = null): void {
+		$this->getConnection()->putObject([
+			'Bucket' => $this->bucket,
+			'Key' => $urn,
+			'Body' => $stream,
+			'ACL' => 'private',
+			'ContentType' => $mimetype,
+		]);
+	}
 
-		$uploader = new MultipartUploader($this->getConnection(), $countStream, [
+
+	/**
+	 * Multipart upload helper that tries to avoid orphaned fragments in S3
+	 *
+	 * @param string $urn the unified resource name used to identify the object
+	 * @param StreamInterface $stream stream with the data to write
+	 * @param string|null $mimetype the mimetype to set for the remove object
+	 * @throws \Exception when something goes wrong, message will be logged
+	 */
+	protected function writeMultiPart(string $urn, StreamInterface $stream, string $mimetype = null): void {
+		$uploader = new MultipartUploader($this->getConnection(), $stream, [
 			'bucket' => $this->bucket,
 			'key' => $urn,
 			'part_size' => $this->uploadPartSize,
 			'params' => [
 				'ContentType' => $mimetype
-			]
+			],
 		]);
 
 		try {
 			$uploader->upload();
 		} catch (S3MultipartUploadException $e) {
-			// This is an empty file so just touch it then
-			if ($count === 0 && feof($countStream)) {
-				$uploader = new ObjectUploader($this->getConnection(), $this->bucket, $urn, '');
-				$uploader->upload();
-			} else {
-				throw $e;
+			// if anything goes wrong with multipart, make sure that you don´t poison and
+			// slow down s3 bucket with orphaned fragments
+			$uploadInfo = $e->getState()->getId();
+			if ($e->getState()->isInitiated() && (array_key_exists('UploadId', $uploadInfo))) {
+				$this->getConnection()->abortMultipartUpload($uploadInfo);
 			}
+			throw $e;
+		}
+	}
+
+
+	/**
+	 * @param string $urn the unified resource name used to identify the object
+	 * @param resource $stream stream with the data to write
+	 * @param string|null $mimetype the mimetype to set for the remove object @since 22.0.0
+	 * @throws \Exception when something goes wrong, message will be logged
+	 * @since 7.0.0
+	 */
+	public function writeObject($urn, $stream, string $mimetype = null) {
+		$psrStream = Utils::streamFor($stream);
+
+		// ($psrStream->isSeekable() && $psrStream->getSize() !== null) evaluates to true for a On-Seekable stream
+		// so the optimisation does not apply
+		$buffer = new Psr7\Stream(fopen("php://memory", 'rwb+'));
+		Utils::copyToStream($psrStream, $buffer, MultipartUploader::PART_MIN_SIZE);
+		$buffer->seek(0);
+		if ($buffer->getSize() < MultipartUploader::PART_MIN_SIZE) {
+			// buffer is fully seekable, so use it directly for the small upload
+			$this->writeSingle($urn, $buffer, $mimetype);
+		} else {
+			$loadStream = new Psr7\AppendStream([$buffer, $psrStream]);
+			$this->writeMultiPart($urn, $loadStream, $mimetype);
 		}
 	}