perf(extensions): bound JSON-RPC payload size (#419)

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` обязательны.

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

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<List<int>, String> boundedUtf8LineSplitter({
+  int maxLineBytes = kDefaultJsonRpcMaxLineBytes,
+}) {
+  return _BoundedUtf8LineSplitter(maxLineBytes: maxLineBytes);
+}
+
+class _BoundedUtf8LineSplitter
+    extends StreamTransformerBase<List<int>, String> {
+  _BoundedUtf8LineSplitter({required this.maxLineBytes});
+
+  final int maxLineBytes;
+
+  @override
+  Stream<String> bind(Stream<List<int>> stream) {
+    final controller = StreamController<String>(sync: true);
+    final pending = BytesBuilder(copy: false);
+    late final StreamSubscription<List<int>> 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;
+  }
+}

lib/core/extensions/rpc/json_rpc_stdio_client.dart

@@ -1,31 +1,49 @@
 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<List<int>> 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<String> _lines;
   final Duration requestTimeout;
+  final int maxLineBytes;
 
   final Map<int, Completer<Object?>> _pending = {};
   var _nextId = 1;
   var _closed = false;
   StreamSubscription<String>? _subscription;
   Object? _fatalError;
 
+  /// Serializes async line handling so large-line isolate decode stays ordered.
+  Future<void> _lineChain = Future<void>.value();
+
   /// Sends a JSON-RPC request and waits for the matching response.
   Future<Object?> sendRequest(
     String method, [
@@ -82,12 +100,21 @@ class JsonRpcStdioClient {
   }
 
   void _onLine(String line) {
+    _lineChain = _lineChain.then((_) => _handleLine(line));
+  }
+
+  Future<void> _handleLine(String line) async {
     if (line.trim().isEmpty) return;
     late final Map<String, dynamic> message;
     try {
-      final decoded = jsonDecode(line);
-      if (decoded is! Map<String, dynamic>) 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<String, dynamic>.from(decoded);
     } catch (_) {
       return;
     }

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

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<List<int>>.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<List<int>>();
+      final errors = <Object>[];
+      final sub = controller.stream
+          .transform(boundedUtf8LineSplitter(maxLineBytes: 8))
+          .listen((_) {}, onError: errors.add);
+
+      controller.add(utf8.encode('123456789')); // 9 bytes, no newline yet
+      await Future<void>.delayed(Duration.zero);
+      expect(errors, isNotEmpty);
+      expect(errors.first, isA<JsonRpcPayloadTooLargeException>());
+      await sub.cancel();
+      await controller.close();
+    });
+  });
+
+  group('JsonRpcStdioClient payload bounds', () {
+    test('completes pending request with payload-too-large error', () async {
+      final stdout = StreamController<List<int>>();
+      final stdin = StreamController<List<int>>();
+      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<int>.filled(40, 0x61)); // 'a' * 40
+      await expectLater(pending, throwsA(isA<JsonRpcPayloadTooLargeException>()));
+
+      await client.close();
+      await stdout.close();
+      await stdin.close();
+    });
+
+    test('decodes normal response', () async {
+      final stdout = StreamController<List<int>>();
+      final stdin = StreamController<List<int>>();
+      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<Map>());
+      expect((result as Map)['ok'], true);
+
+      await client.close();
+      await stdout.close();
+      await stdin.close();
+    });
+  });
+}