diff --git a/lib/Backend/SftpReadStream.php b/lib/Backend/SftpReadStream.php index c66dfc6..99ee417 100644 --- a/lib/Backend/SftpReadStream.php +++ b/lib/Backend/SftpReadStream.php @@ -30,6 +30,8 @@ final class SftpReadStream { 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; + /** Larger files are left to the usual read by readMany(), which holds every file in memory. */ + private const MANY_MAX_BYTES = 16 * 1048576; /** 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). */ @@ -42,6 +44,7 @@ final class SftpReadStream { private const FXP_HANDLE = 102; private const FXP_DATA = 103; private const FXF_READ = 1; + private const FX_OK = 0; private const FX_EOF = 1; /** @@ -182,26 +185,45 @@ final class SftpReadStream { if ($sftp === null) { return null; } - $result = []; + $result = array_fill_keys($paths, null); foreach (array_chunk($paths, self::BATCH) as $batch) { try { - $result += self::readBatch($sftp, $root, $batch); + [$read, $closed] = self::readBatch($sftp, $root, $batch); } catch (\Throwable $e) { self::$failures++; + // Answers to requests already sent may still arrive, and files_external would take + // them for its own: end the connection rather than leave it out of step. + self::disconnect($sftp); throw $e instanceof BackendException ? $e : new BackendException('SFTP connection failed while reading ' . count($batch) . ' files: ' . $e->getMessage(), 0, $e); } + foreach ($read as $path => $contents) { + $result[$path] = $contents; + } + if (!$closed) { + // The server did not confirm closing these files: no more batches, so handles + // cannot pile up. The rest is read the usual way, one at a time. + break; + } } self::$failures = 0; return $result; } - /** Requests of one kind for a whole batch, then their answers in order. @return list */ + private static function disconnect(object $sftp): void { + try { + if (is_callable([$sftp, 'disconnect'])) { + $sftp->disconnect(); + } + } catch (\Throwable) { + } + } + + /** 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; @@ -217,23 +239,28 @@ final class SftpReadStream { return $answers; } - /** @param list $batch @return array */ + /** + * @param list $batch + * @return array{0: array, 1: bool} the files read whole, and whether the + * server confirmed closing every handle + */ private static function readBatch(object $sftp, string $root, array $batch): array { - $result = array_fill_keys($batch, null); $opens = []; + $openPaths = []; foreach ($batch as $path) { $remote = $sftp->_realpath($root . '/' . ltrim($path, '/')); - if (!is_string($remote)) { - throw new BackendException('SFTP path could not be resolved: ' . $path); + if (is_string($remote)) { + $opens[] = pack('Na*N2', strlen($remote), $remote, self::FXF_READ, 0); + $openPaths[] = $path; } - $opens[] = pack('Na*N2', strlen($remote), $remote, self::FXF_READ, 0); + // a path that cannot be resolved is left to the caller's usual read } $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); + $handles[$openPaths[$i]] = substr($response, 4); } - // any other answer: not readable this way, the caller reads it the usual way + // any other answer (missing, no permission): the caller reads it the usual way } $data = array_fill_keys(array_keys($handles), ''); $reading = $handles; @@ -247,14 +274,18 @@ final class SftpReadStream { $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) { + $declared = strlen($response) >= 4 ? unpack('N', $response) : false; + if ($type === self::FXP_DATA && is_array($declared) && $declared[1] > 0 && $declared[1] === strlen($response) - 4 + && $declared[1] <= $length && strlen($data[$path]) + $declared[1] <= self::MANY_MAX_BYTES) { $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 + $status = $type === self::FXP_STATUS && is_array($declared) ? $declared[1] : null; + if ($status !== self::FX_EOF) { + // a server error, a malformed answer or a file too large for this: the caller + // reads it the usual way + unset($data[$path]); } } } @@ -262,11 +293,14 @@ final class SftpReadStream { 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; + $closed = true; + foreach (self::exchange($sftp, self::FXP_CLOSE, $closes) as [$type, $response]) { + $status = $type === self::FXP_STATUS && strlen($response) >= 4 ? unpack('N', $response) : false; + if (!is_array($status) || $status[1] !== self::FX_OK) { + $closed = false; + } } - return $result; + return [$data, $closed]; } /** Test hook: forget the learned READ sizes and earlier failures. */ diff --git a/lib/Repository/BlobIndex.php b/lib/Repository/BlobIndex.php index 581ec4e..a71faa6 100644 --- a/lib/Repository/BlobIndex.php +++ b/lib/Repository/BlobIndex.php @@ -92,31 +92,37 @@ final class BlobIndex { $cached = $verify ? [] : ($this->cache?->all($this->repositoryId) ?? []); $present = []; $this->downloaded = 0; + $order = []; // pack ids in listing order: where a blob is in two packs, the last listed wins $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; - $entriesByPack[$packId] = $cached[$packId] ?? null; + $order[] = $packId; if (!isset($cached[$packId])) { $fetch[] = $path; } } // 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. + // repository these are tens of thousands of small files. getMany() keeps their order, so + // the cached packs listed before each one are added first. + $next = 0; $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; + for (; $next < count($order) && $order[$next] !== $packId; $next++) { + $this->add($order[$next], $cached[$order[$next]] ?? []); + } + $this->add($packId, $entries); + $next++; if ($progress !== null) { $progress($this->downloaded, $total); } } - foreach ($entriesByPack as $packId => $entries) { - $this->add($packId, $entries ?? []); + for (; $next < count($order); $next++) { + $this->add($order[$next], $cached[$order[$next]] ?? []); } if ($verify && $this->cache !== null) { // Refresh the cache from what was just verified. diff --git a/lib/Repository/Repository.php b/lib/Repository/Repository.php index 9c90a9c..5298596 100644 --- a/lib/Repository/Repository.php +++ b/lib/Repository/Repository.php @@ -750,14 +750,11 @@ final class Repository { * referenced by more than one file is only checked once; a missing database manifest is a * failure, not something to silently skip over. * - * @param ?callable $heartbeat called after every blob (shallow) or every blob read (deep); - * 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) + * downloaded), "packs" (pack folders listed), "files" and "database" (files of the + * snapshot checked); callers use it to keep a time-boxed lease alive and to report + * @return array{filesChecked:int, blobsChecked:int, bytesChecked:int, missing:list, failed:list, unreadable:list, readError:string} */ public function verify(string $snapshotId, bool $deep = false, ?callable $heartbeat = null): array { $heartbeat ??= static function (array $progress): void { @@ -805,7 +802,8 @@ final class Repository { /** * 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. + * pack. A folder that cannot be listed, or comes back empty (which a storage also reports when + * it could not look), 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 @@ -816,11 +814,15 @@ final class Repository { 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; + $files = $this->backend->listFiles('packs/' . $dir); } catch (BackendException) { + $files = []; + } + foreach ($files as $path) { + $present[basename($path)] = true; + } + if ($files !== []) { + $listed[$dir] = true; } $heartbeat(['phase' => 'packs', 'done' => $i + 1, 'total' => 256]); } @@ -864,7 +866,6 @@ final class Repository { return $stats; } - /** @param array $packSeen */ /** * @param array $packSeen pack id => present; filled in as packs are checked * @param array $listedDirs pack folders whose listing is in $packSeen: a pack diff --git a/lib/Service/VerifyService.php b/lib/Service/VerifyService.php index d172427..8d7486a 100644 --- a/lib/Service/VerifyService.php +++ b/lib/Service/VerifyService.php @@ -99,7 +99,11 @@ final class VerifyService { $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 { + $this->store($target, $state); + } catch (\Throwable) { + // progress reports are a convenience; the check itself goes ahead + } try { $result = $this->leases->with(PruneService::lockKey($target), LeaseService::SHARED, $ttl, function (callable $refresh) use ($target, $ttl, $fn, $onProgress, &$state): array { $lastRefresh = time(); diff --git a/tests/Unit/Backend/SftpReadStreamTest.php b/tests/Unit/Backend/SftpReadStreamTest.php index 98af084..2542b53 100644 --- a/tests/Unit/Backend/SftpReadStreamTest.php +++ b/tests/Unit/Backend/SftpReadStreamTest.php @@ -264,10 +264,34 @@ class SftpReadStreamTest extends TestCase { $this->assertSame(['a' => $data, 'b' => 'b'], SftpReadStream::readMany(new FakeSftpStorage($sftp), ['a', 'b'])); } - public function testReadManyFailsOnAConnectionError(): void { + public function testReadManyEndsTheConnectionOnAConnectionError(): void { + // Answers to requests already out may still come: files_external must not get them. $sftp = new FakeSftpConnection(['/home/a' => random_bytes(10), '/home/b' => 'b'], loseAt: 0); - $this->expectException(BackendException::class); - SftpReadStream::readMany(new FakeSftpStorage($sftp), ['a', 'b']); + try { + SftpReadStream::readMany(new FakeSftpStorage($sftp), ['a', 'b']); + $this->fail('expected a BackendException'); + } catch (BackendException) { + } + $this->assertTrue($sftp->disconnected); + } + + public function testReadManyStopsBatchingWhenCloseIsNotConfirmed(): void { + $files = []; + for ($i = 0; $i < 40; $i++) { + $files["/home/$i"] = "file $i"; + } + $sftp = new FakeSftpConnection($files, closeStatus: 4); + $got = SftpReadStream::readMany(new FakeSftpStorage($sftp), array_map('strval', range(0, 39))); + $this->assertSame('file 0', $got['0']); + $this->assertSame('file 31', $got['31']); + $this->assertNull($got['32'], 'no second batch after unconfirmed closes'); + $this->assertCount(SftpReadStream::BATCH, array_filter($got, fn ($v) => $v !== null)); + } + + public function testReadManyLeavesAMalformedAnswerToTheCaller(): void { + $sftp = new FakeSftpConnection(['/home/a' => 'aaaa', '/home/b' => 'bbbb'], malformed: '/home/a'); + $this->assertSame(['a' => null, 'b' => 'bbbb'], SftpReadStream::readMany(new FakeSftpStorage($sftp), ['a', 'b'])); + $this->assertSame([], $sftp->openHandles); } public function testReadManyOnAnUnsupportedConnectionIsNull(): void { @@ -458,9 +482,19 @@ final class FakeSftpConnection { private ?int $loseAt = null, private bool $closeFails = false, private bool $channelFails = false, + /** status code the server answers CLOSE with */ + private int $closeStatus = 0, + /** a file whose DATA answers declare a wrong length */ + private ?string $malformed = null, ) { } + public bool $disconnected = false; + + public function disconnect(): void { + $this->disconnected = true; + } + public function pwd(): string|false { if ($this->channelFails) { return false; @@ -509,7 +543,7 @@ final class FakeSftpConnection { } $handle = substr($data, 4, unpack('N', $data)[1]); unset($this->openHandles[$handle]); - $this->answers[$id] = [101, pack('N', 0)]; + $this->answers[$id] = [101, pack('N', $this->closeStatus)]; break; case 5: // READ $len = unpack('N', $data)[1]; @@ -531,7 +565,8 @@ final class FakeSftpConnection { $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)]; + $declared = $this->openHandles[$handle] === $this->malformed ? strlen($chunk) + 7 : strlen($chunk); + $this->answers[$id] = [103, pack('Na*', $declared, $chunk)]; } $this->maxInFlight = max($this->maxInFlight, count($this->answers)); break; diff --git a/tests/Unit/Repository/VerifyTest.php b/tests/Unit/Repository/VerifyTest.php index d34a26b..c210739 100644 --- a/tests/Unit/Repository/VerifyTest.php +++ b/tests/Unit/Repository/VerifyTest.php @@ -117,10 +117,25 @@ class VerifyTest extends TestCase { $this->assertGreaterThan(0, $counting->packExists); } - public function testAPackMissingFromItsFolderListingIsReported(): void { + public function testAnEmptyFolderListingIsNoProofThatItsPacksAreGone(): void { + // A storage also reports an empty folder when it could not look (is_dir() false). $repo = Repository::initWithKey($this->backend, $this->keys, $this->keys->wrap('pw')); $s = $repo->backupDirectory($this->src); - foreach ($this->backend->list('packs') as $pack) { + $counting = new ListingBackend($this->backend, emptyListings: true); + + $result = Repository::openWithKey($counting, $this->keys, null, null)->verify($s['snapshot']); + + $this->assertSame([], $result['missing']); + $this->assertGreaterThan(0, $counting->packExists); + } + + public function testAPackMissingFromAFolderThatIsNotEmptyIsReportedWithoutAskingForIt(): void { + $repo = Repository::initWithKey($this->backend, $this->keys, $this->keys->wrap('pw')); + $s = $repo->backupDirectory($this->src); + $packs = $this->backend->list('packs'); + // Another file in the same folder keeps its listing non-empty. + $this->backend->put(dirname($packs[0]) . '/0000other', fopen('data://text/plain,x', 'r')); + foreach ($packs as $pack) { $this->backend->delete($pack); } $counting = new ListingBackend($this->backend); @@ -131,6 +146,18 @@ class VerifyTest extends TestCase { $this->assertSame(0, $counting->packExists); } + public function testAPackMissingFromAnEmptyFolderIsReportedToo(): 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); + } + + $result = Repository::openWithKey(new ListingBackend($this->backend), $this->keys, null, null)->verify($s['snapshot']); + + $this->assertNotSame([], $result['missing']); + } + public function testDeepVerifyDetectsATamperedPack(): void { $repo = Repository::initWithKey($this->backend, $this->keys, $this->keys->wrap('pw')); $s = $repo->backupDirectory($this->src); @@ -246,6 +273,7 @@ final class ListingBackend implements IBackend { public function __construct( private IBackend $inner, private bool $unlistable = false, + private bool $emptyListings = false, ) { } @@ -280,6 +308,9 @@ final class ListingBackend implements IBackend { if ($this->unlistable && str_starts_with($dir, 'packs/')) { throw new BackendException('Cannot list ' . $dir); } + if ($this->emptyListings && str_starts_with($dir, 'packs/')) { + return []; + } return $this->inner->listFiles($dir); }