Repository: read the blobs of a pack from one open stream
Restores and deep verifies requested every blob separately: a new file handle and a seek on the location each time. They read the blobs of a snapshot in the order they were written, so the stream of the pack read last now stays open and is read on; a jump backwards, to another pack, or more than 8 MiB ahead opens the pack again at the offset, as before. On a Hetzner Storage Box over SMB, restoring 86 MB in 202 files went from 0.78 MB/s to 4.2 MB/s.
This commit is contained in:
@@ -7,6 +7,15 @@ the app store accepts a `-beta` suffix, but Nextcloud servers then only offer th
|
||||
beta/daily update channel, which would hide the app from most admins. The stability stage is
|
||||
instead carried by the app description, the git tag, the release, and this file.
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
### Fixed
|
||||
- Restores and `occ backup:verify --deep` fetched every blob with its own request, so on a
|
||||
remote location they ran far below the speed of the connection. The blobs of a pack are now
|
||||
read from one open stream, in the order they were written. Restoring 86 MB in 202 files from a
|
||||
Hetzner Storage Box over SMB went from 0.78 MB/s to 4.2 MB/s, the speed of reading the packs
|
||||
as plain files over that connection.
|
||||
|
||||
## [0.9.3-beta.4] - 2026-10-08
|
||||
|
||||
Bugfix release.
|
||||
|
||||
@@ -30,12 +30,18 @@ final class Repository {
|
||||
private const FLAG_DEFLATE = 1;
|
||||
/** Reads per blob in a deep verify before it is reported as unreadable (not corrupted). */
|
||||
private const VERIFY_READ_ATTEMPTS = 3;
|
||||
/** A gap up to this size in an open pack stream is read past instead of opening the pack again. */
|
||||
private const SKIP_MAX = 8 * 1048576;
|
||||
|
||||
private StreamCipher $cipher;
|
||||
private BlobIndex $index;
|
||||
private ?PackWriter $runPacks = null;
|
||||
private ?int $maxBytes = null;
|
||||
private Catalog $catalog;
|
||||
/** @var resource|null open read stream of the pack read last, positioned at $packStreamPos */
|
||||
private $packStream = null;
|
||||
private string $packStreamId = '';
|
||||
private int $packStreamPos = 0;
|
||||
|
||||
private function __construct(
|
||||
private IBackend $backend,
|
||||
@@ -708,7 +714,70 @@ final class Repository {
|
||||
|
||||
private function loadBlob(string $id): string {
|
||||
[$pack, $offset, $length] = $this->index->get($id);
|
||||
return $this->decodeBlob($id, $this->backend->getRange('packs/' . substr($pack, 0, 2) . '/' . $pack, $offset, $length));
|
||||
return $this->decodeBlob($id, $this->readPack($pack, $offset, $length));
|
||||
}
|
||||
|
||||
/**
|
||||
* Bytes of a pack. Restores and deep verifies read the blobs of a snapshot in the order they
|
||||
* were written, so the stream of the pack read last is kept open and read on from where it
|
||||
* is: one request per pack instead of one per blob (a round trip each, and on most remote
|
||||
* storages a new connection or file handle too). Anything else opens the pack at the offset,
|
||||
* as a single read always did.
|
||||
*/
|
||||
private function readPack(string $pack, int $offset, int $length): string {
|
||||
if ($this->packStream !== null && $this->packStreamId === $pack && $offset >= $this->packStreamPos && $offset - $this->packStreamPos <= self::SKIP_MAX) {
|
||||
try {
|
||||
while ($this->packStreamPos < $offset) {
|
||||
$this->readExactly(min($offset - $this->packStreamPos, 1048576), $pack);
|
||||
}
|
||||
return $this->readExactly($length, $pack);
|
||||
} catch (\Exception) {
|
||||
// The open stream broke (e.g. a timeout); open the pack again below.
|
||||
}
|
||||
}
|
||||
$this->closePackStream();
|
||||
$fh = $this->backend->get('packs/' . substr($pack, 0, 2) . '/' . $pack);
|
||||
$this->packStream = $fh;
|
||||
$this->packStreamId = $pack;
|
||||
$this->packStreamPos = 0;
|
||||
try {
|
||||
if ($offset > 0 && fseek($fh, $offset) !== 0) {
|
||||
throw new RepositoryException('Seek failed in pack ' . $pack);
|
||||
}
|
||||
$this->packStreamPos = $offset;
|
||||
return $this->readExactly($length, $pack);
|
||||
} catch (\Exception $e) {
|
||||
$this->closePackStream();
|
||||
throw $e;
|
||||
}
|
||||
}
|
||||
|
||||
/** Read $length bytes from the open pack stream; a short read closes it. */
|
||||
private function readExactly(int $length, string $pack): string {
|
||||
$fh = $this->packStream ?? throw new RepositoryException('No open pack stream');
|
||||
$data = '';
|
||||
while (strlen($data) < $length) {
|
||||
$chunk = fread($fh, min($length - strlen($data), 1048576));
|
||||
if ($chunk === false || $chunk === '') {
|
||||
$this->closePackStream();
|
||||
throw new RepositoryException('Short read in pack ' . $pack);
|
||||
}
|
||||
$data .= $chunk;
|
||||
}
|
||||
$this->packStreamPos += $length;
|
||||
return $data;
|
||||
}
|
||||
|
||||
private function closePackStream(): void {
|
||||
$fh = $this->packStream;
|
||||
$this->packStream = null;
|
||||
if ($fh !== null) {
|
||||
@fclose($fh);
|
||||
}
|
||||
}
|
||||
|
||||
public function __destruct() {
|
||||
$this->closePackStream();
|
||||
}
|
||||
|
||||
/** Decrypts and checks a blob read from its pack; throws when it is corrupted or tampered with. */
|
||||
@@ -848,7 +917,7 @@ final class Repository {
|
||||
$encrypted = null;
|
||||
for ($attempt = 1; $attempt <= self::VERIFY_READ_ATTEMPTS && $encrypted === null; $attempt++) {
|
||||
try {
|
||||
$encrypted = $this->backend->getRange('packs/' . substr($pack, 0, 2) . '/' . $pack, $offset, $length);
|
||||
$encrypted = $this->readPack($pack, $offset, $length);
|
||||
} catch (\Exception $e) { // storage adapters throw their own exceptions; \Error (a bug) is not retried
|
||||
$stats['readError'] = $e->getMessage();
|
||||
}
|
||||
|
||||
@@ -0,0 +1,134 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace OCA\NgBackup\Tests\Unit\Repository;
|
||||
|
||||
use OCA\NgBackup\Backend\IBackend;
|
||||
use OCA\NgBackup\Backend\LocalBackend;
|
||||
use OCA\NgBackup\Crypto\KeyRing;
|
||||
use OCA\NgBackup\Repository\Repository;
|
||||
use OCA\NgBackup\Tests\Unit\TempDirTrait;
|
||||
use PHPUnit\Framework\TestCase;
|
||||
|
||||
/** Restores and deep verifies read a pack as one stream instead of one request per blob. */
|
||||
class PackReadTest extends TestCase {
|
||||
use TempDirTrait;
|
||||
|
||||
private string $root;
|
||||
private KeyRing $keys;
|
||||
private string $src;
|
||||
private string $snapshot;
|
||||
|
||||
protected function setUp(): void {
|
||||
$this->root = $this->tempDir();
|
||||
$this->keys = KeyRing::generate();
|
||||
$this->src = $this->tempDir();
|
||||
for ($i = 0; $i < 200; $i++) {
|
||||
file_put_contents(sprintf('%s/f%03d', $this->src, $i), random_bytes(20000));
|
||||
}
|
||||
$backend = new LocalBackend($this->root);
|
||||
$repo = Repository::initWithKey($backend, $this->keys, $this->keys->wrap('pw'));
|
||||
$this->snapshot = $repo->backupDirectory($this->src)['snapshot'];
|
||||
}
|
||||
|
||||
private function open(PackOpenCountingBackend $backend): Repository {
|
||||
return Repository::openWithKey($backend, $this->keys, null, null);
|
||||
}
|
||||
|
||||
public function testFullRestoreOpensEachPackOnce(): void {
|
||||
$backend = new PackOpenCountingBackend(new LocalBackend($this->root));
|
||||
$target = $this->tempDir();
|
||||
|
||||
$this->assertSame(200, $this->open($backend)->restore($this->snapshot, $target));
|
||||
|
||||
for ($i = 0; $i < 200; $i++) {
|
||||
$name = sprintf('f%03d', $i);
|
||||
$this->assertSame(file_get_contents("$this->src/$name"), file_get_contents("$target/$name"));
|
||||
}
|
||||
// One pack: one stream instead of 200 requests.
|
||||
$this->assertSame(1, $backend->packOpens());
|
||||
}
|
||||
|
||||
public function testDeepVerifyOpensEachPackOnce(): void {
|
||||
$backend = new PackOpenCountingBackend(new LocalBackend($this->root));
|
||||
|
||||
$result = $this->open($backend)->verify($this->snapshot, deep: true);
|
||||
|
||||
$this->assertSame(200 * 20000, $result['bytesChecked']);
|
||||
$this->assertSame(1, $backend->packOpens());
|
||||
}
|
||||
|
||||
public function testReadingBackwardsOpensThePackAgain(): void {
|
||||
$backend = new PackOpenCountingBackend(new LocalBackend($this->root));
|
||||
$repo = $this->open($backend);
|
||||
|
||||
$this->assertSame(1, $repo->restore($this->snapshot, $this->tempDir(), 'f150'));
|
||||
$this->assertSame(1, $repo->restore($this->snapshot, $this->tempDir(), 'f050'));
|
||||
|
||||
$this->assertSame(2, $backend->packOpens());
|
||||
}
|
||||
|
||||
public function testDeepVerifyOfATruncatedPackStillChecksTheBlobsBeforeTheDamage(): void {
|
||||
$packs = (new LocalBackend($this->root))->list('packs');
|
||||
$this->assertCount(1, $packs);
|
||||
$pack = $this->root . '/' . $packs[0];
|
||||
$fh = fopen($pack, 'r+b');
|
||||
ftruncate($fh, intdiv(filesize($pack), 2));
|
||||
fclose($fh);
|
||||
|
||||
$result = $this->open(new PackOpenCountingBackend(new LocalBackend($this->root)))->verify($this->snapshot, deep: true);
|
||||
|
||||
$this->assertGreaterThan(80 * 20000, $result['bytesChecked']);
|
||||
$this->assertNotSame([], $result['unreadable']);
|
||||
$this->assertSame([], $result['failed']);
|
||||
}
|
||||
}
|
||||
|
||||
final class PackOpenCountingBackend implements IBackend {
|
||||
/** @var list<string> */
|
||||
public array $gets = [];
|
||||
|
||||
public function __construct(
|
||||
private IBackend $inner,
|
||||
) {
|
||||
}
|
||||
|
||||
public function put(string $path, $stream): void {
|
||||
$this->inner->put($path, $stream);
|
||||
}
|
||||
|
||||
public function packOpens(): int {
|
||||
return count(array_filter($this->gets, fn (string $p): bool => str_starts_with($p, 'packs/')));
|
||||
}
|
||||
|
||||
public function get(string $path) {
|
||||
$this->gets[] = $path;
|
||||
return $this->inner->get($path);
|
||||
}
|
||||
|
||||
public function getRange(string $path, int $offset, int $length): string {
|
||||
$this->gets[] = $path;
|
||||
return $this->inner->getRange($path, $offset, $length);
|
||||
}
|
||||
|
||||
public function exists(string $path): bool {
|
||||
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 {
|
||||
return $this->inner->listFiles($dir);
|
||||
}
|
||||
|
||||
public function delete(string $path): void {
|
||||
$this->inner->delete($path);
|
||||
}
|
||||
}
|
||||
@@ -129,7 +129,7 @@ class VerifyTest extends TestCase {
|
||||
}
|
||||
}
|
||||
|
||||
/** Delegates to a real backend, but getRange() throws for the first $failures calls. */
|
||||
/** Delegates to a real backend, but reading a pack throws for the first $failures reads. */
|
||||
final class FlakyRangeBackend implements IBackend {
|
||||
public function __construct(
|
||||
private IBackend $inner,
|
||||
@@ -143,15 +143,22 @@ final class FlakyRangeBackend implements IBackend {
|
||||
}
|
||||
|
||||
public function get(string $path) {
|
||||
if (str_starts_with($path, 'packs/')) {
|
||||
$this->fail();
|
||||
}
|
||||
return $this->inner->get($path);
|
||||
}
|
||||
|
||||
public function getRange(string $path, int $offset, int $length): string {
|
||||
$this->fail();
|
||||
return $this->inner->getRange($path, $offset, $length);
|
||||
}
|
||||
|
||||
private function fail(): void {
|
||||
if ($this->failures > 0) {
|
||||
$this->failures--;
|
||||
throw $this->error ?? new BackendException('connection reset');
|
||||
}
|
||||
return $this->inner->getRange($path, $offset, $length);
|
||||
}
|
||||
|
||||
public function exists(string $path): bool {
|
||||
|
||||
Reference in New Issue
Block a user