SftpReadStream: fall back to the storage's own stream instead of failing a read
This commit is contained in:
+183
-33
@@ -17,7 +17,10 @@ namespace OCA\NgBackup\Backend;
|
||||
*
|
||||
* Talks to the storage's phpseclib connection through the same internal methods files_external's
|
||||
* read stream uses (_send_sftp_packet, _get_sftp_packet, _realpath); open() returns null when the
|
||||
* connection does not have them, and StorageBackend then falls back to the storage's own stream.
|
||||
* connection does not offer them (phpseclib 3 does not), and StorageBackend then falls back to the
|
||||
* storage's own stream. A restore must rather be slow than fail: when a read fails halfway, the
|
||||
* stream goes on from the same position with the storage's own stream. After MAX_FAILURES streams
|
||||
* in a row failed, open() returns null for the rest of the process; one that works resets the count.
|
||||
* Like that stream, it owns the connection while open: no other request may go over the same
|
||||
* connection until it is closed (PackReaderGuard takes care of that for the repository).
|
||||
*/
|
||||
@@ -47,6 +50,11 @@ final class SftpReadStream {
|
||||
*/
|
||||
private static ?\WeakMap $chunks = null;
|
||||
private static int $requestId = 0x4e470000;
|
||||
/** Streams in a row that failed; at MAX_FAILURES only the storage's own stream is used. */
|
||||
private static int $failures = 0;
|
||||
public const MAX_FAILURES = 3;
|
||||
/** Set by stream_open() when the connection failed rather than the file. */
|
||||
private static bool $openBroke = false;
|
||||
|
||||
/** @var resource|null set by PHP for stream wrappers */
|
||||
public $context;
|
||||
@@ -58,6 +66,11 @@ final class SftpReadStream {
|
||||
* @psalm-suppress PropertyNotSetInConstructor
|
||||
*/
|
||||
private object $sftp;
|
||||
/** The storage and the path on it, to fall back to the storage's own stream. */
|
||||
private ?object $storage = null;
|
||||
private string $file = '';
|
||||
/** @var resource|null the storage's own stream, once this one has failed */
|
||||
private $fallback = null;
|
||||
private string $handle = '';
|
||||
/** Position of the reader: the first byte of $buffer. */
|
||||
private int $pos = 0;
|
||||
@@ -68,7 +81,10 @@ final class SftpReadStream {
|
||||
private array $inFlight = [];
|
||||
private int $depth = 1;
|
||||
private bool $eof = false;
|
||||
/** This stream gave up on its own requests. */
|
||||
private bool $failed = false;
|
||||
/** The connection is in an unknown state: nothing more is sent over it or read from it. */
|
||||
private bool $broken = false;
|
||||
/** Length of the last short answer, until the next answer tells whether it was the end of the file. */
|
||||
private ?int $shortAnswer = null;
|
||||
|
||||
@@ -76,28 +92,52 @@ final class SftpReadStream {
|
||||
* @param object $storage files_external's SFTP storage (getConnection(), getRoot())
|
||||
* @return resource|false|null the stream, false when the file cannot be opened, null when the
|
||||
* connection does not offer what this stream needs
|
||||
* @throws BackendException when the connection failed after the OPEN request went out: the
|
||||
* storage's own stream must not be tried on it, as it would take a
|
||||
* late answer to that request for its own
|
||||
*/
|
||||
public static function open(object $storage, string $path) {
|
||||
if (!method_exists($storage, 'getConnection') || !method_exists($storage, 'getRoot')) {
|
||||
if (self::$failures >= self::MAX_FAILURES) {
|
||||
return null;
|
||||
}
|
||||
$sftp = $storage->getConnection();
|
||||
if (!is_object($sftp) || !method_exists($sftp, '_send_sftp_packet') || !method_exists($sftp, '_get_sftp_packet')
|
||||
|| !method_exists($sftp, '_realpath') || !property_exists($sftp, 'packet_type')) {
|
||||
try {
|
||||
if (!is_callable([$storage, 'getConnection']) || !is_callable([$storage, 'getRoot'])) {
|
||||
return null;
|
||||
}
|
||||
$sftp = $storage->getConnection();
|
||||
// is_callable and get_object_vars from here only see what is public: phpseclib 3 has
|
||||
// methods of these names neither public nor at all.
|
||||
if (!is_object($sftp) || !is_callable([$sftp, '_send_sftp_packet']) || !is_callable([$sftp, '_get_sftp_packet'])
|
||||
|| !is_callable([$sftp, '_realpath']) || !array_key_exists('packet_type', get_object_vars($sftp))) {
|
||||
return null;
|
||||
}
|
||||
if (!in_array(self::PROTOCOL, stream_get_wrappers(), true)) {
|
||||
stream_wrapper_register(self::PROTOCOL, self::class);
|
||||
}
|
||||
$root = rtrim((string)$storage->getRoot(), '/');
|
||||
$context = stream_context_create([self::PROTOCOL => [
|
||||
'session' => $sftp,
|
||||
'path' => $root . '/' . ltrim($path, '/'),
|
||||
'storage' => $storage,
|
||||
'file' => $path,
|
||||
]]);
|
||||
self::$openBroke = false;
|
||||
$fh = @fopen(self::PROTOCOL . '://file', 'r', false, $context);
|
||||
} catch (\Throwable) {
|
||||
// nothing was sent yet
|
||||
self::$failures++;
|
||||
return null;
|
||||
}
|
||||
if (!in_array(self::PROTOCOL, stream_get_wrappers(), true)) {
|
||||
stream_wrapper_register(self::PROTOCOL, self::class);
|
||||
if ($fh === false && self::openBroke()) {
|
||||
throw new BackendException('SFTP connection failed while opening ' . $path);
|
||||
}
|
||||
$root = rtrim((string)$storage->getRoot(), '/');
|
||||
$context = stream_context_create([self::PROTOCOL => ['session' => $sftp, 'path' => $root . '/' . ltrim($path, '/')]]);
|
||||
$fh = @fopen(self::PROTOCOL . '://file', 'r', false, $context);
|
||||
return $fh === false ? false : $fh;
|
||||
return $fh;
|
||||
}
|
||||
|
||||
/** Test hook: forget the learned READ sizes. */
|
||||
/** Test hook: forget the learned READ sizes and earlier failures. */
|
||||
public static function reset(): void {
|
||||
self::$chunks = null;
|
||||
self::$failures = 0;
|
||||
}
|
||||
|
||||
public function stream_open(string $url, string $mode, int $options, ?string &$openedPath): bool {
|
||||
@@ -106,26 +146,51 @@ final class SftpReadStream {
|
||||
return false;
|
||||
}
|
||||
$this->sftp = $options['session'];
|
||||
$remote = $this->sftp->_realpath((string)$options['path']);
|
||||
if (!is_string($remote)) {
|
||||
return false;
|
||||
$storage = $options['storage'] ?? null;
|
||||
$this->storage = is_object($storage) ? $storage : null;
|
||||
$this->file = (string)($options['file'] ?? '');
|
||||
try {
|
||||
$remote = $this->sftp->_realpath((string)$options['path']);
|
||||
if (!is_string($remote)) {
|
||||
return false;
|
||||
}
|
||||
$id = self::nextId();
|
||||
if (!$this->sftp->_send_sftp_packet(self::FXP_OPEN, pack('Na*N2', strlen($remote), $remote, self::FXF_READ, 0), $id)) {
|
||||
return $this->connectionFailed();
|
||||
}
|
||||
$response = $this->sftp->_get_sftp_packet($id);
|
||||
} catch (\Throwable) {
|
||||
return $this->connectionFailed();
|
||||
}
|
||||
$id = self::nextId();
|
||||
if (!$this->sftp->_send_sftp_packet(self::FXP_OPEN, pack('Na*N2', strlen($remote), $remote, self::FXF_READ, 0), $id)) {
|
||||
return false;
|
||||
if (!is_string($response)) {
|
||||
return $this->connectionFailed();
|
||||
}
|
||||
$response = $this->sftp->_get_sftp_packet($id);
|
||||
if ($this->sftp->packet_type !== self::FXP_HANDLE || !is_string($response)) {
|
||||
return false;
|
||||
if ($this->sftp->packet_type !== self::FXP_HANDLE) {
|
||||
return false; // an answer, but no handle: the file cannot be opened
|
||||
}
|
||||
$this->handle = substr($response, 4);
|
||||
return true;
|
||||
}
|
||||
|
||||
/** Not a missing file but a connection that does not behave as expected. */
|
||||
private function connectionFailed(): bool {
|
||||
self::$failures++;
|
||||
self::$openBroke = true;
|
||||
return false;
|
||||
}
|
||||
|
||||
/** Whether the last stream_open() failed on the connection; a call, as fopen() sets it. */
|
||||
private static function openBroke(): bool {
|
||||
return self::$openBroke;
|
||||
}
|
||||
|
||||
public function stream_read(int $count): string|false {
|
||||
if ($this->fallback !== null) {
|
||||
return fread($this->fallback, $count);
|
||||
}
|
||||
while (strlen($this->buffer) < $count && !$this->eof) {
|
||||
if (!$this->receive()) {
|
||||
return false;
|
||||
return $this->useFallback() ? $this->stream_read($count) : false;
|
||||
}
|
||||
}
|
||||
$data = substr($this->buffer, 0, $count);
|
||||
@@ -135,22 +200,36 @@ final class SftpReadStream {
|
||||
}
|
||||
|
||||
public function stream_eof(): bool {
|
||||
if ($this->fallback !== null) {
|
||||
return feof($this->fallback);
|
||||
}
|
||||
return $this->eof && $this->buffer === '';
|
||||
}
|
||||
|
||||
public function stream_tell(): int {
|
||||
if ($this->fallback !== null) {
|
||||
$pos = ftell($this->fallback);
|
||||
return $pos === false ? $this->pos : $pos;
|
||||
}
|
||||
return $this->pos;
|
||||
}
|
||||
|
||||
public function stream_seek(int $offset, int $whence = SEEK_SET): bool {
|
||||
if ($this->fallback !== null) {
|
||||
return fseek($this->fallback, $offset, $whence) === 0;
|
||||
}
|
||||
if ($whence === SEEK_CUR) {
|
||||
$offset += $this->pos;
|
||||
} elseif ($whence !== SEEK_SET) {
|
||||
return false; // the size is never asked for, to save a round trip
|
||||
}
|
||||
if ($offset < 0 || $this->failed) {
|
||||
if ($offset < 0) {
|
||||
return false;
|
||||
}
|
||||
if ($this->failed) {
|
||||
$this->pos = $offset;
|
||||
return $this->useFallback();
|
||||
}
|
||||
if ($offset >= $this->pos && $offset - $this->pos <= strlen($this->buffer)) {
|
||||
$this->buffer = substr($this->buffer, $offset - $this->pos);
|
||||
} else {
|
||||
@@ -171,11 +250,59 @@ final class SftpReadStream {
|
||||
}
|
||||
|
||||
public function stream_close(): void {
|
||||
if (!$this->failed) {
|
||||
self::$failures = 0;
|
||||
}
|
||||
$fallback = $this->fallback;
|
||||
$this->fallback = null;
|
||||
if ($fallback !== null) {
|
||||
fclose($fallback);
|
||||
}
|
||||
$this->closeHandle();
|
||||
}
|
||||
|
||||
/**
|
||||
* Go on with the storage's own stream from the current position, after a server error on this
|
||||
* one. The data still buffered is dropped and read again. False after a connection error, or
|
||||
* when that stream cannot be opened or positioned: then the read fails as it would have
|
||||
* without this class. Reopening by path is safe for a repository: packs and index files are
|
||||
* never rewritten, and every object is authenticated when it is decrypted.
|
||||
*/
|
||||
private function useFallback(): bool {
|
||||
$this->closeHandle();
|
||||
$this->buffer = '';
|
||||
// files_external's stream takes whatever answer comes next on the connection: after a
|
||||
// connection error, answers to this stream's requests may still be on their way.
|
||||
if ($this->broken || $this->storage === null || !is_callable([$this->storage, 'fopen'])) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
$fh = $this->storage->fopen($this->file, 'r');
|
||||
if (!is_resource($fh)) {
|
||||
return false;
|
||||
}
|
||||
if ($this->pos > 0 && fseek($fh, $this->pos) !== 0) {
|
||||
fclose($fh);
|
||||
return false;
|
||||
}
|
||||
} catch (\Throwable) {
|
||||
return false;
|
||||
}
|
||||
$this->fallback = $fh;
|
||||
return true;
|
||||
}
|
||||
|
||||
private function closeHandle(): void {
|
||||
$this->drain();
|
||||
if ($this->handle !== '') {
|
||||
$id = self::nextId();
|
||||
if ($this->sftp->_send_sftp_packet(self::FXP_CLOSE, pack('Na*', strlen($this->handle), $this->handle), $id)) {
|
||||
$this->sftp->_get_sftp_packet($id);
|
||||
if ($this->handle !== '' && !$this->broken) {
|
||||
try {
|
||||
$id = self::nextId();
|
||||
if (!$this->sftp->_send_sftp_packet(self::FXP_CLOSE, pack('Na*', strlen($this->handle), $this->handle), $id)
|
||||
|| !is_string($this->sftp->_get_sftp_packet($id))) {
|
||||
$this->broken = true;
|
||||
}
|
||||
} catch (\Throwable) {
|
||||
$this->broken = true;
|
||||
}
|
||||
}
|
||||
$this->handle = '';
|
||||
@@ -186,11 +313,21 @@ final class SftpReadStream {
|
||||
if ($this->failed) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
return $this->receiveOne();
|
||||
} catch (\Throwable) {
|
||||
$this->broken = true;
|
||||
return $this->fail();
|
||||
}
|
||||
}
|
||||
|
||||
private function receiveOne(): bool {
|
||||
while (count($this->inFlight) < $this->depth) {
|
||||
$id = self::nextId();
|
||||
$length = self::chunk($this->sftp);
|
||||
$packet = pack('Na*N3', strlen($this->handle), $this->handle, intdiv($this->next, 4294967296), $this->next % 4294967296, $length);
|
||||
if (!$this->sftp->_send_sftp_packet(self::FXP_READ, $packet, $id)) {
|
||||
$this->broken = true;
|
||||
return $this->fail();
|
||||
}
|
||||
$this->inFlight[] = [$id, $this->next, $length];
|
||||
@@ -199,10 +336,11 @@ final class SftpReadStream {
|
||||
[$id, $offset, $length] = array_shift($this->inFlight);
|
||||
$response = $this->sftp->_get_sftp_packet($id);
|
||||
if (!is_string($response)) {
|
||||
$this->broken = true;
|
||||
return $this->fail();
|
||||
}
|
||||
if ($this->sftp->packet_type === self::FXP_DATA) {
|
||||
$data = substr($response, 4);
|
||||
$data = substr($response, 4);
|
||||
if ($data === '') {
|
||||
return $this->fail();
|
||||
}
|
||||
@@ -236,18 +374,30 @@ final class SftpReadStream {
|
||||
|
||||
private function fail(): bool {
|
||||
$this->drain();
|
||||
if (!$this->failed) {
|
||||
self::$failures++;
|
||||
}
|
||||
$this->failed = true;
|
||||
return false;
|
||||
}
|
||||
|
||||
/** Read and drop the answers still on their way, so the connection is free again. */
|
||||
private function drain(): void {
|
||||
foreach ($this->inFlight as [$id]) {
|
||||
if (!$this->failed) {
|
||||
$this->sftp->_get_sftp_packet($id);
|
||||
}
|
||||
}
|
||||
$inFlight = $this->inFlight;
|
||||
$this->inFlight = [];
|
||||
if ($this->broken) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
foreach ($inFlight as [$id]) {
|
||||
if (!is_string($this->sftp->_get_sftp_packet($id))) {
|
||||
$this->broken = true;
|
||||
return;
|
||||
}
|
||||
}
|
||||
} catch (\Throwable) {
|
||||
$this->broken = true;
|
||||
}
|
||||
}
|
||||
|
||||
private static function chunk(object $sftp): int {
|
||||
|
||||
@@ -87,10 +87,8 @@ final class StorageBackend implements IBackend {
|
||||
if (is_resource($fh)) {
|
||||
return $fh;
|
||||
}
|
||||
if ($fh === null) {
|
||||
$this->sftpReads = false;
|
||||
}
|
||||
// false: not opened, which need not mean missing; the storage's own stream decides
|
||||
// null: not supported or failing for now; false: not opened, which need not mean
|
||||
// missing. Either way the storage's own stream decides.
|
||||
}
|
||||
$fh = $this->storage->fopen($this->abs($path), 'r');
|
||||
if ($fh === false) {
|
||||
|
||||
Reference in New Issue
Block a user