refactor: runner proto socket transport hardening - refactor agent components and update tests
- Move completed subtasks (02+01_dispatch_state, 03+01,02_runner_actions) to archive - Refactor agent_runner.dart, oto_server_job_client.dart, registration_client.dart - Update smoke tests and server tests for Go services
This commit is contained in:
parent
88c8ff3b07
commit
dde15565a5
15 changed files with 1272 additions and 60 deletions
|
|
@ -51,41 +51,45 @@ task=m-runner-proto-socket-transport-hardening/02+01_dispatch_state, plan=0, tag
|
|||
|
||||
| 항목 | 완료 여부 |
|
||||
|------|---------|
|
||||
| [API-1] socket queued dispatch와 invalid RunRequest failure | [ ] |
|
||||
| [API-2] socket terminal report와 active execution gate | [ ] |
|
||||
| [API-1] socket queued dispatch와 invalid RunRequest failure | [x] |
|
||||
| [API-2] socket terminal report와 active execution gate | [x] |
|
||||
|
||||
## 구현 체크리스트
|
||||
|
||||
- [ ] 선행 `01_socket_lifecycle` PASS complete evidence를 확인한다.
|
||||
- [ ] [API-1] socket queued dispatch와 invalid RunRequest failure 테스트/보강을 구현한다. 검증: `cd services/core && go test -count=1 ./...`
|
||||
- [ ] [API-2] socket terminal report와 active execution gate 테스트/보강을 구현한다. 검증: `cd services/core && go test -count=1 ./...`
|
||||
- [ ] runner socket smoke 관련 회귀를 확인한다. 검증: `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart`
|
||||
- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
|
||||
- [x] 선행 `01_socket_lifecycle` PASS complete evidence를 확인한다.
|
||||
- [x] [API-1] socket queued dispatch와 invalid RunRequest failure 테스트/보강을 구현한다. 검증: `cd services/core && go test -count=1 ./...`
|
||||
- [x] [API-2] socket terminal report와 active execution gate 테스트/보강을 구현한다. 검증: `cd services/core && go test -count=1 ./...`
|
||||
- [x] runner socket smoke 관련 회귀를 확인한다. 검증: `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart`
|
||||
- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
|
||||
|
||||
## 코드리뷰 전용 체크리스트
|
||||
|
||||
> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다.
|
||||
> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다.
|
||||
|
||||
- [ ] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다.
|
||||
- [ ] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다.
|
||||
- [ ] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G07_N.log`로 아카이브한다.
|
||||
- [ ] active `PLAN-*-G??.md`를 `plan_cloud_G07_M.log`로 아카이브한다.
|
||||
- [ ] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다.
|
||||
- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다.
|
||||
- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다.
|
||||
- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G07_N.log`로 아카이브한다.
|
||||
- [x] active `PLAN-*-G??.md`를 `plan_cloud_G07_M.log`로 아카이브한다.
|
||||
- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다.
|
||||
- [ ] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다.
|
||||
- [ ] PASS이면 active task 디렉터리 `agent-task/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/`를 `agent-task/archive/YYYY/MM/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다.
|
||||
- [ ] PASS이고 task group이 `m-<milestone-slug>`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다.
|
||||
- [ ] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-runner-proto-socket-transport-hardening/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다.
|
||||
- [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active plan/review 파일 또는 follow-up plan을 작성하고 `complete.log`를 작성하지 않는다.
|
||||
- [x] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active plan/review 파일 또는 follow-up plan을 작성하고 `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`를 남기지 않는다.
|
||||
|
||||
## 계획 대비 변경 사항
|
||||
|
||||
_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._
|
||||
- `server.go`와 `store.go` 수정 없음: 기존 `validTransitions` 테이블과 `handleExecutionReport`의 implicit guard가 이미 올바른 정책을 구현하고 있어 production 코드 변경 없이 테스트만 추가했다. explicit running check 추가는 계획에서 "필요 시"로 명시되었고, 현재 상태에서 테스트가 모두 통과되므로 불필요하다.
|
||||
- `server_test.go` 한 파일에 API-1과 API-2 모두 추가: 파일 분리 없이 helper 함수(`testServerWithStore`, `registerRunner`, `createQueuedJob`, `setupRunningExecution`)를 공유해 중복을 최소화했다.
|
||||
|
||||
## 주요 설계 결정
|
||||
|
||||
_구현 에이전트가 주요 설계 결정 사항을 기록한다._
|
||||
- **API-1 dispatch test**: `net.Pipe()`의 remote end를 goroutine으로 drain하여 `client.Send()` 블로킹 없이 `DispatchQueuedJobs()`를 동기 완료시켰다. 상태 전이(queued→running)는 Send() 완료 후 동기적으로 반영된다.
|
||||
- **API-1 send failure test**: remote end를 Close() 후 dispatch하여 `client.Send()` 오류를 유발했다. `claimNextQueuedJob`은 execution을 running으로 claim한 뒤 send가 실패하므로, 최종 상태는 execution+job 모두 failed임을 검증했다.
|
||||
- **API-2 guard test**: `handleExecutionReport` 내부 메서드를 직접 호출했다(white-box). socket transport를 통과하는 end-to-end보다 빠르고 결정적이며, 같은 패키지 내 테스트이므로 접근 가능하다. state guard는 `validTransitions` 테이블의 implicit check로 동작하며, queued/succeeded/cancelled 상태 모두 TransitionExecution 실패로 rejection된다.
|
||||
- **cancelled + terminal report 충돌 정책**: `CancelJobExecution` 후 terminal report를 시도하면 `TransitionExecution(execID, StateSucceeded)`가 "invalid transition from canceled to succeeded" 오류를 반환해 rejection된다. 현재 store 정책을 테스트로 고정했다.
|
||||
|
||||
## 사용자 리뷰 요청
|
||||
|
||||
|
|
@ -108,26 +112,39 @@ _기본값은 `없음`이다. 구현 중 새 결정이 필요해 보여도 직
|
|||
|
||||
## 검증 결과
|
||||
|
||||
_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._
|
||||
|
||||
### API-1 중간 검증
|
||||
```bash
|
||||
$ cd services/core && go test -count=1 ./...
|
||||
(output)
|
||||
? github.com/toki/oto/services/core/cmd/oto-core [no test files]
|
||||
ok github.com/toki/oto/services/core/internal/cicdstate 0.002s
|
||||
ok github.com/toki/oto/services/core/internal/httpserver 0.082s
|
||||
ok github.com/toki/oto/services/core/internal/runnerregistry 0.013s
|
||||
ok github.com/toki/oto/services/core/internal/runnersocket 0.049s
|
||||
? github.com/toki/oto/services/core/oto [no test files]
|
||||
```
|
||||
|
||||
### API-2 중간 검증
|
||||
```bash
|
||||
$ cd services/core && go test -count=1 ./...
|
||||
(output)
|
||||
? github.com/toki/oto/services/core/cmd/oto-core [no test files]
|
||||
ok github.com/toki/oto/services/core/internal/cicdstate 0.002s
|
||||
ok github.com/toki/oto/services/core/internal/httpserver 0.082s
|
||||
ok github.com/toki/oto/services/core/internal/runnerregistry 0.013s
|
||||
ok github.com/toki/oto/services/core/internal/runnersocket 0.049s
|
||||
? github.com/toki/oto/services/core/oto [no test files]
|
||||
```
|
||||
|
||||
### 최종 검증
|
||||
```bash
|
||||
$ cd services/core && go test -count=1 ./...
|
||||
(output)
|
||||
? github.com/toki/oto/services/core/cmd/oto-core [no test files]
|
||||
ok github.com/toki/oto/services/core/internal/cicdstate 0.002s
|
||||
ok github.com/toki/oto/services/core/internal/httpserver 0.082s
|
||||
ok github.com/toki/oto/services/core/internal/runnerregistry 0.013s
|
||||
ok github.com/toki/oto/services/core/internal/runnersocket 0.049s
|
||||
? github.com/toki/oto/services/core/oto [no test files]
|
||||
$ cd apps/runner && dart test test/oto_server_connection_smoke_test.dart
|
||||
(output)
|
||||
00:07 +11: All tests passed!
|
||||
```
|
||||
|
||||
---
|
||||
|
|
@ -135,3 +152,21 @@ $ cd apps/runner && dart test test/oto_server_connection_smoke_test.dart
|
|||
> **[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.
|
||||
|
||||
## 코드리뷰 결과
|
||||
|
||||
- 종합 판정: FAIL
|
||||
- 차원별 평가:
|
||||
- correctness: Pass
|
||||
- completeness: Fail
|
||||
- test coverage: Fail
|
||||
- API contract: Fail
|
||||
- code quality: Pass
|
||||
- plan deviation: Fail
|
||||
- verification trust: Pass
|
||||
- spec conformance: Fail
|
||||
- 발견된 문제:
|
||||
- Required: [server_test.go](/config/workspace/oto/services/core/internal/runnersocket/server_test.go:238)에서 `TestServerDispatchQueuedJobOverSocket`가 pipe remote end를 단순 drain만 하므로 `DispatchQueuedJobs()`가 실제 `RunRequest`를 socket으로 보냈는지, 그리고 payload의 runner/job/execution id와 run input이 맞는지 검증하지 못한다. 계획은 loopback/fake proto-socket client로 `RunRequest` 수신을 확인하라고 했고, SDD S03도 socket dispatcher가 idle runner에 `RunRequest`를 push한다는 evidence를 요구한다. 이 테스트는 state transition만으로도 통과하므로 socket dispatch 계약 회귀를 놓친다. 수정: 테스트에서 remote side가 정확히 하나의 `otopb.RunRequest`를 수신/파싱하고 runner id, job id, execution id, pipeline yaml/path, variables/command types를 assertion하도록 보강한다. 가능하면 queued job을 2개 두어 한 dispatch pass에서 하나만 claim되고 나머지는 queued로 남는지도 함께 고정한다.
|
||||
- Required: [server_test.go](/config/workspace/oto/services/core/internal/runnersocket/server_test.go:276)의 실패 케이스는 socket write failure만 검증하고, 계획/SDD가 요구한 invalid `RunRequest`/invalid run input 실패 경로를 검증하지 않는다. `claimNextQueuedJob()`는 invalid `RunInput`일 때 [server.go](/config/workspace/oto/services/core/internal/runnersocket/server.go:299)에서 send 전 오류를 반환하고 [server.go](/config/workspace/oto/services/core/internal/runnersocket/server.go:87) 경로로 job을 failed 처리하므로, 이 정책이 socket dispatch 테스트로 고정되어야 한다. 수정: `PipelineYAML`과 `PipelineYAMLPath`를 동시에 주거나 둘 다 비운 queued job을 만들고 `DispatchQueuedJobs()` 후 job이 failed가 되며 실행 생성 여부가 현재 정책과 일치함을 assertion한다.
|
||||
- Suggested: [server_test.go](/config/workspace/oto/services/core/internal/runnersocket/server_test.go:311)의 terminal report 테스트는 completed/failed state만 확인하고 계획에서 언급한 report message/log field 보존 증거를 남기지 않는다. 수정: success 또는 failure report에 `Message`를 넣고 `store.GetLogs(execID)`가 해당 line을 포함하는지 assertion한다.
|
||||
- 다음 단계: FAIL follow-up plan/review를 생성한다.
|
||||
|
|
@ -0,0 +1,190 @@
|
|||
<!-- task=m-runner-proto-socket-transport-hardening/02+01_dispatch_state plan=1 tag=REVIEW_API -->
|
||||
|
||||
# Code Review Reference - REVIEW_API
|
||||
|
||||
> **[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 selected SDD decision or selected Milestone `구현 잠금 > 결정 필요` item, fill `사용자 리뷰 요청` with evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. Environment/secret/service setup, generic scope conflicts, loop exhaustion, and evidence gaps that a follow-up agent can close are normal follow-up issues, not user-review blockers by themselves.
|
||||
> Do not ask the user directly, present choices in chat, or call `request_user_input` during implementation; record only the linked SDD/Milestone lock decision in `사용자 리뷰 요청` and stop for code-review.
|
||||
> 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-06-20
|
||||
task=m-runner-proto-socket-transport-hardening/02+01_dispatch_state, plan=1, tag=REVIEW_API
|
||||
|
||||
## Roadmap Targets
|
||||
|
||||
- Milestone: `agent-roadmap/phase/control-plane-product-surface/milestones/runner-proto-socket-transport-hardening.md`
|
||||
- Task ids:
|
||||
- `dispatch-state`: queued job dispatch, active execution gate, invalid RunRequest failure, terminal state transition을 socket transport 기준으로 검증한다.
|
||||
- Completion mode: check-on-pass
|
||||
- Split dependency:
|
||||
- `01_socket_lifecycle` PASS complete evidence가 있어야 구현을 시작한다.
|
||||
|
||||
## Spec Targets
|
||||
|
||||
- SDD: `agent-roadmap/sdd/control-plane-product-surface/runner-proto-socket-transport-hardening/SDD.md`
|
||||
- Acceptance scenarios:
|
||||
- `S03`: task=`dispatch-state`; evidence=`Socket RunRequest dispatch와 terminal report가 queued/running/completed/failed/cancelled 상태 전이를 검증한다.`
|
||||
- Completion mode: spec-check-on-pass
|
||||
|
||||
## Archive Evidence Snapshot
|
||||
|
||||
- Current archived plan: `agent-task/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/plan_cloud_G07_0.log`
|
||||
- Current archived review: `agent-task/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/code_review_cloud_G07_0.log`
|
||||
- Verdict: FAIL
|
||||
- Required summary:
|
||||
- `services/core/internal/runnersocket/server_test.go`의 dispatch success 테스트가 remote pipe를 drain만 해서 실제 `otopb.RunRequest` 수신/필드 검증을 하지 않는다.
|
||||
- send failure 테스트는 write failure만 검증하고 invalid `RunInput`/invalid `RunRequest` 실패 정책을 socket dispatch 경로에서 고정하지 않는다.
|
||||
- Suggested summary:
|
||||
- terminal report 테스트가 report message/log field 보존 증거를 남기지 않는다.
|
||||
- Affected files:
|
||||
- `services/core/internal/runnersocket/server_test.go`
|
||||
- 필요 시 `services/core/internal/runnersocket/server.go`
|
||||
- Verification evidence from failed loop:
|
||||
- `cd services/core && go test -count=1 ./...` PASS
|
||||
- `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart` PASS
|
||||
- `git diff --check` PASS
|
||||
- Roadmap/spec carryover:
|
||||
- Roadmap task `dispatch-state`와 SDD S03 completion target은 유지한다.
|
||||
- 선행 `01_socket_lifecycle` PASS evidence는 `agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/01_socket_lifecycle/complete.log`에서 확인되었다.
|
||||
- Allowed narrow archive reread:
|
||||
- 위 archived plan/review log와 선행 dependency `complete.log`만 필요 시 좁게 읽는다. `agent-task/archive/**` 전체 검색은 하지 않는다.
|
||||
|
||||
## 이 파일을 읽는 리뷰 에이전트에게
|
||||
|
||||
> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다.
|
||||
|
||||
각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요.
|
||||
리뷰 완료는 아래 순서까지 끝난 상태를 의미합니다.
|
||||
|
||||
1. 판정을 append한다.
|
||||
2. `CODE_REVIEW-local-G06.md` -> `code_review_local_G06_N.log`, `PLAN-local-G06.md` -> `plan_local_G06_M.log`로 아카이브한다.
|
||||
3. PASS이면 `complete.log` 작성 후 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/`로 이동한다. WARN/FAIL이면 user-review gate를 확인한 뒤 다음 active plan/review 파일 또는 `USER_REVIEW.md`를 작성한다.
|
||||
4. PASS이고 task group이 `m-<milestone-slug>`이면 완료 이벤트 메타데이터를 보고한다. roadmap 상태 체크와 `update-roadmap` 호출은 런타임 책임이다.
|
||||
5. 적용 가능한 `코드리뷰 전용 체크리스트` 항목을 최종 `.log` 위치에서 체크한 뒤 보고한다.
|
||||
|
||||
---
|
||||
|
||||
## 구현 항목별 완료 여부
|
||||
|
||||
| 항목 | 완료 여부 |
|
||||
|------|---------|
|
||||
| [REVIEW_API-1] socket dispatch `RunRequest` payload와 single-claim 증거 | [x] |
|
||||
| [REVIEW_API-2] invalid run input failed 정책과 terminal report message/log 증거 | [x] |
|
||||
|
||||
## 구현 체크리스트
|
||||
|
||||
- [x] [REVIEW_API-1] socket dispatch `RunRequest` payload와 single-claim 증거를 보강한다. 검증: `cd services/core && go test -count=1 ./internal/runnersocket`
|
||||
- [x] [REVIEW_API-2] invalid run input failed 정책과 terminal report message/log 증거를 보강한다. 검증: `cd services/core && go test -count=1 ./internal/runnersocket`
|
||||
- [x] 전체 core/runner smoke 회귀를 확인한다. 검증: `cd services/core && go test -count=1 ./...` 및 `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart`
|
||||
- [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_local_G06_N.log`로 아카이브한다.
|
||||
- [x] active `PLAN-*-G??.md`를 `plan_local_G06_M.log`로 아카이브한다.
|
||||
- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다.
|
||||
- [x] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다.
|
||||
- [x] PASS이면 active task 디렉터리 `agent-task/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/`를 `agent-task/archive/YYYY/MM/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다.
|
||||
- [x] PASS이고 task group이 `m-<milestone-slug>`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다.
|
||||
- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-runner-proto-socket-transport-hardening/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다.
|
||||
- [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active plan/review 파일 또는 follow-up plan을 작성하고 `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`를 남기지 않는다.
|
||||
|
||||
## 계획 대비 변경 사항
|
||||
|
||||
- 계획: `TestServerDispatchQueuedJobOverSocket`를 보강한다. 실제: 기존 테스트를 인라인 job 생성 + `pipeClientPair` 사용으로 전면 교체해 payload 검증 포함. `createQueuedJob` 헬퍼 대신 Variables/CommandTypes를 포함한 `store.CreateJob`을 직접 호출. 별도 테스트 함수를 추가하지 않고 기존 함수를 강화함으로써 중복 제거.
|
||||
- 계획: "기존 `proto_socket` helper로 파싱하기 어렵다면 package-local helper를 작게 만든다" → `AddListenerTyped[*otopb.RunRequest]`로 충분히 파싱 가능하므로 별도 helper 없이 `pipeClientPair` + `runnerParserMap`으로 해결.
|
||||
|
||||
## 주요 설계 결정
|
||||
|
||||
- `runnerParserMap()`: runner 수신 방향에 필요한 `RunRequest`만 포함. `CancelRunRequest` 등 미사용 타입은 추가하지 않음.
|
||||
- `pipeClientPair()`: `net.Pipe()` 양 끝을 각각 `TcpClient`로 래핑. `intervalSec=0`으로 heartbeat 비활성화해 타이머 노이즈 없음. server-side는 기존 `parserMap()`, runner-side는 `runnerParserMap()` 사용.
|
||||
- invalid run input: `validateRunInput` 실패 시 `execID=""`이므로 execution이 생성되지 않는다는 기존 정책을 두 fixture(both-empty, both-set)로 고정. production 코드 변경 없이 테스트만 추가.
|
||||
- terminal report message: `handleExecutionReport`가 `Message != ""` 이면 `store.AppendLog`를 호출하는 기존 동작을 success/failure 두 케이스로 고정.
|
||||
|
||||
## 사용자 리뷰 요청
|
||||
|
||||
_기본값은 `없음`이다. 구현 중 새 결정이 필요해 보여도 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 이 섹션은 선택된 SDD 결정 또는 선택된 Milestone `구현 잠금 > 결정 필요` 항목이 실구현을 차단할 때만 채운다. 외부 환경/secret/서비스 준비, 검증 증거 공백, 반복 실패, 일반 범위 조정은 사용자 리뷰 요청이 아니며 `검증 결과`, `계획 대비 변경 사항`, 또는 code-review의 일반 follow-up plan으로 처리한다._
|
||||
|
||||
- 상태: 없음
|
||||
- 사유 유형: 없음
|
||||
- 연결 대상: 없음
|
||||
- 결정 필요: 없음
|
||||
- 차단 근거: 없음
|
||||
- 실행한 검증/명령: 없음
|
||||
- 자동 후속 불가 이유: 없음
|
||||
- 재개 조건: 없음
|
||||
|
||||
## 리뷰어를 위한 체크포인트
|
||||
|
||||
- dispatch success 테스트가 실제 socket `RunRequest`를 수신/파싱하고 payload를 검증하는지 확인한다.
|
||||
- 한 runner의 한 dispatch pass가 queued job 하나만 claim하고 나머지 queued job을 남기는지 확인한다.
|
||||
- invalid run input 실패 정책이 socket dispatch 경로에서 테스트로 고정되었는지 확인한다.
|
||||
- terminal report message가 execution log로 보존되는지 확인한다.
|
||||
|
||||
## 검증 결과
|
||||
|
||||
_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._
|
||||
|
||||
필수 규칙:
|
||||
- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다.
|
||||
- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다.
|
||||
- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다.
|
||||
- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다.
|
||||
|
||||
### REVIEW_API-1 중간 검증
|
||||
```bash
|
||||
$ cd services/core && go test -count=1 ./internal/runnersocket
|
||||
ok github.com/toki/oto/services/core/internal/runnersocket 0.046s
|
||||
```
|
||||
|
||||
### REVIEW_API-2 중간 검증
|
||||
```bash
|
||||
$ cd services/core && go test -count=1 ./internal/runnersocket
|
||||
ok github.com/toki/oto/services/core/internal/runnersocket 0.046s
|
||||
```
|
||||
|
||||
### 최종 검증
|
||||
```bash
|
||||
$ cd services/core && go test -count=1 ./...
|
||||
? github.com/toki/oto/services/core/cmd/oto-core [no test files]
|
||||
ok github.com/toki/oto/services/core/internal/cicdstate 0.002s
|
||||
ok github.com/toki/oto/services/core/internal/httpserver 0.066s
|
||||
ok github.com/toki/oto/services/core/internal/runnerregistry 0.006s
|
||||
ok github.com/toki/oto/services/core/internal/runnersocket 0.048s
|
||||
? github.com/toki/oto/services/core/oto [no test files]
|
||||
$ cd apps/runner && dart test test/oto_server_connection_smoke_test.dart
|
||||
00:08 +11: All tests passed!
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
> **[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.
|
||||
|
||||
## 코드리뷰 결과
|
||||
|
||||
- 종합 판정: PASS
|
||||
- 차원별 평가:
|
||||
- correctness: Pass
|
||||
- completeness: Pass
|
||||
- test coverage: Pass
|
||||
- API contract: Pass
|
||||
- code quality: Pass
|
||||
- plan deviation: Pass
|
||||
- verification trust: Pass
|
||||
- spec conformance: Pass
|
||||
- 발견된 문제: 없음
|
||||
- 다음 단계: PASS complete.log 작성 후 task directory를 archive로 이동한다.
|
||||
|
|
@ -0,0 +1,51 @@
|
|||
# Complete - m-runner-proto-socket-transport-hardening/02+01_dispatch_state
|
||||
|
||||
## 완료 일시
|
||||
|
||||
2026-06-20
|
||||
|
||||
## 요약
|
||||
|
||||
Runner proto-socket dispatch/state 보강을 2회 리뷰 루프로 완료했다. 최종 판정은 PASS다.
|
||||
|
||||
## 루프 이력
|
||||
|
||||
| Plan | Review | Verdict | 메모 |
|
||||
|------|--------|---------|------|
|
||||
| `plan_cloud_G07_0.log` | `code_review_cloud_G07_0.log` | FAIL | socket `RunRequest` payload 수신 검증, invalid input 실패 정책, terminal report message log 증거가 부족해 follow-up 필요 |
|
||||
| `plan_local_G06_1.log` | `code_review_local_G06_1.log` | PASS | socket payload, single-claim, invalid input failure, terminal report message log 보존 테스트를 보강 |
|
||||
|
||||
## 구현/정리 내용
|
||||
|
||||
- `services/core/internal/runnersocket/server_test.go`에 runner-side proto-socket parser를 둔 `pipeClientPair` helper를 추가해 실제 `RunRequest` payload를 수신/검증했다.
|
||||
- dispatch success, one-job-per-runner claim, send failure, invalid run input failure, terminal report state guard, terminal report message log 보존 테스트를 socket dispatch/state evidence로 정리했다.
|
||||
- production code 변경 없이 기존 `DispatchQueuedJobs`, `claimNextQueuedJob`, `handleExecutionReport` 정책을 테스트로 고정했다.
|
||||
|
||||
## 최종 검증
|
||||
|
||||
- `cd services/core && go test -count=1 ./internal/runnersocket` - PASS; `ok github.com/toki/oto/services/core/internal/runnersocket 0.049s`
|
||||
- `cd services/core && go test -count=1 ./...` - PASS; core packages all passed, including `internal/runnersocket`
|
||||
- `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart` - PASS; `00:12 +11: All tests passed!`
|
||||
- `git diff --check` - PASS; no whitespace errors
|
||||
|
||||
## Roadmap Completion
|
||||
|
||||
- Milestone: `agent-roadmap/phase/control-plane-product-surface/milestones/runner-proto-socket-transport-hardening.md`
|
||||
- Completed task ids:
|
||||
- `dispatch-state`: PASS; evidence=`agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/plan_local_G06_1.log`, `agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/code_review_local_G06_1.log`; verification=`cd services/core && go test -count=1 ./internal/runnersocket`, `cd services/core && go test -count=1 ./...`, `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart`, `git diff --check`
|
||||
- Not completed task ids: 없음
|
||||
|
||||
## Spec Completion
|
||||
|
||||
- SDD: `agent-roadmap/sdd/control-plane-product-surface/runner-proto-socket-transport-hardening/SDD.md`
|
||||
- Completed scenario ids:
|
||||
- `S03`: PASS; task=`dispatch-state`; evidence=`agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/plan_local_G06_1.log`, `agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/code_review_local_G06_1.log`; verification=`cd services/core && go test -count=1 ./internal/runnersocket`, `cd services/core && go test -count=1 ./...`, `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart`, `git diff --check`
|
||||
- Not completed scenario ids: 없음
|
||||
|
||||
## 잔여 Nit
|
||||
|
||||
- 없음
|
||||
|
||||
## 후속 작업
|
||||
|
||||
- 없음
|
||||
|
|
@ -0,0 +1,109 @@
|
|||
<!-- task=m-runner-proto-socket-transport-hardening/02+01_dispatch_state plan=1 tag=REVIEW_API -->
|
||||
|
||||
# Plan - REVIEW_API
|
||||
|
||||
## 이 파일을 읽는 구현 에이전트에게
|
||||
|
||||
`CODE_REVIEW-local-G06.md`의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채우는 것까지가 구현입니다. 검증을 실행하고 실제 stdout/stderr를 붙인 뒤 active 파일을 그대로 두고 review ready로 보고하세요. 선택된 SDD 결정 또는 Milestone `구현 잠금 > 결정 필요` 항목이 막으면 review stub의 `사용자 리뷰 요청` 섹션을 채우고 멈추며, 직접 사용자에게 질문하거나 `USER_REVIEW.md`, `complete.log`, archive를 만들지 않습니다. 환경/secret/서비스 문제나 증거 공백은 일반 follow-up 대상으로 기록합니다.
|
||||
|
||||
## Roadmap Targets
|
||||
|
||||
- Milestone: `agent-roadmap/phase/control-plane-product-surface/milestones/runner-proto-socket-transport-hardening.md`
|
||||
- Task ids:
|
||||
- `dispatch-state`: queued job dispatch, active execution gate, invalid RunRequest failure, terminal state transition을 socket transport 기준으로 검증한다.
|
||||
- Completion mode: check-on-pass
|
||||
- Split dependency:
|
||||
- `01_socket_lifecycle` PASS complete evidence가 있어야 구현을 시작한다.
|
||||
|
||||
## Spec Targets
|
||||
|
||||
- SDD: `agent-roadmap/sdd/control-plane-product-surface/runner-proto-socket-transport-hardening/SDD.md`
|
||||
- Acceptance scenarios:
|
||||
- `S03`: task=`dispatch-state`; evidence=`Socket RunRequest dispatch와 terminal report가 queued/running/completed/failed/cancelled 상태 전이를 검증한다.`
|
||||
- Completion mode: spec-check-on-pass
|
||||
|
||||
## Archive Evidence Snapshot
|
||||
|
||||
- Current archived plan: `agent-task/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/plan_cloud_G07_0.log`
|
||||
- Current archived review: `agent-task/m-runner-proto-socket-transport-hardening/02+01_dispatch_state/code_review_cloud_G07_0.log`
|
||||
- Verdict: FAIL
|
||||
- Required summary:
|
||||
- `services/core/internal/runnersocket/server_test.go`의 dispatch success 테스트가 remote pipe를 drain만 해서 실제 `otopb.RunRequest` 수신/필드 검증을 하지 않는다.
|
||||
- send failure 테스트는 write failure만 검증하고 invalid `RunInput`/invalid `RunRequest` 실패 정책을 socket dispatch 경로에서 고정하지 않는다.
|
||||
- Suggested summary:
|
||||
- terminal report 테스트가 report message/log field 보존 증거를 남기지 않는다.
|
||||
- Affected files:
|
||||
- `services/core/internal/runnersocket/server_test.go`
|
||||
- 필요 시 `services/core/internal/runnersocket/server.go`
|
||||
- Verification evidence from failed loop:
|
||||
- `cd services/core && go test -count=1 ./...` PASS
|
||||
- `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart` PASS
|
||||
- `git diff --check` PASS
|
||||
- Roadmap/spec carryover:
|
||||
- Roadmap task `dispatch-state`와 SDD S03 completion target은 유지한다.
|
||||
- 선행 `01_socket_lifecycle` PASS evidence는 `agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/01_socket_lifecycle/complete.log`에서 확인되었다.
|
||||
- Allowed narrow archive reread:
|
||||
- 위 archived plan/review log와 선행 dependency `complete.log`만 필요 시 좁게 읽는다. `agent-task/archive/**` 전체 검색은 하지 않는다.
|
||||
|
||||
## 배경
|
||||
|
||||
이 follow-up은 production behavior 변경이 아니라 SDD S03과 `dispatch-state` 완료 근거를 충족하기 위한 테스트 증거 보강이다. 기존 구현은 state transition test를 추가했지만, socket payload 자체와 invalid input 실패 정책을 충분히 관찰하지 못했다.
|
||||
|
||||
## 범위 결정 근거
|
||||
|
||||
- 포함: `services/core/internal/runnersocket/server_test.go`의 Go test 보강.
|
||||
- 필요 시 포함: 테스트 가능성을 위해 아주 작은 helper 추가 또는 `server.go`의 오류 정책 명시 보강.
|
||||
- 제외: cancel/status/self-update operator action, HTTP compatibility boundary, Dart runner implementation 변경.
|
||||
- 사용자 리뷰 요청 금지: 이번 범위는 SDD/Milestone 미정 결정이 아니라 테스트 증거 공백이므로 파일 기반 follow-up으로 해결한다.
|
||||
|
||||
## 구현 체크리스트
|
||||
|
||||
- [ ] [REVIEW_API-1] socket dispatch `RunRequest` payload와 single-claim 증거를 보강한다. 검증: `cd services/core && go test -count=1 ./internal/runnersocket`
|
||||
- [ ] [REVIEW_API-2] invalid run input failed 정책과 terminal report message/log 증거를 보강한다. 검증: `cd services/core && go test -count=1 ./internal/runnersocket`
|
||||
- [ ] 전체 core/runner smoke 회귀를 확인한다. 검증: `cd services/core && go test -count=1 ./...` 및 `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart`
|
||||
- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
|
||||
|
||||
### [REVIEW_API-1] socket dispatch `RunRequest` payload와 single-claim 증거
|
||||
|
||||
문제: `TestServerDispatchQueuedJobOverSocket`는 `net.Pipe()`의 remote side를 drain만 하므로 `DispatchQueuedJobs()`가 실제 runner-facing `RunRequest`를 push했는지 검증하지 않는다. state transition만 확인하면 socket payload regression을 놓칠 수 있다.
|
||||
|
||||
해결 방법:
|
||||
|
||||
- remote side에서 proto-socket frame을 실제로 읽어 `otopb.RunRequest`를 수신했음을 검증한다. 기존 `proto_socket` helper로 파싱하기 어렵다면 package-local helper를 작게 만들되, 테스트 전용으로 제한한다.
|
||||
- 수신한 `RunRequest`의 `RunnerId`, `JobId`, `ExecutionId`, `PipelineYaml` 또는 `PipelineYamlPath`, `Variables`, `CommandTypes`를 store state와 대조한다.
|
||||
- queued job을 2개 준비하고 한 runner에 대해 `DispatchQueuedJobs()` 한 번이 하나만 claim하는지 확인한다. 첫 job은 running, 두 번째 job은 queued로 남아야 한다.
|
||||
|
||||
중간 검증:
|
||||
|
||||
```bash
|
||||
cd services/core && go test -count=1 ./internal/runnersocket
|
||||
```
|
||||
|
||||
기대 결과: `internal/runnersocket` package 통과.
|
||||
|
||||
### [REVIEW_API-2] invalid run input failed 정책과 terminal report message/log 증거
|
||||
|
||||
문제: 기존 실패 테스트는 socket write failure만 다루고, SDD S03의 invalid input -> failed 증거를 socket dispatch 경로에서 남기지 않는다. terminal report 테스트도 state만 확인하고 report message가 log로 보존되는지 확인하지 않는다.
|
||||
|
||||
해결 방법:
|
||||
|
||||
- invalid run input fixture를 만든다. 예: `PipelineYAML`과 `PipelineYAMLPath`를 동시에 설정하거나 둘 다 비운 queued job.
|
||||
- `DispatchQueuedJobs()` 후 현재 정책을 명확히 고정한다. 현 production 코드가 유지된다면 invalid input은 send 전 실패하므로 job은 failed가 되고 execution은 생성되지 않는 것이 기대값이다. 정책을 바꾸는 경우에는 execution failed까지 코드와 테스트를 함께 맞춘다.
|
||||
- terminal report success 또는 failure case에 `Message`를 넣고 `store.GetLogs(execID)`가 해당 message를 포함하는지 검증한다.
|
||||
|
||||
중간 검증:
|
||||
|
||||
```bash
|
||||
cd services/core && go test -count=1 ./internal/runnersocket
|
||||
```
|
||||
|
||||
기대 결과: `internal/runnersocket` package 통과.
|
||||
|
||||
## 최종 검증
|
||||
|
||||
```bash
|
||||
cd services/core && go test -count=1 ./...
|
||||
cd apps/runner && dart test test/oto_server_connection_smoke_test.dart
|
||||
```
|
||||
|
||||
기대 결과: Go tests 통과, runner socket smoke 통과. 모든 코드 변경 완료 후 반드시 `CODE_REVIEW-local-G06.md`의 구현 에이전트 소유 섹션을 채운다.
|
||||
|
|
@ -52,42 +52,47 @@ task=m-runner-proto-socket-transport-hardening/03+01,02_runner_actions, plan=0,
|
|||
|
||||
| 항목 | 완료 여부 |
|
||||
|------|---------|
|
||||
| [API-1] cancel/status action의 socket-first 방향과 상태 guard | [ ] |
|
||||
| [API-2] runner-side cancel handling 또는 compatibility 제외 | [ ] |
|
||||
| [API-3] self-update action 방향 명확화 | [ ] |
|
||||
| [API-1] cancel/status action의 socket-first 방향과 상태 guard | [x] |
|
||||
| [API-2] runner-side cancel handling 또는 compatibility 제외 | [x] |
|
||||
| [API-3] self-update action 방향 명확화 | [x] |
|
||||
|
||||
## 구현 체크리스트
|
||||
|
||||
- [ ] 선행 `01_socket_lifecycle`, `02+01_dispatch_state` PASS complete evidence를 확인한다.
|
||||
- [ ] [API-1] cancel/status action의 socket-first 방향과 상태 guard를 구현/검증한다. 검증: `cd services/core && go test -count=1 ./...`
|
||||
- [ ] [API-2] runner-side cancel handling 또는 명시적 compatibility 제외를 구현/검증한다. 검증: `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart`
|
||||
- [ ] [API-3] self-update action을 socket-first 구현 또는 compatibility 제외 범위로 명확히 한다. 검증: `cd apps/runner && dart analyze`
|
||||
- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
|
||||
- [x] 선행 `01_socket_lifecycle`, `02+01_dispatch_state` PASS complete evidence를 확인한다.
|
||||
- [x] [API-1] cancel/status action의 socket-first 방향과 상태 guard를 구현/검증한다. 검증: `cd services/core && go test -count=1 ./...`
|
||||
- [x] [API-2] runner-side cancel handling 또는 명시적 compatibility 제외를 구현/검증한다. 검증: `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart`
|
||||
- [x] [API-3] self-update action을 socket-first 구현 또는 compatibility 제외 범위로 명확히 한다. 검증: `cd apps/runner && dart analyze`
|
||||
- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
|
||||
|
||||
## 코드리뷰 전용 체크리스트
|
||||
|
||||
> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다.
|
||||
> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다.
|
||||
|
||||
- [ ] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다.
|
||||
- [ ] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다.
|
||||
- [ ] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G07_N.log`로 아카이브한다.
|
||||
- [ ] active `PLAN-*-G??.md`를 `plan_cloud_G07_M.log`로 아카이브한다.
|
||||
- [ ] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다.
|
||||
- [ ] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다.
|
||||
- [ ] PASS이면 active task 디렉터리 `agent-task/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/`를 `agent-task/archive/YYYY/MM/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다.
|
||||
- [ ] PASS이고 task group이 `m-<milestone-slug>`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다.
|
||||
- [ ] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-runner-proto-socket-transport-hardening/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다.
|
||||
- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다.
|
||||
- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다.
|
||||
- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G07_N.log`로 아카이브한다.
|
||||
- [x] active `PLAN-*-G??.md`를 `plan_cloud_G07_M.log`로 아카이브한다.
|
||||
- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다.
|
||||
- [x] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다.
|
||||
- [x] PASS이면 active task 디렉터리 `agent-task/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/`를 `agent-task/archive/YYYY/MM/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다.
|
||||
- [x] PASS이고 task group이 `m-<milestone-slug>`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다.
|
||||
- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-runner-proto-socket-transport-hardening/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다.
|
||||
- [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active plan/review 파일 또는 follow-up plan을 작성하고 `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`를 남기지 않는다.
|
||||
|
||||
## 계획 대비 변경 사항
|
||||
|
||||
_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._
|
||||
- `CancelRunRequest` runner-side 수신은 실제 프로세스 중단까지 확장하지 않고 socket stream 수신과 log 처리로 제한했다. 현재 `RemoteRunExecutor`/`Application.build()`에는 cooperative cancel hook이 없어서 실행 중단은 후속 범위로 남기고, 이번 작업은 server store cancel과 late terminal report guard를 source of truth로 둔다.
|
||||
- socket cancel smoke는 처음 HTTP claim으로 running execution을 만들려 했으나, socket runner 연결 상태에서는 job 생성 직후 server가 자동 dispatch하는 것이 정상 동작이라 `runRequests` stream에서 server-assigned execution id를 받아 cancel하는 방식으로 수정했다.
|
||||
- self-update는 socket command로 구현하지 않고 HTTP compatibility action으로 명시했다. 응답 메시지와 Dart client 주석에서 idle accept/active defer만 이번 범위임을 드러냈다.
|
||||
|
||||
## 주요 설계 결정
|
||||
|
||||
_구현 에이전트가 주요 설계 결정 사항을 기록한다._
|
||||
- operator cancel HTTP action은 store를 `canceled` terminal state로 전환한 뒤 runner socket에 best-effort `CancelRunRequest`를 보낸다. socket send 실패는 cancel 응답 성공 여부를 뒤집지 않으며, terminal conflict는 store transition guard가 막는다.
|
||||
- status는 runner socket command가 아니라 registry/store read query로 유지한다. 테스트에서 status path가 dispatcher를 호출하지 않음을 고정했다.
|
||||
- Dart push session은 `runRequests`와 별도 `cancelRequests` stream을 노출한다. 기본 agent loop는 cancel 수신을 log로 남기고, terminal conflict 처리는 server state guard에 맡긴다.
|
||||
- self-update는 `SelfUpdateRequest` proto가 남아 있어도 이번 Milestone에서는 socket dispatch 대상이 아니다. HTTP compatibility response가 idle accepted 또는 active deferred 결과만 반환한다.
|
||||
|
||||
## 사용자 리뷰 요청
|
||||
|
||||
|
|
@ -115,29 +120,54 @@ _구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후
|
|||
### API-1 중간 검증
|
||||
```bash
|
||||
$ cd services/core && go test -count=1 ./...
|
||||
(output)
|
||||
? github.com/toki/oto/services/core/cmd/oto-core [no test files]
|
||||
ok github.com/toki/oto/services/core/internal/cicdstate 0.002s
|
||||
ok github.com/toki/oto/services/core/internal/httpserver 0.082s
|
||||
ok github.com/toki/oto/services/core/internal/runnerregistry 0.003s
|
||||
ok github.com/toki/oto/services/core/internal/runnersocket 0.048s
|
||||
? github.com/toki/oto/services/core/oto [no test files]
|
||||
```
|
||||
|
||||
### API-2 중간 검증
|
||||
```bash
|
||||
$ cd apps/runner && dart test test/oto_server_connection_smoke_test.dart
|
||||
(output)
|
||||
00:00 +0: loading test/oto_server_connection_smoke_test.dart
|
||||
00:00 +0: OTO Dart runner registers with Go OTO Server, goes online, and disconnects on close
|
||||
...
|
||||
00:10 +12: OtoServerSocketRegistrationClient socket session close is idempotent after server disconnect
|
||||
00:10 +12: All tests passed!
|
||||
```
|
||||
|
||||
### API-3 중간 검증
|
||||
```bash
|
||||
$ cd apps/runner && dart analyze
|
||||
(output)
|
||||
Analyzing runner...
|
||||
No issues found!
|
||||
```
|
||||
|
||||
### 최종 검증
|
||||
```bash
|
||||
$ cd services/core && go test -count=1 ./...
|
||||
(output)
|
||||
? github.com/toki/oto/services/core/cmd/oto-core [no test files]
|
||||
ok github.com/toki/oto/services/core/internal/cicdstate 0.002s
|
||||
ok github.com/toki/oto/services/core/internal/httpserver 0.082s
|
||||
ok github.com/toki/oto/services/core/internal/runnerregistry 0.003s
|
||||
ok github.com/toki/oto/services/core/internal/runnersocket 0.048s
|
||||
? github.com/toki/oto/services/core/oto [no test files]
|
||||
|
||||
$ cd apps/runner && dart test test/oto_server_connection_smoke_test.dart
|
||||
(output)
|
||||
00:00 +0: loading test/oto_server_connection_smoke_test.dart
|
||||
00:00 +0: OTO Dart runner registers with Go OTO Server, goes online, and disconnects on close
|
||||
...
|
||||
00:10 +12: OtoServerSocketRegistrationClient socket session close is idempotent after server disconnect
|
||||
00:10 +12: All tests passed!
|
||||
|
||||
$ cd apps/runner && dart analyze
|
||||
(output)
|
||||
Analyzing runner...
|
||||
No issues found!
|
||||
|
||||
$ git diff --check
|
||||
(no output; exit 0)
|
||||
```
|
||||
|
||||
---
|
||||
|
|
@ -145,3 +175,18 @@ $ cd apps/runner && dart analyze
|
|||
> **[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.
|
||||
|
||||
## 코드리뷰 결과
|
||||
|
||||
- 종합 판정: PASS
|
||||
- 차원별 평가:
|
||||
- correctness: Pass
|
||||
- completeness: Pass
|
||||
- test coverage: Pass
|
||||
- API contract: Pass
|
||||
- code quality: Pass
|
||||
- plan deviation: Pass
|
||||
- verification trust: Pass
|
||||
- spec conformance: Pass
|
||||
- 발견된 문제: 없음
|
||||
- 다음 단계: PASS이므로 active plan/review 파일을 로그로 아카이브하고 `complete.log`를 작성한 뒤 task 디렉터리를 archive로 이동한다.
|
||||
|
|
@ -0,0 +1,51 @@
|
|||
# Complete - m-runner-proto-socket-transport-hardening/03+01,02_runner_actions
|
||||
|
||||
## 완료 일시
|
||||
|
||||
2026-06-20
|
||||
|
||||
## 요약
|
||||
|
||||
Runner action 방향성 정리 작업을 1회 리뷰 루프로 완료했다. 최종 판정은 PASS다.
|
||||
|
||||
## 루프 이력
|
||||
|
||||
| Plan | Review | Verdict | 메모 |
|
||||
|------|--------|---------|------|
|
||||
| `plan_cloud_G07_0.log` | `code_review_cloud_G07_0.log` | PASS | cancel/status/self-update의 socket-first 및 HTTP compatibility 경계가 코드, 테스트, 검증 결과로 충족됨 |
|
||||
|
||||
## 구현/정리 내용
|
||||
|
||||
- operator cancel HTTP action이 store를 terminal canceled로 전환한 뒤 runner socket으로 best-effort `CancelRunRequest`를 전달하도록 검증했다.
|
||||
- cancel 이후 late terminal report가 canceled 상태를 뒤집지 못하도록 state guard를 테스트로 고정했다.
|
||||
- status는 socket command가 아닌 registry/store read query로 유지하고, self-update는 HTTP compatibility action으로 명시했다.
|
||||
- Dart socket push session에 `cancelRequests` stream을 추가해 socket cancel 수신을 smoke로 검증했다.
|
||||
|
||||
## 최종 검증
|
||||
|
||||
- `cd services/core && go test -count=1 ./...` - PASS; core packages all passed.
|
||||
- `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart` - PASS; `+12: All tests passed!`
|
||||
- `cd apps/runner && dart analyze` - PASS; `No issues found!`
|
||||
- `git diff --check` - PASS; no output.
|
||||
|
||||
## Roadmap Completion
|
||||
|
||||
- Milestone: `agent-roadmap/phase/control-plane-product-surface/milestones/runner-proto-socket-transport-hardening.md`
|
||||
- Completed task ids:
|
||||
- `runner-actions`: PASS; evidence=`agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/plan_cloud_G07_0.log`, `agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/code_review_cloud_G07_0.log`; verification=`cd services/core && go test -count=1 ./...`, `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart`, `cd apps/runner && dart analyze`, `git diff --check`
|
||||
- Not completed task ids: 없음
|
||||
|
||||
## Spec Completion
|
||||
|
||||
- SDD: `agent-roadmap/sdd/control-plane-product-surface/runner-proto-socket-transport-hardening/SDD.md`
|
||||
- Completed scenario ids:
|
||||
- `S04`: PASS; task=`runner-actions`; evidence=`agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/plan_cloud_G07_0.log`, `agent-task/archive/2026/06/m-runner-proto-socket-transport-hardening/03+01,02_runner_actions/code_review_cloud_G07_0.log`; verification=`cd services/core && go test -count=1 ./...`, `cd apps/runner && dart test test/oto_server_connection_smoke_test.dart`, `cd apps/runner && dart analyze`, `git diff --check`
|
||||
- Not completed scenario ids: 없음
|
||||
|
||||
## 잔여 Nit
|
||||
|
||||
- 없음
|
||||
|
||||
## 후속 작업
|
||||
|
||||
- 없음
|
||||
|
|
@ -106,7 +106,7 @@ class DefaultAgentRunner implements AgentRunner {
|
|||
);
|
||||
|
||||
var runChain = Future<void>.value();
|
||||
final subscription = session.runRequests.listen((request) {
|
||||
final runSubscription = session.runRequests.listen((request) {
|
||||
runChain = runChain.then((_) async {
|
||||
if (request.jobId.isEmpty || request.executionId.isEmpty) {
|
||||
_log('Received RunRequest without job_id or execution_id; skipping.');
|
||||
|
|
@ -125,10 +125,22 @@ class DefaultAgentRunner implements AgentRunner {
|
|||
}
|
||||
});
|
||||
});
|
||||
final cancelSubscription = session.cancelRequests.listen((request) {
|
||||
final executionId = request.executionId;
|
||||
if (executionId.isEmpty) {
|
||||
_log('Received CancelRunRequest without execution_id; skipping.');
|
||||
return;
|
||||
}
|
||||
_log(
|
||||
'Received cancel for execution $executionId; '
|
||||
'server state guard owns terminal conflict handling.',
|
||||
);
|
||||
});
|
||||
|
||||
_log('Agent connected via proto-socket. Waiting for pushed jobs...');
|
||||
await shutdownFuture;
|
||||
await subscription.cancel();
|
||||
await runSubscription.cancel();
|
||||
await cancelSubscription.cancel();
|
||||
await runChain;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -343,6 +343,10 @@ class OtoServerJobClient implements RemoteJobReporter {
|
|||
return RunnerStatusResult.fromJson(response);
|
||||
}
|
||||
|
||||
/// Requests the operator-facing HTTP compatibility self-update action.
|
||||
///
|
||||
/// This milestone does not dispatch a socket self-update command; the server
|
||||
/// either accepts an idle compatibility request or defers while a run is active.
|
||||
Future<RunnerSelfUpdateResult> requestSelfUpdate({
|
||||
required String version,
|
||||
required String downloadUrl,
|
||||
|
|
|
|||
|
|
@ -103,6 +103,7 @@ abstract class OtoServerJobSession implements AgentSession {
|
|||
abstract class OtoServerPushJobSession implements AgentSession {
|
||||
RemoteJobReporter get jobReporter;
|
||||
Stream<oto.RunRequest> get runRequests;
|
||||
Stream<oto.CancelRunRequest> get cancelRequests;
|
||||
}
|
||||
|
||||
abstract class RegistrationClient {
|
||||
|
|
@ -319,6 +320,8 @@ class _OtoServerSocketAgentSession implements OtoServerPushJobSession {
|
|||
final String _runnerId;
|
||||
final StreamController<oto.RunRequest> _runRequests =
|
||||
StreamController<oto.RunRequest>();
|
||||
final StreamController<oto.CancelRunRequest> _cancelRequests =
|
||||
StreamController<oto.CancelRunRequest>();
|
||||
Timer? _heartbeatTimer;
|
||||
bool _closed = false;
|
||||
|
||||
|
|
@ -331,6 +334,9 @@ class _OtoServerSocketAgentSession implements OtoServerPushJobSession {
|
|||
@override
|
||||
Stream<oto.RunRequest> get runRequests => _runRequests.stream;
|
||||
|
||||
@override
|
||||
Stream<oto.CancelRunRequest> get cancelRequests => _cancelRequests.stream;
|
||||
|
||||
_OtoServerSocketAgentSession(
|
||||
this.result,
|
||||
this._client,
|
||||
|
|
@ -346,6 +352,11 @@ class _OtoServerSocketAgentSession implements OtoServerPushJobSession {
|
|||
_runRequests.add(request);
|
||||
}
|
||||
});
|
||||
_client.addListener<oto.CancelRunRequest>((request) {
|
||||
if (!_cancelRequests.isClosed) {
|
||||
_cancelRequests.add(request);
|
||||
}
|
||||
});
|
||||
if (result.accepted) {
|
||||
_heartbeatTimer = Timer.periodic(
|
||||
heartbeatInterval,
|
||||
|
|
@ -376,6 +387,7 @@ class _OtoServerSocketAgentSession implements OtoServerPushJobSession {
|
|||
_heartbeatTimer?.cancel();
|
||||
_heartbeatTimer = null;
|
||||
unawaited(_runRequests.close());
|
||||
unawaited(_cancelRequests.close());
|
||||
await _client.close();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import 'dart:io';
|
|||
import 'package:http/http.dart' as http;
|
||||
import 'package:test/test.dart';
|
||||
import 'package:oto/oto/agent/agent_config.dart';
|
||||
import 'package:oto/oto/agent/oto_server_job_client.dart';
|
||||
import 'package:oto/oto/agent/registration_client.dart';
|
||||
import 'package:oto/oto/agent/remote_run_executor.dart';
|
||||
import 'package:oto/oto/application.dart';
|
||||
|
|
@ -1398,11 +1399,17 @@ pipeline:
|
|||
expect(session.result.accepted, isTrue);
|
||||
expect(session.result.runnerId, _runnerId);
|
||||
|
||||
// Capture runRequests stream completion after close().
|
||||
final streamDone = Completer<void>();
|
||||
final sub = (session as OtoServerPushJobSession).runRequests.listen(
|
||||
// Capture stream completion after close().
|
||||
final pushSession = session as OtoServerPushJobSession;
|
||||
final runRequestsDone = Completer<void>();
|
||||
final cancelRequestsDone = Completer<void>();
|
||||
final runSub = pushSession.runRequests.listen(
|
||||
(_) {},
|
||||
onDone: streamDone.complete,
|
||||
onDone: runRequestsDone.complete,
|
||||
);
|
||||
final cancelSub = pushSession.cancelRequests.listen(
|
||||
(_) {},
|
||||
onDone: cancelRequestsDone.complete,
|
||||
);
|
||||
|
||||
final httpClient = http.Client();
|
||||
|
|
@ -1419,13 +1426,21 @@ pipeline:
|
|||
reason: 'socket runner did not go online in registry',
|
||||
);
|
||||
|
||||
// close() must cancel the heartbeat timer and close runRequests.
|
||||
// close() must cancel the heartbeat timer and close socket streams.
|
||||
await session.close();
|
||||
await streamDone.future.timeout(
|
||||
await runRequestsDone.future.timeout(
|
||||
const Duration(seconds: 2),
|
||||
onTimeout: () => fail('runRequests stream did not close after session.close()'),
|
||||
onTimeout: () =>
|
||||
fail('runRequests stream did not close after session.close()'),
|
||||
);
|
||||
await sub.cancel();
|
||||
await cancelRequestsDone.future.timeout(
|
||||
const Duration(seconds: 2),
|
||||
onTimeout: () => fail(
|
||||
'cancelRequests stream did not close after session.close()',
|
||||
),
|
||||
);
|
||||
await runSub.cancel();
|
||||
await cancelSub.cancel();
|
||||
|
||||
final isDisconnected = await _pollRunnerStatus(
|
||||
httpClient,
|
||||
|
|
@ -1458,6 +1473,137 @@ pipeline:
|
|||
timeout: const Timeout(Duration(seconds: 30)),
|
||||
);
|
||||
|
||||
test(
|
||||
'OtoServerSocketRegistrationClient receives socket cancel requests from HTTP cancel action',
|
||||
() async {
|
||||
final port = await _freePort();
|
||||
final serverAddr = '$_host:$port';
|
||||
const jobId = 'smoke-job-socket-cancel';
|
||||
|
||||
final (:process, :socketAddr) = await _startCoreWithSocket(serverAddr);
|
||||
|
||||
final output = StringBuffer();
|
||||
final stdoutSub = process.stdout
|
||||
.transform(systemEncoding.decoder)
|
||||
.listen(output.write, onError: output.write);
|
||||
final stderrSub = process.stderr
|
||||
.transform(systemEncoding.decoder)
|
||||
.listen(output.write, onError: output.write);
|
||||
|
||||
try {
|
||||
await _waitForPort(_host, port, process, output);
|
||||
|
||||
final agentConfig = AgentConfig(
|
||||
agent: const AgentIdentityConfig(
|
||||
id: _runnerId,
|
||||
alias: _runnerAlias,
|
||||
enrollmentToken: _token,
|
||||
),
|
||||
server: ServerConnectionConfig(
|
||||
url: 'http://$serverAddr',
|
||||
socketUrl: 'tcp://$socketAddr',
|
||||
),
|
||||
runtime: const AgentRuntimeConfig(
|
||||
installDir: '/tmp/install',
|
||||
workspaceRoot: '/tmp/workspace',
|
||||
logDir: '/tmp/log',
|
||||
),
|
||||
);
|
||||
|
||||
final client = OtoServerSocketRegistrationClient(
|
||||
commandTypes: ['Shell'],
|
||||
heartbeatInterval: const Duration(milliseconds: 200),
|
||||
);
|
||||
final session = await client.openSession(agentConfig);
|
||||
expect(session.result.accepted, isTrue);
|
||||
final pushSession = session as OtoServerPushJobSession;
|
||||
|
||||
final runReceived = Completer<String>();
|
||||
final runSub = pushSession.runRequests.listen((request) {
|
||||
if (!runReceived.isCompleted) {
|
||||
expect(request.runnerId, _runnerId);
|
||||
expect(request.jobId, jobId);
|
||||
expect(request.executionId, isNotEmpty);
|
||||
runReceived.complete(request.executionId);
|
||||
}
|
||||
});
|
||||
var expectedExecutionId = '';
|
||||
final cancelReceived = Completer<String>();
|
||||
final cancelSub = pushSession.cancelRequests.listen((request) {
|
||||
if (!cancelReceived.isCompleted) {
|
||||
expect(request.runnerId, _runnerId);
|
||||
expect(request.executionId, expectedExecutionId);
|
||||
expect(request.reason, 'socket cancel smoke');
|
||||
cancelReceived.complete(request.executionId);
|
||||
}
|
||||
});
|
||||
|
||||
final httpClient = http.Client();
|
||||
final jobs = OtoServerJobClient(
|
||||
serverUrl: 'http://$serverAddr',
|
||||
runnerId: _runnerId,
|
||||
client: httpClient,
|
||||
);
|
||||
try {
|
||||
final createResp = await httpClient.post(
|
||||
Uri.parse('http://$serverAddr/api/v1/jobs'),
|
||||
headers: {'content-type': 'application/json'},
|
||||
body: jsonEncode({
|
||||
'id': jobId,
|
||||
'name': 'socket cancel smoke',
|
||||
'run_request': {
|
||||
'pipeline_yaml': 'commands:\n - type: Shell',
|
||||
'command_types': ['Shell'],
|
||||
},
|
||||
}),
|
||||
);
|
||||
expect(createResp.statusCode, equals(201));
|
||||
|
||||
expectedExecutionId = await runReceived.future.timeout(
|
||||
const Duration(seconds: 3),
|
||||
onTimeout: () => fail('socket RunRequest was not received'),
|
||||
);
|
||||
|
||||
final cancelRes = await jobs.cancelRun(
|
||||
executionId: expectedExecutionId,
|
||||
reason: 'socket cancel smoke',
|
||||
);
|
||||
expect(cancelRes.success, isTrue);
|
||||
|
||||
await cancelReceived.future.timeout(
|
||||
const Duration(seconds: 3),
|
||||
onTimeout: () => fail('socket CancelRunRequest was not received'),
|
||||
);
|
||||
|
||||
final execResp = await httpClient.get(
|
||||
Uri.parse(
|
||||
'http://$serverAddr/api/v1/executions/$expectedExecutionId',
|
||||
),
|
||||
);
|
||||
final execJson = jsonDecode(execResp.body) as Map<String, dynamic>;
|
||||
expect(execJson['state'], equals('canceled'));
|
||||
} finally {
|
||||
await runSub.cancel();
|
||||
await cancelSub.cancel();
|
||||
httpClient.close();
|
||||
await session.close();
|
||||
}
|
||||
} finally {
|
||||
process.kill(ProcessSignal.sigterm);
|
||||
await process.exitCode.timeout(
|
||||
const Duration(seconds: 5),
|
||||
onTimeout: () {
|
||||
process.kill(ProcessSignal.sigkill);
|
||||
return process.exitCode;
|
||||
},
|
||||
);
|
||||
await stdoutSub.cancel();
|
||||
await stderrSub.cancel();
|
||||
}
|
||||
},
|
||||
timeout: const Timeout(Duration(seconds: 30)),
|
||||
);
|
||||
|
||||
test(
|
||||
'OtoServerSocketRegistrationClient socket session close is idempotent after server disconnect',
|
||||
() async {
|
||||
|
|
|
|||
|
|
@ -474,10 +474,10 @@ func handleRunnerSelfUpdate(store *cicdstate.Store, registry *runnerregistry.Reg
|
|||
}
|
||||
|
||||
if hasActiveExecution {
|
||||
writeResponse(w, http.StatusOK, selfUpdateToJSON(false, false, true, req.GetVersion(), "deferred: runner is currently running a job"))
|
||||
writeResponse(w, http.StatusOK, selfUpdateToJSON(false, false, true, req.GetVersion(), "deferred: runner is currently running a job; self-update remains an HTTP compatibility action"))
|
||||
return
|
||||
}
|
||||
|
||||
writeResponse(w, http.StatusOK, selfUpdateToJSON(true, true, false, req.GetVersion(), "self-update request accepted"))
|
||||
writeResponse(w, http.StatusOK, selfUpdateToJSON(true, true, false, req.GetVersion(), "self-update compatibility request accepted; no socket self-update command was dispatched"))
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,6 +18,27 @@ import (
|
|||
otopb "github.com/toki/oto/services/core/oto"
|
||||
)
|
||||
|
||||
type runnerCancelCall struct {
|
||||
runnerID string
|
||||
execID string
|
||||
reason string
|
||||
}
|
||||
|
||||
type recordingRunnerControlDispatcher struct {
|
||||
dispatchCount int
|
||||
cancelCalls []runnerCancelCall
|
||||
cancelErr error
|
||||
}
|
||||
|
||||
func (d *recordingRunnerControlDispatcher) DispatchQueuedJobs() {
|
||||
d.dispatchCount++
|
||||
}
|
||||
|
||||
func (d *recordingRunnerControlDispatcher) CancelExecution(runnerID, execID, reason string) error {
|
||||
d.cancelCalls = append(d.cancelCalls, runnerCancelCall{runnerID: runnerID, execID: execID, reason: reason})
|
||||
return d.cancelErr
|
||||
}
|
||||
|
||||
func TestHandleHealthz(t *testing.T) {
|
||||
req := httptest.NewRequest(http.MethodGet, "/healthz", nil)
|
||||
rr := httptest.NewRecorder()
|
||||
|
|
@ -1489,6 +1510,7 @@ func TestHandleRunnerRegisterRejectsIncompatibleCapability(t *testing.T) {
|
|||
func TestHandleRunnerCancelExecution(t *testing.T) {
|
||||
store := cicdstate.NewStore()
|
||||
registry := runnerregistry.New()
|
||||
dispatcher := &recordingRunnerControlDispatcher{}
|
||||
|
||||
registry.Register(&otopb.RegisterRunnerRequest{
|
||||
EnrollmentToken: "token-123",
|
||||
|
|
@ -1507,7 +1529,7 @@ func TestHandleRunnerCancelExecution(t *testing.T) {
|
|||
body := bytes.NewBufferString(`{"runner_id":"runner-1", "execution_id":"exec-1", "reason":"user request"}`)
|
||||
req := httptest.NewRequest(http.MethodPost, "/api/v1/runners/runner-1/executions/exec-1/cancel", body)
|
||||
rr := httptest.NewRecorder()
|
||||
handleRouter(store, registry)(rr, req)
|
||||
handleRouterWithDispatcher(store, registry, dispatcher)(rr, req)
|
||||
|
||||
if rr.Code != http.StatusOK {
|
||||
t.Fatalf("cancel status = %v, want %v; body=%s", rr.Code, http.StatusOK, rr.Body.String())
|
||||
|
|
@ -1517,11 +1539,40 @@ func TestHandleRunnerCancelExecution(t *testing.T) {
|
|||
if !cancelResp.Success {
|
||||
t.Fatal("expected success cancel response")
|
||||
}
|
||||
if len(dispatcher.cancelCalls) != 1 {
|
||||
t.Fatalf("cancel dispatcher calls = %d, want 1", len(dispatcher.cancelCalls))
|
||||
}
|
||||
call := dispatcher.cancelCalls[0]
|
||||
if call.runnerID != "runner-1" || call.execID != "exec-1" || call.reason != "user request" {
|
||||
t.Fatalf("cancel dispatcher call = %+v, want runner-1/exec-1/user request", call)
|
||||
}
|
||||
|
||||
job, _ := store.GetJob("job-1")
|
||||
if job.State != cicdstate.StateCanceled {
|
||||
t.Fatalf("job state = %q, want canceled", job.State)
|
||||
}
|
||||
exec, _ := store.GetExecution("exec-1")
|
||||
if exec.State != cicdstate.StateCanceled {
|
||||
t.Fatalf("execution state = %q, want canceled", exec.State)
|
||||
}
|
||||
|
||||
// Late terminal reports from the runner must not overwrite the canceled
|
||||
// terminal state established by the operator action.
|
||||
lateReportBody := bytes.NewBufferString(`{"runner_id":"runner-1", "execution_id":"exec-1", "job_id":"job-1", "success":true}`)
|
||||
lateReportReq := httptest.NewRequest(http.MethodPost, "/api/v1/runners/runner-1/executions/exec-1/report", lateReportBody)
|
||||
lateReportRR := httptest.NewRecorder()
|
||||
handleRouterWithDispatcher(store, registry, dispatcher)(lateReportRR, lateReportReq)
|
||||
if lateReportRR.Code != http.StatusBadRequest {
|
||||
t.Fatalf("late report status = %v, want %v; body=%s", lateReportRR.Code, http.StatusBadRequest, lateReportRR.Body.String())
|
||||
}
|
||||
job, _ = store.GetJob("job-1")
|
||||
if job.State != cicdstate.StateCanceled {
|
||||
t.Fatalf("late report changed job state = %q, want canceled", job.State)
|
||||
}
|
||||
exec, _ = store.GetExecution("exec-1")
|
||||
if exec.State != cicdstate.StateCanceled {
|
||||
t.Fatalf("late report changed execution state = %q, want canceled", exec.State)
|
||||
}
|
||||
|
||||
// 2. Mismatch cancellation
|
||||
bodyMismatch := bytes.NewBufferString(`{"runner_id":"runner-wrong", "execution_id":"exec-1"}`)
|
||||
|
|
@ -1585,6 +1636,7 @@ func TestHandleRunnerCancelExecution(t *testing.T) {
|
|||
func TestHandleRunnerStatus(t *testing.T) {
|
||||
store := cicdstate.NewStore()
|
||||
registry := runnerregistry.New()
|
||||
dispatcher := &recordingRunnerControlDispatcher{}
|
||||
|
||||
registry.Register(&otopb.RegisterRunnerRequest{
|
||||
EnrollmentToken: "token-123",
|
||||
|
|
@ -1599,7 +1651,7 @@ func TestHandleRunnerStatus(t *testing.T) {
|
|||
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/v1/runners/runner-1/status", nil)
|
||||
rr := httptest.NewRecorder()
|
||||
handleRouter(store, registry)(rr, req)
|
||||
handleRouterWithDispatcher(store, registry, dispatcher)(rr, req)
|
||||
|
||||
if rr.Code != http.StatusOK {
|
||||
t.Fatalf("status code = %v, want %v", rr.Code, http.StatusOK)
|
||||
|
|
@ -1616,6 +1668,9 @@ func TestHandleRunnerStatus(t *testing.T) {
|
|||
if res["current_job_id"] != "job-1" {
|
||||
t.Fatalf("current_job_id = %v", res["current_job_id"])
|
||||
}
|
||||
if dispatcher.dispatchCount != 0 || len(dispatcher.cancelCalls) != 0 {
|
||||
t.Fatalf("status query must not dispatch socket commands; dispatcher=%+v", dispatcher)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleRunnerSelfUpdate(t *testing.T) {
|
||||
|
|
@ -1651,6 +1706,9 @@ func TestHandleRunnerSelfUpdate(t *testing.T) {
|
|||
if resOk["accepted"] != true || resOk["deferred"] != false {
|
||||
t.Fatalf("idle update response = %+v", resOk)
|
||||
}
|
||||
if !strings.Contains(fmt.Sprint(resOk["message"]), "compatibility") {
|
||||
t.Fatalf("idle update message must mark compatibility scope, got %+v", resOk)
|
||||
}
|
||||
|
||||
// 3. Active execution defer
|
||||
store.CreateJob("job-1", "build", nil)
|
||||
|
|
@ -1671,6 +1729,9 @@ func TestHandleRunnerSelfUpdate(t *testing.T) {
|
|||
if resActive["accepted"] != false || resActive["deferred"] != true {
|
||||
t.Fatalf("active update response = %+v", resActive)
|
||||
}
|
||||
if !strings.Contains(fmt.Sprint(resActive["message"]), "HTTP compatibility") {
|
||||
t.Fatalf("active update message must mark HTTP compatibility scope, got %+v", resActive)
|
||||
}
|
||||
}
|
||||
|
||||
func waitServerReady(t *testing.T, baseURL string) {
|
||||
|
|
|
|||
|
|
@ -26,6 +26,29 @@ func pipeClient(t *testing.T) (*protoSocket.TcpClient, net.Conn) {
|
|||
return client, b
|
||||
}
|
||||
|
||||
// runnerParserMap returns a parser map for the runner (client) side of a
|
||||
// proto-socket connection. It includes server-to-runner frames so tests can
|
||||
// verify dispatch and cancel delivery.
|
||||
func runnerParserMap() protoSocket.ParserMap {
|
||||
return protoSocket.ParserMap{
|
||||
protoSocket.TypeNameOf(&otopb.RunRequest{}): parserFor(func() *otopb.RunRequest { return &otopb.RunRequest{} }),
|
||||
protoSocket.TypeNameOf(&otopb.CancelRunRequest{}): parserFor(func() *otopb.CancelRunRequest { return &otopb.CancelRunRequest{} }),
|
||||
}
|
||||
}
|
||||
|
||||
// pipeClientPair creates two TcpClients backed by the two ends of a net.Pipe.
|
||||
// serverSide uses parserMap(); runnerSide uses runnerParserMap() so tests can
|
||||
// capture and inspect RunRequest frames sent by DispatchQueuedJobs.
|
||||
// Cleanup closes both connections.
|
||||
func pipeClientPair(t *testing.T) (serverSide, runnerSide *protoSocket.TcpClient) {
|
||||
t.Helper()
|
||||
a, b := net.Pipe()
|
||||
serverSide = protoSocket.NewTcpClient(a, 0, 0, parserMap())
|
||||
runnerSide = protoSocket.NewTcpClient(b, 0, 0, runnerParserMap())
|
||||
t.Cleanup(func() { _ = a.Close(); _ = b.Close() })
|
||||
return serverSide, runnerSide
|
||||
}
|
||||
|
||||
// TestServerSetClientReplacesDuplicate verifies that registering a second client
|
||||
// for the same runner ID closes the old one and stores the new one.
|
||||
func TestServerReplacesDuplicateRunnerClient(t *testing.T) {
|
||||
|
|
@ -176,6 +199,479 @@ func TestServerDuplicateReplaceWithListenerDoesNotDeadlock(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
// testServerWithStore creates a Server sharing the given registry and returns
|
||||
// the associated Store for state inspection in tests.
|
||||
func testServerWithStore(registry *runnerregistry.Registry) (*Server, *cicdstate.Store) {
|
||||
store := cicdstate.NewStore()
|
||||
return New("127.0.0.1:0", registry, store), store
|
||||
}
|
||||
|
||||
// registerRunner adds a runner to the registry with default test fields.
|
||||
func registerRunner(t *testing.T, registry *runnerregistry.Registry, runnerID string) {
|
||||
t.Helper()
|
||||
resp := registry.Register(&otopb.RegisterRunnerRequest{
|
||||
EnrollmentToken: "tok",
|
||||
RunnerId: runnerID,
|
||||
ProtocolVersion: "oto.runner.v1",
|
||||
Capability: &otopb.RunnerCapability{Name: "oto-runner", Version: "1.0.0"},
|
||||
})
|
||||
if !resp.GetAccepted() {
|
||||
t.Fatalf("registerRunner %s: rejected: %s", runnerID, resp.GetRejectReason())
|
||||
}
|
||||
}
|
||||
|
||||
// createQueuedJob adds a job with a valid inline YAML to the store.
|
||||
func createQueuedJob(t *testing.T, store *cicdstate.Store, jobID string) {
|
||||
t.Helper()
|
||||
if _, err := store.CreateJob(jobID, jobID, &cicdstate.RunInput{PipelineYAML: "pipeline: test"}); err != nil {
|
||||
t.Fatalf("CreateJob %s: %v", jobID, err)
|
||||
}
|
||||
}
|
||||
|
||||
// setupRunningExecution creates a queued job, registers the runner, creates an
|
||||
// execution, and transitions both job and execution to running. Returns execID.
|
||||
func setupRunningExecution(t *testing.T, store *cicdstate.Store, registry *runnerregistry.Registry, jobID, runnerID string) string {
|
||||
t.Helper()
|
||||
createQueuedJob(t, store, jobID)
|
||||
registerRunner(t, registry, runnerID)
|
||||
execID := store.NextExecutionID()
|
||||
if _, err := store.CreateExecution(jobID, execID); err != nil {
|
||||
t.Fatalf("CreateExecution: %v", err)
|
||||
}
|
||||
_ = store.SetExecutionRunnerID(execID, runnerID)
|
||||
if err := store.TransitionJob(jobID, cicdstate.StateRunning); err != nil {
|
||||
t.Fatalf("TransitionJob running: %v", err)
|
||||
}
|
||||
if err := store.TransitionExecution(execID, cicdstate.StateRunning); err != nil {
|
||||
t.Fatalf("TransitionExecution running: %v", err)
|
||||
}
|
||||
return execID
|
||||
}
|
||||
|
||||
// TestServerDispatchQueuedJobOverSocket verifies that DispatchQueuedJobs
|
||||
// transitions the queued job and its execution to running over the socket
|
||||
// transport, assigns the dispatched runner ID to the execution, and sends a
|
||||
// RunRequest frame whose fields match the store state.
|
||||
func TestServerDispatchQueuedJobOverSocket(t *testing.T) {
|
||||
registry := runnerregistry.New()
|
||||
s, store := testServerWithStore(registry)
|
||||
registerRunner(t, registry, "runner-dispatch")
|
||||
|
||||
jobID := "job-dispatch"
|
||||
if _, err := store.CreateJob(jobID, jobID, &cicdstate.RunInput{
|
||||
PipelineYAML: "pipeline: dispatch-test",
|
||||
Variables: map[string]string{"ENV": "ci"},
|
||||
CommandTypes: []string{"flutter"},
|
||||
}); err != nil {
|
||||
t.Fatalf("CreateJob: %v", err)
|
||||
}
|
||||
|
||||
// Use a runner-side TcpClient so we can parse the RunRequest frame.
|
||||
serverClient, runnerClient := pipeClientPair(t)
|
||||
|
||||
received := make(chan *otopb.RunRequest, 1)
|
||||
protoSocket.AddListenerTyped[*otopb.RunRequest](&runnerClient.Communicator, func(req *otopb.RunRequest) {
|
||||
select {
|
||||
case received <- req:
|
||||
default:
|
||||
}
|
||||
})
|
||||
|
||||
s.setClient("runner-dispatch", serverClient)
|
||||
s.DispatchQueuedJobs()
|
||||
|
||||
var rr *otopb.RunRequest
|
||||
select {
|
||||
case rr = <-received:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("timed out waiting for RunRequest")
|
||||
}
|
||||
|
||||
job, err := store.GetJob(jobID)
|
||||
if err != nil {
|
||||
t.Fatalf("GetJob: %v", err)
|
||||
}
|
||||
if job.State != cicdstate.StateRunning {
|
||||
t.Errorf("job state: want %s, got %s", cicdstate.StateRunning, job.State)
|
||||
}
|
||||
if job.ExecutionID == "" {
|
||||
t.Fatal("no execution created for dispatched job")
|
||||
}
|
||||
exec, err := store.GetExecution(job.ExecutionID)
|
||||
if err != nil {
|
||||
t.Fatalf("GetExecution: %v", err)
|
||||
}
|
||||
if exec.State != cicdstate.StateRunning {
|
||||
t.Errorf("execution state: want %s, got %s", cicdstate.StateRunning, exec.State)
|
||||
}
|
||||
if exec.RunnerID != "runner-dispatch" {
|
||||
t.Errorf("execution runnerID: want runner-dispatch, got %s", exec.RunnerID)
|
||||
}
|
||||
|
||||
// Verify RunRequest payload fields match the store state.
|
||||
if rr.RunnerId != "runner-dispatch" {
|
||||
t.Errorf("RunRequest.RunnerId: want runner-dispatch, got %s", rr.RunnerId)
|
||||
}
|
||||
if rr.JobId != jobID {
|
||||
t.Errorf("RunRequest.JobId: want %s, got %s", jobID, rr.JobId)
|
||||
}
|
||||
if rr.ExecutionId != job.ExecutionID {
|
||||
t.Errorf("RunRequest.ExecutionId: want %s, got %s", job.ExecutionID, rr.ExecutionId)
|
||||
}
|
||||
if rr.PipelineYaml != "pipeline: dispatch-test" {
|
||||
t.Errorf("RunRequest.PipelineYaml: want %q, got %q", "pipeline: dispatch-test", rr.PipelineYaml)
|
||||
}
|
||||
if len(rr.Variables) == 0 || rr.Variables["ENV"] != "ci" {
|
||||
t.Errorf("RunRequest.Variables: want {ENV:ci}, got %v", rr.Variables)
|
||||
}
|
||||
if len(rr.CommandTypes) != 1 || rr.CommandTypes[0] != "flutter" {
|
||||
t.Errorf("RunRequest.CommandTypes: want [flutter], got %v", rr.CommandTypes)
|
||||
}
|
||||
}
|
||||
|
||||
// TestServerCancelExecutionSendsSocketCancelToRunner verifies the server-side
|
||||
// cancel action uses the runner-facing socket as a best-effort signal channel.
|
||||
func TestServerCancelExecutionSendsSocketCancelToRunner(t *testing.T) {
|
||||
s := testServer(runnerregistry.New())
|
||||
serverClient, runnerClient := pipeClientPair(t)
|
||||
|
||||
received := make(chan *otopb.CancelRunRequest, 1)
|
||||
protoSocket.AddListenerTyped[*otopb.CancelRunRequest](&runnerClient.Communicator, func(req *otopb.CancelRunRequest) {
|
||||
select {
|
||||
case received <- req:
|
||||
default:
|
||||
}
|
||||
})
|
||||
|
||||
s.setClient("runner-cancel", serverClient)
|
||||
|
||||
if err := s.CancelExecution("runner-cancel", "exec-cancel", "operator request"); err != nil {
|
||||
t.Fatalf("CancelExecution: %v", err)
|
||||
}
|
||||
|
||||
var req *otopb.CancelRunRequest
|
||||
select {
|
||||
case req = <-received:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("timed out waiting for CancelRunRequest")
|
||||
}
|
||||
if req.RunnerId != "runner-cancel" {
|
||||
t.Errorf("CancelRunRequest.RunnerId: want runner-cancel, got %s", req.RunnerId)
|
||||
}
|
||||
if req.ExecutionId != "exec-cancel" {
|
||||
t.Errorf("CancelRunRequest.ExecutionId: want exec-cancel, got %s", req.ExecutionId)
|
||||
}
|
||||
if req.Reason != "operator request" {
|
||||
t.Errorf("CancelRunRequest.Reason: want operator request, got %s", req.Reason)
|
||||
}
|
||||
|
||||
if err := s.CancelExecution("runner-missing", "exec-cancel", "operator request"); err == nil {
|
||||
t.Fatal("expected missing connected runner to return an error")
|
||||
}
|
||||
}
|
||||
|
||||
// TestServerDispatchClaimsOnlyOneJobPerRunner verifies that a single
|
||||
// DispatchQueuedJobs call dispatches at most one job per runner, leaving any
|
||||
// remaining queued jobs untouched.
|
||||
func TestServerDispatchClaimsOnlyOneJobPerRunner(t *testing.T) {
|
||||
registry := runnerregistry.New()
|
||||
s, store := testServerWithStore(registry)
|
||||
registerRunner(t, registry, "runner-claim")
|
||||
createQueuedJob(t, store, "job-claim-1")
|
||||
createQueuedJob(t, store, "job-claim-2")
|
||||
|
||||
client, remote := pipeClient(t)
|
||||
// Drain the remote end so Send does not block.
|
||||
go func() {
|
||||
buf := make([]byte, 4096)
|
||||
for {
|
||||
if _, err := remote.Read(buf); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
s.setClient("runner-claim", client)
|
||||
s.DispatchQueuedJobs()
|
||||
|
||||
job1, err := store.GetJob("job-claim-1")
|
||||
if err != nil {
|
||||
t.Fatalf("GetJob job-claim-1: %v", err)
|
||||
}
|
||||
job2, err := store.GetJob("job-claim-2")
|
||||
if err != nil {
|
||||
t.Fatalf("GetJob job-claim-2: %v", err)
|
||||
}
|
||||
|
||||
running, queued := 0, 0
|
||||
for _, j := range []*cicdstate.Job{job1, job2} {
|
||||
switch j.State {
|
||||
case cicdstate.StateRunning:
|
||||
running++
|
||||
case cicdstate.StateQueued:
|
||||
queued++
|
||||
}
|
||||
}
|
||||
if running != 1 {
|
||||
t.Errorf("want exactly 1 running job after one dispatch pass, got %d", running)
|
||||
}
|
||||
if queued != 1 {
|
||||
t.Errorf("want exactly 1 queued job after one dispatch pass, got %d", queued)
|
||||
}
|
||||
}
|
||||
|
||||
// TestServerFailsExecutionWhenRunRequestCannotBeSent verifies that a send
|
||||
// error over the socket transitions the claimed execution and job to failed
|
||||
// rather than leaving them stuck in running.
|
||||
func TestServerFailsExecutionWhenRunRequestCannotBeSent(t *testing.T) {
|
||||
registry := runnerregistry.New()
|
||||
s, store := testServerWithStore(registry)
|
||||
registerRunner(t, registry, "runner-sendfail")
|
||||
createQueuedJob(t, store, "job-sendfail")
|
||||
|
||||
client, remote := pipeClient(t)
|
||||
// Close the remote end before dispatch so any write to the client fails.
|
||||
_ = remote.Close()
|
||||
|
||||
s.setClient("runner-sendfail", client)
|
||||
s.DispatchQueuedJobs()
|
||||
|
||||
job, err := store.GetJob("job-sendfail")
|
||||
if err != nil {
|
||||
t.Fatalf("GetJob: %v", err)
|
||||
}
|
||||
if job.State != cicdstate.StateFailed {
|
||||
t.Errorf("job state: want %s, got %s", cicdstate.StateFailed, job.State)
|
||||
}
|
||||
if job.ExecutionID == "" {
|
||||
t.Fatal("expected execution to be created before send failure")
|
||||
}
|
||||
exec, err := store.GetExecution(job.ExecutionID)
|
||||
if err != nil {
|
||||
t.Fatalf("GetExecution: %v", err)
|
||||
}
|
||||
if exec.State != cicdstate.StateFailed {
|
||||
t.Errorf("execution state: want %s, got %s", cicdstate.StateFailed, exec.State)
|
||||
}
|
||||
}
|
||||
|
||||
// TestServerReportsRunResultOverSocket verifies that handleExecutionReport
|
||||
// accepts success and failure terminal reports from the owning runner and
|
||||
// transitions execution and job to the correct terminal state.
|
||||
func TestServerReportsRunResultOverSocket(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
success bool
|
||||
wantState string
|
||||
}{
|
||||
{"success", true, cicdstate.StateSucceeded},
|
||||
{"failure", false, cicdstate.StateFailed},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
registry := runnerregistry.New()
|
||||
s, store := testServerWithStore(registry)
|
||||
jobID := "job-report-" + tc.name
|
||||
runnerID := "runner-report-" + tc.name
|
||||
execID := setupRunningExecution(t, store, registry, jobID, runnerID)
|
||||
|
||||
resp := s.handleExecutionReport(&otopb.ExecutionReportRequest{
|
||||
RunnerId: runnerID,
|
||||
JobId: jobID,
|
||||
ExecutionId: execID,
|
||||
Success: tc.success,
|
||||
})
|
||||
|
||||
if !resp.GetAccepted() {
|
||||
t.Fatalf("report rejected: %s", resp.GetErrorMessage())
|
||||
}
|
||||
exec, err := store.GetExecution(execID)
|
||||
if err != nil {
|
||||
t.Fatalf("GetExecution: %v", err)
|
||||
}
|
||||
if exec.State != tc.wantState {
|
||||
t.Errorf("execution state: want %s, got %s", tc.wantState, exec.State)
|
||||
}
|
||||
job, err := store.GetJob(jobID)
|
||||
if err != nil {
|
||||
t.Fatalf("GetJob: %v", err)
|
||||
}
|
||||
if job.State != tc.wantState {
|
||||
t.Errorf("job state: want %s, got %s", tc.wantState, job.State)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestServerRejectsTerminalReportForInactiveExecution verifies the active
|
||||
// execution gate: terminal reports on queued, succeeded, or cancelled
|
||||
// executions are rejected, and reports from the wrong runner are rejected.
|
||||
func TestServerRejectsTerminalReportForInactiveExecution(t *testing.T) {
|
||||
t.Run("queued_execution", func(t *testing.T) {
|
||||
registry := runnerregistry.New()
|
||||
s, store := testServerWithStore(registry)
|
||||
registerRunner(t, registry, "runner-guard")
|
||||
createQueuedJob(t, store, "job-guard-queued")
|
||||
execID := store.NextExecutionID()
|
||||
if _, err := store.CreateExecution("job-guard-queued", execID); err != nil {
|
||||
t.Fatalf("CreateExecution: %v", err)
|
||||
}
|
||||
_ = store.SetExecutionRunnerID(execID, "runner-guard")
|
||||
// Execution remains in queued state; terminal report must be rejected.
|
||||
resp := s.handleExecutionReport(&otopb.ExecutionReportRequest{
|
||||
RunnerId: "runner-guard", JobId: "job-guard-queued", ExecutionId: execID, Success: true,
|
||||
})
|
||||
if resp.GetAccepted() {
|
||||
t.Error("expected rejection for queued execution, got accepted")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("succeeded_execution", func(t *testing.T) {
|
||||
registry := runnerregistry.New()
|
||||
s, store := testServerWithStore(registry)
|
||||
execID := setupRunningExecution(t, store, registry, "job-guard-succ", "runner-guard-succ")
|
||||
_ = store.TransitionExecution(execID, cicdstate.StateSucceeded)
|
||||
_ = store.TransitionJob("job-guard-succ", cicdstate.StateSucceeded)
|
||||
|
||||
resp := s.handleExecutionReport(&otopb.ExecutionReportRequest{
|
||||
RunnerId: "runner-guard-succ", JobId: "job-guard-succ", ExecutionId: execID, Success: true,
|
||||
})
|
||||
if resp.GetAccepted() {
|
||||
t.Error("expected rejection for succeeded execution, got accepted")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("cancelled_execution", func(t *testing.T) {
|
||||
registry := runnerregistry.New()
|
||||
s, store := testServerWithStore(registry)
|
||||
execID := setupRunningExecution(t, store, registry, "job-guard-cancel", "runner-guard-cancel")
|
||||
if err := store.CancelJobExecution("job-guard-cancel", execID); err != nil {
|
||||
t.Fatalf("CancelJobExecution: %v", err)
|
||||
}
|
||||
resp := s.handleExecutionReport(&otopb.ExecutionReportRequest{
|
||||
RunnerId: "runner-guard-cancel", JobId: "job-guard-cancel", ExecutionId: execID, Success: true,
|
||||
})
|
||||
if resp.GetAccepted() {
|
||||
t.Error("expected rejection for cancelled execution, got accepted")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("wrong_runner", func(t *testing.T) {
|
||||
registry := runnerregistry.New()
|
||||
s, store := testServerWithStore(registry)
|
||||
execID := setupRunningExecution(t, store, registry, "job-guard-runner", "runner-owner")
|
||||
registerRunner(t, registry, "runner-other")
|
||||
|
||||
resp := s.handleExecutionReport(&otopb.ExecutionReportRequest{
|
||||
RunnerId: "runner-other", JobId: "job-guard-runner", ExecutionId: execID, Success: true,
|
||||
})
|
||||
if resp.GetAccepted() {
|
||||
t.Error("expected rejection for wrong runner, got accepted")
|
||||
}
|
||||
if resp.GetErrorMessage() == "" {
|
||||
t.Error("expected non-empty error message for wrong runner")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// TestServerDispatchInvalidRunInputFails verifies that DispatchQueuedJobs
|
||||
// transitions a job with an invalid RunInput directly to failed without creating
|
||||
// an execution. Two invalid fixture variants are covered: both fields empty and
|
||||
// both fields set simultaneously.
|
||||
func TestServerDispatchInvalidRunInputFails(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
input *cicdstate.RunInput
|
||||
}{
|
||||
{
|
||||
"both_empty",
|
||||
&cicdstate.RunInput{PipelineYAML: "", PipelineYAMLPath: ""},
|
||||
},
|
||||
{
|
||||
"both_set",
|
||||
&cicdstate.RunInput{PipelineYAML: "pipeline: test", PipelineYAMLPath: "/some/path.yaml"},
|
||||
},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
registry := runnerregistry.New()
|
||||
s, store := testServerWithStore(registry)
|
||||
runnerID := "runner-invalid-" + tc.name
|
||||
jobID := "job-invalid-" + tc.name
|
||||
registerRunner(t, registry, runnerID)
|
||||
|
||||
if _, err := store.CreateJob(jobID, jobID, tc.input); err != nil {
|
||||
t.Fatalf("CreateJob: %v", err)
|
||||
}
|
||||
|
||||
// No drain needed: validateRunInput fails before client.Send is called.
|
||||
client, _ := pipeClient(t)
|
||||
s.setClient(runnerID, client)
|
||||
s.DispatchQueuedJobs()
|
||||
|
||||
job, err := store.GetJob(jobID)
|
||||
if err != nil {
|
||||
t.Fatalf("GetJob: %v", err)
|
||||
}
|
||||
if job.State != cicdstate.StateFailed {
|
||||
t.Errorf("job state: want %s, got %s", cicdstate.StateFailed, job.State)
|
||||
}
|
||||
if job.ExecutionID != "" {
|
||||
t.Errorf("no execution should be created for invalid run input; got execID %s", job.ExecutionID)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestServerReportMessageSavedToLog verifies that a non-empty Message in an
|
||||
// ExecutionReportRequest is appended to the execution log after the report is
|
||||
// accepted.
|
||||
func TestServerReportMessageSavedToLog(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
success bool
|
||||
message string
|
||||
}{
|
||||
{"success_with_message", true, "build succeeded: artifacts uploaded"},
|
||||
{"failure_with_message", false, "build failed: compilation error"},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
registry := runnerregistry.New()
|
||||
s, store := testServerWithStore(registry)
|
||||
jobID := "job-log-" + tc.name
|
||||
runnerID := "runner-log-" + tc.name
|
||||
execID := setupRunningExecution(t, store, registry, jobID, runnerID)
|
||||
|
||||
resp := s.handleExecutionReport(&otopb.ExecutionReportRequest{
|
||||
RunnerId: runnerID,
|
||||
JobId: jobID,
|
||||
ExecutionId: execID,
|
||||
Success: tc.success,
|
||||
Message: tc.message,
|
||||
})
|
||||
if !resp.GetAccepted() {
|
||||
t.Fatalf("report rejected: %s", resp.GetErrorMessage())
|
||||
}
|
||||
|
||||
logs, err := store.GetLogs(execID)
|
||||
if err != nil {
|
||||
t.Fatalf("GetLogs: %v", err)
|
||||
}
|
||||
found := false
|
||||
for _, entry := range logs {
|
||||
if entry.Line == tc.message {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Errorf("expected execution log to contain %q; got entries: %v", tc.message, logs)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestServerHeartbeatUsesRegisteredRunnerID verifies that connectedRunnerIDs
|
||||
// reflects only currently active clients and that clientFor returns nil after
|
||||
// the current client is removed.
|
||||
|
|
|
|||
Loading…
Reference in a new issue