From 379b81db02c34a0c161a3aebda47fa73136832ae Mon Sep 17 00:00:00 2001 From: Dillard Blom Date: Sun, 11 Oct 2026 13:39:38 +0200 Subject: [PATCH] Verify: index files several at a time over SFTP, pack folders listed, progress stored and shown --- CHANGELOG.md | 11 ++ lib/Backend/GetsOneByOne.php | 30 +++++ lib/Backend/IBackend.php | 10 ++ lib/Backend/LocalBackend.php | 2 + lib/Backend/SftpReadStream.php | 157 ++++++++++++++++++++-- lib/Backend/StorageBackend.php | 29 ++++ lib/Command/Status.php | 6 + lib/Command/Verify.php | 10 +- lib/Repository/BlobIndex.php | 39 ++++-- lib/Repository/PackReaderGuard.php | 5 + lib/Repository/Repository.php | 70 ++++++++-- lib/Service/VerifyService.php | 132 ++++++++++++++++-- tests/Unit/Backend/SftpReadStreamTest.php | 44 ++++++ tests/Unit/Repository/PackReadTest.php | 2 + tests/Unit/Repository/VerifyTest.php | 106 +++++++++++++++ 15 files changed, 600 insertions(+), 53 deletions(-) create mode 100644 lib/Backend/GetsOneByOne.php diff --git a/CHANGELOG.md b/CHANGELOG.md index ca0c7d2..b93f283 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,7 +21,15 @@ instead carried by the app description, the git tag, the release, and this file. repository; stored passwords are never sent to the browser, and an empty password field keeps the stored one. Administrators are notified of the change. +- `occ backup:verify` reports how far it is every 30 seconds, and `occ backup:status` shows the + progress of a running check and how the last check of each location ended. On a large + repository a check takes hours, a deep one days. + ### Changed +- Checking and opening a large repository over SFTP is much faster: the index files (one per + pack, 27,547 for 1 TB) are downloaded 32 at a time instead of one after the other, 114 instead + of 6 files per second to a Hetzner Storage Box (about 4 minutes instead of 77 for 1 TB). A check + also lists the pack folders (256 requests) instead of asking for every pack. - App store listing: the author's e-mail address is shown. - App store listing: the homepage link points to the GitHub repository instead of back to the app store page itself. @@ -31,6 +39,9 @@ instead carried by the app description, the git tag, the release, and this file. SFTP reads sent their first request before phpseclib had opened the SFTP channel, which only happens on the first SFTP call after login. Every backup then failed with "No repository at this location". The channel is now opened first. +- A check (`occ backup:verify`) did not renew its hold on the location while it downloaded the + index; on a large repository that takes longer than the hold lasts, so a weekly cleanup could + have started during the check. - A long first backup could end as failed although its snapshot was complete: when its final step took longer than the step's lease (150 s with `occ backup:run`), the background job took the run over, finished it a second time and failed on the snapshot it had already recorded, diff --git a/lib/Backend/GetsOneByOne.php b/lib/Backend/GetsOneByOne.php new file mode 100644 index 0000000..8385260 --- /dev/null +++ b/lib/Backend/GetsOneByOne.php @@ -0,0 +1,30 @@ + $paths + * @return \Generator + */ + public function getMany(array $paths): \Generator { + foreach ($paths as $path) { + $fh = $this->get($path); + try { + $data = stream_get_contents($fh); + } finally { + fclose($fh); + } + if ($data === false) { + throw new BackendException('Read failed for ' . $path); + } + yield $path => $data; + } + } +} diff --git a/lib/Backend/IBackend.php b/lib/Backend/IBackend.php index 2d7eb6d..256e0d7 100644 --- a/lib/Backend/IBackend.php +++ b/lib/Backend/IBackend.php @@ -22,6 +22,16 @@ interface IBackend { /** @return resource readable stream of the whole object */ public function get(string $path); + /** + * Many small objects whole, in the order given (the repository's index files). A backend + * where every request costs a round trip can fetch them several at a time. + * + * @param list $paths + * @return \Generator contents by path + * @throws BackendException for an object that cannot be read + */ + public function getMany(array $paths): \Generator; + /** @return string bytes [$offset, $offset + $length) of the object (used to fetch single blobs from a pack) */ public function getRange(string $path, int $offset, int $length): string; diff --git a/lib/Backend/LocalBackend.php b/lib/Backend/LocalBackend.php index 1a577bb..266b517 100644 --- a/lib/Backend/LocalBackend.php +++ b/lib/Backend/LocalBackend.php @@ -8,6 +8,8 @@ declare(strict_types=1); namespace OCA\NgBackup\Backend; final class LocalBackend implements IBackend { + use GetsOneByOne; + public function __construct( private string $root, ) { diff --git a/lib/Backend/SftpReadStream.php b/lib/Backend/SftpReadStream.php index 9fa4a5a..c66dfc6 100644 --- a/lib/Backend/SftpReadStream.php +++ b/lib/Backend/SftpReadStream.php @@ -28,6 +28,8 @@ final class SftpReadStream { private const PROTOCOL = 'ngbsftpread'; /** Requests in flight once a stream reads sequentially; it starts at one and doubles. */ public const DEPTH = 8; + /** Files read at once by readMany(): each holds a handle on the server while its batch runs. */ + public const BATCH = 32; /** OpenSSH and the Hetzner Storage Box both answer at most 255 KiB per READ. */ private const CHUNK = 261120; /** Every SFTP server must handle packets of this size (draft-ietf-secsh-filexfer-02, section 3). */ @@ -101,20 +103,8 @@ final class SftpReadStream { return null; } try { - if (!is_callable([$storage, 'getConnection']) || !is_callable([$storage, 'getRoot'])) { - return null; - } - $sftp = $storage->getConnection(); - // is_callable and get_object_vars from here only see what is public: phpseclib 3 has - // methods of these names neither public nor at all. - if (!is_object($sftp) || !is_callable([$sftp, '_send_sftp_packet']) || !is_callable([$sftp, '_get_sftp_packet']) - || !is_callable([$sftp, '_realpath']) || !is_callable([$sftp, 'pwd']) - || !array_key_exists('packet_type', get_object_vars($sftp))) { - return null; - } - // phpseclib opens the SFTP channel on its first SFTP call, not at login: on a fresh - // connection a packet sent directly would go nowhere. pwd() opens it like any call does. - if ($sftp->pwd() === false) { + $sftp = self::connection($storage); + if ($sftp === null) { return null; } if (!in_array(self::PROTOCOL, stream_get_wrappers(), true)) { @@ -140,6 +130,145 @@ final class SftpReadStream { return $fh; } + /** + * The storage's phpseclib connection, with its SFTP channel open, when it offers what these + * streams need; null otherwise. + */ + private static function connection(object $storage): ?object { + 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']) || !is_callable([$sftp, 'pwd']) + || !array_key_exists('packet_type', get_object_vars($sftp))) { + return null; + } + // phpseclib opens the SFTP channel on its first SFTP call, not at login: on a fresh + // connection a packet sent directly would go nowhere. pwd() opens it like any call does. + if ($sftp->pwd() === false) { + return null; + } + return $sftp; + } + + /** + * Read many small files whole, BATCH at a time: the OPEN, READ and CLOSE requests of a batch + * all go out before their answers are read, so a batch costs a few round trips instead of + * four per file. Meant for files of up to a few hundred KiB (the repository's index files): + * every file is held in memory. + * + * @param list $paths below the storage's root + * @return array|null contents by path; null for a file this could not + * read (missing, or a server error), so the caller + * reads it the usual way; null for all when the + * connection does not offer what this needs + * @throws BackendException when the connection failed while requests were out: the storage's + * own stream must not be tried on it + */ + public static function readMany(object $storage, array $paths): ?array { + if (self::$failures >= self::MAX_FAILURES) { + return null; + } + try { + $sftp = self::connection($storage); + $root = $sftp === null ? '' : rtrim((string)$storage->getRoot(), '/'); + } catch (\Throwable) { + self::$failures++; + return null; + } + if ($sftp === null) { + return null; + } + $result = []; + foreach (array_chunk($paths, self::BATCH) as $batch) { + try { + $result += self::readBatch($sftp, $root, $batch); + } catch (\Throwable $e) { + self::$failures++; + throw $e instanceof BackendException ? $e : new BackendException('SFTP connection failed while reading ' . count($batch) . ' files: ' . $e->getMessage(), 0, $e); + } + } + self::$failures = 0; + return $result; + } + + /** Requests of one kind for a whole batch, then their answers in order. @return list */ + private static function exchange(object $sftp, int $type, array $packets): array { + $ids = []; + foreach ($packets as $packet) { + $id = self::nextId(); + if (!$sftp->_send_sftp_packet($type, $packet, $id)) { + // Requests already sent cannot be taken back: the connection is unusable now. + throw new BackendException('SFTP connection failed while sending a request'); + } + $ids[] = $id; + } + $answers = []; + foreach ($ids as $id) { + $response = $sftp->_get_sftp_packet($id); + if (!is_string($response)) { + throw new BackendException('SFTP connection failed while waiting for an answer'); + } + $answers[] = [$sftp->packet_type, $response]; + } + return $answers; + } + + /** @param list $batch @return array */ + private static function readBatch(object $sftp, string $root, array $batch): array { + $result = array_fill_keys($batch, null); + $opens = []; + foreach ($batch as $path) { + $remote = $sftp->_realpath($root . '/' . ltrim($path, '/')); + if (!is_string($remote)) { + throw new BackendException('SFTP path could not be resolved: ' . $path); + } + $opens[] = pack('Na*N2', strlen($remote), $remote, self::FXF_READ, 0); + } + $handles = []; // path => handle + foreach (self::exchange($sftp, self::FXP_OPEN, $opens) as $i => [$type, $response]) { + if ($type === self::FXP_HANDLE) { + $handles[$batch[$i]] = substr($response, 4); + } + // any other answer: not readable this way, the caller reads it the usual way + } + $data = array_fill_keys(array_keys($handles), ''); + $reading = $handles; + while ($reading !== []) { + $length = self::chunk($sftp); + $reads = []; + foreach ($reading as $path => $handle) { + $offset = strlen($data[$path]); + $reads[] = pack('Na*N3', strlen($handle), $handle, intdiv($offset, 4294967296), $offset % 4294967296, $length); + } + $paths = array_keys($reading); + foreach (self::exchange($sftp, self::FXP_READ, $reads) as $i => [$type, $response]) { + $path = $paths[$i]; + if ($type === self::FXP_DATA && strlen($response) > 4) { + $data[$path] .= substr($response, 4); // more may follow: read on from there + continue; + } + unset($reading[$path]); + $status = $type === self::FXP_STATUS ? unpack('N', $response) : false; + if (!is_array($status) || ($status[1] ?? null) !== self::FX_EOF) { + unset($data[$path]); // a server error on this file: read it the usual way + } + } + } + $closes = []; + foreach ($handles as $handle) { + $closes[] = pack('Na*', strlen($handle), $handle); + } + self::exchange($sftp, self::FXP_CLOSE, $closes); + foreach ($data as $path => $contents) { + $result[$path] = $contents; + } + return $result; + } + /** Test hook: forget the learned READ sizes and earlier failures. */ public static function reset(): void { self::$chunks = null; diff --git a/lib/Backend/StorageBackend.php b/lib/Backend/StorageBackend.php index ae6a8c1..0b4818e 100644 --- a/lib/Backend/StorageBackend.php +++ b/lib/Backend/StorageBackend.php @@ -98,6 +98,35 @@ final class StorageBackend implements IBackend { return $fh; } + /** Over SFTP several at a time (SftpReadStream::readMany()); elsewhere one after the other. */ + public function getMany(array $paths): \Generator { + foreach (array_chunk($paths, 8 * SftpReadStream::BATCH) as $chunk) { + $fast = null; + if ($this->sftpReads) { + $abs = array_map(fn (string $p): string => $this->abs($p), $chunk); + $fast = SftpReadStream::readMany($this->storage, $abs); + $fast = $fast === null ? null : array_combine($chunk, array_map(fn (string $a): ?string => $fast[$a] ?? null, $abs)); + } + foreach ($chunk as $path) { + $data = $fast[$path] ?? null; + if ($data === null) { + // not read in the batch (or no batches here): the usual way, which also + // reports a missing file as such + $fh = $this->get($path); + try { + $data = stream_get_contents($fh); + } finally { + fclose($fh); + } + if ($data === false) { + throw new BackendException('Read failed for ' . $path); + } + } + yield $path => $data; + } + } + } + public function getRange(string $path, int $offset, int $length): string { $fh = $this->get($path); try { diff --git a/lib/Command/Status.php b/lib/Command/Status.php index 9e637ac..d698e6b 100644 --- a/lib/Command/Status.php +++ b/lib/Command/Status.php @@ -10,6 +10,7 @@ namespace OCA\NgBackup\Command; use OCA\NgBackup\Db\RunMapper; use OCA\NgBackup\Service\KeyService; use OCA\NgBackup\Service\TargetService; +use OCA\NgBackup\Service\VerifyService; use Symfony\Component\Console\Command\Command; use Symfony\Component\Console\Helper\Table; use Symfony\Component\Console\Input\InputInterface; @@ -20,6 +21,7 @@ final class Status extends Command { private KeyService $keys, private TargetService $targets, private RunMapper $runs, + private VerifyService $verify, ) { parent::__construct(); } @@ -51,6 +53,10 @@ final class Status extends Command { if ($rows !== []) { (new Table($output))->setHeaders(['run', 'location', 'kind', 'status', 'started', 'took', 'files', 'uploaded', 'error'])->setRows($rows)->render(); } + foreach ($this->verify->progress() as $targetId => $state) { + $output->writeln(sprintf('Verify of %s on %s: %s', mb_strimwidth((string)($state['what'] ?? ''), 0, 40, '…'), + $names[$targetId] ?? '#' . $targetId, VerifyService::describe($state))); + } return 0; } diff --git a/lib/Command/Verify.php b/lib/Command/Verify.php index 37997ab..9480f83 100644 --- a/lib/Command/Verify.php +++ b/lib/Command/Verify.php @@ -37,6 +37,9 @@ blob is also downloaded and decrypted, which fails loudly on exactly that even i lists it -- the same verification an actual restore would do, run here deliberately instead of waiting to find out during one. +While it runs it prints how far it is every 30 seconds; occ backup:status shows the same, and +how the last check of each location ended. + Run this periodically (cron, picking a sample of recent snapshots) rather than only right before you'd need to rely on a restore. @@ -51,9 +54,12 @@ HELP); $target = $this->targets->get($input->getArgument('target')); $ref = (string)$input->getArgument('snapshot'); $deep = (bool)$input->getOption('deep'); + $onProgress = static function (array $state) use ($output): void { + $output->writeln(date('H:i:s') . ' ' . VerifyService::describe($state)); + }; $result = str_starts_with($ref, 'users/') - ? $this->verify->verifyUserExport($target, $ref, $deep) - : $this->verify->verify($target, $ref, $deep); + ? $this->verify->verifyUserExport($target, $ref, $deep, $onProgress) + : $this->verify->verify($target, $ref, $deep, $onProgress); } catch (\Throwable $e) { $output->writeln('' . $e->getMessage() . ''); return 1; diff --git a/lib/Repository/BlobIndex.php b/lib/Repository/BlobIndex.php index eb7269e..581ec4e 100644 --- a/lib/Repository/BlobIndex.php +++ b/lib/Repository/BlobIndex.php @@ -87,27 +87,37 @@ final class BlobIndex { * and only the index files that are not cached yet are downloaded. Cached packs that no * longer exist on the backend are dropped from the cache. */ - public function load(bool $verify = false): void { + public function load(bool $verify = false, ?callable $progress = null): void { $this->blobs = []; $cached = $verify ? [] : ($this->cache?->all($this->repositoryId) ?? []); $present = []; $this->downloaded = 0; + $fetch = []; + $entriesByPack = []; // in listing order: where a blob is in two packs, the last one listed wins foreach ($this->backend->listFiles('index') as $path) { $packId = basename($path); $present[$packId] = true; - if (isset($cached[$packId])) { - $entries = $cached[$packId]; - } else { - $fh = $this->backend->get($path); - $entries = json_decode($this->cipher->decryptString((string)stream_get_contents($fh), 'index:' . $packId), true, 512, JSON_THROW_ON_ERROR); - fclose($fh); - $this->downloaded++; - $this->cache?->put($this->repositoryId, $packId, $entries); + $entriesByPack[$packId] = $cached[$packId] ?? null; + if (!isset($cached[$packId])) { + $fetch[] = $path; } - foreach ($entries as $e) { - $this->blobs[$e['id']] = [$packId, $e['offset'], $e['length'], $e['raw'], $e['flags']]; + } + // The rest is downloaded several at a time where the location allows it: on a large + // repository these are tens of thousands of small files. + $total = count($fetch); + foreach ($this->backend->getMany($fetch) as $path => $data) { + $packId = basename($path); + $entries = json_decode($this->cipher->decryptString($data, 'index:' . $packId), true, 512, JSON_THROW_ON_ERROR); + $this->downloaded++; + $this->cache?->put($this->repositoryId, $packId, $entries); + $entriesByPack[$packId] = $entries; + if ($progress !== null) { + $progress($this->downloaded, $total); } } + foreach ($entriesByPack as $packId => $entries) { + $this->add($packId, $entries ?? []); + } if ($verify && $this->cache !== null) { // Refresh the cache from what was just verified. $this->cache->remove($this->repositoryId, array_keys($this->cache->all($this->repositoryId))); @@ -128,6 +138,13 @@ final class BlobIndex { } } + /** @param list $entries */ + private function add(string $packId, array $entries): void { + foreach ($entries as $e) { + $this->blobs[$e['id']] = [$packId, $e['offset'], $e['length'], $e['raw'], $e['flags']]; + } + } + public function has(string $id): bool { return isset($this->blobs[$id]) || isset($this->pending[$id]); } diff --git a/lib/Repository/PackReaderGuard.php b/lib/Repository/PackReaderGuard.php index 0004f4e..0679298 100644 --- a/lib/Repository/PackReaderGuard.php +++ b/lib/Repository/PackReaderGuard.php @@ -28,6 +28,11 @@ final class PackReaderGuard implements IBackend { return $this->inner->get($path); } + public function getMany(array $paths): \Generator { + $this->reader->close(); + return $this->inner->getMany($paths); + } + public function getRange(string $path, int $offset, int $length): string { $this->reader->close(); return $this->inner->getRange($path, $offset, $length); diff --git a/lib/Repository/Repository.php b/lib/Repository/Repository.php index 3c75cbf..9c90a9c 100644 --- a/lib/Repository/Repository.php +++ b/lib/Repository/Repository.php @@ -754,13 +754,22 @@ final class Repository { * callers use it to keep a time-boxed lease alive during a large audit * @return array{filesChecked:int, blobsChecked:int, bytesChecked:int, missing:list, failed:list, unreadable:list, readError:string} */ + /** + * @param ?callable(array{phase:string, done:int, total:?int, blobs?:int, bytes?:int}):void $heartbeat + * called often during the check, with how far it is: phase "index" (index files + * downloaded), "packs" (pack folders listed), "files" (files of the snapshot checked) + */ public function verify(string $snapshotId, bool $deep = false, ?callable $heartbeat = null): array { - $heartbeat ??= static function (): void { + $heartbeat ??= static function (array $progress): void { }; - $this->index->load(true); + $meta = $this->snapshotMeta($snapshotId); + $filesTotal = isset($meta['stats']['files']) ? (int)$meta['stats']['files'] : null; + $this->index->load(true, static function (int $done, int $total) use ($heartbeat): void { + $heartbeat(['phase' => 'index', 'done' => $done, 'total' => $total]); + }); $stats = ['filesChecked' => 0, 'blobsChecked' => 0, 'bytesChecked' => 0, 'missing' => [], 'failed' => [], 'unreadable' => [], 'readError' => '']; $seen = []; - $packSeen = []; // pack id => exists on the backend; avoids one exists() call per blob + [$packSeen, $listedDirs] = $this->presentPacks($heartbeat); foreach ($this->streamEntries($snapshotId) as $entry) { $stats['filesChecked']++; foreach ($entry['b'] as $id) { @@ -768,11 +777,11 @@ final class Repository { continue; } $seen[$id] = true; - $this->verifyBlob($id, $deep, $stats, $packSeen); - $heartbeat(); + $this->verifyBlob($id, $deep, $stats, $packSeen, $listedDirs); + $heartbeat(['phase' => 'files', 'done' => $stats['filesChecked'], 'total' => $filesTotal, + 'blobs' => $stats['blobsChecked'], 'bytes' => $stats['bytesChecked']]); } } - $meta = $this->snapshotMeta($snapshotId); if (isset($meta['db']) && is_string($meta['db'])) { if (!$this->backend->exists($meta['db'])) { $stats['missing'][] = $meta['db']; @@ -784,8 +793,9 @@ final class Repository { continue; } $seen[$id] = true; - $this->verifyBlob($id, $deep, $stats, $packSeen); - $heartbeat(); + $this->verifyBlob($id, $deep, $stats, $packSeen, $listedDirs); + $heartbeat(['phase' => 'database', 'done' => $stats['filesChecked'], 'total' => $filesTotal, + 'blobs' => $stats['blobsChecked'], 'bytes' => $stats['bytesChecked']]); } } } @@ -793,6 +803,30 @@ final class Repository { return $stats; } + /** + * Which packs are on the location, from one listing per pack folder instead of a request per + * pack. A folder that cannot be listed is left out: its packs are then checked one by one. + * + * @return array{0: array, 1: array} pack id => present, and the + * folders that were listed + */ + private function presentPacks(callable $heartbeat): array { + $present = []; + $listed = []; + for ($i = 0; $i < 256; $i++) { + $dir = sprintf('%02x', $i); + try { + foreach ($this->backend->listFiles('packs/' . $dir) as $path) { + $present[basename($path)] = true; + } + $listed[$dir] = true; + } catch (BackendException) { + } + $heartbeat(['phase' => 'packs', 'done' => $i + 1, 'total' => 256]); + } + return [$present, $listed]; + } + /** * Same checks as verify(), for a user export: every blob its manifest references must be in * the freshly re-authenticated index with its pack present; $deep also decrypts each. @@ -803,9 +837,11 @@ final class Repository { if (!$this->isRecordedUserExport($manifestPath)) { throw new RepositoryException("$manifestPath is not a recorded user export for this location"); } - $heartbeat ??= static function (): void { + $heartbeat ??= static function (array $progress): void { }; - $this->index->load(true); + $this->index->load(true, static function (int $done, int $total) use ($heartbeat): void { + $heartbeat(['phase' => 'index', 'done' => $done, 'total' => $total]); + }); $stats = ['filesChecked' => 0, 'blobsChecked' => 0, 'bytesChecked' => 0, 'missing' => [], 'failed' => [], 'unreadable' => [], 'readError' => '']; $seen = []; $packSeen = []; @@ -821,20 +857,28 @@ final class Repository { } $seen[$id] = true; $this->verifyBlob($id, $deep, $stats, $packSeen); - $heartbeat(); + $heartbeat(['phase' => 'files', 'done' => $stats['filesChecked'], 'total' => null, + 'blobs' => $stats['blobsChecked'], 'bytes' => $stats['bytesChecked']]); } } return $stats; } /** @param array $packSeen */ - private function verifyBlob(string $id, bool $deep, array &$stats, array &$packSeen): void { + /** + * @param array $packSeen pack id => present; filled in as packs are checked + * @param array $listedDirs pack folders whose listing is in $packSeen: a pack + * not in it is not there + */ + private function verifyBlob(string $id, bool $deep, array &$stats, array &$packSeen, array $listedDirs = []): void { if (!$this->index->has($id)) { $stats['missing'][] = $id; return; } [$pack] = $this->index->get($id); - $packExists = $packSeen[$pack] ??= $this->backend->exists('packs/' . substr($pack, 0, 2) . '/' . $pack); + $packExists = $packSeen[$pack] ??= isset($listedDirs[substr($pack, 0, 2)]) + ? false + : $this->backend->exists('packs/' . substr($pack, 0, 2) . '/' . $pack); if (!$packExists) { $stats['missing'][] = $id; return; diff --git a/lib/Service/VerifyService.php b/lib/Service/VerifyService.php index fcfd68e..d172427 100644 --- a/lib/Service/VerifyService.php +++ b/lib/Service/VerifyService.php @@ -7,42 +7,148 @@ declare(strict_types=1); namespace OCA\NgBackup\Service; +use OCA\NgBackup\AppInfo\Application; use OCA\NgBackup\Db\Target; +use OCP\IAppConfig; use OCP\Lock\LockedException; -/** Checksum audit of a snapshot (occ backup:verify), see Repository::verify(). */ +/** + * Checksum audit of a snapshot (occ backup:verify), see Repository::verify(). On a large + * repository a check takes hours (a deep one days), so how far it is gets stored every + * PROGRESS_SECONDS, together with how it ended: `occ backup:status` shows it. + */ final class VerifyService { + public const PROGRESS_SECONDS = 30; + private const KEY_PREFIX = 'verify_progress_'; + public function __construct( private TargetService $targets, private LeaseService $leases, + private IAppConfig $appConfig, ) { } - /** @return array{filesChecked:int, blobsChecked:int, bytesChecked:int, missing:list, failed:list, unreadable:list, readError:string} */ - public function verify(Target $target, string $snapshotId, bool $deep = false): array { - return $this->run($target, $deep, fn (\OCA\NgBackup\Repository\Repository $repo, callable $heartbeat): array => $repo->verify($snapshotId, $deep, $heartbeat)); + /** + * @param ?callable(array):void $onProgress called with the stored progress every PROGRESS_SECONDS + * @return array{filesChecked:int, blobsChecked:int, bytesChecked:int, missing:list, failed:list, unreadable:list, readError:string} + */ + public function verify(Target $target, string $snapshotId, bool $deep = false, ?callable $onProgress = null): array { + return $this->run($target, $snapshotId, $deep, $onProgress, fn (\OCA\NgBackup\Repository\Repository $repo, callable $heartbeat): array => $repo->verify($snapshotId, $deep, $heartbeat)); } - /** @return array{filesChecked:int, blobsChecked:int, bytesChecked:int, missing:list, failed:list, unreadable:list, readError:string} */ - public function verifyUserExport(Target $target, string $manifestPath, bool $deep = false): array { - return $this->run($target, $deep, fn (\OCA\NgBackup\Repository\Repository $repo, callable $heartbeat): array => $repo->verifyUserExport($manifestPath, $deep, $heartbeat)); + /** + * @param ?callable(array):void $onProgress called with the stored progress every PROGRESS_SECONDS + * @return array{filesChecked:int, blobsChecked:int, bytesChecked:int, missing:list, failed:list, unreadable:list, readError:string} + */ + public function verifyUserExport(Target $target, string $manifestPath, bool $deep = false, ?callable $onProgress = null): array { + return $this->run($target, $manifestPath, $deep, $onProgress, fn (\OCA\NgBackup\Repository\Repository $repo, callable $heartbeat): array => $repo->verifyUserExport($manifestPath, $deep, $heartbeat)); } - private function run(Target $target, bool $deep, callable $fn): array { + /** One line on a check's progress or outcome, for occ (see progress()). */ + public static function describe(array $s): string { + $size = static fn (int $b): string => $b >= 1073741824 ? sprintf('%.1f GiB', (float)$b / 1073741824.0) : sprintf('%.1f MiB', (float)$b / 1048576.0); + $count = static fn (int $done, mixed $total): string => number_format($done) + . (is_int($total) && $total > 0 ? '/' . number_format($total) . sprintf(' (%d%%)', intdiv(100 * $done, $total)) : ''); + $deep = ($s['deep'] ?? false) === true; + $kind = $deep ? 'deep check' : 'check'; + if (($s['status'] ?? '') === 'running') { + $phase = match ($s['phase'] ?? '') { + 'index' => 'index files ' . $count((int)$s['done'], $s['total'] ?? null), + 'packs' => 'pack folders ' . $count((int)$s['done'], $s['total'] ?? null), + 'files', 'database' => 'files ' . $count((int)$s['done'], $s['total'] ?? null) . ', ' . number_format((int)($s['blobs'] ?? 0)) . ' blobs' + . ($deep ? ', ' . $size((int)($s['bytes'] ?? 0)) . ' decrypted' : ''), + default => 'starting', + }; + $age = time() - (int)($s['updated'] ?? 0); + return sprintf('%s running since %s: %s; updated %s ago%s', $kind, date('Y-m-d H:i', (int)$s['started']), $phase, + $age < 120 ? $age . ' s' : intdiv($age, 60) . ' min', + $age > 600 ? ' (no progress for a while: the process may have stopped)' : ''); + } + $took = gmdate('H:i:s', max(0, (int)($s['finished'] ?? 0) - (int)($s['started'] ?? 0))); + return match ($s['status'] ?? '') { + 'ok' => sprintf('%s OK on %s (took %s): %s files, %s blobs%s', $kind, date('Y-m-d H:i', (int)$s['finished']), $took, + number_format((int)($s['files'] ?? 0)), number_format((int)($s['blobs'] ?? 0)), $deep ? ', ' . $size((int)($s['bytes'] ?? 0)) . ' decrypted' : ''), + 'problems' => sprintf('%s found PROBLEMS on %s: %d missing, %d failed, %d unreadable blob(s); run occ backup:verify again for the details', + $kind, date('Y-m-d H:i', (int)$s['finished']), (int)($s['missing'] ?? 0), (int)($s['failedBlobs'] ?? 0), (int)($s['unreadable'] ?? 0)), + default => sprintf('%s FAILED on %s: %s', $kind, date('Y-m-d H:i', (int)($s['finished'] ?? 0)), (string)($s['error'] ?? 'unknown error')), + }; + } + + /** + * The stored state of the last check per location: running (with how far it is) or how it + * ended. + * + * @return array> by location id + */ + public function progress(): array { + $result = []; + foreach ($this->appConfig->getKeys(Application::APP_ID) as $key) { + if (!str_starts_with($key, self::KEY_PREFIX)) { + continue; + } + $state = json_decode($this->appConfig->getValueString(Application::APP_ID, $key, '', lazy: true), true); + if (is_array($state)) { + $result[(int)substr($key, strlen(self::KEY_PREFIX))] = $state; + } + } + ksort($result); + return $result; + } + + private function run(Target $target, string $what, bool $deep, ?callable $onProgress, callable $fn): array { $ttl = $deep ? 600 : 300; + $state = ['what' => $what, 'deep' => $deep, 'status' => 'running', 'started' => time(), 'updated' => time(), + 'pid' => getmypid(), 'phase' => 'starting', 'done' => 0, 'total' => null, 'blobs' => 0, 'bytes' => 0]; + $this->store($target, $state); try { - return $this->leases->with(PruneService::lockKey($target), LeaseService::SHARED, $ttl, function (callable $refresh) use ($target, $deep, $ttl, $fn): array { - $last = time(); - $heartbeat = function () use ($refresh, $ttl, &$last): void { - if (time() - $last >= intdiv($ttl, 2)) { + $result = $this->leases->with(PruneService::lockKey($target), LeaseService::SHARED, $ttl, function (callable $refresh) use ($target, $ttl, $fn, $onProgress, &$state): array { + $lastRefresh = time(); + $lastStore = time(); + $heartbeat = function (?array $progress = null) use ($refresh, $ttl, $target, $onProgress, &$lastRefresh, &$lastStore, &$state): void { + $now = time(); + if ($now - $lastRefresh >= intdiv($ttl, 2)) { $refresh(); - $last = time(); + $lastRefresh = $now; + } + if ($progress !== null && $now - $lastStore >= self::PROGRESS_SECONDS) { + $state = array_merge($state, $progress, ['updated' => $now]); + try { + $this->store($target, $state); + } catch (\Throwable) { + // a progress report that cannot be stored must not stop the check + } + $lastStore = $now; + if ($onProgress !== null) { + $onProgress($state); + } } }; return $fn($this->targets->repository($target), $heartbeat); }); } catch (LockedException) { + $this->finish($target, $state, 'failed', ['error' => 'A cleanup of the location was in progress']); throw new \RuntimeException('A cleanup of ' . $target->getName() . ' is in progress; try again later'); + } catch (\Throwable $e) { + $this->finish($target, $state, 'failed', ['error' => $e->getMessage()]); + throw $e; + } + $problems = count($result['missing']) + count($result['failed']) + count($result['unreadable']); + $this->finish($target, $state, $problems === 0 ? 'ok' : 'problems', [ + 'phase' => 'done', 'files' => $result['filesChecked'], 'blobs' => $result['blobsChecked'], 'bytes' => $result['bytesChecked'], + 'missing' => count($result['missing']), 'failedBlobs' => count($result['failed']), 'unreadable' => count($result['unreadable']), + ]); + return $result; + } + + private function finish(Target $target, array $state, string $status, array $extra): void { + try { + $this->store($target, array_merge($state, $extra, ['status' => $status, 'updated' => time(), 'finished' => time()])); + } catch (\Throwable) { + // the check's own result counts; its stored summary is a convenience } } + + private function store(Target $target, array $state): void { + $this->appConfig->setValueString(Application::APP_ID, self::KEY_PREFIX . $target->getId(), json_encode($state, JSON_THROW_ON_ERROR), lazy: true); + } } diff --git a/tests/Unit/Backend/SftpReadStreamTest.php b/tests/Unit/Backend/SftpReadStreamTest.php index b0a6279..98af084 100644 --- a/tests/Unit/Backend/SftpReadStreamTest.php +++ b/tests/Unit/Backend/SftpReadStreamTest.php @@ -231,6 +231,50 @@ class SftpReadStreamTest extends TestCase { $this->assertNull(SftpReadStream::open($storage, 'f')); } + public function testReadManyReadsSmallFilesInBatches(): void { + $files = []; + for ($i = 0; $i < 70; $i++) { + $files["/home/index/$i"] = random_bytes(random_int(0, 5000)); + } + $files['/home/index/big'] = random_bytes(600000); // more than one READ + $sftp = new FakeSftpConnection($files); + $paths = array_map(fn ($p) => substr($p, 6), array_keys($files)); + + $got = SftpReadStream::readMany(new FakeSftpStorage($sftp), $paths); + + $this->assertSame(array_combine($paths, array_values($files)), $got); + $this->assertSame(SftpReadStream::BATCH, $sftp->maxInFlight, 'a batch goes out at once'); + $this->assertSame(0, $sftp->pending()); + $this->assertSame([], $sftp->openHandles); + } + + public function testReadManyLeavesAMissingOrFailingFileToTheCaller(): void { + $sftp = new FakeSftpConnection(['/home/a' => 'aaa', '/home/b' => random_bytes(400000), '/home/c' => 'ccc'], failAt: 300000); + + $got = SftpReadStream::readMany(new FakeSftpStorage($sftp), ['a', 'missing', 'b', 'c']); + + $this->assertSame(['a' => 'aaa', 'missing' => null, 'b' => null, 'c' => 'ccc'], $got); + $this->assertSame(0, $sftp->pending()); + $this->assertSame([], $sftp->openHandles); + } + + public function testReadManyAcrossACappedServer(): void { + $data = random_bytes(300000); + $sftp = new FakeSftpConnection(['/home/a' => $data, '/home/b' => 'b'], cap: 100000); + $this->assertSame(['a' => $data, 'b' => 'b'], SftpReadStream::readMany(new FakeSftpStorage($sftp), ['a', 'b'])); + } + + public function testReadManyFailsOnAConnectionError(): void { + $sftp = new FakeSftpConnection(['/home/a' => random_bytes(10), '/home/b' => 'b'], loseAt: 0); + $this->expectException(BackendException::class); + SftpReadStream::readMany(new FakeSftpStorage($sftp), ['a', 'b']); + } + + public function testReadManyOnAnUnsupportedConnectionIsNull(): void { + $this->assertNull(SftpReadStream::readMany(new \stdClass(), ['a'])); + $this->assertNull(SftpReadStream::readMany(new FakeSftpStorage(new FakeSftpConnection(['/home/a' => 'a'], channelFails: true)), ['a'])); + } + public function testAFreshConnectionHasItsSftpChannelOpenedFirst(): void { // phpseclib 2 logs in at connect time but opens the SFTP channel on the first SFTP call. $sftp = new FakeSftpConnection(['/home/config' => 'repository config']); diff --git a/tests/Unit/Repository/PackReadTest.php b/tests/Unit/Repository/PackReadTest.php index 70a1f30..d3a8670 100644 --- a/tests/Unit/Repository/PackReadTest.php +++ b/tests/Unit/Repository/PackReadTest.php @@ -106,6 +106,8 @@ class PackReadTest extends TestCase { } final class PackOpenCountingBackend implements IBackend { + use \OCA\NgBackup\Backend\GetsOneByOne; + /** @var list */ public array $gets = []; /** @var list */ diff --git a/tests/Unit/Repository/VerifyTest.php b/tests/Unit/Repository/VerifyTest.php index c8fcb17..d34a26b 100644 --- a/tests/Unit/Repository/VerifyTest.php +++ b/tests/Unit/Repository/VerifyTest.php @@ -78,6 +78,59 @@ class VerifyTest extends TestCase { $this->assertSame(['db/does-not-exist'], $result['missing']); } + public function testVerifyReportsHowFarItIs(): void { + $repo = Repository::initWithKey($this->backend, $this->keys, $this->keys->wrap('pw')); + $s = $repo->backupDirectory($this->src); + $seen = []; + + $this->open()->verify($s['snapshot'], heartbeat: function (array $p) use (&$seen): void { + $seen[] = $p; + }); + + $phases = array_values(array_unique(array_column($seen, 'phase'))); + $this->assertSame(['index', 'packs', 'files'], $phases); + $this->assertSame(['phase' => 'packs', 'done' => 256, 'total' => 256], $seen[array_key_last(array_filter($seen, fn ($p) => $p['phase'] === 'packs'))]); + $last = end($seen); + $this->assertSame(2, $last['done']); + $this->assertNull($last['total'], 'this snapshot has no file count in its metadata (a backup run records one)'); + } + + public function testVerifyListsPackFoldersInsteadOfCheckingEveryPack(): void { + $repo = Repository::initWithKey($this->backend, $this->keys, $this->keys->wrap('pw')); + $s = $repo->backupDirectory($this->src); + $counting = new ListingBackend($this->backend); + + $result = Repository::openWithKey($counting, $this->keys, null, null)->verify($s['snapshot']); + + $this->assertSame([], $result['missing']); + $this->assertSame(0, $counting->packExists); + } + + public function testAPackFolderThatCannotBeListedIsCheckedPackByPack(): void { + $repo = Repository::initWithKey($this->backend, $this->keys, $this->keys->wrap('pw')); + $s = $repo->backupDirectory($this->src); + $counting = new ListingBackend($this->backend, unlistable: true); + + $result = Repository::openWithKey($counting, $this->keys, null, null)->verify($s['snapshot']); + + $this->assertSame([], $result['missing']); + $this->assertGreaterThan(0, $counting->packExists); + } + + public function testAPackMissingFromItsFolderListingIsReported(): void { + $repo = Repository::initWithKey($this->backend, $this->keys, $this->keys->wrap('pw')); + $s = $repo->backupDirectory($this->src); + foreach ($this->backend->list('packs') as $pack) { + $this->backend->delete($pack); + } + $counting = new ListingBackend($this->backend); + + $result = Repository::openWithKey($counting, $this->keys, null, null)->verify($s['snapshot']); + + $this->assertNotSame([], $result['missing']); + $this->assertSame(0, $counting->packExists); + } + public function testDeepVerifyDetectsATamperedPack(): void { $repo = Repository::initWithKey($this->backend, $this->keys, $this->keys->wrap('pw')); $s = $repo->backupDirectory($this->src); @@ -131,6 +184,8 @@ class VerifyTest extends TestCase { /** Delegates to a real backend, but reading a pack throws for the first $failures reads. */ final class FlakyRangeBackend implements IBackend { + use \OCA\NgBackup\Backend\GetsOneByOne; + public function __construct( private IBackend $inner, private int $failures, @@ -181,3 +236,54 @@ final class FlakyRangeBackend implements IBackend { $this->inner->delete($path); } } + +/** Delegates to a real backend; counts exists() on packs, and can refuse to list pack folders. */ +final class ListingBackend implements IBackend { + use \OCA\NgBackup\Backend\GetsOneByOne; + + public int $packExists = 0; + + public function __construct( + private IBackend $inner, + private bool $unlistable = false, + ) { + } + + public function put(string $path, $stream): void { + $this->inner->put($path, $stream); + } + + public function get(string $path) { + return $this->inner->get($path); + } + + public function getRange(string $path, int $offset, int $length): string { + return $this->inner->getRange($path, $offset, $length); + } + + public function exists(string $path): bool { + if (str_starts_with($path, 'packs/')) { + $this->packExists++; + } + 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 { + if ($this->unlistable && str_starts_with($dir, 'packs/')) { + throw new BackendException('Cannot list ' . $dir); + } + return $this->inner->listFiles($dir); + } + + public function delete(string $path): void { + $this->inner->delete($path); + } +}