Pre-beta review: data safety in storage, catalog and backup runs

Storage and repository:
- Listing skipped every name containing ".part-", directories too, so the
  exports of a user like "john.part-time" were invisible to prune on S3,
  SFTP, FTP and SMB and their data was deleted. Only real temporary uploads
  ("<name>.part-<8 hex>") are skipped now.
- prune refuses to run while a recorded per-user export's manifest is
  missing, instead of freeing that export's data.
- forget() and untrash() copy first, switch the catalog, and delete the
  originals last, so a crash can no longer leave the catalog pointing at a
  missing snapshot or database dump.
- Writes are checked against the expected size, so a short upload is an
  error instead of a truncated pack; raw restores check rename() and touch().
- A catalog write refuses to overwrite a generation another process just
  wrote, and the catalog lock lasts 300 s, so opening a large repository
  inside it no longer lets the lease run out.

Backup runs and scheduling:
- A file that keeps changing is read again at most three times; then the
  last read is kept and reported, instead of restarting forever and
  blocking cleanup.
- Unreadable files and folders are recorded, logged, shown by occ
  backup:run, stored in the snapshot and notified to admins, instead of
  being skipped silently.
- After a failed scheduled backup, at most three attempts a day, an hour
  apart, instead of a new run and notification every five minutes.
- The job no longer spins on a run whose lease another process holds, and
  the weekly cleanup catches up after a Sunday spent backing up.
- Notifications for the per-user export retention events.

lib/Backend/StorageBackend.php

@@ -30,7 +30,13 @@ final class StorageBackend implements IBackend {
 		}
 	}
 
+	/** A temporary upload of put(): "<name>.part-<8 hex>". */
+	public static function isTemporary(string $name): bool {
+		return preg_match('/\.part-[0-9a-f]{8}$/', $name) === 1;
+	}
+
 	public function put(string $path, $stream): void {
+		$expected = self::remaining($stream);
 		$target = $this->abs($path);
 		$this->mkdirs(dirname($target));
 		$writeTo = $this->atomicPut ? $target : $target . '.part-' . bin2hex(random_bytes(4));
@@ -54,6 +60,10 @@ final class StorageBackend implements IBackend {
 			if (is_resource($stream)) {
 				fclose($stream);
 			}
+			// Some adapters report success after a short write or a failed flush: check the size.
+			if ($expected !== null && $this->storage->filesize($writeTo) !== $expected) {
+				throw new BackendException('Incomplete write for ' . $path);
+			}
 			if ($writeTo !== $target && !$this->storage->rename($writeTo, $target)) {
 				throw new BackendException('Cannot finalise ' . $path);
 			}
@@ -132,19 +142,35 @@ final class StorageBackend implements IBackend {
 			return;
 		}
 		while (($name = readdir($dh)) !== false) {
-			if ($name === '.' || $name === '..' || str_contains($name, '.part-')) {
+			if ($name === '.' || $name === '..') {
 				continue;
 			}
 			$child = $dir . '/' . $name;
 			if ($this->storage->is_dir($child)) {
 				$this->listInto($child, $result);
-			} else {
+			} elseif (!self::isTemporary($name)) {
 				$result[] = $child;
 			}
 		}
 		closedir($dh);
 	}
 
+	/** Bytes left to read in a seekable $stream, or null when it cannot tell. @param resource $stream */
+	public static function remaining(mixed $stream): ?int {
+		if (!is_resource($stream)) {
+			return null;
+		}
+		if (!(stream_get_meta_data($stream)['seekable'] ?? false)) {
+			return null;
+		}
+		$stat = fstat($stream);
+		$pos = ftell($stream);
+		if ($stat === false || $pos === false) {
+			return null;
+		}
+		return max(0, $stat['size'] - $pos);
+	}
+
 	private function mkdirs(string $dir): void {
 		if ($dir === '' || $dir === '.' || $this->storage->is_dir($dir)) {
 			return;