Split chunked upload phases

src/Storage/Device.php

@@ -97,6 +97,33 @@ abstract public function getPath(string $filename, ?string $prefix = null): stri
      */
     abstract public function upload(string $source, string $path, int $chunk = 1, int $chunks = 1, array &$metadata = []): int;
 
+    /**
+     * Prepare Upload.
+     *
+     * Initialize adapter-specific upload state without transferring a chunk body.
+     *
+     * @throws Exception
+     */
+    abstract public function prepareUpload(string $path, string $contentType, int $chunks = 1, array &$metadata = []): void;
+
+    /**
+     * Upload Chunk.
+     *
+     * Upload exactly one chunk without finalizing the full upload.
+     *
+     * @throws Exception
+     */
+    abstract public function uploadChunk(string $source, string $path, int $chunk = 1, int $chunks = 1, array &$metadata = []): int;
+
+    /**
+     * Finalize Upload.
+     *
+     * Complete a prepared upload once all chunks are known to be present.
+     *
+     * @throws Exception
+     */
+    abstract public function finalizeUpload(string $path, int $chunks = 1, array &$metadata = []): bool;
+
     /**
      * Upload Data.
      *

src/Storage/Device/Local.php

@@ -55,16 +55,39 @@ public function getPath(string $filename, ?string $prefix = null): string
      * @throws Exception
      */
     public function upload(string $source, string $path, int $chunk = 1, int $chunks = 1, array &$metadata = []): int
+    {
+        $this->prepareUpload($path, '', $chunks, $metadata);
+        $chunksReceived = $this->uploadChunk($source, $path, $chunk, $chunks, $metadata);
+
+        if ($chunks > 1 && $chunks === $chunksReceived && ! $this->finalizeUpload($path, $chunks, $metadata)) {
+            throw new Exception('Failed to finalize upload '.$path);
+        }
+
+        return $chunksReceived;
+    }
+
+    public function prepareUpload(string $path, string $contentType, int $chunks = 1, array &$metadata = []): void
     {
         $this->createDirectory(\dirname($path));
+        $metadata['parts'] ??= [];
+        $metadata['chunks'] ??= 0;
+    }
+
+    public function uploadChunk(string $source, string $path, int $chunk = 1, int $chunks = 1, array &$metadata = []): int
+    {
+        $this->createDirectory(\dirname($path));
+        $metadata['parts'] ??= [];
+        $metadata['chunks'] ??= 0;
 
-        // move_uploaded_file() verifies the file is not tampered with
         if ($chunks === 1) {
-            if (! \move_uploaded_file($source, $path)) {
+            if (! \move_uploaded_file($source, $path) && ! \rename($source, $path)) {
                 throw new Exception('Can\'t upload file '.$path);
             }
 
-            return $chunks;
+            $metadata['parts'][$chunk] = true;
+            $metadata['chunks'] = 1;
+
+            return 1;
         }
 
         $tmp = \dirname($path).DIRECTORY_SEPARATOR.'tmp_'.\basename($path);
@@ -84,14 +107,33 @@ public function upload(string $source, string $path, int $chunk = 1, int $chunks
         }
 
         $chunksReceived = $this->countChunks($tmp, $path);
+        $metadata['parts'][$chunk] = true;
+        $metadata['chunks'] = $chunksReceived;
 
-        if ($chunks === $chunksReceived) {
-            $this->joinChunks($path, $chunks);
+        return $chunksReceived;
+    }
 
-            return $chunksReceived;
+    public function finalizeUpload(string $path, int $chunks = 1, array &$metadata = []): bool
+    {
+        if (\file_exists($path)) {
+            return true;
         }
 
-        return $chunksReceived;
+        if ($chunks === 1) {
+            return false;
+        }
+
+        $tmp = \dirname($path).DIRECTORY_SEPARATOR.'tmp_'.\basename($path);
+        for ($i = 1; $i <= $chunks; $i++) {
+            $part = $tmp.DIRECTORY_SEPARATOR.\pathinfo($path, PATHINFO_FILENAME).'.part.'.$i;
+            if (! \file_exists($part)) {
+                throw new Exception('Missing chunk '.$i);
+            }
+        }
+
+        $this->joinChunks($path, $chunks);
+
+        return true;
     }
 
     /**
@@ -108,7 +150,7 @@ public function upload(string $source, string $path, int $chunk = 1, int $chunks
      */
     public function uploadData(string $data, string $path, string $contentType, int $chunk = 1, int $chunks = 1, array &$metadata = []): int
     {
-        $this->createDirectory(\dirname($path));
+        $this->prepareUpload($path, $contentType, $chunks, $metadata);
 
         if ($chunks === 1) {
             if (! \file_put_contents($path, $data)) {
@@ -131,11 +173,11 @@ public function uploadData(string $data, string $path, string $contentType, int
         }
 
         $chunksReceived = $this->countChunks($tmp, $path);
+        $metadata['parts'][$chunk] = true;
+        $metadata['chunks'] = $chunksReceived;
 
-        if ($chunks === $chunksReceived) {
-            $this->joinChunks($path, $chunks);
-
-            return $chunksReceived;
+        if ($chunks > 1 && $chunks === $chunksReceived && ! $this->finalizeUpload($path, $chunks, $metadata)) {
+            throw new Exception('Failed to finalize upload '.$path);
         }
 
         return $chunksReceived;

src/Storage/Device/S3.php

@@ -167,7 +167,64 @@ public static function setRetryDelay(int $delay): void
      */
     public function upload(string $source, string $path, int $chunk = 1, int $chunks = 1, array &$metadata = []): int
     {
-        return $this->uploadData(\file_get_contents($source), $path, \mime_content_type($source), $chunk, $chunks, $metadata);
+        $contentType = \mime_content_type($source) ?: '';
+        $this->prepareUpload($path, $contentType, $chunks, $metadata);
+        $chunksReceived = $this->uploadChunk($source, $path, $chunk, $chunks, $metadata);
+
+        if ($chunks > 1 && $chunks === $chunksReceived && ! $this->finalizeUpload($path, $chunks, $metadata)) {
+            throw new Exception('Failed to finalize upload '.$path);
+        }
+
+        return $chunksReceived;
+    }
+
+    public function prepareUpload(string $path, string $contentType, int $chunks = 1, array &$metadata = []): void
+    {
+        $metadata['parts'] ??= [];
+        $metadata['chunks'] ??= 0;
+        $metadata['content_type'] ??= $contentType;
+
+        if ($chunks === 1 || ! empty($metadata['uploadId'])) {
+            return;
+        }
+
+        $metadata['uploadId'] = $this->createMultipartUpload($path, $contentType);
+    }
+
+    public function uploadChunk(string $source, string $path, int $chunk = 1, int $chunks = 1, array &$metadata = []): int
+    {
+        $data = \file_get_contents($source);
+        if ($data === false) {
+            throw new Exception('Can\'t read file '.$source);
+        }
+
+        return $this->uploadChunkData($data, $path, $metadata['content_type'] ?? (\mime_content_type($source) ?: ''), $chunk, $chunks, $metadata);
+    }
+
+    public function finalizeUpload(string $path, int $chunks = 1, array &$metadata = []): bool
+    {
+        if ($this->exists($path)) {
+            return true;
+        }
+
+        if ($chunks === 1) {
+            return false;
+        }
+
+        if (empty($metadata['uploadId'])) {
+            throw new Exception('Missing multipart upload ID');
+        }
+
+        $metadata['parts'] ??= [];
+        for ($i = 1; $i <= $chunks; $i++) {
+            if (! array_key_exists($i, $metadata['parts'])) {
+                throw new Exception('Missing chunk '.$i);
+            }
+        }
+
+        $this->completeMultipartUpload($path, $metadata['uploadId'], $metadata['parts']);
+
+        return true;
     }
 
     /**
@@ -183,38 +240,40 @@ public function upload(string $source, string $path, int $chunk = 1, int $chunks
      * @throws Exception
      */
     public function uploadData(string $data, string $path, string $contentType, int $chunk = 1, int $chunks = 1, array &$metadata = []): int
+    {
+        $this->prepareUpload($path, $contentType, $chunks, $metadata);
+        $chunksReceived = $this->uploadChunkData($data, $path, $contentType, $chunk, $chunks, $metadata);
+
+        if ($chunks > 1 && $chunks === $chunksReceived && ! $this->finalizeUpload($path, $chunks, $metadata)) {
+            throw new Exception('Failed to finalize upload '.$path);
+        }
+
+        return $chunksReceived;
+    }
+
+    private function uploadChunkData(string $data, string $path, string $contentType, int $chunk = 1, int $chunks = 1, array &$metadata = []): int
     {
         if ($chunk == 1 && $chunks == 1) {
-            return $this->write($path, $data, $contentType);
+            $this->write($path, $data, $contentType);
+            $metadata['parts'][$chunk] = true;
+            $metadata['chunks'] = 1;
+
+            return 1;
         }
-        $uploadId = $metadata['uploadId'] ?? null;
-        if (empty($uploadId)) {
-            $uploadId = $this->createMultipartUpload($path, $contentType);
-            $metadata['uploadId'] = $uploadId;
+
+        if (empty($metadata['uploadId'])) {
+            throw new Exception('Missing multipart upload ID');
         }
 
         $metadata['parts'] ??= [];
         $metadata['chunks'] ??= 0;
 
-        $etag = $this->uploadPart($data, $path, $contentType, $chunk, $uploadId);
+        $etag = $this->uploadPart($data, $path, $contentType, $chunk, $metadata['uploadId']);
         // skip incrementing if the chunk was re-uploaded
         if (! array_key_exists($chunk, $metadata['parts'])) {
             $metadata['chunks']++;
         }
         $metadata['parts'][$chunk] = $etag;
-        if ($metadata['chunks'] == $chunks) {
-            $headers = $this->headers;
-            $amzHeaders = $this->amzHeaders;
-
-            if ($this->exists($path)) {
-                return $metadata['chunks'];
-            }
-
-            $this->headers = $headers;
-            $this->amzHeaders = $amzHeaders;
-
-            $this->completeMultipartUpload($path, $uploadId, $metadata['parts']);
-        }
 
         return $metadata['chunks'];
     }
@@ -307,14 +366,15 @@ protected function completeMultipartUpload(string $path, string $uploadId, array
     {
         $uri = $path !== '' ? '/'.\str_replace(['%2F', '%3F'], ['/', '?'], \rawurlencode($path)) : '/';
 
-        \ksort($parts);
+        \ksort($parts, SORT_NUMERIC);
 
         $body = '<CompleteMultipartUpload>';
         foreach ($parts as $key => $etag) {
             $body .= "<Part><ETag>{$etag}</ETag><PartNumber>{$key}</PartNumber></Part>";
         }
         $body .= '</CompleteMultipartUpload>';
 
+        $this->headers['content-type'] = 'application/xml';
         $this->amzHeaders['x-amz-content-sha256'] = \hash('sha256', $body);
         $this->headers['content-md5'] = \base64_encode(md5($body, true));
         $this->call('s3:completeMultipartUpload', self::METHOD_POST, $uri, $body, ['uploadId' => $uploadId]);

src/Storage/Device/Telemetry.php

@@ -64,6 +64,21 @@ public function upload(string $source, string $path, int $chunk = 1, int $chunks
         return $this->measure(__FUNCTION__, $source, $path, $chunk, $chunks, $metadata);
     }
 
+    public function prepareUpload(string $path, string $contentType, int $chunks = 1, array &$metadata = []): void
+    {
+        $this->measure(__FUNCTION__, $path, $contentType, $chunks, $metadata);
+    }
+
+    public function uploadChunk(string $source, string $path, int $chunk = 1, int $chunks = 1, array &$metadata = []): int
+    {
+        return $this->measure(__FUNCTION__, $source, $path, $chunk, $chunks, $metadata);
+    }
+
+    public function finalizeUpload(string $path, int $chunks = 1, array &$metadata = []): bool
+    {
+        return $this->measure(__FUNCTION__, $path, $chunks, $metadata);
+    }
+
     public function uploadData(string $data, string $path, string $contentType, int $chunk = 1, int $chunks = 1, array &$metadata = []): int
     {
         return $this->measure(__FUNCTION__, $data, $path, $contentType, $chunk, $chunks, $metadata);

tests/Storage/Device/LocalTest.php

@@ -190,6 +190,51 @@ public function testPartUpload()
         return $dest;
     }
 
+    public function testUploadChunkDoesNotFinalizeUntilFinalizeUpload(): void
+    {
+        $dest = $this->object->getPath('chunked-phase-upload.txt');
+        $metadata = [];
+        $parts = [
+            2 => 'bbb',
+            1 => 'aaa',
+            3 => 'ccc',
+        ];
+
+        foreach ($parts as $chunk => $data) {
+            $source = __DIR__.'/chunk-'.$chunk.'.part';
+            file_put_contents($source, $data);
+
+            $this->object->uploadChunk($source, $dest, $chunk, 3, $metadata);
+            $this->assertFalse($this->object->exists($dest));
+        }
+
+        $this->assertSame(3, $metadata['chunks']);
+        $this->assertTrue($this->object->finalizeUpload($dest, 3, $metadata));
+        $this->assertSame('aaabbbccc', $this->object->read($dest));
+        $this->assertTrue($this->object->finalizeUpload($dest, 3, $metadata));
+
+        $this->object->delete($dest);
+    }
+
+    public function testFinalizeUploadRequiresAllLocalChunks(): void
+    {
+        $dest = $this->object->getPath('chunked-phase-missing.txt');
+        $metadata = [];
+        $source = __DIR__.'/chunk-missing.part';
+        file_put_contents($source, 'aaa');
+
+        $this->object->uploadChunk($source, $dest, 1, 2, $metadata);
+
+        try {
+            $this->object->finalizeUpload($dest, 2, $metadata);
+            $this->fail('Expected missing chunk exception');
+        } catch (\Exception $e) {
+            $this->assertSame('Missing chunk 2', $e->getMessage());
+        } finally {
+            $this->object->abort($dest);
+        }
+    }
+
     public function testPartUploadRetry()
     {
         $source = __DIR__.'/../../resources/disk-a/large_file.mp4';