diff --git a/CHANGELOG.md b/CHANGELOG.md index cd58819..139f25a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,25 @@ the app store accepts a `-beta` suffix, but Nextcloud servers then only offer th beta/daily update channel, which would hide the app from most admins. The stability stage is instead carried by the app description, the git tag, the release, and this file. +## [0.9.4-beta.5] - 2026-10-09 + +Bugfix release: faster restores. + +### Fixed +- Restores and `occ backup:verify --deep` fetched every blob with its own request, so on a + remote location they ran far below the speed of the connection. The blobs of a pack are now + read from one open stream, in the order they were written. Restoring 86 MB in 202 files from a + Hetzner Storage Box over SMB went from 0.78 MB/s to 4.2 MB/s, the speed of reading the packs + as plain files over that connection. +- Reading from an SFTP location waited a full round trip for every 256 KiB, whatever the line + speed. Reads now keep up to 8 requests in flight. Restoring 1.97 GB in 500 files from a Hetzner + Storage Box (47 ms away) took 694 s before the fix above, 499 s with it and 142 s with both + (2.8 → 3.9 → 13.9 MB/s); a full 51 GB restore ran at 13.0 MB/s. When a server error + interrupts such a read, it goes on from the same position with Nextcloud's own SFTP stream; + after three failing files in a row, or on a Nextcloud whose SFTP library lacks what this needs, + only Nextcloud's own stream is used. +- README: a WebDAV location on a slow upload needs a higher `davstorage.request_timeout`. + ## [0.9.3-beta.4] - 2026-10-08 Bugfix release. @@ -135,6 +154,7 @@ and not implemented yet. - See the "Storage backend experiences" section of the README for provider-specific quirks found so far (e.g. Hetzner Storage Box sub-accounts). +[0.9.4-beta.5]: https://github.com/dillardblom/ng_backup/releases/tag/v0.9.4-beta.5 [0.9.3-beta.4]: https://github.com/dillardblom/ng_backup/releases/tag/v0.9.3-beta.4 [0.9.2-beta.3]: https://github.com/dillardblom/ng_backup/releases/tag/v0.9.2-beta.3 [0.9.1-beta.2]: https://github.com/dillardblom/ng_backup/releases/tag/v0.9.1-beta.2 diff --git a/README.md b/README.md index 7aa4343..c7bc7e7 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ installations and hosted Nextcloud. (Choosing SMB as a storage backend is the on see "Storage backend experiences" below — that requirement comes from files_external, not from ng_backup, and applies to any Nextcloud app that writes to an SMB share.) -**Status: beta (0.9.3).** Tested thoroughly, usable alongside your current backup, but not yet a +**Status: beta (0.9.4).** Tested thoroughly, usable alongside your current backup, but not yet a drop-in replacement for it. See "Known limitations" below, `PLAN.md` for the design and `docs/HOWTO.md` for a step-by-step guide from setup to a tested restore. @@ -136,9 +136,10 @@ Contributions welcome via PR — add what you ran into with your own provider. Conflict`. SMB: share = the sub-account name, port 445 must be reachable from your server (it was from a home connection here, but some providers block it), and the `smbclient` PECL extension is required (see the SMB entry below). Both have a noticeable fixed cost per run: an incremental run - that uploads nothing took about 20 s over WebDAV and 50 s over SMB. A deep verify is slow over - remote storage, since each blob is fetched with its own request (19 minutes for 66 MiB over - WebDAV); run it occasionally, not after every backup. + that uploads nothing took about 20 s over WebDAV and 50 s over SMB. Over WebDAV, Nextcloud gives + every request at most 30 s by default, also an upload of a whole 32 MiB pack: on an upload + slower than about 9 Mbit/s a backup fails with `cURL error 28: Operation timed out`. Raise the + limit with `occ config:system:set davstorage.request_timeout --value=600 --type=integer`. - **FTP, server-dependent absolute-path handling:** files_external's FTP backend always builds absolute paths (e.g. `MKD /repo`) for the configured root, never relative ones. `atmoz/sftp`-style diff --git a/appinfo/info.xml b/appinfo/info.xml index 3f4f45e..26e0f1d 100644 --- a/appinfo/info.xml +++ b/appinfo/info.xml @@ -9,7 +9,7 @@ directory to S3, SFTP, WebDAV, FTP, SMB or a local path, with restore of files a single users (via user_migration) and of the whole server onto a fresh installation. Tested thoroughly and usable alongside your current backup, but not yet a drop-in replacement for it. See the README for known limitations and the recovery kit before relying on it.]]> - 0.9.3 + 0.9.4 agpl Dillard Blom NgBackup diff --git a/lib/Backend/SftpReadStream.php b/lib/Backend/SftpReadStream.php new file mode 100644 index 0000000..c982991 --- /dev/null +++ b/lib/Backend/SftpReadStream.php @@ -0,0 +1,412 @@ +|null + */ + private static ?\WeakMap $chunks = null; + private static int $requestId = 0x4e470000; + /** Streams in a row that failed; at MAX_FAILURES only the storage's own stream is used. */ + private static int $failures = 0; + public const MAX_FAILURES = 3; + /** Set by stream_open() when the connection failed rather than the file. */ + private static bool $openBroke = false; + + /** @var resource|null set by PHP for stream wrappers */ + public $context; + + /** + * The phpseclib connection. PHP creates stream wrappers without constructor arguments; it is + * set in stream_open(), and no other method is called when that fails. + * + * @psalm-suppress PropertyNotSetInConstructor + */ + private object $sftp; + /** The storage and the path on it, to fall back to the storage's own stream. */ + private ?object $storage = null; + private string $file = ''; + /** @var resource|null the storage's own stream, once this one has failed */ + private $fallback = null; + private string $handle = ''; + /** Position of the reader: the first byte of $buffer. */ + private int $pos = 0; + /** Offset of the next READ to send. */ + private int $next = 0; + private string $buffer = ''; + /** @var list request id, offset, length, in the order sent */ + private array $inFlight = []; + private int $depth = 1; + private bool $eof = false; + /** This stream gave up on its own requests. */ + private bool $failed = false; + /** The connection is in an unknown state: nothing more is sent over it or read from it. */ + private bool $broken = false; + /** Length of the last short answer, until the next answer tells whether it was the end of the file. */ + private ?int $shortAnswer = null; + + /** + * @param object $storage files_external's SFTP storage (getConnection(), getRoot()) + * @return resource|false|null the stream, false when the file cannot be opened, null when the + * connection does not offer what this stream needs + * @throws BackendException when the connection failed after the OPEN request went out: the + * storage's own stream must not be tried on it, as it would take a + * late answer to that request for its own + */ + public static function open(object $storage, string $path) { + if (self::$failures >= self::MAX_FAILURES) { + return null; + } + try { + if (!is_callable([$storage, 'getConnection']) || !is_callable([$storage, 'getRoot'])) { + return null; + } + $sftp = $storage->getConnection(); + // is_callable and get_object_vars from here only see what is public: phpseclib 3 has + // methods of these names neither public nor at all. + if (!is_object($sftp) || !is_callable([$sftp, '_send_sftp_packet']) || !is_callable([$sftp, '_get_sftp_packet']) + || !is_callable([$sftp, '_realpath']) || !array_key_exists('packet_type', get_object_vars($sftp))) { + return null; + } + if (!in_array(self::PROTOCOL, stream_get_wrappers(), true)) { + stream_wrapper_register(self::PROTOCOL, self::class); + } + $root = rtrim((string)$storage->getRoot(), '/'); + $context = stream_context_create([self::PROTOCOL => [ + 'session' => $sftp, + 'path' => $root . '/' . ltrim($path, '/'), + 'storage' => $storage, + 'file' => $path, + ]]); + self::$openBroke = false; + $fh = @fopen(self::PROTOCOL . '://file', 'r', false, $context); + } catch (\Throwable) { + // nothing was sent yet + self::$failures++; + return null; + } + if ($fh === false && self::openBroke()) { + throw new BackendException('SFTP connection failed while opening ' . $path); + } + return $fh; + } + + /** Test hook: forget the learned READ sizes and earlier failures. */ + public static function reset(): void { + self::$chunks = null; + self::$failures = 0; + } + + public function stream_open(string $url, string $mode, int $options, ?string &$openedPath): bool { + $options = stream_context_get_options($this->context ?? stream_context_get_default())[self::PROTOCOL] ?? []; + if (!isset($options['session'], $options['path']) || !is_object($options['session'])) { + return false; + } + $this->sftp = $options['session']; + $storage = $options['storage'] ?? null; + $this->storage = is_object($storage) ? $storage : null; + $this->file = (string)($options['file'] ?? ''); + try { + $remote = $this->sftp->_realpath((string)$options['path']); + if (!is_string($remote)) { + return false; + } + $id = self::nextId(); + if (!$this->sftp->_send_sftp_packet(self::FXP_OPEN, pack('Na*N2', strlen($remote), $remote, self::FXF_READ, 0), $id)) { + return $this->connectionFailed(); + } + $response = $this->sftp->_get_sftp_packet($id); + } catch (\Throwable) { + return $this->connectionFailed(); + } + if (!is_string($response)) { + return $this->connectionFailed(); + } + if ($this->sftp->packet_type !== self::FXP_HANDLE) { + return false; // an answer, but no handle: the file cannot be opened + } + $this->handle = substr($response, 4); + return true; + } + + /** Not a missing file but a connection that does not behave as expected. */ + private function connectionFailed(): bool { + self::$failures++; + self::$openBroke = true; + return false; + } + + /** Whether the last stream_open() failed on the connection; a call, as fopen() sets it. */ + private static function openBroke(): bool { + return self::$openBroke; + } + + public function stream_read(int $count): string|false { + if ($this->fallback !== null) { + return fread($this->fallback, $count); + } + while (strlen($this->buffer) < $count && !$this->eof) { + if (!$this->receive()) { + return $this->useFallback() ? $this->stream_read($count) : false; + } + } + $data = substr($this->buffer, 0, $count); + $this->buffer = substr($this->buffer, strlen($data)); + $this->pos += strlen($data); + return $data; + } + + public function stream_eof(): bool { + if ($this->fallback !== null) { + return feof($this->fallback); + } + return $this->eof && $this->buffer === ''; + } + + public function stream_tell(): int { + if ($this->fallback !== null) { + $pos = ftell($this->fallback); + return $pos === false ? $this->pos : $pos; + } + return $this->pos; + } + + public function stream_seek(int $offset, int $whence = SEEK_SET): bool { + if ($this->fallback !== null) { + return fseek($this->fallback, $offset, $whence) === 0; + } + if ($whence === SEEK_CUR) { + $offset += $this->pos; + } elseif ($whence !== SEEK_SET) { + return false; // the size is never asked for, to save a round trip + } + if ($offset < 0) { + return false; + } + if ($this->failed) { + $this->pos = $offset; + return $this->useFallback(); + } + if ($offset >= $this->pos && $offset - $this->pos <= strlen($this->buffer)) { + $this->buffer = substr($this->buffer, $offset - $this->pos); + } else { + $this->drain(); + $this->buffer = ''; + $this->next = $offset; + $this->eof = false; + $this->depth = 1; + $this->shortAnswer = null; + } + $this->pos = $offset; + return true; + } + + /** @return array */ + public function stream_stat(): array { + return []; + } + + public function stream_close(): void { + if (!$this->failed) { + self::$failures = 0; + } + $fallback = $this->fallback; + $this->fallback = null; + if ($fallback !== null) { + fclose($fallback); + } + $this->closeHandle(); + } + + /** + * Go on with the storage's own stream from the current position, after a server error on this + * one. The data still buffered is dropped and read again. False after a connection error, or + * when that stream cannot be opened or positioned: then the read fails as it would have + * without this class. Reopening by path is safe for a repository: packs and index files are + * never rewritten, and every object is authenticated when it is decrypted. + */ + private function useFallback(): bool { + $this->closeHandle(); + $this->buffer = ''; + // files_external's stream takes whatever answer comes next on the connection: after a + // connection error, answers to this stream's requests may still be on their way. + if ($this->broken || $this->storage === null || !is_callable([$this->storage, 'fopen'])) { + return false; + } + try { + $fh = $this->storage->fopen($this->file, 'r'); + if (!is_resource($fh)) { + return false; + } + if ($this->pos > 0 && fseek($fh, $this->pos) !== 0) { + fclose($fh); + return false; + } + } catch (\Throwable) { + return false; + } + $this->fallback = $fh; + return true; + } + + private function closeHandle(): void { + $this->drain(); + if ($this->handle !== '' && !$this->broken) { + try { + $id = self::nextId(); + if (!$this->sftp->_send_sftp_packet(self::FXP_CLOSE, pack('Na*', strlen($this->handle), $this->handle), $id) + || !is_string($this->sftp->_get_sftp_packet($id))) { + $this->broken = true; + } + } catch (\Throwable) { + $this->broken = true; + } + } + $this->handle = ''; + } + + /** Keep the pipeline full, then take the oldest answer. False on a connection or server error. */ + private function receive(): bool { + if ($this->failed) { + return false; + } + try { + return $this->receiveOne(); + } catch (\Throwable) { + $this->broken = true; + return $this->fail(); + } + } + + private function receiveOne(): bool { + while (count($this->inFlight) < $this->depth) { + $id = self::nextId(); + $length = self::chunk($this->sftp); + $packet = pack('Na*N3', strlen($this->handle), $this->handle, intdiv($this->next, 4294967296), $this->next % 4294967296, $length); + if (!$this->sftp->_send_sftp_packet(self::FXP_READ, $packet, $id)) { + $this->broken = true; + return $this->fail(); + } + $this->inFlight[] = [$id, $this->next, $length]; + $this->next += $length; + } + [$id, $offset, $length] = array_shift($this->inFlight); + $response = $this->sftp->_get_sftp_packet($id); + if (!is_string($response)) { + $this->broken = true; + return $this->fail(); + } + if ($this->sftp->packet_type === self::FXP_DATA) { + $data = substr($response, 4); + if ($data === '') { + return $this->fail(); + } + $this->buffer .= $data; + $this->depth = min(self::DEPTH, $this->depth * 2); + if ($this->shortAnswer !== null) { + // More data after a short answer: that one was the server's cap, not the end. + // Never below what every server must support: a one-off small answer is no cap. + self::$chunks ??= new \WeakMap(); + self::$chunks[$this->sftp] = max(self::MIN_CHUNK, min(self::chunk($this->sftp), $this->shortAnswer)); + $this->shortAnswer = null; + } + if (strlen($data) < $length) { + // Fewer bytes than asked: the end of the file, or a server that caps a READ. The + // requests behind this one assumed a full answer, so drop them and ask again from here. + $this->drain(); + $this->next = $offset + strlen($data); + $this->shortAnswer = strlen($data); + } + return true; + } + $status = $this->sftp->packet_type === self::FXP_STATUS ? unpack('N', $response) : false; + if (is_array($status) && ($status[1] ?? null) === self::FX_EOF) { + $this->eof = true; + $this->shortAnswer = null; + $this->drain(); + return true; + } + return $this->fail(); + } + + private function fail(): bool { + $this->drain(); + if (!$this->failed) { + self::$failures++; + } + $this->failed = true; + return false; + } + + /** Read and drop the answers still on their way, so the connection is free again. */ + private function drain(): void { + $inFlight = $this->inFlight; + $this->inFlight = []; + if ($this->broken) { + return; + } + try { + foreach ($inFlight as [$id]) { + if (!is_string($this->sftp->_get_sftp_packet($id))) { + $this->broken = true; + return; + } + } + } catch (\Throwable) { + $this->broken = true; + } + } + + private static function chunk(object $sftp): int { + $learned = self::$chunks?->offsetExists($sftp) === true ? self::$chunks->offsetGet($sftp) : null; + return is_int($learned) ? $learned : self::CHUNK; + } + + private static function nextId(): int { + self::$requestId = self::$requestId >= 0x7fffffff ? 0x4e470000 : self::$requestId + 1; + return self::$requestId; + } +} diff --git a/lib/Backend/StorageBackend.php b/lib/Backend/StorageBackend.php index 58f53e2..95672f3 100644 --- a/lib/Backend/StorageBackend.php +++ b/lib/Backend/StorageBackend.php @@ -17,6 +17,8 @@ use OCP\Files\Storage\IWriteStreamStorage; final class StorageBackend implements IBackend { /** Object stores write whole objects atomically; other storages get a temp name + rename. */ private bool $atomicPut; + /** SFTP: read through SftpReadStream, which keeps several requests in flight. */ + private bool $sftpReads; public function __construct( private IStorage $storage, @@ -25,6 +27,7 @@ final class StorageBackend implements IBackend { $this->base = trim($base, '/'); self::validate($this->base, true); $this->atomicPut = $storage->instanceOfStorage(\OCA\Files_External\Lib\Storage\AmazonS3::class); + $this->sftpReads = $storage->instanceOfStorage(\OCA\Files_External\Lib\Storage\SFTP::class); if ($this->base !== '') { $this->mkdirs($this->base); } @@ -79,6 +82,14 @@ final class StorageBackend implements IBackend { } public function get(string $path) { + if ($this->sftpReads) { + $fh = SftpReadStream::open($this->storage, $this->abs($path)); + if (is_resource($fh)) { + return $fh; + } + // null: not supported or failing for now; false: not opened, which need not mean + // missing. Either way the storage's own stream decides. + } $fh = $this->storage->fopen($this->abs($path), 'r'); if ($fh === false) { throw new BackendException('Not found: ' . $path); 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 d6c4f1d..3c75cbf 100644 --- a/lib/Repository/Repository.php +++ b/lib/Repository/Repository.php @@ -36,6 +36,7 @@ final class Repository { private ?PackWriter $runPacks = null; private ?int $maxBytes = null; private Catalog $catalog; + private PackReader $packReader; private function __construct( private IBackend $backend, @@ -43,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 { @@ -708,7 +711,7 @@ final class Repository { private function loadBlob(string $id): string { [$pack, $offset, $length] = $this->index->get($id); - return $this->decodeBlob($id, $this->backend->getRange('packs/' . substr($pack, 0, 2) . '/' . $pack, $offset, $length)); + 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. */ @@ -848,7 +851,7 @@ final class Repository { $encrypted = null; for ($attempt = 1; $attempt <= self::VERIFY_READ_ATTEMPTS && $encrypted === null; $attempt++) { try { - $encrypted = $this->backend->getRange('packs/' . substr($pack, 0, 2) . '/' . $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(); } @@ -859,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/Backend/SftpReadStreamTest.php b/tests/Unit/Backend/SftpReadStreamTest.php new file mode 100644 index 0000000..abef338 --- /dev/null +++ b/tests/Unit/Backend/SftpReadStreamTest.php @@ -0,0 +1,478 @@ + $data]); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'repo/packs/ab/abcd'); + + $this->assertIsResource($fh); + $this->assertSame($data, stream_get_contents($fh)); + $this->assertTrue(feof($fh)); + fclose($fh); + $this->assertSame(SftpReadStream::DEPTH, $sftp->maxInFlight); + $this->assertSame(0, $sftp->pending(), 'nothing may be left on the connection after close'); + $this->assertSame([], $sftp->openHandles); + } + + public function testAServerThatCapsAReadStillGivesEveryByteInOrder(): void { + $data = random_bytes(3 * 1048576 + 7); + $sftp = new FakeSftpConnection(['/home/f' => $data], cap: 100000); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + + $this->assertSame($data, stream_get_contents($fh)); + fclose($fh); + $this->assertSame(0, $sftp->pending()); + + // The cap is learned: a next stream asks for no more than the server gives. + $sftp->reads = []; + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + $this->assertSame($data, stream_get_contents($fh)); + fclose($fh); + $this->assertSame([100000], array_values(array_unique($sftp->reads))); + + // Only for that connection. + $other = new FakeSftpConnection(['/home/f' => $data]); + $fh = SftpReadStream::open(new FakeSftpStorage($other), 'f'); + $this->assertSame($data, stream_get_contents($fh)); + fclose($fh); + $this->assertSame(261120, max($other->reads)); + } + + public function testATinyAnswerDoesNotShrinkTheReadsBelowWhatEveryServerSupports(): void { + $data = random_bytes(1048576); + $sftp = new FakeSftpConnection(['/home/f' => $data], cap: 1000); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + $this->assertSame($data, stream_get_contents($fh)); + fclose($fh); + $this->assertSame(32768, min($sftp->reads)); + } + + public function testSeekingReadsFromTheNewOffset(): void { + $data = random_bytes(3 * 1048576); + $sftp = new FakeSftpConnection(['/home/f' => $data]); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + + $this->assertSame(0, fseek($fh, 2000000)); + $this->assertSame(2000000, ftell($fh)); + $this->assertSame(substr($data, 2000000, 1000), fread($fh, 1000)); + $this->assertSame(0, fseek($fh, 10)); // backwards + $this->assertSame(substr($data, 10, 70000), $this->readExactly($fh, 70000)); + $this->assertSame(0, fseek($fh, 100, SEEK_CUR)); // within what is already buffered + $this->assertSame(substr($data, 70110, 5000), $this->readExactly($fh, 5000)); + fclose($fh); + $this->assertSame(0, $sftp->pending()); + } + + public function testASmallReadDoesNotFetchTheWholePipelineAhead(): void { + $sftp = new FakeSftpConnection(['/home/f' => random_bytes(10 * 1048576)]); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + fread($fh, 100); + fclose($fh); + $this->assertCount(1, $sftp->reads, 'the first request goes alone, the pipeline grows while reading on'); + } + + public function testAnEmptyFileIsAtEndAtOnce(): void { + $sftp = new FakeSftpConnection(['/home/f' => '']); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + $this->assertSame('', stream_get_contents($fh)); + $this->assertTrue(feof($fh)); + fclose($fh); + } + + public function testAMissingFileCannotBeOpened(): void { + $sftp = new FakeSftpConnection([]); + $this->assertFalse(SftpReadStream::open(new FakeSftpStorage($sftp), 'nope')); + $this->assertSame(0, $sftp->pending()); + } + + public function testAServerErrorGoesOnWithTheStoragesOwnStreamFromTheSamePosition(): void { + $data = random_bytes(3 * 1048576); + $sftp = new FakeSftpConnection(['/home/f' => $data], failAt: 1048576); + $storage = new FakeSftpStorage($sftp); + $fh = SftpReadStream::open($storage, 'f'); + + $this->assertSame($data, $this->readAll($fh)); + $this->assertSame(['f'], $storage->fopens); + $this->assertSame(strlen($data), ftell($fh)); + fclose($fh); + $this->assertSame(0, $sftp->pending(), 'the connection is clean for the storage\'s own stream'); + $this->assertSame([], $sftp->openHandles, 'the failed handle is closed'); + + // One failure is no reason to give up on the fast reads. + $fh = SftpReadStream::open($storage, 'f'); + $this->assertIsResource($fh); + fclose($fh); + } + + public function testOnlyFailuresInARowSwitchTheFastReadsOff(): void { + $data = random_bytes(2 * 1048576); + $failing = new FakeSftpStorage(new FakeSftpConnection(['/home/f' => $data], failAt: 1048576)); + $working = new FakeSftpStorage(new FakeSftpConnection(['/home/f' => $data])); + for ($i = 1; $i < SftpReadStream::MAX_FAILURES; $i++) { + $fh = SftpReadStream::open($failing, 'f'); + $this->assertSame($data, $this->readAll($fh)); + fclose($fh); + } + $fh = SftpReadStream::open($working, 'f'); // a stream that works resets the count + $this->assertSame($data, $this->readAll($fh)); + fclose($fh); + for ($i = 1; $i <= SftpReadStream::MAX_FAILURES; $i++) { + $fh = SftpReadStream::open($failing, 'f'); + $this->assertIsResource($fh, "failure $i"); + $this->assertSame($data, $this->readAll($fh)); + fclose($fh); + } + $this->assertNull(SftpReadStream::open($working, 'f')); + $this->assertSame(SftpReadStream::MAX_FAILURES * 2 - 1, count($failing->fopens), 'every failed stream still gave all its data'); + } + + public function testAServerErrorAfterASeekGoesOnFromTheSeekedPosition(): void { + $data = random_bytes(3 * 1048576); + $sftp = new FakeSftpConnection(['/home/f' => $data], failAt: 2500000); + $storage = new FakeSftpStorage($sftp); + $fh = SftpReadStream::open($storage, 'f'); + $this->assertSame(0, fseek($fh, 2400000)); + $this->assertSame(substr($data, 2400000), $this->readAll($fh)); + $this->assertSame(0, fseek($fh, 100)); // seeks go to the storage's stream now + $this->assertSame(substr($data, 100, 1000), $this->readExactly($fh, 1000)); + fclose($fh); + } + + public function testAConnectionErrorDoesNotFallBackOnTheSameConnection(): void { + // After an exception, answers may still be on their way; files_external's stream would + // take them for its own. + $sftp = new FakeSftpConnection(['/home/f' => random_bytes(3 * 1048576)], throwAt: 1048576); + $storage = new FakeSftpStorage($sftp); + $fh = SftpReadStream::open($storage, 'f'); + $this->assertLessThan(3 * 1048576, strlen($this->readAll($fh))); + fclose($fh); + $this->assertSame([], $storage->fopens); + } + + public function testALostConnectionDoesNotFallBackEither(): void { + $sftp = new FakeSftpConnection(['/home/f' => random_bytes(3 * 1048576)], loseAt: 1048576); + $storage = new FakeSftpStorage($sftp); + $fh = SftpReadStream::open($storage, 'f'); + $this->assertLessThan(3 * 1048576, strlen($this->readAll($fh))); + fclose($fh); + $this->assertSame([], $storage->fopens); + } + + public function testWhenTheStoragesOwnStreamFailsTooTheReadFails(): void { + $sftp = new FakeSftpConnection(['/home/f' => random_bytes(3 * 1048576)], failAt: 1048576); + $storage = new FakeSftpStorage($sftp, fopenFails: true); + $fh = SftpReadStream::open($storage, 'f'); + $this->assertLessThan(3 * 1048576, strlen($this->readAll($fh))); + fclose($fh); + $this->assertSame(['f'], $storage->fopens); + } + + public function testAConnectionErrorWhileOpeningIsAnErrorNotAMissingFile(): void { + // The OPEN request went out: the storage's own stream must not be tried on this connection. + $storage = new FakeSftpStorage(new FakeSftpConnection(['/home/f' => 'x'], throwAt: -1)); + try { + SftpReadStream::open($storage, 'f'); + $this->fail('expected a BackendException'); + } catch (BackendException $e) { + $this->assertStringContainsString('connection', $e->getMessage()); + } + $lost = new FakeSftpStorage(new FakeSftpConnection(['/home/f' => 'x'], loseAt: -1)); + $this->expectException(BackendException::class); + SftpReadStream::open($lost, 'f'); + } + + public function testAConnectionErrorWhileOpeningDoesNotStopTheNextFile(): void { + try { + SftpReadStream::open(new FakeSftpStorage(new FakeSftpConnection(['/home/f' => 'x'], throwAt: -1)), 'f'); + } catch (BackendException) { + } + $fh = SftpReadStream::open(new FakeSftpStorage(new FakeSftpConnection(['/home/f' => 'x'])), 'f'); + $this->assertSame('x', stream_get_contents($fh)); + fclose($fh); + } + + public function testAFailedCloseDoesNotFallBackOnTheSameConnection(): void { + $sftp = new FakeSftpConnection(['/home/f' => random_bytes(3 * 1048576)], failAt: 1048576, closeFails: true); + $storage = new FakeSftpStorage($sftp); + $fh = SftpReadStream::open($storage, 'f'); + $this->assertLessThan(3 * 1048576, strlen($this->readAll($fh))); + fclose($fh); + $this->assertSame([], $storage->fopens); + } + + public function testAStorageThatThrowsIsNotUsed(): void { + $storage = new class { + public function getRoot(): string { + return '/'; + } + + public function getConnection(): object { + throw new \RuntimeException('Login failed'); + } + }; + $this->assertNull(SftpReadStream::open($storage, 'f')); + } + + public function testAConnectionWithoutTheNeededMethodsIsNotUsed(): void { + $storage = new class { + public function getRoot(): string { + return '/'; + } + + public function getConnection(): object { + return new \stdClass(); + } + }; + $this->assertNull(SftpReadStream::open($storage, 'f')); + $this->assertNull(SftpReadStream::open(new \stdClass(), 'f')); + } + + public function testAPhpseclib3ConnectionIsNotUsed(): void { + // phpseclib 3 has no _-prefixed methods; its packet_type and packet methods are private. + $v3 = new class { + private int $packet_type = -1; + + private function _send_sftp_packet(): bool { + return true; + } + + private function _get_sftp_packet(): string { + return ''; + } + + public function _realpath(string $path): string { + return $path; + } + + public function realpath(string $path): string { + return $path . (string)$this->packet_type . $this->_send_sftp_packet() . $this->_get_sftp_packet(); + } + }; + $storage = new class($v3) { + public function __construct( + private object $connection, + ) { + } + + public function getRoot(): string { + return '/'; + } + + public function getConnection(): object { + return $this->connection; + } + }; + $this->assertNull(SftpReadStream::open($storage, 'f')); + } + + public function testAPrivatePacketTypeIsNotUsed(): void { + $conn = new class { + private int $packet_type = -1; + + public function _send_sftp_packet(): bool { + return true; + } + + public function _get_sftp_packet(): string { + return (string)$this->packet_type; + } + + public function _realpath(string $path): string { + return $path; + } + }; + $storage = new class($conn) { + public function __construct( + private object $connection, + ) { + } + + public function getRoot(): string { + return '/'; + } + + public function getConnection(): object { + return $this->connection; + } + }; + $this->assertNull(SftpReadStream::open($storage, 'f')); + } + + /** @param resource $fh */ + private function readAll($fh): string { + $data = ''; + while (!feof($fh)) { + $chunk = fread($fh, 65536); + if ($chunk === false || $chunk === '') { + break; + } + $data .= $chunk; + } + return $data; + } + + /** @param resource $fh */ + private function readExactly($fh, int $length): string { + $data = ''; + while (strlen($data) < $length && !feof($fh)) { + $data .= fread($fh, $length - strlen($data)); + } + return $data; + } +} + +final class FakeSftpStorage { + /** @var list paths opened through the storage's own stream */ + public array $fopens = []; + + public function __construct( + private FakeSftpConnection $connection, + private bool $fopenFails = false, + ) { + } + + /** files_external's own read stream: here the file from memory, seekable. */ + public function fopen(string $path, string $mode) { + $this->fopens[] = $path; + $data = $this->connection->file('/home/' . $path); + if ($this->fopenFails || $data === null) { + return false; + } + $fh = fopen('php://memory', 'r+'); + fwrite($fh, $data); + rewind($fh); + return $fh; + } + + public function getRoot(): string { + return '/home/'; + } + + public function getConnection(): FakeSftpConnection { + return $this->connection; + } +} + +/** The part of phpseclib\Net\SFTP (2.x) that SftpReadStream uses. */ +final class FakeSftpConnection { + public int $packet_type = -1; + public int $maxInFlight = 0; + /** @var list lengths asked for by READ requests */ + public array $reads = []; + /** @var array */ + public array $openHandles = []; + /** @var array answers by request id, in the order the server sent them */ + private array $answers = []; + private int $handles = 0; + + /** @param array $files */ + public function __construct( + private array $files, + private int $cap = 261120, + private ?int $failAt = null, + /** READ at or past this offset throws; -1: OPEN throws */ + private ?int $throwAt = null, + /** READ at or past this offset finds the connection gone (phpseclib returns false); -1: OPEN */ + private ?int $loseAt = null, + private bool $closeFails = false, + ) { + } + + public function file(string $path): ?string { + return $this->files[$path] ?? null; + } + + public function pending(): int { + return count($this->answers); + } + + public function _realpath(string $path): string { + return '/' . trim(preg_replace('#/+#', '/', $path), '/'); + } + + public function _send_sftp_packet(int $type, string $data, int $id = 1): bool { + switch ($type) { + case 3: // OPEN + if ($this->throwAt === -1) { + throw new \RuntimeException('Connection closed by server'); + } + if ($this->loseAt === -1) { + return false; + } + $path = substr($data, 4, unpack('N', $data)[1]); + if (!isset($this->files[$path])) { + $this->answers[$id] = [101, pack('N', 2)]; + break; + } + $handle = 'h' . ++$this->handles; + $this->openHandles[$handle] = $path; + $this->answers[$id] = [102, pack('Na*', strlen($handle), $handle)]; + break; + case 4: // CLOSE + if ($this->closeFails) { + return false; + } + $handle = substr($data, 4, unpack('N', $data)[1]); + unset($this->openHandles[$handle]); + $this->answers[$id] = [101, pack('N', 0)]; + break; + case 5: // READ + $len = unpack('N', $data)[1]; + $handle = substr($data, 4, $len); + ['hi' => $hi, 'lo' => $lo, 'n' => $n] = unpack('Nhi/Nlo/Nn', substr($data, 4 + $len)); + $offset = $hi * 4294967296 + $lo; + $this->reads[] = $n; + if ($this->throwAt !== null && $this->throwAt >= 0 && $offset >= $this->throwAt) { + throw new \RuntimeException('Connection closed by server'); + } + if ($this->loseAt !== null && $this->loseAt >= 0 && $offset >= $this->loseAt) { + $this->answers = []; + return false; + } + $file = $this->files[$this->openHandles[$handle]]; + if ($this->failAt !== null && $offset >= $this->failAt) { + $this->answers[$id] = [101, pack('N', 4)]; // FX_FAILURE + } elseif ($offset >= strlen($file)) { + $this->answers[$id] = [101, pack('N', 1)]; // FX_EOF + } else { + $chunk = substr($file, $offset, min($n, $this->cap)); + $this->answers[$id] = [103, pack('Na*', strlen($chunk), $chunk)]; + } + $this->maxInFlight = max($this->maxInFlight, count($this->answers)); + break; + default: + throw new \LogicException("unexpected packet type $type"); + } + return true; + } + + public function _get_sftp_packet(?int $id = null): string|false { + if ($this->answers === []) { + throw new \LogicException('waiting for an answer that will never come'); + } + if ($id === null || !isset($this->answers[$id])) { + throw new \LogicException("no answer for request $id"); + } + [$this->packet_type, $payload] = $this->answers[$id]; + unset($this->answers[$id]); + return $payload; + } +} diff --git a/tests/Unit/Repository/PackReadTest.php b/tests/Unit/Repository/PackReadTest.php new file mode 100644 index 0000000..70a1f30 --- /dev/null +++ b/tests/Unit/Repository/PackReadTest.php @@ -0,0 +1,172 @@ +root = $this->tempDir(); + $this->keys = KeyRing::generate(); + $this->src = $this->tempDir(); + for ($i = 0; $i < 200; $i++) { + file_put_contents(sprintf('%s/f%03d', $this->src, $i), random_bytes(20000)); + } + $backend = new LocalBackend($this->root); + $repo = Repository::initWithKey($backend, $this->keys, $this->keys->wrap('pw')); + $this->snapshot = $repo->backupDirectory($this->src)['snapshot']; + } + + private function open(PackOpenCountingBackend $backend): Repository { + return Repository::openWithKey($backend, $this->keys, null, null); + } + + public function testFullRestoreOpensEachPackOnce(): void { + $backend = new PackOpenCountingBackend(new LocalBackend($this->root)); + $target = $this->tempDir(); + + $this->assertSame(200, $this->open($backend)->restore($this->snapshot, $target)); + + for ($i = 0; $i < 200; $i++) { + $name = sprintf('f%03d', $i); + $this->assertSame(file_get_contents("$this->src/$name"), file_get_contents("$target/$name")); + } + // One pack: one stream instead of 200 requests. + $this->assertSame(1, $backend->packOpens()); + } + + public function testDeepVerifyOpensEachPackOnce(): void { + $backend = new PackOpenCountingBackend(new LocalBackend($this->root)); + + $result = $this->open($backend)->verify($this->snapshot, deep: true); + + $this->assertSame(200 * 20000, $result['bytesChecked']); + $this->assertSame(1, $backend->packOpens()); + } + + public function testReadingBackwardsOpensThePackAgain(): void { + $backend = new PackOpenCountingBackend(new LocalBackend($this->root)); + $repo = $this->open($backend); + + $this->assertSame(1, $repo->restore($this->snapshot, $this->tempDir(), 'f150')); + $this->assertSame(1, $repo->restore($this->snapshot, $this->tempDir(), 'f050')); + + $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); + $pack = $this->root . '/' . $packs[0]; + $fh = fopen($pack, 'r+b'); + ftruncate($fh, intdiv(filesize($pack), 2)); + fclose($fh); + + $result = $this->open(new PackOpenCountingBackend(new LocalBackend($this->root)))->verify($this->snapshot, deep: true); + + $this->assertGreaterThan(80 * 20000, $result['bytesChecked']); + $this->assertNotSame([], $result['unreadable']); + $this->assertSame([], $result['failed']); + } +} + +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); + } + + public function packOpens(): int { + return count(array_filter($this->gets, fn (string $p): bool => str_starts_with($p, 'packs/'))); + } + + public function get(string $path) { + $this->gets[] = $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 { + $this->gets[] = $path; + return $this->inner->getRange($path, $offset, $length); + } + + public function exists(string $path): bool { + return $this->inner->exists($path); + } + + public function move(string $from, string $to): void { + $this->inner->move($from, $to); + } + + public function list(string $prefix): array { + return $this->inner->list($prefix); + } + + public function listFiles(string $dir): array { + return $this->inner->listFiles($dir); + } + + public function delete(string $path): void { + $this->inner->delete($path); + } +} diff --git a/tests/Unit/Repository/VerifyTest.php b/tests/Unit/Repository/VerifyTest.php index fccdcb7..c8fcb17 100644 --- a/tests/Unit/Repository/VerifyTest.php +++ b/tests/Unit/Repository/VerifyTest.php @@ -129,7 +129,7 @@ class VerifyTest extends TestCase { } } -/** Delegates to a real backend, but getRange() throws for the first $failures calls. */ +/** Delegates to a real backend, but reading a pack throws for the first $failures reads. */ final class FlakyRangeBackend implements IBackend { public function __construct( private IBackend $inner, @@ -143,15 +143,22 @@ final class FlakyRangeBackend implements IBackend { } public function get(string $path) { + if (str_starts_with($path, 'packs/')) { + $this->fail(); + } return $this->inner->get($path); } public function getRange(string $path, int $offset, int $length): string { + $this->fail(); + return $this->inner->getRange($path, $offset, $length); + } + + private function fail(): void { if ($this->failures > 0) { $this->failures--; throw $this->error ?? new BackendException('connection reset'); } - return $this->inner->getRange($path, $offset, $length); } public function exists(string $path): bool {