diff --git a/agent-task/edge_node_integration/code_review_0.log b/agent-task/edge_node_integration/code_review_0.log new file mode 100644 index 0000000..c6bddbc --- /dev/null +++ b/agent-task/edge_node_integration/code_review_0.log @@ -0,0 +1,147 @@ + + +# Code Review Reference - TEST + +## 개요 + +date=2026-05-02 +task=edge_node_integration, plan=0, tag=TEST + +## 이 파일을 읽는 리뷰 에이전트에게 + +각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. +리뷰 완료 후 반드시 아래 순서로 아카이브하세요. + +1. `CODE_REVIEW.md` → `code_review_N.log` (N = 기존 code_review_*.log 수) +2. `PLAN.md` → `plan_M.log` (M = 기존 plan_*.log 수) +3. PASS인 경우 `complete.log` 작성 후 종료. WARN/FAIL인 경우 새 `PLAN.md` + `CODE_REVIEW.md` 스텁 작성. + +--- + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [TEST-1] Edge Server 통합 테스트 | [x] | +| [TEST-2] Node Client 통합 테스트 | [x] | + +## 계획 대비 변경 사항 + +- 계획서의 요구 파일(`apps/edge/internal/transport/integration_test.go`, `apps/node/internal/transport/integration_test.go`)만 수정했다. +- `apps/node/internal/transport/integration_test.go`에서는 고정 `time.Sleep` 대신 서버 accept 채널 대기를 사용해 플래키 가능성을 줄였다. +- 최종 검증은 계획서의 명령(`go test -v ./apps/edge/... ./apps/node/...`)까지 확대 실행해 transport 외 연관 패키지 회귀도 함께 확인했다. + +## 주요 설계 결정 + +- Go `internal` 규칙을 지키기 위해 edge/node 통합 테스트를 각 앱의 `internal/transport` 패키지 안에서 분리 유지했다. +- 양쪽 테스트 모두 실제 TCP 소켓을 열고, 반대편 역할은 `common-proto-socket` 기반 mock client/server로 대체해 네트워크 계층을 직접 검증했다. +- Edge 테스트는 registry 등록 여부뿐 아니라 등록된 `NodeEntry`의 `NodeID`/adapter 메타데이터까지 확인하도록 구성했다. +- Node 테스트는 `CapabilityRequest` 응답 확인에 더해 `RunRequest` 수신과 `RunEvent` 반환까지 왕복 흐름으로 검증했다. + +## 리뷰어를 위한 체크포인트 + +- `apps/edge/internal/transport/integration_test.go`에서 소켓 기반 연결을 통해 Node 등록 흐름이 정상 작동하는가? +- `apps/node/internal/transport/integration_test.go`에서 Mock Server와의 소켓 연결 및 메시지 처리(`CapabilityRequest`, `RunRequest` 등)가 정상 작동하는가? +- 테스트 종료 시 서버 포트가 정상 릴리즈되는가? + +## 검증 결과 + +### 중간 검증 +``` +$ go test -count=1 -v ./apps/edge/internal/transport ./apps/node/internal/transport +=== RUN TestEdgeServerIntegration +--- PASS: TestEdgeServerIntegration (0.03s) +PASS +ok iop/apps/edge/internal/transport 0.655s +=== RUN TestNodeParserMap_RunRequest +--- PASS: TestNodeParserMap_RunRequest (0.00s) +=== RUN TestNodeParserMap_CancelRequest +--- PASS: TestNodeParserMap_CancelRequest (0.00s) +=== RUN TestNodeParserMap_CapabilityRequest +--- PASS: TestNodeParserMap_CapabilityRequest (0.00s) +=== RUN TestNodeClientIntegration +--- PASS: TestNodeClientIntegration (0.00s) +=== RUN TestSession_RegisterCancel_CancelRun +--- PASS: TestSession_RegisterCancel_CancelRun (0.00s) +=== RUN TestSession_DeregisterCancel +--- PASS: TestSession_DeregisterCancel (0.00s) +=== RUN TestSession_CancelRun_UnknownID +--- PASS: TestSession_CancelRun_UnknownID (0.00s) +=== RUN TestSession_ConcurrentRegisterCancel +--- PASS: TestSession_ConcurrentRegisterCancel (0.00s) +PASS +ok iop/apps/node/internal/transport 1.143s +``` + +### 최종 검증 +``` +$ go test -count=1 -v ./apps/edge/... ./apps/node/... +? iop/apps/edge/cmd/edge [no test files] +? iop/apps/edge/internal/bootstrap [no test files] +=== RUN TestRegistry_RegisterAndCount +--- PASS: TestRegistry_RegisterAndCount (0.00s) +=== RUN TestRegistry_Pick_Empty +--- PASS: TestRegistry_Pick_Empty (0.00s) +=== RUN TestRegistry_Unregister +--- PASS: TestRegistry_Unregister (0.00s) +PASS +ok iop/apps/edge/internal/node 0.265s +=== RUN TestEdgeServerIntegration +--- PASS: TestEdgeServerIntegration (0.03s) +PASS +ok iop/apps/edge/internal/transport 0.807s +? iop/apps/node/cmd/node [no test files] +? iop/apps/node/internal/adapters [no test files] +? iop/apps/node/internal/adapters/cli [no test files] +? iop/apps/node/internal/adapters/mock [no test files] +? iop/apps/node/internal/adapters/ollama [no test files] +? iop/apps/node/internal/adapters/vllm [no test files] +? iop/apps/node/internal/bootstrap [no test files] +=== RUN TestOnCapabilityRequest +--- PASS: TestOnCapabilityRequest (0.00s) +=== RUN TestOnRunRequest_RouterError +--- PASS: TestOnRunRequest_RouterError (0.00s) +=== RUN TestOnRunRequest_AdapterNotFound +--- PASS: TestOnRunRequest_AdapterNotFound (0.00s) +=== RUN TestOnRunRequest_Success +--- PASS: TestOnRunRequest_Success (0.00s) +=== RUN TestOnCancel_CallsCancelFn +--- PASS: TestOnCancel_CallsCancelFn (0.00s) +PASS +ok iop/apps/node/internal/node 1.299s +? iop/apps/node/internal/router [no test files] +? iop/apps/node/internal/runtime [no test files] +? iop/apps/node/internal/store [no test files] +=== RUN TestNodeParserMap_RunRequest +--- PASS: TestNodeParserMap_RunRequest (0.00s) +=== RUN TestNodeParserMap_CancelRequest +--- PASS: TestNodeParserMap_CancelRequest (0.00s) +=== RUN TestNodeParserMap_CapabilityRequest +--- PASS: TestNodeParserMap_CapabilityRequest (0.00s) +=== RUN TestNodeClientIntegration +--- PASS: TestNodeClientIntegration (0.10s) +=== RUN TestSession_RegisterCancel_CancelRun +--- PASS: TestSession_RegisterCancel_CancelRun (0.00s) +=== RUN TestSession_DeregisterCancel +--- PASS: TestSession_DeregisterCancel (0.00s) +=== RUN TestSession_CancelRun_UnknownID +--- PASS: TestSession_CancelRun_UnknownID (0.00s) +=== RUN TestSession_ConcurrentRegisterCancel +--- PASS: TestSession_ConcurrentRegisterCancel (0.00s) +PASS +ok iop/apps/node/internal/transport 1.367s +``` + +## 코드리뷰 결과 + +- **종합 판정**: `PASS` +- **차원별 평가**: + - Correctness: Pass + - Completeness: Pass + - Test coverage: Pass + - API contract: Pass + - Code quality: Pass + - Plan deviation: Pass (internal 패키지 정책에 맞춘 분리 구현은 타당함) + - Verification trust: Pass +- **발견된 문제**: 없음 +- **다음 단계**: PASS (완료) diff --git a/agent-task/edge_node_integration/complete.log b/agent-task/edge_node_integration/complete.log new file mode 100644 index 0000000..a178f21 --- /dev/null +++ b/agent-task/edge_node_integration/complete.log @@ -0,0 +1,23 @@ + + +# Complete - edge_node_integration + +date=2026-05-02 +result=PASS + +## 요약 + +1회 plan 루프를 거쳐 edge_node_integration 태스크 완료. + +## 루프 이력 + +| 파일 쌍 | 판정 | +|---------|------| +| `plan_0.log` / `code_review_0.log` | PASS | + +## 최종 리뷰 요약 + +- Edge Server 통합 네트워크 테스트 (`apps/edge/internal/transport/integration_test.go`) 작성 및 통과 확인 +- Node Client 통합 네트워크 테스트 (`apps/node/internal/transport/integration_test.go`) 작성 및 통과 확인 +- Mock Client/Server를 사용하여 Go의 `internal` 패키지 규칙 준수 하에 양측의 패킷(CapabilityRequest, RunRequest 등) 통신 흐름 완전 검증 +- 테스트 종료 시 서버 포트 정상 릴리즈 등 자원 누수 없음 확인 diff --git a/agent-task/edge_node_integration/plan_0.log b/agent-task/edge_node_integration/plan_0.log new file mode 100644 index 0000000..b096c03 --- /dev/null +++ b/agent-task/edge_node_integration/plan_0.log @@ -0,0 +1,38 @@ + +# Edge-Node Integration Test + +## 이 파일을 읽는 구현 에이전트에게 +작업을 진행하며 `CODE_REVIEW.md` 스텁의 체크리스트를 채우고, +중간/최종 검증 결과를 작성해 주십시오. + +## 배경 +Go의 `internal` 패키지 규칙으로 인해 `apps/edge`와 `apps/node`를 하나의 외부 패키지에서 동시에 임포트할 수 없습니다. 따라서 양측의 네트워크 통합 테스트를 각각의 모듈 내부(`apps/edge/internal/transport`, `apps/node/internal/transport`)에서 수행하도록 나누어 구성합니다. + +### [TEST-1] Edge Server 통합 테스트 +- **문제**: Edge 서버의 네트워크 등록 절차가 실제 소켓 환경에서 검증되지 않음. +- **해결 방법**: `apps/edge/internal/transport/integration_test.go`를 작성하여 `Server`를 구동하고, `toki.TcpClient`를 통해 Node의 역할을 모방해 접속한 뒤 Registry 등록을 검증. +- **수정 파일 및 체크리스트**: + - `apps/edge/internal/transport/integration_test.go` + - [ ] `transport.Server` 구동 + - [ ] Node 클라이언트 소켓 연결 및 `CapabilityResponse` 전송 + - [ ] Edge Registry 내 정상 등록 대기 및 확인 + +### [TEST-2] Node Client 통합 테스트 +- **문제**: Node 클라이언트의 서버 접속 및 메시지 처리 흐름이 네트워크 레벨에서 검증되지 않음. +- **해결 방법**: `apps/node/internal/transport/integration_test.go`를 작성하여 임의의 `toki.TcpServer`를 구동해 Edge 서버를 모방하고, Node 클라이언트가 `DialEdge`로 접속하여 패킷을 교환하는지 검증. +- **수정 파일 및 체크리스트**: + - `apps/node/internal/transport/integration_test.go` + - [ ] Mock Edge 서버 구동 (임의 포트) + - [ ] `transport.DialEdge` 호출로 접속 + - [ ] Mock Handler 동작 검증 (`CapabilityRequest`, `RunRequest` 등) + +## 수정 파일 요약 +| 파일 경로 | 작업 항목 | +|-----------|-----------| +| `apps/edge/internal/transport/integration_test.go` | [TEST-1] | +| `apps/node/internal/transport/integration_test.go` | [TEST-2] | + +## 최종 검증 +```bash +go test -v ./apps/edge/... ./apps/node/... +``` diff --git a/apps/edge/internal/transport/integration_test.go b/apps/edge/internal/transport/integration_test.go new file mode 100644 index 0000000..6a98812 --- /dev/null +++ b/apps/edge/internal/transport/integration_test.go @@ -0,0 +1,122 @@ +package transport_test + +import ( + "context" + "fmt" + "net" + "testing" + "time" + + "go.uber.org/zap" + "google.golang.org/protobuf/proto" + + toki "git.toki-labs.com/toki/common-proto-socket/go" + + edgenode "iop/apps/edge/internal/node" + "iop/apps/edge/internal/transport" + iop "iop/proto/gen/iop" +) + +func getFreePort(t *testing.T) string { + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + addr := l.Addr().String() + l.Close() + return addr +} + +func waitForRegistryEntry(ctx context.Context, registry *edgenode.Registry, nodeID string) (*edgenode.NodeEntry, bool) { + ticker := time.NewTicker(25 * time.Millisecond) + defer ticker.Stop() + + for { + if entry, ok := registry.Get(nodeID); ok { + return entry, true + } + + select { + case <-ctx.Done(): + return nil, false + case <-ticker.C: + } + } +} + +func TestEdgeServerIntegration(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + logger := zap.NewNop() + listenAddr := getFreePort(t) + registry := edgenode.NewRegistry() + + // 1. Edge 서버 구동 + server, err := transport.NewServer(listenAddr, registry, logger) + if err != nil { + t.Fatalf("failed to create server: %v", err) + } + if err := server.Start(ctx); err != nil { + t.Fatalf("failed to start server: %v", err) + } + defer server.Stop() + + // 2. Mock Node 클라이언트 접속 + host, portStr, _ := net.SplitHostPort(listenAddr) + port := 0 + fmt.Sscanf(portStr, "%d", &port) + + // Node parser map (mock) + nodeParser := toki.ParserMap{ + toki.TypeNameOf(&iop.CapabilityRequest{}): func(b []byte) (proto.Message, error) { + m := &iop.CapabilityRequest{} + return m, proto.Unmarshal(b, m) + }, + toki.TypeNameOf(&iop.RunRequest{}): func(b []byte) (proto.Message, error) { + m := &iop.RunRequest{} + return m, proto.Unmarshal(b, m) + }, + } + + client, err := toki.DialTcp(ctx, host, port, 30, 10, nodeParser) + if err != nil { + t.Fatalf("failed to dial edge server: %v", err) + } + defer client.Close() + + // 3. 서버가 CapabilityRequest를 보내오면 CapabilityResponse로 응답 + toki.AddRequestListenerTyped[*iop.CapabilityRequest, *iop.CapabilityResponse]( + &client.Communicator, + func(_ *iop.CapabilityRequest) (*iop.CapabilityResponse, error) { + return &iop.CapabilityResponse{ + NodeId: "test-node", + Adapters: []*iop.AdapterInfo{ + {Name: "test-adapter", Models: []string{"v1"}, MaxConcurrency: 1}, + }, + }, nil + }, + ) + + // 4. Registry 등록 여부 확인 + entry, ok := waitForRegistryEntry(ctx, registry, "test-node") + if !ok { + t.Fatal("node was not registered in edge registry within timeout") + } + + if got := registry.Count(); got != 1 { + t.Fatalf("expected registry count 1, got %d", got) + } + if entry == nil { + t.Fatal("expected registry entry, got nil") + } + if got := entry.NodeID; got != "test-node" { + t.Fatalf("expected node id %q, got %q", "test-node", got) + } + if len(entry.Adapters) != 1 { + t.Fatalf("expected 1 adapter, got %d", len(entry.Adapters)) + } + if got := entry.Adapters[0].GetName(); got != "test-adapter" { + t.Fatalf("expected adapter name %q, got %q", "test-adapter", got) + } +} diff --git a/apps/node/internal/transport/integration_test.go b/apps/node/internal/transport/integration_test.go new file mode 100644 index 0000000..e222b49 --- /dev/null +++ b/apps/node/internal/transport/integration_test.go @@ -0,0 +1,158 @@ +package transport_test + +import ( + "context" + "fmt" + "net" + "testing" + "time" + + "go.uber.org/zap" + "google.golang.org/protobuf/proto" + + toki "git.toki-labs.com/toki/common-proto-socket/go" + + "iop/apps/node/internal/transport" + iop "iop/proto/gen/iop" +) + +type mockHandler struct { + runReqCh chan *iop.RunRequest +} + +func (m *mockHandler) OnRunRequest(ctx context.Context, sess *transport.Session, req *iop.RunRequest) error { + m.runReqCh <- req + return sess.Send(&iop.RunEvent{RunId: req.GetRunId(), Type: "test_event"}) +} + +func (m *mockHandler) OnCapabilityRequest(ctx context.Context, sess *transport.Session) (*iop.CapabilityResponse, error) { + return &iop.CapabilityResponse{ + NodeId: "test-node", + Adapters: []*iop.AdapterInfo{ + {Name: "test-adapter", Models: []string{"v1"}, MaxConcurrency: 1}, + }, + }, nil +} + +func (m *mockHandler) OnCancel(ctx context.Context, sess *transport.Session, req *iop.CancelRequest) error { + return nil +} + +func getFreePort(t *testing.T) string { + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + addr := l.Addr().String() + l.Close() + return addr +} + +func edgeParserMap() toki.ParserMap { + return toki.ParserMap{ + toki.TypeNameOf(&iop.RunEvent{}): func(b []byte) (proto.Message, error) { + m := &iop.RunEvent{} + return m, proto.Unmarshal(b, m) + }, + toki.TypeNameOf(&iop.CapabilityResponse{}): func(b []byte) (proto.Message, error) { + m := &iop.CapabilityResponse{} + return m, proto.Unmarshal(b, m) + }, + } +} + +func waitForAcceptedClient(t *testing.T, acceptedCh <-chan *toki.TcpClient) *toki.TcpClient { + t.Helper() + + select { + case client := <-acceptedCh: + return client + case <-time.After(2 * time.Second): + t.Fatal("edge server did not accept connection") + return nil + } +} + +func TestNodeClientIntegration(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + logger := zap.NewNop() + listenAddr := getFreePort(t) + host, portStr, _ := net.SplitHostPort(listenAddr) + port := 0 + fmt.Sscanf(portStr, "%d", &port) + + // 1. Mock Edge 서버 구동 + acceptedCh := make(chan *toki.TcpClient, 1) + server := toki.NewTcpServer(host, port, func(conn net.Conn) *toki.TcpClient { + client := toki.NewTcpClient(conn, 30, 10, edgeParserMap()) + acceptedCh <- client + return client + }) + if err := server.Start(ctx); err != nil { + t.Fatalf("failed to start mock edge server: %v", err) + } + defer server.Stop() + + // 2. Node 클라이언트 접속 + handler := &mockHandler{ + runReqCh: make(chan *iop.RunRequest, 1), + } + + sess, err := transport.DialEdge(ctx, listenAddr, handler, "test-node", logger) + if err != nil { + t.Fatalf("failed to dial edge: %v", err) + } + defer sess.Close() + + edgeClient := waitForAcceptedClient(t, acceptedCh) + runEventCh := make(chan *iop.RunEvent, 1) + toki.AddListenerTyped[*iop.RunEvent](&edgeClient.Communicator, func(event *iop.RunEvent) { + runEventCh <- event + }) + + // 3. Edge -> Node 로 CapabilityRequest 전송 + resp, err := toki.SendRequestTyped[*iop.CapabilityRequest, *iop.CapabilityResponse]( + &edgeClient.Communicator, + &iop.CapabilityRequest{}, + 2*time.Second, + ) + if err != nil { + t.Fatalf("failed to send capability request: %v", err) + } + if resp.GetNodeId() != "test-node" { + t.Fatalf("unexpected node id from capability response: %q", resp.GetNodeId()) + } + + // 4. Edge -> Node 로 RunRequest 전송 + runReq := &iop.RunRequest{ + RunId: "test-run", + Adapter: "test-adapter", + } + if err := edgeClient.Send(runReq); err != nil { + t.Fatalf("failed to send run request: %v", err) + } + + // 5. Node에서 RunRequest 수신 확인 + select { + case receivedReq := <-handler.runReqCh: + if receivedReq.GetRunId() != runReq.GetRunId() { + t.Fatalf("expected run id %q, got %q", runReq.GetRunId(), receivedReq.GetRunId()) + } + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for run request on node handler") + } + + select { + case event := <-runEventCh: + if event.GetRunId() != runReq.GetRunId() { + t.Fatalf("expected run event id %q, got %q", runReq.GetRunId(), event.GetRunId()) + } + if event.GetType() != "test_event" { + t.Fatalf("expected run event type %q, got %q", "test_event", event.GetType()) + } + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for run event from node session") + } +}