From 9957075edd48c9dd8ad829ce926be8c5ae274777 Mon Sep 17 00:00:00 2001 From: ZhuchkaTriplesix Date: Sun, 30 Aug 2026 18:37:32 +0300 Subject: [PATCH] fix(extensions): harden plugin rpc bridge against stdin flush hangs and manifest request timeouts (close #666) --- .../extensions/extension_driver_session.dart | 27 ++++- .../models/sandbox_capabilities.dart | 6 + .../extensions/rpc/json_rpc_stdio_client.dart | 12 +- .../extensions/rpc/plugin_rpc_bridge.dart | 16 ++- .../extensions/rpc/plugin_rpc_exceptions.dart | 17 +++ .../sandbox/sandbox_stderr_pipe.dart | 9 ++ .../extension_driver_session_test.dart | 14 +++ .../rpc/plugin_rpc_bridge_test.dart | 104 ++++++++++++++++++ 8 files changed, 199 insertions(+), 6 deletions(-) diff --git a/lib/core/extensions/extension_driver_session.dart b/lib/core/extensions/extension_driver_session.dart index 47b0c3f0..74e6c789 100644 --- a/lib/core/extensions/extension_driver_session.dart +++ b/lib/core/extensions/extension_driver_session.dart @@ -52,7 +52,16 @@ class ExtensionDriverSession { final manifest = await _resolveManifestForRow(row); final hydrated = await _hydrateSecrets(row); - final bridge = await _startBridge(manifest); + late final PluginRpcBridge bridge; + bridge = await _startBridge( + manifest, + onProcessExited: (code) { + if (_bridges[id] == bridge) { + _bridges.remove(id); + _manifests.remove(id); + } + }, + ); try { await _injectAndConnect(bridge, connectionId: id, row: hydrated); @@ -99,7 +108,10 @@ class ExtensionDriverSession { } } - Future _startBridge(ExtensionManifest manifest) async { + Future _startBridge( + ExtensionManifest manifest, { + void Function(int code)? onProcessExited, + }) async { final root = manifest.installPath; if (root == null || root.isEmpty) { throw StateError('Extension "${manifest.id}" has no install path'); @@ -127,7 +139,16 @@ class ExtensionDriverSession { } } - final bridge = bridgeFactory?.call() ?? PluginRpcBridge(); + final manifestTimeoutSeconds = + manifest.sandbox?.resources.timeoutSeconds; + final bridge = bridgeFactory?.call() ?? + PluginRpcBridge( + requestTimeout: + manifestTimeoutSeconds != null && manifestTimeoutSeconds > 0 + ? Duration(seconds: manifestTimeoutSeconds) + : const Duration(seconds: 60), + onProcessExited: onProcessExited, + ); await _startBridgeWithConsent( bridge: bridge, manifest: manifest, diff --git a/lib/core/extensions/models/sandbox_capabilities.dart b/lib/core/extensions/models/sandbox_capabilities.dart index 3037e839..8ec210bd 100644 --- a/lib/core/extensions/models/sandbox_capabilities.dart +++ b/lib/core/extensions/models/sandbox_capabilities.dart @@ -118,6 +118,7 @@ class ResourceLimits { const ResourceLimits({ this.memoryMb = defaultMemoryMb, this.maxOpenFiles = defaultMaxOpenFiles, + this.timeoutSeconds, }); static const defaultMemoryMb = 256; @@ -129,16 +130,21 @@ class ResourceLimits { /// Maximum number of open file descriptors (`ulimit -n`). final int maxOpenFiles; + /// Optional custom request/query timeout declared in manifest, in seconds. + final int? timeoutSeconds; + factory ResourceLimits.fromJson(Map json) { return ResourceLimits( memoryMb: json['memory_mb'] as int? ?? defaultMemoryMb, maxOpenFiles: json['max_open_files'] as int? ?? defaultMaxOpenFiles, + timeoutSeconds: json['timeout_seconds'] as int?, ); } Map toJson() => { 'memory_mb': memoryMb, 'max_open_files': maxOpenFiles, + if (timeoutSeconds != null) 'timeout_seconds': timeoutSeconds, }; } diff --git a/lib/core/extensions/rpc/json_rpc_stdio_client.dart b/lib/core/extensions/rpc/json_rpc_stdio_client.dart index 4e0e5445..d1cb9537 100644 --- a/lib/core/extensions/rpc/json_rpc_stdio_client.dart +++ b/lib/core/extensions/rpc/json_rpc_stdio_client.dart @@ -74,7 +74,17 @@ class JsonRpcStdioClient { }; _stdin.writeln(jsonEncode(payload)); - await _stdin.flush(); + try { + await _stdin.flush().timeout(const Duration(seconds: 3)); + } on TimeoutException { + _pending.remove(id); + throw TimeoutException( + 'Failed to flush JSON-RPC request "$method" to plugin stdin within 3s', + ); + } catch (e) { + _pending.remove(id); + rethrow; + } try { return await completer.future.timeout(requestTimeout); diff --git a/lib/core/extensions/rpc/plugin_rpc_bridge.dart b/lib/core/extensions/rpc/plugin_rpc_bridge.dart index 4b0a61b9..1b4fbdec 100644 --- a/lib/core/extensions/rpc/plugin_rpc_bridge.dart +++ b/lib/core/extensions/rpc/plugin_rpc_bridge.dart @@ -18,11 +18,12 @@ import 'package:querya_desktop/core/extensions/sandbox/sandbox_auto_recovery.dar class PluginRpcBridge { PluginRpcBridge({ SandboxProcessRunner? processRunner, - this.handshakeTimeout = const Duration(seconds: 3), + this.handshakeTimeout = const Duration(seconds: 6), this.shutdownTimeout = const Duration(seconds: 3), this.requestTimeout = const Duration(seconds: 30), this.enableWatchdog = true, this.enableStderrPipe = true, + this.onProcessExited, SandboxSecurityAudit? audit, SandboxAutoRecovery? recovery, }) : _runner = processRunner ?? SandboxProcessRunner(), @@ -35,6 +36,7 @@ class PluginRpcBridge { final Duration requestTimeout; final bool enableWatchdog; final bool enableStderrPipe; + final void Function(int exitCode)? onProcessExited; final SandboxSecurityAudit? _audit; final SandboxAutoRecovery _recovery; @@ -125,6 +127,11 @@ class PluginRpcBridge { pluginId: handle.pluginId, detail: 'watchdog deadlock', )); + _failPending(PluginDeadlockException( + pluginId: handle.pluginId, + message: 'Watchdog detected plugin deadlock (ping timeout)', + )); + unawaited(_disposeLocal(keepRecovery: true)); } }, ); @@ -138,9 +145,13 @@ class PluginRpcBridge { _recovery.recordSuccess(); return result; } on TimeoutException { + final recent = _stderrPipe?.recentLines ?? const []; + final stderrTail = recent.isNotEmpty + ? '\nStderr output:\n${recent.join('\n')}' + : ''; await _forceKill(); throw PluginProtocolTimeoutException( - 'system.handshake timed out after $handshakeTimeout', + 'system.handshake timed out after $handshakeTimeout$stderrTail', ); } } @@ -212,6 +223,7 @@ class PluginRpcBridge { pluginId: _handle?.pluginId ?? 'unknown', exitCode: code, )); + onProcessExited?.call(code); unawaited(_disposeLocal(keepRecovery: true)); } diff --git a/lib/core/extensions/rpc/plugin_rpc_exceptions.dart b/lib/core/extensions/rpc/plugin_rpc_exceptions.dart index 257d9b4c..175fcfed 100644 --- a/lib/core/extensions/rpc/plugin_rpc_exceptions.dart +++ b/lib/core/extensions/rpc/plugin_rpc_exceptions.dart @@ -26,3 +26,20 @@ class PluginProtocolTimeoutException implements Exception { @override String toString() => 'PluginProtocolTimeoutException: $message'; } + +/// Thrown when the watchdog detects that the plugin process is deadlocked (ping timeout). +class PluginDeadlockException implements Exception { + PluginDeadlockException({ + required this.pluginId, + this.message, + }); + + final String pluginId; + final String? message; + + @override + String toString() { + final detail = message == null ? '' : ': $message'; + return 'PluginDeadlockException($pluginId)$detail'; + } +} diff --git a/lib/core/extensions/sandbox/sandbox_stderr_pipe.dart b/lib/core/extensions/sandbox/sandbox_stderr_pipe.dart index 5eb1f8b0..63e8f1a9 100644 --- a/lib/core/extensions/sandbox/sandbox_stderr_pipe.dart +++ b/lib/core/extensions/sandbox/sandbox_stderr_pipe.dart @@ -39,6 +39,11 @@ class SandboxStderrPipe { final int maxCarryChars; final int maxSanitizeChars; + final List _recentLines = []; + + /// Rolling buffer of recent sanitized stderr lines (last 20 lines max). + List get recentLines => List.unmodifiable(_recentLines); + StreamSubscription>? _subscription; final StringBuffer _carry = StringBuffer(); int _carryLength = 0; @@ -193,6 +198,10 @@ class SandboxStderrPipe { ); } onSanitizedLine?.call(sanitized); + _recentLines.add(sanitized); + if (_recentLines.length > 20) { + _recentLines.removeAt(0); + } await log.appendLine(sanitized); } catch (e, st) { debugPrint('SandboxStderrPipe($pluginId) write failed: $e\n$st'); diff --git a/test/core/extensions/extension_driver_session_test.dart b/test/core/extensions/extension_driver_session_test.dart index 18301551..0c16ec6f 100644 --- a/test/core/extensions/extension_driver_session_test.dart +++ b/test/core/extensions/extension_driver_session_test.dart @@ -3,6 +3,7 @@ import 'package:querya_desktop/core/extensions/extension_driver_session.dart'; import 'package:querya_desktop/core/extensions/models/extension_driver_capabilities.dart'; import 'package:querya_desktop/core/extensions/models/extension_object_metadata.dart'; import 'package:querya_desktop/core/extensions/models/extension_server_stats.dart'; +import 'package:querya_desktop/core/extensions/models/sandbox_capabilities.dart'; import 'package:querya_desktop/core/storage/local_db.dart'; void main() { @@ -136,5 +137,18 @@ void main() { expect(metadata.columns[0].comment, 'Primary ID'); expect(metadata.properties['engine'], 'MergeTree'); }); + + test('ResourceLimits parses timeout_seconds correctly', () { + final limits = ResourceLimits.fromJson({ + 'memory_mb': 512, + 'max_open_files': 128, + 'timeout_seconds': 600, + }); + + expect(limits.memoryMb, 512); + expect(limits.maxOpenFiles, 128); + expect(limits.timeoutSeconds, 600); + expect(limits.toJson()['timeout_seconds'], 600); + }); }); } diff --git a/test/core/extensions/rpc/plugin_rpc_bridge_test.dart b/test/core/extensions/rpc/plugin_rpc_bridge_test.dart index c7bee5e8..f864b30e 100644 --- a/test/core/extensions/rpc/plugin_rpc_bridge_test.dart +++ b/test/core/extensions/rpc/plugin_rpc_bridge_test.dart @@ -8,6 +8,7 @@ import 'package:querya_desktop/core/extensions/models/extension_type.dart'; import 'package:querya_desktop/core/extensions/models/sandbox_capabilities.dart'; import 'package:querya_desktop/core/extensions/rpc/plugin_rpc_bridge.dart'; import 'package:querya_desktop/core/extensions/rpc/plugin_rpc_exceptions.dart'; +import 'package:querya_desktop/core/extensions/sandbox/sandbox_log_paths.dart'; import 'package:querya_desktop/core/extensions/sandbox/sandbox_process_runner.dart'; class _FakeProcess implements Process { @@ -66,9 +67,11 @@ void main() { setUp(() async { tempBase = await Directory.systemTemp.createTemp('querya_rpc_bridge_'); + SandboxLogPaths.mockLogsDirectory = tempBase; }); tearDown(() async { + SandboxLogPaths.mockLogsDirectory = null; try { if (await tempBase.exists()) { await tempBase.delete(recursive: true); @@ -316,4 +319,105 @@ void main() { expect(bridge.isStarted, isFalse); await sub.cancel(); }); + + test('unexpected exit fires onProcessExited callback', () async { + final process = _FakeProcess(); + final sub = process.stdinLines.listen((line) { + final req = jsonDecode(line) as Map; + if (req['method'] == 'system.handshake') { + process.reply({ + 'jsonrpc': '2.0', + 'id': req['id'] as int, + 'result': {'ok': true}, + }); + } + }); + + final runner = SandboxProcessRunner( + platformOverride: 'windows', + scratchBaseDirectory: tempBase, + processStarter: ( + exe, + args, { + String? workingDirectory, + Map? environment, + bool includeParentEnvironment = true, + bool runInShell = false, + ProcessStartMode mode = ProcessStartMode.normal, + }) async => + process, + ); + + int? capturedExitCode; + final bridge = PluginRpcBridge( + processRunner: runner, + enableWatchdog: false, + enableStderrPipe: false, + onProcessExited: (code) { + capturedExitCode = code; + }, + ); + + await bridge.start( + manifest: testManifest, + pluginExecutable: '/opt/driver', + allowUnsandboxedLaunch: true, + ); + + process.completeExit(137); // SIGKILL + await Future.delayed(const Duration(milliseconds: 20)); + + expect(capturedExitCode, 137); + expect(bridge.isStarted, isFalse); + await sub.cancel(); + }); + + test('handshake timeout includes recent stderr output in exception', () async { + final process = _FakeProcess(); + final sub = process.stdinLines.listen((_) { + process._stderrController.add( + utf8.encode('FATAL: libclickhouse.so not found\n'), + ); + }); + + final runner = SandboxProcessRunner( + platformOverride: 'windows', + scratchBaseDirectory: tempBase, + processStarter: ( + exe, + args, { + String? workingDirectory, + Map? environment, + bool includeParentEnvironment = true, + bool runInShell = false, + ProcessStartMode mode = ProcessStartMode.normal, + }) async => + process, + ); + + final bridge = PluginRpcBridge( + processRunner: runner, + handshakeTimeout: const Duration(milliseconds: 50), + enableWatchdog: false, + enableStderrPipe: true, + ); + + final startFuture = bridge.start( + manifest: testManifest, + pluginExecutable: '/opt/driver', + allowUnsandboxedLaunch: true, + ); + + await expectLater( + startFuture, + throwsA( + isA().having( + (e) => e.message, + 'message', + contains('FATAL: libclickhouse.so not found'), + ), + ), + ); + await sub.cancel(); + }); }