diff --git a/agent-task/archive/2026/06/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/code_review_cloud_G08_0.log b/agent-task/archive/2026/06/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/code_review_cloud_G08_0.log new file mode 100644 index 0000000..b8264ed --- /dev/null +++ b/agent-task/archive/2026/06/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/code_review_cloud_G08_0.log @@ -0,0 +1,126 @@ + + +# Code Review Reference - REFACTOR + +> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.** +> Fill implementation-owned sections, then stop with active files in place and report ready for review. Do not create `complete.log` or archive files. + +## 개요 + +date=2026-06-06 +task=m-performance-hotspot-optimization/10_dart_isolate_gateway_contract, plan=0, tag=REFACTOR + +## Roadmap Targets + +- Milestone: `agent-roadmap/milestones/performance-hotspot-optimization.md` +- Task ids: + - `dart-isolate-backpressure`: bounded backpressure와 backlog 계측 + - `dart-isolate-error-ordering`: decode error ordered result와 disconnect semantics + - `dart-isolate-cleanup`: lifecycle cleanup hard gate +- Completion mode: check-on-pass + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 종결 절차는 코드리뷰 에이전트 전용이다. + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [REFACTOR-1] Async submit and bounded backlog | [x] | +| [REFACTOR-2] Ordered decode error result | [x] | +| [REFACTOR-3] Cleanup snapshot | [x] | + +## 구현 체크리스트 + +- [x] `InboundGateway.submit`을 bounded acceptance/backpressure가 가능한 async contract로 바꾸고 current call sites를 갱신한다. +- [x] gateway result model이 decoded frame과 ordered decode error를 같은 seq stream으로 표현하게 한다. +- [x] `FrameReorderBuffer`에 pending count/clear 또는 snapshot을 추가해 cleanup/backlog evidence를 만들 수 있게 한다. +- [x] IO/Web gateway가 close 후 pending submit/result stream/reorder state를 남기지 않게 한다. +- [x] backpressure, ordered error, repeated cleanup tests를 `dart/test/communicator_test.dart`에 추가한다. +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 코드리뷰 전용 체크리스트 + +> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. + +- [x] 판정을 append한다. +- [x] active 파일을 `.log`로 아카이브하고 PASS이면 `complete.log`를 작성한다. +- [x] PASS이면 런타임 완료 이벤트 메타데이터를 보고한다. + +## 계획 대비 변경 사항 + +- 계획 sketch는 `int get queuedCount; int get inFlightCount;`를 제안했고, 인터페이스에 두 getter를 모두 추가했다. 의미는 `inFlightCount`=worker에 보냈으나 결과가 오지 않은 frame 수(mailbox backlog), `queuedCount`=reorder buffer에 잔류 중인 frame 수(`FrameReorderBuffer.pendingCount`)로 나눠 정의했다. cleanup hard gate가 두 backlog를 따로 0으로 증명할 수 있게 하기 위함이다. +- ordered decode error는 별도 result type을 만들지 않고 기존 `DecodedFrame`에 `error` 필드와 `DecodedFrame.error()` 명명 생성자를 추가했다. 같은 `seq` stream/`FrameReorderBuffer`를 그대로 재사용해 reorder hang을 막기 위함이며, `results` 스트림 타입(`Stream`)을 바꾸지 않아 호출부 변경을 최소화했다. +- isolate worker reply를 5-element에서 6-element(`[seq, typeName, data, nonce, responseNonce, error]`)로 확장했다. error는 isolate 경계에서 항상 직렬화되도록 `Object`가 아니라 문자열(`'decode failed: ...'`)로 전달한다. +- web fallback `IsolateInboundGateway`에 무시되는 `capacity` 선택 인자를 추가해 conditional-export 생성자 시그니처를 IO 버전과 일치시켰다. in-process 동기 디코드에는 bound할 worker mailbox가 없어 동작에는 영향이 없다. +- `onReceivedFrame`은 `gateway.submit(...)`을 `await`하도록 바꿔 backpressure가 transport read loop로 전파되게 했다. 현재 production wiring(attach 호출부)은 없고 `attachInboundGateway`는 테스트 전용 경로뿐이므로 실제 transport 연결은 후속 `11+10`에서 처리한다. + +## 주요 설계 결정 + +- **submit readiness drop 제거(REFACTOR-1)**: `submit`은 `_ready` completer가 미완료면 await한 뒤 hand-off한다. isolate spawn 중 도착한 frame이 `_toWorker == null`로 조용히 drop되던 경로를 제거했다. +- **bounded backpressure(REFACTOR-1)**: `_inFlight >= capacity`이면 `_capacityWaiter`로 대기하고, worker 결과 수신 시 `_inFlight--` 후 slot을 release한다. `defaultCapacity=1024`로 무제한 `SendPort` mailbox를 bound한다. +- **ordered error → connection teardown(REFACTOR-2)**: malformed frame은 worker에서 catch되어 같은 seq의 error result로 돌아오고, `FrameReorderBuffer`를 통과해 후속 seq의 영구 잔류를 막는다. coordinator는 `onGatewayDecodeError`에서 `close()`를 호출해 connection close + pending 정리로 연결한다. `BaseClient.close()` override가 transport close까지 수행한다. +- **cleanup hard gate(REFACTOR-3)**: `close()`는 isolate kill/ReceivePort close에 더해 `_inFlight=0`, `_reorder.clear()`, 대기 중 completer 해제까지 수행한다. 반복 start/submit/close 20회 후 `inFlightCount`/`queuedCount`가 0으로 유지됨을 테스트로 증명한다. +- **isolate 생존(REFACTOR-2)**: worker `_gatewayEntry`의 디코드를 try/catch로 감싸 malformed frame이 isolate를 죽이지 않고 error result만 반환하게 했다. + +## 사용자 리뷰 요청 + +_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다._ + +- 상태: 없음 +- 사유 유형: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- submit readiness에서 frame drop이 없는지 확인한다. +- malformed frame이 seq gap/reorder hang을 남기지 않는지 확인한다. +- close/repeated cleanup test가 actual counts를 검증하는지 확인한다. + +## 검증 결과 + +### REFACTOR-1 중간 검증 +```text +$ cd dart && dart test test/communicator_test.dart -r expanded +00:00 +21: ... IsolateInboundGateway submit waits for ready and preserves all frames under capacity pressure +00:00 +22: ... SyncInboundGateway: malformed frame이 ordered decode error를 emit한다 +00:00 +23: ... IsolateInboundGateway: malformed frame이 isolate를 죽이지 않고 connection close + pending cleanup으로 이어진다 +00:00 +24: ... FrameReorderBuffer: pendingCount와 clear가 backlog를 노출하고 초기화한다 +00:00 +25: ... IsolateInboundGateway: 반복 start/submit/close가 residual backlog를 남기지 않는다 +00:00 +26: All tests passed! +``` + +### 최종 검증 +```text +$ cd dart && dart test test/communicator_test.dart -r expanded +00:00 +26: All tests passed! + +$ cd dart && dart test test/socket_test.dart -r expanded +00:16 +41: All tests passed! + +$ cd dart && dart compile js test/browser_ws_import_compile.dart -o /tmp/proto_socket_browser_ws_import_compile.js +Compiled 10,871,530 input bytes (5,872,758 characters source) to 11,692 characters JavaScript in 0.47 seconds +(/tmp/proto_socket_browser_ws_import_compile.js 산출물 생성됨) + +$ cd dart && dart test # dart-smoke 필수 전체 실행 +00:17 +67: All tests passed! +``` + +## 코드리뷰 결과 + +- 종합 판정: PASS +- 차원별 평가: + - correctness: Pass + - completeness: Pass + - test coverage: Pass + - API contract: Pass + - code quality: Pass + - plan deviation: Pass + - verification trust: Pass +- 발견된 문제: 없음 +- 다음 단계: PASS이므로 active plan/review를 `.log`로 아카이브하고 `complete.log` 작성 후 task 디렉터리를 archive로 이동한다. diff --git a/agent-task/archive/2026/06/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/complete.log b/agent-task/archive/2026/06/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/complete.log new file mode 100644 index 0000000..4ecec31 --- /dev/null +++ b/agent-task/archive/2026/06/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/complete.log @@ -0,0 +1,46 @@ +# Complete - m-performance-hotspot-optimization/10_dart_isolate_gateway_contract + +## 완료 일시 + +2026-06-06 + +## 요약 + +Dart isolate inbound gateway contract hardening 구현을 1회 리뷰했고 최종 판정은 PASS다. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `plan_cloud_G08_0.log` | `code_review_cloud_G08_0.log` | PASS | Async submit/backpressure, ordered decode error, cleanup snapshot 구현과 검증이 계획 범위를 충족했다. | + +## 구현/정리 내용 + +- `InboundGateway.submit`을 async backpressure contract로 바꾸고 `inFlightCount`/`queuedCount` 계측을 추가했다. +- `DecodedFrame.error`와 worker try/catch를 통해 malformed frame을 ordered error result로 전달하고 connection cleanup으로 연결했다. +- `FrameReorderBuffer.clear` 및 IO/Web gateway cleanup contract를 정리하고 관련 Dart tests를 추가했다. + +## 최종 검증 + +- `dart pub get` - PASS; dependencies resolved, newer incompatible package notices only. +- `dart test test/communicator_test.dart -r expanded` - PASS; 26 tests passed. +- `dart test` - PASS; 67 tests passed. +- `dart compile js test/browser_ws_import_compile.dart -o /tmp/proto_socket_browser_ws_import_compile.js` - PASS; JavaScript output generated. +- `dart analyze` - PASS for changed files; existing info remains at `lib/src/heartbeat_mixin.dart:13:16` for missing `@override`. + +## Roadmap Completion + +- Milestone: `agent-roadmap/milestones/performance-hotspot-optimization.md` +- Completed task ids: + - `dart-isolate-backpressure`: PASS; evidence=`agent-task/archive/2026/06/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/plan_cloud_G08_0.log`, `agent-task/archive/2026/06/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/code_review_cloud_G08_0.log`; verification=`dart test test/communicator_test.dart -r expanded`, `dart test`, `dart compile js test/browser_ws_import_compile.dart -o /tmp/proto_socket_browser_ws_import_compile.js` + - `dart-isolate-error-ordering`: PASS; evidence=`agent-task/archive/2026/06/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/plan_cloud_G08_0.log`, `agent-task/archive/2026/06/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/code_review_cloud_G08_0.log`; verification=`dart test test/communicator_test.dart -r expanded`, `dart test` + - `dart-isolate-cleanup`: PASS; evidence=`agent-task/archive/2026/06/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/plan_cloud_G08_0.log`, `agent-task/archive/2026/06/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/code_review_cloud_G08_0.log`; verification=`dart test test/communicator_test.dart -r expanded`, `dart test` +- Not completed task ids: 없음 + +## 잔여 Nit + +- 없음 + +## 후속 작업 + +- 없음 diff --git a/agent-task/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/PLAN-cloud-G08.md b/agent-task/archive/2026/06/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/plan_cloud_G08_0.log similarity index 100% rename from agent-task/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/PLAN-cloud-G08.md rename to agent-task/archive/2026/06/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/plan_cloud_G08_0.log diff --git a/agent-task/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/CODE_REVIEW-cloud-G08.md b/agent-task/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/CODE_REVIEW-cloud-G08.md deleted file mode 100644 index b8ceabd..0000000 --- a/agent-task/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/CODE_REVIEW-cloud-G08.md +++ /dev/null @@ -1,93 +0,0 @@ - - -# Code Review Reference - REFACTOR - -> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.** -> Fill implementation-owned sections, then stop with active files in place and report ready for review. Do not create `complete.log` or archive files. - -## 개요 - -date=2026-06-06 -task=m-performance-hotspot-optimization/10_dart_isolate_gateway_contract, plan=0, tag=REFACTOR - -## Roadmap Targets - -- Milestone: `agent-roadmap/milestones/performance-hotspot-optimization.md` -- Task ids: - - `dart-isolate-backpressure`: bounded backpressure와 backlog 계측 - - `dart-isolate-error-ordering`: decode error ordered result와 disconnect semantics - - `dart-isolate-cleanup`: lifecycle cleanup hard gate -- Completion mode: check-on-pass - -## 이 파일을 읽는 리뷰 에이전트에게 - -> **[REVIEW AGENT ONLY]** 종결 절차는 코드리뷰 에이전트 전용이다. - -## 구현 항목별 완료 여부 - -| 항목 | 완료 여부 | -|------|---------| -| [REFACTOR-1] Async submit and bounded backlog | [ ] | -| [REFACTOR-2] Ordered decode error result | [ ] | -| [REFACTOR-3] Cleanup snapshot | [ ] | - -## 구현 체크리스트 - -- [ ] `InboundGateway.submit`을 bounded acceptance/backpressure가 가능한 async contract로 바꾸고 current call sites를 갱신한다. -- [ ] gateway result model이 decoded frame과 ordered decode error를 같은 seq stream으로 표현하게 한다. -- [ ] `FrameReorderBuffer`에 pending count/clear 또는 snapshot을 추가해 cleanup/backlog evidence를 만들 수 있게 한다. -- [ ] IO/Web gateway가 close 후 pending submit/result stream/reorder state를 남기지 않게 한다. -- [ ] backpressure, ordered error, repeated cleanup tests를 `dart/test/communicator_test.dart`에 추가한다. -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. - -## 코드리뷰 전용 체크리스트 - -> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. - -- [ ] 판정을 append한다. -- [ ] active 파일을 `.log`로 아카이브하고 PASS이면 `complete.log`를 작성한다. -- [ ] PASS이면 런타임 완료 이벤트 메타데이터를 보고한다. - -## 계획 대비 변경 사항 - -_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._ - -## 주요 설계 결정 - -_구현 에이전트가 주요 설계 결정 사항을 기록한다._ - -## 사용자 리뷰 요청 - -_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다._ - -- 상태: 없음 -- 사유 유형: 없음 -- 결정 필요: 없음 -- 차단 근거: 없음 -- 실행한 검증/명령: 없음 -- 자동 후속 불가 이유: 없음 -- 재개 조건: 없음 - -## 리뷰어를 위한 체크포인트 - -- submit readiness에서 frame drop이 없는지 확인한다. -- malformed frame이 seq gap/reorder hang을 남기지 않는지 확인한다. -- close/repeated cleanup test가 actual counts를 검증하는지 확인한다. - -## 검증 결과 - -### REFACTOR-1 중간 검증 -```text -$ cd dart && dart test test/communicator_test.dart -r expanded -(output) -``` - -### 최종 검증 -```text -$ cd dart && dart test test/communicator_test.dart -r expanded -(output) -$ cd dart && dart test test/socket_test.dart -r expanded -(output) -$ cd dart && dart compile js test/browser_ws_import_compile.dart -o /tmp/proto_socket_browser_ws_import_compile.js -(output) -``` diff --git a/dart/lib/src/communicator.dart b/dart/lib/src/communicator.dart index 153e6e9..9d2cba3 100644 --- a/dart/lib/src/communicator.dart +++ b/dart/lib/src/communicator.dart @@ -240,6 +240,10 @@ abstract class Communicator { void attachInboundGateway(InboundGateway gateway) { _inboundGateway = gateway; _gatewaySubscription = gateway.results.listen((frame) { + if (frame.isError) { + onGatewayDecodeError(frame.seq, frame.error!); + return; + } onReceivedData( frame.typeName, frame.data, @@ -249,6 +253,17 @@ abstract class Communicator { }); } + /// Handles an ordered decode failure surfaced by the gateway. + /// + /// A malformed envelope means the inbound framing can no longer be trusted, so + /// the connection is torn down and pending requests are cleaned up. Overrides + /// (e.g. [BaseClient]) close the transport as part of [close]. + @protected + void onGatewayDecodeError(int seq, Object error) { + if (!isAlive || _isClosing) return; + unawaited(close()); + } + int _nextFrameSeq() { _frameSeq += 1; return _frameSeq; @@ -264,7 +279,8 @@ abstract class Communicator { if (!isAlive || _isClosing) return; final gateway = _inboundGateway; if (gateway != null) { - gateway.submit(InboundFrame(seq: _nextFrameSeq(), bytes: frame)); + // Await so gateway backpressure propagates to the transport read loop. + await gateway.submit(InboundFrame(seq: _nextFrameSeq(), bytes: frame)); return; } final common = PacketBase.fromBuffer(frame); diff --git a/dart/lib/src/inbound_gateway.dart b/dart/lib/src/inbound_gateway.dart index b4fa165..aaf0126 100644 --- a/dart/lib/src/inbound_gateway.dart +++ b/dart/lib/src/inbound_gateway.dart @@ -19,6 +19,11 @@ class InboundFrame { /// /// Carries the same envelope fields that [Communicator.onReceivedData] expects, /// plus the originating [seq] so results can be reordered before dispatch. +/// +/// A decode failure is also a result: [DecodedFrame.error] produces a frame +/// with the same [seq] but [error] set. Emitting it through the ordered +/// [InboundGateway.results] stream keeps the reorder buffer from stalling on a +/// missing seq and lets the coordinator tear the connection down deterministically. class DecodedFrame { final int seq; final String typeName; @@ -26,13 +31,29 @@ class DecodedFrame { final int incomingNonce; final int responseNonce; + /// Non-null when decoding this [seq] failed. Success frames leave it null. + final Object? error; + const DecodedFrame({ required this.seq, required this.typeName, required this.data, required this.incomingNonce, required this.responseNonce, - }); + }) : error = null; + + /// An ordered decode failure for [seq]. Occupies its seq slot so successors + /// still release, but carries no envelope payload. + const DecodedFrame.error({ + required this.seq, + required this.error, + }) : typeName = '', + data = const [], + incomingNonce = 0, + responseNonce = 0; + + /// Whether this frame represents a decode failure rather than a payload. + bool get isError => error != null; } /// Worker gateway contract. @@ -47,12 +68,26 @@ abstract interface class InboundGateway { /// Reordered decode results, emitted in input [InboundFrame.seq] order. Stream get results; + /// Frames submitted to the worker whose result has not yet come back. + /// + /// Bounded by the gateway's capacity; used as backlog/cleanup evidence. + int get inFlightCount; + + /// Frames already decoded but buffered in the reorder buffer waiting for an + /// earlier seq. Drains to zero once the contiguous run is released. + int get queuedCount; + /// Prepares the gateway (e.g. spawns a long-running isolate). Idempotent. Future start(); - /// Submits a raw frame for decoding. Fire-and-forget; the decoded result is - /// later emitted on [results]. - void submit(InboundFrame frame); + /// Submits a raw frame for decoding. + /// + /// Awaits worker readiness before handing off, so a frame is never silently + /// dropped while the isolate is still spawning, and applies backpressure when + /// [inFlightCount] reaches capacity: the future does not complete until an + /// earlier result releases a slot. The decoded result (or an ordered + /// [DecodedFrame.error]) is later emitted on [results]. + Future submit(InboundFrame frame); /// Releases gateway resources and closes [results]. Future close(); @@ -67,6 +102,9 @@ class FrameReorderBuffer { int _nextSeq = 1; final Map _pending = {}; + /// Frames buffered out of order, still waiting for an earlier seq. + int get pendingCount => _pending.length; + /// Buffers [frame] and returns the frames now releasable in seq order. List release(DecodedFrame frame) { if (frame.seq < _nextSeq) { @@ -81,6 +119,13 @@ class FrameReorderBuffer { } return ready; } + + /// Drops all buffered frames and resets the expected sequence to its initial + /// value, so a reused gateway leaves no residual backlog after close. + void clear() { + _pending.clear(); + _nextSeq = 1; + } } /// In-process gateway that decodes synchronously on the calling isolate. @@ -98,20 +143,34 @@ class SyncInboundGateway implements InboundGateway { @override Stream get results => _controller.stream; + // Synchronous decode never leaves work outstanding across an await. + @override + int get inFlightCount => 0; + + @override + int get queuedCount => _reorder.pendingCount; + @override Future start() async {} @override - void submit(InboundFrame frame) { + Future submit(InboundFrame frame) async { if (_closed) return; - final common = PacketBase.fromBuffer(frame.bytes); - final decoded = DecodedFrame( - seq: frame.seq, - typeName: common.typeName, - data: common.data, - incomingNonce: common.nonce, - responseNonce: common.responseNonce, - ); + DecodedFrame decoded; + try { + final common = PacketBase.fromBuffer(frame.bytes); + decoded = DecodedFrame( + seq: frame.seq, + typeName: common.typeName, + data: common.data, + incomingNonce: common.nonce, + responseNonce: common.responseNonce, + ); + } catch (error) { + // A malformed envelope still occupies its seq so successors release and + // the coordinator can tear the connection down in order. + decoded = DecodedFrame.error(seq: frame.seq, error: error); + } for (final ready in _reorder.release(decoded)) { if (!_controller.isClosed) _controller.add(ready); } @@ -121,6 +180,7 @@ class SyncInboundGateway implements InboundGateway { Future close() async { if (_closed) return; _closed = true; + _reorder.clear(); await _controller.close(); } } diff --git a/dart/lib/src/inbound_gateway_io.dart b/dart/lib/src/inbound_gateway_io.dart index 74db936..6fe600d 100644 --- a/dart/lib/src/inbound_gateway_io.dart +++ b/dart/lib/src/inbound_gateway_io.dart @@ -16,6 +16,12 @@ import 'packets/message_common.pb.dart'; /// handlers, the pending-request map, and the outbound write queue all remain /// on the main isolate's receive coordinator. class IsolateInboundGateway implements InboundGateway { + /// Max frames handed to the worker without a result back before [submit] + /// applies backpressure. Bounds the otherwise unbounded [SendPort] mailbox. + static const int defaultCapacity = 1024; + + final int capacity; + final StreamController _controller = StreamController.broadcast(); final FrameReorderBuffer _reorder = FrameReorderBuffer(); @@ -26,9 +32,21 @@ class IsolateInboundGateway implements InboundGateway { Completer? _ready; bool _closed = false; + int _inFlight = 0; + Completer? _capacityWaiter; + + IsolateInboundGateway({this.capacity = defaultCapacity}) + : assert(capacity > 0, 'capacity must be positive'); + @override Stream get results => _controller.stream; + @override + int get inFlightCount => _inFlight; + + @override + int get queuedCount => _reorder.pendingCount; + @override Future start() async { if (_isolate != null || _closed) return; @@ -44,18 +62,23 @@ class IsolateInboundGateway implements InboundGateway { void _onWorkerMessage(dynamic message) { if (message is SendPort) { _toWorker = message; - _ready?.complete(); + if (_ready != null && !_ready!.isCompleted) _ready!.complete(); _ready = null; return; } if (message is List) { - final decoded = DecodedFrame( - seq: message[0] as int, - typeName: message[1] as String, - data: (message[2] as List).cast(), - incomingNonce: message[3] as int, - responseNonce: message[4] as int, - ); + _inFlight--; + _releaseCapacity(); + final error = message[5]; + final decoded = error != null + ? DecodedFrame.error(seq: message[0] as int, error: error) + : DecodedFrame( + seq: message[0] as int, + typeName: message[1] as String, + data: (message[2] as List).cast(), + incomingNonce: message[3] as int, + responseNonce: message[4] as int, + ); for (final ready in _reorder.release(decoded)) { if (!_controller.isClosed) _controller.add(ready); } @@ -63,9 +86,29 @@ class IsolateInboundGateway implements InboundGateway { } @override - void submit(InboundFrame frame) { + Future submit(InboundFrame frame) async { if (_closed) return; - _toWorker?.send([frame.seq, frame.bytes]); + // Wait for the worker's SendPort so an early frame is never dropped. + final ready = _ready; + if (ready != null) await ready.future; + // Backpressure: hold until an in-flight result frees a slot. + while (!_closed && _inFlight >= capacity) { + final waiter = _capacityWaiter ??= Completer(); + await waiter.future; + } + if (_closed) return; + final port = _toWorker; + if (port == null) return; + _inFlight++; + port.send([frame.seq, frame.bytes]); + } + + void _releaseCapacity() { + final waiter = _capacityWaiter; + if (waiter != null && _inFlight < capacity) { + _capacityWaiter = null; + waiter.complete(); + } } @override @@ -77,6 +120,14 @@ class IsolateInboundGateway implements InboundGateway { _isolate = null; _fromWorker?.close(); _fromWorker = null; + _inFlight = 0; + _reorder.clear(); + // Unblock any submit awaiting readiness or capacity so it observes _closed. + if (_ready != null && !_ready!.isCompleted) _ready!.complete(); + _ready = null; + final waiter = _capacityWaiter; + _capacityWaiter = null; + waiter?.complete(); await _controller.close(); } } @@ -85,6 +136,11 @@ class IsolateInboundGateway implements InboundGateway { /// /// Runs only pure decode work. The first message back is the worker's /// [SendPort]; subsequent messages are `[seq, bytes]` decode requests. +/// +/// Each reply is a 6-element list `[seq, typeName, data, nonce, responseNonce, +/// error]`. A malformed frame is caught and reported as an ordered error result +/// (index 5 non-null) instead of crashing the isolate, so the reorder buffer +/// never stalls on the dropped seq. void _gatewayEntry(SendPort toMain) { final fromMain = ReceivePort(); toMain.send(fromMain.sendPort); @@ -96,13 +152,18 @@ void _gatewayEntry(SendPort toMain) { final request = message as List; final seq = request[0] as int; final bytes = (request[1] as List).cast(); - final common = PacketBase.fromBuffer(bytes); - toMain.send([ - seq, - common.typeName, - common.data, - common.nonce, - common.responseNonce, - ]); + try { + final common = PacketBase.fromBuffer(bytes); + toMain.send([ + seq, + common.typeName, + common.data, + common.nonce, + common.responseNonce, + null, + ]); + } catch (error) { + toMain.send([seq, '', const [], 0, 0, 'decode failed: $error']); + } }); } diff --git a/dart/lib/src/inbound_gateway_web.dart b/dart/lib/src/inbound_gateway_web.dart index e316881..e8e9efb 100644 --- a/dart/lib/src/inbound_gateway_web.dart +++ b/dart/lib/src/inbound_gateway_web.dart @@ -7,6 +7,10 @@ import 'inbound_gateway.dart'; /// preserving the same `seq`-ordered [InboundGateway] contract. This keeps the /// `dart.library.html` conditional export building without pulling `dart:io` /// or `dart:isolate` into the web bundle. +/// +/// Accepts (and ignores) [capacity] so the conditional-export constructor +/// signature matches the IO isolate gateway; in-process decode has no worker +/// mailbox to bound. class IsolateInboundGateway extends SyncInboundGateway { - IsolateInboundGateway(); + IsolateInboundGateway({int capacity = 0}); } diff --git a/dart/test/communicator_test.dart b/dart/test/communicator_test.dart index f72d89f..d2fd022 100644 --- a/dart/test/communicator_test.dart +++ b/dart/test/communicator_test.dart @@ -604,6 +604,140 @@ void main() { expect(received, [1, 2, 3, 4, 5]); await communicator.close(); }); + + test( + 'IsolateInboundGateway submit waits for ready and preserves all frames under capacity pressure', + () async { + final gateway = IsolateInboundGateway(capacity: 4); + final received = []; + final sub = gateway.results.listen((frame) => received.add(frame.seq)); + + // start()를 await하지 않고 곧바로 submit해 ready 대기와 drop 없음을 검증한다. + final startFuture = gateway.start(); + + const total = 50; + for (var seq = 1; seq <= total; seq++) { + await gateway.submit(InboundFrame(seq: seq, bytes: _frameBytes(seq))); + // capacity(4)를 초과하는 in-flight backlog가 쌓이지 않는다. + expect(gateway.inFlightCount, lessThanOrEqualTo(4)); + } + await startFuture; + + for (var i = 0; i < 400; i++) { + if (received.length == total) break; + await Future.delayed(const Duration(milliseconds: 5)); + } + + expect(received, List.generate(total, (i) => i + 1)); + await sub.cancel(); + await gateway.close(); + }); + + test('SyncInboundGateway: malformed frame이 ordered decode error를 emit한다', + () async { + final gateway = SyncInboundGateway(); + final errors = []; + final dispatched = []; + final sub = gateway.results.listen((frame) { + if (frame.isError) { + errors.add(frame.seq); + } else { + dispatched.add(frame.incomingNonce); + } + }); + + // seq=2가 malformed여도 seq=1은 정상 dispatch되고 seq=2는 error로 release되어 + // 이후 seq가 reorder buffer에 영구 잔류하지 않는다. + await gateway.submit(InboundFrame(seq: 1, bytes: _frameBytes(1, incomingNonce: 1))); + await gateway.submit(const InboundFrame(seq: 2, bytes: [0xFF, 0x01])); + await gateway.submit(InboundFrame(seq: 3, bytes: _frameBytes(3, incomingNonce: 3))); + + await Future.delayed(Duration.zero); + + expect(dispatched, [1, 3]); + expect(errors, [2]); + expect(gateway.queuedCount, 0); + await sub.cancel(); + await gateway.close(); + }); + + test( + 'IsolateInboundGateway: malformed frame이 isolate를 죽이지 않고 connection close + pending cleanup으로 이어진다', + () async { + final communicator = _FakeCommunicator(); + final gateway = IsolateInboundGateway(); + await gateway.start(); + communicator.attachGatewayForTest(gateway); + + final future = communicator.sendRequest( + TestData()..index = 1, + timeout: const Duration(seconds: 5), + ); + await Future.delayed(Duration.zero); + + // malformed PacketBase bytes: invalid wire type → worker decode 실패. + await communicator.receiveFrameForTest(const [0xFF, 0x01]); + + await expectLater( + future, + throwsA( + isA().having( + (error) => error.message, + 'message', + contains('connection closed'), + ), + ), + ); + expect(communicator.isAlive, isFalse); + }); + + test('FrameReorderBuffer: pendingCount와 clear가 backlog를 노출하고 초기화한다', () { + final buffer = FrameReorderBuffer(); + DecodedFrame frame(int seq) => DecodedFrame( + seq: seq, + typeName: 'T', + data: const [], + incomingNonce: seq, + responseNonce: 0, + ); + + // seq=2,3은 seq=1을 기다리며 buffer에 잔류한다. + buffer.release(frame(2)); + buffer.release(frame(3)); + expect(buffer.pendingCount, 2); + + // clear는 backlog를 비우고 nextSeq를 1로 초기화한다. + buffer.clear(); + expect(buffer.pendingCount, 0); + expect(buffer.release(frame(1)).map((f) => f.seq), [1]); + }); + + test( + 'IsolateInboundGateway: 반복 start/submit/close가 residual backlog를 남기지 않는다', + () async { + for (var round = 0; round < 20; round++) { + final gateway = IsolateInboundGateway(capacity: 8); + await gateway.start(); + final received = []; + final sub = gateway.results.listen((frame) => received.add(frame.seq)); + + for (var seq = 1; seq <= 10; seq++) { + await gateway.submit(InboundFrame(seq: seq, bytes: _frameBytes(seq))); + } + for (var i = 0; i < 200; i++) { + if (received.length == 10) break; + await Future.delayed(const Duration(milliseconds: 5)); + } + expect(received, List.generate(10, (i) => i + 1)); + + await sub.cancel(); + await gateway.close(); + + // close 후 in-flight/queued backlog가 모두 0이다. + expect(gateway.inFlightCount, 0); + expect(gateway.queuedCount, 0); + } + }); }); }