From effc5e17bd36fe7bb52a2a0017dcd5340fe61d16 Mon Sep 17 00:00:00 2001 From: toki Date: Mon, 1 Jun 2026 19:22:26 +0900 Subject: [PATCH] add inbound queue ordering milestone and update test matrix remote root --- .../scripts/run_matrix.sh | 2 +- agent-roadmap/ROADMAP.md | 1 + agent-roadmap/current.md | 4 +- .../milestones/inbound-queue-ordering.md | 94 +++++++++++++++++++ 4 files changed, 99 insertions(+), 2 deletions(-) create mode 100644 agent-roadmap/milestones/inbound-queue-ordering.md diff --git a/agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_matrix.sh b/agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_matrix.sh index 9760a55..11d65f8 100755 --- a/agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_matrix.sh +++ b/agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_matrix.sh @@ -28,7 +28,7 @@ log_dir="$(mktemp -d "${TMPDIR:-/tmp}/proto-socket-matrix.XXXXXX")" run_started_at="$(date -u +%Y-%m-%dT%H:%M:%SZ)" run_record_stamp="$(date -u +%Y%m%d-%H%M%S)" remote_web_host="${PROTO_SOCKET_DART_WEB_REMOTE:-toki@toki-labs.com}" -remote_web_root_expr="${PROTO_SOCKET_DART_WEB_REMOTE_ROOT:-~/works}" +remote_web_root_expr="${PROTO_SOCKET_DART_WEB_REMOTE_ROOT:-~/agent-work}" remote_web_repo_name="${PROTO_SOCKET_DART_WEB_REMOTE_REPO:-proto-socket-dart-web}" remote_web_repo_url="${PROTO_SOCKET_DART_WEB_REMOTE_URL:-$(git -C "$repo_root" config --get remote.origin.url 2>/dev/null || true)}" remote_web_ref="${PROTO_SOCKET_DART_WEB_REF:-$(git -C "$repo_root" rev-parse --abbrev-ref HEAD 2>/dev/null || echo master)}" diff --git a/agent-roadmap/ROADMAP.md b/agent-roadmap/ROADMAP.md index bccdd62..9e283de 100644 --- a/agent-roadmap/ROADMAP.md +++ b/agent-roadmap/ROADMAP.md @@ -28,6 +28,7 @@ Proto Socket은 여러 언어와 플랫폼에서 일관되게 동작하는 얇 ### 안정화와 유지 - [안정화 기준선](milestones/stability-baseline.md) - 상태: 완료; 목표: 현재 5개 언어 구현을 완성형 후보로 보고 안정성 판단과 유지 기준을 정리한다. +- [수신 큐와 처리 순서 보장](milestones/inbound-queue-ordering.md) - 상태: [계획]; 목표: 현재 5개 언어 구현에 per-connection 수신 큐와 언어별 worker gateway 후보를 도입해 수신 처리 순서, 자동 응답 출력 순서, thread-safe한 공유 상태 접근, 대량 처리 성능 개선 가능성을 보장한다. ### 남은 native platform 포팅 diff --git a/agent-roadmap/current.md b/agent-roadmap/current.md index a9f9e9c..7ff7ddb 100644 --- a/agent-roadmap/current.md +++ b/agent-roadmap/current.md @@ -2,7 +2,9 @@ ## 활성 Milestone -- 없음: 현재 진행 중인 활성 Milestone은 없다. 미완료로 남은 Milestone은 실제 소비 수요 대기 상태인 C#과 Swift 포팅 보류 항목이다. +- [계획] 수신 큐와 처리 순서 보장 + - 단계: 안정화와 유지 + - 경로: `agent-roadmap/milestones/inbound-queue-ordering.md` ## 선택 규칙 diff --git a/agent-roadmap/milestones/inbound-queue-ordering.md b/agent-roadmap/milestones/inbound-queue-ordering.md new file mode 100644 index 0000000..b56dd10 --- /dev/null +++ b/agent-roadmap/milestones/inbound-queue-ordering.md @@ -0,0 +1,94 @@ +# 수신 큐와 처리 순서 보장 + +## 목표 + +현재 Dart, Go, Kotlin, Python, TypeScript 구현에 per-connection 수신 큐를 도입해 메시지 수신, 핸들러 실행, 자동 응답 송신의 순서를 명시적으로 보장한다. read loop는 프레임 파싱 후 큐에 넣고, 단일 수신 coordinator가 FIFO로 처리해 thread-safe한 상태 접근과 예측 가능한 출력 순서를 제공한다. 대량 수신과 CPU 비용이 큰 decode/전처리는 언어별 worker gateway로 병렬화하되, 최종 dispatch는 coordinator가 `seq` 순서대로만 수행한다. + +## 단계 + +안정화와 유지 + +## 상태 + +[계획] + +## 구현 잠금 + +- 상태: 해제 +- 결정 필요: 없음 + +## 범위 + +- Dart, Go, Kotlin, Python, TypeScript의 공통 수신 파이프라인을 per-connection FIFO 큐 모델로 맞춘다. +- 일반 listener, request listener, pending response 처리의 순서와 동시성 경계를 언어별 구현에 반영한다. +- bounded inbound queue, close/drain/cancel, backpressure 동작을 각 런타임에 맞게 구현한다. +- 대량 raw packet 처리와 순수 decode/전처리는 언어별 worker gateway 후보로 분리하고, `seq` 기반 reorder로 출력 순서를 보존한다. +- 송신 `writeQueue`와 수신 큐가 충돌하지 않도록 응답 송신은 기존 송신 큐를 통해 직렬화한다. +- 순서 보장, 느린 핸들러, 큐 포화, 연결 종료 시 pending 처리에 대한 동일 언어 테스트와 크로스 언어 검증을 추가한다. + +## 기능 + +### Epic: [contract] 수신 처리 계약 + +수신 큐가 제공해야 하는 ordering, backpressure, close semantics를 모든 언어 구현의 공통 기준으로 정리한다. + +- [ ] [recv-contract] per-connection inbound queue 계약을 문서화한다. 검증: 수신 순서, 자동 응답 순서, 큐 capacity, close/drain/cancel 정책이 README 또는 PROTOCOL/PORTING_GUIDE 중 적절한 문서에 명시되어 있다. +- [ ] [thread-model] handlers, request handlers, pending response map 접근의 thread-safety 기준을 언어별 구현 메모에 반영한다. 검증: 각 언어 구현에서 수신 worker와 공유 상태 보호 방식이 코드 또는 테스트로 확인된다. +- [ ] [gateway-contract] worker gateway 계약을 문서화한다. 검증: gateway는 순수 raw bytes/decode/전처리만 담당하고, coordinator가 `seq` 순서 복원과 stateful dispatch를 단독 소유한다는 기준이 문서와 테스트에 반영되어 있다. + +### Epic: [impl] 언어별 수신 큐 구현 + +각 언어에서 read loop와 message dispatch 사이에 bounded FIFO 큐와 단일 consumer를 둔다. + +- [ ] [dart] Dart 구현에 inbound queue와 receive worker를 추가하고 request handler를 순차 처리한다. 검증: Dart 테스트에서 수신 순서, 자동 응답 순서, close 시 큐/pending 정리가 통과한다. +- [ ] [go] Go 구현에 inbound queue channel과 receive worker를 추가하고 request handler goroutine 분기를 제거하거나 순서 보장 경계 뒤로 옮긴다. 검증: Go 테스트에서 수신 순서, 자동 응답 순서, close 시 큐/pending 정리가 통과한다. +- [ ] [kotlin] Kotlin 구현에 inbound `Channel`과 receive coroutine을 추가하고 request handler launch 정책을 FIFO 처리로 맞춘다. 검증: Kotlin 테스트에서 수신 순서, 자동 응답 순서, close 시 큐/pending 정리가 통과한다. +- [ ] [python] Python 구현에 `asyncio.Queue` 기반 inbound queue와 receive task를 추가하고 request handler task 분기를 순서 보장 모델로 바꾼다. 검증: Python 테스트에서 수신 순서, 자동 응답 순서, close 시 큐/pending 정리가 통과한다. +- [ ] [typescript] TypeScript 구현에 inbound queue와 async receive worker를 추가하고 request handler 실행을 FIFO로 직렬화한다. 검증: TypeScript 테스트에서 수신 순서, 자동 응답 순서, close 시 큐/pending 정리가 통과한다. + +### Epic: [gateway] 언어별 worker gateway + +수신 큐 앞단에서 병렬화 가능한 순수 작업을 처리하되, stateful dispatch와 출력 순서는 receive coordinator가 보장한다. + +- [ ] [dart-gateway] Dart IO 구현에서 `../dart-app-core`의 isolate service 패턴을 참고해, 연결 또는 runtime 단위로 상시 유지되는 isolate gateway를 두고 `PacketBase` envelope decode, 대량 raw packet 처리, 병렬화 가능한 순수 bytes 작업을 분리한다. 검증: Dart IO 테스트에서 gateway 사용 시에도 수신 순서와 응답 순서가 유지되고, Dart web 조건부 export/stub 빌드가 깨지지 않는다. +- [ ] [go-gateway] Go 구현에서 goroutine worker pool과 bounded channel을 gateway 후보로 적용하고, 결과를 `seq` 기준으로 receive coordinator에 전달한다. 검증: 대량 packet decode 테스트에서 coordinator dispatch 순서가 입력 순서와 일치하고 race detector 기준 공유 상태 경합이 없다. +- [ ] [kotlin-gateway] Kotlin 구현에서 coroutine worker pool 또는 제한된 `CoroutineDispatcher`를 gateway 후보로 적용하고, 결과를 `seq` 기준으로 receive actor/coordinator에 전달한다. 검증: 대량 packet decode 테스트에서 coordinator dispatch 순서가 입력 순서와 일치하고 공유 상태는 actor/coordinator로 한정된다. +- [ ] [python-gateway] Python 구현에서 CPU-bound decode는 `ProcessPoolExecutor` 또는 장기 실행 process worker 후보로, IO/경량 전처리는 `asyncio` worker 후보로 분리해 benchmark 기반 적용 여부를 결정한다. 검증: 적용 시 `seq` 순서 복원과 close/cancel 처리가 통과하고, 작은 packet에서는 gateway 우회 fallback이 가능하다. +- [ ] [typescript-gateway] TypeScript 구현에서 Node는 `worker_threads`, browser target은 Web Worker 또는 fallback을 후보로 적용하고 transferable `ArrayBuffer`로 복사 비용을 줄인다. 검증: Node 테스트에서 gateway 사용 시 순서가 보존되고, browser build/runtime에서는 worker 미지원 환경 fallback이 동작한다. + +### Epic: [verify] 검증과 호환성 + +수신 큐 도입이 프로토콜 호환성과 기존 송신 직렬화 보장을 깨지 않는지 확인한다. + +- [ ] [unit-order] 모든 지원 언어에 순서 보장 테스트를 추가한다. 검증: 연속 수신된 일반 메시지와 request 메시지가 등록 순서대로 listener/handler에 전달된다. +- [ ] [slow-handler] 모든 지원 언어에 느린 request handler 테스트를 추가한다. 검증: 먼저 받은 요청의 응답이 먼저 송신 큐에 들어가며, 뒤 요청이 앞 요청을 추월하지 않는다. +- [ ] [queue-close] 모든 지원 언어에 큐 포화와 연결 종료 테스트를 추가한다. 검증: inbound queue가 꽉 찼을 때 backpressure가 적용되고, close 시 queued item과 pending request가 일관되게 취소된다. +- [ ] [gateway-bench] 언어별 worker gateway 성능 기준을 검증한다. 검증: 큰 payload 또는 burst traffic에서 gateway on/off 비교 결과를 기록하고, 이득이 불명확한 언어/환경은 안전한 fallback을 기본으로 둔다. +- [ ] [matrix] 전체 로컬 검증 매트릭스를 실행해 변경 후에도 동일 언어 및 크로스 언어 통신이 유지되는지 확인한다. 검증: `bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_matrix.sh --all` 통과. + +## 완료 리뷰 + +- 상태: 없음 +- 요청일: 없음 +- 완료 근거: 아직 구현 전 계획 상태다. +- 리뷰 필요: + - [ ] 사용자가 완료 결과를 확인했다 + - [ ] archive 이동을 승인했다 +- 리뷰 코멘트: 없음 + +## 범위 제외 + +- wire format, proto schema, nonce 필드 의미를 변경하지 않는다. +- 메시지 우선순위, 병렬 handler pool, application-level ordering key 같은 고수준 라우팅 정책은 추가하지 않는다. +- 수신 큐 구현을 이유로 C# 또는 Swift 포팅을 시작하지 않는다. +- package registry 릴리즈 작업을 시작하지 않는다. + +## 작업 컨텍스트 + +- 관련 경로: `dart/lib/src/communicator.dart`, `dart/lib/src/protobuf_client.dart`, `dart/lib/src/ws_protobuf_client_io.dart`, `go/communicator.go`, `go/tcp_client.go`, `go/ws_client.go`, `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`, `python/proto_socket/communicator.py`, `python/proto_socket/tcp_client.py`, `python/proto_socket/ws_client.py`, `typescript/src/communicator.ts`, `typescript/src/tcp_client.ts`, `typescript/src/node_ws_client.ts` +- 표준선: 각 connection은 bounded FIFO inbound queue 하나와 단일 receive worker 하나를 가진다. read loop는 프레임을 파싱해 큐에 넣고, worker가 response matching, request handling, listener dispatch를 순서대로 수행한다. 자동 응답은 기존 송신 큐에 enqueue되어 출력 순서를 유지한다. +- Dart isolate 후보: `../dart-app-core/lib/platform/isolate_manager.dart`의 장기 실행 isolate + `SendPort`/`ReceivePort` 이벤트 패턴은 참고하되, `dart:io`/앱 전역 system 초기화/범용 broadcast 구조는 복사하지 않는다. proto-socket에는 IO 전용 상시 worker gateway와 web fallback/stub을 둔다. gateway는 isolate spawn 비용을 반복하지 않고 `seq`가 붙은 raw bytes를 받아 병렬화 가능한 순수 작업을 처리한다. user listener/request handler, `pendingRequests`, transport/write queue는 isolate로 옮기지 않고 main isolate의 receive worker에서 순서와 상태 일관성을 유지한다. +- 언어별 gateway 후보: Go는 goroutine worker pool, Kotlin은 coroutine worker pool/dispatcher, Python은 process worker 또는 asyncio fallback, TypeScript는 Node `worker_threads`/browser Web Worker를 후보로 둔다. 모든 gateway 결과는 `seq`로 reorder된 뒤 coordinator가 dispatch하며, gateway는 수신 큐와 thread-safe 상태 소유 모델을 대체하지 않는다. +- 선행 작업: 안정화 기준선에서 확인된 현재 5개 언어 구현 +- 후속 작업: 수신 큐 구현 완료 후 안정화 기준선 재평가 +- 확인 필요: 없음