Full restore over SFTP: read packs ahead on four connections at once
A restore read one pack at a time over one SFTP connection, which over a long line is bound by the round trip and by what one TCP connection gets: about 2-3 MiB/s from a Hetzner Storage Box to Canada, with PHP mostly waiting. Repository::restore() now first lists the ranges of packs the restore needs, in order (ReadAheadPlan), and the location reads them ahead into unlinked files in a spool folder in the target (IReadsAhead). Over SFTP, SftpRangeFetcher does that on four connections of its own, 16 requests in flight each, taking the answers of whichever connection has one. A range that fails, a damaged copy, or a blob that moved since the plan is read the usual way, which reports what is wrong. Other locations restore as before. On a test line with 120 ms RTT and 0.2% loss, a 1.6 GiB restore went from 5.1 to 17.8 MiB/s, all files identical.
This commit is contained in:
@@ -0,0 +1,30 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
// SPDX-FileCopyrightText: 2026 Dillard Blom
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
namespace OCA\NgBackup\Backend;
|
||||
|
||||
/**
|
||||
* A backend that can read many byte ranges ahead of use, several at a time over connections of
|
||||
* its own (a full restore, where one connection at a time is bound by the round trip).
|
||||
*/
|
||||
interface IReadsAhead {
|
||||
/** Whether readAhead() is worth planning for: false when it would return null anyway. */
|
||||
public function canReadAhead(): bool;
|
||||
|
||||
/**
|
||||
* The ranges in the order given, each as a local stream at offset 0 holding its bytes, read a
|
||||
* few ranges ahead of the one the caller is at. A stream is closed when the generator is
|
||||
* resumed. Null for a range that could not be read this way (a missing object, a server or
|
||||
* connection error): the caller reads it the usual way, which reports why.
|
||||
*
|
||||
* @param list<array{0:string, 1:int, 2:int}> $ranges [path, offset, length] each
|
||||
* @param string $spool an existing local directory for the files that hold the ranges
|
||||
* @return \Generator<int, resource|null>|null by position in $ranges; null when this backend
|
||||
* cannot read ahead now
|
||||
*/
|
||||
public function readAhead(array $ranges, string $spool): ?\Generator;
|
||||
}
|
||||
@@ -0,0 +1,344 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
// SPDX-FileCopyrightText: 2026 Dillard Blom
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
namespace OCA\NgBackup\Backend;
|
||||
|
||||
/**
|
||||
* Reads byte ranges of files on an SFTP location ahead of use, over several connections at once,
|
||||
* into local files (StorageBackend::readAhead(), for a full restore).
|
||||
*
|
||||
* One connection is bound by the round trip and by what a single TCP connection gets over a long
|
||||
* line: a restore from a Hetzner Storage Box to Canada (RTT 122 ms) reached about 2 MiB/s, with
|
||||
* the PHP process mostly waiting. Here every connection reads its own range with DEPTH requests
|
||||
* in flight. PHP runs one thing at a time: while it waits for an answer on one connection, the
|
||||
* answers of the others arrive in the kernel's buffers, so the connections still transfer at the
|
||||
* same time.
|
||||
*
|
||||
* Uses the same internal phpseclib 2 methods as SftpReadStream (see SftpReadStream::connection()).
|
||||
* The connections are this class's own, never the storage's: no other request goes over them.
|
||||
* A range that fails (missing file, server error, a connection that breaks) comes out as null,
|
||||
* for the caller to read the usual way; a broken connection is not used again.
|
||||
*/
|
||||
final class SftpRangeFetcher {
|
||||
/** Connections used at once by a full restore. A Hetzner Storage Box allows 10 per account. */
|
||||
public const CONNECTIONS = 4;
|
||||
/** READ requests in flight per connection. */
|
||||
public const DEPTH = 16;
|
||||
/** Ranges read ahead of the one the caller is at, in flight or waiting in the spool. */
|
||||
public const AHEAD = 8;
|
||||
/** Seconds ready() waits for any connection to answer before it waits for each in turn. */
|
||||
private const SELECT_TIMEOUT = 10;
|
||||
/** See SftpReadStream::CHUNK and MIN_CHUNK. */
|
||||
private const CHUNK = 261120;
|
||||
private const MIN_CHUNK = 32768;
|
||||
|
||||
private const FXP_OPEN = 3;
|
||||
private const FXP_CLOSE = 4;
|
||||
private const FXP_READ = 5;
|
||||
private const FXP_STATUS = 101;
|
||||
private const FXP_HANDLE = 102;
|
||||
private const FXP_DATA = 103;
|
||||
private const FXF_READ = 1;
|
||||
private const FX_OK = 0;
|
||||
|
||||
private const OPEN = 0;
|
||||
private const READ = 1;
|
||||
private const CLOSE = 2;
|
||||
|
||||
/** @var array<int, int> READ size per connection, lowered when a server answers less than asked */
|
||||
private array $chunks = [];
|
||||
/** @var array<int, SftpRangeJob> the range each busy connection reads */
|
||||
private array $jobs = [];
|
||||
/** @var array<int, resource|null> ranges read (or failed) and not yet given to the caller */
|
||||
private array $done = [];
|
||||
private int $requestId = 0x4e520000;
|
||||
|
||||
/**
|
||||
* @param array<int, object> $connections phpseclib 2 connections with their SFTP channel open
|
||||
* @param list<array{0:string, 1:int, 2:int}> $ranges paths below $root
|
||||
*/
|
||||
private function __construct(
|
||||
private array $connections,
|
||||
private string $root,
|
||||
private array $ranges,
|
||||
private string $spool,
|
||||
) {
|
||||
foreach (array_keys($connections) as $c) {
|
||||
$this->chunks[$c] = self::CHUNK;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* See IReadsAhead::readAhead(). The connections are ended when the generator ends.
|
||||
*
|
||||
* @param list<object> $connections
|
||||
* @param list<array{0:string, 1:int, 2:int}> $ranges
|
||||
* @return \Generator<int, resource|null>
|
||||
*/
|
||||
public static function fetch(array $connections, string $root, array $ranges, string $spool): \Generator {
|
||||
$fetcher = new self($connections, rtrim($root, '/'), $ranges, $spool);
|
||||
try {
|
||||
yield from $fetcher->run();
|
||||
} finally {
|
||||
$fetcher->end();
|
||||
}
|
||||
}
|
||||
|
||||
/** @return \Generator<int, resource|null> */
|
||||
private function run(): \Generator {
|
||||
$count = count($this->ranges);
|
||||
$next = 0;
|
||||
for ($want = 0; $want < $count; $want++) {
|
||||
while (!array_key_exists($want, $this->done)) {
|
||||
foreach (array_keys($this->connections) as $c) {
|
||||
if (!isset($this->jobs[$c]) && $next < $count && $next < $want + self::AHEAD) {
|
||||
$this->start($c, $next++);
|
||||
}
|
||||
}
|
||||
if ($this->jobs === [] && !array_key_exists($want, $this->done)) {
|
||||
// no connection left: the rest the usual way
|
||||
$this->done[$want] = null;
|
||||
$next = max($next, $want + 1);
|
||||
break;
|
||||
}
|
||||
foreach ($this->ready() as $c) {
|
||||
$this->step($c);
|
||||
}
|
||||
}
|
||||
$fh = $this->done[$want];
|
||||
unset($this->done[$want]);
|
||||
try {
|
||||
yield $want => $fh;
|
||||
} finally {
|
||||
if (is_resource($fh)) {
|
||||
fclose($fh);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The busy connections that have an answer waiting, so that a connection that waits for a
|
||||
* lost packet to be sent again does not hold up the others; all of them when the connections
|
||||
* do not tell (then they are taken in turn).
|
||||
*
|
||||
* @return list<int>
|
||||
*/
|
||||
private function ready(): array {
|
||||
$sockets = [];
|
||||
$ready = [];
|
||||
foreach ($this->jobs as $c => $job) {
|
||||
$sftp = $this->connections[$c];
|
||||
$vars = get_object_vars($sftp); // only what is public: phpseclib 2's are
|
||||
$buffered = $vars['requestBuffer'] ?? null;
|
||||
if (is_array($buffered) && isset($job->queue[0]) && array_key_exists($job->queue[0][0], $buffered)) {
|
||||
$ready[] = $c; // read already, while waiting for another answer
|
||||
continue;
|
||||
}
|
||||
$socket = $vars['fsock'] ?? null;
|
||||
if (!is_resource($socket)) {
|
||||
return array_keys($this->jobs);
|
||||
}
|
||||
$sockets[$c] = $socket;
|
||||
}
|
||||
if ($ready !== [] || $sockets === []) {
|
||||
return $ready;
|
||||
}
|
||||
$read = $sockets;
|
||||
$write = null;
|
||||
$except = null;
|
||||
// Data in PHP's own buffer of a socket counts as readable. Nothing within the time:
|
||||
// each in turn, where phpseclib's own timeout applies.
|
||||
$n = @stream_select($read, $write, $except, self::SELECT_TIMEOUT);
|
||||
return $n === false || $n === 0 ? array_keys($this->jobs) : array_keys($read);
|
||||
}
|
||||
|
||||
private function start(int $c, int $i): void {
|
||||
[$path, $offset, $length] = $this->ranges[$i];
|
||||
$fh = $this->spoolFile();
|
||||
if ($fh === null) {
|
||||
$this->done[$i] = null;
|
||||
return;
|
||||
}
|
||||
if ($length <= 0) {
|
||||
$this->done[$i] = $fh;
|
||||
return;
|
||||
}
|
||||
$this->jobs[$c] = new SftpRangeJob($i, $fh, $offset, $offset + $length);
|
||||
try {
|
||||
$remote = $this->connections[$c]->_realpath($this->root . '/' . ltrim($path, '/'));
|
||||
if (!is_string($remote)) {
|
||||
$this->finish($c);
|
||||
return;
|
||||
}
|
||||
$this->send($c, self::FXP_OPEN, pack('Na*N2', strlen($remote), $remote, self::FXF_READ, 0), self::OPEN);
|
||||
} catch (\Throwable) {
|
||||
$this->drop($c);
|
||||
}
|
||||
}
|
||||
|
||||
/** Take the oldest answer on connection $c, then keep its requests in flight. */
|
||||
private function step(int $c): void {
|
||||
$job = $this->jobs[$c];
|
||||
[$id, $kind, $offset, $length] = array_shift($job->queue) ?? throw new \LogicException('no request in flight');
|
||||
$sftp = $this->connections[$c];
|
||||
try {
|
||||
$response = $sftp->_get_sftp_packet($id);
|
||||
if (!is_string($response)) {
|
||||
throw new BackendException('SFTP connection lost');
|
||||
}
|
||||
$type = $sftp->packet_type;
|
||||
if ($kind === self::OPEN) {
|
||||
if ($type === self::FXP_HANDLE) {
|
||||
$job->handle = substr($response, 4);
|
||||
} else {
|
||||
$job->failed = true; // missing, or no permission
|
||||
}
|
||||
} elseif ($kind === self::READ) {
|
||||
if (!$job->failed) {
|
||||
$this->received($c, $type, $response, $offset, $length);
|
||||
}
|
||||
} else {
|
||||
$status = $type === self::FXP_STATUS && strlen($response) >= 4 ? unpack('N', $response) : false;
|
||||
$closed = is_array($status) && $status[1] === self::FX_OK;
|
||||
$this->finish($c);
|
||||
if (!$closed) {
|
||||
// handles might pile up on the server: not used again
|
||||
$this->drop($c);
|
||||
}
|
||||
return;
|
||||
}
|
||||
$this->refill($c);
|
||||
} catch (\Throwable) {
|
||||
$this->drop($c);
|
||||
}
|
||||
}
|
||||
|
||||
private function received(int $c, int $type, string $response, int $offset, int $length): void {
|
||||
$job = $this->jobs[$c];
|
||||
$declared = strlen($response) >= 4 ? unpack('N', $response) : false;
|
||||
$n = is_array($declared) ? $declared[1] : 0;
|
||||
if ($type !== self::FXP_DATA || $n <= 0 || $n !== strlen($response) - 4 || $n > $length) {
|
||||
// end of file before the end of the range, a server error or a malformed answer
|
||||
$job->failed = true;
|
||||
return;
|
||||
}
|
||||
if (fseek($job->fh, $offset - $job->start) !== 0 || fwrite($job->fh, substr($response, 4)) !== $n) {
|
||||
$job->failed = true;
|
||||
return;
|
||||
}
|
||||
$job->got += $n;
|
||||
if ($n < $length) {
|
||||
// a server that caps a READ: ask for the rest, and no more than it gives from now on
|
||||
$job->redo[] = [$offset + $n, $length - $n];
|
||||
$this->chunks[$c] = max(self::MIN_CHUNK, min($this->chunks[$c], $n));
|
||||
}
|
||||
}
|
||||
|
||||
/** Send READs until DEPTH are in flight; when nothing is left to ask for, close the file. */
|
||||
private function refill(int $c): void {
|
||||
$job = $this->jobs[$c];
|
||||
if ($job->handle !== '' && !$job->failed) {
|
||||
while (count($job->queue) < self::DEPTH) {
|
||||
if ($job->redo !== []) {
|
||||
[$offset, $length] = array_shift($job->redo);
|
||||
} elseif ($job->next < $job->end) {
|
||||
$offset = $job->next;
|
||||
$length = min($this->chunks[$c], $job->end - $offset);
|
||||
$job->next += $length;
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
$packet = pack('Na*N3', strlen($job->handle), $job->handle, intdiv($offset, 4294967296), $offset % 4294967296, $length);
|
||||
$this->send($c, self::FXP_READ, $packet, self::READ, $offset, $length);
|
||||
}
|
||||
}
|
||||
if ($job->queue !== []) {
|
||||
return;
|
||||
}
|
||||
if ($job->handle === '') {
|
||||
$this->finish($c); // not opened: nothing to close
|
||||
return;
|
||||
}
|
||||
$this->send($c, self::FXP_CLOSE, pack('Na*', strlen($job->handle), $job->handle), self::CLOSE);
|
||||
}
|
||||
|
||||
private function send(int $c, int $type, string $packet, int $kind, int $offset = 0, int $length = 0): void {
|
||||
$this->requestId = $this->requestId >= 0x7fffffff ? 0x4e520000 : $this->requestId + 1;
|
||||
if (!$this->connections[$c]->_send_sftp_packet($type, $packet, $this->requestId)) {
|
||||
throw new BackendException('SFTP connection failed while sending a request');
|
||||
}
|
||||
$this->jobs[$c]->queue[] = [$this->requestId, $kind, $offset, $length];
|
||||
}
|
||||
|
||||
/** The range of connection $c is done: its stream for the caller, or null when it failed. */
|
||||
private function finish(int $c): void {
|
||||
$job = $this->jobs[$c];
|
||||
unset($this->jobs[$c]);
|
||||
if ($job->failed || $job->got !== $job->end - $job->start || !rewind($job->fh)) {
|
||||
self::closeFile($job->fh);
|
||||
$this->done[$job->i] = null;
|
||||
return;
|
||||
}
|
||||
$this->done[$job->i] = $job->fh;
|
||||
}
|
||||
|
||||
/** A connection in an unknown state: its range fails, and it is not used again. */
|
||||
private function drop(int $c): void {
|
||||
if (isset($this->jobs[$c])) {
|
||||
$job = $this->jobs[$c];
|
||||
unset($this->jobs[$c]);
|
||||
self::closeFile($job->fh);
|
||||
$this->done[$job->i] = null;
|
||||
}
|
||||
self::disconnect($this->connections[$c]);
|
||||
unset($this->connections[$c]);
|
||||
}
|
||||
|
||||
/** @return resource|null a new file in the spool, already unlinked: nothing is left behind if the process dies */
|
||||
private function spoolFile() {
|
||||
$path = $this->spool . '/' . bin2hex(random_bytes(8));
|
||||
$fh = @fopen($path, 'x+b');
|
||||
if ($fh === false) {
|
||||
return null;
|
||||
}
|
||||
@unlink($path);
|
||||
return $fh;
|
||||
}
|
||||
|
||||
private function end(): void {
|
||||
foreach ($this->jobs as $job) {
|
||||
self::closeFile($job->fh);
|
||||
}
|
||||
foreach ($this->done as $fh) {
|
||||
if (is_resource($fh)) {
|
||||
fclose($fh);
|
||||
}
|
||||
}
|
||||
$this->jobs = [];
|
||||
$this->done = [];
|
||||
foreach ($this->connections as $sftp) {
|
||||
self::disconnect($sftp);
|
||||
}
|
||||
$this->connections = [];
|
||||
}
|
||||
|
||||
private static function closeFile(mixed $fh): void {
|
||||
if (is_resource($fh)) {
|
||||
fclose($fh);
|
||||
}
|
||||
}
|
||||
|
||||
private static function disconnect(object $sftp): void {
|
||||
try {
|
||||
if (is_callable([$sftp, 'disconnect'])) {
|
||||
$sftp->disconnect();
|
||||
}
|
||||
} catch (\Throwable) {
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
// SPDX-FileCopyrightText: 2026 Dillard Blom
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
namespace OCA\NgBackup\Backend;
|
||||
|
||||
/**
|
||||
* One range that SftpRangeFetcher reads on one connection.
|
||||
*
|
||||
* @internal
|
||||
*/
|
||||
final class SftpRangeJob {
|
||||
public string $handle = '';
|
||||
/** Offset of the next READ to send. */
|
||||
public int $next;
|
||||
/** @var list<array{int, int}> offset and length of parts to ask for again (after a short answer) */
|
||||
public array $redo = [];
|
||||
/** @var list<array{int, int, int, int}> request id, kind, offset and length, in the order sent */
|
||||
public array $queue = [];
|
||||
/** Bytes received. */
|
||||
public int $got = 0;
|
||||
public bool $failed = false;
|
||||
|
||||
/** @var resource the spool file that holds the range */
|
||||
public $fh;
|
||||
|
||||
/** @param resource $fh */
|
||||
public function __construct(
|
||||
public int $i,
|
||||
$fh,
|
||||
public int $start,
|
||||
public int $end,
|
||||
) {
|
||||
$this->fh = $fh;
|
||||
$this->next = $start;
|
||||
}
|
||||
}
|
||||
@@ -144,9 +144,9 @@ final class SftpReadStream {
|
||||
|
||||
/**
|
||||
* The storage's phpseclib connection, with its SFTP channel open, when it offers what these
|
||||
* streams need; null otherwise.
|
||||
* streams (and SftpRangeFetcher) need; null otherwise.
|
||||
*/
|
||||
private static function connection(object $storage): ?object {
|
||||
public static function connection(object $storage): ?object {
|
||||
if (!is_callable([$storage, 'getConnection']) || !is_callable([$storage, 'getRoot'])) {
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -14,7 +14,7 @@ use OCP\Files\Storage\IWriteStreamStorage;
|
||||
* IBackend on top of a Nextcloud storage object (from files_external or core), below a base
|
||||
* directory on that storage.
|
||||
*/
|
||||
final class StorageBackend implements IBackend {
|
||||
final class StorageBackend implements IBackend, IReadsAhead {
|
||||
/** Object stores write whole objects atomically; other storages get a temp name + rename. */
|
||||
private bool $atomicPut;
|
||||
/** SFTP: read through SftpReadStream, which keeps several requests in flight. */
|
||||
@@ -243,6 +243,51 @@ final class StorageBackend implements IBackend {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Over SFTP: SftpRangeFetcher, on CONNECTIONS new connections of its own (built with the
|
||||
* reconnect function), so the storage's connection stays free for other requests. Elsewhere
|
||||
* not (yet): null.
|
||||
*/
|
||||
public function canReadAhead(): bool {
|
||||
return $this->sftpReads && $this->reconnect !== null;
|
||||
}
|
||||
|
||||
public function readAhead(array $ranges, string $spool): ?\Generator {
|
||||
$reconnect = $this->reconnect;
|
||||
if (!$this->sftpReads || $reconnect === null || $ranges === []) {
|
||||
return null;
|
||||
}
|
||||
$storages = [];
|
||||
$connections = [];
|
||||
$root = '';
|
||||
for ($n = 0; $n < SftpRangeFetcher::CONNECTIONS; $n++) {
|
||||
try {
|
||||
$storage = $reconnect();
|
||||
$sftp = SftpReadStream::connection($storage);
|
||||
if ($sftp === null || !is_callable([$storage, 'getRoot'])) {
|
||||
break;
|
||||
}
|
||||
$root = (string)$storage->getRoot();
|
||||
} catch (\Throwable) {
|
||||
break; // the server may allow fewer connections: go on with those that work
|
||||
}
|
||||
$storages[] = $storage;
|
||||
$connections[] = $sftp;
|
||||
}
|
||||
if ($connections === []) {
|
||||
return null;
|
||||
}
|
||||
$abs = array_map(fn (array $r): array => [$this->abs($r[0]), $r[1], $r[2]], $ranges);
|
||||
return (static function () use ($storages, $connections, $root, $abs, $spool): \Generator {
|
||||
// the storages hold their connections: kept until the fetcher is done
|
||||
try {
|
||||
yield from SftpRangeFetcher::fetch($connections, $root, $abs, $spool);
|
||||
} finally {
|
||||
unset($storages);
|
||||
}
|
||||
})();
|
||||
}
|
||||
|
||||
public function getRange(string $path, int $offset, int $length): string {
|
||||
return $this->call(fn () => $this->readRange($path, $offset, $length));
|
||||
}
|
||||
|
||||
@@ -8,9 +8,10 @@ declare(strict_types=1);
|
||||
namespace OCA\NgBackup\Repository;
|
||||
|
||||
use OCA\NgBackup\Backend\IBackend;
|
||||
use OCA\NgBackup\Backend\IReadsAhead;
|
||||
|
||||
/** The location as Repository sees it: closes the PackReader's open stream before every other request. */
|
||||
final class PackReaderGuard implements IBackend {
|
||||
final class PackReaderGuard implements IBackend, IReadsAhead {
|
||||
public function __construct(
|
||||
private IBackend $inner,
|
||||
private PackReader $reader,
|
||||
@@ -43,6 +44,15 @@ final class PackReaderGuard implements IBackend {
|
||||
return $this->inner->getRanges($ranges);
|
||||
}
|
||||
|
||||
public function canReadAhead(): bool {
|
||||
return $this->inner instanceof IReadsAhead && $this->inner->canReadAhead();
|
||||
}
|
||||
|
||||
/** Over connections of the location's own: the open stream may stay open. */
|
||||
public function readAhead(array $ranges, string $spool): ?\Generator {
|
||||
return $this->inner instanceof IReadsAhead ? $this->inner->readAhead($ranges, $spool) : null;
|
||||
}
|
||||
|
||||
public function exists(string $path): bool {
|
||||
$this->reader->close();
|
||||
return $this->inner->exists($path);
|
||||
|
||||
@@ -0,0 +1,123 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
// SPDX-FileCopyrightText: 2026 Dillard Blom
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
namespace OCA\NgBackup\Repository;
|
||||
|
||||
/**
|
||||
* The ranges of packs a restore reads, in order, for IReadsAhead::readAhead(), and the reading of
|
||||
* blobs from them as the restore goes.
|
||||
*
|
||||
* add() takes the blobs in the order the restore needs them. A blob in the same pack as the range
|
||||
* before it, not before that range and at most GAP after its end, extends that range; anything
|
||||
* else starts a new one. read() takes the same blobs in the same order: a blob is in the current
|
||||
* range or in the next one, which then becomes current; a blob not in the plan is read elsewhere.
|
||||
*/
|
||||
final class ReadAheadPlan {
|
||||
/** Unused bytes between two blobs of a pack that are read along rather than asked for apart. */
|
||||
public const GAP = 4 * 1048576;
|
||||
/** Longest range: packs are 32 MiB, repacked ones too. */
|
||||
public const MAX_RANGE = 64 * 1048576;
|
||||
/** Ranges after the current one that read() looks in for a blob. */
|
||||
private const SEARCH = 16;
|
||||
|
||||
/** @var list<array{0:string, 1:int, 2:int}> [path, offset, length] */
|
||||
private array $ranges = [];
|
||||
/** @var \Generator<int, resource|null>|null */
|
||||
private ?\Generator $streams = null;
|
||||
private int $current = -1;
|
||||
/** @var resource|null */
|
||||
private $stream = null;
|
||||
|
||||
/** @param string $spool local folder for the ranges read ahead (see IReadsAhead::readAhead()) */
|
||||
public function __construct(
|
||||
public readonly string $spool,
|
||||
) {
|
||||
}
|
||||
|
||||
public function add(string $path, int $offset, int $length): void {
|
||||
$last = array_pop($this->ranges);
|
||||
if ($last !== null) {
|
||||
[$p, $start, $len] = $last;
|
||||
if ($p === $path && $offset >= $start && $offset <= $start + $len + self::GAP
|
||||
&& $offset + $length - $start <= self::MAX_RANGE) {
|
||||
$this->ranges[] = [$p, $start, max($len, $offset + $length - $start)];
|
||||
return;
|
||||
}
|
||||
$this->ranges[] = $last;
|
||||
}
|
||||
$this->ranges[] = [$path, $offset, $length];
|
||||
}
|
||||
|
||||
/** @return list<array{0:string, 1:int, 2:int}> */
|
||||
public function ranges(): array {
|
||||
return $this->ranges;
|
||||
}
|
||||
|
||||
/** @param \Generator<int, resource|null> $streams what IReadsAhead::readAhead() gave for ranges() */
|
||||
public function readFrom(\Generator $streams): void {
|
||||
$this->streams = $streams;
|
||||
}
|
||||
|
||||
/** The bytes of a blob, or null when its range could not be read ahead (read it the usual way). */
|
||||
public function read(string $path, int $offset, int $length): ?string {
|
||||
if ($this->streams === null) {
|
||||
return null;
|
||||
}
|
||||
if (!$this->contains($this->current, $path, $offset, $length)) {
|
||||
// Normally the next range. Further on when blobs before it were read elsewhere: the
|
||||
// index was reloaded after another server's cleanup moved them to new packs.
|
||||
$to = null;
|
||||
for ($i = $this->current + 1; $i <= $this->current + self::SEARCH && $to === null; $i++) {
|
||||
if ($this->contains($i, $path, $offset, $length)) {
|
||||
$to = $i;
|
||||
}
|
||||
}
|
||||
if ($to === null) {
|
||||
return null; // not where the plan has it
|
||||
}
|
||||
if ($this->current < 0) {
|
||||
$this->streams->current();
|
||||
$this->current = 0;
|
||||
}
|
||||
while ($this->current < $to) {
|
||||
$this->streams->next();
|
||||
$this->current++;
|
||||
}
|
||||
$stream = $this->streams->valid() ? $this->streams->current() : null;
|
||||
$this->stream = is_resource($stream) ? $stream : null;
|
||||
}
|
||||
$fh = $this->stream;
|
||||
if ($fh === null || fseek($fh, $offset - $this->ranges[$this->current][1]) !== 0) {
|
||||
return null;
|
||||
}
|
||||
$data = '';
|
||||
while (strlen($data) < $length) {
|
||||
$chunk = fread($fh, $length - strlen($data));
|
||||
if ($chunk === false || $chunk === '') {
|
||||
return null;
|
||||
}
|
||||
$data .= $chunk;
|
||||
}
|
||||
return $data;
|
||||
}
|
||||
|
||||
/** Ends the reads ahead, also halfway: closes their streams and connections. */
|
||||
public function close(): void {
|
||||
$this->stream = null;
|
||||
$streams = $this->streams;
|
||||
$this->streams = null;
|
||||
unset($streams); // the generator's finally blocks run when it is destroyed
|
||||
}
|
||||
|
||||
private function contains(int $i, string $path, int $offset, int $length): bool {
|
||||
if ($i < 0 || !isset($this->ranges[$i])) {
|
||||
return false;
|
||||
}
|
||||
[$p, $start, $len] = $this->ranges[$i];
|
||||
return $p === $path && $offset >= $start && $offset + $length <= $start + $len;
|
||||
}
|
||||
}
|
||||
@@ -9,6 +9,7 @@ namespace OCA\NgBackup\Repository;
|
||||
|
||||
use OCA\NgBackup\Backend\BackendException;
|
||||
use OCA\NgBackup\Backend\IBackend;
|
||||
use OCA\NgBackup\Backend\IReadsAhead;
|
||||
use OCA\NgBackup\Crypto\KeyRing;
|
||||
use OCA\NgBackup\Crypto\StreamCipher;
|
||||
|
||||
@@ -216,6 +217,53 @@ final class Repository {
|
||||
if ($prefix !== '' && !self::isSafePath(rtrim($prefix, '/'))) {
|
||||
throw new RepositoryException('Invalid path prefix');
|
||||
}
|
||||
$plan = $this->readAhead($snapshotId, $prefix, $target);
|
||||
try {
|
||||
return $this->restoreEntries($snapshotId, $target, $prefix, $stripPrefix, $heartbeat, $plan);
|
||||
} finally {
|
||||
$plan?->close();
|
||||
if ($plan !== null) {
|
||||
@rmdir($plan->spool);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Start reading the packs of a restore ahead, when the location can (IReadsAhead): a first
|
||||
* pass over the snapshot lists the ranges of packs it needs, in order. The spool is a
|
||||
* folder in the target, which has room for the restore anyway.
|
||||
*/
|
||||
private function readAhead(string $snapshotId, string $prefix, string $target): ?ReadAheadPlan {
|
||||
if (!$this->backend instanceof IReadsAhead || !$this->backend->canReadAhead()) {
|
||||
return null;
|
||||
}
|
||||
$plan = new ReadAheadPlan(rtrim($target, '/') . '/.ngb-restore-spool-' . bin2hex(random_bytes(4)));
|
||||
foreach ($this->streamEntries($snapshotId) as $entry) {
|
||||
if ($prefix !== '' && !str_starts_with($entry['p'], $prefix)) {
|
||||
continue;
|
||||
}
|
||||
foreach ($entry['b'] as $id) {
|
||||
try {
|
||||
[$pack, $offset, $length] = $this->index->get($id);
|
||||
} catch (RepositoryException) {
|
||||
continue; // the restore reports it when it gets there
|
||||
}
|
||||
$plan->add('packs/' . substr($pack, 0, 2) . '/' . $pack, $offset, $length);
|
||||
}
|
||||
}
|
||||
if (count($plan->ranges()) < 2 || (!is_dir($plan->spool) && !@mkdir($plan->spool, 0700, true))) {
|
||||
return null;
|
||||
}
|
||||
$streams = $this->backend->readAhead($plan->ranges(), $plan->spool);
|
||||
if ($streams === null) {
|
||||
@rmdir($plan->spool);
|
||||
return null;
|
||||
}
|
||||
$plan->readFrom($streams);
|
||||
return $plan;
|
||||
}
|
||||
|
||||
private function restoreEntries(string $snapshotId, string $target, string $prefix, bool $stripPrefix, ?callable $heartbeat, ?ReadAheadPlan $plan): int {
|
||||
$count = 0;
|
||||
foreach ($this->streamEntries($snapshotId) as $entry) {
|
||||
if ($prefix !== '' && !str_starts_with($entry['p'], $prefix)) {
|
||||
@@ -240,7 +288,7 @@ final class Repository {
|
||||
throw new RepositoryException('Cannot create a temporary file for ' . $entry['p']);
|
||||
}
|
||||
foreach ($entry['b'] as $id) {
|
||||
fwrite($out, $this->loadBlob($id));
|
||||
fwrite($out, $this->restoreBlob($id, $plan));
|
||||
if ($heartbeat !== null) {
|
||||
$heartbeat();
|
||||
}
|
||||
@@ -789,6 +837,22 @@ final class Repository {
|
||||
}
|
||||
}
|
||||
|
||||
/** A blob for restore(): read ahead when it could be, otherwise (or when that copy is bad) the usual way. */
|
||||
private function restoreBlob(string $id, ?ReadAheadPlan $plan): string {
|
||||
if ($plan !== null) {
|
||||
[$pack, $offset, $length] = $this->index->get($id);
|
||||
$encrypted = $plan->read('packs/' . substr($pack, 0, 2) . '/' . $pack, $offset, $length);
|
||||
if ($encrypted !== null) {
|
||||
try {
|
||||
return $this->decodeBlob($id, $encrypted);
|
||||
} catch (\Exception) {
|
||||
// read again from the location below, which reports what is wrong
|
||||
}
|
||||
}
|
||||
}
|
||||
return $this->loadBlob($id);
|
||||
}
|
||||
|
||||
private function loadBlob(string $id): string {
|
||||
[$pack, $offset, $length] = $this->index->get($id);
|
||||
try {
|
||||
|
||||
Reference in New Issue
Block a user