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 {