Ensembia/nextcloud-ng_backup
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases 17 Wiki Activity

Repository: close the pack stream before any other request on the location

Browse Source 
Review notes on the open pack stream: it stayed open while a backup run went on to upload
(e.g. the snapshot after reading the file list back), which SFTP cannot do on one connection.
Reading moves to PackReader, and Repository reaches the location through PackReaderGuard,
which closes the stream before every other request. A pack stream that cannot seek is read up
to the offset. Deep verify no longer reports an \Error during decoding as damage.
This commit is contained in:
Dillard Blom
2026-10-08 22:34:02 +02:00
parent 5f8d5faee7
commit b309081d1a
4 changed files with 203 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/Repository/PackReader.php
+97
View File
@@ -0,0 +1,97 @@
<?php
declare(strict_types=1);
// SPDX-FileCopyrightText: 2026 Dillard Blom
// SPDX-License-Identifier: AGPL-3.0-or-later
namespace OCA\NgBackup\Repository;
use OCA\NgBackup\Backend\IBackend;
/**
* Reads blobs from packs. 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.
*
* The open stream must not overlap other requests on the location (SFTP cannot read and write
* on one connection at once): Repository reaches the location through a PackReaderGuard, which
* closes it first.
*/
final class PackReader {
/** A gap up to this size in the open stream is read past instead of opening the pack again. */
private const SKIP_MAX = 8 * 1048576;
private const CHUNK = 1048576;
/** @var resource|null */
private $stream = null;
private string $pack = '';
private int $pos = 0;
public function __construct(
private IBackend $backend,
) {
}
public function read(string $pack, int $offset, int $length): string {
if ($this->stream !== null && $this->pack === $pack && $offset >= $this->pos && $offset - $this->pos <= self::SKIP_MAX) {
try {
$this->skipTo($offset);
return $this->readExactly($length);
} catch (\Exception) {
// The open stream broke (e.g. a timeout); open the pack again below.
}
}
$this->close();
$fh = $this->backend->get('packs/' . substr($pack, 0, 2) . '/' . $pack);
$this->stream = $fh;
$this->pack = $pack;
$this->pos = 0;
try {
if ($offset > 0 && (stream_get_meta_data($fh)['seekable'] ?? false) && fseek($fh, $offset) === 0) {
$this->pos = $offset;
}
$this->skipTo($offset); // a stream that cannot seek is read up to the offset
return $this->readExactly($length);
} catch (\Exception $e) {
$this->close();
throw $e;
}
}
public function close(): void {
$fh = $this->stream;
$this->stream = null;
if ($fh !== null) {
@fclose($fh);
}
}
public function __destruct() {
$this->close();
}
private function skipTo(int $offset): void {
while ($this->pos < $offset) {
$this->readExactly(min($offset - $this->pos, self::CHUNK));
}
}
/** Read $length bytes from the open stream; a short read closes it. */
private function readExactly(int $length): string {
$fh = $this->stream ?? throw new RepositoryException('No open pack stream');
$data = '';
while (strlen($data) < $length) {
$chunk = fread($fh, min($length - strlen($data), self::CHUNK));
if ($chunk === false || $chunk === '') {
$this->close();
throw new RepositoryException('Short read in pack ' . $this->pack);
}
$data .= $chunk;
}
$this->pos += $length;
return $data;
}
}
lib/Repository/PackReaderGuard.php
+60
View File
@@ -0,0 +1,60 @@
<?php
declare(strict_types=1);
// SPDX-FileCopyrightText: 2026 Dillard Blom
// SPDX-License-Identifier: AGPL-3.0-or-later
namespace OCA\NgBackup\Repository;
use OCA\NgBackup\Backend\IBackend;
/** The location as Repository sees it: closes the PackReader's open stream before every other request. */
final class PackReaderGuard implements IBackend {
public function __construct(
private IBackend $inner,
private PackReader $reader,
) {
}
/** @param resource $stream */
public function put(string $path, $stream): void {
$this->reader->close();
$this->inner->put($path, $stream);
}
public function get(string $path) {
$this->reader->close();
return $this->inner->get($path);
}
public function getRange(string $path, int $offset, int $length): string {
$this->reader->close();
return $this->inner->getRange($path, $offset, $length);
}
public function exists(string $path): bool {
$this->reader->close();
return $this->inner->exists($path);
}
public function move(string $from, string $to): void {
$this->reader->close();
$this->inner->move($from, $to);
}
public function list(string $prefix): array {
$this->reader->close();
return $this->inner->list($prefix);
}
public function listFiles(string $dir): array {
$this->reader->close();
return $this->inner->listFiles($dir);
}
public function delete(string $path): void {
$this->reader->close();
$this->inner->delete($path);
}
}
lib/Repository/Repository.php
+7 -73
View File
@@ -30,18 +30,13 @@ 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 PackReader $packReader;
private function __construct(
private IBackend $backend,
@@ -49,7 +44,9 @@ final class Repository {
private string $repositoryId,
) {
$this->cipher = new StreamCipher($keys);
$this->index = new BlobIndex($backend, $this->cipher);
$this->packReader = new PackReader($backend);
$this->backend = new PackReaderGuard($backend, $this->packReader);
$this->index = new BlobIndex($this->backend, $this->cipher);
}
public static function init(IBackend $backend, #[\SensitiveParameter] string $passphrase): self {
@@ -714,70 +711,7 @@ final class Repository {
private function loadBlob(string $id): string {
[$pack, $offset, $length] = $this->index->get($id);
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();
return $this->decodeBlob($id, $this->packReader->read($pack, $offset, $length));
}
/** Decrypts and checks a blob read from its pack; throws when it is corrupted or tampered with. */
@@ -917,7 +851,7 @@ final class Repository {
$encrypted = null;
for ($attempt = 1; $attempt <= self::VERIFY_READ_ATTEMPTS && $encrypted === null; $attempt++) {
try {
$encrypted = $this->readPack($pack, $offset, $length);
$encrypted = $this->packReader->read($pack, $offset, $length);
} catch (\Exception $e) { // storage adapters throw their own exceptions; \Error (a bug) is not retried
$stats['readError'] = $e->getMessage();
}
@@ -928,7 +862,7 @@ final class Repository {
}
try {
$stats['bytesChecked'] += strlen($this->decodeBlob($id, $encrypted));
} catch (\Throwable) {
} catch (\Exception) { // damage; an \Error is a bug and must not be reported as damage
$stats['failed'][] = $id;
}
}
tests/Unit/Repository/PackReadTest.php
+39 -1
View File
@@ -69,6 +69,26 @@ class PackReadTest extends TestCase {
$this->assertSame(2, $backend->packOpens());
}
public function testThePackStreamIsClosedBeforeAnyOtherRequest(): void {
$backend = new PackOpenCountingBackend(new LocalBackend($this->root));
$repo = $this->open($backend);
$this->assertSame(1, $repo->restore($this->snapshot, $this->tempDir(), 'f010'));
$this->assertTrue($backend->packStreamOpen(), 'kept open for the next read');
$repo->putObject('objects/after-restore', 'data');
$this->assertSame([false], $backend->packStreamOpenAtPut);
}
public function testAPackStreamThatCannotSeekIsReadUpToTheOffset(): void {
$backend = new PackOpenCountingBackend(new LocalBackend($this->root), $this->root);
$target = $this->tempDir();
$this->assertSame(1, $this->open($backend)->restore($this->snapshot, $target, 'f150'));
$this->assertSame(file_get_contents("$this->src/f150"), file_get_contents("$target/f150"));
}
public function testDeepVerifyOfATruncatedPackStillChecksTheBlobsBeforeTheDamage(): void {
$packs = (new LocalBackend($this->root))->list('packs');
$this->assertCount(1, $packs);
@@ -88,13 +108,24 @@ class PackReadTest extends TestCase {
final class PackOpenCountingBackend implements IBackend {
/** @var list<string> */
public array $gets = [];
/** @var list<bool> */
public array $packStreamOpenAtPut = [];
/** @var resource|null */
private $lastPackStream = null;
/** @param ?string $pipeFrom serve packs as pipes (streams that cannot seek) from this local repository root */
public function __construct(
private IBackend $inner,
private ?string $pipeFrom = null,
) {
}
public function packStreamOpen(): bool {
return is_resource($this->lastPackStream);
}
public function put(string $path, $stream): void {
$this->packStreamOpenAtPut[] = $this->packStreamOpen();
$this->inner->put($path, $stream);
}
@@ -104,7 +135,14 @@ final class PackOpenCountingBackend implements IBackend {
public function get(string $path) {
$this->gets[] = $path;
return $this->inner->get($path);
if (!str_starts_with($path, 'packs/')) {
return $this->inner->get($path);
}
$fh = $this->pipeFrom !== null
? popen('cat ' . escapeshellarg($this->pipeFrom . '/' . $path), 'r')
: $this->inner->get($path);
$this->lastPackStream = $fh;
return $fh;
}
public function getRange(string $path, int $offset, int $length): string {