From 570cd8173cdadfb13e75942cc0865da1f0a9c30f Mon Sep 17 00:00:00 2001 From: toki Date: Tue, 2 Jun 2026 13:53:37 +0900 Subject: [PATCH] Update Kotlin files and archive G08 task artifacts --- .../code_review_cloud_G08_0.log | 113 ++++++++ .../code_review_cloud_G08_1.log | 139 ++++++++++ .../06_kotlin_gateway/complete.log | 42 +++ .../06_kotlin_gateway/plan_cloud_G08_0.log} | 0 .../06_kotlin_gateway/plan_cloud_G08_1.log | 110 ++++++++ .../CODE_REVIEW-cloud-G08.md | 71 ----- .../com/tokilabs/proto_socket/Communicator.kt | 256 ++++++++++++++++++ .../com/tokilabs/proto_socket/TcpClient.kt | 15 +- .../com/tokilabs/proto_socket/WsClient.kt | 17 +- .../tokilabs/proto_socket/CommunicatorTest.kt | 238 +++++++++++++++- 10 files changed, 912 insertions(+), 89 deletions(-) create mode 100644 agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/code_review_cloud_G08_0.log create mode 100644 agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/code_review_cloud_G08_1.log create mode 100644 agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/complete.log rename agent-task/{m-inbound-queue-ordering/06_kotlin_gateway/PLAN-cloud-G08.md => archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/plan_cloud_G08_0.log} (100%) create mode 100644 agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/plan_cloud_G08_1.log delete mode 100644 agent-task/m-inbound-queue-ordering/06_kotlin_gateway/CODE_REVIEW-cloud-G08.md diff --git a/agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/code_review_cloud_G08_0.log b/agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/code_review_cloud_G08_0.log new file mode 100644 index 0000000..529c8ae --- /dev/null +++ b/agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/code_review_cloud_G08_0.log @@ -0,0 +1,113 @@ + + +# Code Review Reference - API + +> **[IMPLEMENTING AGENT — READ FIRST]** 구현 후 이 파일을 채우고 active 파일을 그대로 둔 채 리뷰를 요청한다. 최종화는 code-review 전용이다. + +## 개요 + +date=2026-06-02 +task=m-inbound-queue-ordering/06_kotlin_gateway, plan=0, tag=API + +## Roadmap Targets + +- Milestone: `agent-roadmap/milestones/inbound-queue-ordering.md` +- Task ids: + - `kotlin-gateway`: Kotlin coroutine worker gateway 후보를 적용한다. +- Completion mode: check-on-pass + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [API-1] Kotlin coroutine gateway | [x] | + +## 구현 체크리스트 + +- [x] Kotlin gateway input/output에 내부 `seq`를 포함하고 coroutine worker pool 또는 제한 dispatcher를 추가한다. +- [x] gateway는 decode/전처리만 수행하고 pending/listener/request/write queue 상태를 소유하지 않는다. +- [x] 결과 collector가 `seq` 순서로 receive coordinator에 전달한다. +- [x] Kotlin tests에 out-of-order worker completion, 대량 packet 순서, close/cancel anchor를 추가한다. +- [x] 중간 검증으로 `cd kotlin && env JAVA_HOME=/config/opt/jdk/jdk-17.0.10+7 GRADLE_USER_HOME=/tmp/gradle ./gradlew test --tests com.tokilabs.proto_socket.CommunicatorTest --rerun-tasks`를 실행한다. +- [x] 최종 검증으로 `bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_matrix.sh --all`을 실행한다. +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 코드리뷰 전용 체크리스트 + +- [x] verdict append, log archive, FAIL follow-up PLAN/CODE_REVIEW 작성을 수행했다. + +## 계획 대비 변경 사항 + +- `Communicator.kt`에 coroutine worker-pool gateway를 추가했다(`InboundFrame`, `DecodedFrame`, `InboundGateway`, `CoroutineWorkerGateway`, 내부 `GatewayResult`/`FrameReorderBuffer`). Go의 `go/inbound_gateway.go` 설계를 coroutine 모델로 대응시켰다. 계획의 "수정 파일 요약"대로 신규 파일을 만들지 않고 기존 `Communicator.kt`에 추가했다. +- `CommunicatorTest.kt`에 gateway 테스트 5개를 추가했다(reorder buffer, out-of-order worker completion, decode error 전파, communicator 통합 대량 순서, close 멱등/전달 중단). +- **계획 범위 외 변경(사용자 승인)**: 계획은 "OkHttp WS staging 정책"을 범위 제외로 명시했다. staging 정책 코드(`WsClient.kt`)는 건드리지 않았으나, 기존 flaky 테스트 `testWsInboundStagingOverflowDisconnects`가 베이스라인에서도 5회 중 4회 실패하여 전체 검증 게이트를 막았다. 사용자 결정에 따라 해당 테스트의 대기 방식만 안정화했다(가상 시간 `delay(100)` → 실제 시간 폴링). 프로덕션 staging 정책은 변경하지 않았다. + +## 주요 설계 결정 + +- **gateway 책임 경계**: gateway는 순수 `PacketBase` decode/전처리만 수행한다. `pendingRequests`, listener, request handler, write queue 상태는 소유하지 않고 receive coordinator(`Communicator.enqueueInbound`)에 남긴다. sink를 호출하는 coroutine이 단일 collector뿐이므로 dispatch 호출은 직렬화되고 입력 순서가 보존된다. +- **seq 기반 reorder**: `seq`는 수신 측이 gateway 제출 전에 부여하는 내부 단조 증가 시퀀스로, wire format/공개 프로토콜 필드에는 노출하지 않는다. `FrameReorderBuffer`가 out-of-order 결과를 버퍼해 연속 prefix가 완성될 때만 순서대로 release한다. decode 실패 프레임도 seq를 진행시켜 reorder window를 막지 않으며, 오류는 sink가 아니라 `onError`로 seq 순서에 맞춰 보고한다(transport 종료 결정은 호출자 몫). +- **backpressure / 수명주기**: bounded jobs `Channel`이 backpressure를 제공한다(가득 차면 `submit` suspend). `close()`는 jobs 채널을 닫아 worker가 잔여 프레임을 drain한 뒤 종료하게 하며 멱등이다. 모든 worker가 종료하면 results 채널을 닫아 collector도 정상 종료시킨다. close 이후 `submit`은 drop한다. +- **테스트 결정성**: out-of-order 완료를 결정적으로 재현하기 위해 `decode`를 suspend로 두고 `runTest` 가상 시간 `delay`로 worker 완료 순서를 제어했다(Go의 sleep 기반 테스트에 대응). + +## 사용자 리뷰 요청 + +- 상태: 없음 +- 사유 유형: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- gateway가 actor/coordinator state ownership을 침범하지 않는지 확인한다. + +## 검증 결과 + +### API-1 중간 검증 +``` +$ cd kotlin && env JAVA_HOME=/config/opt/jdk/jdk-17.0.10+7 GRADLE_USER_HOME=/tmp/gradle ./gradlew test --tests com.tokilabs.proto_socket.CommunicatorTest --rerun-tasks +# 안정성 확인을 위해 5회 반복 실행 +=== run 1 === BUILD SUCCESSFUL in 7s +=== run 2 === BUILD SUCCESSFUL in 6s +=== run 3 === BUILD SUCCESSFUL in 6s +=== run 4 === BUILD SUCCESSFUL in 5s +=== run 5 === BUILD SUCCESSFUL in 5s +# 신규 gateway 테스트 5개 단독 실행도 3/3 안정 통과: +# testFrameReorderBufferReleasesInOrder +# testWorkerGatewayReordersOutOfOrderCompletion +# testWorkerGatewayReportsDecodeErrorAndAdvances +# testWorkerGatewayDispatchesThroughCommunicatorInOrder +# testWorkerGatewayCloseIsIdempotentAndStopsDelivery +``` + +flaky 안정화 근거: 동일 명령을 변경 전 베이스라인에서 5회 실행 시 `testWsInboundStagingOverflowDisconnects`가 4회 FAIL(가상 시간 vs Dispatchers.IO 실제 코루틴 레이스). 테스트 대기를 실제 시간 폴링으로 바꾼 뒤 5/5 PASS. + +### 최종 검증 +``` +$ bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_matrix.sh --all + +**Proto 동기화**: schema sync PASS +**동일언어**: Dart PASS / Go PASS / Kotlin PASS / Python PASS / TypeScript PASS +**언어 PASS 매트릭스**: 모든 서버×클라이언트 조합 PASS (Dart.io/Go/Kotlin/Python/TypeScript × Dart.io/Dart.web/Dart.web(WSS)/Go/Kotlin/Python/TypeScript) +**크로스테스트 상세**: 전 방향 PASS (언어간 16/16, Dart.web 방향 2/2), FAIL lines 0 + +결과 기록 파일: agent-test/runs/20260602-033833-proto-socket-full-matrix.md +``` + +## 코드리뷰 결과 + +- 종합 판정: FAIL +- 차원별 평가: + - correctness: Fail + - completeness: Fail + - test coverage: Fail + - API contract: Fail + - code quality: Pass + - plan deviation: Fail + - verification trust: Pass +- 발견된 문제: + - Required: [kotlin/src/main/kotlin/com/tokilabs/proto_socket/Communicator.kt:437](/config/workspace/proto-socket/kotlin/src/main/kotlin/com/tokilabs/proto_socket/Communicator.kt:437) 새 `CoroutineWorkerGateway`가 생성되었지만 `Communicator`에 gateway를 attach/enable하거나 raw frame을 gateway로 제출하는 API가 없습니다. 현재 실제 ingress는 [TcpClient.kt:71](/config/workspace/proto-socket/kotlin/src/main/kotlin/com/tokilabs/proto_socket/TcpClient.kt:71)와 [WsClient.kt:64](/config/workspace/proto-socket/kotlin/src/main/kotlin/com/tokilabs/proto_socket/WsClient.kt:64)에서 여전히 `PacketBase.parseFrom` 후 `enqueueInbound`를 직접 호출하므로 gateway worker pool이 production TCP/WS 수신 경로에 적용되지 않습니다. `Communicator`에 `enableInboundGateway`/`attachInboundGateway`/`onReceivedFrame`와 내부 frame seq, gateway decode error handler를 추가하고, TCP/WS read loop가 raw bytes를 `onReceivedFrame`으로 전달하게 수정하세요. + - Required: [kotlin/src/test/kotlin/com/tokilabs/proto_socket/CommunicatorTest.kt:533](/config/workspace/proto-socket/kotlin/src/test/kotlin/com/tokilabs/proto_socket/CommunicatorTest.kt:533) gateway 테스트가 gateway 인스턴스를 수동 생성한 뒤 sink에서 `communicator.enqueueInbound`를 직접 호출하는 단위 검증에 머물러 있어, 실제 read loop가 gateway를 우회해도 실패하지 않습니다. custom counting gateway를 communicator에 attach한 뒤 TCP/WS ingress가 raw frame을 gateway에 submit하는지 검증하고, gateway decode error가 transport disconnect/error path로 이어지는 테스트를 추가하세요. +- 다음 단계: FAIL이므로 user-review gate 없이 후속 `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.md`를 작성한다. diff --git a/agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/code_review_cloud_G08_1.log b/agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/code_review_cloud_G08_1.log new file mode 100644 index 0000000..f8b2636 --- /dev/null +++ b/agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/code_review_cloud_G08_1.log @@ -0,0 +1,139 @@ + + +# Code Review Reference - REVIEW_API + +> **[IMPLEMENTING AGENT — READ FIRST]** 구현 후 이 파일을 채우고 active 파일을 그대로 둔 채 리뷰를 요청한다. 최종화는 code-review 전용이다. + +## 개요 + +date=2026-06-02 +task=m-inbound-queue-ordering/06_kotlin_gateway, plan=1, tag=REVIEW_API + +## Roadmap Targets + +- Milestone: `agent-roadmap/milestones/inbound-queue-ordering.md` +- Task ids: + - `kotlin-gateway`: Kotlin coroutine worker gateway 후보를 적용한다. +- Completion mode: check-on-pass + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [REVIEW_API-1] Kotlin gateway wiring | [x] | + +## 구현 체크리스트 + +- [x] `Communicator`에 내부 frame seq, gateway attach/enable API, raw `onReceivedFrame` ingress, gateway decode error handler를 추가한다. +- [x] `Communicator.close()`/`shutdown()`에서 attached gateway lifecycle을 정리하고, gateway가 coordinator state를 소유하지 않게 유지한다. +- [x] `TcpClient` read loop가 raw packet bytes를 `Communicator.onReceivedFrame`으로 전달하고 inline/gateway parse error 모두 기존 disconnect semantics를 보존한다. +- [x] `WsClient.receiveBytes`가 raw websocket bytes를 `Communicator.onReceivedFrame`으로 전달하고 inline/gateway parse error 모두 기존 disconnect semantics를 보존한다. +- [x] Kotlin tests에 read loop가 attached gateway를 우회하지 않는 검증과 gateway parse-error path 검증을 추가한다. +- [x] 중간 검증으로 `cd kotlin && env JAVA_HOME=/config/opt/jdk/jdk-17.0.10+7 GRADLE_USER_HOME=/tmp/gradle ./gradlew test --tests com.tokilabs.proto_socket.CommunicatorTest --rerun-tasks`를 실행한다. +- [x] 최종 검증으로 `bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_matrix.sh --all`을 실행한다. +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 코드리뷰 전용 체크리스트 + +> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. +> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다. + +- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. +- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_{review_lane}_GNN_N.log`로 아카이브한다. +- [x] active `PLAN-*-G??.md`를 `plan_{build_lane}_GNN_M.log`로 아카이브한다. +- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하여 plan/review/archive 산출물이 추적 가능한지 확인한다. +- [x] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. +- [x] PASS이면 active task 디렉터리 `agent-task/{task_name}/`를 `agent-task/archive/YYYY/MM/{task_name}/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. +- [x] PASS이고 task group이 `m-`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. +- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/{task_group}/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. +- [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-{build_lane}-GNN.md`와 `CODE_REVIEW-{review_lane}-GNN.md`를 작성하고 `complete.log`를 작성하지 않는다. +- [ ] USER_REVIEW이면 `agent-ops/skills/common/code-review/templates/user-review-template.md` 기준으로 `USER_REVIEW.md`를 작성하고 active `PLAN-*.md`, `CODE_REVIEW-*.md`, `complete.log`를 남기지 않는다. +- [ ] USER_REVIEW가 사용자 결정으로 완료/PASS 해소되면 `USER_REVIEW.md`를 해소 상태로 갱신하고 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다. + +## 계획 대비 변경 사항 + +- 계획의 "수정 파일 요약" 4개 파일을 모두 수정했고 추가/대체 파일은 없다. 검증 명령도 계약 그대로 사용했다. +- `Communicator`에 `frameSeq`(AtomicLong), `gateway` 필드, `frameErrorHandler`, `setFrameErrorHandler`/`onFrameError`, `enableInboundGateway`/`attachInboundGateway`/`closeGateway`, suspend `onReceivedFrame(raw): Throwable?`를 추가했다. Go 선행 구현(`go/communicator.go`의 `OnReceivedFrame`/`EnableInboundGateway`/`SetFrameErrorHandler`)과 동일한 계약이다. +- `onReceivedFrame`은 Go가 `error`를 반환하는 것에 대응해 inline decode 오류를 `Throwable?`로 반환한다(null=정상). gateway path 오류는 `frameErrorHandler`로 비동기 보고한다. +- `TcpClient.readLoop`/`WsClient.receiveBytes`는 `PacketBase.parseFrom`+`enqueueInbound` 직접 호출을 `communicator.onReceivedFrame(bytes)` 호출로 교체하고, init에서 `setFrameErrorHandler { onDisconnected() }`를 등록했다. inline/gateway 두 경로 모두 기존 disconnect semantics를 보존한다. +- 직전 plan=0에서 manual `gateway.submit`+`sink→enqueueInbound`로 검증하던 통합 테스트를 production 경로(`attachInboundGateway`+`onReceivedFrame`)로 전환했다. + +## 주요 설계 결정 + +- **opt-in gateway, coordinator 소유권 유지**: gateway는 기본 미부착(inline decode)이며 `enableInboundGateway`/`attachInboundGateway`로만 활성화한다. gateway sink는 decoded frame을 `enqueueInbound`로만 전달하므로 `pendingRequests`/listener/request handler/write queue 상태는 receive coordinator가 단독 소유한다. Go 선행 구현과 동일한 경계다. +- **frame seq 부여 위치**: `onReceivedFrame`이 `frameSeq.incrementAndGet()`으로 seq를 부여한다. WS는 `receiveMutex` 안에서, TCP는 단일 read loop에서 호출하므로 seq 부여 순서가 수신 순서와 일치한다. seq는 내부 reorder 전용이며 wire format에 노출하지 않는다. +- **parse-error semantics 일원화**: gateway path decode 오류는 `onError → onFrameError → frameErrorHandler(=onDisconnected)`로, inline path 오류는 `onReceivedFrame` 반환값 → `onDisconnected`로 흐른다. 양쪽 모두 transport disconnect로 수렴하며 gateway 자체는 transport를 종료하지 않는다. +- **lifecycle**: `close()`/`shutdown()` 모두 `closeGateway()`로 attached gateway를 분리·close한다. worker/collector coroutine은 communicator scope에 속해 BaseClient의 `scope.cancel()` 시 정리된다. + +## 사용자 리뷰 요청 + +_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._ + +- 상태: 없음 +- 사유 유형: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- TCP/WS read loop가 raw frame을 gateway-aware communicator ingress로 넘기는지 확인한다. +- gateway가 pending/listener/request/write queue 상태를 직접 소유하지 않는지 확인한다. +- gateway parse error가 기존 transport disconnect/error semantics와 충돌하지 않는지 확인한다. + +## 검증 결과 + +_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._ + +필수 규칙: +- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다. +- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다. +- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다. +- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다. +- mobile/UI hang, timeout, 또는 2분 무진행은 blind retry를 중단하고 focused rerun 명령과 screenshot/window/UI-tree evidence path를 남기며, 불가능하면 정확한 사유를 남긴다. + +### REVIEW_API-1 중간 검증 +``` +$ cd kotlin && env JAVA_HOME=/config/opt/jdk/jdk-17.0.10+7 GRADLE_USER_HOME=/tmp/gradle ./gradlew test --tests com.tokilabs.proto_socket.CommunicatorTest --rerun-tasks +# 안정성 확인을 위해 3회 반복 +=== run 1 === BUILD SUCCESSFUL in 6s +=== run 2 === BUILD SUCCESSFUL in 6s +=== run 3 === BUILD SUCCESSFUL in 6s +# CommunicatorTest = 21 tests, 전체 통과 + +# read loop wiring 검증 신규/전환 테스트: +# testWsIngressSubmitsRawFramesToAttachedGateway - WsClient.receiveBytes가 attached gateway에 raw frame submit (inline 우회 시 submitted 비어 FAIL) +# testWsGatewayDecodeErrorTriggersDisconnect - gateway decode 오류가 frameErrorHandler→onDisconnected로 transport disconnect 유발 +# testAttachedGatewayDispatchesThroughOnReceivedFrameInOrder - attach+onReceivedFrame production 경로, out-of-order 완료 후에도 listener dispatch 순서=입력 순서(50개) + +# 전체 Kotlin 테스트(./gradlew test --rerun-tasks)도 BUILD SUCCESSFUL (TcpClient/WsClient 회귀 없음) +``` + +### 최종 검증 +``` +$ bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_matrix.sh --all + +**Proto 동기화**: schema sync PASS +**동일언어**: Dart PASS / Go PASS / Kotlin PASS / Python PASS / TypeScript PASS +**언어 PASS 매트릭스**: 모든 서버×클라이언트 조합 PASS (Dart.io/Go/Kotlin/Python/TypeScript × Dart.io/Dart.web/Dart.web(WSS)/Go/Kotlin/Python/TypeScript) +**크로스테스트 상세**: 전 방향 PASS (언어간 16/16, Dart.web 방향 2/2), FAIL lines 0 + +결과 기록 파일: agent-test/runs/20260602-041819-proto-socket-full-matrix.md +``` + +## 코드리뷰 결과 + +- 종합 판정: PASS +- 차원별 평가: + - correctness: Pass + - completeness: Pass + - test coverage: Pass + - API contract: Pass + - code quality: Pass + - plan deviation: Pass + - verification trust: Pass +- 발견된 문제: 없음 +- 다음 단계: PASS이므로 active plan/review를 로그로 아카이브하고 `complete.log`를 작성한 뒤 task 디렉터리를 archive로 이동한다. diff --git a/agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/complete.log b/agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/complete.log new file mode 100644 index 0000000..06fa6db --- /dev/null +++ b/agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/complete.log @@ -0,0 +1,42 @@ +# Complete - m-inbound-queue-ordering/06_kotlin_gateway + +## 완료 일시 + +2026-06-02 + +## 요약 + +Kotlin coroutine worker gateway wiring follow-up까지 2회 리뷰 루프로 완료했으며 최종 판정은 PASS. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `plan_cloud_G08_0.log` | `code_review_cloud_G08_0.log` | FAIL | gateway 타입과 단위 테스트는 있었지만 실제 TCP/WS ingress가 gateway를 우회해 follow-up 필요 | +| `plan_cloud_G08_1.log` | `code_review_cloud_G08_1.log` | PASS | communicator raw frame ingress, gateway attach/enable, TCP/WS wiring, parse-error path 테스트까지 충족 | + +## 구현/정리 내용 + +- Kotlin `Communicator`에 frame seq, gateway attach/enable, raw `onReceivedFrame`, frame error handler, lifecycle cleanup을 추가했다. +- `TcpClient`와 `WsClient`가 raw frame을 communicator ingress로 전달하도록 변경해 inline/gateway parse-error disconnect semantics를 보존했다. +- Kotlin tests에 gateway reorder, attached ingress, WS gateway submit, gateway decode error disconnect 검증을 추가했다. + +## 최종 검증 + +- `cd kotlin && env JAVA_HOME=/config/opt/jdk/jdk-17.0.10+7 GRADLE_USER_HOME=/tmp/gradle ./gradlew test --tests com.tokilabs.proto_socket.CommunicatorTest --rerun-tasks` - PASS; 리뷰 중 재실행 결과 `BUILD SUCCESSFUL in 6s`. +- `bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_matrix.sh --all` - PASS; 결과 기록 `agent-test/runs/20260602-041819-proto-socket-full-matrix.md`. + +## Roadmap Completion + +- Milestone: `agent-roadmap/milestones/inbound-queue-ordering.md` +- Completed task ids: + - `kotlin-gateway`: PASS; evidence=`plan_cloud_G08_1.log`, `code_review_cloud_G08_1.log`; verification=`agent-test/runs/20260602-041819-proto-socket-full-matrix.md` +- Not completed task ids: 없음 + +## 잔여 Nit + +- 없음 + +## 후속 작업 + +- 없음 diff --git a/agent-task/m-inbound-queue-ordering/06_kotlin_gateway/PLAN-cloud-G08.md b/agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/plan_cloud_G08_0.log similarity index 100% rename from agent-task/m-inbound-queue-ordering/06_kotlin_gateway/PLAN-cloud-G08.md rename to agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/plan_cloud_G08_0.log diff --git a/agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/plan_cloud_G08_1.log b/agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/plan_cloud_G08_1.log new file mode 100644 index 0000000..8961c43 --- /dev/null +++ b/agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/plan_cloud_G08_1.log @@ -0,0 +1,110 @@ + + +# Kotlin Gateway Wiring 후속 구현 계획 + +## 이 파일을 읽는 구현 에이전트에게 + +이 계획은 직전 리뷰 FAIL의 Required 이슈만 해결한다. 구현 후 `CODE_REVIEW-cloud-G08.md`의 구현 에이전트 소유 섹션에 실제 변경 내용과 검증 stdout/stderr를 채운 뒤 active 파일을 그대로 두고 리뷰를 요청한다. 최종 판정과 archive는 code-review 전용이다. + +## 배경 + +직전 구현은 `CoroutineWorkerGateway`와 reorder 테스트를 추가했지만, 실제 Kotlin TCP/WS read loop는 여전히 raw frame을 inline decode한 뒤 `Communicator.enqueueInbound`로 직접 넘긴다. 따라서 gateway worker pool이 production ingress path에 적용되지 않았다. + +## 사용자 리뷰 요청 흐름 + +구현 중 user-only blocker가 있으면 active review stub의 `사용자 리뷰 요청` 섹션을 채운다. + +## Roadmap Targets + +- Milestone: `agent-roadmap/milestones/inbound-queue-ordering.md` +- Task ids: + - `kotlin-gateway`: Kotlin coroutine worker gateway 후보를 적용한다. +- Completion mode: check-on-pass + +## 분석 결과 + +### 읽은 파일 + +- `agent-task/m-inbound-queue-ordering/06_kotlin_gateway/plan_cloud_G08_0.log` +- `agent-task/m-inbound-queue-ordering/06_kotlin_gateway/code_review_cloud_G08_0.log` +- `kotlin/src/main/kotlin/com/tokilabs/proto_socket/Communicator.kt` +- `kotlin/src/main/kotlin/com/tokilabs/proto_socket/TcpClient.kt` +- `kotlin/src/main/kotlin/com/tokilabs/proto_socket/WsClient.kt` +- `kotlin/src/test/kotlin/com/tokilabs/proto_socket/CommunicatorTest.kt` +- `go/communicator.go` +- `go/tcp_client.go` +- `go/ws_client.go` + +### 실패 원인 + +- Kotlin `Communicator`에 raw frame ingress API와 gateway attach/enable API가 없다. +- `TcpClient`와 `WsClient`가 gateway를 거칠 수 있는 `onReceivedFrame` equivalent 대신 `PacketBase.parseFrom`을 직접 호출한다. +- 테스트가 gateway를 수동 생성하고 sink에서 `enqueueInbound`를 직접 호출해 실제 read loop wiring gap을 잡지 못한다. + +### 범위 결정 근거 + +- 기존 worker gateway 구현을 유지하고, communicator/transport wiring과 그 wiring을 증명하는 테스트만 추가한다. +- proto schema, 다른 언어 구현, OkHttp staging 정책은 변경하지 않는다. + +### 빌드 등급 + +- build/review: `cloud-G08`. 이유: coroutine gateway lifecycle, transport parse-error semantics, TCP/WS ingress API contract를 함께 다룬다. + +## 구현 체크리스트 + +- [ ] `Communicator`에 내부 frame seq, gateway attach/enable API, raw `onReceivedFrame` ingress, gateway decode error handler를 추가한다. +- [ ] `Communicator.close()`/`shutdown()`에서 attached gateway lifecycle을 정리하고, gateway가 coordinator state를 소유하지 않게 유지한다. +- [ ] `TcpClient` read loop가 raw packet bytes를 `Communicator.onReceivedFrame`으로 전달하고 inline/gateway parse error 모두 기존 disconnect semantics를 보존한다. +- [ ] `WsClient.receiveBytes`가 raw websocket bytes를 `Communicator.onReceivedFrame`으로 전달하고 inline/gateway parse error 모두 기존 disconnect semantics를 보존한다. +- [ ] Kotlin tests에 read loop가 attached gateway를 우회하지 않는 검증과 gateway parse-error path 검증을 추가한다. +- [ ] 중간 검증으로 `cd kotlin && env JAVA_HOME=/config/opt/jdk/jdk-17.0.10+7 GRADLE_USER_HOME=/tmp/gradle ./gradlew test --tests com.tokilabs.proto_socket.CommunicatorTest --rerun-tasks`를 실행한다. +- [ ] 최종 검증으로 `bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_matrix.sh --all`을 실행한다. +- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +### [REVIEW_API-1] Kotlin gateway wiring + +#### 문제 + +- `CoroutineWorkerGateway`가 실제 ingress path에서 사용되지 않아 `kotlin-gateway` Task 완료 조건을 충족하지 못한다. + +#### 해결 방법 + +- Go 선행 구현과 같은 구조로 `Communicator`에 gateway attach/enable과 raw frame ingress를 추가한다. +- `TcpClient`/`WsClient`는 raw frame을 직접 decode하지 말고 communicator raw ingress에 넘긴다. +- gateway sink는 decoded frame을 `enqueueInbound` 또는 기존 coordinator path로만 전달한다. + +#### 수정 파일 및 체크리스트 + +- [ ] `kotlin/src/main/kotlin/com/tokilabs/proto_socket/Communicator.kt` +- [ ] `kotlin/src/main/kotlin/com/tokilabs/proto_socket/TcpClient.kt` +- [ ] `kotlin/src/main/kotlin/com/tokilabs/proto_socket/WsClient.kt` +- [ ] `kotlin/src/test/kotlin/com/tokilabs/proto_socket/CommunicatorTest.kt` + +#### 테스트 작성 + +- attached counting gateway가 raw frame submit을 기록하고, read loop/receiveBytes가 inline decode로 우회하면 실패하게 만든다. +- gateway decode error가 caller-owned error/disconnect handler로 전달되는지 확인한다. +- production `CoroutineWorkerGateway` attached path에서도 out-of-order completion 후 communicator listener dispatch 순서가 입력 순서인지 확인한다. + +#### 중간 검증 + +```bash +cd kotlin && env JAVA_HOME=/config/opt/jdk/jdk-17.0.10+7 GRADLE_USER_HOME=/tmp/gradle ./gradlew test --tests com.tokilabs.proto_socket.CommunicatorTest --rerun-tasks +``` + +## 수정 파일 요약 + +| 파일 | 항목 | +|------|------| +| `kotlin/src/main/kotlin/com/tokilabs/proto_socket/Communicator.kt` | REVIEW_API-1 | +| `kotlin/src/main/kotlin/com/tokilabs/proto_socket/TcpClient.kt` | REVIEW_API-1 | +| `kotlin/src/main/kotlin/com/tokilabs/proto_socket/WsClient.kt` | REVIEW_API-1 | +| `kotlin/src/test/kotlin/com/tokilabs/proto_socket/CommunicatorTest.kt` | REVIEW_API-1 | + +## 최종 검증 + +```bash +bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_matrix.sh --all +``` + +모든 코드 변경 완료 후 반드시 `CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다. diff --git a/agent-task/m-inbound-queue-ordering/06_kotlin_gateway/CODE_REVIEW-cloud-G08.md b/agent-task/m-inbound-queue-ordering/06_kotlin_gateway/CODE_REVIEW-cloud-G08.md deleted file mode 100644 index 15f54c7..0000000 --- a/agent-task/m-inbound-queue-ordering/06_kotlin_gateway/CODE_REVIEW-cloud-G08.md +++ /dev/null @@ -1,71 +0,0 @@ - - -# Code Review Reference - API - -> **[IMPLEMENTING AGENT — READ FIRST]** 구현 후 이 파일을 채우고 active 파일을 그대로 둔 채 리뷰를 요청한다. 최종화는 code-review 전용이다. - -## 개요 - -date=2026-06-02 -task=m-inbound-queue-ordering/06_kotlin_gateway, plan=0, tag=API - -## Roadmap Targets - -- Milestone: `agent-roadmap/milestones/inbound-queue-ordering.md` -- Task ids: - - `kotlin-gateway`: Kotlin coroutine worker gateway 후보를 적용한다. -- Completion mode: check-on-pass - -## 구현 항목별 완료 여부 - -| 항목 | 완료 여부 | -|------|---------| -| [API-1] Kotlin coroutine gateway | [ ] | - -## 구현 체크리스트 - -- [ ] Kotlin gateway input/output에 내부 `seq`를 포함하고 coroutine worker pool 또는 제한 dispatcher를 추가한다. -- [ ] gateway는 decode/전처리만 수행하고 pending/listener/request/write queue 상태를 소유하지 않는다. -- [ ] 결과 collector가 `seq` 순서로 receive coordinator에 전달한다. -- [ ] Kotlin tests에 out-of-order worker completion, 대량 packet 순서, close/cancel anchor를 추가한다. -- [ ] 중간 검증으로 `cd kotlin && env JAVA_HOME=/config/opt/jdk/jdk-17.0.10+7 GRADLE_USER_HOME=/tmp/gradle ./gradlew test --tests com.tokilabs.proto_socket.CommunicatorTest --rerun-tasks`를 실행한다. -- [ ] 최종 검증으로 `bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_matrix.sh --all`을 실행한다. -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. - -## 코드리뷰 전용 체크리스트 - -- [ ] verdict append, log archive, PASS complete.log/archive move를 수행한다. - -## 계획 대비 변경 사항 - -_구현 에이전트가 기록한다._ - -## 주요 설계 결정 - -_구현 에이전트가 기록한다._ - -## 사용자 리뷰 요청 - -- 상태: 없음 -- 사유 유형: 없음 -- 결정 필요: 없음 -- 차단 근거: 없음 -- 실행한 검증/명령: 없음 -- 자동 후속 불가 이유: 없음 -- 재개 조건: 없음 - -## 리뷰어를 위한 체크포인트 - -- gateway가 actor/coordinator state ownership을 침범하지 않는지 확인한다. - -## 검증 결과 - -### API-1 중간 검증 -``` -$ cd kotlin && env JAVA_HOME=/config/opt/jdk/jdk-17.0.10+7 GRADLE_USER_HOME=/tmp/gradle ./gradlew test --tests com.tokilabs.proto_socket.CommunicatorTest --rerun-tasks -``` - -### 최종 검증 -``` -$ bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_matrix.sh --all -``` diff --git a/kotlin/src/main/kotlin/com/tokilabs/proto_socket/Communicator.kt b/kotlin/src/main/kotlin/com/tokilabs/proto_socket/Communicator.kt index de96336..5e663ca 100644 --- a/kotlin/src/main/kotlin/com/tokilabs/proto_socket/Communicator.kt +++ b/kotlin/src/main/kotlin/com/tokilabs/proto_socket/Communicator.kt @@ -8,12 +8,14 @@ import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.ClosedSendChannelException +import kotlinx.coroutines.joinAll import kotlinx.coroutines.launch import kotlinx.coroutines.selects.select import kotlinx.coroutines.withTimeout import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicLong import java.util.concurrent.locks.ReentrantReadWriteLock import kotlin.concurrent.read import kotlin.concurrent.write @@ -71,6 +73,7 @@ class Communicator( ) { private val rwLock = ReentrantReadWriteLock() private val nonce = AtomicInteger(0) + private val frameSeq = AtomicLong(0) private val isAlive = AtomicBoolean(false) private val parserMap: MutableMap MessageLite> = ConcurrentHashMap() private val handlers = mutableMapOf Unit>>() @@ -84,6 +87,8 @@ class Communicator( @Volatile private var receiveLoopJob: kotlinx.coroutines.Job? = null private var writeErrorHandler: ((Throwable) -> Unit)? = null + private var frameErrorHandler: ((Throwable) -> Unit)? = null + private var gateway: InboundGateway? = null init { initialize(parserMap) @@ -104,6 +109,89 @@ class Communicator( rwLock.write { writeErrorHandler = fn } } + /** + * inbound gateway가 raw frame decode에 실패했을 때 호출되는 handler를 등록한다. transport는 이를 통해 + * gateway path에서도 inline path와 동일한 parse-error disconnect semantics를 적용한다(write path의 + * [setWriteErrorHandler]와 대응). gateway가 없으면 [onReceivedFrame]이 decode 오류를 반환값으로 + * 노출하므로 이 handler는 쓰이지 않는다. + */ + fun setFrameErrorHandler(fn: (Throwable) -> Unit) { + rwLock.write { frameErrorHandler = fn } + } + + private fun onFrameError(err: Throwable) { + val handler = rwLock.read { frameErrorHandler } + handler?.invoke(err) + } + + /** + * receive coordinator 앞단에 default coroutine worker-pool gateway를 attach한다. [onReceivedFrame]으로 + * 제출된 raw frame은 pool이 decode하고 내부 `seq`로 reorder해 coordinator에 전달하므로 FIFO dispatch와 + * stateful 처리 소유권은 coordinator에 유지된다. + * + * gateway는 순수 `PacketBase` envelope decode만 수행하고 pending-request map, listener, write queue를 + * 건드리지 않는다. opt-in이며, attach하지 않으면 [onReceivedFrame]은 호출 coroutine에서 inline decode한다. + */ + fun enableInboundGateway(workers: Int = 0, queueSize: Int = 64) { + val gw = CoroutineWorkerGateway( + scope = scope, + workers = workers, + queueSize = queueSize, + sink = { f -> enqueueInbound(f.typeName, f.data, f.incomingNonce, f.responseNonce) }, + onError = { err -> + // gateway decode 오류를 transport의 frame error handler로 노출해 inline path와 동일한 + // parse-error disconnect semantics를 적용한다. gateway 자체는 transport를 종료하지 않는다. + onFrameError(err) + }, + ) + attachInboundGateway(gw) + } + + /** + * custom [InboundGateway]를 receive coordinator 앞단에 설치한다. gateway sink는 coordinator 소유권을 + * 보존하기 위해 decoded frame을 [enqueueInbound]로 전달해야 한다. default worker-pool gateway는 + * [enableInboundGateway]를 사용한다. + */ + fun attachInboundGateway(gateway: InboundGateway) { + rwLock.write { this.gateway = gateway } + } + + private fun closeGateway() { + val gw = rwLock.write { + val current = gateway + gateway = null + current + } + gw?.close() + } + + /** + * transport read loop가 raw `PacketBase` frame을 ingest한다. + * + * gateway가 attach되어 있으면 frame에 내부 `seq`를 부여해 off-coordinator decode + reorder로 제출하고 + * null을 반환한다. decode에 실패한 frame은 `seq` 순서에 맞춰 비동기로 [setFrameErrorHandler]에 등록된 + * frame error handler에 보고되어 transport가 inline path와 동일한 parse-error disconnect semantics를 + * 적용할 수 있게 한다. gateway 자체는 transport를 종료하지 않는다. + * + * gateway가 없으면 호출 coroutine에서 inline decode 후 [onReceivedData]로 전달하고, decode 오류를 + * 반환해 transport가 기존 parse-error disconnect semantics를 적용하게 한다. + */ + suspend fun onReceivedFrame(raw: ByteArray): Throwable? { + if (!isAlive() || isClosing.get()) return null + val gw = rwLock.read { gateway } + if (gw != null) { + gw.submit(InboundFrame(seq = frameSeq.incrementAndGet(), bytes = raw)) + return null + } + return try { + val base = PacketBase.parseFrom(raw) + onReceivedData(base.typeName, base.data.toByteArray(), base.nonce, base.responseNonce) + null + } catch (e: Throwable) { + e + } + } + @PublishedApi internal fun nextNonce(): Int { while (true) { @@ -117,6 +205,7 @@ class Communicator( if (!shutdownStarted.compareAndSet(false, true)) return isAlive.set(false) isClosing.set(false) + closeGateway() writeQueue.close() inboundQueue.cancel() closed.complete(Unit) @@ -132,6 +221,7 @@ class Communicator( suspend fun close() { if (!shutdownStarted.compareAndSet(false, true)) return isClosing.set(true) + closeGateway() inboundQueue.close() receiveLoopJob?.join() @@ -353,6 +443,172 @@ class Communicator( rwLock.write { pendingRequests.remove(nonce) } } +/** + * 수신 측이 gateway 제출 전에 부여하는 내부 raw 프레임이다. + * + * `seq`는 단조 증가하는 내부 시퀀스로, 병렬 decode 후 입력 순서를 복원하는 용도로만 쓰인다. + * wire format이나 공개 프로토콜 필드에는 절대 노출하지 않는다. + */ +data class InboundFrame(val seq: Long, val bytes: ByteArray) + +/** + * gateway가 PacketBase envelope을 decode한 결과다. `Communicator`의 inbound dispatch가 + * 기대하는 envelope 필드와, 입력 순서 복원을 위한 원본 `seq`를 함께 싣는다. + */ +data class DecodedFrame( + val seq: Long, + val typeName: String, + val data: ByteArray, + val incomingNonce: Int, + val responseNonce: Int, +) + +/** + * 수신 coordinator 앞단에서 순수 raw-bytes 작업(PacketBase envelope decode/경량 전처리)만 + * 병렬 처리하는 worker gateway 계약이다. + * + * gateway는 stateful dispatch, `pendingRequests`, listener, write queue를 소유하지 않는다. + * decode 결과는 입력 `seq` 순서로 sink에 전달되어 coordinator가 결정적으로 dispatch할 수 있게 한다. + */ +interface InboundGateway { + /** raw 프레임을 gateway에 제출한다. fire-and-forget이며 decode 결과는 이후 sink에 `seq` 순서로 전달된다. */ + suspend fun submit(frame: InboundFrame) + + /** gateway 자원을 해제하고 추가 전달을 멈춘다. */ + fun close() +} + +/** reorder 버퍼를 통과하는 per-frame decode 결과다. 실패 프레임도 seq를 진행시켜 reorder window를 막지 않는다. */ +internal data class GatewayResult( + val seq: Long, + val frame: DecodedFrame?, + val error: Throwable?, +) + +/** + * out-of-order로 도착할 수 있는 결과에서 입력 순서를 복원한다. seq는 1부터 연속이라고 가정한다. + * 결과는 앞선 모든 seq가 release될 때까지 버퍼되며, 이후 자신과 연속된 후속 결과까지 순서대로 반환한다. + * concurrent 사용에 안전하지 않으며 collector coroutine 단독에서만 호출한다. + */ +internal class FrameReorderBuffer { + private var nextSeq: Long = 1 + private val pending = HashMap() + + /** res를 버퍼하고 지금 순서대로 release 가능한 결과들을 seq 순으로 반환한다. */ + fun release(res: GatewayResult): List { + if (res.seq < nextSeq) { + // 이미 release되었거나 중복/지연 도착이므로 두 번 방출하지 않도록 무시한다. + return emptyList() + } + pending[res.seq] = res + val ready = mutableListOf() + while (true) { + val r = pending.remove(nextSeq) ?: break + ready.add(r) + nextSeq++ + } + return ready + } +} + +/** + * coroutine worker-pool gateway다. bounded jobs `Channel`이 N개 worker에 프레임을 공급해 PacketBase + * envelope을 병렬 decode하고, 단일 collector coroutine이 결과를 `seq`로 reorder해 입력 순서로 sink에 전달한다. + * sink를 호출하는 coroutine이 collector 하나뿐이므로 sink 호출은 직렬화되고 순서가 보존되어, stateful dispatch는 + * 수신 coordinator에 남는다. + * + * decode에 실패한 프레임은 sink가 아니라 `onError`로 `seq` 순서에 맞춰 보고되어 transport가 기존 parse-error + * 정책을 적용할 수 있게 한다. gateway 자체는 transport를 종료하지 않는다. `onError`는 null일 수 있다. + * + * `workers <= 0`은 사용 가능한 프로세서 수로, `queueSize <= 0`은 64로 기본값을 적용한다. bounded jobs + * 채널이 backpressure를 제공한다. 채널이 가득 차면 worker가 슬롯을 비울 때까지 `submit`이 suspend한다. + */ +class CoroutineWorkerGateway( + scope: CoroutineScope, + workers: Int = 0, + queueSize: Int = 64, + private val decode: suspend (ByteArray) -> PacketBase = { PacketBase.parseFrom(it) }, + private val sink: suspend (DecodedFrame) -> Unit, + private val onError: ((Throwable) -> Unit)? = null, +) : InboundGateway { + private val jobs: Channel + private val results: Channel + private val closed = AtomicBoolean(false) + + init { + val workerCount = if (workers <= 0) Runtime.getRuntime().availableProcessors() else workers + val capacity = if (queueSize <= 0) 64 else queueSize + jobs = Channel(capacity = capacity) + results = Channel(capacity = capacity) + + val workerJobs = (1..workerCount).map { scope.launch { workerLoop() } } + scope.launch { collectorLoop() } + // 모든 worker가 jobs를 drain하고 종료하면 results를 닫아 collector도 종료시킨다. + scope.launch { + workerJobs.joinAll() + results.close() + } + } + + private suspend fun workerLoop() { + for (frame in jobs) { + val result = try { + val base = decode(frame.bytes) + GatewayResult( + seq = frame.seq, + frame = DecodedFrame( + seq = frame.seq, + typeName = base.typeName, + data = base.data.toByteArray(), + incomingNonce = base.nonce, + responseNonce = base.responseNonce, + ), + error = null, + ) + } catch (e: Throwable) { + GatewayResult(seq = frame.seq, frame = null, error = e) + } + try { + results.send(result) + } catch (e: ClosedSendChannelException) { + return + } + } + } + + private suspend fun collectorLoop() { + val reorder = FrameReorderBuffer() + for (result in results) { + for (ready in reorder.release(result)) { + val error = ready.error + if (error != null) { + // decode 오류를 seq 순서로 보고하고 계속 진행해 한 프레임의 실패가 이후 프레임을 막지 않게 한다. + // transport 종료 여부는 호출자의 onError가 결정한다. + onError?.invoke(error) + continue + } + ready.frame?.let { sink(it) } + } + } + } + + /** raw 프레임을 worker pool에 넘긴다. bounded jobs 채널이 가득 차면 suspend(backpressure)하고, 닫혀 있으면 즉시 drop한다. */ + override suspend fun submit(frame: InboundFrame) { + if (closed.get()) return + try { + jobs.send(frame) + } catch (e: ClosedSendChannelException) { + // gateway가 닫혔으므로 drop한다. + } + } + + /** jobs 채널을 닫아 worker가 남은 프레임을 drain한 뒤 종료하게 한다. 멱등하며 in-flight sink 호출을 기다리지 않는다. */ + override fun close() { + if (!closed.compareAndSet(false, true)) return + jobs.close() + } +} + inline fun addListenerTyped( communicator: Communicator, noinline fn: (T) -> Unit, diff --git a/kotlin/src/main/kotlin/com/tokilabs/proto_socket/TcpClient.kt b/kotlin/src/main/kotlin/com/tokilabs/proto_socket/TcpClient.kt index 0a00910..165cba5 100644 --- a/kotlin/src/main/kotlin/com/tokilabs/proto_socket/TcpClient.kt +++ b/kotlin/src/main/kotlin/com/tokilabs/proto_socket/TcpClient.kt @@ -34,6 +34,7 @@ class TcpClient private constructor( init { communicator.setWriteErrorHandler { onDisconnected() } + communicator.setFrameErrorHandler { onDisconnected() } communicator.addListener(typeNameOf()) { onHeartBeat() } readLoop() sendHeartBeat() @@ -68,13 +69,13 @@ class TcpClient private constructor( } val bytes = ByteArray(length) input.readFully(bytes) - val base = PacketBase.parseFrom(bytes) - communicator.enqueueInbound( - base.typeName, - base.data.toByteArray(), - base.nonce, - base.responseNonce, - ) + // raw frame을 communicator ingress로 넘긴다. gateway가 attach되어 있으면 worker pool이 + // decode/reorder하고, 없으면 inline decode한다. inline decode 오류는 반환값으로, + // gateway decode 오류는 frame error handler로 동일한 disconnect semantics를 적용한다. + if (communicator.onReceivedFrame(bytes) != null) { + onDisconnected() + return@launch + } sendHeartBeat() } catch (_: Exception) { onDisconnected() diff --git a/kotlin/src/main/kotlin/com/tokilabs/proto_socket/WsClient.kt b/kotlin/src/main/kotlin/com/tokilabs/proto_socket/WsClient.kt index aa9e457..2b1c2fe 100644 --- a/kotlin/src/main/kotlin/com/tokilabs/proto_socket/WsClient.kt +++ b/kotlin/src/main/kotlin/com/tokilabs/proto_socket/WsClient.kt @@ -40,6 +40,7 @@ class WsClient internal constructor( init { communicator.setWriteErrorHandler { onDisconnected() } + communicator.setFrameErrorHandler { onDisconnected() } communicator.addListener(typeNameOf()) { onHeartBeat() } sendHeartBeat() } @@ -60,17 +61,13 @@ class WsClient internal constructor( scope.launch { try { receiveMutex.withLock { - runCatching { - val base = PacketBase.parseFrom(bytes) - communicator.enqueueInbound( - base.typeName, - base.data.toByteArray(), - base.nonce, - base.responseNonce, - ) - sendHeartBeat() - }.onFailure { + // raw frame을 communicator ingress로 넘긴다. gateway가 attach되어 있으면 worker pool이 + // decode/reorder하고, 없으면 inline decode한다. inline decode 오류는 반환값으로, + // gateway decode 오류는 frame error handler로 동일한 disconnect semantics를 적용한다. + if (communicator.onReceivedFrame(bytes) != null) { onDisconnected() + } else { + sendHeartBeat() } } } finally { diff --git a/kotlin/src/test/kotlin/com/tokilabs/proto_socket/CommunicatorTest.kt b/kotlin/src/test/kotlin/com/tokilabs/proto_socket/CommunicatorTest.kt index c24e573..92afa7a 100644 --- a/kotlin/src/test/kotlin/com/tokilabs/proto_socket/CommunicatorTest.kt +++ b/kotlin/src/test/kotlin/com/tokilabs/proto_socket/CommunicatorTest.kt @@ -11,6 +11,7 @@ import kotlinx.coroutines.delay import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking import kotlinx.coroutines.test.runTest +import kotlinx.coroutines.withContext import java.util.Collections import java.util.concurrent.atomic.AtomicInteger import kotlin.test.Test @@ -332,7 +333,14 @@ class CommunicatorTest { client.receiveBytes(f) } - delay(100) + // staging 코루틴은 Dispatchers.IO(실제 스레드)에서 동작하므로 runTest 가상 시간 delay로는 + // onDisconnected의 실제 close 완료를 기다릴 수 없다. 실제 시간 기준으로 connection.closed를 폴링한다. + withContext(Dispatchers.IO) { + val deadline = System.nanoTime() + 2_000_000_000L + while (System.nanoTime() < deadline && !connection.closed) { + Thread.sleep(5) + } + } assertTrue(connection.closed, "connection should be closed forcefully due to staging overflow") assertTrue(!client.communicator.isAlive(), "communicator should be inactive") @@ -434,5 +442,233 @@ class CommunicatorTest { assertEquals(listOf(1, 2), received, "seq ordering was not preserved") communicator.close() } + + @Test + fun testFrameReorderBufferReleasesInOrder() { + val buffer = FrameReorderBuffer() + + assertTrue( + buffer.release(GatewayResult(seq = 3, frame = null, error = null)).isEmpty(), + "seq 3 should buffer until 1,2 arrive", + ) + assertTrue( + buffer.release(GatewayResult(seq = 2, frame = null, error = null)).isEmpty(), + "seq 2 should still buffer until 1 arrives", + ) + val flushed = buffer.release(GatewayResult(seq = 1, frame = null, error = null)) + assertEquals(listOf(1L, 2L, 3L), flushed.map { it.seq }, "1,2,3 should flush together in order") + + // nextSeq 아래 seq는 중복/지연 도착이므로 무시되어야 한다. + assertTrue( + buffer.release(GatewayResult(seq = 1, frame = null, error = null)).isEmpty(), + "late duplicate seq should be ignored", + ) + } + + @Test + fun testWorkerGatewayReordersOutOfOrderCompletion() = runTest { + val total = 8 + val dispatched = mutableListOf() + + // 낮은 nonce(앞선 seq)일수록 더 오래 delay시켜 worker 완료 순서를 역전시킨다. + val gateway = CoroutineWorkerGateway( + scope = this, + workers = 4, + queueSize = 16, + decode = { bytes -> + val base = PacketBase.parseFrom(bytes) + delay(((total - base.nonce) * 5).toLong()) + base + }, + sink = { frame -> dispatched.add(frame.incomingNonce) }, + ) + + for (seq in 1..total) { + val frame = PacketBase.newBuilder().setTypeName("x").setNonce(seq).build().toByteArray() + gateway.submit(InboundFrame(seq = seq.toLong(), bytes = frame)) + } + + var elapsed = 0 + while (dispatched.size < total && elapsed < 5000) { + delay(5) + elapsed += 5 + } + + assertEquals((1..total).toList(), dispatched, "seq ordering not preserved despite out-of-order completion") + gateway.close() + } + + @Test + fun testWorkerGatewayReportsDecodeErrorAndAdvances() = runTest { + val seen = mutableListOf() + var errorCount = 0 + + val gateway = CoroutineWorkerGateway( + scope = this, + workers = 2, + queueSize = 8, + sink = { frame -> seen.add(frame.incomingNonce) }, + onError = { errorCount++ }, + ) + + val good1 = PacketBase.newBuilder().setTypeName("x").setNonce(1).build().toByteArray() + val good2 = PacketBase.newBuilder().setTypeName("x").setNonce(3).build().toByteArray() + + gateway.submit(InboundFrame(seq = 1, bytes = good1)) + gateway.submit(InboundFrame(seq = 2, bytes = byteArrayOf(0xff.toByte(), 0xff.toByte(), 0xff.toByte()))) + gateway.submit(InboundFrame(seq = 3, bytes = good2)) + + var elapsed = 0 + while (seen.size < 2 && elapsed < 2000) { + delay(5) + elapsed += 5 + } + + assertEquals(listOf(1, 3), seen, "sink should see good frames in order, skipping the bad one") + assertEquals(1, errorCount, "exactly one decode error should be reported") + gateway.close() + } + + @Test + fun testAttachedGatewayDispatchesThroughOnReceivedFrameInOrder() = runTest { + val transport = FakeTransport() + val communicator = Communicator(transport, testParserMap(), this) + + val received = mutableListOf() + addListenerTyped(communicator) { msg -> received.add(msg.index) } + + val total = 50 + // production attached path: gateway를 communicator에 attach하고 raw frame을 onReceivedFrame으로 + // ingest한다. gateway sink는 decoded frame을 coordinator inbound queue로만 전달한다. + val gateway = CoroutineWorkerGateway( + scope = this, + workers = 4, + queueSize = 64, + decode = { bytes -> + val base = PacketBase.parseFrom(bytes) + // 짝수 nonce를 더 오래 delay시켜 worker 완료 순서를 섞는다. + delay(if (base.nonce % 2 == 0) 8L else 2L) + base + }, + sink = { frame -> + communicator.enqueueInbound(frame.typeName, frame.data, frame.incomingNonce, frame.responseNonce) + }, + ) + communicator.attachInboundGateway(gateway) + + for (i in 1..total) { + val data = TestData.newBuilder().setIndex(i).build().toByteArray() + val raw = PacketBase.newBuilder() + .setTypeName(typeNameOf()) + .setNonce(i) + .setData(com.google.protobuf.ByteString.copyFrom(data)) + .build() + .toByteArray() + communicator.onReceivedFrame(raw) + } + + var elapsed = 0 + while (received.size < total && elapsed < 5000) { + delay(5) + elapsed += 5 + } + + assertEquals((1..total).toList(), received, "coordinator dispatch order should match input order") + communicator.close() + } + + @Test + fun testWsIngressSubmitsRawFramesToAttachedGateway() = runTest { + val connection = object : WsConnection { + override fun send(bytes: ByteArray): Boolean = true + override fun close() {} + } + val client = WsClient(connection, intervalSec = 30, waitSec = 10, parserMap = testParserMap()) + + // counting gateway: raw frame submit만 기록하고 decode하지 않는다. + // read loop가 gateway를 우회해 inline decode하면 submitted가 비어 테스트가 실패한다. + val submitted = Collections.synchronizedList(mutableListOf()) + val gateway = object : InboundGateway { + override suspend fun submit(frame: InboundFrame) { + submitted.add(frame.seq) + } + + override fun close() {} + } + client.communicator.attachInboundGateway(gateway) + + val total = 5 + for (i in 1..total) { + val data = TestData.newBuilder().setIndex(i).build().toByteArray() + val frame = PacketBase.newBuilder() + .setTypeName(typeNameOf()) + .setNonce(i) + .setData(com.google.protobuf.ByteString.copyFrom(data)) + .build() + .toByteArray() + client.receiveBytes(frame) + } + + withContext(Dispatchers.IO) { + val deadline = System.nanoTime() + 2_000_000_000L + while (System.nanoTime() < deadline && submitted.size < total) { + Thread.sleep(5) + } + } + + // ingress가 gateway에 frame을 submit했고 내부 seq가 수신 순서대로 부여되었는지 확인한다. + assertEquals((1L..total.toLong()).toList(), submitted.sorted(), "WS ingress should submit raw frames to the attached gateway") + client.communicator.shutdown() + } + + @Test + fun testWsGatewayDecodeErrorTriggersDisconnect() = runTest { + val connection = object : WsConnection { + var closed = false + override fun send(bytes: ByteArray): Boolean = true + override fun close() { + closed = true + } + } + val client = WsClient(connection, intervalSec = 30, waitSec = 10, parserMap = testParserMap()) + + // production worker-pool gateway를 enable한다. decode 불가 frame은 gateway onError → + // frameErrorHandler(WsClient가 onDisconnected로 등록) 경로로 transport disconnect를 유발해야 한다. + client.communicator.enableInboundGateway() + + client.receiveBytes(byteArrayOf(0xff.toByte(), 0xff.toByte(), 0xff.toByte())) + + withContext(Dispatchers.IO) { + val deadline = System.nanoTime() + 2_000_000_000L + while (System.nanoTime() < deadline && !connection.closed) { + Thread.sleep(5) + } + } + + assertTrue(connection.closed, "gateway decode error should reach the transport disconnect path") + assertTrue(!client.communicator.isAlive(), "communicator should be inactive after gateway decode error") + } + + @Test + fun testWorkerGatewayCloseIsIdempotentAndStopsDelivery() = runTest { + val dispatched = mutableListOf() + + val gateway = CoroutineWorkerGateway( + scope = this, + workers = 2, + queueSize = 8, + sink = { frame -> dispatched.add(frame.incomingNonce) }, + ) + + gateway.close() + gateway.close() // 멱등: 두 번 호출해도 안전해야 한다. + + // close 이후 submit은 drop되어 sink에 도달하지 않아야 한다. + val frame = PacketBase.newBuilder().setTypeName("x").setNonce(1).build().toByteArray() + gateway.submit(InboundFrame(seq = 1, bytes = frame)) + + delay(50) + assertTrue(dispatched.isEmpty(), "submit after close should be dropped") + } }