proto-socket/agent-roadmap/archive/phase/stability-maintenance/milestones/inbound-queue-ordering.md

수신 큐와 처리 순서 보장

목표

현재 Dart, Go, Kotlin, Python, TypeScript 구현에 per-connection 수신 큐를 도입해 메시지 수신, 핸들러 실행, 자동 응답 송신의 순서를 명시적으로 보장한다. read loop는 프레임 파싱 후 큐에 넣고, 단일 수신 coordinator가 FIFO로 처리해 thread-safe한 상태 접근과 예측 가능한 출력 순서를 제공한다. 대량 수신과 CPU 비용이 큰 decode/전처리는 언어별 worker gateway로 병렬화하되, 최종 dispatch는 coordinator가 seq 순서대로만 수행한다. 이 Milestone의 안정화 기준은 단순 protobuf 직렬화 호환을 넘어서, protobuf payload를 기반으로 한 고속 병렬 request-response와 burst inbound traffic에서도 nonce 매칭, 순서 보장, backpressure, close cleanup이 무너지지 않는 구조를 확인하는 것이다.

단계

안정화와 유지

상태

완료

구현 잠금

• 상태: 해제
• 결정 필요: 없음

범위

• 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 처리에 대한 동일 언어 테스트와 크로스 언어 검증을 추가한다.
• 빠른 왕복 메시징과 burst traffic 스트레스 기준선을 추가해, high-concurrency 환경에서 timeout, nonce mismatch, 응답 추월, 무제한 메모리 증가가 없는지 측정한다.

기능

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가 일관되게 취소된다.
• [rt-bench] 빠른 request-response 왕복 메시징 벤치마크를 추가한다. 검증: 지원 언어의 same-language TCP/WS 경로에서 순차와 동시 sendRequest 부하를 측정하고, concurrency 1/16/64/256 기준의 throughput, p50/p95/p99 latency, timeout/error count, nonce mismatch count를 기록한다. 기준선은 각 실행에서 nonce mismatch 0, response type mismatch 0, unexpected timeout 0이며, 절대 성능 수치는 환경 의존 baseline으로 저장한다.
• [burst-stress] burst inbound traffic 스트레스 테스트를 추가한다. 검증: 1k/10k 이상 연속 inbound message와 request burst에서 listener/handler dispatch 순서가 입력 순서와 일치하고, queue full 구간에서 backpressure 또는 disconnect 정책이 계약대로 동작하며, 처리 완료 후 queue/pending 상태가 비어 있다.
• [sustained-load] 지속 부하 테스트 기준선을 추가한다. 검증: 최소 30초 이상 configurable sustained request-response 부하를 실행해 처리량, p95/p99 latency, peak memory 또는 runtime별 관측 가능한 메모리 지표를 기록하고, 테스트 종료 후 close/drain/cancel cleanup이 완료된다.
• [parallel-clients] 다중 connection 병렬 스트레스 테스트를 추가한다. 검증: 여러 client connection이 동시에 request-response를 수행할 때 connection별 FIFO와 nonce 매칭이 독립적으로 유지되고, 한 connection의 느린 handler나 queue 포화가 다른 connection의 pending response 상태를 오염시키지 않는다.
• [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 통과.

완료 리뷰

• 상태: 승인됨
• 요청일: 2026-06-02
• 완료 근거: contract, impl, gateway Epic 완료 근거에 더해 09_verify_ordering과 10+09_stress_baseline 완료 로그가 PASS로 정리되어 verify Epic과 스트레스 기준선까지 완료 후보가 되었다.
• 리뷰 필요:
  • 사용자가 완료 결과를 확인했다
  • archive 이동을 승인했다
• 리뷰 코멘트: 모든 기능 Task와 명시 검증이 완료 근거와 함께 충족되어 완료 처리했다. 후속 성능 보강은 고성능 병렬 운용 기준선과 최적화 Milestone에서 진행한다.

범위 제외

• 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되어 출력 순서를 유지한다.
• 성능/스트레스 표준선: 이 프로젝트의 제품 차별점은 protobuf 직렬화 자체가 아니라 protobuf payload를 고속 병렬 통신에서 안전하게 운반하는 런타임 구조다. benchmark는 절대 성능 합격선을 고정하기보다 local 환경별 baseline을 남기고, 안정성 합격선은 nonce mismatch 0, response type mismatch 0, unexpected timeout 0, connection별 FIFO 위반 0, 종료 후 pending/queue leak 0으로 둔다. roundtrip benchmark는 순차/동시, burst stress, sustained load, multi-client 축을 분리해 병목 위치를 판별할 수 있어야 한다.
• 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개 언어 구현
• 후속 작업: 고성능 병렬 운용 성능 보강은 agent-roadmap/milestones/high-performance-parallel-operations.md로 분리한다.
• 완료 근거:
  • [recv-contract] agent-task/archive/2026/06/m-inbound-queue-ordering/01_contract_docs/complete.log PASS. 검증: agent-test/runs/20260601-105247-proto-socket-full-matrix.md.
  • [thread-model] agent-task/archive/2026/06/m-inbound-queue-ordering/02+01_receive_thread_model/complete.log PASS. 검증: agent-test/runs/20260601-201443-proto-socket-full-matrix.md.
  • [gateway-contract] agent-task/archive/2026/06/m-inbound-queue-ordering/03+02_gateway_contract/complete.log PASS. 검증: agent-test/runs/20260601-213027-proto-socket-full-matrix.md, Dart/Kotlin/TypeScript communicator reorder anchor 테스트 통과.
  • [dart] dart/lib/src/communicator.dart의 inbound queue/receive worker와 dart/test/communicator_test.dart PASS. 검증: dart test test/communicator_test.dart 16/16 PASS.
  • [go] go/communicator.go의 inbound queue channel/receive loop와 go/test/communicator_test.go PASS. 검증: go test -count=1 ./... PASS.
  • [kotlin] kotlin/src/main/kotlin/com/tokilabs/proto_socket/Communicator.kt의 inbound Channel/receive coroutine과 CommunicatorTest.kt PASS. 검증: ./gradlew test --tests com.tokilabs.proto_socket.CommunicatorTest --rerun-tasks PASS.
  • [python] python/proto_socket/communicator.py의 asyncio.Queue/receive task와 python/test/test_communicator.py PASS. 검증: python3 -m pytest -q test/test_communicator.py 13 passed.
  • [typescript] typescript/src/communicator.ts의 inbound queue/async worker와 typescript/test/communicator.test.ts PASS. 검증: npm test -- communicator 16/16 PASS.
  • [dart-gateway] agent-task/archive/2026/06/m-inbound-queue-ordering/04_dart_gateway/complete.log PASS. 검증: agent-test/runs/20260602-012917-proto-socket-full-matrix.md.
  • [go-gateway] agent-task/archive/2026/06/m-inbound-queue-ordering/05_go_gateway/complete.log PASS. 검증: agent-test/runs/20260602-022853-proto-socket-full-matrix.md.
  • [kotlin-gateway] agent-task/archive/2026/06/m-inbound-queue-ordering/06_kotlin_gateway/complete.log PASS. 검증: agent-test/runs/20260602-041819-proto-socket-full-matrix.md.
  • [python-gateway] agent-task/archive/2026/06/m-inbound-queue-ordering/07_python_gateway/complete.log PASS. 검증: cd python && python3 -m pytest -q test/test_communicator.py, bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_matrix.sh --all, git diff --check.
  • [typescript-gateway] agent-task/archive/2026/06/m-inbound-queue-ordering/08_typescript_gateway/complete.log PASS. 검증: agent-test/runs/20260602-074418-proto-socket-full-matrix.md.
  • [unit-order] agent-task/archive/2026/06/m-inbound-queue-ordering/09_verify_ordering/complete.log PASS. 검증: agent-test/runs/20260602-093257-proto-socket-full-matrix.md, focused communicator suites PASS.
  • [slow-handler] agent-task/archive/2026/06/m-inbound-queue-ordering/09_verify_ordering/complete.log PASS. 검증: Dart 21 tests, Go full suite, Kotlin CommunicatorTest, Python 20 passed, TypeScript 24 passed.
  • [queue-close] agent-task/archive/2026/06/m-inbound-queue-ordering/09_verify_ordering/complete.log PASS. 검증: queue full backpressure, graceful close drain, pending cleanup focused suites PASS.
  • [rt-bench] agent-task/archive/2026/06/m-inbound-queue-ordering/10+09_stress_baseline/complete.log PASS. 검증: run_stress.sh --quick, SUMMARY|status=PASS|mode=quick|profiles=roundtrip,burst,sustained,parallel,gateway|rows=9|stability_violations=0.
  • [burst-stress] agent-task/archive/2026/06/m-inbound-queue-ordering/10+09_stress_baseline/complete.log PASS. 검증: run_stress.sh --quick --profile roundtrip,burst,sustained,parallel, stability_violations=0.
  • [sustained-load] agent-task/archive/2026/06/m-inbound-queue-ordering/10+09_stress_baseline/complete.log PASS. 검증: run_stress.sh --quick --profile roundtrip,burst,sustained,parallel, sustained profile included.
  • [parallel-clients] agent-task/archive/2026/06/m-inbound-queue-ordering/10+09_stress_baseline/complete.log PASS. 검증: run_stress.sh --quick --profile roundtrip,burst,sustained,parallel, parallel profile included.
  • [gateway-bench] agent-task/archive/2026/06/m-inbound-queue-ordering/10+09_stress_baseline/complete.log PASS. 검증: run_stress.sh --quick, gateway profile included.
  • [matrix] agent-task/archive/2026/06/m-inbound-queue-ordering/10+09_stress_baseline/complete.log PASS. 검증: agent-test/runs/20260602-102155-proto-socket-full-matrix.md.
• 확인 필요: 없음