SFTP: replace a dropped connection; write the index cache in one transaction

A Hetzner Storage Box drops an SFTP connection that carried a lot of data
and then stayed quiet for a few minutes; files_external keeps using the dead
client, on which every file looks missing. StorageBackend now checks the
connection after a minute without requests, and replaces a connection that
is dead after a failed request (once per request; put, move and delete are
not repeated). SftpReadStream closes a connection it left in an unknown state.

The verify cache refresh took more than 6 minutes on PostgreSQL for 27,547
packs, one autocommitted insert per pack: putMany()/replaceAll() write in one
transaction. After a verify the cache holds the index files as they are, so a
blob in two packs stays cached for both.

lib/Backend/SftpReadStream.php

@@ -362,6 +362,8 @@ final class SftpReadStream {
 		self::$failures++;
 		self::$openBroke = true;
 		self::$openCause = $cause . self::connectionState($this->sftp);
+		// unknown state: closed, so that nothing else uses it (and the backend can build a new one)
+		self::disconnect($this->sftp);
 		return false;
 	}
 
@@ -514,6 +516,9 @@ final class SftpReadStream {
 			} catch (\Throwable) {
 				$this->broken = true;
 			}
+			if ($this->broken) {
+				self::disconnect($this->sftp);
+			}
 		}
 		$this->handle = '';
 	}
@@ -595,19 +600,20 @@ final class SftpReadStream {
 	private function drain(): void {
 		$inFlight = $this->inFlight;
 		$this->inFlight = [];
-		if ($this->broken) {
-			return;
-		}
 		try {
-			foreach ($inFlight as [$id]) {
+			foreach ($this->broken ? [] : $inFlight as [$id]) {
 				if (!is_string($this->sftp->_get_sftp_packet($id))) {
 					$this->broken = true;
-					return;
+					break;
 				}
 			}
 		} catch (\Throwable) {
 			$this->broken = true;
 		}
+		if ($this->broken) {
+			// unknown state: closed, so that nothing else uses it (and the backend can build a new one)
+			self::disconnect($this->sftp);
+		}
 	}
 
 	private static function chunk(object $sftp): int {

lib/Backend/StorageBackend.php

@@ -19,12 +19,23 @@ final class StorageBackend implements IBackend {
 	private bool $atomicPut;
 	/** SFTP: read through SftpReadStream, which keeps several requests in flight. */
 	private bool $sftpReads;
+	/** Seconds without a request after which an SFTP connection is checked before it is used. */
+	public const IDLE_CHECK = 60;
+	private float $lastUsed;
+	/** Times the storage was built anew after its connection died (for tests and diagnostics). */
+	public int $reconnects = 0;
 
+	/**
+	 * @param (\Closure():IStorage)|null $reconnect builds the same storage anew, with a new
+	 *                                             connection
+	 */
 	public function __construct(
 		private IStorage $storage,
 		private string $base = '',
 		bool $createBase = true,
+		private ?\Closure $reconnect = null,
 	) {
+		$this->lastUsed = microtime(true);
 		$this->base = trim($base, '/');
 		self::validate($this->base, true);
 		$this->atomicPut = $storage->instanceOfStorage(\OCA\Files_External\Lib\Storage\AmazonS3::class);
@@ -40,6 +51,8 @@ final class StorageBackend implements IBackend {
 	}
 
 	public function put(string $path, $stream): void {
+		// not repeated on a new connection: the stream is (partly) read
+		$this->idleCheck();
 		$expected = self::remaining($stream);
 		$target = $this->abs($path);
 		$this->mkdirs(dirname($target));
@@ -83,6 +96,11 @@ final class StorageBackend implements IBackend {
 	}
 
 	public function get(string $path) {
+		return $this->call(fn () => $this->open($path));
+	}
+
+	/** @return resource */
+	private function open(string $path) {
 		if ($this->sftpReads) {
 			$fh = SftpReadStream::open($this->storage, $this->abs($path));
 			if (is_resource($fh)) {
@@ -104,7 +122,7 @@ final class StorageBackend implements IBackend {
 			$fast = null;
 			if ($this->sftpReads) {
 				$abs = array_map(fn (string $p): string => $this->abs($p), $chunk);
-				$fast = SftpReadStream::readMany($this->storage, $abs);
+				$fast = $this->call(fn () => 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) {
@@ -128,6 +146,10 @@ final class StorageBackend implements IBackend {
 	}
 
 	public function getRange(string $path, int $offset, int $length): string {
+		return $this->call(fn () => $this->readRange($path, $offset, $length));
+	}
+
+	private function readRange(string $path, int $offset, int $length): string {
 		$fh = $this->get($path);
 		try {
 			if (fseek($fh, $offset) !== 0) {
@@ -160,22 +182,30 @@ final class StorageBackend implements IBackend {
 	}
 
 	public function exists(string $path): bool {
-		return $this->storage->file_exists($this->abs($path));
+		$abs = $this->abs($path);
+		return $this->call(fn () => $this->storage->file_exists($abs), fn (bool $found) => !$found);
 	}
 
 	public function list(string $prefix): array {
-		$result = [];
-		$this->listInto($this->abs($prefix), $result);
+		$abs = $this->abs($prefix);
+		$result = $this->call(function () use ($abs): array {
+			$result = [];
+			$this->listInto($abs, $result);
+			return $result;
+		}, fn (array $found) => $found === []);
 		sort($result);
 		$cut = $this->base === '' ? 0 : strlen($this->base) + 1;
 		return array_map(fn ($p) => substr($p, $cut), $result);
 	}
 
 	public function delete(string $path): void {
+		$this->idleCheck();
 		$this->storage->unlink($this->abs($path));
 	}
 
 	public function move(string $from, string $to): void {
+		// not repeated on a new connection: the server may have done it before the connection died
+		$this->idleCheck();
 		$target = $this->abs($to);
 		$this->mkdirs(dirname($target));
 		if (!$this->storage->rename($this->abs($from), $target)) {
@@ -185,6 +215,11 @@ final class StorageBackend implements IBackend {
 
 	public function listFiles(string $dir): array {
 		$abs = $this->abs($dir);
+		return $this->call(fn () => $this->filesIn($dir, $abs), fn (array $found) => $found === []);
+	}
+
+	/** @return list<string> */
+	private function filesIn(string $dir, string $abs): array {
 		if (!$this->storage->is_dir($abs)) {
 			return []; // not created yet
 		}
@@ -205,6 +240,94 @@ final class StorageBackend implements IBackend {
 		return $result;
 	}
 
+	/**
+	 * Run $op; when it throws, or $failed says its result may come from a dead connection, and
+	 * the SFTP connection turns out to be dead, run it once more on a new connection.
+	 *
+	 * @template T
+	 * @param \Closure():T $op
+	 * @param (\Closure(T):bool)|null $failed
+	 * @return T
+	 */
+	private function call(\Closure $op, ?\Closure $failed = null): mixed {
+		$this->idleCheck();
+		try {
+			$result = $op();
+		} catch (\Throwable $e) {
+			if (!$this->reopenIfDead()) {
+				throw $e;
+			}
+			return $op();
+		}
+		if ($failed !== null && $failed($result) && $this->reopenIfDead()) {
+			return $op();
+		}
+		return $result;
+	}
+
+	/**
+	 * A Hetzner Storage Box drops an SFTP connection that carried a lot of data and then stayed
+	 * quiet for a few minutes, while phpseclib still takes it for open; files_external never
+	 * builds a new one, and every request on it fails as if the file were missing. So after
+	 * such a pause the connection is checked (one round trip) before it is used again.
+	 */
+	private function idleCheck(): void {
+		$now = microtime(true);
+		if ($this->reconnect !== null && $this->sftpReads && $now - $this->lastUsed >= self::IDLE_CHECK
+			&& $this->alive(true) === false) {
+			$this->reopen();
+		}
+		$this->lastUsed = microtime(true);
+	}
+
+	private function reopenIfDead(): bool {
+		if ($this->reconnect === null || !$this->sftpReads || $this->alive(false) !== false) {
+			return false;
+		}
+		$this->reopen();
+		return true;
+	}
+
+	/**
+	 * Whether the storage's SFTP connection is alive; null when it cannot tell.
+	 * $ask: send a request instead of trusting what phpseclib saw last.
+	 */
+	private function alive(bool $ask): ?bool {
+		try {
+			if (!is_callable([$this->storage, 'getConnection'])) {
+				return null;
+			}
+			$sftp = $this->storage->getConnection();
+			if (!is_object($sftp) || !is_callable([$sftp, 'isConnected'])) {
+				return null;
+			}
+			if (!$sftp->isConnected()) {
+				return false;
+			}
+			if (!$ask || !is_callable([$sftp, 'stat'])) {
+				return true;
+			}
+			if (is_callable([$sftp, 'clearStatCache'])) {
+				$sftp->clearStatCache(); // or stat() answers from memory
+			}
+			$sftp->stat('.');
+			return (bool)$sftp->isConnected();
+		} catch (\Throwable) {
+			return false;
+		}
+	}
+
+	private function reopen(): void {
+		/** @var \Closure():IStorage $reconnect */
+		$reconnect = $this->reconnect;
+		$this->storage = $reconnect();
+		$this->reconnects++;
+		$this->lastUsed = microtime(true);
+		if (function_exists('OCP\Log\logger')) {
+			\OCP\Log\logger('ng_backup')->info('The SFTP connection to the backup location was lost; reconnected');
+		}
+	}
+
 	private function listInto(string $dir, array &$result): void {
 		if (!$this->storage->is_dir($dir)) {
 			return;

lib/Db/DbIndexCache.php

@@ -30,19 +30,84 @@ final class DbIndexCache implements IIndexCache {
 	}
 
 	public function put(string $repositoryId, string $packId, array $entries): void {
+		try {
+			$this->insert($repositoryId, $packId, $entries);
+		} catch (Exception $e) {
+			if ($e->getReason() !== Exception::REASON_UNIQUE_CONSTRAINT_VIOLATION) {
+				throw $e;
+			}
+		}
+	}
+
+	/**
+	 * In one transaction: on PostgreSQL with its default settings every statement of its own
+	 * waits for a disk sync, which made 27,547 rows take more than 6 minutes. Rows already there
+	 * are replaced. Should the transaction fail (another process writing the same rows), the
+	 * rows are written one by one, keeping rows that are already there.
+	 */
+	public function putMany(string $repositoryId, array $packs): void {
+		if ($packs === []) {
+			return;
+		}
+		try {
+			$this->transaction(function () use ($repositoryId, $packs): void {
+				$this->remove($repositoryId, array_keys($packs));
+				$this->insertAll($repositoryId, $packs);
+			});
+		} catch (\Throwable $e) {
+			if ($this->db->inTransaction()) {
+				throw $e; // a transaction of the caller's, which may be spoilt now
+			}
+			foreach ($packs as $packId => $entries) {
+				$this->put($repositoryId, $packId, $entries);
+			}
+		}
+	}
+
+	/** All or nothing: when it fails, the cache stays as it was. */
+	public function replaceAll(string $repositoryId, array $packs): void {
+		$this->transaction(function () use ($repositoryId, $packs): void {
+			$qb = $this->db->getQueryBuilder();
+			$qb->delete('ngb_index')->where($qb->expr()->eq('repository_id', $qb->createNamedParameter($repositoryId)));
+			$qb->executeStatement();
+			$this->insertAll($repositoryId, $packs);
+		});
+	}
+
+	/** Run $write in a transaction of its own, or as part of the caller's when there is one. */
+	private function transaction(\Closure $write): void {
+		if ($this->db->inTransaction()) {
+			$write();
+			return;
+		}
+		$this->db->beginTransaction();
+		try {
+			$write();
+			$this->db->commit();
+		} catch (\Throwable $e) {
+			if ($this->db->inTransaction()) {
+				$this->db->rollBack();
+			}
+			throw $e;
+		}
+	}
+
+	/** @param array<string, list<array{id:string, offset:int, length:int, raw:int, flags:int}>> $packs */
+	private function insertAll(string $repositoryId, array $packs): void {
+		foreach ($packs as $packId => $entries) {
+			$this->insert($repositoryId, $packId, $entries);
+		}
+	}
+
+	/** @param list<array{id:string, offset:int, length:int, raw:int, flags:int}> $entries */
+	private function insert(string $repositoryId, string $packId, array $entries): void {
 		$qb = $this->db->getQueryBuilder();
 		$qb->insert('ngb_index')->values([
 			'repository_id' => $qb->createNamedParameter($repositoryId),
 			'pack_id' => $qb->createNamedParameter($packId),
 			'entries' => $qb->createNamedParameter(json_encode($entries, JSON_THROW_ON_ERROR)),
 		]);
-		try {
-			$qb->executeStatement();
-		} catch (Exception $e) {
-			if ($e->getReason() !== Exception::REASON_UNIQUE_CONSTRAINT_VIOLATION) {
-				throw $e;
-			}
-		}
+		$qb->executeStatement();
 	}
 
 	public function remove(string $repositoryId, array $packIds): void {

lib/Repository/BlobIndex.php

@@ -28,6 +28,8 @@ final class BlobIndex {
 	/** @var list<\Closure():void>|null cache writes held back by holdCache(), or null when writing directly */
 	private ?array $heldCacheWrites = null;
 	private string $repositoryId = '';
+	/** Index files written to the cache per database transaction while loading. */
+	private const CACHE_LOT = 256;
 	/** @var int index files downloaded by the last load() (for tests and diagnostics) */
 	public int $downloaded = 0;
 
@@ -107,11 +109,19 @@ final class BlobIndex {
 		// the cached packs listed before each one are added first.
 		$next = 0;
 		$total = count($fetch);
+		$toCache = []; // written to the cache a few hundred at a time, each lot in one transaction
 		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);
+			if ($this->cache !== null) {
+				$toCache[$packId] = $entries;
+				// a check writes the whole cache afresh at the end
+				if (!$verify && count($toCache) >= self::CACHE_LOT) {
+					$this->cache->putMany($this->repositoryId, $toCache);
+					$toCache = [];
+				}
+			}
 			for (; $next < count($order) && $order[$next] !== $packId; $next++) {
 				$this->add($order[$next], $cached[$order[$next]] ?? []);
 			}
@@ -125,20 +135,16 @@ final class BlobIndex {
 			$this->add($order[$next], $cached[$order[$next]] ?? []);
 		}
 		if ($verify && $this->cache !== null) {
-			// Refresh the cache from what was just verified.
-			// One pass over the blobs: a pass per pack took half an hour on a 1 TB repository
-			// (27,547 packs x 400,000 blobs), long enough for the location to drop the idle
-			// connection before the check went on.
-			$byPack = [];
-			foreach ($this->blobs as $id => [$pack, $offset, $length, $raw, $flags]) {
-				$byPack[$pack][] = ['id' => $id, 'offset' => $offset, 'length' => $length, 'raw' => $raw, 'flags' => $flags];
-			}
-			$this->cache->remove($this->repositoryId, array_keys($this->cache->all($this->repositoryId)));
-			foreach ($byPack as $packId => $entries) {
-				$this->cache->put($this->repositoryId, $packId, $entries);
-			}
+			// Refresh the cache from what was just verified, as the index files list it: a blob in
+			// two packs is in both rows, so that removing one of the packs leaves the other right.
+			// In one transaction: one row at a time took more than 6 minutes on a 1 TB repository
+			// (27,547 packs), long enough for the location to drop the connection.
+			$this->cache->replaceAll($this->repositoryId, $toCache);
 			return;
 		}
+		if ($toCache !== []) {
+			$this->cache?->putMany($this->repositoryId, $toCache);
+		}
 		$stale = array_keys(array_diff_key($cached, $present));
 		if ($stale !== []) {
 			$this->cache?->remove($this->repositoryId, $stale);

lib/Repository/IIndexCache.php

@@ -15,6 +15,20 @@ interface IIndexCache {
 	/** @param list<array{id:string, offset:int, length:int, raw:int, flags:int}> $entries */
 	public function put(string $repositoryId, string $packId, array $entries): void;
 
+	/**
+	 * put() for many packs at once (one database transaction).
+	 *
+	 * @param array<string, list<array{id:string, offset:int, length:int, raw:int, flags:int}>> $packs packId => entries
+	 */
+	public function putMany(string $repositoryId, array $packs): void;
+
+	/**
+	 * Replace everything cached for the repository with $packs (one database transaction).
+	 *
+	 * @param array<string, list<array{id:string, offset:int, length:int, raw:int, flags:int}>> $packs packId => entries
+	 */
+	public function replaceAll(string $repositoryId, array $packs): void;
+
 	/** @param list<string> $packIds */
 	public function remove(string $repositoryId, array $packIds): void;
 }

lib/Service/TargetService.php

@@ -17,6 +17,7 @@ use OCA\NgBackup\Db\TargetMapper;
 use OCA\NgBackup\Repository\Repository;
 use OCP\App\IAppManager;
 use OCP\AppFramework\Db\DoesNotExistException;
+use OCP\Files\Storage\IStorage;
 use OCP\Security\ICrypto;
 use OCP\Server;
 
@@ -279,7 +280,8 @@ final class TargetService {
 	}
 
 	private function buildBackend(string $backend, string $auth, array $options, string $basePath, bool $createBase = true): StorageBackend {
-		return new StorageBackend($this->factory()->create($backend, $auth, $options), $basePath, $createBase);
+		$create = fn (): IStorage => $this->factory()->create($backend, $auth, $options);
+		return new StorageBackend($create(), $basePath, $createBase, $create);
 	}
 
 	private ?ExternalStorageFactory $factory = null;