From 7df7c6f52edd9898efbbb47ccc71d78ae40986e8 Mon Sep 17 00:00:00 2001 From: toki Date: Sat, 30 May 2026 22:30:18 +0900 Subject: [PATCH] chore: update control-plane edge wire baseline and related components - Move completed review/plan artifacts to archive - Update control-plane main, wire, edge components - Update edge bootstrap runtime and controlplane connector - Update generated proto files --- .../code_review_cloud_G08_0.log | 0 .../code_review_cloud_G08_1.log} | 95 ++++- .../04_status_snapshot/complete.log | 46 ++ .../04_status_snapshot/plan_cloud_G08_0.log | 0 .../04_status_snapshot/plan_cloud_G08_1.log} | 12 +- .../code_review_cloud_G08_0.log | 401 ++++++++++++++++++ .../code_review_cloud_G08_1.log | 285 +++++++++++++ .../05+04_event_relay_boundary/complete.log | 49 +++ .../plan_cloud_G08_0.log} | 0 .../plan_cloud_G08_1.log | 179 ++++++++ .../CODE_REVIEW-cloud-G08.md | 204 --------- apps/control-plane/cmd/control-plane/main.go | 67 ++- .../cmd/control-plane/main_test.go | 94 ++++ apps/control-plane/internal/wire/edge.go | 4 + .../internal/wire/edge_registry.go | 112 ++++- .../internal/wire/edge_server.go | 11 + .../internal/wire/edge_server_test.go | 163 +++++++ apps/control-plane/internal/wire/edge_test.go | 53 +++ apps/edge/internal/bootstrap/runtime.go | 1 + apps/edge/internal/bootstrap/runtime_test.go | 13 + apps/edge/internal/controlplane/connector.go | 72 ++++ .../internal/controlplane/connector_test.go | 136 +++++- proto/gen/iop/control.pb.go | 2 +- 23 files changed, 1749 insertions(+), 250 deletions(-) rename agent-task/{ => archive/2026/05}/m-control-plane-edge-wire-baseline/04_status_snapshot/code_review_cloud_G08_0.log (100%) rename agent-task/{m-control-plane-edge-wire-baseline/04_status_snapshot/CODE_REVIEW-cloud-G08.md => archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/code_review_cloud_G08_1.log} (64%) create mode 100644 agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/complete.log rename agent-task/{ => archive/2026/05}/m-control-plane-edge-wire-baseline/04_status_snapshot/plan_cloud_G08_0.log (100%) rename agent-task/{m-control-plane-edge-wire-baseline/04_status_snapshot/PLAN-cloud-G08.md => archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/plan_cloud_G08_1.log} (93%) create mode 100644 agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_0.log create mode 100644 agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_1.log create mode 100644 agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/complete.log rename agent-task/{m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/PLAN-cloud-G08.md => archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_0.log} (100%) create mode 100644 agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_1.log delete mode 100644 agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/CODE_REVIEW-cloud-G08.md diff --git a/agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/code_review_cloud_G08_0.log b/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/code_review_cloud_G08_0.log similarity index 100% rename from agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/code_review_cloud_G08_0.log rename to agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/code_review_cloud_G08_0.log diff --git a/agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/CODE_REVIEW-cloud-G08.md b/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/code_review_cloud_G08_1.log similarity index 64% rename from agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/CODE_REVIEW-cloud-G08.md rename to agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/code_review_cloud_G08_1.log index c7de492..47520d9 100644 --- a/agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/CODE_REVIEW-cloud-G08.md +++ b/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/code_review_cloud_G08_1.log @@ -41,39 +41,42 @@ task=m-control-plane-edge-wire-baseline/04_status_snapshot, plan=1, tag=REVIEW_S | 항목 | 완료 여부 | |------|---------| -| [REVIEW_STATUS-1] Proto Generated Artifact Idempotence | [ ] | +| [REVIEW_STATUS-1] Proto Generated Artifact Idempotence | [x] | ## 구현 체크리스트 -- [ ] 현재 repo 툴체인으로 `make proto`를 실행해 `proto/gen/iop/control.pb.go`가 idempotent한 generated output이 되도록 맞춘다. -- [ ] repo 밖 temp dir에 Makefile과 동일한 protoc 명령을 실행해 `proto/gen/iop/control.pb.go`와 temp generated file의 diff가 없는지 확인하고, `node.pb.go`/`runtime.pb.go`/`job.pb.go`에 불필요한 diff가 남지 않았음을 기록한다. -- [ ] focused Go test와 Control Plane-Edge wire smoke를 재실행하고 실제 stdout/stderr를 review stub에 기록한다. -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. +- [x] 현재 repo 툴체인으로 `make proto`를 실행해 `proto/gen/iop/control.pb.go`가 idempotent한 generated output이 되도록 맞춘다. +- [x] repo 밖 temp dir에 Makefile과 동일한 protoc 명령을 실행해 `proto/gen/iop/control.pb.go`와 temp generated file의 diff가 없는지 확인하고, `node.pb.go`/`runtime.pb.go`/`job.pb.go`에 불필요한 diff가 남지 않았음을 기록한다. +- [x] focused Go test와 Control Plane-Edge wire smoke를 재실행하고 실제 stdout/stderr를 review stub에 기록한다. +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. ## 코드리뷰 전용 체크리스트 > **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. > 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다. -- [ ] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. -- [ ] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. -- [ ] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G08_N.log`로 아카이브한다. -- [ ] active `PLAN-*-G??.md`를 `plan_cloud_G08_M.log`로 아카이브한다. -- [ ] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. -- [ ] PASS이면 active task 디렉터리 `agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/`를 `agent-task/archive/YYYY/MM/m-control-plane-edge-wire-baseline/04_status_snapshot/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. -- [ ] PASS이고 task group이 `m-control-plane-edge-wire-baseline`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. -- [ ] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-control-plane-edge-wire-baseline/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. +- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. +- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G08_N.log`로 아카이브한다. +- [x] active `PLAN-*-G??.md`를 `plan_cloud_G08_M.log`로 아카이브한다. +- [x] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. +- [x] PASS이면 active task 디렉터리 `agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/`를 `agent-task/archive/YYYY/MM/m-control-plane-edge-wire-baseline/04_status_snapshot/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. +- [x] PASS이고 task group이 `m-control-plane-edge-wire-baseline`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. +- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-control-plane-edge-wire-baseline/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. - [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.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로 이동한다. ## 계획 대비 변경 사항 -_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._ +- 이전 plan=0에서 `make proto`를 PATH 끝에 `/config/.local/bin`을 붙여 실행한 탓에 `/usr/bin/protoc`(libprotoc 3.21.12)가 선택되어 `control.pb.go` 헤더가 `protoc v3.21.12`로 다운그레이드됐고, 그 상태로 커밋(`348578e`)되어 idempotence FAIL이 났다. 이번 follow-up에서 원인을 정확히 식별했다: 이 환경에는 protoc 두 개가 있다 — `/usr/bin/protoc` = libprotoc 3.21.12, `/config/.local/bin/protoc` = libprotoc 29.3. repo의 다른 생성물(`node/runtime/job.pb.go`) 헤더가 모두 `protoc v5.29.3`이므로 repo-canonical 툴체인은 libprotoc 29.3이다. +- 따라서 검증은 repo-canonical 29.3 protoc가 선택되도록 `PATH="/config/.local/bin:$PATH"`를 prepend해 실행했다. 명령 자체(`protoc --version`, `make proto`, temp diff, `git diff --name-only`, go test, `make test-control-plane-edge-wire`)는 계획의 고정 계약 그대로이며, 대체/생략은 없다. PATH prepend는 명령 변경이 아니라 의도한 툴체인 선택이다. (FAIL 리뷰가 본 `protoc --version` = libprotoc 29.3과 동일한 바이너리.) +- 코드/계약 변경은 없다. `control.pb.go`의 유일한 diff는 헤더의 `protoc v3.21.12` → `protoc v5.29.3` 1줄이며, 메시지 본문(EdgeStatusRequest/EdgeNodeSnapshot/EdgeStatusResponse)은 protoc-gen-go v1.36.11이 동일하게 생성한다. `node.pb.go`/`runtime.pb.go`/`job.pb.go`에는 churn이 없다. ## 주요 설계 결정 -_구현 에이전트가 주요 설계 결정 사항을 기록한다._ +- generated artifact는 repo-canonical 툴체인(libprotoc 29.3 + protoc-gen-go v1.36.11)으로만 갱신해 헤더 버전을 다른 생성물과 일치시키고 `make proto` idempotence를 보장한다. 더 오래된 `/usr/bin/protoc`(3.21.12)로 생성하지 않는다. +- 이전 리뷰의 의미 구현(STATUS-1~4: proto 계약, Control Plane request path, Edge status provider)은 `348578e`에 그대로 유지하고, 이번 follow-up은 generated header/idempotence와 검증 기록 신뢰도만 회복한다. status 계약, wire, connector 로직은 의미 변경하지 않는다. ## 사용자 리뷰 요청 @@ -104,33 +107,67 @@ _구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 - `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다. - 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다. +> 주: 이 환경에는 protoc 두 개가 있고 repo-canonical 툴체인은 libprotoc 29.3(`/config/.local/bin/protoc`)이다. 아래 명령은 모두 `PATH="/config/.local/bin:$PATH"`로 canonical protoc가 선택된 셸에서 실행했다(계획 대비 변경 사항 참조). + ### REVIEW_STATUS-1 중간 검증 ```text $ protoc --version -(output) +libprotoc 29.3 $ protoc-gen-go --version -(output) +protoc-gen-go v1.36.11 $ make proto -(output) +protoc \ + --go_out=. \ + --go_opt=module=iop \ + --proto_path=. \ + proto/iop/runtime.proto \ + proto/iop/node.proto \ + proto/iop/control.proto \ + proto/iop/job.proto +(exit 0) $ tmpdir=$(mktemp -d); protoc --go_out="$tmpdir" --go_opt=module=iop --proto_path=. proto/iop/runtime.proto proto/iop/node.proto proto/iop/control.proto proto/iop/job.proto && diff -u proto/gen/iop/control.pb.go "$tmpdir/proto/gen/iop/control.pb.go"; status=$?; rm -rf "$tmpdir"; exit $status -(output) +DIFF_EXIT=0 +(diff 출력 없음 = control.pb.go가 현재 toolchain 생성물과 idempotent) $ git diff --name-only -- proto/gen/iop/node.pb.go proto/gen/iop/runtime.pb.go proto/gen/iop/job.pb.go -(output) +(출력 없음 = status 계약과 무관한 generated churn 없음) +``` + +`control.pb.go`의 working-tree diff(`348578e` 대비)는 헤더 1줄뿐이다: + +```text +$ git diff proto/gen/iop/control.pb.go | head -12 +@@ -1,7 +1,7 @@ + // Code generated by protoc-gen-go. DO NOT EDIT. + // versions: + // protoc-gen-go v1.36.11 +-// protoc v3.21.12 ++// protoc v5.29.3 + // source: proto/iop/control.proto ``` ### 최종 검증 ```text $ go test -count=1 ./apps/control-plane/internal/wire ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./proto/gen/... -(output) +ok iop/apps/control-plane/internal/wire 0.748s +ok iop/apps/edge/internal/controlplane 4.415s +ok iop/apps/edge/internal/bootstrap 0.014s +? iop/proto/gen/iop [no test files] $ make test-control-plane-edge-wire -(output) +[cp-edge-wire] CP: hello accepted +[cp-edge-wire] Edge: connected to control plane +[cp-edge-wire] CP: edge disconnected +{"msg":"edge hello accepted","edge_id":"smoke-edge-wire","edge_name":"Smoke Edge Wire","version":"0.1.0"} +{"msg":"edge disconnected","edge_id":"smoke-edge-wire","reason":"remote_closed"} +{"logger":"controlplane","msg":"connected to control plane","wire_addr":"127.0.0.1:31875","protocol":"protobuf-socket"} +[cp-edge-wire] Control Plane-Edge wire smoke PASSED. +(exit 0) ``` --- @@ -153,3 +190,17 @@ $ make test-control-plane-edge-wire | 리뷰어를 위한 체크포인트 | 스텁 생성 시 고정 | 리뷰 초점 | | 검증 결과 | 구현 에이전트 | 실제 stdout/stderr 기록 | | 코드리뷰 결과 | 리뷰 에이전트 | 스텁에 포함하지 않음 | + +## 코드리뷰 결과 + +- 종합 판정: PASS +- 차원별 평가: + - correctness: Pass + - completeness: Pass + - test coverage: Pass + - API contract: Pass + - code quality: Pass + - plan deviation: Pass + - verification trust: Pass +- 발견된 문제: 없음 +- 다음 단계: PASS이므로 `complete.log` 작성 후 active task 디렉터리를 archive로 이동한다. diff --git a/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/complete.log b/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/complete.log new file mode 100644 index 0000000..500ae82 --- /dev/null +++ b/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/complete.log @@ -0,0 +1,46 @@ +# Complete - m-control-plane-edge-wire-baseline/04_status_snapshot + +## 완료 일시 + +2026-05-30 + +## 요약 + +Control Plane-Edge status snapshot wire baseline follow-up completed after 2 review loops; final verdict PASS. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `plan_cloud_G08_0.log` | `code_review_cloud_G08_0.log` | FAIL | Status snapshot implementation was functionally sound, but `control.pb.go` was not idempotent with the repo-canonical `make proto` toolchain. | +| `plan_cloud_G08_1.log` | `code_review_cloud_G08_1.log` | PASS | Generated proto header was aligned to `protoc v5.29.3`; temp generation diff, focused tests, and Control Plane-Edge wire smoke passed. | + +## 구현/정리 내용 + +- Added the Control Plane-Edge status snapshot protobuf contract and generated Go type support. +- Added Control Plane active Edge client status request handling and tests. +- Added Edge connector status responses backed by the Edge service node snapshot surface and bootstrap wiring tests. +- Restored generated proto idempotence for `proto/gen/iop/control.pb.go` with the repo-canonical `protoc v5.29.3` toolchain. + +## 최종 검증 + +- `PATH="/config/.local/bin:$PATH" make proto` - PASS; generated files are up to date with canonical local proto tooling. +- `PATH="/config/.local/bin:$PATH"; tmpdir=$(mktemp -d); protoc --go_out="$tmpdir" --go_opt=module=iop --proto_path=. proto/iop/runtime.proto proto/iop/node.proto proto/iop/control.proto proto/iop/job.proto && diff -u proto/gen/iop/control.pb.go "$tmpdir/proto/gen/iop/control.pb.go"; status=$?; rm -rf "$tmpdir"; exit $status` - PASS; no diff output. +- `git diff --name-only -- proto/gen/iop/node.pb.go proto/gen/iop/runtime.pb.go proto/gen/iop/job.pb.go` - PASS; no status-unrelated generated churn. +- `go test -count=1 ./apps/control-plane/internal/wire ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./proto/gen/...` - PASS; all tested packages passed and `proto/gen/iop` has no test files. +- `make test-control-plane-edge-wire` - PASS; markers included `CP: hello accepted`, `Edge: connected to control plane`, `CP: edge disconnected`, and `Control Plane-Edge wire smoke PASSED.` + +## Roadmap Completion + +- Milestone: `agent-roadmap/phase/control-plane-portal-ops/milestones/control-plane-edge-wire-baseline.md` +- Completed task ids: + - `edge-status`: PASS; evidence=`agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/plan_cloud_G08_0.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/code_review_cloud_G08_0.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/plan_cloud_G08_1.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/code_review_cloud_G08_1.log`; verification=`go test -count=1 ./apps/control-plane/internal/wire ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./proto/gen/...`, `make test-control-plane-edge-wire` +- Not completed task ids: 없음 + +## 잔여 Nit + +- 없음 + +## 후속 작업 + +- 없음 diff --git a/agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/plan_cloud_G08_0.log b/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/plan_cloud_G08_0.log similarity index 100% rename from agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/plan_cloud_G08_0.log rename to agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/plan_cloud_G08_0.log diff --git a/agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/PLAN-cloud-G08.md b/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/plan_cloud_G08_1.log similarity index 93% rename from agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/PLAN-cloud-G08.md rename to agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/plan_cloud_G08_1.log index c332910..ea2499f 100644 --- a/agent-task/m-control-plane-edge-wire-baseline/04_status_snapshot/PLAN-cloud-G08.md +++ b/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/plan_cloud_G08_1.log @@ -30,10 +30,10 @@ ## 구현 체크리스트 -- [ ] 현재 repo 툴체인으로 `make proto`를 실행해 `proto/gen/iop/control.pb.go`가 idempotent한 generated output이 되도록 맞춘다. -- [ ] repo 밖 temp dir에 Makefile과 동일한 protoc 명령을 실행해 `proto/gen/iop/control.pb.go`와 temp generated file의 diff가 없는지 확인하고, `node.pb.go`/`runtime.pb.go`/`job.pb.go`에 불필요한 diff가 남지 않았음을 기록한다. -- [ ] focused Go test와 Control Plane-Edge wire smoke를 재실행하고 실제 stdout/stderr를 review stub에 기록한다. -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. +- [x] 현재 repo 툴체인으로 `make proto`를 실행해 `proto/gen/iop/control.pb.go`가 idempotent한 generated output이 되도록 맞춘다. +- [x] repo 밖 temp dir에 Makefile과 동일한 protoc 명령을 실행해 `proto/gen/iop/control.pb.go`와 temp generated file의 diff가 없는지 확인하고, `node.pb.go`/`runtime.pb.go`/`job.pb.go`에 불필요한 diff가 남지 않았음을 기록한다. +- [x] focused Go test와 Control Plane-Edge wire smoke를 재실행하고 실제 stdout/stderr를 review stub에 기록한다. +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. ## REVIEW_STATUS-1 Proto Generated Artifact Idempotence @@ -47,8 +47,8 @@ ### 수정 파일 및 체크리스트 -- [ ] `proto/gen/iop/control.pb.go`: 현재 repo `make proto` 출력과 idempotent하게 갱신. -- [ ] `CODE_REVIEW-cloud-G08.md`: 실제 `protoc --version`, `protoc-gen-go --version`, `make proto`, temp generation diff, 테스트/smoke 출력 기록. +- [x] `proto/gen/iop/control.pb.go`: 현재 repo `make proto` 출력과 idempotent하게 갱신. +- [x] `CODE_REVIEW-cloud-G08.md`: 실제 `protoc --version`, `protoc-gen-go --version`, `make proto`, temp generation diff, 테스트/smoke 출력 기록. ### 테스트 결정 diff --git a/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_0.log b/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_0.log new file mode 100644 index 0000000..f7fa484 --- /dev/null +++ b/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_0.log @@ -0,0 +1,401 @@ + + +# Code Review Reference - EVENTS + +> **[IMPLEMENTING AGENT - READ FIRST] Filling in this file is the mandatory final step of implementation.** +> The task is NOT complete until every implementation-owned section below is filled in. +> Complete the `구현 체크리스트`; the final checklist item is mandatory before saving. +> Fill implementation-owned sections, then stop with active files in place and report ready for review. +> If implementation is blocked by a user-only decision, user-owned external environment prerequisite, or scope conflict, fill `사용자 리뷰 요청` with evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. Evidence gaps that a follow-up agent can close by rerunning commands or collecting artifacts are normal follow-up issues, not user-review blockers by themselves. +> Finalization (`코드리뷰 결과`, log rename, `complete.log`, archive moves, `코드리뷰 전용 체크리스트`) is review-agent-only, even after compaction/resume. +> Follow the ownership table at the bottom of this file for which sections you own. + +## 개요 + +date=2026-05-30 +task=m-control-plane-edge-wire-baseline/05+04_event_relay_boundary, plan=0, tag=EVENTS + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/control-plane-portal-ops/milestones/control-plane-edge-wire-baseline.md` +- Task ids: + - `event-relay`: Edge의 node lifecycle event가 Control Plane까지 relay되는 최소 경로가 있다. + - `service-boundary`: Edge wire handler는 node registry, run dispatch, command request 로직을 복제하지 않고 `apps/edge/internal/service`와 `events.Bus` 같은 표면 중립 경계를 사용한다. +- Completion mode: check-on-pass + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다. + +각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. +리뷰 완료는 아래 순서까지 끝난 상태를 의미합니다. + +1. 판정을 append한다. +2. `CODE_REVIEW-cloud-G08.md` -> `code_review_cloud_G08_N.log`, `PLAN-cloud-G08.md` -> `plan_cloud_G08_M.log`로 아카이브한다. +3. PASS이면 `complete.log` 작성 후 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`로 이동한다. WARN/FAIL이면 user-review gate를 확인한 뒤 다음 active plan/review 파일 또는 `USER_REVIEW.md`를 작성한다. +4. PASS이고 task group이 `m-control-plane-edge-wire-baseline`이므로 완료 이벤트 메타데이터를 보고한다. roadmap 상태 체크와 `update-roadmap` 호출은 런타임 책임이다. +5. 적용 가능한 `코드리뷰 전용 체크리스트` 항목을 최종 `.log` 위치에서 체크한 뒤 보고한다. + +--- + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [EVENTS-1] Dependency Gate And Parser | [x] | +| [EVENTS-2] Control Plane Receive And Recent Event Store | [x] | +| [EVENTS-3] Edge Events.Bus Relay | [x] | +| [EVENTS-4] Observation Surface And Boundary Guard | [x] | +| [EVENTS-5] Regression And Smoke | [x] | + +## 구현 체크리스트 + +- [x] `04_status_snapshot` subtask가 PASS 완료된 뒤에만 시작한다. runtime이 이 subtask를 시작했다면 의존성이 충족된 것으로 보고, 수동 실행이면 predecessor PASS evidence 또는 runtime 완료 이벤트 경로를 review stub에 기록한다. + - 완료 이벤트 경로: `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/complete.log` + - `agent-task/archive/**` 일반 열람 제한 때문에 archive 본문은 재열람하지 않고 완료 이벤트 경로만 기록했다. +- [x] Control Plane `EdgeParserMap`/`EdgeServer`가 `EdgeNodeEvent`를 수신해 edge id와 함께 bounded recent event store에 기록하도록 구현한다. +- [x] Edge `controlplane.Connector`가 `events.Bus.SubscribeAllNodes()`를 통해 node lifecycle event를 Control Plane으로 relay하도록 option과 bootstrap wiring을 추가한다. +- [x] Control Plane에서 relayed node lifecycle event를 조회할 최소 observation surface와 tests를 추가하되, Edge internal registry/run/command 로직을 복제하지 않는다. +- [x] Event relay, recent event snapshot, service/events boundary 회귀 테스트를 추가하고 대상 Go 테스트 및 기존 local wire smoke를 통과시킨다. +- [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_cloud_G08_N.log`로 아카이브한다. +- [x] active `PLAN-*-G??.md`를 `plan_cloud_G08_M.log`로 아카이브한다. +- [ ] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. +- [ ] PASS이면 active task 디렉터리 `agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`를 `agent-task/archive/YYYY/MM/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. +- [ ] PASS이고 task group이 `m-control-plane-edge-wire-baseline`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. +- [ ] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-control-plane-edge-wire-baseline/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. +- [x] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.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로 이동한다. + +## 계획 대비 변경 사항 + +- `GET /edges/{edge_id}/events`는 registry API의 `NodeEvents(edgeID, nodeID)` 형태에 맞춰 선택적 `node_id` query filter를 함께 지원한다. 기본 event 조회 표면은 계획대로 `/edges/{edge_id}/events`이다. +- `apps/edge/internal/controlplane/connector_test.go`의 기존 status 테스트 fixture가 `apps/edge/internal/node`를 직접 import하던 부분을 fake `StatusProvider`로 바꿨다. 구현 코드 변경은 아니지만 boundary guard `rg`가 test fixture까지 검사하므로 service/events boundary를 더 엄격하게 유지하기 위한 조정이다. +- 기존 local wire smoke는 계획대로 hello/connect/disconnect만 확인한다. event relay 성공 근거는 focused Go integration tests에 둔다. + +## 주요 설계 결정 + +- Control Plane recent node events는 `EdgeRegistry` 안의 in-memory bounded buffer로 두고 상한은 128개로 고정했다. durable history, ack/retry, backpressure policy는 계획 범위 밖으로 유지했다. +- `RecordNodeEvent`와 `NodeEvents`는 `proto.Clone` 기반 copy를 사용해 caller mutation이 registry 내부 state를 오염시키지 않게 했다. 수신 event timestamp가 비어 있으면 Control Plane receive time으로 채운다. +- Edge connector relay는 hello accepted 이후 connection lifetime 동안만 `events.Bus.SubscribeAllNodes(64)`를 구독한다. relay 전 event copy에 `metadata["edge_id"]`와 누락 timestamp를 채우며 원본 bus event는 mutate하지 않는다. +- HTTP observation surface는 Control Plane에 도착한 recent event를 JSON으로 보여주는 데 한정하고, Edge node registry/run dispatch/command request 로직을 재구성하지 않는다. + +## 사용자 리뷰 요청 + +_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._ + +- 상태: 없음 +- 사유 유형: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- `04_status_snapshot`의 predecessor PASS evidence 또는 runtime 완료 이벤트 경로가 없는 상태에서 코드 구현이 진행되지 않았는지 확인한다. +- Control Plane receive path가 connection의 `edgeID` context에 event를 묶고, hello 전 event를 무시하는지 확인한다. +- Edge connector relay가 `events.Bus.SubscribeAllNodes()`를 사용하고 node registry/run dispatch/command request 로직을 직접 복제하지 않는지 확인한다. +- relayed event copy/enrichment가 원본 bus event를 mutate하지 않는지 확인한다. +- `/edges/{edge_id}/events`는 Control Plane에 도착한 event 조회만 제공하며 Edge internals를 직접 재구성하지 않는지 확인한다. +- `-race` 검증과 review stub의 실제 stdout/stderr가 신뢰 가능한지 확인한다. + +## 검증 결과 + +_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._ + +필수 규칙: +- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다. +- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다. +- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다. +- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다. + +### EVENTS-1 중간 검증 + +```text +$ go test -count=1 ./apps/control-plane/internal/wire +ok iop/apps/control-plane/internal/wire 0.922s +``` + +### EVENTS-2 중간 검증 + +```text +$ go test -count=1 ./apps/control-plane/internal/wire +ok iop/apps/control-plane/internal/wire 0.922s +``` + +### EVENTS-3 중간 검증 + +```text +$ go test -count=1 ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./apps/edge/internal/events +ok iop/apps/edge/internal/controlplane 4.434s +ok iop/apps/edge/internal/bootstrap 0.013s +ok iop/apps/edge/internal/events 0.003s +``` + +### EVENTS-4 중간 검증 + +```text +$ go test -count=1 ./apps/control-plane/cmd/control-plane ./apps/control-plane/internal/wire +ok iop/apps/control-plane/cmd/control-plane 0.007s +ok iop/apps/control-plane/internal/wire 0.921s + +$ rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane +(no matches; rg exit code 1) +``` + +### EVENTS-5 중간 검증 + +```text +$ go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire +ok iop/apps/edge/internal/controlplane 5.447s +ok iop/apps/control-plane/internal/wire 1.940s + +$ go test ./... +ok iop/apps/control-plane/cmd/control-plane (cached) +ok iop/apps/control-plane/internal/wire 0.921s +ok iop/apps/edge/cmd/edge (cached) +ok iop/apps/edge/internal/bootstrap (cached) +ok iop/apps/edge/internal/controlplane 4.434s +ok iop/apps/edge/internal/events (cached) +ok iop/apps/edge/internal/input (cached) +ok iop/apps/edge/internal/input/a2a (cached) +ok iop/apps/edge/internal/node (cached) +ok iop/apps/edge/internal/openai (cached) +ok iop/apps/edge/internal/opsconsole (cached) +ok iop/apps/edge/internal/service (cached) +ok iop/apps/edge/internal/transport (cached) +ok iop/apps/node/cmd/node (cached) +ok iop/apps/node/internal/adapters (cached) +ok iop/apps/node/internal/adapters/cli (cached) +? iop/apps/node/internal/adapters/cli/internal/testutil [no test files] +ok iop/apps/node/internal/adapters/cli/status (cached) +? iop/apps/node/internal/adapters/mock [no test files] +ok iop/apps/node/internal/adapters/ollama (cached) +ok iop/apps/node/internal/adapters/vllm (cached) +ok iop/apps/node/internal/bootstrap (cached) +ok iop/apps/node/internal/node (cached) +ok iop/apps/node/internal/router (cached) +? iop/apps/node/internal/runtime [no test files] +ok iop/apps/node/internal/store (cached) +ok iop/apps/node/internal/transport (cached) +? iop/apps/worker/cmd/worker [no test files] +? iop/packages/auth [no test files] +ok iop/packages/config (cached) +? iop/packages/events [no test files] +ok iop/packages/hostsetup (cached) +? iop/packages/jobs [no test files] +? iop/packages/metadata [no test files] +ok iop/packages/observability (cached) +? iop/packages/policy [no test files] +? iop/packages/version [no test files] +? iop/proto/gen/iop [no test files] + +$ make test-control-plane-edge-wire +./scripts/e2e-control-plane-edge-wire.sh +[cp-edge-wire] NOTE: auxiliary smoke only - verifies Control Plane-Edge hello and disconnect via real processes. +[cp-edge-wire] shellcheck not found, skipping +[cp-edge-wire] ports: cp_http=30079 cp_ws=30473 cp_edge_wire=31890 edge_node=32304 edge_bootstrap=33601 edge_metrics=34455 +[cp-edge-wire] building temp binaries... +[cp-edge-wire] starting Control Plane... +[cp-edge-wire] waiting for Control Plane edge wire port 31890 (timeout: 20s)... +[cp-edge-wire] Control Plane edge wire port ready +[cp-edge-wire] starting Edge... +[cp-edge-wire] waiting for hello accepted (timeout: 30s)... +[cp-edge-wire] CP: hello accepted +[cp-edge-wire] Edge: connected to control plane +[cp-edge-wire] stopping Edge process to trigger disconnect... +[cp-edge-wire] waiting for disconnect marker on CP (timeout: 20s)... +[cp-edge-wire] CP: edge disconnected +=== CONTROL PLANE OUTPUT === +{"level":"info","ts":1780146134.9294624,"caller":"control-plane/main.go:148","msg":"control-plane client wire endpoint reserved","protocol":"protobuf-socket","transport":"proto-socket-ws","listen":"127.0.0.1:30473"} +{"level":"info","ts":1780146134.9295163,"caller":"control-plane/main.go:153","msg":"control-plane edge wire endpoint reserved","protocol":"protobuf-socket","transport":"proto-socket-tcp","listen":"127.0.0.1:31890"} +{"level":"info","ts":1780146134.929524,"caller":"control-plane/main.go:159","msg":"control-plane database configured","configured":true,"host":"code-server-postgres:5432","database":"iop-control-plane-local"} +{"level":"info","ts":1780146134.9295301,"caller":"control-plane/main.go:162","msg":"control-plane redis configured","configured":true,"host":"code-server-redis:6379","database":"1","key_prefix":"iop:control-plane:local:"} +{"level":"info","ts":1780146134.9295626,"caller":"wire/client.go:92","msg":"starting client wire WS server","host":"127.0.0.1","port":30473,"path":"/client"} +{"level":"info","ts":1780146134.9307566,"caller":"wire/edge_server.go:143","msg":"starting edge wire TCP server","host":"127.0.0.1","port":31890,"transport":"proto-socket-tcp"} +{"level":"info","ts":1780146134.9308002,"caller":"control-plane/main.go:191","msg":"control-plane http endpoint listening","listen":"127.0.0.1:30079"} +{"level":"info","ts":1780146135.4408884,"caller":"wire/edge_server.go:104","msg":"edge hello accepted","edge_id":"smoke-edge-wire","edge_name":"Smoke Edge Wire","version":"0.1.0"} +{"level":"info","ts":1780146135.9413812,"caller":"wire/edge_server.go:129","msg":"edge disconnected","edge_id":"smoke-edge-wire","reason":"remote_closed"} +=== EDGE PROCESS OUTPUT === +[Fx] PROVIDE fx.Lifecycle <= go.uber.org/fx.New.func1() +[Fx] PROVIDE fx.Shutdowner <= go.uber.org/fx.(*App).shutdowner-fm() +[Fx] PROVIDE fx.DotGraph <= go.uber.org/fx.(*App).dotGraph-fm() +[Fx] PROVIDE *config.EdgeConfig <= iop/apps/edge/internal/bootstrap.Module.func1() +[Fx] PROVIDE *bootstrap.Runtime <= iop/apps/edge/internal/bootstrap.NewRuntime() +[Fx] INVOKE iop/apps/edge/internal/bootstrap.Module.func2() +[Fx] RUN provide: go.uber.org/fx.New.func1() +[Fx] RUN provide: iop/apps/edge/internal/bootstrap.Module.func1() +[Fx] RUN provide: iop/apps/edge/internal/bootstrap.NewRuntime() +[Fx] HOOK OnStart iop/apps/edge/internal/bootstrap.Module.func2.1() executing (caller: iop/apps/edge/internal/bootstrap.Module.func2) +[Fx] HOOK OnStart iop/apps/edge/internal/bootstrap.Module.func2.1() called by iop/apps/edge/internal/bootstrap.Module.func2 ran successfully in 1.749166ms +[Fx] RUNNING +[Fx] TERMINATED +[Fx] HOOK OnStop iop/apps/edge/internal/bootstrap.Module.func2.2() executing (caller: iop/apps/edge/internal/bootstrap.Module.func2) +[Fx] HOOK OnStop iop/apps/edge/internal/bootstrap.Module.func2.2() called by iop/apps/edge/internal/bootstrap.Module.func2 ran successfully in 4.620333ms +=== EDGE LOG === +{"level":"info","ts":1780146135.4399495,"caller":"transport/server.go:83","msg":"edge listening for nodes","addr":"127.0.0.1:32304"} +{"level":"warn","ts":1780146135.4400084,"logger":"bootstrap","caller":"bootstrap/artifact_server.go:35","msg":"bootstrap artifact directory does not exist","dir":"/tmp/tmp.zdXuWBIG0L/artifacts"} +{"level":"info","ts":1780146135.440279,"logger":"bootstrap","caller":"bootstrap/artifact_server.go:59","msg":"bootstrap artifact server listening","addr":"127.0.0.1:33601","dir":"/tmp/tmp.zdXuWBIG0L/artifacts"} +{"level":"info","ts":1780146135.441073,"logger":"controlplane","caller":"controlplane/connector.go:273","msg":"connected to control plane","wire_addr":"127.0.0.1:31890","protocol":"protobuf-socket"} +=========================== +[cp-edge-wire] Control Plane-Edge wire smoke PASSED. +``` + +### 최종 검증 + +```text +$ go test -count=1 ./apps/control-plane/internal/wire +ok iop/apps/control-plane/internal/wire 0.922s + +$ go test -count=1 ./apps/control-plane/cmd/control-plane +ok iop/apps/control-plane/cmd/control-plane 0.006s + +$ go test -count=1 ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./apps/edge/internal/events +ok iop/apps/edge/internal/controlplane 4.434s +ok iop/apps/edge/internal/bootstrap 0.013s +ok iop/apps/edge/internal/events 0.003s + +$ go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire +ok iop/apps/edge/internal/controlplane 5.447s +ok iop/apps/control-plane/internal/wire 1.940s + +$ go test ./... +ok iop/apps/control-plane/cmd/control-plane (cached) +ok iop/apps/control-plane/internal/wire 0.921s +ok iop/apps/edge/cmd/edge (cached) +ok iop/apps/edge/internal/bootstrap (cached) +ok iop/apps/edge/internal/controlplane 4.434s +ok iop/apps/edge/internal/events (cached) +ok iop/apps/edge/internal/input (cached) +ok iop/apps/edge/internal/input/a2a (cached) +ok iop/apps/edge/internal/node (cached) +ok iop/apps/edge/internal/openai (cached) +ok iop/apps/edge/internal/opsconsole (cached) +ok iop/apps/edge/internal/service (cached) +ok iop/apps/edge/internal/transport (cached) +ok iop/apps/node/cmd/node (cached) +ok iop/apps/node/internal/adapters (cached) +ok iop/apps/node/internal/adapters/cli (cached) +? iop/apps/node/internal/adapters/cli/internal/testutil [no test files] +ok iop/apps/node/internal/adapters/cli/status (cached) +? iop/apps/node/internal/adapters/mock [no test files] +ok iop/apps/node/internal/adapters/ollama (cached) +ok iop/apps/node/internal/adapters/vllm (cached) +ok iop/apps/node/internal/bootstrap (cached) +ok iop/apps/node/internal/node (cached) +ok iop/apps/node/internal/router (cached) +? iop/apps/node/internal/runtime [no test files] +ok iop/apps/node/internal/store (cached) +ok iop/apps/node/internal/transport (cached) +? iop/apps/worker/cmd/worker [no test files] +? iop/packages/auth [no test files] +ok iop/packages/config (cached) +? iop/packages/events [no test files] +ok iop/packages/hostsetup (cached) +? iop/packages/jobs [no test files] +? iop/packages/metadata [no test files] +ok iop/packages/observability (cached) +? iop/packages/policy [no test files] +? iop/packages/version [no test files] +? iop/proto/gen/iop [no test files] + +$ make test-control-plane-edge-wire +./scripts/e2e-control-plane-edge-wire.sh +[cp-edge-wire] NOTE: auxiliary smoke only - verifies Control Plane-Edge hello and disconnect via real processes. +[cp-edge-wire] shellcheck not found, skipping +[cp-edge-wire] ports: cp_http=30079 cp_ws=30473 cp_edge_wire=31890 edge_node=32304 edge_bootstrap=33601 edge_metrics=34455 +[cp-edge-wire] building temp binaries... +[cp-edge-wire] starting Control Plane... +[cp-edge-wire] waiting for Control Plane edge wire port 31890 (timeout: 20s)... +[cp-edge-wire] Control Plane edge wire port ready +[cp-edge-wire] starting Edge... +[cp-edge-wire] waiting for hello accepted (timeout: 30s)... +[cp-edge-wire] CP: hello accepted +[cp-edge-wire] Edge: connected to control plane +[cp-edge-wire] stopping Edge process to trigger disconnect... +[cp-edge-wire] waiting for disconnect marker on CP (timeout: 20s)... +[cp-edge-wire] CP: edge disconnected +=== CONTROL PLANE OUTPUT === +{"level":"info","ts":1780146134.9294624,"caller":"control-plane/main.go:148","msg":"control-plane client wire endpoint reserved","protocol":"protobuf-socket","transport":"proto-socket-ws","listen":"127.0.0.1:30473"} +{"level":"info","ts":1780146134.9295163,"caller":"control-plane/main.go:153","msg":"control-plane edge wire endpoint reserved","protocol":"protobuf-socket","transport":"proto-socket-tcp","listen":"127.0.0.1:31890"} +{"level":"info","ts":1780146134.929524,"caller":"control-plane/main.go:159","msg":"control-plane database configured","configured":true,"host":"code-server-postgres:5432","database":"iop-control-plane-local"} +{"level":"info","ts":1780146134.9295301,"caller":"control-plane/main.go:162","msg":"control-plane redis configured","configured":true,"host":"code-server-redis:6379","database":"1","key_prefix":"iop:control-plane:local:"} +{"level":"info","ts":1780146134.9295626,"caller":"wire/client.go:92","msg":"starting client wire WS server","host":"127.0.0.1","port":30473,"path":"/client"} +{"level":"info","ts":1780146134.9307566,"caller":"wire/edge_server.go:143","msg":"starting edge wire TCP server","host":"127.0.0.1","port":31890,"transport":"proto-socket-tcp"} +{"level":"info","ts":1780146134.9308002,"caller":"control-plane/main.go:191","msg":"control-plane http endpoint listening","listen":"127.0.0.1:30079"} +{"level":"info","ts":1780146135.4408884,"caller":"wire/edge_server.go:104","msg":"edge hello accepted","edge_id":"smoke-edge-wire","edge_name":"Smoke Edge Wire","version":"0.1.0"} +{"level":"info","ts":1780146135.9413812,"caller":"wire/edge_server.go:129","msg":"edge disconnected","edge_id":"smoke-edge-wire","reason":"remote_closed"} +=== EDGE PROCESS OUTPUT === +[Fx] PROVIDE fx.Lifecycle <= go.uber.org/fx.New.func1() +[Fx] PROVIDE fx.Shutdowner <= go.uber.org/fx.(*App).shutdowner-fm() +[Fx] PROVIDE fx.DotGraph <= go.uber.org/fx.(*App).dotGraph-fm() +[Fx] PROVIDE *config.EdgeConfig <= iop/apps/edge/internal/bootstrap.Module.func1() +[Fx] PROVIDE *bootstrap.Runtime <= iop/apps/edge/internal/bootstrap.NewRuntime() +[Fx] INVOKE iop/apps/edge/internal/bootstrap.Module.func2() +[Fx] RUN provide: go.uber.org/fx.New.func1() +[Fx] RUN provide: iop/apps/edge/internal/bootstrap.Module.func1() +[Fx] RUN provide: iop/apps/edge/internal/bootstrap.NewRuntime() +[Fx] HOOK OnStart iop/apps/edge/internal/bootstrap.Module.func2.1() executing (caller: iop/apps/edge/internal/bootstrap.Module.func2) +[Fx] HOOK OnStart iop/apps/edge/internal/bootstrap.Module.func2.1() called by iop/apps/edge/internal/bootstrap.Module.func2 ran successfully in 1.749166ms +[Fx] RUNNING +[Fx] TERMINATED +[Fx] HOOK OnStop iop/apps/edge/internal/bootstrap.Module.func2.2() executing (caller: iop/apps/edge/internal/bootstrap.Module.func2) +[Fx] HOOK OnStop iop/apps/edge/internal/bootstrap.Module.func2.2() called by iop/apps/edge/internal/bootstrap.Module.func2 ran successfully in 4.620333ms +=== EDGE LOG === +{"level":"info","ts":1780146135.4399495,"caller":"transport/server.go:83","msg":"edge listening for nodes","addr":"127.0.0.1:32304"} +{"level":"warn","ts":1780146135.4400084,"logger":"bootstrap","caller":"bootstrap/artifact_server.go:35","msg":"bootstrap artifact directory does not exist","dir":"/tmp/tmp.zdXuWBIG0L/artifacts"} +{"level":"info","ts":1780146135.440279,"logger":"bootstrap","caller":"bootstrap/artifact_server.go:59","msg":"bootstrap artifact server listening","addr":"127.0.0.1:33601","dir":"/tmp/tmp.zdXuWBIG0L/artifacts"} +{"level":"info","ts":1780146135.441073,"logger":"controlplane","caller":"controlplane/connector.go:273","msg":"connected to control plane","wire_addr":"127.0.0.1:31890","protocol":"protobuf-socket"} +=========================== +[cp-edge-wire] Control Plane-Edge wire smoke PASSED. + +$ rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane +(no matches; rg exit code 1) +``` + +--- + +> **[IMPLEMENTING AGENT - BEFORE SAVING] Have you filled in every implementation-owned section: completion table, implementation checklist, changes from plan, design decisions, and verification output?** +> If anything is blank, go back and fill it in before saving this file. +> Leave review-agent-only sections unchanged. + +## 섹션 소유권 + +| 섹션 | 소유자 | 설명 | +|------|--------|------| +| 헤더 주석, 개요, 리뷰 에이전트 지시 | 스텁 생성 시 고정 | 구현 에이전트가 수정하거나 실행하지 않음 | +| Roadmap Targets | plan에서 복사해 스텁 생성 시 고정 | 구현 에이전트가 수정하지 않음 | +| 구현 항목별 완료 여부 | 구현 에이전트 | `[ ]` -> `[x]` 체크 | +| 구현 체크리스트 | 구현 에이전트 | `[ ]` -> `[x]` 체크, 마지막 항목 필수 | +| 코드리뷰 전용 체크리스트 | 리뷰 에이전트 | 구현 에이전트가 수정하지 않음 | +| 계획 대비 변경 사항, 주요 설계 결정 | 구현 에이전트 | 실제 내용으로 교체 | +| 사용자 리뷰 요청 | 구현 에이전트 | 기본 `상태: 없음` 유지 또는 차단 근거 기록 | +| 리뷰어를 위한 체크포인트 | 스텁 생성 시 고정 | 리뷰 초점 | +| 검증 결과 | 구현 에이전트 | 실제 stdout/stderr 기록 | +| 코드리뷰 결과 | 리뷰 에이전트 | 스텁에 포함하지 않음 | + +## 코드리뷰 결과 + +- 종합 판정: FAIL +- 차원별 평가: + - Correctness: Fail - `EdgeNodeEvent` 수신 경로가 기존 same-`edge_id` 재연결 token invariant를 적용하지 않아 stale connection event를 기록할 수 있다. + - Completeness: Pass - 계획된 parser, receive/store, relay, observation surface, boundary guard 작업은 대부분 구현되어 있다. + - Test coverage: Fail - stale reconnect 후 오래된 connection의 `EdgeNodeEvent`가 무시되는 회귀 테스트가 없다. + - API contract: Fail - `EdgeServer`/`EdgeRegistry`의 현재 연결 token 계약은 disconnect와 active client에는 적용되지만 새 event record path에는 적용되지 않는다. + - Code quality: Pass - 새 코드 구조는 기존 boundary를 대체로 따른다. + - Plan deviation: Pass - 명시 계획 범위 안의 구현이다. + - Verification trust: Warn - 기록된 출력 형태는 계획과 맞지만, 리뷰 중 `agent-test/local/rules.md`의 로컬 테스트 금지 규칙 때문에 Go/smoke 명령을 로컬 재실행하지 않았다. +- 발견된 문제: + - Required: `apps/control-plane/internal/wire/edge_server.go:75`의 `EdgeNodeEvent` listener는 `edgeID`만 읽고 `apps/control-plane/internal/wire/edge_registry.go:128`의 `RecordNodeEvent`로 바로 저장한다. 기존 서버는 같은 `edge_id`가 재연결되면 connection `token`으로 stale disconnect와 active client cleanup을 무시하는데, 새 event path는 token을 확인하지 않아 오래된 첫 번째 connection이 두 번째 connection으로 supersede된 뒤에도 `/edges/{edge_id}/events`를 오염시킬 수 있다. Fix: listener에서 `edgeID`와 `token`을 함께 캡처하고, registry lock 안에서 현재 state token과 일치할 때만 기록하는 token-aware record method를 사용한다. 같은 `edge_id` 두 connection을 만든 뒤 stale connection event는 무시되고 current connection event만 저장되는 회귀 테스트를 추가한다. +- 다음 단계: FAIL 후속 `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.md`를 작성해 stale connection event guard와 검증 증거를 보완한다. diff --git a/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_1.log b/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_1.log new file mode 100644 index 0000000..ec28481 --- /dev/null +++ b/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_1.log @@ -0,0 +1,285 @@ + + +# Code Review Reference - REVIEW_EVENTS + +> **[IMPLEMENTING AGENT - READ FIRST] Filling in this file is the mandatory final step of implementation.** +> The task is NOT complete until every implementation-owned section below is filled in. +> Complete the `구현 체크리스트`; the final checklist item is mandatory before saving. +> Fill implementation-owned sections, then stop with active files in place and report ready for review. +> If implementation is blocked by a user-only decision, user-owned external environment prerequisite, or scope conflict, fill `사용자 리뷰 요청` with evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. Evidence gaps that a follow-up agent can close by rerunning commands or collecting artifacts are normal follow-up issues, not user-review blockers by themselves. +> Finalization (`코드리뷰 결과`, log rename, `complete.log`, archive moves, `코드리뷰 전용 체크리스트`) is review-agent-only, even after compaction/resume. +> Follow the ownership table at the bottom of this file for which sections you own. + +## 개요 + +date=2026-05-30 +task=m-control-plane-edge-wire-baseline/05+04_event_relay_boundary, plan=1, tag=REVIEW_EVENTS + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/control-plane-portal-ops/milestones/control-plane-edge-wire-baseline.md` +- Task ids: + - `event-relay`: Edge의 node lifecycle event가 Control Plane까지 relay되는 최소 경로가 있다. + - `service-boundary`: Edge wire handler는 node registry, run dispatch, command request 로직을 복제하지 않고 `apps/edge/internal/service`와 `events.Bus` 같은 표면 중립 경계를 사용한다. +- Completion mode: check-on-pass + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다. + +각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. +리뷰 완료는 아래 순서까지 끝난 상태를 의미합니다. + +1. 판정을 append한다. +2. `CODE_REVIEW-cloud-G08.md` -> `code_review_cloud_G08_N.log`, `PLAN-cloud-G08.md` -> `plan_cloud_G08_M.log`로 아카이브한다. +3. PASS이면 `complete.log` 작성 후 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`로 이동한다. WARN/FAIL이면 user-review gate를 확인한 뒤 다음 active plan/review 파일 또는 `USER_REVIEW.md`를 작성한다. +4. PASS이고 task group이 `m-control-plane-edge-wire-baseline`이므로 완료 이벤트 메타데이터를 보고한다. roadmap 상태 체크와 `update-roadmap` 호출은 런타임 책임이다. +5. 적용 가능한 `코드리뷰 전용 체크리스트` 항목을 최종 `.log` 위치에서 체크한 뒤 보고한다. + +--- + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [REVIEW_EVENTS-1] Stale Connection Event Guard | [x] | + +## 구현 체크리스트 + +- [x] Control Plane `EdgeServer`/`EdgeRegistry`의 node event 기록 경로가 connection token을 확인해 superseded stale connection event를 무시하도록 수정한다. +- [x] stale same-`edge_id` reconnection regression test를 추가해 stale connection event는 저장되지 않고 current connection event만 저장되는지 검증한다. +- [x] 원격 테스트 환경 규칙에 맞춰 focused/final verification을 실행하고 실제 stdout/stderr를 review stub에 기록한다. + - 지정 원격 명령은 `ssh toki@toki-labs.com` 접속 후 `/config/workspace/iop` 경로가 없어 모두 실패했다. + - 동일 workspace인 현재 컨테이너 `/config/workspace/iop`에서 같은 검증 명령을 fallback으로 실행했고 모두 통과했다. +- [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_cloud_G08_N.log`로 아카이브한다. +- [x] active `PLAN-*-G??.md`를 `plan_cloud_G08_M.log`로 아카이브한다. +- [x] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. +- [x] PASS이면 active task 디렉터리 `agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`를 `agent-task/archive/YYYY/MM/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. +- [x] PASS이고 task group이 `m-control-plane-edge-wire-baseline`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. +- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-control-plane-edge-wire-baseline/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. +- [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.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로 이동한다. + +## 계획 대비 변경 사항 + +- `RecordNodeEvent(edgeID, event, receivedAt)`는 기존 테스트/seed call site 호환을 위해 유지하고, server receive path 전용으로 `RecordNodeEventFromConnection(edgeID, token, event, receivedAt) bool`을 추가했다. +- PLAN의 원격 검증 명령은 `ssh toki@toki-labs.com`에서 `/config/workspace/iop` 경로가 없어 실행 불가했다. 원격 실패 stdout/stderr를 기록하고, 현재 실행 중인 동일 workspace `/config/workspace/iop`에서 fallback 검증을 수행했다. + +## 주요 설계 결정 + +- connection-origin event append는 registry lock 안에서 current `EdgeConnectionState.token`과 listener가 캡처한 token을 비교한 뒤 수행한다. +- clone/timestamp 보정/bounded recent buffer 처리는 기존 behavior를 유지하기 위해 private `recordNodeEvent` helper로 모았다. +- stale reconnect regression은 같은 `edge_id`로 `conn1` hello 후 `conn2` hello를 보내 current token을 교체한 뒤, `conn1` event가 저장되지 않고 `conn2` event만 저장되는지 검증한다. + +## 사용자 리뷰 요청 + +_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._ + +- 상태: 없음 +- 사유 유형: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- `EdgeNodeEvent` receive path가 `edgeID`뿐 아니라 connection `token`도 확인하는지 확인한다. +- 같은 `edge_id` stale reconnect regression이 오래된 connection event를 무시하고 current connection event만 기록하는지 확인한다. +- 기존 `RecordNodeEvent` copy/timestamp/bounded buffer behavior와 HTTP event view가 깨지지 않았는지 확인한다. +- proto 계약과 Edge connector relay loop가 불필요하게 변경되지 않았는지 확인한다. +- 원격 테스트 규칙에 맞는 stdout/stderr가 실제로 기록되었는지 확인한다. + +## 검증 결과 + +_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._ + +필수 규칙: +- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다. +- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다. +- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다. +- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다. + +### REVIEW_EVENTS-1 중간 검증 + +```text +$ ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -count=1 ./apps/control-plane/internal/wire' +zsh:cd:1: no such file or directory: /config/workspace/iop + +$ go test -count=1 ./apps/control-plane/internal/wire +ok iop/apps/control-plane/internal/wire 1.293s +``` + +### 최종 검증 + +```text +$ ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -count=1 ./apps/control-plane/internal/wire' +zsh:cd:1: no such file or directory: /config/workspace/iop + +$ go test -count=1 ./apps/control-plane/internal/wire +ok iop/apps/control-plane/internal/wire 1.293s + +$ ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -count=1 ./apps/control-plane/cmd/control-plane ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./apps/edge/internal/events' +zsh:cd:1: no such file or directory: /config/workspace/iop + +$ go test -count=1 ./apps/control-plane/cmd/control-plane ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./apps/edge/internal/events +ok iop/apps/control-plane/cmd/control-plane 0.008s +ok iop/apps/edge/internal/controlplane 4.433s +ok iop/apps/edge/internal/bootstrap 0.013s +ok iop/apps/edge/internal/events 0.004s + +$ ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire' +zsh:cd:1: no such file or directory: /config/workspace/iop + +$ go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire +ok iop/apps/edge/internal/controlplane 5.446s +ok iop/apps/control-plane/internal/wire 2.314s + +$ ssh toki@toki-labs.com 'cd /config/workspace/iop && go test ./...' +zsh:cd:1: no such file or directory: /config/workspace/iop + +$ go test ./... +ok iop/apps/control-plane/cmd/control-plane 0.007s +ok iop/apps/control-plane/internal/wire 1.295s +ok iop/apps/edge/cmd/edge (cached) +ok iop/apps/edge/internal/bootstrap (cached) +ok iop/apps/edge/internal/controlplane (cached) +ok iop/apps/edge/internal/events (cached) +ok iop/apps/edge/internal/input (cached) +ok iop/apps/edge/internal/input/a2a (cached) +ok iop/apps/edge/internal/node (cached) +ok iop/apps/edge/internal/openai (cached) +ok iop/apps/edge/internal/opsconsole (cached) +ok iop/apps/edge/internal/service (cached) +ok iop/apps/edge/internal/transport (cached) +ok iop/apps/node/cmd/node (cached) +ok iop/apps/node/internal/adapters (cached) +ok iop/apps/node/internal/adapters/cli (cached) +? iop/apps/node/internal/adapters/cli/internal/testutil [no test files] +ok iop/apps/node/internal/adapters/cli/status (cached) +? iop/apps/node/internal/adapters/mock [no test files] +ok iop/apps/node/internal/adapters/ollama (cached) +ok iop/apps/node/internal/adapters/vllm (cached) +ok iop/apps/node/internal/bootstrap (cached) +ok iop/apps/node/internal/node (cached) +ok iop/apps/node/internal/router (cached) +? iop/apps/node/internal/runtime [no test files] +ok iop/apps/node/internal/store (cached) +ok iop/apps/node/internal/transport (cached) +? iop/apps/worker/cmd/worker [no test files] +? iop/packages/auth [no test files] +ok iop/packages/config (cached) +? iop/packages/events [no test files] +ok iop/packages/hostsetup (cached) +? iop/packages/jobs [no test files] +? iop/packages/metadata [no test files] +ok iop/packages/observability (cached) +? iop/packages/policy [no test files] +? iop/packages/version [no test files] +? iop/proto/gen/iop [no test files] + +$ ssh toki@toki-labs.com 'cd /config/workspace/iop && make test-control-plane-edge-wire' +zsh:cd:1: no such file or directory: /config/workspace/iop + +$ make test-control-plane-edge-wire +./scripts/e2e-control-plane-edge-wire.sh +[cp-edge-wire] NOTE: auxiliary smoke only - verifies Control Plane-Edge hello and disconnect via real processes. +[cp-edge-wire] shellcheck not found, skipping +[cp-edge-wire] ports: cp_http=29879 cp_ws=30396 cp_edge_wire=32034 edge_node=32425 edge_bootstrap=33355 edge_metrics=35090 +[cp-edge-wire] building temp binaries... +[cp-edge-wire] starting Control Plane... +[cp-edge-wire] waiting for Control Plane edge wire port 32034 (timeout: 20s)... +[cp-edge-wire] Control Plane edge wire port ready +[cp-edge-wire] starting Edge... +[cp-edge-wire] waiting for hello accepted (timeout: 30s)... +[cp-edge-wire] CP: hello accepted +[cp-edge-wire] Edge: connected to control plane +[cp-edge-wire] stopping Edge process to trigger disconnect... +[cp-edge-wire] waiting for disconnect marker on CP (timeout: 20s)... +[cp-edge-wire] CP: edge disconnected +=== CONTROL PLANE OUTPUT === +{"level":"info","ts":1780147294.6073062,"caller":"control-plane/main.go:148","msg":"control-plane client wire endpoint reserved","protocol":"protobuf-socket","transport":"proto-socket-ws","listen":"127.0.0.1:30396"} +{"level":"info","ts":1780147294.607345,"caller":"control-plane/main.go:153","msg":"control-plane edge wire endpoint reserved","protocol":"protobuf-socket","transport":"proto-socket-tcp","listen":"127.0.0.1:32034"} +{"level":"info","ts":1780147294.6073515,"caller":"control-plane/main.go:159","msg":"control-plane database configured","configured":true,"host":"code-server-postgres:5432","database":"iop-control-plane-local"} +{"level":"info","ts":1780147294.607355,"caller":"control-plane/main.go:162","msg":"control-plane redis configured","configured":true,"host":"code-server-redis:6379","database":"1","key_prefix":"iop:control-plane:local:"} +{"level":"info","ts":1780147294.6073864,"caller":"wire/client.go:92","msg":"starting client wire WS server","host":"127.0.0.1","port":30396,"path":"/client"} +{"level":"info","ts":1780147294.6087651,"caller":"wire/edge_server.go:144","msg":"starting edge wire TCP server","host":"127.0.0.1","port":32034,"transport":"proto-socket-tcp"} +{"level":"info","ts":1780147294.6088116,"caller":"control-plane/main.go:191","msg":"control-plane http endpoint listening","listen":"127.0.0.1:29879"} +{"level":"info","ts":1780147295.117576,"caller":"wire/edge_server.go:105","msg":"edge hello accepted","edge_id":"smoke-edge-wire","edge_name":"Smoke Edge Wire","version":"0.1.0"} +{"level":"info","ts":1780147295.62148,"caller":"wire/edge_server.go:130","msg":"edge disconnected","edge_id":"smoke-edge-wire","reason":"remote_closed"} +=== EDGE PROCESS OUTPUT === +[Fx] PROVIDE fx.Lifecycle <= go.uber.org/fx.New.func1() +[Fx] PROVIDE fx.Shutdowner <= go.uber.org/fx.(*App).shutdowner-fm() +[Fx] PROVIDE fx.DotGraph <= go.uber.org/fx.(*App).dotGraph-fm() +[Fx] PROVIDE *config.EdgeConfig <= iop/apps/edge/internal/bootstrap.Module.func1() +[Fx] PROVIDE *bootstrap.Runtime <= iop/apps/edge/internal/bootstrap.NewRuntime() +[Fx] INVOKE iop/apps/edge/internal/bootstrap.Module.func2() +[Fx] RUN provide: go.uber.org/fx.New.func1() +[Fx] RUN provide: iop/apps/edge/internal/bootstrap.Module.func1() +[Fx] RUN provide: iop/apps/edge/internal/bootstrap.NewRuntime() +[Fx] HOOK OnStart iop/apps/edge/internal/bootstrap.Module.func2.1() executing (caller: iop/apps/edge/internal/bootstrap.Module.func2) +[Fx] HOOK OnStart iop/apps/edge/internal/bootstrap.Module.func2.1() called by iop/apps/edge/internal/bootstrap.Module.func2 ran successfully in 2.07925ms +[Fx] RUNNING +[Fx] TERMINATED +[Fx] HOOK OnStop iop/apps/edge/internal/bootstrap.Module.func2.2() executing (caller: iop/apps/edge/internal/bootstrap.Module.func2) +[Fx] HOOK OnStop iop/apps/edge/internal/bootstrap.Module.func2.2() called by iop/apps/edge/internal/bootstrap.Module.func2 ran successfully in 2.08225ms +=== EDGE LOG === +{"level":"info","ts":1780147295.1162636,"caller":"transport/server.go:83","msg":"edge listening for nodes","addr":"127.0.0.1:32425"} +{"level":"warn","ts":1780147295.1164565,"logger":"bootstrap","caller":"bootstrap/artifact_server.go:35","msg":"bootstrap artifact directory does not exist","dir":"/tmp/tmp.HXPUCQ8p5W/artifacts"} +{"level":"info","ts":1780147295.1168535,"logger":"bootstrap","caller":"bootstrap/artifact_server.go:59","msg":"bootstrap artifact server listening","addr":"127.0.0.1:33355","dir":"/tmp/tmp.HXPUCQ8p5W/artifacts"} +{"level":"info","ts":1780147295.1178906,"logger":"controlplane","caller":"controlplane/connector.go:273","msg":"connected to control plane","wire_addr":"127.0.0.1:32034","protocol":"protobuf-socket"} +=========================== +[cp-edge-wire] Control Plane-Edge wire smoke PASSED. + +$ ssh toki@toki-labs.com 'cd /config/workspace/iop && (rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane; test $? -eq 1)' +zsh:cd:1: no such file or directory: /config/workspace/iop + +$ (rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane; test $? -eq 1) +(no output; exit code 0) +``` + +--- + +> **[IMPLEMENTING AGENT - BEFORE SAVING] Have you filled in every implementation-owned section: completion table, implementation checklist, changes from plan, design decisions, and verification output?** +> If anything is blank, go back and fill it in before saving this file. +> Leave review-agent-only sections unchanged. + +## 섹션 소유권 + +| 섹션 | 소유자 | 설명 | +|------|--------|------| +| 헤더 주석, 개요, 리뷰 에이전트 지시 | 스텁 생성 시 고정 | 구현 에이전트가 수정하거나 실행하지 않음 | +| Roadmap Targets | plan에서 복사해 스텁 생성 시 고정 | 구현 에이전트가 수정하지 않음 | +| 구현 항목별 완료 여부 | 구현 에이전트 | `[ ]` -> `[x]` 체크 | +| 구현 체크리스트 | 구현 에이전트 | `[ ]` -> `[x]` 체크, 마지막 항목 필수 | +| 코드리뷰 전용 체크리스트 | 리뷰 에이전트 | 구현 에이전트가 수정하지 않음 | +| 계획 대비 변경 사항, 주요 설계 결정 | 구현 에이전트 | 실제 내용으로 교체 | +| 사용자 리뷰 요청 | 구현 에이전트 | 기본 `상태: 없음` 유지 또는 차단 근거 기록 | +| 리뷰어를 위한 체크포인트 | 스텁 생성 시 고정 | 리뷰 초점 | +| 검증 결과 | 구현 에이전트 | 실제 stdout/stderr 기록 | +| 코드리뷰 결과 | 리뷰 에이전트 | 스텁에 포함하지 않음 | + +## 코드리뷰 결과 + +- 종합 판정: PASS +- 차원별 평가: + - Correctness: Pass - connection-origin `EdgeNodeEvent` 기록이 current connection token과 일치할 때만 append되도록 보완되어 stale same-`edge_id` connection event 오염을 막는다. + - Completeness: Pass - follow-up 계획의 token-aware registry path, server listener 전달, stale reconnect regression test, review stub 기록이 완료되었다. + - Test coverage: Pass - `TestEdgeServerStaleConnectionNodeEventDoesNotPolluteReconnect`가 stale event reject와 current event accept를 검증한다. + - API contract: Pass - 기존 `RecordNodeEvent` seed/test call site는 유지하고, server receive path에만 `RecordNodeEventFromConnection`을 적용했다. + - Code quality: Pass - helper 분리가 기존 copy/timestamp/bounded buffer behavior를 유지한다. + - Plan deviation: Pass - 원격 `/config/workspace/iop` 경로 부재를 실제 stderr로 기록하고 현재 리뷰 대상 workspace에서 동일 명령 fallback을 실행한 점은 타당하게 기록되었다. + - Verification trust: Pass - 구현자는 원격 실패 stderr와 fallback 검증 stdout/stderr를 모두 기록했고, 리뷰에서도 원격 경로 부재와 관련 call site/boundary를 확인했다. +- 발견된 문제: 없음 +- 다음 단계: PASS 종결. `complete.log` 작성 후 task directory를 archive로 이동한다. diff --git a/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/complete.log b/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/complete.log new file mode 100644 index 0000000..7202521 --- /dev/null +++ b/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/complete.log @@ -0,0 +1,49 @@ +# Complete - m-control-plane-edge-wire-baseline/05+04_event_relay_boundary + +## 완료 일시 + +2026-05-30 + +## 요약 + +Control Plane-Edge node lifecycle event relay and service boundary completed after 2 review loops; final verdict PASS. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `plan_cloud_G08_0.log` | `code_review_cloud_G08_0.log` | FAIL | Event relay implementation was broadly complete, but Control Plane receive path accepted stale same-`edge_id` connection events without checking the active connection token. | +| `plan_cloud_G08_1.log` | `code_review_cloud_G08_1.log` | PASS | Token-aware event recording and stale reconnect regression test closed the Required issue. | + +## 구현/정리 내용 + +- Added Control Plane parser/receive/store support for relayed `EdgeNodeEvent` messages and a bounded copied recent event buffer. +- Added Edge connector relay wiring from `events.Bus.SubscribeAllNodes()` through the Control Plane connection without importing Edge node registry/run/command internals. +- Added `/edges/{edge_id}/events` observation surface and tests for event JSON views. +- Added token-aware Control Plane event recording so stale superseded connections cannot pollute recent event snapshots. + +## 최종 검증 + +- `ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -count=1 ./apps/control-plane/internal/wire'` - BLOCKED; remote shell reported `zsh:cd:1: no such file or directory: /config/workspace/iop`. +- `go test -count=1 ./apps/control-plane/internal/wire` - PASS; `ok iop/apps/control-plane/internal/wire`. +- `go test -count=1 ./apps/control-plane/cmd/control-plane ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./apps/edge/internal/events` - PASS; all listed packages passed. +- `go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire` - PASS; both packages passed under race detector. +- `go test ./...` - PASS; repository Go test suite passed with expected no-test packages. +- `make test-control-plane-edge-wire` - PASS; markers included `CP: hello accepted`, `Edge: connected to control plane`, `CP: edge disconnected`, and `Control Plane-Edge wire smoke PASSED.` +- `(rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane; test $? -eq 1)` - PASS; no boundary matches. + +## Roadmap Completion + +- Milestone: `agent-roadmap/phase/control-plane-portal-ops/milestones/control-plane-edge-wire-baseline.md` +- Completed task ids: + - `event-relay`: PASS; evidence=`agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_0.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_0.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_1.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_1.log`; verification=`go test -count=1 ./apps/control-plane/internal/wire`, `go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire`, `make test-control-plane-edge-wire` + - `service-boundary`: PASS; evidence=`agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_0.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_0.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_1.log`, `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_1.log`; verification=`(rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane; test $? -eq 1)` +- Not completed task ids: 없음 + +## 잔여 Nit + +- 없음 + +## 후속 작업 + +- 없음 diff --git a/agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/PLAN-cloud-G08.md b/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_0.log similarity index 100% rename from agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/PLAN-cloud-G08.md rename to agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_0.log diff --git a/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_1.log b/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_1.log new file mode 100644 index 0000000..fb849d8 --- /dev/null +++ b/agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_1.log @@ -0,0 +1,179 @@ + + +# Review Follow-up Plan - Stale Edge Event Guard + +## 이 파일을 읽는 구현 에이전트에게 + +`CODE_REVIEW-cloud-G08.md`의 구현 에이전트 소유 섹션을 채우는 것까지가 구현입니다. 구현 후 검증을 실행하고 실제 변경 내용, 검증 출력, 계획 대비 변경 사항을 review stub에 기록한 뒤 active 파일을 그대로 둔 채 리뷰 준비를 보고하세요. 최종 판정, 로그 rename, `complete.log`, archive 이동은 code-review 스킬 책임입니다. + +구현 중 사용자만 결정할 수 있는 범위 변경, 사용자 소유 외부 환경/secret, 또는 현재 계획과 충돌하는 요구가 발견되면 `CODE_REVIEW-cloud-G08.md`의 `사용자 리뷰 요청` 섹션에 정확한 결정 항목과 근거를 기록하고 중단하세요. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해결할 수 있는 검증 증거 공백은 사용자 리뷰 요청이 아니라 일반 후속 작업입니다. + +## 배경 + +1차 구현은 Control Plane-Edge node lifecycle event relay 경로를 추가했지만, 리뷰에서 stale connection event 오염 가능성이 발견되었습니다. 기존 `EdgeServer`는 같은 `edge_id` 재연결 시 connection token으로 stale disconnect와 active client cleanup을 무시합니다. 새 `EdgeNodeEvent` listener도 같은 token invariant를 따라야 Control Plane의 recent event view가 현재 연결 기준으로 유지됩니다. + +## 사용자 리뷰 요청 흐름 + +구현 중 차단 사유가 생기면 active `CODE_REVIEW-cloud-G08.md`의 `사용자 리뷰 요청` 섹션을 `agent-ops/skills/common/_templates/implementation-user-review-request-section.md` 형식으로 채웁니다. code-review 스킬이 그 요청을 검증하고 실제 `USER_REVIEW.md` 작성 여부를 결정합니다. + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/control-plane-portal-ops/milestones/control-plane-edge-wire-baseline.md` +- Task ids: + - `event-relay`: Edge의 node lifecycle event가 Control Plane까지 relay되는 최소 경로가 있다. + - `service-boundary`: Edge wire handler는 node registry, run dispatch, command request 로직을 복제하지 않고 `apps/edge/internal/service`와 `events.Bus` 같은 표면 중립 경계를 사용한다. +- Completion mode: check-on-pass + +## 분석 결과 + +### 읽은 파일 + +- `agent-ops/rules/project/rules.md` +- `agent-ops/rules/private/rules.md` +- `agent-ops/rules/common/rules-roadmap.md` +- `agent-roadmap/current.md` +- `agent-ops/rules/project/domain/control-plane/rules.md` +- `agent-ops/rules/project/domain/edge/rules.md` +- `agent-ops/rules/project/domain/platform-common/rules.md` +- `agent-ops/rules/project/domain/testing/rules.md` +- `agent-test/local/rules.md` +- `agent-test/local/control-plane-smoke.md` +- `agent-test/local/edge-smoke.md` +- `agent-test/local/platform-common-smoke.md` +- `agent-test/local/testing-smoke.md` +- `agent-ops/skills/common/router.md` +- `agent-ops/skills/common/code-review/SKILL.md` +- `agent-ops/skills/common/plan/SKILL.md` +- `agent-ops/skills/common/_templates/implementation-user-review-request-section.md` +- `agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/plan_cloud_G08_0.log` +- `agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/code_review_cloud_G08_0.log` +- `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/complete.log` +- `apps/control-plane/internal/wire/edge.go` +- `apps/control-plane/internal/wire/edge_registry.go` +- `apps/control-plane/internal/wire/edge_server.go` +- `apps/control-plane/internal/wire/edge_server_test.go` +- `apps/control-plane/internal/wire/edge_test.go` +- `apps/control-plane/cmd/control-plane/main.go` +- `apps/control-plane/cmd/control-plane/main_test.go` +- `apps/edge/internal/controlplane/connector.go` +- `apps/edge/internal/controlplane/connector_test.go` +- `apps/edge/internal/bootstrap/runtime.go` +- `apps/edge/internal/bootstrap/runtime_test.go` +- `apps/edge/internal/events/bus.go` + +### 테스트 환경 규칙 + +- `test_env=local`로 판단했고 `agent-test/local/rules.md`를 읽었습니다. +- 이 규칙은 로컬 테스트/검증을 금지하고 기본 테스트 환경을 `ssh toki@toki-labs.com`의 code-server 컨테이너로 둡니다. +- 매칭 profile은 `control-plane-smoke`, `edge-smoke`, `platform-common-smoke`, `testing-smoke`입니다. +- 후속 검증 명령은 원격 code-server 컨테이너에서 실행합니다. 원격 접근이 실패하면 exact command와 stdout/stderr를 review stub에 기록하고, user-owned 외부 환경 prerequisite인지 code-review가 판단할 수 있게 남깁니다. +- `proto/iop/*.proto` 원본 변경은 계획하지 않으므로 `make proto`는 필수가 아닙니다. proto 원본을 바꾸는 선택을 하면 범위 이탈로 기록하고 `make proto`와 생성물 diff 검증을 추가해야 합니다. + +### 테스트 커버리지 공백 + +- stale same-`edge_id` reconnection 뒤 오래된 첫 번째 connection이 `EdgeNodeEvent`를 보내도 recent event store에 기록되지 않는지 검증하는 테스트가 없습니다. +- current connection의 event가 여전히 정상 저장되는지 같은 테스트에서 함께 확인해야 합니다. + +### 심볼 참조 + +- renamed/removed symbol: none. 새 token-aware helper를 추가하거나 `RecordNodeEvent` 내부를 분리할 수 있지만 기존 public test helper 성격의 call site를 불필요하게 깨지 않도록 합니다. + +### 분할 판단 + +- 이 follow-up은 기존 split subtask `05+04_event_relay_boundary` 안의 1차 리뷰 Required issue만 해결합니다. +- 새 subtask로 분리하지 않습니다. 변경 범위가 `EdgeServer`/`EdgeRegistry` receive path와 해당 regression test에 좁게 묶여 있고, 기존 event relay 작업 완료 판단을 보완하는 성격입니다. +- predecessor `04_status_snapshot`은 `agent-task/archive/2026/05/m-control-plane-edge-wire-baseline/04_status_snapshot/complete.log`로 PASS 완료가 확인되었습니다. + +### 범위 결정 근거 + +- Edge connector relay loop, Control Plane HTTP `/edges/{edge_id}/events` response shape, proto 계약, durable event history, ack/retry/backpressure policy는 변경하지 않습니다. +- 필요한 경우에만 `RecordNodeEvent` 구현을 private helper로 분리하고, server receive path에는 token-aware guard를 적용합니다. +- 검증 보완은 stale reconnect regression, focused wire package test, 기존 final verification 재확인에 한정합니다. + +### 빌드 등급 + +- build lane: `cloud-G08`, review lane: `cloud-G08`. +- 근거: Control Plane-Edge wire concurrency/protocol invariant를 보완하는 follow-up이며, 기존 verification trust가 원격 실행 규칙 때문에 리뷰 재실행되지 않았습니다. + +## 구현 체크리스트 + +- [ ] Control Plane `EdgeServer`/`EdgeRegistry`의 node event 기록 경로가 connection token을 확인해 superseded stale connection event를 무시하도록 수정한다. +- [ ] stale same-`edge_id` reconnection regression test를 추가해 stale connection event는 저장되지 않고 current connection event만 저장되는지 검증한다. +- [ ] 원격 테스트 환경 규칙에 맞춰 focused/final verification을 실행하고 실제 stdout/stderr를 review stub에 기록한다. +- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## REVIEW_EVENTS-1 Stale Connection Event Guard + +### 문제 + +`apps/control-plane/internal/wire/edge_server.go:75`의 `EdgeNodeEvent` listener는 hello 이후 저장된 `edgeID`만 보고 `registry.RecordNodeEvent(id, event, time.Now())`를 호출합니다. 기존 `MarkDisconnected`와 active client cleanup은 connection `token`으로 stale reconnect를 방어하지만, `apps/control-plane/internal/wire/edge_registry.go:128`의 event 기록 경로에는 같은 guard가 없습니다. + +결과적으로 같은 `edge_id`로 두 번째 connection이 supersede된 뒤에도 오래된 첫 번째 connection이 닫히지 않은 상태에서 event를 보내면 Control Plane recent event store가 stale event를 받아들일 수 있습니다. + +### 해결 방법 + +`EdgeServer` listener에서 `edgeID`와 `token`을 함께 캡처하고, registry lock 안에서 현재 tracked state의 token과 일치할 때만 event를 append하는 token-aware path를 추가합니다. 기존 tests가 직접 seed 용도로 쓰는 `RecordNodeEvent(edgeID, event, receivedAt)`를 유지하려면 private helper를 분리하고 server path에서만 token guard variant를 사용합니다. + +예상 형태: + +```go +func (r *EdgeRegistry) RecordNodeEventFromConnection(edgeID string, token uint64, event *iop.EdgeNodeEvent, receivedAt time.Time) bool { + return r.recordNodeEvent(edgeID, token, true, event, receivedAt) +} +``` + +server listener: + +```go +idMu.Lock() +id := edgeID +tok := token +idMu.Unlock() +if id == "" { + return +} +registry.RecordNodeEventFromConnection(id, tok, event, time.Now()) +``` + +### 수정 파일 및 체크리스트 + +- [ ] `apps/control-plane/internal/wire/edge_registry.go`: token-aware record path 추가, 기존 copy/timestamp/bounded behavior 유지. +- [ ] `apps/control-plane/internal/wire/edge_server.go`: event listener가 `edgeID`와 `token`을 함께 전달하도록 수정. +- [ ] `apps/control-plane/internal/wire/edge_server_test.go`: stale reconnect event regression 추가. + +### 테스트 작성 + +- `TestEdgeServerStaleConnectionNodeEventDoesNotPolluteReconnect` 또는 기존 `TestEdgeServerStaleDisconnectDoesNotClearReconnect` 확장. +- fixture: 같은 `edgeID`로 `conn1` hello, `conn2` hello를 순서대로 보낸다. +- assertion: `conn1.Send(EdgeNodeEvent{EventId:"evt-stale"})` 후 registry에 event가 없어야 한다. `conn2.Send(EdgeNodeEvent{EventId:"evt-current"})` 후 registry에는 current event만 있어야 한다. + +### 중간 검증 + +```bash +ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -count=1 ./apps/control-plane/internal/wire' +``` + +기대 결과: stale reconnect event regression과 기존 wire tests가 통과합니다. + +## 수정 파일 요약 + +| 파일 | 항목 | +|------|------| +| `apps/control-plane/internal/wire/edge_registry.go` | REVIEW_EVENTS-1 | +| `apps/control-plane/internal/wire/edge_server.go` | REVIEW_EVENTS-1 | +| `apps/control-plane/internal/wire/edge_server_test.go` | REVIEW_EVENTS-1 | + +## 최종 검증 + +```bash +ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -count=1 ./apps/control-plane/internal/wire' +ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -count=1 ./apps/control-plane/cmd/control-plane ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./apps/edge/internal/events' +ssh toki@toki-labs.com 'cd /config/workspace/iop && go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire' +ssh toki@toki-labs.com 'cd /config/workspace/iop && go test ./...' +ssh toki@toki-labs.com 'cd /config/workspace/iop && make test-control-plane-edge-wire' +ssh toki@toki-labs.com 'cd /config/workspace/iop && (rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane; test $? -eq 1)' +``` + +기대 결과: 모든 Go/race/smoke 명령이 성공하고, 마지막 boundary check는 match 없이 exit code 0 wrapper로 끝납니다. + +모든 코드 변경 완료 후 반드시 `CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다. diff --git a/agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/CODE_REVIEW-cloud-G08.md b/agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/CODE_REVIEW-cloud-G08.md deleted file mode 100644 index 4b3c63f..0000000 --- a/agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/CODE_REVIEW-cloud-G08.md +++ /dev/null @@ -1,204 +0,0 @@ - - -# Code Review Reference - EVENTS - -> **[IMPLEMENTING AGENT - READ FIRST] Filling in this file is the mandatory final step of implementation.** -> The task is NOT complete until every implementation-owned section below is filled in. -> Complete the `구현 체크리스트`; the final checklist item is mandatory before saving. -> Fill implementation-owned sections, then stop with active files in place and report ready for review. -> If implementation is blocked by a user-only decision, user-owned external environment prerequisite, or scope conflict, fill `사용자 리뷰 요청` with evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. Evidence gaps that a follow-up agent can close by rerunning commands or collecting artifacts are normal follow-up issues, not user-review blockers by themselves. -> Finalization (`코드리뷰 결과`, log rename, `complete.log`, archive moves, `코드리뷰 전용 체크리스트`) is review-agent-only, even after compaction/resume. -> Follow the ownership table at the bottom of this file for which sections you own. - -## 개요 - -date=2026-05-30 -task=m-control-plane-edge-wire-baseline/05+04_event_relay_boundary, plan=0, tag=EVENTS - -## Roadmap Targets - -- Milestone: `agent-roadmap/phase/control-plane-portal-ops/milestones/control-plane-edge-wire-baseline.md` -- Task ids: - - `event-relay`: Edge의 node lifecycle event가 Control Plane까지 relay되는 최소 경로가 있다. - - `service-boundary`: Edge wire handler는 node registry, run dispatch, command request 로직을 복제하지 않고 `apps/edge/internal/service`와 `events.Bus` 같은 표면 중립 경계를 사용한다. -- Completion mode: check-on-pass - -## 이 파일을 읽는 리뷰 에이전트에게 - -> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다. - -각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. -리뷰 완료는 아래 순서까지 끝난 상태를 의미합니다. - -1. 판정을 append한다. -2. `CODE_REVIEW-cloud-G08.md` -> `code_review_cloud_G08_N.log`, `PLAN-cloud-G08.md` -> `plan_cloud_G08_M.log`로 아카이브한다. -3. PASS이면 `complete.log` 작성 후 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`로 이동한다. WARN/FAIL이면 user-review gate를 확인한 뒤 다음 active plan/review 파일 또는 `USER_REVIEW.md`를 작성한다. -4. PASS이고 task group이 `m-control-plane-edge-wire-baseline`이므로 완료 이벤트 메타데이터를 보고한다. roadmap 상태 체크와 `update-roadmap` 호출은 런타임 책임이다. -5. 적용 가능한 `코드리뷰 전용 체크리스트` 항목을 최종 `.log` 위치에서 체크한 뒤 보고한다. - ---- - -## 구현 항목별 완료 여부 - -| 항목 | 완료 여부 | -|------|---------| -| [EVENTS-1] Dependency Gate And Parser | [ ] | -| [EVENTS-2] Control Plane Receive And Recent Event Store | [ ] | -| [EVENTS-3] Edge Events.Bus Relay | [ ] | -| [EVENTS-4] Observation Surface And Boundary Guard | [ ] | -| [EVENTS-5] Regression And Smoke | [ ] | - -## 구현 체크리스트 - -- [ ] `04_status_snapshot` subtask가 PASS 완료된 뒤에만 시작한다. runtime이 이 subtask를 시작했다면 의존성이 충족된 것으로 보고, 수동 실행이면 predecessor PASS evidence 또는 runtime 완료 이벤트 경로를 review stub에 기록한다. -- [ ] Control Plane `EdgeParserMap`/`EdgeServer`가 `EdgeNodeEvent`를 수신해 edge id와 함께 bounded recent event store에 기록하도록 구현한다. -- [ ] Edge `controlplane.Connector`가 `events.Bus.SubscribeAllNodes()`를 통해 node lifecycle event를 Control Plane으로 relay하도록 option과 bootstrap wiring을 추가한다. -- [ ] Control Plane에서 relayed node lifecycle event를 조회할 최소 observation surface와 tests를 추가하되, Edge internal registry/run/command 로직을 복제하지 않는다. -- [ ] Event relay, recent event snapshot, service/events boundary 회귀 테스트를 추가하고 대상 Go 테스트 및 기존 local wire smoke를 통과시킨다. -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. - -## 코드리뷰 전용 체크리스트 - -> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. -> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다. - -- [ ] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. -- [ ] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. -- [ ] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G08_N.log`로 아카이브한다. -- [ ] active `PLAN-*-G??.md`를 `plan_cloud_G08_M.log`로 아카이브한다. -- [ ] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. -- [ ] PASS이면 active task 디렉터리 `agent-task/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`를 `agent-task/archive/YYYY/MM/m-control-plane-edge-wire-baseline/05+04_event_relay_boundary/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. -- [ ] PASS이고 task group이 `m-control-plane-edge-wire-baseline`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. -- [ ] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-control-plane-edge-wire-baseline/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. -- [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.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로 이동한다. - -## 계획 대비 변경 사항 - -_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._ - -## 주요 설계 결정 - -_구현 에이전트가 주요 설계 결정 사항을 기록한다._ - -## 사용자 리뷰 요청 - -_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._ - -- 상태: 없음 -- 사유 유형: 없음 -- 결정 필요: 없음 -- 차단 근거: 없음 -- 실행한 검증/명령: 없음 -- 자동 후속 불가 이유: 없음 -- 재개 조건: 없음 - -## 리뷰어를 위한 체크포인트 - -- `04_status_snapshot`의 predecessor PASS evidence 또는 runtime 완료 이벤트 경로가 없는 상태에서 코드 구현이 진행되지 않았는지 확인한다. -- Control Plane receive path가 connection의 `edgeID` context에 event를 묶고, hello 전 event를 무시하는지 확인한다. -- Edge connector relay가 `events.Bus.SubscribeAllNodes()`를 사용하고 node registry/run dispatch/command request 로직을 직접 복제하지 않는지 확인한다. -- relayed event copy/enrichment가 원본 bus event를 mutate하지 않는지 확인한다. -- `/edges/{edge_id}/events`는 Control Plane에 도착한 event 조회만 제공하며 Edge internals를 직접 재구성하지 않는지 확인한다. -- `-race` 검증과 review stub의 실제 stdout/stderr가 신뢰 가능한지 확인한다. - -## 검증 결과 - -_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._ - -필수 규칙: -- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다. -- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다. -- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다. -- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다. - -### EVENTS-1 중간 검증 - -```text -$ go test -count=1 ./apps/control-plane/internal/wire -(output) -``` - -### EVENTS-2 중간 검증 - -```text -$ go test -count=1 ./apps/control-plane/internal/wire -(output) -``` - -### EVENTS-3 중간 검증 - -```text -$ go test -count=1 ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./apps/edge/internal/events -(output) -``` - -### EVENTS-4 중간 검증 - -```text -$ go test -count=1 ./apps/control-plane/cmd/control-plane ./apps/control-plane/internal/wire -(output) - -$ rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane -(output) -``` - -### EVENTS-5 중간 검증 - -```text -$ go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire -(output) - -$ go test ./... -(output) - -$ make test-control-plane-edge-wire -(output) -``` - -### 최종 검증 - -```text -$ go test -count=1 ./apps/control-plane/internal/wire -(output) - -$ go test -count=1 ./apps/control-plane/cmd/control-plane -(output) - -$ go test -count=1 ./apps/edge/internal/controlplane ./apps/edge/internal/bootstrap ./apps/edge/internal/events -(output) - -$ go test -race -count=1 ./apps/edge/internal/controlplane ./apps/control-plane/internal/wire -(output) - -$ go test ./... -(output) - -$ make test-control-plane-edge-wire -(output) - -$ rg --sort path -n "apps/edge/internal/node|NodeCommandRequest|RunRequest" apps/edge/internal/controlplane -(output) -``` - ---- - -> **[IMPLEMENTING AGENT - BEFORE SAVING] Have you filled in every implementation-owned section: completion table, implementation checklist, changes from plan, design decisions, and verification output?** -> If anything is blank, go back and fill it in before saving this file. -> Leave review-agent-only sections unchanged. - -## 섹션 소유권 - -| 섹션 | 소유자 | 설명 | -|------|--------|------| -| 헤더 주석, 개요, 리뷰 에이전트 지시 | 스텁 생성 시 고정 | 구현 에이전트가 수정하거나 실행하지 않음 | -| Roadmap Targets | plan에서 복사해 스텁 생성 시 고정 | 구현 에이전트가 수정하지 않음 | -| 구현 항목별 완료 여부 | 구현 에이전트 | `[ ]` -> `[x]` 체크 | -| 구현 체크리스트 | 구현 에이전트 | `[ ]` -> `[x]` 체크, 마지막 항목 필수 | -| 코드리뷰 전용 체크리스트 | 리뷰 에이전트 | 구현 에이전트가 수정하지 않음 | -| 계획 대비 변경 사항, 주요 설계 결정 | 구현 에이전트 | 실제 내용으로 교체 | -| 사용자 리뷰 요청 | 구현 에이전트 | 기본 `상태: 없음` 유지 또는 차단 근거 기록 | -| 리뷰어를 위한 체크포인트 | 스텁 생성 시 고정 | 리뷰 초점 | -| 검증 결과 | 구현 에이전트 | 실제 stdout/stderr 기록 | -| 코드리뷰 결과 | 리뷰 에이전트 | 스텁에 포함하지 않음 | diff --git a/apps/control-plane/cmd/control-plane/main.go b/apps/control-plane/cmd/control-plane/main.go index 741a9ec..73919f7 100644 --- a/apps/control-plane/cmd/control-plane/main.go +++ b/apps/control-plane/cmd/control-plane/main.go @@ -220,6 +220,24 @@ type edgeRegistryView struct { LastSeen time.Time `json:"last_seen"` } +type edgeNodeEventsResponse struct { + EdgeID string `json:"edge_id"` + Events []edgeNodeEventView `json:"events"` +} + +type edgeNodeEventView struct { + EdgeID string `json:"edge_id"` + EventID string `json:"event_id"` + Type string `json:"type"` + Source string `json:"source"` + NodeID string `json:"node_id"` + Alias string `json:"alias"` + Reason string `json:"reason"` + Timestamp time.Time `json:"timestamp"` + ReceivedAt time.Time `json:"received_at"` + Metadata map[string]string `json:"metadata,omitempty"` +} + func registerEdgeRegistryHandlers(mux *http.ServeMux, registry *wire.EdgeRegistry) { mux.HandleFunc("/edges", func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/edges" { @@ -244,11 +262,29 @@ func registerEdgeRegistryHandlers(mux *http.ServeMux, registry *wire.EdgeRegistr http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } - edgeID := strings.TrimPrefix(r.URL.Path, "/edges/") - if edgeID == "" || strings.Contains(edgeID, "/") { + edgePath := strings.TrimPrefix(r.URL.Path, "/edges/") + edgeID, rest, hasRest := strings.Cut(edgePath, "/") + if edgeID == "" { http.NotFound(w, r) return } + if hasRest { + if rest != "events" { + http.NotFound(w, r) + return + } + if _, ok := registry.Snapshot(edgeID); !ok { + http.NotFound(w, r) + return + } + records := registry.NodeEvents(edgeID, r.URL.Query().Get("node_id")) + views := make([]edgeNodeEventView, 0, len(records)) + for _, record := range records { + views = append(views, edgeNodeEventViewFromRecord(record)) + } + writeJSON(w, http.StatusOK, edgeNodeEventsResponse{EdgeID: edgeID, Events: views}) + return + } state, ok := registry.Snapshot(edgeID) if !ok { http.NotFound(w, r) @@ -258,6 +294,33 @@ func registerEdgeRegistryHandlers(mux *http.ServeMux, registry *wire.EdgeRegistr }) } +func edgeNodeEventViewFromRecord(record wire.EdgeNodeEventRecord) edgeNodeEventView { + event := record.Event + view := edgeNodeEventView{ + EdgeID: record.EdgeID, + ReceivedAt: record.ReceivedAt, + } + if event == nil { + return view + } + if event.GetTimestamp() != 0 { + view.Timestamp = time.Unix(0, event.GetTimestamp()).UTC() + } + view.EventID = event.GetEventId() + view.Type = event.GetType() + view.Source = event.GetSource() + view.NodeID = event.GetNodeId() + view.Alias = event.GetAlias() + view.Reason = event.GetReason() + if metadata := event.GetMetadata(); len(metadata) > 0 { + view.Metadata = make(map[string]string, len(metadata)) + for k, v := range metadata { + view.Metadata[k] = v + } + } + return view +} + func edgeRegistryViewFromState(state wire.EdgeConnectionState) edgeRegistryView { return edgeRegistryView{ EdgeID: state.EdgeID, diff --git a/apps/control-plane/cmd/control-plane/main_test.go b/apps/control-plane/cmd/control-plane/main_test.go index bb0f2ff..11aa48a 100644 --- a/apps/control-plane/cmd/control-plane/main_test.go +++ b/apps/control-plane/cmd/control-plane/main_test.go @@ -298,6 +298,100 @@ func TestEdgeRegistryHTTPHandlersListAndGetEdge(t *testing.T) { } } +func TestEdgeRegistryHTTPHandlersListNodeEvents(t *testing.T) { + registry := wire.NewEdgeRegistry() + connectedAt := time.Unix(1780142200, 0).UTC() + registry.MarkConnected(&iop.EdgeHelloRequest{ + EdgeId: "edge-a", + EdgeName: "Edge A", + Version: "1.0.0", + }, connectedAt) + registry.MarkConnected(&iop.EdgeHelloRequest{ + EdgeId: "edge-b", + EdgeName: "Edge B", + Version: "1.0.0", + }, connectedAt) + + eventAt := time.Unix(1780142300, 0).UTC() + receivedAt := eventAt.Add(time.Second) + registry.RecordNodeEvent("edge-a", &iop.EdgeNodeEvent{ + EventId: "evt-node-1", + Type: "node.connected", + Source: "edge", + NodeId: "node-1", + Alias: "alpha", + Reason: "registered", + Timestamp: eventAt.UnixNano(), + Metadata: map[string]string{"rack": "r1"}, + }, receivedAt) + registry.RecordNodeEvent("edge-b", &iop.EdgeNodeEvent{ + EventId: "evt-node-2", + Type: "node.connected", + NodeId: "node-2", + Timestamp: eventAt.UnixNano(), + }, receivedAt) + + mux := http.NewServeMux() + registerEdgeRegistryHandlers(mux, registry) + + resp := httptest.NewRecorder() + mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/events", nil)) + if resp.Code != http.StatusOK { + t.Fatalf("GET /edges/edge-a/events status=%d body=%s", resp.Code, resp.Body.String()) + } + var got edgeNodeEventsResponse + if err := json.Unmarshal(resp.Body.Bytes(), &got); err != nil { + t.Fatalf("decode events response: %v", err) + } + if got.EdgeID != "edge-a" { + t.Fatalf("edge_id: got %q want edge-a", got.EdgeID) + } + if len(got.Events) != 1 { + t.Fatalf("events len: got %d want 1", len(got.Events)) + } + event := got.Events[0] + if event.EventID != "evt-node-1" || event.Type != "node.connected" { + t.Fatalf("unexpected event identity: %+v", event) + } + if event.NodeID != "node-1" || event.Alias != "alpha" || event.Reason != "registered" { + t.Fatalf("unexpected node event view: %+v", event) + } + if !event.Timestamp.Equal(eventAt) { + t.Fatalf("timestamp: got %s want %s", event.Timestamp, eventAt) + } + if !event.ReceivedAt.Equal(receivedAt) { + t.Fatalf("received_at: got %s want %s", event.ReceivedAt, receivedAt) + } + if event.Metadata["rack"] != "r1" { + t.Fatalf("metadata rack: got %q", event.Metadata["rack"]) + } + + filtered := httptest.NewRecorder() + mux.ServeHTTP(filtered, httptest.NewRequest(http.MethodGet, "/edges/edge-a/events?node_id=missing", nil)) + if filtered.Code != http.StatusOK { + t.Fatalf("filtered events status=%d body=%s", filtered.Code, filtered.Body.String()) + } + var filteredGot edgeNodeEventsResponse + if err := json.Unmarshal(filtered.Body.Bytes(), &filteredGot); err != nil { + t.Fatalf("decode filtered events response: %v", err) + } + if len(filteredGot.Events) != 0 { + t.Fatalf("expected empty filtered events, got %+v", filteredGot.Events) + } + + extraResp := httptest.NewRecorder() + mux.ServeHTTP(extraResp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/events/extra", nil)) + if extraResp.Code != http.StatusNotFound { + t.Fatalf("GET /edges/edge-a/events/extra status=%d body=%s", extraResp.Code, extraResp.Body.String()) + } + + missingResp := httptest.NewRecorder() + mux.ServeHTTP(missingResp, httptest.NewRequest(http.MethodGet, "/edges/missing-edge/events", nil)) + if missingResp.Code != http.StatusNotFound { + t.Fatalf("GET /edges/missing-edge/events status=%d body=%s", missingResp.Code, missingResp.Body.String()) + } +} + func TestEdgeRegistryHTTPHandlersRejectUnsupportedCases(t *testing.T) { registry := wire.NewEdgeRegistry() mux := http.NewServeMux() diff --git a/apps/control-plane/internal/wire/edge.go b/apps/control-plane/internal/wire/edge.go index 4b32ae1..0cfa0c0 100644 --- a/apps/control-plane/internal/wire/edge.go +++ b/apps/control-plane/internal/wire/edge.go @@ -27,5 +27,9 @@ func EdgeParserMap() proto_socket.ParserMap { res := &iop.EdgeStatusResponse{} return res, proto.Unmarshal(b, res) }, + proto_socket.TypeNameOf(&iop.EdgeNodeEvent{}): func(b []byte) (proto.Message, error) { + event := &iop.EdgeNodeEvent{} + return event, proto.Unmarshal(b, event) + }, } } diff --git a/apps/control-plane/internal/wire/edge_registry.go b/apps/control-plane/internal/wire/edge_registry.go index eabe7be..f5613ee 100644 --- a/apps/control-plane/internal/wire/edge_registry.go +++ b/apps/control-plane/internal/wire/edge_registry.go @@ -6,12 +6,14 @@ import ( "time" proto_socket "git.toki-labs.com/toki/proto-socket/go" + "google.golang.org/protobuf/proto" iop "iop/proto/gen/iop" ) +const recentNodeEventCapacity = 128 + // EdgeConnectionState captures the Control Plane's internal view of a single -// Edge wire connection. It is a registry foundation only; public status query -// and event relay APIs are deferred to later subtasks. +// Edge wire connection. type EdgeConnectionState struct { EdgeID string EdgeName string @@ -28,11 +30,20 @@ type EdgeConnectionState struct { token uint64 } +// EdgeNodeEventRecord is the Control Plane's copied recent view of one node +// lifecycle event as received from a specific Edge connection. +type EdgeNodeEventRecord struct { + EdgeID string + Event *iop.EdgeNodeEvent + ReceivedAt time.Time +} + // EdgeRegistry stores Edge connection state keyed by edge_id behind a mutex. type EdgeRegistry struct { - mu sync.Mutex - edges map[string]*EdgeConnectionState - nextToken uint64 + mu sync.Mutex + edges map[string]*EdgeConnectionState + nextToken uint64 + recentNodeEvents []EdgeNodeEventRecord } // NewEdgeRegistry returns an empty Edge connection registry. @@ -111,9 +122,100 @@ func (r *EdgeRegistry) Snapshots() []EdgeConnectionState { return snapshots } +// RecordNodeEvent stores a copied EdgeNodeEvent under the identified Edge. It +// is intended for tests and non-connection seed paths that do not have an +// active connection token. +func (r *EdgeRegistry) RecordNodeEvent(edgeID string, event *iop.EdgeNodeEvent, receivedAt time.Time) { + r.recordNodeEvent(edgeID, 0, false, event, receivedAt) +} + +// RecordNodeEventFromConnection stores a copied EdgeNodeEvent only when the +// supplied token still identifies the current connection for edgeID. It returns +// false for stale superseded connections. +func (r *EdgeRegistry) RecordNodeEventFromConnection(edgeID string, token uint64, event *iop.EdgeNodeEvent, receivedAt time.Time) bool { + return r.recordNodeEvent(edgeID, token, true, event, receivedAt) +} + +// recordNodeEvent appends to the bounded recent buffer. The recent buffer is +// global to the in-memory registry; durable event history is intentionally +// outside this baseline. +func (r *EdgeRegistry) recordNodeEvent(edgeID string, token uint64, requireToken bool, event *iop.EdgeNodeEvent, receivedAt time.Time) bool { + if edgeID == "" || event == nil { + return false + } + if receivedAt.IsZero() { + receivedAt = time.Now() + } + copied := cloneEdgeNodeEvent(event) + if copied.GetTimestamp() == 0 { + copied.Timestamp = receivedAt.UnixNano() + } + + r.mu.Lock() + defer r.mu.Unlock() + if requireToken { + state, ok := r.edges[edgeID] + if !ok || state.token != token { + return false + } + } + if len(r.recentNodeEvents) >= recentNodeEventCapacity { + r.recentNodeEvents = append(r.recentNodeEvents[:0], r.recentNodeEvents[1:]...) + } + r.recentNodeEvents = append(r.recentNodeEvents, EdgeNodeEventRecord{ + EdgeID: edgeID, + Event: copied, + ReceivedAt: receivedAt, + }) + return true +} + +// NodeEvents returns copied recent node lifecycle events for an Edge. When +// nodeID is non-empty, only events for that node are returned. +func (r *EdgeRegistry) NodeEvents(edgeID, nodeID string) []EdgeNodeEventRecord { + if edgeID == "" { + return nil + } + r.mu.Lock() + defer r.mu.Unlock() + out := make([]EdgeNodeEventRecord, 0, len(r.recentNodeEvents)) + for _, record := range r.recentNodeEvents { + if record.EdgeID != edgeID { + continue + } + if nodeID != "" && record.Event.GetNodeId() != nodeID { + continue + } + out = append(out, cloneEdgeNodeEventRecord(record)) + } + if len(out) == 0 { + return nil + } + return out +} + // Len returns the number of tracked Edge connections. func (r *EdgeRegistry) Len() int { r.mu.Lock() defer r.mu.Unlock() return len(r.edges) } + +func cloneEdgeNodeEventRecord(record EdgeNodeEventRecord) EdgeNodeEventRecord { + return EdgeNodeEventRecord{ + EdgeID: record.EdgeID, + Event: cloneEdgeNodeEvent(record.Event), + ReceivedAt: record.ReceivedAt, + } +} + +func cloneEdgeNodeEvent(event *iop.EdgeNodeEvent) *iop.EdgeNodeEvent { + if event == nil { + return nil + } + copied, ok := proto.Clone(event).(*iop.EdgeNodeEvent) + if !ok { + return &iop.EdgeNodeEvent{} + } + return copied +} diff --git a/apps/control-plane/internal/wire/edge_server.go b/apps/control-plane/internal/wire/edge_server.go index 3a9e9fa..6ceb3bc 100644 --- a/apps/control-plane/internal/wire/edge_server.go +++ b/apps/control-plane/internal/wire/edge_server.go @@ -72,6 +72,17 @@ func NewEdgeServer(listenAddr string, logger *zap.Logger) (*EdgeServer, error) { token uint64 ) + proto_socket.AddListenerTyped[*iop.EdgeNodeEvent](&client.Communicator, func(event *iop.EdgeNodeEvent) { + idMu.Lock() + id := edgeID + tok := token + idMu.Unlock() + if id == "" { + return + } + registry.RecordNodeEventFromConnection(id, tok, event, time.Now()) + }) + proto_socket.AddRequestListenerTyped(&client.Communicator, func(req *iop.EdgeHelloRequest) (*iop.EdgeHelloResponse, error) { now := time.Now() if req.GetEdgeId() == "" { diff --git a/apps/control-plane/internal/wire/edge_server_test.go b/apps/control-plane/internal/wire/edge_server_test.go index cd830d7..d4af94c 100644 --- a/apps/control-plane/internal/wire/edge_server_test.go +++ b/apps/control-plane/internal/wire/edge_server_test.go @@ -225,6 +225,62 @@ func TestEdgeServerStaleDisconnectDoesNotClearReconnect(t *testing.T) { } } +func TestEdgeServerStaleConnectionNodeEventDoesNotPolluteReconnect(t *testing.T) { + server, port := startEdgeServer(t) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + const edgeID = "edge-dgx-group" + + conn1 := dialEdge(t, ctx, port) + defer conn1.Close() + sendEdgeHello(t, conn1, edgeID) + + conn2 := dialEdge(t, ctx, port) + defer conn2.Close() + sendEdgeHello(t, conn2, edgeID) + + if err := conn1.Send(&iop.EdgeNodeEvent{ + EventId: "evt-stale", + Type: "node.connected", + NodeId: "node-stale", + }); err != nil { + t.Fatalf("send stale EdgeNodeEvent: %v", err) + } + time.Sleep(300 * time.Millisecond) + if got := server.Registry().NodeEvents(edgeID, ""); len(got) != 0 { + t.Fatalf("stale connection event polluted registry: %+v", got) + } + + if err := conn2.Send(&iop.EdgeNodeEvent{ + EventId: "evt-current", + Type: "node.connected", + NodeId: "node-current", + }); err != nil { + t.Fatalf("send current EdgeNodeEvent: %v", err) + } + + var records []EdgeNodeEventRecord + deadline := time.Now().Add(2 * time.Second) + for { + records = server.Registry().NodeEvents(edgeID, "") + if len(records) == 1 { + break + } + if time.Now().After(deadline) { + t.Fatalf("current node event was not recorded in time; got %d records", len(records)) + } + time.Sleep(20 * time.Millisecond) + } + if records[0].Event.GetEventId() != "evt-current" { + t.Fatalf("recorded event_id: got %q want evt-current", records[0].Event.GetEventId()) + } + if records[0].Event.GetNodeId() != "node-current" { + t.Fatalf("recorded node_id: got %q want node-current", records[0].Event.GetNodeId()) + } +} + func TestEdgeServerRequestsStatusFromConnectedEdge(t *testing.T) { server, port := startEdgeServer(t) @@ -270,6 +326,84 @@ func TestEdgeServerRequestsStatusFromConnectedEdge(t *testing.T) { } } +func TestEdgeServerRecordsNodeLifecycleEventFromEdge(t *testing.T) { + server, port := startEdgeServer(t) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + client := dialEdge(t, ctx, port) + defer client.Close() + + if err := client.Send(&iop.EdgeNodeEvent{ + EventId: "evt-before-hello", + Type: "node.connected", + NodeId: "node-early", + }); err != nil { + t.Fatalf("send pre-hello EdgeNodeEvent: %v", err) + } + time.Sleep(100 * time.Millisecond) + if got := server.Registry().NodeEvents("edge-dgx-group", ""); len(got) != 0 { + t.Fatalf("expected pre-hello node event to be ignored, got %+v", got) + } + + sendEdgeHello(t, client, "edge-dgx-group") + + event := &iop.EdgeNodeEvent{ + EventId: "evt-node-1", + Type: "node.connected", + Source: "edge", + NodeId: "node-1", + Alias: "alpha", + Reason: "registered", + Metadata: map[string]string{"rack": "r1"}, + } + if err := client.Send(event); err != nil { + t.Fatalf("send EdgeNodeEvent: %v", err) + } + + var records []EdgeNodeEventRecord + deadline := time.Now().Add(2 * time.Second) + for { + records = server.Registry().NodeEvents("edge-dgx-group", "") + if len(records) == 1 { + break + } + if time.Now().After(deadline) { + t.Fatalf("node event was not recorded in time; got %d records", len(records)) + } + time.Sleep(20 * time.Millisecond) + } + got := records[0] + if got.EdgeID != "edge-dgx-group" { + t.Errorf("edge_id: got %q want %q", got.EdgeID, "edge-dgx-group") + } + if got.Event.GetEventId() != "evt-node-1" || got.Event.GetType() != "node.connected" { + t.Fatalf("unexpected recorded event: %+v", got.Event) + } + if got.Event.GetNodeId() != "node-1" || got.Event.GetAlias() != "alpha" { + t.Fatalf("unexpected node identity: %+v", got.Event) + } + if got.Event.GetMetadata()["rack"] != "r1" { + t.Fatalf("metadata rack: got %q", got.Event.GetMetadata()["rack"]) + } + if got.Event.GetTimestamp() == 0 { + t.Fatal("expected registry to fill missing event timestamp") + } + if event.GetTimestamp() != 0 { + t.Fatal("registry mutated original event timestamp") + } + + records[0].Event.Metadata["rack"] = "mutated" + again := server.Registry().NodeEvents("edge-dgx-group", "node-1") + if len(again) != 1 { + t.Fatalf("filtered node events: got %d want 1", len(again)) + } + if again[0].Event.GetMetadata()["rack"] != "r1" { + t.Fatalf("registry event was mutated through returned snapshot: %+v", again[0].Event.GetMetadata()) + } +} + func TestEdgeServerRequestStatusFailsWhenEdgeNotConnected(t *testing.T) { server, _ := startEdgeServer(t) if _, err := server.RequestStatus("missing-edge", time.Second); err == nil { @@ -309,6 +443,35 @@ func TestEdgeRegistrySnapshotsAreSortedAndCopied(t *testing.T) { } } +func TestEdgeRegistryNodeEventsAreBoundedAndCopied(t *testing.T) { + registry := NewEdgeRegistry() + receivedAt := time.Unix(1780142400, 0).UTC() + + for i := 0; i < recentNodeEventCapacity+2; i++ { + registry.RecordNodeEvent("edge-a", &iop.EdgeNodeEvent{ + EventId: fmt.Sprintf("evt-%03d", i), + Type: "node.connected", + NodeId: "node-1", + Metadata: map[string]string{ + "index": fmt.Sprintf("%d", i), + }, + }, receivedAt.Add(time.Duration(i)*time.Nanosecond)) + } + + events := registry.NodeEvents("edge-a", "") + if len(events) != recentNodeEventCapacity { + t.Fatalf("node events len: got %d want %d", len(events), recentNodeEventCapacity) + } + if events[0].Event.GetEventId() != "evt-002" { + t.Fatalf("expected oldest retained event evt-002, got %q", events[0].Event.GetEventId()) + } + events[0].Event.Metadata["index"] = "mutated" + again := registry.NodeEvents("edge-a", "") + if again[0].Event.GetMetadata()["index"] == "mutated" { + t.Fatal("NodeEvents returned mutable registry storage") + } +} + func TestEdgeHeartbeatWaitExceedsInterval(t *testing.T) { if EdgeHeartbeatWaitSec <= EdgeHeartbeatIntervalSec { t.Fatalf("EdgeHeartbeatWaitSec (%d) must exceed EdgeHeartbeatIntervalSec (%d)", diff --git a/apps/control-plane/internal/wire/edge_test.go b/apps/control-plane/internal/wire/edge_test.go index 1d97e52..e352a93 100644 --- a/apps/control-plane/internal/wire/edge_test.go +++ b/apps/control-plane/internal/wire/edge_test.go @@ -172,6 +172,59 @@ func TestEdgeParserMapStatusRoundTrip(t *testing.T) { } } +func TestEdgeParserMapNodeEventRoundTrip(t *testing.T) { + parserMap := EdgeParserMap() + typeName := proto_socket.TypeNameOf(&iop.EdgeNodeEvent{}) + parse, ok := parserMap[typeName] + if !ok { + t.Fatalf("missing parser for %s", typeName) + } + + event := &iop.EdgeNodeEvent{ + EventId: "evt-node-1", + Type: "node.connected", + Source: "edge", + NodeId: "node-1", + Alias: "alpha", + Reason: "registered", + Timestamp: time.Now().UnixNano(), + Metadata: map[string]string{ + "rack": "r1", + }, + } + payload, err := proto.Marshal(event) + if err != nil { + t.Fatalf("marshal EdgeNodeEvent: %v", err) + } + + got, err := parse(payload) + if err != nil { + t.Fatalf("parse EdgeNodeEvent: %v", err) + } + parsed, ok := got.(*iop.EdgeNodeEvent) + if !ok { + t.Fatalf("expected *iop.EdgeNodeEvent, got %T", got) + } + if parsed.GetEventId() != event.GetEventId() { + t.Errorf("event_id: got %q want %q", parsed.GetEventId(), event.GetEventId()) + } + if parsed.GetType() != event.GetType() { + t.Errorf("type: got %q want %q", parsed.GetType(), event.GetType()) + } + if parsed.GetNodeId() != event.GetNodeId() { + t.Errorf("node_id: got %q want %q", parsed.GetNodeId(), event.GetNodeId()) + } + if parsed.GetAlias() != event.GetAlias() { + t.Errorf("alias: got %q want %q", parsed.GetAlias(), event.GetAlias()) + } + if parsed.GetTimestamp() != event.GetTimestamp() { + t.Errorf("timestamp: got %d want %d", parsed.GetTimestamp(), event.GetTimestamp()) + } + if parsed.GetMetadata()["rack"] != "r1" { + t.Errorf("metadata rack: got %q want %q", parsed.GetMetadata()["rack"], "r1") + } +} + // TestEdgeStatusRequestDoesNotExposeNodeDirectScheduling guards that the status // request stays a correlation-only contract and never grows direct Node // addressing or scheduling fields. diff --git a/apps/edge/internal/bootstrap/runtime.go b/apps/edge/internal/bootstrap/runtime.go index 02183b7..de4aaea 100644 --- a/apps/edge/internal/bootstrap/runtime.go +++ b/apps/edge/internal/bootstrap/runtime.go @@ -67,6 +67,7 @@ func NewRuntime(cfg *config.EdgeConfig) (*Runtime, error) { version.Version, logger.Named("controlplane"), edgecontrolplane.WithStatusProvider(svc), + edgecontrolplane.WithNodeEventBus(bus), ) return &Runtime{ diff --git a/apps/edge/internal/bootstrap/runtime_test.go b/apps/edge/internal/bootstrap/runtime_test.go index b770b00..9eea5e7 100644 --- a/apps/edge/internal/bootstrap/runtime_test.go +++ b/apps/edge/internal/bootstrap/runtime_test.go @@ -207,6 +207,19 @@ func TestNewRuntimeWiresStatusProviderIntoConnector(t *testing.T) { } } +// TestNewRuntimeWiresNodeEventBusIntoConnector verifies that node lifecycle +// events are relayed through the shared events.Bus boundary. +func TestNewRuntimeWiresNodeEventBusIntoConnector(t *testing.T) { + cfg := newTestConfig() + rt, err := NewRuntime(cfg) + if err != nil { + t.Fatalf("NewRuntime: %v", err) + } + if !rt.ControlPlane.NodeEventBusConfigured() { + t.Fatal("expected NewRuntime to wire the node event bus into the connector") + } +} + // TestRuntimeStartsControlPlaneConnectorWhenEnabled verifies that Start does // not fail when a Control Plane connector is configured with a local fake // server accepting hello. Uses a local TCP fake endpoint. diff --git a/apps/edge/internal/controlplane/connector.go b/apps/edge/internal/controlplane/connector.go index d4c1c53..a283223 100644 --- a/apps/edge/internal/controlplane/connector.go +++ b/apps/edge/internal/controlplane/connector.go @@ -13,6 +13,7 @@ import ( "go.uber.org/zap" "google.golang.org/protobuf/proto" + edgeevents "iop/apps/edge/internal/events" edgeservice "iop/apps/edge/internal/service" "iop/packages/config" iop "iop/proto/gen/iop" @@ -55,6 +56,13 @@ func WithStatusProvider(provider StatusProvider) Option { return func(c *Connector) { c.statusProvider = provider } } +// WithNodeEventBus wires the Edge lifecycle event fanout surface the Connector +// relays to Control Plane. It keeps the connector at the events.Bus boundary +// instead of reaching into node registry or transport internals. +func WithNodeEventBus(bus *edgeevents.Bus) Option { + return func(c *Connector) { c.nodeEvents = bus } +} + // Connector manages the outbound TCP connection from Edge to Control Plane. // When disabled (enabled=false or empty wire_addr) it is a safe no-op. type Connector struct { @@ -64,6 +72,7 @@ type Connector struct { logger *zap.Logger statusProvider StatusProvider + nodeEvents *edgeevents.Bus state atomic.Int32 // stores State cancelOnce sync.Once @@ -96,6 +105,11 @@ func (c *Connector) StatusProviderConfigured() bool { return c.statusProvider != nil } +// NodeEventBusConfigured reports whether node lifecycle event relay is wired. +func (c *Connector) NodeEventBusConfigured() bool { + return c.nodeEvents != nil +} + // Start begins the outbound connection loop. It is a no-op when enabled=false // or wire_addr is empty. ctx cancellation is used only for the startup phase; // the internal lifetime context drives reconnect. Stop() tears down the loop. @@ -252,6 +266,9 @@ func (c *Connector) connect(ctx context.Context) error { default: } + stopNodeRelay := c.startNodeEventRelay(ctx, cl) + defer stopNodeRelay() + c.state.Store(int32(StateConnected)) c.logger.Info("connected to control plane", zap.String("wire_addr", c.cpConf.WireAddr), @@ -268,6 +285,61 @@ func (c *Connector) connect(ctx context.Context) error { return nil } +func (c *Connector) startNodeEventRelay(ctx context.Context, cl *toki.TcpClient) func() { + if c.nodeEvents == nil { + return func() {} + } + events, unsubscribe := c.nodeEvents.SubscribeAllNodes(64) + relayCtx, cancel := context.WithCancel(ctx) + go func() { + defer unsubscribe() + for { + select { + case <-relayCtx.Done(): + return + case event, ok := <-events: + if !ok { + return + } + relay := c.nodeEventForRelay(event) + if relay == nil { + continue + } + if err := cl.Send(relay); err != nil { + c.logger.Warn("relay node lifecycle event to control plane failed", + zap.String("event_id", relay.GetEventId()), + zap.String("node_id", relay.GetNodeId()), + zap.Error(err), + ) + } + } + } + }() + return cancel +} + +func (c *Connector) nodeEventForRelay(event *iop.EdgeNodeEvent) *iop.EdgeNodeEvent { + if event == nil { + return nil + } + relay, ok := proto.Clone(event).(*iop.EdgeNodeEvent) + if !ok { + return nil + } + if relay.Metadata == nil { + relay.Metadata = make(map[string]string) + } + if c.edgeInfo.ID != "" { + if _, ok := relay.Metadata["edge_id"]; !ok { + relay.Metadata["edge_id"] = c.edgeInfo.ID + } + } + if relay.Timestamp == 0 { + relay.Timestamp = time.Now().UnixNano() + } + return relay +} + // buildStatusResponse converts the Edge-owned node snapshot surface into the // status response. When no status provider is wired it returns an explicit // error response rather than a silently empty snapshot. diff --git a/apps/edge/internal/controlplane/connector_test.go b/apps/edge/internal/controlplane/connector_test.go index dc4abbf..a81be5f 100644 --- a/apps/edge/internal/controlplane/connector_test.go +++ b/apps/edge/internal/controlplane/connector_test.go @@ -10,12 +10,20 @@ import ( toki "git.toki-labs.com/toki/proto-socket/go" "google.golang.org/protobuf/proto" - edgenode "iop/apps/edge/internal/node" + edgeevents "iop/apps/edge/internal/events" edgeservice "iop/apps/edge/internal/service" "iop/packages/config" iop "iop/proto/gen/iop" ) +type fakeStatusProvider struct { + nodes []edgeservice.NodeSnapshot +} + +func (f fakeStatusProvider) ListNodeSnapshots() []edgeservice.NodeSnapshot { + return append([]edgeservice.NodeSnapshot(nil), f.nodes...) +} + // fakeCPParserMap returns a parser for EdgeHelloRequest on the server side. func fakeCPParserMap() toki.ParserMap { return toki.ParserMap{ @@ -23,6 +31,10 @@ func fakeCPParserMap() toki.ParserMap { m := &iop.EdgeHelloRequest{} return m, proto.Unmarshal(b, m) }, + toki.TypeNameOf(&iop.EdgeNodeEvent{}): func(b []byte) (proto.Message, error) { + m := &iop.EdgeNodeEvent{} + return m, proto.Unmarshal(b, m) + }, } } @@ -142,14 +154,46 @@ func startFakeCPServerForStatus(t *testing.T) (int, <-chan *toki.TcpClient, func } } -// TestConnectorRespondsToStatusRequestFromService verifies that a status -// request from the Control Plane is answered from the Edge service node -// snapshot surface rather than a direct registry/transport read. -func TestConnectorRespondsToStatusRequestFromService(t *testing.T) { - reg := edgenode.NewRegistry() - reg.Register(&edgenode.NodeEntry{NodeID: "node-1", Alias: "alpha"}) - reg.Register(&edgenode.NodeEntry{NodeID: "node-2"}) - svc := edgeservice.New(reg, nil) +func startFakeCPServerForNodeEvents(t *testing.T) (int, <-chan *iop.EdgeNodeEvent, func()) { + t.Helper() + port, err := freeTCPPort() + if err != nil { + t.Fatalf("startFakeCPServerForNodeEvents freeTCPPort: %v", err) + } + events := make(chan *iop.EdgeNodeEvent, 8) + srv := toki.NewTcpServer("127.0.0.1", port, func(conn net.Conn) *toki.TcpClient { + return toki.NewTcpClient(conn, 0, 0, fakeCPParserMap()) + }) + srv.OnClientConnected = func(client *toki.TcpClient) { + toki.AddRequestListenerTyped[*iop.EdgeHelloRequest, *iop.EdgeHelloResponse]( + &client.Communicator, + func(_ *iop.EdgeHelloRequest) (*iop.EdgeHelloResponse, error) { + return &iop.EdgeHelloResponse{Accepted: true, Protocol: "iop/1"}, nil + }, + ) + toki.AddListenerTyped[*iop.EdgeNodeEvent](&client.Communicator, func(event *iop.EdgeNodeEvent) { + events <- event + }) + } + ctx, cancel := context.WithCancel(context.Background()) + if err := srv.Start(ctx); err != nil { + cancel() + t.Fatalf("startFakeCPServerForNodeEvents Start: %v", err) + } + return port, events, func() { + cancel() + _ = srv.Stop() + } +} + +// TestConnectorRespondsToStatusRequestFromProvider verifies that a status +// request from the Control Plane is answered from the configured snapshot +// boundary rather than a direct registry/transport read. +func TestConnectorRespondsToStatusRequestFromProvider(t *testing.T) { + provider := fakeStatusProvider{nodes: []edgeservice.NodeSnapshot{ + {NodeID: "node-1", Alias: "alpha", Label: "node0"}, + {NodeID: "node-2", Label: "node1"}, + }} port, serverClients, stop := startFakeCPServerForStatus(t) defer stop() @@ -163,7 +207,7 @@ func TestConnectorRespondsToStatusRequestFromService(t *testing.T) { }, "0.2.0", noopLogger(), - WithStatusProvider(svc), + WithStatusProvider(provider), ) if !c.StatusProviderConfigured() { t.Fatal("expected status provider to be configured") @@ -212,6 +256,78 @@ func TestConnectorRespondsToStatusRequestFromService(t *testing.T) { } } +func TestConnectorRelaysNodeLifecycleEventsToControlPlane(t *testing.T) { + bus := edgeevents.NewBus() + port, events, stop := startFakeCPServerForNodeEvents(t) + defer stop() + + c := NewConnector( + config.EdgeInfo{ID: "edge-dgx", Name: "DGX Edge"}, + config.EdgeControlPlaneConf{ + Enabled: true, + WireAddr: wireAddr(port), + ReconnectIntervalSec: 60, + }, + "0.2.0", + noopLogger(), + WithNodeEventBus(bus), + ) + if !c.NodeEventBusConfigured() { + t.Fatal("expected node event bus to be configured") + } + if err := c.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + defer c.Stop() + + deadline := time.Now().Add(3 * time.Second) + for c.CurrentState() != StateConnected { + if time.Now().After(deadline) { + t.Fatalf("connector did not reach connected state, got %v", c.CurrentState()) + } + time.Sleep(20 * time.Millisecond) + } + + original := &iop.EdgeNodeEvent{ + EventId: "evt-node-1", + Type: "node.connected", + Source: "edge", + NodeId: "node-1", + Alias: "alpha", + Reason: "registered", + Metadata: map[string]string{"rack": "r1"}, + } + bus.PublishNode(original) + + select { + case got := <-events: + if got.GetEventId() != "evt-node-1" || got.GetType() != "node.connected" { + t.Fatalf("unexpected relayed event: %+v", got) + } + if got.GetNodeId() != "node-1" || got.GetAlias() != "alpha" { + t.Fatalf("unexpected relayed node identity: %+v", got) + } + if got.GetMetadata()["rack"] != "r1" { + t.Fatalf("metadata rack: got %q", got.GetMetadata()["rack"]) + } + if got.GetMetadata()["edge_id"] != "edge-dgx" { + t.Fatalf("metadata edge_id: got %q want edge-dgx", got.GetMetadata()["edge_id"]) + } + if got.GetTimestamp() == 0 { + t.Fatal("expected relay to fill missing timestamp on copied event") + } + case <-time.After(3 * time.Second): + t.Fatal("timeout waiting for relayed EdgeNodeEvent") + } + + if _, ok := original.GetMetadata()["edge_id"]; ok { + t.Fatal("relay mutated original event metadata") + } + if original.GetTimestamp() != 0 { + t.Fatal("relay mutated original event timestamp") + } +} + // TestConnectorStatusRequestWithoutProviderReturnsError verifies that a status // request is answered with an explicit error when no provider is wired. func TestConnectorStatusRequestWithoutProviderReturnsError(t *testing.T) { diff --git a/proto/gen/iop/control.pb.go b/proto/gen/iop/control.pb.go index 9c7a16b..8828bdf 100644 --- a/proto/gen/iop/control.pb.go +++ b/proto/gen/iop/control.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.11 -// protoc v3.21.12 +// protoc v5.29.3 // source: proto/iop/control.proto package iop