From ccbbf01c54988023123829f10fe0fb40f5718ea4 Mon Sep 17 00:00:00 2001 From: Dillard Blom Date: Mon, 19 Oct 2026 01:40:49 +0200 Subject: [PATCH] Full restore over SFTP: read packs ahead on four connections at once A restore read one pack at a time over one SFTP connection, which over a long line is bound by the round trip and by what one TCP connection gets: about 2-3 MiB/s from a Hetzner Storage Box to Canada, with PHP mostly waiting. Repository::restore() now first lists the ranges of packs the restore needs, in order (ReadAheadPlan), and the location reads them ahead into unlinked files in a spool folder in the target (IReadsAhead). Over SFTP, SftpRangeFetcher does that on four connections of its own, 16 requests in flight each, taking the answers of whichever connection has one. A range that fails, a damaged copy, or a blob that moved since the plan is read the usual way, which reports what is wrong. Other locations restore as before. On a test line with 120 ms RTT and 0.2% loss, a 1.6 GiB restore went from 5.1 to 17.8 MiB/s, all files identical. --- lib/Backend/IReadsAhead.php | 30 ++ lib/Backend/SftpRangeFetcher.php | 344 ++++++++++++++++++ lib/Backend/SftpRangeJob.php | 40 ++ lib/Backend/SftpReadStream.php | 4 +- lib/Backend/StorageBackend.php | 47 ++- lib/Repository/PackReaderGuard.php | 12 +- lib/Repository/ReadAheadPlan.php | 123 +++++++ lib/Repository/Repository.php | 66 +++- tests/Unit/Backend/SftpRangeFetcherTest.php | 189 ++++++++++ tests/Unit/Backend/SftpReadStreamTest.php | 3 + .../Backend/StorageBackendReadAheadTest.php | 103 ++++++ tests/Unit/Repository/ReadAheadTest.php | 272 ++++++++++++++ 12 files changed, 1228 insertions(+), 5 deletions(-) create mode 100644 lib/Backend/IReadsAhead.php create mode 100644 lib/Backend/SftpRangeFetcher.php create mode 100644 lib/Backend/SftpRangeJob.php create mode 100644 lib/Repository/ReadAheadPlan.php create mode 100644 tests/Unit/Backend/SftpRangeFetcherTest.php create mode 100644 tests/Unit/Backend/StorageBackendReadAheadTest.php create mode 100644 tests/Unit/Repository/ReadAheadTest.php diff --git a/lib/Backend/IReadsAhead.php b/lib/Backend/IReadsAhead.php new file mode 100644 index 0000000..873577e --- /dev/null +++ b/lib/Backend/IReadsAhead.php @@ -0,0 +1,30 @@ + $ranges [path, offset, length] each + * @param string $spool an existing local directory for the files that hold the ranges + * @return \Generator|null by position in $ranges; null when this backend + * cannot read ahead now + */ + public function readAhead(array $ranges, string $spool): ?\Generator; +} diff --git a/lib/Backend/SftpRangeFetcher.php b/lib/Backend/SftpRangeFetcher.php new file mode 100644 index 0000000..cb059b8 --- /dev/null +++ b/lib/Backend/SftpRangeFetcher.php @@ -0,0 +1,344 @@ + READ size per connection, lowered when a server answers less than asked */ + private array $chunks = []; + /** @var array the range each busy connection reads */ + private array $jobs = []; + /** @var array ranges read (or failed) and not yet given to the caller */ + private array $done = []; + private int $requestId = 0x4e520000; + + /** + * @param array $connections phpseclib 2 connections with their SFTP channel open + * @param list $ranges paths below $root + */ + private function __construct( + private array $connections, + private string $root, + private array $ranges, + private string $spool, + ) { + foreach (array_keys($connections) as $c) { + $this->chunks[$c] = self::CHUNK; + } + } + + /** + * See IReadsAhead::readAhead(). The connections are ended when the generator ends. + * + * @param list $connections + * @param list $ranges + * @return \Generator + */ + public static function fetch(array $connections, string $root, array $ranges, string $spool): \Generator { + $fetcher = new self($connections, rtrim($root, '/'), $ranges, $spool); + try { + yield from $fetcher->run(); + } finally { + $fetcher->end(); + } + } + + /** @return \Generator */ + private function run(): \Generator { + $count = count($this->ranges); + $next = 0; + for ($want = 0; $want < $count; $want++) { + while (!array_key_exists($want, $this->done)) { + foreach (array_keys($this->connections) as $c) { + if (!isset($this->jobs[$c]) && $next < $count && $next < $want + self::AHEAD) { + $this->start($c, $next++); + } + } + if ($this->jobs === [] && !array_key_exists($want, $this->done)) { + // no connection left: the rest the usual way + $this->done[$want] = null; + $next = max($next, $want + 1); + break; + } + foreach ($this->ready() as $c) { + $this->step($c); + } + } + $fh = $this->done[$want]; + unset($this->done[$want]); + try { + yield $want => $fh; + } finally { + if (is_resource($fh)) { + fclose($fh); + } + } + } + } + + /** + * The busy connections that have an answer waiting, so that a connection that waits for a + * lost packet to be sent again does not hold up the others; all of them when the connections + * do not tell (then they are taken in turn). + * + * @return list + */ + private function ready(): array { + $sockets = []; + $ready = []; + foreach ($this->jobs as $c => $job) { + $sftp = $this->connections[$c]; + $vars = get_object_vars($sftp); // only what is public: phpseclib 2's are + $buffered = $vars['requestBuffer'] ?? null; + if (is_array($buffered) && isset($job->queue[0]) && array_key_exists($job->queue[0][0], $buffered)) { + $ready[] = $c; // read already, while waiting for another answer + continue; + } + $socket = $vars['fsock'] ?? null; + if (!is_resource($socket)) { + return array_keys($this->jobs); + } + $sockets[$c] = $socket; + } + if ($ready !== [] || $sockets === []) { + return $ready; + } + $read = $sockets; + $write = null; + $except = null; + // Data in PHP's own buffer of a socket counts as readable. Nothing within the time: + // each in turn, where phpseclib's own timeout applies. + $n = @stream_select($read, $write, $except, self::SELECT_TIMEOUT); + return $n === false || $n === 0 ? array_keys($this->jobs) : array_keys($read); + } + + private function start(int $c, int $i): void { + [$path, $offset, $length] = $this->ranges[$i]; + $fh = $this->spoolFile(); + if ($fh === null) { + $this->done[$i] = null; + return; + } + if ($length <= 0) { + $this->done[$i] = $fh; + return; + } + $this->jobs[$c] = new SftpRangeJob($i, $fh, $offset, $offset + $length); + try { + $remote = $this->connections[$c]->_realpath($this->root . '/' . ltrim($path, '/')); + if (!is_string($remote)) { + $this->finish($c); + return; + } + $this->send($c, self::FXP_OPEN, pack('Na*N2', strlen($remote), $remote, self::FXF_READ, 0), self::OPEN); + } catch (\Throwable) { + $this->drop($c); + } + } + + /** Take the oldest answer on connection $c, then keep its requests in flight. */ + private function step(int $c): void { + $job = $this->jobs[$c]; + [$id, $kind, $offset, $length] = array_shift($job->queue) ?? throw new \LogicException('no request in flight'); + $sftp = $this->connections[$c]; + try { + $response = $sftp->_get_sftp_packet($id); + if (!is_string($response)) { + throw new BackendException('SFTP connection lost'); + } + $type = $sftp->packet_type; + if ($kind === self::OPEN) { + if ($type === self::FXP_HANDLE) { + $job->handle = substr($response, 4); + } else { + $job->failed = true; // missing, or no permission + } + } elseif ($kind === self::READ) { + if (!$job->failed) { + $this->received($c, $type, $response, $offset, $length); + } + } else { + $status = $type === self::FXP_STATUS && strlen($response) >= 4 ? unpack('N', $response) : false; + $closed = is_array($status) && $status[1] === self::FX_OK; + $this->finish($c); + if (!$closed) { + // handles might pile up on the server: not used again + $this->drop($c); + } + return; + } + $this->refill($c); + } catch (\Throwable) { + $this->drop($c); + } + } + + private function received(int $c, int $type, string $response, int $offset, int $length): void { + $job = $this->jobs[$c]; + $declared = strlen($response) >= 4 ? unpack('N', $response) : false; + $n = is_array($declared) ? $declared[1] : 0; + if ($type !== self::FXP_DATA || $n <= 0 || $n !== strlen($response) - 4 || $n > $length) { + // end of file before the end of the range, a server error or a malformed answer + $job->failed = true; + return; + } + if (fseek($job->fh, $offset - $job->start) !== 0 || fwrite($job->fh, substr($response, 4)) !== $n) { + $job->failed = true; + return; + } + $job->got += $n; + if ($n < $length) { + // a server that caps a READ: ask for the rest, and no more than it gives from now on + $job->redo[] = [$offset + $n, $length - $n]; + $this->chunks[$c] = max(self::MIN_CHUNK, min($this->chunks[$c], $n)); + } + } + + /** Send READs until DEPTH are in flight; when nothing is left to ask for, close the file. */ + private function refill(int $c): void { + $job = $this->jobs[$c]; + if ($job->handle !== '' && !$job->failed) { + while (count($job->queue) < self::DEPTH) { + if ($job->redo !== []) { + [$offset, $length] = array_shift($job->redo); + } elseif ($job->next < $job->end) { + $offset = $job->next; + $length = min($this->chunks[$c], $job->end - $offset); + $job->next += $length; + } else { + break; + } + $packet = pack('Na*N3', strlen($job->handle), $job->handle, intdiv($offset, 4294967296), $offset % 4294967296, $length); + $this->send($c, self::FXP_READ, $packet, self::READ, $offset, $length); + } + } + if ($job->queue !== []) { + return; + } + if ($job->handle === '') { + $this->finish($c); // not opened: nothing to close + return; + } + $this->send($c, self::FXP_CLOSE, pack('Na*', strlen($job->handle), $job->handle), self::CLOSE); + } + + private function send(int $c, int $type, string $packet, int $kind, int $offset = 0, int $length = 0): void { + $this->requestId = $this->requestId >= 0x7fffffff ? 0x4e520000 : $this->requestId + 1; + if (!$this->connections[$c]->_send_sftp_packet($type, $packet, $this->requestId)) { + throw new BackendException('SFTP connection failed while sending a request'); + } + $this->jobs[$c]->queue[] = [$this->requestId, $kind, $offset, $length]; + } + + /** The range of connection $c is done: its stream for the caller, or null when it failed. */ + private function finish(int $c): void { + $job = $this->jobs[$c]; + unset($this->jobs[$c]); + if ($job->failed || $job->got !== $job->end - $job->start || !rewind($job->fh)) { + self::closeFile($job->fh); + $this->done[$job->i] = null; + return; + } + $this->done[$job->i] = $job->fh; + } + + /** A connection in an unknown state: its range fails, and it is not used again. */ + private function drop(int $c): void { + if (isset($this->jobs[$c])) { + $job = $this->jobs[$c]; + unset($this->jobs[$c]); + self::closeFile($job->fh); + $this->done[$job->i] = null; + } + self::disconnect($this->connections[$c]); + unset($this->connections[$c]); + } + + /** @return resource|null a new file in the spool, already unlinked: nothing is left behind if the process dies */ + private function spoolFile() { + $path = $this->spool . '/' . bin2hex(random_bytes(8)); + $fh = @fopen($path, 'x+b'); + if ($fh === false) { + return null; + } + @unlink($path); + return $fh; + } + + private function end(): void { + foreach ($this->jobs as $job) { + self::closeFile($job->fh); + } + foreach ($this->done as $fh) { + if (is_resource($fh)) { + fclose($fh); + } + } + $this->jobs = []; + $this->done = []; + foreach ($this->connections as $sftp) { + self::disconnect($sftp); + } + $this->connections = []; + } + + private static function closeFile(mixed $fh): void { + if (is_resource($fh)) { + fclose($fh); + } + } + + private static function disconnect(object $sftp): void { + try { + if (is_callable([$sftp, 'disconnect'])) { + $sftp->disconnect(); + } + } catch (\Throwable) { + } + } +} diff --git a/lib/Backend/SftpRangeJob.php b/lib/Backend/SftpRangeJob.php new file mode 100644 index 0000000..1e36d4d --- /dev/null +++ b/lib/Backend/SftpRangeJob.php @@ -0,0 +1,40 @@ + offset and length of parts to ask for again (after a short answer) */ + public array $redo = []; + /** @var list request id, kind, offset and length, in the order sent */ + public array $queue = []; + /** Bytes received. */ + public int $got = 0; + public bool $failed = false; + + /** @var resource the spool file that holds the range */ + public $fh; + + /** @param resource $fh */ + public function __construct( + public int $i, + $fh, + public int $start, + public int $end, + ) { + $this->fh = $fh; + $this->next = $start; + } +} diff --git a/lib/Backend/SftpReadStream.php b/lib/Backend/SftpReadStream.php index cd91855..1be1ee3 100644 --- a/lib/Backend/SftpReadStream.php +++ b/lib/Backend/SftpReadStream.php @@ -144,9 +144,9 @@ final class SftpReadStream { /** * The storage's phpseclib connection, with its SFTP channel open, when it offers what these - * streams need; null otherwise. + * streams (and SftpRangeFetcher) need; null otherwise. */ - private static function connection(object $storage): ?object { + public static function connection(object $storage): ?object { if (!is_callable([$storage, 'getConnection']) || !is_callable([$storage, 'getRoot'])) { return null; } diff --git a/lib/Backend/StorageBackend.php b/lib/Backend/StorageBackend.php index c9c0ab3..8d59e6c 100644 --- a/lib/Backend/StorageBackend.php +++ b/lib/Backend/StorageBackend.php @@ -14,7 +14,7 @@ use OCP\Files\Storage\IWriteStreamStorage; * IBackend on top of a Nextcloud storage object (from files_external or core), below a base * directory on that storage. */ -final class StorageBackend implements IBackend { +final class StorageBackend implements IBackend, IReadsAhead { /** 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. */ @@ -243,6 +243,51 @@ final class StorageBackend implements IBackend { } } + /** + * Over SFTP: SftpRangeFetcher, on CONNECTIONS new connections of its own (built with the + * reconnect function), so the storage's connection stays free for other requests. Elsewhere + * not (yet): null. + */ + public function canReadAhead(): bool { + return $this->sftpReads && $this->reconnect !== null; + } + + public function readAhead(array $ranges, string $spool): ?\Generator { + $reconnect = $this->reconnect; + if (!$this->sftpReads || $reconnect === null || $ranges === []) { + return null; + } + $storages = []; + $connections = []; + $root = ''; + for ($n = 0; $n < SftpRangeFetcher::CONNECTIONS; $n++) { + try { + $storage = $reconnect(); + $sftp = SftpReadStream::connection($storage); + if ($sftp === null || !is_callable([$storage, 'getRoot'])) { + break; + } + $root = (string)$storage->getRoot(); + } catch (\Throwable) { + break; // the server may allow fewer connections: go on with those that work + } + $storages[] = $storage; + $connections[] = $sftp; + } + if ($connections === []) { + return null; + } + $abs = array_map(fn (array $r): array => [$this->abs($r[0]), $r[1], $r[2]], $ranges); + return (static function () use ($storages, $connections, $root, $abs, $spool): \Generator { + // the storages hold their connections: kept until the fetcher is done + try { + yield from SftpRangeFetcher::fetch($connections, $root, $abs, $spool); + } finally { + unset($storages); + } + })(); + } + public function getRange(string $path, int $offset, int $length): string { return $this->call(fn () => $this->readRange($path, $offset, $length)); } diff --git a/lib/Repository/PackReaderGuard.php b/lib/Repository/PackReaderGuard.php index 99577e6..24ca5c7 100644 --- a/lib/Repository/PackReaderGuard.php +++ b/lib/Repository/PackReaderGuard.php @@ -8,9 +8,10 @@ declare(strict_types=1); namespace OCA\NgBackup\Repository; use OCA\NgBackup\Backend\IBackend; +use OCA\NgBackup\Backend\IReadsAhead; /** The location as Repository sees it: closes the PackReader's open stream before every other request. */ -final class PackReaderGuard implements IBackend { +final class PackReaderGuard implements IBackend, IReadsAhead { public function __construct( private IBackend $inner, private PackReader $reader, @@ -43,6 +44,15 @@ final class PackReaderGuard implements IBackend { return $this->inner->getRanges($ranges); } + public function canReadAhead(): bool { + return $this->inner instanceof IReadsAhead && $this->inner->canReadAhead(); + } + + /** Over connections of the location's own: the open stream may stay open. */ + public function readAhead(array $ranges, string $spool): ?\Generator { + return $this->inner instanceof IReadsAhead ? $this->inner->readAhead($ranges, $spool) : null; + } + public function exists(string $path): bool { $this->reader->close(); return $this->inner->exists($path); diff --git a/lib/Repository/ReadAheadPlan.php b/lib/Repository/ReadAheadPlan.php new file mode 100644 index 0000000..dc74395 --- /dev/null +++ b/lib/Repository/ReadAheadPlan.php @@ -0,0 +1,123 @@ + [path, offset, length] */ + private array $ranges = []; + /** @var \Generator|null */ + private ?\Generator $streams = null; + private int $current = -1; + /** @var resource|null */ + private $stream = null; + + /** @param string $spool local folder for the ranges read ahead (see IReadsAhead::readAhead()) */ + public function __construct( + public readonly string $spool, + ) { + } + + public function add(string $path, int $offset, int $length): void { + $last = array_pop($this->ranges); + if ($last !== null) { + [$p, $start, $len] = $last; + if ($p === $path && $offset >= $start && $offset <= $start + $len + self::GAP + && $offset + $length - $start <= self::MAX_RANGE) { + $this->ranges[] = [$p, $start, max($len, $offset + $length - $start)]; + return; + } + $this->ranges[] = $last; + } + $this->ranges[] = [$path, $offset, $length]; + } + + /** @return list */ + public function ranges(): array { + return $this->ranges; + } + + /** @param \Generator $streams what IReadsAhead::readAhead() gave for ranges() */ + public function readFrom(\Generator $streams): void { + $this->streams = $streams; + } + + /** The bytes of a blob, or null when its range could not be read ahead (read it the usual way). */ + public function read(string $path, int $offset, int $length): ?string { + if ($this->streams === null) { + return null; + } + if (!$this->contains($this->current, $path, $offset, $length)) { + // Normally the next range. Further on when blobs before it were read elsewhere: the + // index was reloaded after another server's cleanup moved them to new packs. + $to = null; + for ($i = $this->current + 1; $i <= $this->current + self::SEARCH && $to === null; $i++) { + if ($this->contains($i, $path, $offset, $length)) { + $to = $i; + } + } + if ($to === null) { + return null; // not where the plan has it + } + if ($this->current < 0) { + $this->streams->current(); + $this->current = 0; + } + while ($this->current < $to) { + $this->streams->next(); + $this->current++; + } + $stream = $this->streams->valid() ? $this->streams->current() : null; + $this->stream = is_resource($stream) ? $stream : null; + } + $fh = $this->stream; + if ($fh === null || fseek($fh, $offset - $this->ranges[$this->current][1]) !== 0) { + return null; + } + $data = ''; + while (strlen($data) < $length) { + $chunk = fread($fh, $length - strlen($data)); + if ($chunk === false || $chunk === '') { + return null; + } + $data .= $chunk; + } + return $data; + } + + /** Ends the reads ahead, also halfway: closes their streams and connections. */ + public function close(): void { + $this->stream = null; + $streams = $this->streams; + $this->streams = null; + unset($streams); // the generator's finally blocks run when it is destroyed + } + + private function contains(int $i, string $path, int $offset, int $length): bool { + if ($i < 0 || !isset($this->ranges[$i])) { + return false; + } + [$p, $start, $len] = $this->ranges[$i]; + return $p === $path && $offset >= $start && $offset + $length <= $start + $len; + } +} diff --git a/lib/Repository/Repository.php b/lib/Repository/Repository.php index 83b4abc..ad71021 100644 --- a/lib/Repository/Repository.php +++ b/lib/Repository/Repository.php @@ -9,6 +9,7 @@ namespace OCA\NgBackup\Repository; use OCA\NgBackup\Backend\BackendException; use OCA\NgBackup\Backend\IBackend; +use OCA\NgBackup\Backend\IReadsAhead; use OCA\NgBackup\Crypto\KeyRing; use OCA\NgBackup\Crypto\StreamCipher; @@ -216,6 +217,53 @@ final class Repository { if ($prefix !== '' && !self::isSafePath(rtrim($prefix, '/'))) { throw new RepositoryException('Invalid path prefix'); } + $plan = $this->readAhead($snapshotId, $prefix, $target); + try { + return $this->restoreEntries($snapshotId, $target, $prefix, $stripPrefix, $heartbeat, $plan); + } finally { + $plan?->close(); + if ($plan !== null) { + @rmdir($plan->spool); + } + } + } + + /** + * Start reading the packs of a restore ahead, when the location can (IReadsAhead): a first + * pass over the snapshot lists the ranges of packs it needs, in order. The spool is a + * folder in the target, which has room for the restore anyway. + */ + private function readAhead(string $snapshotId, string $prefix, string $target): ?ReadAheadPlan { + if (!$this->backend instanceof IReadsAhead || !$this->backend->canReadAhead()) { + return null; + } + $plan = new ReadAheadPlan(rtrim($target, '/') . '/.ngb-restore-spool-' . bin2hex(random_bytes(4))); + foreach ($this->streamEntries($snapshotId) as $entry) { + if ($prefix !== '' && !str_starts_with($entry['p'], $prefix)) { + continue; + } + foreach ($entry['b'] as $id) { + try { + [$pack, $offset, $length] = $this->index->get($id); + } catch (RepositoryException) { + continue; // the restore reports it when it gets there + } + $plan->add('packs/' . substr($pack, 0, 2) . '/' . $pack, $offset, $length); + } + } + if (count($plan->ranges()) < 2 || (!is_dir($plan->spool) && !@mkdir($plan->spool, 0700, true))) { + return null; + } + $streams = $this->backend->readAhead($plan->ranges(), $plan->spool); + if ($streams === null) { + @rmdir($plan->spool); + return null; + } + $plan->readFrom($streams); + return $plan; + } + + private function restoreEntries(string $snapshotId, string $target, string $prefix, bool $stripPrefix, ?callable $heartbeat, ?ReadAheadPlan $plan): int { $count = 0; foreach ($this->streamEntries($snapshotId) as $entry) { if ($prefix !== '' && !str_starts_with($entry['p'], $prefix)) { @@ -240,7 +288,7 @@ final class Repository { throw new RepositoryException('Cannot create a temporary file for ' . $entry['p']); } foreach ($entry['b'] as $id) { - fwrite($out, $this->loadBlob($id)); + fwrite($out, $this->restoreBlob($id, $plan)); if ($heartbeat !== null) { $heartbeat(); } @@ -789,6 +837,22 @@ final class Repository { } } + /** A blob for restore(): read ahead when it could be, otherwise (or when that copy is bad) the usual way. */ + private function restoreBlob(string $id, ?ReadAheadPlan $plan): string { + if ($plan !== null) { + [$pack, $offset, $length] = $this->index->get($id); + $encrypted = $plan->read('packs/' . substr($pack, 0, 2) . '/' . $pack, $offset, $length); + if ($encrypted !== null) { + try { + return $this->decodeBlob($id, $encrypted); + } catch (\Exception) { + // read again from the location below, which reports what is wrong + } + } + } + return $this->loadBlob($id); + } + private function loadBlob(string $id): string { [$pack, $offset, $length] = $this->index->get($id); try { diff --git a/tests/Unit/Backend/SftpRangeFetcherTest.php b/tests/Unit/Backend/SftpRangeFetcherTest.php new file mode 100644 index 0000000..57036f4 --- /dev/null +++ b/tests/Unit/Backend/SftpRangeFetcherTest.php @@ -0,0 +1,189 @@ + */ + private array $files = []; + + protected function setUp(): void { + SftpReadStream::reset(); + for ($i = 0; $i < 10; $i++) { + $this->files["/home/repo/packs/p$i"] = random_bytes(3 * 1048576 + $i * 1000); + } + } + + /** @return list */ + private function connections(int $n, array $options = []): array { + $result = []; + for ($c = 0; $c < $n; $c++) { + $result[] = new FakeSftpConnection($this->files, ...($options[$c] ?? [])); + } + return $result; + } + + /** @return list what StorageBackend passes on: each with its SFTP channel open */ + private static function opened(array $connections): array { + return array_map(fn (FakeSftpConnection $c): object => SftpReadStream::connection(new FakeSftpStorage($c)), $connections); + } + + /** @return list */ + private function ranges(): array { + $ranges = []; + for ($i = 0; $i < 10; $i++) { + $ranges[] = ["repo/packs/p$i", 1000 * $i, 2 * 1048576 + 333]; + } + return $ranges; + } + + /** @return array sha1 of the contents by position (a failing comparison of MiBs would not fit in memory) */ + private static function collect(\Generator $streams): array { + $result = []; + foreach ($streams as $i => $fh) { + $result[$i] = $fh === null ? null : sha1(stream_get_contents($fh)); + } + return $result; + } + + private function expected(array $ranges): array { + return array_map(fn (array $r): string => sha1(substr($this->files['/home/' . $r[0]], $r[1], $r[2])), $ranges); + } + + public function testReadsEveryRangeInOrderOverAllConnections(): void { + $connections = $this->connections(4); + $ranges = $this->ranges(); + $spool = $this->tempDir(); + + $this->assertSame($this->expected($ranges), self::collect(SftpRangeFetcher::fetch(self::opened($connections), '/home/', $ranges, $spool))); + + foreach ($connections as $c) { + $this->assertGreaterThan(0, count($c->reads), 'every connection reads'); + $this->assertLessThanOrEqual(SftpRangeFetcher::DEPTH, $c->maxInFlight); + $this->assertSame([], $c->openHandles); + $this->assertSame(0, $c->pending()); + $this->assertTrue($c->disconnected, 'its own connections end with it'); + } + $this->assertSame([], glob("$spool/*"), 'nothing left in the spool'); + } + + public function testKeepsSeveralRequestsInFlightOnEachConnection(): void { + $this->files['/home/repo/packs/big'] = random_bytes(6 * 1048576); + $connections = $this->connections(2); + self::collect(SftpRangeFetcher::fetch(self::opened($connections), '/home/', [['repo/packs/big', 0, 6 * 1048576], ['repo/packs/big', 5, 6 * 1048576 - 5]], $this->tempDir())); + + $this->assertSame(SftpRangeFetcher::DEPTH, $connections[0]->maxInFlight); + $this->assertSame(SftpRangeFetcher::DEPTH, $connections[1]->maxInFlight); + } + + public function testAServerThatCapsAReadStillGivesEveryByte(): void { + $connections = $this->connections(3, array_fill(0, 3, ['cap' => 100000])); + $ranges = $this->ranges(); + + $this->assertSame($this->expected($ranges), self::collect(SftpRangeFetcher::fetch(self::opened($connections), '/home/', $ranges, $this->tempDir()))); + // learned: after the first short answers (and asking for their rest) it asks for no more than the server gives + foreach ($connections as $c) { + $this->assertLessThanOrEqual(2 * SftpRangeFetcher::DEPTH, count(array_filter($c->reads, fn (int $n): bool => $n > 100000))); + } + } + + public function testAMissingFileIsLeftToTheCaller(): void { + $connections = $this->connections(2); + $ranges = $this->ranges(); + $ranges[3][0] = 'repo/packs/gone'; + + $result = self::collect(SftpRangeFetcher::fetch(self::opened($connections), '/home/', $ranges, $this->tempDir())); + + $expected = $this->expected($this->ranges()); + $expected[3] = null; + $this->assertSame($expected, $result); + $this->assertSame([], $connections[0]->openHandles); + } + + public function testARangePastTheEndOfTheFileIsLeftToTheCaller(): void { + $ranges = $this->ranges(); + $ranges[5][2] = 10 * 1048576; + + $result = self::collect(SftpRangeFetcher::fetch(self::opened($this->connections(2)), '/home/', $ranges, $this->tempDir())); + + $this->assertNull($result[5]); + $this->assertSame($this->expected($this->ranges())[6], $result[6]); + } + + public function testAServerErrorFailsOnlyThatRange(): void { + // every read from 1 MiB on fails on connection 0: its ranges, not those of the others + $connections = $this->connections(2, [['failAt' => 1048576]]); + $ranges = $this->ranges(); + + $result = self::collect(SftpRangeFetcher::fetch(self::opened($connections), '/home/', $ranges, $this->tempDir())); + + $expected = $this->expected($ranges); + $failed = array_keys(array_filter($result, fn (?string $r): bool => $r === null)); + $this->assertNotSame([], $failed); + foreach ($result as $i => $data) { + if ($data !== null) { + $this->assertSame($expected[$i], $data); + } + } + $this->assertLessThan(10, count($failed), 'the other connection goes on'); + $this->assertSame([], $connections[0]->openHandles, 'a failed range is closed too'); + } + + public function testALostConnectionIsNotUsedAgainAndTheOthersGoOn(): void { + $connections = $this->connections(3, [['loseAt' => 1048576]]); + $ranges = $this->ranges(); + + $result = self::collect(SftpRangeFetcher::fetch(self::opened($connections), '/home/', $ranges, $this->tempDir())); + + $expected = $this->expected($ranges); + $this->assertSame(1, count(array_filter($result, fn (?string $r): bool => $r === null)), 'only the range it was reading'); + foreach ($result as $i => $data) { + if ($data !== null) { + $this->assertSame($expected[$i], $data); + } + } + $this->assertTrue($connections[0]->disconnected); + $this->assertLessThanOrEqual(1048576 / 261120 + SftpRangeFetcher::DEPTH + 1, count($connections[0]->reads), 'not used after it broke'); + } + + public function testWhenEveryConnectionIsLostTheRestIsLeftToTheCaller(): void { + $connections = $this->connections(2, [['throwAt' => 1048576], ['loseAt' => 1048576]]); + + $result = self::collect(SftpRangeFetcher::fetch(self::opened($connections), '/home/', $this->ranges(), $this->tempDir())); + + $this->assertSame(array_fill(0, 10, null), $result); + } + + public function testStoppingHalfwayEndsTheConnectionsAndLeavesNothingBehind(): void { + $connections = $this->connections(2); + $spool = $this->tempDir(); + $streams = SftpRangeFetcher::fetch(self::opened($connections), '/home/', $this->ranges(), $spool); + $this->assertIsResource($streams->current()); + + unset($streams); + + $this->assertTrue($connections[0]->disconnected); + $this->assertTrue($connections[1]->disconnected); + $this->assertSame([], glob("$spool/*")); + } + + public function testReadsNoMoreThanAheadOfTheCaller(): void { + $connections = $this->connections(4); + $streams = SftpRangeFetcher::fetch(self::opened($connections), '/home/', $this->ranges(), $this->tempDir()); + $streams->current(); + + $opened = array_sum(array_map(fn (FakeSftpConnection $c): int => count(array_unique(array_values($c->opened))), $connections)); + $this->assertLessThanOrEqual(SftpRangeFetcher::AHEAD, $opened); + } +} diff --git a/tests/Unit/Backend/SftpReadStreamTest.php b/tests/Unit/Backend/SftpReadStreamTest.php index e152baa..0718d0e 100644 --- a/tests/Unit/Backend/SftpReadStreamTest.php +++ b/tests/Unit/Backend/SftpReadStreamTest.php @@ -466,6 +466,8 @@ final class FakeSftpConnection { public array $reads = []; /** @var array */ public array $openHandles = []; + /** @var list paths of every OPEN request */ + public array $opened = []; /** Whether the SFTP channel is open: phpseclib 2 opens it on the first SFTP call. */ public bool $initialised = false; /** Packets sent */ @@ -532,6 +534,7 @@ final class FakeSftpConnection { return false; } $path = substr($data, 4, unpack('N', $data)[1]); + $this->opened[] = $path; if (!isset($this->files[$path])) { $this->answers[$id] = [101, pack('N', 2)]; break; diff --git a/tests/Unit/Backend/StorageBackendReadAheadTest.php b/tests/Unit/Backend/StorageBackendReadAheadTest.php new file mode 100644 index 0000000..5bf7b7c --- /dev/null +++ b/tests/Unit/Backend/StorageBackendReadAheadTest.php @@ -0,0 +1,103 @@ + */ + private array $connections = []; + /** @var array */ + private array $files = []; + /** New storages after this many fail to connect. */ + private ?int $connectLimit = null; + + protected function setUp(): void { + SftpReadStream::reset(); + for ($i = 0; $i < 6; $i++) { + $this->files["/home/base/repo/packs/p$i"] = random_bytes(1048576 + $i); + } + } + + private function storage(): IStorage { + if ($this->connectLimit !== null && count($this->connections) >= $this->connectLimit) { + throw new \RuntimeException('Too many connections'); + } + $sftp = new FakeSftpConnection($this->files); + $this->connections[] = $sftp; + $storage = $this->createMock(SftpLikeRootStorage::class); + $storage->method('instanceOfStorage')->willReturnCallback(fn (string $class): bool => $class === \OCA\Files_External\Lib\Storage\SFTP::class); + $storage->method('getConnection')->willReturn($sftp); + $storage->method('getRoot')->willReturn('/home/'); + $storage->method('is_dir')->willReturn(true); + return $storage; + } + + /** @return list */ + private static function ranges(): array { + return array_map(fn (int $i): array => ["packs/p$i", 7, 1048576 - 7], range(0, 5)); + } + + public function testReadsAheadOnNewConnectionsOnly(): void { + $backend = new StorageBackend($this->storage(), 'base/repo', false, fn (): IStorage => $this->storage()); + $this->assertTrue($backend->canReadAhead()); + + $result = []; + foreach ($backend->readAhead(self::ranges(), $this->tempDir()) as $i => $fh) { + $result[$i] = sha1(stream_get_contents($fh)); + } + + $expected = array_map(fn (int $i): string => sha1(substr($this->files["/home/base/repo/packs/p$i"], 7, 1048576 - 7)), range(0, 5)); + $this->assertSame($expected, $result); + $this->assertCount(1 + SftpRangeFetcher::CONNECTIONS, $this->connections); + $this->assertSame(0, $this->connections[0]->sent, 'the storage\'s own connection stays free'); + foreach (array_slice($this->connections, 1) as $c) { + $this->assertTrue($c->disconnected); + } + } + + public function testGoesOnWithTheConnectionsTheServerAllows(): void { + $this->connectLimit = 3; + $backend = new StorageBackend($this->storage(), 'base/repo', false, fn (): IStorage => $this->storage()); + + $count = 0; + foreach ($backend->readAhead(self::ranges(), $this->tempDir()) as $fh) { + $this->assertIsResource($fh); + $count++; + } + + $this->assertSame(6, $count); + $this->assertCount(3, $this->connections); + } + + public function testNoNewConnectionMeansNoReadingAhead(): void { + $this->connectLimit = 1; + $backend = new StorageBackend($this->storage(), 'base/repo', false, fn (): IStorage => $this->storage()); + + $this->assertNull($backend->readAhead(self::ranges(), $this->tempDir())); + } + + public function testWithoutAWayToConnectAgainItDoesNotReadAhead(): void { + $backend = new StorageBackend($this->storage(), 'base/repo', false, null); + + $this->assertFalse($backend->canReadAhead()); + $this->assertNull($backend->readAhead(self::ranges(), $this->tempDir())); + } +} + +interface SftpLikeRootStorage extends SftpLikeStorage { + public function getRoot(): string; +} diff --git a/tests/Unit/Repository/ReadAheadTest.php b/tests/Unit/Repository/ReadAheadTest.php new file mode 100644 index 0000000..276639f --- /dev/null +++ b/tests/Unit/Repository/ReadAheadTest.php @@ -0,0 +1,272 @@ +root = $this->tempDir(); + $this->keys = KeyRing::generate(); + $this->src = $this->tempDir(); + // three packs of 32 MiB, and small files in between + for ($i = 0; $i < 4; $i++) { + file_put_contents(sprintf('%s/big%d', $this->src, $i), random_bytes(20 * 1048576)); + for ($j = 0; $j < 20; $j++) { + file_put_contents(sprintf('%s/small%d-%02d', $this->src, $i, $j), random_bytes(30000)); + } + } + file_put_contents("$this->src/same", file_get_contents("$this->src/small0-00")); // the same blob again, later + $repo = Repository::initWithKey(new LocalBackend($this->root), $this->keys, $this->keys->wrap('pw')); + $this->snapshot = $repo->backupDirectory($this->src)['snapshot']; + } + + private function restore(ReadAheadBackend $backend, string $prefix = ''): string { + $target = $this->tempDir(); + $repo = Repository::openWithKey($backend, $this->keys, null, null); + $repo->restore($this->snapshot, $target, $prefix); + return $target; + } + + private function assertRestored(string $target, string $prefix = ''): void { + $expected = array_filter(self::hashTree($this->src), fn (string $p): bool => str_starts_with($p, $prefix), ARRAY_FILTER_USE_KEY); + $this->assertSame($expected, self::hashTree($target)); + $this->assertSame([], glob("$target/.ngb-restore-spool-*"), 'the spool is removed'); + } + + public function testARestoreReadsEveryPackAhead(): void { + $backend = new ReadAheadBackend(new LocalBackend($this->root)); + + $this->assertRestored($this->restore($backend)); + + $this->assertSame(0, $backend->counter->packOpens(), 'nothing read the usual way'); + $this->assertGreaterThanOrEqual(3, count($backend->asked)); + // in the order of the files, one range per pack the restore goes through (the same blob + // again later is a range of its own) + $this->assertLessThanOrEqual(count($backend->inner->list('packs')) + 1, count($backend->asked)); + } + + public function testARangeThatCouldNotBeReadAheadIsReadTheUsualWay(): void { + $backend = new ReadAheadBackend(new LocalBackend($this->root), fail: [1]); + + $this->assertRestored($this->restore($backend)); + + $this->assertGreaterThan(0, $backend->counter->packOpens()); + } + + public function testADamagedCopyIsReadAgainTheUsualWay(): void { + $backend = new ReadAheadBackend(new LocalBackend($this->root), damage: [0, 2]); + + $this->assertRestored($this->restore($backend)); + + $this->assertGreaterThan(0, $backend->counter->packOpens()); + } + + public function testAPartialRestoreReadsOnlyWhatItNeeds(): void { + $backend = new ReadAheadBackend(new LocalBackend($this->root)); + + $this->assertRestored($this->restore($backend, 'big1'), 'big1'); + + $this->assertSame(0, $backend->counter->packOpens()); + $this->assertCount(2, $backend->asked, 'the end of one pack and the start of the next'); + $total = array_sum(array_map(fn (array $r): int => $r[2], $backend->asked)); + $this->assertLessThan(21 * 1048576, $total, 'not the whole packs'); + } + + public function testABackendThatCannotReadAheadNowRestoresAsBefore(): void { + $backend = new ReadAheadBackend(new LocalBackend($this->root), unavailable: true); + + $this->assertRestored($this->restore($backend)); + + $this->assertGreaterThan(0, $backend->counter->packOpens()); + } + + public function testAFailingRestoreStillEndsTheReadsAheadAndRemovesTheSpool(): void { + // the second range fails, and so does reading it the usual way + $backend = new ReadAheadBackend(new LocalBackend($this->root), fail: [1], packsGone: true); + $target = $this->tempDir(); + $repo = Repository::openWithKey($backend, $this->keys, null, null); + + try { + $repo->restore($this->snapshot, $target); + $this->fail('restore should fail'); + } catch (BackendException $e) { + $this->assertStringStartsWith('Not found', $e->getMessage()); + } + + $this->assertTrue($backend->ended, 'the generator was ended'); + $this->assertSame([], glob("$target/.ngb-restore-spool-*")); + } + + public function testPlanJoinsBlobsThatFollowEachOtherInAPack(): void { + $plan = new ReadAheadPlan('/nowhere'); + $plan->add('packs/aa/a', 0, 100); + $plan->add('packs/aa/a', 100, 100); + $plan->add('packs/aa/a', 50, 10); // the same blob again: inside the range + $plan->add('packs/aa/a', 200 + ReadAheadPlan::GAP, 100); // a gap it reads along + $plan->add('packs/aa/a', 300 + 3 * ReadAheadPlan::GAP, 100); // too far: a new range + $plan->add('packs/bb/b', 0, 100); + $plan->add('packs/bb/b', 0, 100); // backwards is fine as long as it is inside + $plan->add('packs/aa/a', 0, 100); // back to an earlier pack: a new range + + $this->assertSame([ + ['packs/aa/a', 0, 300 + ReadAheadPlan::GAP], + ['packs/aa/a', 300 + 3 * ReadAheadPlan::GAP, 100], + ['packs/bb/b', 0, 100], + ['packs/aa/a', 0, 100], + ], $plan->ranges()); + } + + public function testPlanSkipsRangesWhoseBlobsWereReadElsewhere(): void { + $plan = new ReadAheadPlan('/nowhere'); + foreach (range(0, 4) as $p) { + $plan->add("packs/0$p/p$p", 0, 10); + } + $plan->readFrom((function (): \Generator { + foreach (range(0, 4) as $p) { + $fh = fopen('php://temp', 'w+b'); + fwrite($fh, str_repeat((string)$p, 10)); + rewind($fh); + yield $p => $fh; + } + })()); + + $this->assertSame('00000', $plan->read('packs/00/p0', 0, 5)); + $this->assertNull($plan->read('packs/77/moved', 0, 5), 'moved: not in the plan'); + $this->assertSame('333', $plan->read('packs/03/p3', 2, 3), 'the ranges in between were read elsewhere'); + $this->assertNull($plan->read('packs/01/p1', 0, 5), 'never backwards'); + $this->assertSame('44', $plan->read('packs/04/p4', 0, 2)); + } + + public function testPlanKeepsARangeShorterThanTheMaximum(): void { + $plan = new ReadAheadPlan('/nowhere'); + for ($o = 0; $o < 3 * ReadAheadPlan::MAX_RANGE; $o += 4 * 1048576) { + $plan->add('packs/aa/a', $o, 4 * 1048576); + } + $this->assertCount(3, $plan->ranges()); + foreach ($plan->ranges() as $r) { + $this->assertLessThanOrEqual(ReadAheadPlan::MAX_RANGE, $r[2]); + } + } +} + +/** Reads ahead from a local repository, one range after the other; can fail or damage ranges. */ +final class ReadAheadBackend implements IBackend, IReadsAhead { + public PackOpenCountingBackend $counter; + /** @var list */ + public array $asked = []; + public bool $ended = false; + + /** + * @param list $fail ranges given as null + * @param list $damage ranges given with a flipped byte in every blob + */ + public function __construct( + public LocalBackend $inner, + private array $fail = [], + private array $damage = [], + private bool $unavailable = false, + /** reading a pack the usual way fails */ + private bool $packsGone = false, + ) { + $this->counter = new PackOpenCountingBackend($inner); + } + + public function canReadAhead(): bool { + return true; + } + + public function readAhead(array $ranges, string $spool): ?\Generator { + if ($this->unavailable) { + return null; + } + \PHPUnit\Framework\Assert::assertDirectoryExists($spool); + $this->asked = $ranges; + return (function () use ($ranges): \Generator { + try { + foreach ($ranges as $i => [$path, $offset, $length]) { + if (in_array($i, $this->fail, true)) { + yield $i => null; + continue; + } + $data = $this->inner->getRange($path, $offset, $length); + if (in_array($i, $this->damage, true)) { + for ($k = 0; $k < strlen($data); $k += 4096) { + $data[$k] = chr(ord($data[$k]) ^ 1); + } + } + $fh = fopen('php://temp', 'w+b'); + fwrite($fh, $data); + rewind($fh); + yield $i => $fh; + fclose($fh); + } + } finally { + $this->ended = true; + } + })(); + } + + public function put(string $path, $stream): void { + $this->counter->put($path, $stream); + } + + public function get(string $path) { + if ($this->packsGone && str_starts_with($path, 'packs/')) { + throw new BackendException('Not found: ' . $path); + } + return $this->counter->get($path); + } + + public function getMany(array $paths): \Generator { + return $this->counter->getMany($paths); + } + + public function getRange(string $path, int $offset, int $length): string { + return $this->counter->getRange($path, $offset, $length); + } + + public function getRanges(array $ranges): \Generator { + return $this->counter->getRanges($ranges); + } + + public function exists(string $path): bool { + return $this->counter->exists($path); + } + + public function move(string $from, string $to): void { + $this->counter->move($from, $to); + } + + public function list(string $prefix): array { + return $this->counter->list($prefix); + } + + public function listFiles(string $dir): array { + return $this->counter->listFiles($dir); + } + + public function delete(string $path): void { + $this->counter->delete($path); + } +}