From 5f8d5faee7ef56c69e55b66090f55b9f31215904 Mon Sep 17 00:00:00 2001 From: Dillard Blom Date: Thu, 8 Oct 2026 22:17:53 +0200 Subject: [PATCH 1/7] Repository: read the blobs of a pack from one open stream Restores and deep verifies requested every blob separately: a new file handle and a seek on the location each time. They read the blobs of a snapshot in the order they were written, so the stream of the pack read last now stays open and is read on; a jump backwards, to another pack, or more than 8 MiB ahead opens the pack again at the offset, as before. On a Hetzner Storage Box over SMB, restoring 86 MB in 202 files went from 0.78 MB/s to 4.2 MB/s. --- CHANGELOG.md | 9 ++ lib/Repository/Repository.php | 73 +++++++++++++- tests/Unit/Repository/PackReadTest.php | 134 +++++++++++++++++++++++++ tests/Unit/Repository/VerifyTest.php | 11 +- 4 files changed, 223 insertions(+), 4 deletions(-) create mode 100644 tests/Unit/Repository/PackReadTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index cd58819..6a7d9f9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,15 @@ 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. +## [Unreleased] + +### 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. + ## [0.9.3-beta.4] - 2026-10-08 Bugfix release. diff --git a/lib/Repository/Repository.php b/lib/Repository/Repository.php index d6c4f1d..8c57bc7 100644 --- a/lib/Repository/Repository.php +++ b/lib/Repository/Repository.php @@ -30,12 +30,18 @@ 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 function __construct( private IBackend $backend, @@ -708,7 +714,70 @@ 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->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(); } /** Decrypts and checks a blob read from its pack; throws when it is corrupted or tampered with. */ @@ -848,7 +917,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->readPack($pack, $offset, $length); } catch (\Exception $e) { // storage adapters throw their own exceptions; \Error (a bug) is not retried $stats['readError'] = $e->getMessage(); } diff --git a/tests/Unit/Repository/PackReadTest.php b/tests/Unit/Repository/PackReadTest.php new file mode 100644 index 0000000..936b573 --- /dev/null +++ b/tests/Unit/Repository/PackReadTest.php @@ -0,0 +1,134 @@ +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 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 = []; + + public function __construct( + private IBackend $inner, + ) { + } + + public function put(string $path, $stream): void { + $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; + return $this->inner->get($path); + } + + 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 { From b309081d1acd30b4f9529d5b061ef500b478cb80 Mon Sep 17 00:00:00 2001 From: Dillard Blom Date: Thu, 8 Oct 2026 22:34:02 +0200 Subject: [PATCH 2/7] 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 { From bcfec17cfe5862567a06459a25fa5e8f90324f22 Mon Sep 17 00:00:00 2001 From: Dillard Blom Date: Thu, 8 Oct 2026 22:38:32 +0200 Subject: [PATCH 3/7] README: WebDAV request timeout on slow uploads; deep verify no longer fetches per blob --- CHANGELOG.md | 1 + README.md | 7 ++++--- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6a7d9f9..3ade99b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,7 @@ instead carried by the app description, the git tag, the release, and this file. 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. +- README: a WebDAV location on a slow upload needs a higher `davstorage.request_timeout`. ## [0.9.3-beta.4] - 2026-10-08 diff --git a/README.md b/README.md index 7aa4343..3bbcfe9 100644 --- a/README.md +++ b/README.md @@ -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 From d8e772137fe9db75f5358b40f8799e78749b4f45 Mon Sep 17 00:00:00 2001 From: Dillard Blom Date: Fri, 9 Oct 2026 04:27:21 +0200 Subject: [PATCH 4/7] StorageBackend: read SFTP locations with several requests in flight files_external's SFTP read stream asks for the next 256 KiB only after the previous answer has arrived, so reading costs one round trip per request whatever the line speed: 4.9 MB/s at 47 ms to a Hetzner Storage Box. SftpReadStream keeps up to eight READ requests in flight over the same phpseclib connection (15 MB/s on that link), starting with one so a small read does not fetch megabytes ahead. A server that answers less than asked is handled by asking again from where the answer ended; when more data follows, that size is used for later reads over the same connection, never below the 32 KiB every server must support. Without the needed connection methods, or when the open fails, the storage's own stream is used as before. --- lib/Backend/SftpReadStream.php | 262 ++++++++++++++++++++++ lib/Backend/StorageBackend.php | 13 ++ tests/Unit/Backend/SftpReadStreamTest.php | 243 ++++++++++++++++++++ 3 files changed, 518 insertions(+) create mode 100644 lib/Backend/SftpReadStream.php create mode 100644 tests/Unit/Backend/SftpReadStreamTest.php diff --git a/lib/Backend/SftpReadStream.php b/lib/Backend/SftpReadStream.php new file mode 100644 index 0000000..9bd9a33 --- /dev/null +++ b/lib/Backend/SftpReadStream.php @@ -0,0 +1,262 @@ +|null + */ + private static ?\WeakMap $chunks = null; + private static int $requestId = 0x4e470000; + + /** @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; + 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; + private bool $failed = 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 + */ + public static function open(object $storage, string $path) { + if (!method_exists($storage, 'getConnection') || !method_exists($storage, 'getRoot')) { + return null; + } + $sftp = $storage->getConnection(); + if (!is_object($sftp) || !method_exists($sftp, '_send_sftp_packet') || !method_exists($sftp, '_get_sftp_packet') + || !method_exists($sftp, '_realpath') || !property_exists($sftp, 'packet_type')) { + 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, '/')]]); + $fh = @fopen(self::PROTOCOL . '://file', 'r', false, $context); + return $fh === false ? false : $fh; + } + + /** Test hook: forget the learned READ sizes. */ + public static function reset(): void { + self::$chunks = null; + } + + 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']; + $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 false; + } + $response = $this->sftp->_get_sftp_packet($id); + if ($this->sftp->packet_type !== self::FXP_HANDLE || !is_string($response)) { + return false; + } + $this->handle = substr($response, 4); + return true; + } + + public function stream_read(int $count): string|false { + while (strlen($this->buffer) < $count && !$this->eof) { + if (!$this->receive()) { + return 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 { + return $this->eof && $this->buffer === ''; + } + + public function stream_tell(): int { + return $this->pos; + } + + public function stream_seek(int $offset, int $whence = SEEK_SET): bool { + 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 || $this->failed) { + return false; + } + 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 { + $this->drain(); + if ($this->handle !== '') { + $id = self::nextId(); + if ($this->sftp->_send_sftp_packet(self::FXP_CLOSE, pack('Na*', strlen($this->handle), $this->handle), $id)) { + $this->sftp->_get_sftp_packet($id); + } + } + $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; + } + 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)) { + 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)) { + 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(); + $this->failed = true; + return false; + } + + /** Read and drop the answers still on their way, so the connection is free again. */ + private function drain(): void { + foreach ($this->inFlight as [$id]) { + if (!$this->failed) { + $this->sftp->_get_sftp_packet($id); + } + } + $this->inFlight = []; + } + + 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..80f4c7d 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,16 @@ 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; + } + if ($fh === null) { + $this->sftpReads = false; + } + // false: not opened, which need not mean missing; 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/tests/Unit/Backend/SftpReadStreamTest.php b/tests/Unit/Backend/SftpReadStreamTest.php new file mode 100644 index 0000000..3d40b4a --- /dev/null +++ b/tests/Unit/Backend/SftpReadStreamTest.php @@ -0,0 +1,243 @@ + $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 testAServerErrorFailsTheReadAndLeavesTheConnectionUsable(): void { + $sftp = new FakeSftpConnection(['/home/f' => random_bytes(3 * 1048576), '/home/g' => 'second'], failAt: 1048576); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + $read = ''; + while (!feof($fh)) { + $chunk = fread($fh, 65536); + if ($chunk === false || $chunk === '') { + break; + } + $read .= $chunk; + } + $this->assertLessThan(3 * 1048576, strlen($read)); + fclose($fh); + $this->assertSame(0, $sftp->pending()); + + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'g'); + $this->assertSame('second', stream_get_contents($fh)); + fclose($fh); + } + + 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')); + } + + /** @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 { + public function __construct( + private FakeSftpConnection $connection, + ) { + } + + 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, + ) { + } + + 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 + $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 + $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; + $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; + } +} From c775e27dd2e718960d6c884712d44cdb68e8c27a Mon Sep 17 00:00:00 2001 From: Dillard Blom Date: Fri, 9 Oct 2026 09:59:10 +0200 Subject: [PATCH 5/7] CHANGELOG: SFTP reads with several requests in flight --- CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3ade99b..1ec9e8f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,10 @@ instead carried by the app description, the git tag, the release, and this file. 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. - README: a WebDAV location on a slow upload needs a higher `davstorage.request_timeout`. ## [0.9.3-beta.4] - 2026-10-08 From b3e3054737c1baf26d2f12c052280dee0ca895cd Mon Sep 17 00:00:00 2001 From: Dillard Blom Date: Fri, 9 Oct 2026 11:24:37 +0200 Subject: [PATCH 6/7] SftpReadStream: fall back to the storage's own stream instead of failing a read --- CHANGELOG.md | 5 +- lib/Backend/SftpReadStream.php | 216 ++++++++++++++--- lib/Backend/StorageBackend.php | 6 +- tests/Unit/Backend/SftpReadStreamTest.php | 267 ++++++++++++++++++++-- 4 files changed, 440 insertions(+), 54 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ec9e8f..f9823bc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,7 +18,10 @@ instead carried by the app description, the git tag, the release, and this file. - 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. + (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 diff --git a/lib/Backend/SftpReadStream.php b/lib/Backend/SftpReadStream.php index 9bd9a33..c982991 100644 --- a/lib/Backend/SftpReadStream.php +++ b/lib/Backend/SftpReadStream.php @@ -17,7 +17,10 @@ namespace OCA\NgBackup\Backend; * * Talks to the storage's phpseclib connection through the same internal methods files_external's * read stream uses (_send_sftp_packet, _get_sftp_packet, _realpath); open() returns null when the - * connection does not have them, and StorageBackend then falls back to the storage's own stream. + * connection does not offer them (phpseclib 3 does not), and StorageBackend then falls back to the + * storage's own stream. A restore must rather be slow than fail: when a read fails halfway, the + * stream goes on from the same position with the storage's own stream. After MAX_FAILURES streams + * in a row failed, open() returns null for the rest of the process; one that works resets the count. * Like that stream, it owns the connection while open: no other request may go over the same * connection until it is closed (PackReaderGuard takes care of that for the repository). */ @@ -47,6 +50,11 @@ final class SftpReadStream { */ 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; @@ -58,6 +66,11 @@ final class SftpReadStream { * @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; @@ -68,7 +81,10 @@ final class SftpReadStream { 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; @@ -76,28 +92,52 @@ final class SftpReadStream { * @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 (!method_exists($storage, 'getConnection') || !method_exists($storage, 'getRoot')) { + if (self::$failures >= self::MAX_FAILURES) { return null; } - $sftp = $storage->getConnection(); - if (!is_object($sftp) || !method_exists($sftp, '_send_sftp_packet') || !method_exists($sftp, '_get_sftp_packet') - || !method_exists($sftp, '_realpath') || !property_exists($sftp, 'packet_type')) { + 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 (!in_array(self::PROTOCOL, stream_get_wrappers(), true)) { - stream_wrapper_register(self::PROTOCOL, self::class); + if ($fh === false && self::openBroke()) { + throw new BackendException('SFTP connection failed while opening ' . $path); } - $root = rtrim((string)$storage->getRoot(), '/'); - $context = stream_context_create([self::PROTOCOL => ['session' => $sftp, 'path' => $root . '/' . ltrim($path, '/')]]); - $fh = @fopen(self::PROTOCOL . '://file', 'r', false, $context); - return $fh === false ? false : $fh; + return $fh; } - /** Test hook: forget the learned READ sizes. */ + /** 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 { @@ -106,26 +146,51 @@ final class SftpReadStream { return false; } $this->sftp = $options['session']; - $remote = $this->sftp->_realpath((string)$options['path']); - if (!is_string($remote)) { - return false; + $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(); } - $id = self::nextId(); - if (!$this->sftp->_send_sftp_packet(self::FXP_OPEN, pack('Na*N2', strlen($remote), $remote, self::FXF_READ, 0), $id)) { - return false; + if (!is_string($response)) { + return $this->connectionFailed(); } - $response = $this->sftp->_get_sftp_packet($id); - if ($this->sftp->packet_type !== self::FXP_HANDLE || !is_string($response)) { - return false; + 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 false; + return $this->useFallback() ? $this->stream_read($count) : false; } } $data = substr($this->buffer, 0, $count); @@ -135,22 +200,36 @@ final class SftpReadStream { } 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 || $this->failed) { + 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 { @@ -171,11 +250,59 @@ final class SftpReadStream { } 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 !== '') { - $id = self::nextId(); - if ($this->sftp->_send_sftp_packet(self::FXP_CLOSE, pack('Na*', strlen($this->handle), $this->handle), $id)) { - $this->sftp->_get_sftp_packet($id); + 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 = ''; @@ -186,11 +313,21 @@ final class SftpReadStream { 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]; @@ -199,10 +336,11 @@ final class SftpReadStream { [$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); + $data = substr($response, 4); if ($data === '') { return $this->fail(); } @@ -236,18 +374,30 @@ final class SftpReadStream { 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 { - foreach ($this->inFlight as [$id]) { - if (!$this->failed) { - $this->sftp->_get_sftp_packet($id); - } - } + $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 { diff --git a/lib/Backend/StorageBackend.php b/lib/Backend/StorageBackend.php index 80f4c7d..95672f3 100644 --- a/lib/Backend/StorageBackend.php +++ b/lib/Backend/StorageBackend.php @@ -87,10 +87,8 @@ final class StorageBackend implements IBackend { if (is_resource($fh)) { return $fh; } - if ($fh === null) { - $this->sftpReads = false; - } - // false: not opened, which need not mean missing; the storage's own stream decides + // 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) { diff --git a/tests/Unit/Backend/SftpReadStreamTest.php b/tests/Unit/Backend/SftpReadStreamTest.php index 3d40b4a..abef338 100644 --- a/tests/Unit/Backend/SftpReadStreamTest.php +++ b/tests/Unit/Backend/SftpReadStreamTest.php @@ -4,6 +4,7 @@ declare(strict_types=1); namespace OCA\NgBackup\Tests\Unit\Backend; +use OCA\NgBackup\Backend\BackendException; use OCA\NgBackup\Backend\SftpReadStream; use PHPUnit\Framework\TestCase; @@ -102,24 +103,132 @@ class SftpReadStreamTest extends TestCase { $this->assertSame(0, $sftp->pending()); } - public function testAServerErrorFailsTheReadAndLeavesTheConnectionUsable(): void { - $sftp = new FakeSftpConnection(['/home/f' => random_bytes(3 * 1048576), '/home/g' => 'second'], failAt: 1048576); - $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); - $read = ''; - while (!feof($fh)) { - $chunk = fread($fh, 65536); - if ($chunk === false || $chunk === '') { - break; - } - $read .= $chunk; - } - $this->assertLessThan(3 * 1048576, strlen($read)); - fclose($fh); - $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'); - $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'g'); - $this->assertSame('second', stream_get_contents($fh)); + $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 { @@ -136,6 +245,90 @@ class SftpReadStreamTest extends TestCase { $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 = ''; @@ -147,11 +340,28 @@ class SftpReadStreamTest extends TestCase { } 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/'; } @@ -178,9 +388,18 @@ final class FakeSftpConnection { 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); } @@ -192,6 +411,12 @@ final class FakeSftpConnection { 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)]; @@ -202,6 +427,9 @@ final class FakeSftpConnection { $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)]; @@ -212,6 +440,13 @@ final class FakeSftpConnection { ['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 From fe92ef70bf09f2eec208be8c34b97b4c4e1de734 Mon Sep 17 00:00:00 2001 From: Dillard Blom Date: Fri, 9 Oct 2026 11:58:52 +0200 Subject: [PATCH 7/7] Release 0.9.4-beta.5 --- CHANGELOG.md | 5 ++++- README.md | 2 +- appinfo/info.xml | 2 +- 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f9823bc..139f25a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,7 +7,9 @@ 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. -## [Unreleased] +## [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 @@ -152,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 3bbcfe9..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. 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