Verify and batched SFTP reads: address review notes

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<array{int, string|false}> */
+	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<array{int, string}> */
 	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<string> $batch @return array<string, string|null> */
+	/**
+	 * @param list<string> $batch
+	 * @return array{0: array<string, string>, 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. */

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.

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<string>, failed:list<string>, unreadable:list<string>, 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<string>, failed:list<string>, unreadable:list<string>, 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<string, bool>, 1: array<string, true>} 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<string, bool> $packSeen */
 	/**
 	 * @param array<string, bool> $packSeen pack id => present; filled in as packs are checked
 	 * @param array<string, true> $listedDirs pack folders whose listing is in $packSeen: a pack

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();

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;

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);
 	}