From b309081d1acd30b4f9529d5b061ef500b478cb80 Mon Sep 17 00:00:00 2001 From: Dillard Blom Date: Thu, 8 Oct 2026 22:34:02 +0200 Subject: [PATCH] Repository: close the pack stream before any other request on the location Review notes on the open pack stream: it stayed open while a backup run went on to upload (e.g. the snapshot after reading the file list back), which SFTP cannot do on one connection. Reading moves to PackReader, and Repository reaches the location through PackReaderGuard, which closes the stream before every other request. A pack stream that cannot seek is read up to the offset. Deep verify no longer reports an \Error during decoding as damage. --- lib/Repository/PackReader.php | 97 ++++++++++++++++++++++++++ lib/Repository/PackReaderGuard.php | 60 ++++++++++++++++ lib/Repository/Repository.php | 80 ++------------------- tests/Unit/Repository/PackReadTest.php | 40 ++++++++++- 4 files changed, 203 insertions(+), 74 deletions(-) create mode 100644 lib/Repository/PackReader.php create mode 100644 lib/Repository/PackReaderGuard.php diff --git a/lib/Repository/PackReader.php b/lib/Repository/PackReader.php new file mode 100644 index 0000000..a7b5b87 --- /dev/null +++ b/lib/Repository/PackReader.php @@ -0,0 +1,97 @@ +stream !== null && $this->pack === $pack && $offset >= $this->pos && $offset - $this->pos <= self::SKIP_MAX) { + try { + $this->skipTo($offset); + return $this->readExactly($length); + } catch (\Exception) { + // The open stream broke (e.g. a timeout); open the pack again below. + } + } + $this->close(); + $fh = $this->backend->get('packs/' . substr($pack, 0, 2) . '/' . $pack); + $this->stream = $fh; + $this->pack = $pack; + $this->pos = 0; + try { + if ($offset > 0 && (stream_get_meta_data($fh)['seekable'] ?? false) && fseek($fh, $offset) === 0) { + $this->pos = $offset; + } + $this->skipTo($offset); // a stream that cannot seek is read up to the offset + return $this->readExactly($length); + } catch (\Exception $e) { + $this->close(); + throw $e; + } + } + + public function close(): void { + $fh = $this->stream; + $this->stream = null; + if ($fh !== null) { + @fclose($fh); + } + } + + public function __destruct() { + $this->close(); + } + + private function skipTo(int $offset): void { + while ($this->pos < $offset) { + $this->readExactly(min($offset - $this->pos, self::CHUNK)); + } + } + + /** Read $length bytes from the open stream; a short read closes it. */ + private function readExactly(int $length): string { + $fh = $this->stream ?? throw new RepositoryException('No open pack stream'); + $data = ''; + while (strlen($data) < $length) { + $chunk = fread($fh, min($length - strlen($data), self::CHUNK)); + if ($chunk === false || $chunk === '') { + $this->close(); + throw new RepositoryException('Short read in pack ' . $this->pack); + } + $data .= $chunk; + } + $this->pos += $length; + return $data; + } +} diff --git a/lib/Repository/PackReaderGuard.php b/lib/Repository/PackReaderGuard.php new file mode 100644 index 0000000..0004f4e --- /dev/null +++ b/lib/Repository/PackReaderGuard.php @@ -0,0 +1,60 @@ +reader->close(); + $this->inner->put($path, $stream); + } + + public function get(string $path) { + $this->reader->close(); + return $this->inner->get($path); + } + + public function getRange(string $path, int $offset, int $length): string { + $this->reader->close(); + return $this->inner->getRange($path, $offset, $length); + } + + public function exists(string $path): bool { + $this->reader->close(); + return $this->inner->exists($path); + } + + public function move(string $from, string $to): void { + $this->reader->close(); + $this->inner->move($from, $to); + } + + public function list(string $prefix): array { + $this->reader->close(); + return $this->inner->list($prefix); + } + + public function listFiles(string $dir): array { + $this->reader->close(); + return $this->inner->listFiles($dir); + } + + public function delete(string $path): void { + $this->reader->close(); + $this->inner->delete($path); + } +} diff --git a/lib/Repository/Repository.php b/lib/Repository/Repository.php index 8c57bc7..3c75cbf 100644 --- a/lib/Repository/Repository.php +++ b/lib/Repository/Repository.php @@ -30,18 +30,13 @@ final class Repository { private const FLAG_DEFLATE = 1; /** Reads per blob in a deep verify before it is reported as unreadable (not corrupted). */ private const VERIFY_READ_ATTEMPTS = 3; - /** A gap up to this size in an open pack stream is read past instead of opening the pack again. */ - private const SKIP_MAX = 8 * 1048576; private StreamCipher $cipher; private BlobIndex $index; private ?PackWriter $runPacks = null; private ?int $maxBytes = null; private Catalog $catalog; - /** @var resource|null open read stream of the pack read last, positioned at $packStreamPos */ - private $packStream = null; - private string $packStreamId = ''; - private int $packStreamPos = 0; + private PackReader $packReader; private function __construct( private IBackend $backend, @@ -49,7 +44,9 @@ final class Repository { private string $repositoryId, ) { $this->cipher = new StreamCipher($keys); - $this->index = new BlobIndex($backend, $this->cipher); + $this->packReader = new PackReader($backend); + $this->backend = new PackReaderGuard($backend, $this->packReader); + $this->index = new BlobIndex($this->backend, $this->cipher); } public static function init(IBackend $backend, #[\SensitiveParameter] string $passphrase): self { @@ -714,70 +711,7 @@ final class Repository { private function loadBlob(string $id): string { [$pack, $offset, $length] = $this->index->get($id); - return $this->decodeBlob($id, $this->readPack($pack, $offset, $length)); - } - - /** - * Bytes of a pack. Restores and deep verifies read the blobs of a snapshot in the order they - * were written, so the stream of the pack read last is kept open and read on from where it - * is: one request per pack instead of one per blob (a round trip each, and on most remote - * storages a new connection or file handle too). Anything else opens the pack at the offset, - * as a single read always did. - */ - private function readPack(string $pack, int $offset, int $length): string { - if ($this->packStream !== null && $this->packStreamId === $pack && $offset >= $this->packStreamPos && $offset - $this->packStreamPos <= self::SKIP_MAX) { - try { - while ($this->packStreamPos < $offset) { - $this->readExactly(min($offset - $this->packStreamPos, 1048576), $pack); - } - return $this->readExactly($length, $pack); - } catch (\Exception) { - // The open stream broke (e.g. a timeout); open the pack again below. - } - } - $this->closePackStream(); - $fh = $this->backend->get('packs/' . substr($pack, 0, 2) . '/' . $pack); - $this->packStream = $fh; - $this->packStreamId = $pack; - $this->packStreamPos = 0; - try { - if ($offset > 0 && fseek($fh, $offset) !== 0) { - throw new RepositoryException('Seek failed in pack ' . $pack); - } - $this->packStreamPos = $offset; - return $this->readExactly($length, $pack); - } catch (\Exception $e) { - $this->closePackStream(); - throw $e; - } - } - - /** Read $length bytes from the open pack stream; a short read closes it. */ - private function readExactly(int $length, string $pack): string { - $fh = $this->packStream ?? throw new RepositoryException('No open pack stream'); - $data = ''; - while (strlen($data) < $length) { - $chunk = fread($fh, min($length - strlen($data), 1048576)); - if ($chunk === false || $chunk === '') { - $this->closePackStream(); - throw new RepositoryException('Short read in pack ' . $pack); - } - $data .= $chunk; - } - $this->packStreamPos += $length; - return $data; - } - - private function closePackStream(): void { - $fh = $this->packStream; - $this->packStream = null; - if ($fh !== null) { - @fclose($fh); - } - } - - public function __destruct() { - $this->closePackStream(); + return $this->decodeBlob($id, $this->packReader->read($pack, $offset, $length)); } /** Decrypts and checks a blob read from its pack; throws when it is corrupted or tampered with. */ @@ -917,7 +851,7 @@ final class Repository { $encrypted = null; for ($attempt = 1; $attempt <= self::VERIFY_READ_ATTEMPTS && $encrypted === null; $attempt++) { try { - $encrypted = $this->readPack($pack, $offset, $length); + $encrypted = $this->packReader->read($pack, $offset, $length); } catch (\Exception $e) { // storage adapters throw their own exceptions; \Error (a bug) is not retried $stats['readError'] = $e->getMessage(); } @@ -928,7 +862,7 @@ final class Repository { } try { $stats['bytesChecked'] += strlen($this->decodeBlob($id, $encrypted)); - } catch (\Throwable) { + } catch (\Exception) { // damage; an \Error is a bug and must not be reported as damage $stats['failed'][] = $id; } } diff --git a/tests/Unit/Repository/PackReadTest.php b/tests/Unit/Repository/PackReadTest.php index 936b573..70a1f30 100644 --- a/tests/Unit/Repository/PackReadTest.php +++ b/tests/Unit/Repository/PackReadTest.php @@ -69,6 +69,26 @@ class PackReadTest extends TestCase { $this->assertSame(2, $backend->packOpens()); } + public function testThePackStreamIsClosedBeforeAnyOtherRequest(): void { + $backend = new PackOpenCountingBackend(new LocalBackend($this->root)); + $repo = $this->open($backend); + $this->assertSame(1, $repo->restore($this->snapshot, $this->tempDir(), 'f010')); + $this->assertTrue($backend->packStreamOpen(), 'kept open for the next read'); + + $repo->putObject('objects/after-restore', 'data'); + + $this->assertSame([false], $backend->packStreamOpenAtPut); + } + + public function testAPackStreamThatCannotSeekIsReadUpToTheOffset(): void { + $backend = new PackOpenCountingBackend(new LocalBackend($this->root), $this->root); + $target = $this->tempDir(); + + $this->assertSame(1, $this->open($backend)->restore($this->snapshot, $target, 'f150')); + + $this->assertSame(file_get_contents("$this->src/f150"), file_get_contents("$target/f150")); + } + public function testDeepVerifyOfATruncatedPackStillChecksTheBlobsBeforeTheDamage(): void { $packs = (new LocalBackend($this->root))->list('packs'); $this->assertCount(1, $packs); @@ -88,13 +108,24 @@ class PackReadTest extends TestCase { final class PackOpenCountingBackend implements IBackend { /** @var list */ public array $gets = []; + /** @var list */ + public array $packStreamOpenAtPut = []; + /** @var resource|null */ + private $lastPackStream = null; + /** @param ?string $pipeFrom serve packs as pipes (streams that cannot seek) from this local repository root */ public function __construct( private IBackend $inner, + private ?string $pipeFrom = null, ) { } + public function packStreamOpen(): bool { + return is_resource($this->lastPackStream); + } + public function put(string $path, $stream): void { + $this->packStreamOpenAtPut[] = $this->packStreamOpen(); $this->inner->put($path, $stream); } @@ -104,7 +135,14 @@ final class PackOpenCountingBackend implements IBackend { public function get(string $path) { $this->gets[] = $path; - return $this->inner->get($path); + if (!str_starts_with($path, 'packs/')) { + return $this->inner->get($path); + } + $fh = $this->pipeFrom !== null + ? popen('cat ' . escapeshellarg($this->pipeFrom . '/' . $path), 'r') + : $this->inner->get($path); + $this->lastPackStream = $fh; + return $fh; } public function getRange(string $path, int $offset, int $length): string {