Update Kotlin files and archive G08 task artifacts
This commit is contained in:
parent
2ec43f2d00
commit
570cd8173c
10 changed files with 912 additions and 89 deletions
|
|
@ -0,0 +1,113 @@
|
|||
<!-- task=m-inbound-queue-ordering/06_kotlin_gateway plan=0 tag=API -->
|
||||
|
||||
# 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`를 작성한다.
|
||||
|
|
@ -0,0 +1,139 @@
|
|||
<!-- task=m-inbound-queue-ordering/06_kotlin_gateway plan=1 tag=REVIEW_API -->
|
||||
|
||||
# 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-<milestone-slug>`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, 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로 이동한다.
|
||||
|
|
@ -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
|
||||
|
||||
- 없음
|
||||
|
||||
## 후속 작업
|
||||
|
||||
- 없음
|
||||
|
|
@ -0,0 +1,110 @@
|
|||
<!-- task=m-inbound-queue-ordering/06_kotlin_gateway plan=1 tag=REVIEW_API -->
|
||||
|
||||
# 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`의 구현 에이전트 소유 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다.
|
||||
|
|
@ -1,71 +0,0 @@
|
|||
<!-- task=m-inbound-queue-ordering/06_kotlin_gateway plan=0 tag=API -->
|
||||
|
||||
# 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
|
||||
```
|
||||
|
|
@ -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<String, (ByteArray) -> MessageLite> = ConcurrentHashMap()
|
||||
private val handlers = mutableMapOf<String, MutableList<(MessageLite) -> 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<Long, GatewayResult>()
|
||||
|
||||
/** res를 버퍼하고 지금 순서대로 release 가능한 결과들을 seq 순으로 반환한다. */
|
||||
fun release(res: GatewayResult): List<GatewayResult> {
|
||||
if (res.seq < nextSeq) {
|
||||
// 이미 release되었거나 중복/지연 도착이므로 두 번 방출하지 않도록 무시한다.
|
||||
return emptyList()
|
||||
}
|
||||
pending[res.seq] = res
|
||||
val ready = mutableListOf<GatewayResult>()
|
||||
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<InboundFrame>
|
||||
private val results: Channel<GatewayResult>
|
||||
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 <reified T : Message> addListenerTyped(
|
||||
communicator: Communicator,
|
||||
noinline fn: (T) -> Unit,
|
||||
|
|
|
|||
|
|
@ -34,6 +34,7 @@ class TcpClient private constructor(
|
|||
|
||||
init {
|
||||
communicator.setWriteErrorHandler { onDisconnected() }
|
||||
communicator.setFrameErrorHandler { onDisconnected() }
|
||||
communicator.addListener(typeNameOf<HeartBeat>()) { 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()
|
||||
|
|
|
|||
|
|
@ -40,6 +40,7 @@ class WsClient internal constructor(
|
|||
|
||||
init {
|
||||
communicator.setWriteErrorHandler { onDisconnected() }
|
||||
communicator.setFrameErrorHandler { onDisconnected() }
|
||||
communicator.addListener(typeNameOf<HeartBeat>()) { 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 {
|
||||
|
|
|
|||
|
|
@ -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<Int>()
|
||||
|
||||
// 낮은 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<Int>()
|
||||
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<Int>()
|
||||
addListenerTyped<TestData>(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<TestData>())
|
||||
.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<Long>())
|
||||
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<TestData>())
|
||||
.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<Int>()
|
||||
|
||||
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")
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue