feat: handle parallel chunk upload assembly race in Local and S3 devices

Author: TorstenDittmann

.gitignore

@@ -3,4 +3,5 @@
 .phpunit.result.cache
 tests/chunk.php
 .idea/
-.env
+.env
+.DS_Store

src/Storage/Device/Local.php

@@ -164,6 +164,10 @@ private function countChunks(string $tmp, string $path): int
 
     private function joinChunks(string $path, int $chunks): void
     {
+        if (\file_exists($path)) {
+            return;
+        }
+
         $tmp = \dirname($path).DIRECTORY_SEPARATOR.'tmp_'.\basename($path);
         $tmpAssemble = \dirname($path).DIRECTORY_SEPARATOR.'tmp_assemble_'.\basename($path);
 
@@ -195,6 +199,11 @@ private function joinChunks(string $path, int $chunks): void
         \fclose($dest);
 
         if (! \rename($tmpAssemble, $path)) {
+            if (\file_exists($path)) {
+                \unlink($tmpAssemble);
+
+                return;
+            }
             \unlink($tmpAssemble);
             throw new Exception('Failed to finalize assembled file '.$path);
         }

src/Storage/Device/S3.php

@@ -203,6 +203,9 @@ public function uploadData(string $data, string $path, string $contentType, int
         }
         $metadata['parts'][$chunk] = $etag;
         if ($metadata['chunks'] == $chunks) {
+            if ($this->exists($path)) {
+                return $metadata['chunks'];
+            }
             $this->completeMultipartUpload($path, $uploadId, $metadata['parts']);
         }
 

tests/Storage/Device/LocalTest.php

@@ -577,4 +577,37 @@ public function testOutOfOrderUploadWithRetry(): void
 
         $storage->delete($storage->getRoot(), true);
     }
+
+    public function testParallelChunkUpload(): void
+    {
+        $storage = $this->makeJoinTestStorage();
+        $dest = $storage->getRoot().DIRECTORY_SEPARATOR.'parallel.dat';
+
+        // Upload chunk 1 (creates temp directory)
+        $storage->uploadData('AAAA', $dest, 'application/octet-stream', 1, 2);
+
+        // Upload chunk 2 (assembles the file)
+        $storage->uploadData('BBBB', $dest, 'application/octet-stream', 2, 2);
+
+        // Verify file exists and is correct
+        $this->assertTrue(\file_exists($dest));
+        $this->assertSame('AAAABBBB', \file_get_contents($dest));
+
+        // Simulate the race where another request already assembled the file
+        // by calling joinChunks directly when the file already exists
+        $reflection = new \ReflectionClass($storage);
+        $method = $reflection->getMethod('joinChunks');
+        $method->setAccessible(true);
+
+        try {
+            $method->invoke($storage, $dest, 2);
+        } catch (\Exception $e) {
+            $this->fail('Duplicate assembly should not throw: '.$e->getMessage());
+        }
+
+        $this->assertTrue(\file_exists($dest), 'File should still exist after duplicate assembly attempt');
+        $this->assertSame('AAAABBBB', \file_get_contents($dest), 'File content must not be corrupted');
+
+        $storage->delete($storage->getRoot(), true);
+    }
 }