From d8e772137fe9db75f5358b40f8799e78749b4f45 Mon Sep 17 00:00:00 2001 From: Dillard Blom Date: Fri, 9 Oct 2026 04:27:21 +0200 Subject: [PATCH] StorageBackend: read SFTP locations with several requests in flight files_external's SFTP read stream asks for the next 256 KiB only after the previous answer has arrived, so reading costs one round trip per request whatever the line speed: 4.9 MB/s at 47 ms to a Hetzner Storage Box. SftpReadStream keeps up to eight READ requests in flight over the same phpseclib connection (15 MB/s on that link), starting with one so a small read does not fetch megabytes ahead. A server that answers less than asked is handled by asking again from where the answer ended; when more data follows, that size is used for later reads over the same connection, never below the 32 KiB every server must support. Without the needed connection methods, or when the open fails, the storage's own stream is used as before. --- lib/Backend/SftpReadStream.php | 262 ++++++++++++++++++++++ lib/Backend/StorageBackend.php | 13 ++ tests/Unit/Backend/SftpReadStreamTest.php | 243 ++++++++++++++++++++ 3 files changed, 518 insertions(+) create mode 100644 lib/Backend/SftpReadStream.php create mode 100644 tests/Unit/Backend/SftpReadStreamTest.php diff --git a/lib/Backend/SftpReadStream.php b/lib/Backend/SftpReadStream.php new file mode 100644 index 0000000..9bd9a33 --- /dev/null +++ b/lib/Backend/SftpReadStream.php @@ -0,0 +1,262 @@ +|null + */ + private static ?\WeakMap $chunks = null; + private static int $requestId = 0x4e470000; + + /** @var resource|null set by PHP for stream wrappers */ + public $context; + + /** + * The phpseclib connection. PHP creates stream wrappers without constructor arguments; it is + * set in stream_open(), and no other method is called when that fails. + * + * @psalm-suppress PropertyNotSetInConstructor + */ + private object $sftp; + private string $handle = ''; + /** Position of the reader: the first byte of $buffer. */ + private int $pos = 0; + /** Offset of the next READ to send. */ + private int $next = 0; + private string $buffer = ''; + /** @var list request id, offset, length, in the order sent */ + private array $inFlight = []; + private int $depth = 1; + private bool $eof = false; + private bool $failed = false; + /** Length of the last short answer, until the next answer tells whether it was the end of the file. */ + private ?int $shortAnswer = null; + + /** + * @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 + */ + public static function open(object $storage, string $path) { + if (!method_exists($storage, 'getConnection') || !method_exists($storage, 'getRoot')) { + 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')) { + 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, '/')]]); + $fh = @fopen(self::PROTOCOL . '://file', 'r', false, $context); + return $fh === false ? false : $fh; + } + + /** Test hook: forget the learned READ sizes. */ + public static function reset(): void { + self::$chunks = null; + } + + public function stream_open(string $url, string $mode, int $options, ?string &$openedPath): bool { + $options = stream_context_get_options($this->context ?? stream_context_get_default())[self::PROTOCOL] ?? []; + if (!isset($options['session'], $options['path']) || !is_object($options['session'])) { + return false; + } + $this->sftp = $options['session']; + $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 false; + } + $response = $this->sftp->_get_sftp_packet($id); + if ($this->sftp->packet_type !== self::FXP_HANDLE || !is_string($response)) { + return false; + } + $this->handle = substr($response, 4); + return true; + } + + public function stream_read(int $count): string|false { + while (strlen($this->buffer) < $count && !$this->eof) { + if (!$this->receive()) { + return false; + } + } + $data = substr($this->buffer, 0, $count); + $this->buffer = substr($this->buffer, strlen($data)); + $this->pos += strlen($data); + return $data; + } + + public function stream_eof(): bool { + return $this->eof && $this->buffer === ''; + } + + public function stream_tell(): int { + return $this->pos; + } + + public function stream_seek(int $offset, int $whence = SEEK_SET): bool { + 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) { + return false; + } + if ($offset >= $this->pos && $offset - $this->pos <= strlen($this->buffer)) { + $this->buffer = substr($this->buffer, $offset - $this->pos); + } else { + $this->drain(); + $this->buffer = ''; + $this->next = $offset; + $this->eof = false; + $this->depth = 1; + $this->shortAnswer = null; + } + $this->pos = $offset; + return true; + } + + /** @return array */ + public function stream_stat(): array { + return []; + } + + public function stream_close(): 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); + } + } + $this->handle = ''; + } + + /** Keep the pipeline full, then take the oldest answer. False on a connection or server error. */ + private function receive(): bool { + if ($this->failed) { + return false; + } + 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)) { + return $this->fail(); + } + $this->inFlight[] = [$id, $this->next, $length]; + $this->next += $length; + } + [$id, $offset, $length] = array_shift($this->inFlight); + $response = $this->sftp->_get_sftp_packet($id); + if (!is_string($response)) { + return $this->fail(); + } + if ($this->sftp->packet_type === self::FXP_DATA) { + $data = substr($response, 4); + if ($data === '') { + return $this->fail(); + } + $this->buffer .= $data; + $this->depth = min(self::DEPTH, $this->depth * 2); + if ($this->shortAnswer !== null) { + // More data after a short answer: that one was the server's cap, not the end. + // Never below what every server must support: a one-off small answer is no cap. + self::$chunks ??= new \WeakMap(); + self::$chunks[$this->sftp] = max(self::MIN_CHUNK, min(self::chunk($this->sftp), $this->shortAnswer)); + $this->shortAnswer = null; + } + if (strlen($data) < $length) { + // Fewer bytes than asked: the end of the file, or a server that caps a READ. The + // requests behind this one assumed a full answer, so drop them and ask again from here. + $this->drain(); + $this->next = $offset + strlen($data); + $this->shortAnswer = strlen($data); + } + return true; + } + $status = $this->sftp->packet_type === self::FXP_STATUS ? unpack('N', $response) : false; + if (is_array($status) && ($status[1] ?? null) === self::FX_EOF) { + $this->eof = true; + $this->shortAnswer = null; + $this->drain(); + return true; + } + return $this->fail(); + } + + private function fail(): bool { + $this->drain(); + $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); + } + } + $this->inFlight = []; + } + + private static function chunk(object $sftp): int { + $learned = self::$chunks?->offsetExists($sftp) === true ? self::$chunks->offsetGet($sftp) : null; + return is_int($learned) ? $learned : self::CHUNK; + } + + private static function nextId(): int { + self::$requestId = self::$requestId >= 0x7fffffff ? 0x4e470000 : self::$requestId + 1; + return self::$requestId; + } +} diff --git a/lib/Backend/StorageBackend.php b/lib/Backend/StorageBackend.php index 58f53e2..80f4c7d 100644 --- a/lib/Backend/StorageBackend.php +++ b/lib/Backend/StorageBackend.php @@ -17,6 +17,8 @@ use OCP\Files\Storage\IWriteStreamStorage; final class StorageBackend implements IBackend { /** 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. */ + private bool $sftpReads; public function __construct( private IStorage $storage, @@ -25,6 +27,7 @@ final class StorageBackend implements IBackend { $this->base = trim($base, '/'); self::validate($this->base, true); $this->atomicPut = $storage->instanceOfStorage(\OCA\Files_External\Lib\Storage\AmazonS3::class); + $this->sftpReads = $storage->instanceOfStorage(\OCA\Files_External\Lib\Storage\SFTP::class); if ($this->base !== '') { $this->mkdirs($this->base); } @@ -79,6 +82,16 @@ final class StorageBackend implements IBackend { } public function get(string $path) { + if ($this->sftpReads) { + $fh = SftpReadStream::open($this->storage, $this->abs($path)); + 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 + } $fh = $this->storage->fopen($this->abs($path), 'r'); if ($fh === false) { throw new BackendException('Not found: ' . $path); diff --git a/tests/Unit/Backend/SftpReadStreamTest.php b/tests/Unit/Backend/SftpReadStreamTest.php new file mode 100644 index 0000000..3d40b4a --- /dev/null +++ b/tests/Unit/Backend/SftpReadStreamTest.php @@ -0,0 +1,243 @@ + $data]); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'repo/packs/ab/abcd'); + + $this->assertIsResource($fh); + $this->assertSame($data, stream_get_contents($fh)); + $this->assertTrue(feof($fh)); + fclose($fh); + $this->assertSame(SftpReadStream::DEPTH, $sftp->maxInFlight); + $this->assertSame(0, $sftp->pending(), 'nothing may be left on the connection after close'); + $this->assertSame([], $sftp->openHandles); + } + + public function testAServerThatCapsAReadStillGivesEveryByteInOrder(): void { + $data = random_bytes(3 * 1048576 + 7); + $sftp = new FakeSftpConnection(['/home/f' => $data], cap: 100000); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + + $this->assertSame($data, stream_get_contents($fh)); + fclose($fh); + $this->assertSame(0, $sftp->pending()); + + // The cap is learned: a next stream asks for no more than the server gives. + $sftp->reads = []; + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + $this->assertSame($data, stream_get_contents($fh)); + fclose($fh); + $this->assertSame([100000], array_values(array_unique($sftp->reads))); + + // Only for that connection. + $other = new FakeSftpConnection(['/home/f' => $data]); + $fh = SftpReadStream::open(new FakeSftpStorage($other), 'f'); + $this->assertSame($data, stream_get_contents($fh)); + fclose($fh); + $this->assertSame(261120, max($other->reads)); + } + + public function testATinyAnswerDoesNotShrinkTheReadsBelowWhatEveryServerSupports(): void { + $data = random_bytes(1048576); + $sftp = new FakeSftpConnection(['/home/f' => $data], cap: 1000); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + $this->assertSame($data, stream_get_contents($fh)); + fclose($fh); + $this->assertSame(32768, min($sftp->reads)); + } + + public function testSeekingReadsFromTheNewOffset(): void { + $data = random_bytes(3 * 1048576); + $sftp = new FakeSftpConnection(['/home/f' => $data]); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + + $this->assertSame(0, fseek($fh, 2000000)); + $this->assertSame(2000000, ftell($fh)); + $this->assertSame(substr($data, 2000000, 1000), fread($fh, 1000)); + $this->assertSame(0, fseek($fh, 10)); // backwards + $this->assertSame(substr($data, 10, 70000), $this->readExactly($fh, 70000)); + $this->assertSame(0, fseek($fh, 100, SEEK_CUR)); // within what is already buffered + $this->assertSame(substr($data, 70110, 5000), $this->readExactly($fh, 5000)); + fclose($fh); + $this->assertSame(0, $sftp->pending()); + } + + public function testASmallReadDoesNotFetchTheWholePipelineAhead(): void { + $sftp = new FakeSftpConnection(['/home/f' => random_bytes(10 * 1048576)]); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + fread($fh, 100); + fclose($fh); + $this->assertCount(1, $sftp->reads, 'the first request goes alone, the pipeline grows while reading on'); + } + + public function testAnEmptyFileIsAtEndAtOnce(): void { + $sftp = new FakeSftpConnection(['/home/f' => '']); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + $this->assertSame('', stream_get_contents($fh)); + $this->assertTrue(feof($fh)); + fclose($fh); + } + + public function testAMissingFileCannotBeOpened(): void { + $sftp = new FakeSftpConnection([]); + $this->assertFalse(SftpReadStream::open(new FakeSftpStorage($sftp), 'nope')); + $this->assertSame(0, $sftp->pending()); + } + + public function testAServerErrorFailsTheReadAndLeavesTheConnectionUsable(): void { + $sftp = new FakeSftpConnection(['/home/f' => random_bytes(3 * 1048576), '/home/g' => 'second'], failAt: 1048576); + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'f'); + $read = ''; + while (!feof($fh)) { + $chunk = fread($fh, 65536); + if ($chunk === false || $chunk === '') { + break; + } + $read .= $chunk; + } + $this->assertLessThan(3 * 1048576, strlen($read)); + fclose($fh); + $this->assertSame(0, $sftp->pending()); + + $fh = SftpReadStream::open(new FakeSftpStorage($sftp), 'g'); + $this->assertSame('second', stream_get_contents($fh)); + fclose($fh); + } + + public function testAConnectionWithoutTheNeededMethodsIsNotUsed(): void { + $storage = new class { + public function getRoot(): string { + return '/'; + } + + public function getConnection(): object { + return new \stdClass(); + } + }; + $this->assertNull(SftpReadStream::open($storage, 'f')); + $this->assertNull(SftpReadStream::open(new \stdClass(), 'f')); + } + + /** @param resource $fh */ + private function readExactly($fh, int $length): string { + $data = ''; + while (strlen($data) < $length && !feof($fh)) { + $data .= fread($fh, $length - strlen($data)); + } + return $data; + } +} + +final class FakeSftpStorage { + public function __construct( + private FakeSftpConnection $connection, + ) { + } + + public function getRoot(): string { + return '/home/'; + } + + public function getConnection(): FakeSftpConnection { + return $this->connection; + } +} + +/** The part of phpseclib\Net\SFTP (2.x) that SftpReadStream uses. */ +final class FakeSftpConnection { + public int $packet_type = -1; + public int $maxInFlight = 0; + /** @var list lengths asked for by READ requests */ + public array $reads = []; + /** @var array */ + public array $openHandles = []; + /** @var array answers by request id, in the order the server sent them */ + private array $answers = []; + private int $handles = 0; + + /** @param array $files */ + public function __construct( + private array $files, + private int $cap = 261120, + private ?int $failAt = null, + ) { + } + + public function pending(): int { + return count($this->answers); + } + + public function _realpath(string $path): string { + return '/' . trim(preg_replace('#/+#', '/', $path), '/'); + } + + public function _send_sftp_packet(int $type, string $data, int $id = 1): bool { + switch ($type) { + case 3: // OPEN + $path = substr($data, 4, unpack('N', $data)[1]); + if (!isset($this->files[$path])) { + $this->answers[$id] = [101, pack('N', 2)]; + break; + } + $handle = 'h' . ++$this->handles; + $this->openHandles[$handle] = $path; + $this->answers[$id] = [102, pack('Na*', strlen($handle), $handle)]; + break; + case 4: // CLOSE + $handle = substr($data, 4, unpack('N', $data)[1]); + unset($this->openHandles[$handle]); + $this->answers[$id] = [101, pack('N', 0)]; + break; + case 5: // READ + $len = unpack('N', $data)[1]; + $handle = substr($data, 4, $len); + ['hi' => $hi, 'lo' => $lo, 'n' => $n] = unpack('Nhi/Nlo/Nn', substr($data, 4 + $len)); + $offset = $hi * 4294967296 + $lo; + $this->reads[] = $n; + $file = $this->files[$this->openHandles[$handle]]; + if ($this->failAt !== null && $offset >= $this->failAt) { + $this->answers[$id] = [101, pack('N', 4)]; // FX_FAILURE + } elseif ($offset >= strlen($file)) { + $this->answers[$id] = [101, pack('N', 1)]; // FX_EOF + } else { + $chunk = substr($file, $offset, min($n, $this->cap)); + $this->answers[$id] = [103, pack('Na*', strlen($chunk), $chunk)]; + } + $this->maxInFlight = max($this->maxInFlight, count($this->answers)); + break; + default: + throw new \LogicException("unexpected packet type $type"); + } + return true; + } + + public function _get_sftp_packet(?int $id = null): string|false { + if ($this->answers === []) { + throw new \LogicException('waiting for an answer that will never come'); + } + if ($id === null || !isset($this->answers[$id])) { + throw new \LogicException("no answer for request $id"); + } + [$this->packet_type, $payload] = $this->answers[$id]; + unset($this->answers[$id]); + return $payload; + } +}