gsadmin/projectsend
1
0
Fork 0
mirror of https://github.com/projectsend/projectsend.git synced 2026-09-19 18:15:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1724 from fix/assemble-keeps-parts-for-retry

Browse Source 
Keep an upload's parts until its bytes are stored
This commit is contained in:
ignacionelson
2026-09-07 19:24:06 -03:00
parent 02946abf85 f2b705beee
commit b758fca19c
2 changed files with 194 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/Modules/Files/Uploads/LocalPartStore.php
+139 -73
View File
@@ -27,6 +27,12 @@ use Throwable;
*/
class LocalPartStore
{
/**
* Said twice, because a full temp volume can announce itself in the
* middle of the copy or only when the last buffer is flushed.
*/
private const WRITE_FAILED = 'Could not assemble the upload: writing to the temporary directory failed.';
/**
* The route name is a parameter because the same flow is mounted twice:
* once on the session-authenticated web routes for the browser, once on
@@ -140,9 +146,21 @@ class LocalPartStore
}
/**
* Stream-append parts in order onto the files disk, hashing as we
* go. Peak temp usage ≈ file size + one part (parts are unlinked
* as they are consumed).
* Stream-append parts in order onto the files disk, hashing as we go.
*
* The parts stay on disk until the assembled bytes are safely on the
* target disk. ChunkedUploadsController's completion lock promises that
* "a later retry still works", and everything that can fail after the
* concatenation — reopening the copy, a disk refusing the write, the
* File row itself — happens while the client has nothing but this
* session to retry with. Unlinking each part as it was consumed left
* listParts() empty, so every later complete() answered "Upload is
* incomplete: missing parts" for good.
*
* The cost is temp space: peak usage is the whole file twice over
* (every part, plus the assembled copy) rather than the file plus one
* part. Both are freed by the abort() below the moment the write lands,
* and by the failure path the moment it does not.
*
* @return array{path: string, disk: string, size: int, checksum: string}
*/
@@ -158,6 +176,93 @@ class LocalPartStore
}
$assembledPath = $this->directory($session).'/assembled';
try {
[$size, $checksum] = $this->concatenate($session, $parts, $assembledPath);
$readStream = fopen($assembledPath, 'rb');
if ($readStream === false) {
throw new RuntimeException('Could not reopen assembled file.');
}
$diskEvent = new ResolvingUploadDisk($session->user);
Event::dispatch($diskEvent);
$disk = $diskEvent->disk;
$written = Storage::disk($disk)->writeStream($targetPath, $readStream);
if (is_resource($readStream)) {
fclose($readStream);
}
// The disks are configured with 'throw' => false, so a refused
// write is a `false` return rather than an exception — and the
// caller goes on to record a File row for bytes that were never
// stored. Losing an upload silently is worse than failing it, and
// this is the only place that can tell the difference: a real
// instance of it was a GCS bucket rejecting the adapter's ACL,
// which looked exactly like a successful upload.
if ($written === false) {
// The reason is lost by the time it gets here — 'throw' => false
// means Flysystem swallowed the exception rather than passing it
// on — so log what was attempted. Which bucket it was is the
// difference between reading this as "my credentials expired"
// and "I typed the wrong bucket name", and only the log can say
// it: the message below is shown to whoever was uploading, which
// includes clients, and a bucket name is not theirs to see.
Log::error('Upload could not be written to storage.', [
'disk' => $disk,
'bucket' => config('filesystems.disks.'.$disk.'.bucket'),
'driver' => config('filesystems.disks.'.$disk.'.driver'),
'path' => $targetPath,
]);
throw new RuntimeException(
'Could not write the assembled upload to the "'.$disk.'" disk. '
.'Check the storage backend is reachable and its credentials are still valid.'
);
}
} catch (Throwable $failure) {
// The half-written copy belongs to this attempt and the next one
// makes its own; the parts belong to the client, and they are
// what a retry needs. Deleting the copy here is also the only
// thing that removes it at all on this path — it used to sit in
// the session directory until the sweeper came round.
FileSystem::delete($assembledPath);
throw $failure;
}
$this->abort($session);
return [
'path' => $targetPath,
'disk' => $disk,
'size' => $size,
'checksum' => $checksum,
];
}
/**
* Concatenate the parts into $assembledPath, returning the byte count
* and the sha256 of what was written.
*
* Every read and every write is checked. They were not, and while a
* failing fwrite on a full volume is loud in practice — Laravel's
* error handler turns the warning into an ErrorException — loud there
* means a 500 carrying a PHP message, where the disk-refused-the-write
* case a few lines above becomes a sentence the person uploading can
* act on. A short write arriving without a warning would be worse
* still: $size and the hash describe the buffer that was read, so an
* unchecked one yields a truncated file with a checksum matching bytes
* that were never stored.
*
* @param list<array{PartNumber: int, Size: int, ETag: string}> $parts
* @return array{0: int, 1: string}
*/
private function concatenate(UploadSession $session, array $parts, string $assembledPath): array
{
$out = fopen($assembledPath, 'wb');
if ($out === false) {
@@ -167,85 +272,46 @@ class LocalPartStore
$hash = hash_init('sha256');
$size = 0;
foreach ($parts as $part) {
$partPath = $this->partPath($session, $part['PartNumber']);
$in = fopen($partPath, 'rb');
try {
foreach ($parts as $part) {
$in = fopen($this->partPath($session, $part['PartNumber']), 'rb');
if ($in === false) {
fclose($out);
throw new RuntimeException('Could not read part '.$part['PartNumber'].'.');
}
while (! feof($in)) {
$buffer = fread($in, 1024 * 1024);
if ($buffer === false) {
break;
if ($in === false) {
throw new RuntimeException('Could not read part '.$part['PartNumber'].'.');
}
fwrite($out, $buffer);
hash_update($hash, $buffer);
$size += strlen($buffer);
try {
while (! feof($in)) {
$buffer = fread($in, 1024 * 1024);
if ($buffer === false) {
throw new RuntimeException('Could not read part '.$part['PartNumber'].'.');
}
if ($buffer !== '' && @fwrite($out, $buffer) !== strlen($buffer)) {
throw new RuntimeException(self::WRITE_FAILED);
}
hash_update($hash, $buffer);
$size += strlen($buffer);
}
} finally {
fclose($in);
}
}
} catch (Throwable $failure) {
fclose($out);
fclose($in);
unlink($partPath);
throw $failure;
}
fclose($out);
$readStream = fopen($assembledPath, 'rb');
if ($readStream === false) {
throw new RuntimeException('Could not reopen assembled file.');
// fclose flushes, so a volume that filled up on the last buffer
// fails here rather than in the loop.
if (! fclose($out)) {
throw new RuntimeException(self::WRITE_FAILED);
}
$diskEvent = new ResolvingUploadDisk($session->user);
Event::dispatch($diskEvent);
$disk = $diskEvent->disk;
$written = Storage::disk($disk)->writeStream($targetPath, $readStream);
if (is_resource($readStream)) {
fclose($readStream);
}
// The disks are configured with 'throw' => false, so a refused
// write is a `false` return rather than an exception — and the
// caller goes on to record a File row for bytes that were never
// stored. Losing an upload silently is worse than failing it, and
// this is the only place that can tell the difference: a real
// instance of it was a GCS bucket rejecting the adapter's ACL,
// which looked exactly like a successful upload.
if ($written === false) {
// The reason is lost by the time it gets here — 'throw' => false
// means Flysystem swallowed the exception rather than passing it
// on — so log what was attempted. Which bucket it was is the
// difference between reading this as "my credentials expired"
// and "I typed the wrong bucket name", and only the log can say
// it: the message below is shown to whoever was uploading, which
// includes clients, and a bucket name is not theirs to see.
Log::error('Upload could not be written to storage.', [
'disk' => $disk,
'bucket' => config('filesystems.disks.'.$disk.'.bucket'),
'driver' => config('filesystems.disks.'.$disk.'.driver'),
'path' => $targetPath,
]);
throw new RuntimeException(
'Could not write the assembled upload to the "'.$disk.'" disk. '
.'Check the storage backend is reachable and its credentials are still valid.'
);
}
$this->abort($session);
return [
'path' => $targetPath,
'disk' => $disk,
'size' => $size,
'checksum' => hash_final($hash),
];
return [$size, hash_final($hash)];
}
public function abort(UploadSession $session): void
tests/Feature/Files/ChunkedUploadsTest.php
+55
View File
@@ -329,6 +329,61 @@ test('a storage backend that refuses the write fails the upload instead of recor
expect(File::query()->count())->toBe($before);
});
// What survives a completion that failed. The lock in complete() promises
// the client may try again once whatever went wrong is fixed -- "the
// lock's TTL releases the claim if a completion dies mid-flight, so a
// later retry still works" -- and a retry has nothing to work from but
// the parts.
test('a refused write leaves the parts for the retry the lock promises', function () {
$this->actingAs($this->admin);
$sessionId = createSession(11, 'assembled.txt');
putPart($sessionId, 1, 'hello-');
putPart($sessionId, 2, 'world');
$refusing = Mockery::mock(Illuminate\Contracts\Filesystem\Filesystem::class);
$refusing->shouldReceive('writeStream')->once()->andReturnFalse();
Storage::set('files', $refusing);
$this->postJson("/uploads/{$sessionId}/complete")->assertStatus(422);
// Both parts are still there, and the half-written copy is not.
expect($this->getJson("/uploads/{$sessionId}/parts")->json())->toHaveCount(2)
->and(file_exists(partsRoot().'/'.$sessionId.'/assembled'))->toBeFalse();
// The operator fixes the bucket; the same session completes.
Storage::fake('files');
$this->postJson("/uploads/{$sessionId}/complete")->assertOk();
$file = File::query()->latest('id')->firstOrFail();
expect(Storage::disk('files')->get($file->path))->toBe('hello-world')
->and($file->size)->toBe(11);
});
test('a temporary directory that cannot be written fails the upload, and says so', function () {
// /dev/full accepts an open and refuses every write with ENOSPC, which
// is the failure this guards against without needing a full volume.
// Unchecked, the byte count and the checksum describe what was read
// rather than what landed; checked, it reads like the other storage
// failure a few lines below it in the same method.
$this->actingAs($this->admin);
$sessionId = createSession(11, 'assembled.txt');
putPart($sessionId, 1, 'hello-');
putPart($sessionId, 2, 'world');
symlink('/dev/full', partsRoot().'/'.$sessionId.'/assembled');
$this->postJson("/uploads/{$sessionId}/complete")
->assertStatus(422)
->assertJsonPath('errors.parts.0', fn (string $message): bool => str_contains($message, 'Could not assemble the upload'));
// Nothing recorded, and the parts are still the client's to retry with.
expect(File::query()->count())->toBe(0)
->and($this->getJson("/uploads/{$sessionId}/parts")->json())->toHaveCount(2);
})->skip(! file_exists('/dev/full'), 'needs /dev/full, which only exists on Linux');
// A chunked upload is two requests, and store()'s rule only ever sees the
// first one. Delete the folder while the bytes are in flight and the
// session still names it -- the version of this that nobody can ask for