Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 24 additions & 3 deletions lib/core/extensions/extension_driver_session.dart
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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');
Expand Down Expand Up @@ -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,
Expand Down
6 changes: 6 additions & 0 deletions lib/core/extensions/models/sandbox_capabilities.dart
Original file line number Diff line number Diff line change
Expand Up @@ -118,6 +118,7 @@ class ResourceLimits {
const ResourceLimits({
this.memoryMb = defaultMemoryMb,
this.maxOpenFiles = defaultMaxOpenFiles,
this.timeoutSeconds,
});

static const defaultMemoryMb = 256;
Expand All @@ -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,
};
}

Expand Down
12 changes: 11 additions & 1 deletion lib/core/extensions/rpc/json_rpc_stdio_client.dart
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
16 changes: 14 additions & 2 deletions lib/core/extensions/rpc/plugin_rpc_bridge.dart
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand All @@ -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;

Expand Down Expand Up @@ -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));
}
},
);
Expand All @@ -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',
);
}
}
Expand Down Expand Up @@ -212,6 +223,7 @@ class PluginRpcBridge {
pluginId: _handle?.pluginId ?? 'unknown',
exitCode: code,
));
onProcessExited?.call(code);
unawaited(_disposeLocal(keepRecovery: true));
}

Expand Down
17 changes: 17 additions & 0 deletions lib/core/extensions/rpc/plugin_rpc_exceptions.dart
Original file line number Diff line number Diff line change
Expand Up @@ -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';
}
}
9 changes: 9 additions & 0 deletions lib/core/extensions/sandbox/sandbox_stderr_pipe.dart
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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');
Expand Down
14 changes: 14 additions & 0 deletions test/core/extensions/extension_driver_session_test.dart
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand Down Expand Up @@ -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);
});
});
}
104 changes: 104 additions & 0 deletions test/core/extensions/rpc/plugin_rpc_bridge_test.dart
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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();
});
}
Loading