Merge pull request #668 from QueryaHub/issue/666-plugin-rpc-bridge-resilience

fix(extensions): harden plugin rpc bridge against stdin flush hangs and manifest request timeouts

Author: ZhuchkaTriplesix

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<PluginRpcBridge> _startBridge(ExtensionManifest manifest) async {
+  Future<PluginRpcBridge> _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,

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<String, dynamic> 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<String, dynamic> toJson() => {
         'memory_mb': memoryMb,
         'max_open_files': maxOpenFiles,
+        if (timeoutSeconds != null) 'timeout_seconds': timeoutSeconds,
       };
 }
 

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);

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));
   }
 

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';
+  }
+}

lib/core/extensions/sandbox/sandbox_stderr_pipe.dart

@@ -39,6 +39,11 @@ class SandboxStderrPipe {
   final int maxCarryChars;
   final int maxSanitizeChars;
 
+  final List<String> _recentLines = [];
+
+  /// Rolling buffer of recent sanitized stderr lines (last 20 lines max).
+  List<String> get recentLines => List.unmodifiable(_recentLines);
+
   StreamSubscription<List<int>>? _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');

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);
+    });
   });
 }

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<String, dynamic>;
+      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<String, String>? 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<void>.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<String, String>? 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<PluginProtocolTimeoutException>().having(
+          (e) => e.message,
+          'message',
+          contains('FATAL: libclickhouse.so not found'),
+        ),
+      ),
+    );
+    await sub.cancel();
+  });
 }