diff --git a/docs/tz-block-c-rpc-bridge.md b/docs/tz-block-c-rpc-bridge.md index 1d0d6b36..609c9a2b 100644 --- a/docs/tz-block-c-rpc-bridge.md +++ b/docs/tz-block-c-rpc-bridge.md @@ -46,7 +46,15 @@ final result = await rpcClient.sendRequest('db.connect', credentialsMap); --- -## 4. Контракт Ошибок (Error Mapping) -Плагин должен возвращать ошибки согласно спецификации JSON-RPC. RPC Bridge должен уметь парсить эти ошибки и превращать их в понятные Dart-exceptions: -- Ошибка подключения (Timeout, Wrong Password) -> Показывается в UI в красном Snackbar. -- Синтаксическая ошибка SQL -> Выделяется красным в SQL редакторе. +## 5. Лимиты полезной нагрузки (NDJSON) + +Каждый ответ — **одна JSON-строка** на `stdout` (newline-delimited). Хост (`JsonRpcStdioClient`) применяет: + +| Лимит | Значение по умолчанию | Поведение | +|-------|----------------------|-----------| +| Макс. длина одной строки ответа | **32 MiB** UTF-8 | Fail closed: `JsonRpcPayloadTooLargeException`, все pending RPC завершаются ошибкой | +| Декод больших строк | **> 64 KiB** | `jsonDecode` уходит в isolate | + +Для `db.query` хост всегда передаёт `params.limit` (Preferences → Max rows in results), чтобы драйвер обрезал результат **до** сериализации. Драйверы обязаны уважать `limit`. + +Чанкованный / бинарный framing для очень больших выборок — follow-up; до него bound + `limit` обязательны. diff --git a/lib/core/extensions/extension_driver_session.dart b/lib/core/extensions/extension_driver_session.dart index ee8a3d7a..a082edde 100644 --- a/lib/core/extensions/extension_driver_session.dart +++ b/lib/core/extensions/extension_driver_session.dart @@ -14,6 +14,7 @@ import 'package:querya_desktop/core/extensions/rpc/plugin_rpc_bridge.dart'; import 'package:querya_desktop/core/extensions/sandbox/sandbox_os_isolation.dart'; import 'package:querya_desktop/core/extensions/sandbox/unsandboxed_launch_consent_gate.dart'; import 'package:querya_desktop/core/sdui/sdui_tree_schema.dart'; +import 'package:querya_desktop/core/storage/app_settings.dart'; import 'package:querya_desktop/core/storage/connection_secrets_store.dart'; import 'package:querya_desktop/core/storage/local_db.dart'; @@ -276,16 +277,21 @@ class ExtensionDriverSession { } /// Executes SQL through the plugin (`db.query`) and returns the raw result. + /// + /// When [limit] is omitted, Preferences **Max rows in results** is sent so + /// drivers can bound the NDJSON response before it hits the host. Future query( ConnectionRow row, String sql, { int? limit, }) async { final bridge = await ensureConnected(row); + final effectiveLimit = + limit ?? await AppSettings.instance.getSqlResultMaxRows(); final result = await bridge.sendRequest('db.query', { 'connectionId': row.id, 'sql': sql, - if (limit != null) 'limit': limit, + 'limit': effectiveLimit, }); return compute(_parseExtensionQueryResultRpc, result); } diff --git a/lib/core/extensions/rpc/json_rpc_payload_limits.dart b/lib/core/extensions/rpc/json_rpc_payload_limits.dart new file mode 100644 index 00000000..72750292 --- /dev/null +++ b/lib/core/extensions/rpc/json_rpc_payload_limits.dart @@ -0,0 +1,114 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:typed_data'; + +/// Default max UTF-8 byte length of one JSON-RPC stdout line (NDJSON). +/// +/// Large `db.query` results are one object per line; without a bound the host +/// can OOM before Preferences row caps apply. Drivers should honor `limit`. +const int kDefaultJsonRpcMaxLineBytes = 32 * 1024 * 1024; + +/// Lines above this UTF-8 length are `jsonDecode`d off the UI isolate. +const int kJsonRpcOffIsolateDecodeThresholdBytes = 64 * 1024; + +/// Thrown when a plugin emits a newline-delimited JSON line larger than the +/// configured maximum. +class JsonRpcPayloadTooLargeException implements Exception { + JsonRpcPayloadTooLargeException({ + required this.maxLineBytes, + required this.receivedBytes, + }); + + final int maxLineBytes; + final int receivedBytes; + + @override + String toString() => + 'JsonRpcPayloadTooLargeException: JSON-RPC line is $receivedBytes bytes ' + '(max $maxLineBytes). Reduce result size or pass a smaller `limit`.'; +} + +/// Splits a byte stream into UTF-8 lines, failing closed if any line exceeds +/// [maxLineBytes] (counted before decode). +StreamTransformer, String> boundedUtf8LineSplitter({ + int maxLineBytes = kDefaultJsonRpcMaxLineBytes, +}) { + return _BoundedUtf8LineSplitter(maxLineBytes: maxLineBytes); +} + +class _BoundedUtf8LineSplitter + extends StreamTransformerBase, String> { + _BoundedUtf8LineSplitter({required this.maxLineBytes}); + + final int maxLineBytes; + + @override + Stream bind(Stream> stream) { + final controller = StreamController(sync: true); + final pending = BytesBuilder(copy: false); + late final StreamSubscription> sub; + + void fail(Object error, [StackTrace? st]) { + if (!controller.isClosed) { + controller.addError(error, st); + controller.close(); + } + sub.cancel(); + } + + void emitLine() { + var bytes = pending.takeBytes(); + if (bytes.isNotEmpty && bytes.last == 0x0d) { + bytes = Uint8List.sublistView(bytes, 0, bytes.length - 1); + } + if (bytes.isEmpty) return; + try { + controller.add(utf8.decode(bytes)); + } catch (e, st) { + fail(e, st); + } + } + + sub = stream.listen( + (chunk) { + for (var i = 0; i < chunk.length; i++) { + final b = chunk[i]; + if (b == 0x0a) { + emitLine(); + continue; + } + if (pending.length >= maxLineBytes) { + fail( + JsonRpcPayloadTooLargeException( + maxLineBytes: maxLineBytes, + receivedBytes: pending.length + 1, + ), + ); + return; + } + pending.addByte(b); + } + }, + onError: fail, + onDone: () { + if (pending.length > 0) { + if (pending.length > maxLineBytes) { + fail( + JsonRpcPayloadTooLargeException( + maxLineBytes: maxLineBytes, + receivedBytes: pending.length, + ), + ); + return; + } + emitLine(); + } + controller.close(); + }, + cancelOnError: true, + ); + + controller.onCancel = () => sub.cancel(); + return controller.stream; + } +} diff --git a/lib/core/extensions/rpc/json_rpc_stdio_client.dart b/lib/core/extensions/rpc/json_rpc_stdio_client.dart index e277ee2d..130f56f5 100644 --- a/lib/core/extensions/rpc/json_rpc_stdio_client.dart +++ b/lib/core/extensions/rpc/json_rpc_stdio_client.dart @@ -1,24 +1,39 @@ import 'dart:async'; import 'dart:convert'; import 'dart:io'; +import 'dart:isolate'; + +import 'package:querya_desktop/core/extensions/rpc/json_rpc_payload_limits.dart'; /// Minimal JSON-RPC 2.0 client over newline-delimited JSON on stdio. /// /// Enough for Block E credential injection and later Block C methods without /// pulling `json_rpc_2` yet. One JSON object per line on stdin/stdout. +/// +/// Incoming lines are bounded by [maxLineBytes] (see +/// [kDefaultJsonRpcMaxLineBytes]); oversized payloads fail closed. class JsonRpcStdioClient { JsonRpcStdioClient({ required Stream> stdout, required IOSink stdin, this.requestTimeout = const Duration(seconds: 10), + this.maxLineBytes = kDefaultJsonRpcMaxLineBytes, }) : _stdin = stdin, - _lines = utf8.decoder.bind(stdout).transform(const LineSplitter()) { - _subscription = _lines.listen(_onLine, onError: _onError, onDone: _onDone); + _lines = stdout.transform( + boundedUtf8LineSplitter(maxLineBytes: maxLineBytes), + ) { + _subscription = _lines.listen( + _onLine, + onError: _onError, + onDone: _onDone, + cancelOnError: false, + ); } final IOSink _stdin; final Stream _lines; final Duration requestTimeout; + final int maxLineBytes; final Map> _pending = {}; var _nextId = 1; @@ -26,6 +41,9 @@ class JsonRpcStdioClient { StreamSubscription? _subscription; Object? _fatalError; + /// Serializes async line handling so large-line isolate decode stays ordered. + Future _lineChain = Future.value(); + /// Sends a JSON-RPC request and waits for the matching response. Future sendRequest( String method, [ @@ -82,12 +100,21 @@ class JsonRpcStdioClient { } void _onLine(String line) { + _lineChain = _lineChain.then((_) => _handleLine(line)); + } + + Future _handleLine(String line) async { if (line.trim().isEmpty) return; late final Map message; try { - final decoded = jsonDecode(line); - if (decoded is! Map) return; - message = decoded; + final Object decoded; + if (line.length > kJsonRpcOffIsolateDecodeThresholdBytes) { + decoded = await Isolate.run(() => jsonDecode(line)); + } else { + decoded = jsonDecode(line); + } + if (decoded is! Map) return; + message = Map.from(decoded); } catch (_) { return; } diff --git a/lib/features/extensions/extension_sql_workspace.dart b/lib/features/extensions/extension_sql_workspace.dart index a1dce00b..25dd8c25 100644 --- a/lib/features/extensions/extension_sql_workspace.dart +++ b/lib/features/extensions/extension_sql_workspace.dart @@ -50,6 +50,7 @@ class _ExtensionSqlWorkspaceState String? _statusLine; int _historyMaxEntries = kDefaultSqlHistoryMaxEntries; + int _resultMaxRows = kDefaultSqlResultMaxRows; double _editorFontSize = kDefaultSqlEditorFontSize; static const _previewRowLimit = 200; @@ -89,10 +90,12 @@ class _ExtensionSqlWorkspaceState Future _loadWorkspaceSettings() async { final hist = await AppSettings.instance.getSqlHistoryMaxEntries(); + final rows = await AppSettings.instance.getSqlResultMaxRows(); final font = await AppSettings.instance.getSqlEditorFontSize(); if (!mounted) return; setState(() { _historyMaxEntries = hist; + _resultMaxRows = rows; _editorFontSize = font; }); } @@ -124,8 +127,11 @@ class _ExtensionSqlWorkspaceState }); try { - final result = await ExtensionDriverSession.instance - .query(widget.connectionRow, userSql); + final result = await ExtensionDriverSession.instance.query( + widget.connectionRow, + userSql, + limit: _resultMaxRows, + ); if (!mounted) return; setState(() { @@ -136,7 +142,10 @@ class _ExtensionSqlWorkspaceState } else { final elapsed = result.elapsedMs != null ? ' in ${result.elapsedMs}ms' : ''; - _statusLine = '${result.rows.length} row(s)$elapsed.'; + final capped = result.rows.length >= _resultMaxRows; + _statusLine = capped + ? 'Showing first $_resultMaxRows row(s)$elapsed (result capped).' + : '${result.rows.length} row(s)$elapsed.'; } _running = false; }); diff --git a/test/core/extensions/rpc/json_rpc_stdio_client_test.dart b/test/core/extensions/rpc/json_rpc_stdio_client_test.dart new file mode 100644 index 00000000..b8d454e6 --- /dev/null +++ b/test/core/extensions/rpc/json_rpc_stdio_client_test.dart @@ -0,0 +1,88 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:querya_desktop/core/extensions/rpc/json_rpc_payload_limits.dart'; +import 'package:querya_desktop/core/extensions/rpc/json_rpc_stdio_client.dart'; + +void main() { + group('boundedUtf8LineSplitter', () { + test('splits lines and strips CR', () async { + final lines = await Stream>.fromIterable([ + utf8.encode('one\r\n'), + utf8.encode('two\n'), + ]).transform(boundedUtf8LineSplitter(maxLineBytes: 1024)).toList(); + expect(lines, ['one', 'two']); + }); + + test('fails closed when line exceeds max bytes', () async { + final controller = StreamController>(); + final errors = []; + final sub = controller.stream + .transform(boundedUtf8LineSplitter(maxLineBytes: 8)) + .listen((_) {}, onError: errors.add); + + controller.add(utf8.encode('123456789')); // 9 bytes, no newline yet + await Future.delayed(Duration.zero); + expect(errors, isNotEmpty); + expect(errors.first, isA()); + await sub.cancel(); + await controller.close(); + }); + }); + + group('JsonRpcStdioClient payload bounds', () { + test('completes pending request with payload-too-large error', () async { + final stdout = StreamController>(); + final stdin = StreamController>(); + final client = JsonRpcStdioClient( + stdout: stdout.stream, + stdin: IOSink(stdin.sink), + maxLineBytes: 32, + requestTimeout: const Duration(seconds: 2), + ); + + final pending = client.sendRequest('db.query', {'sql': 'SELECT 1'}); + // Drain request line from fake stdin. + await stdin.stream.first; + + // Oversized reply line (no newline until after overflow). + stdout.add(List.filled(40, 0x61)); // 'a' * 40 + await expectLater(pending, throwsA(isA())); + + await client.close(); + await stdout.close(); + await stdin.close(); + }); + + test('decodes normal response', () async { + final stdout = StreamController>(); + final stdin = StreamController>(); + final client = JsonRpcStdioClient( + stdout: stdout.stream, + stdin: IOSink(stdin.sink), + requestTimeout: const Duration(seconds: 2), + ); + + final pending = client.sendRequest('ping'); + await stdin.stream.first; + stdout.add( + utf8.encode( + '${jsonEncode({ + 'jsonrpc': '2.0', + 'id': 1, + 'result': {'ok': true}, + })}\n', + ), + ); + final result = await pending; + expect(result, isA()); + expect((result as Map)['ok'], true); + + await client.close(); + await stdout.close(); + await stdin.close(); + }); + }); +}