feat(node): runtime concurrency support and task module cleanup
- Add node concurrency integration test - Update bootstrap module for runtime concurrency - Refactor node.go with concurrency improvements - Update run_manager for concurrent task handling - Archive completed subtask documents (07+02_runtime_concurrency) - Remove obsolete code review and plan files
This commit is contained in:
parent
fa910ef06c
commit
137b3a5301
15 changed files with 1548 additions and 104 deletions
|
|
@ -0,0 +1,129 @@
|
|||
<!-- task=m-node-multi-target-serving-foundation/07+02_runtime_concurrency plan=0 tag=REFACTOR -->
|
||||
|
||||
# Code Review Reference - REFACTOR
|
||||
|
||||
> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.**
|
||||
> Fill implementation-owned sections and actual verification output, then stop with active files in place.
|
||||
> Do not ask the user directly; record user-only blockers in `사용자 리뷰 요청`.
|
||||
|
||||
## 개요
|
||||
|
||||
date=2026-06-10
|
||||
task=m-node-multi-target-serving-foundation/07+02_runtime_concurrency, plan=0, tag=REFACTOR
|
||||
|
||||
## Roadmap Targets
|
||||
|
||||
- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/node-multi-target-serving-foundation.md`
|
||||
- Task ids:
|
||||
- `runtime-concurrency`: `runtime.concurrency`와 adapter capability가 Node 멀티 호출에서 실제 제한 또는 스케줄링 정책으로 적용된다.
|
||||
- Completion mode: check-on-pass
|
||||
|
||||
## 이 파일을 읽는 리뷰 에이전트에게
|
||||
|
||||
> **[REVIEW AGENT ONLY]** 종결 절차는 코드리뷰 에이전트 전용이다.
|
||||
|
||||
concurrency policy가 deterministic하고 foreground/background 양쪽에서 같은지 검토한다.
|
||||
|
||||
## 구현 항목별 완료 여부
|
||||
|
||||
| 항목 | 완료 여부 |
|
||||
|------|---------|
|
||||
| [REFACTOR-1] Runtime Concurrency Gate | [x] |
|
||||
|
||||
## 구현 체크리스트
|
||||
|
||||
- [x] Node runtime global concurrency와 adapter/profile concurrency 중 적용 우선순위를 정의한다.
|
||||
- [x] Foreground/background run 모두 동일한 scheduling/limit 정책을 통과한다.
|
||||
- [x] 제한 초과 요청의 queue/reject/timeout 중 하나를 선택하고 error/event/store 상태를 테스트로 고정한다.
|
||||
- [x] race/order test를 추가하고 최종 검증을 실행한다.
|
||||
- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
|
||||
|
||||
## 코드리뷰 전용 체크리스트
|
||||
|
||||
- [x] `코드리뷰 결과`에 판정을 append한다.
|
||||
- [x] active `CODE_REVIEW-cloud-G08.md`를 `code_review_cloud_G08_N.log`로 아카이브한다.
|
||||
- [x] active `PLAN-cloud-G07.md`를 `plan_cloud_G07_M.log`로 아카이브한다.
|
||||
- [ ] PASS이면 `complete.log` 작성 후 archive 이동한다.
|
||||
- [x] WARN/FAIL이면 후속 상태를 작성한다.
|
||||
|
||||
## 계획 대비 변경 사항
|
||||
|
||||
- `proto/iop/runtime.proto`: 변경 없음. `NodeRuntimeConfig.concurrency` 필드가 이미 존재하며 충분하다.
|
||||
- `apps/node/internal/runtime/types.go`: 변경 없음. `Capabilities.MaxConcurrency`는 advisory 값으로 유지한다. runtime gate에는 사용하지 않는다.
|
||||
- `run_manager.go` 책임 분리 방식: 별도 파일 신설 대신 `run_manager.go`에 `permitManager`를 추가했다. scheduling/limit 책임을 같은 파일에 두는 것이 더 응집도가 높다.
|
||||
|
||||
## 주요 설계 결정
|
||||
|
||||
- **정책: reject-on-exceed** — queue/timeout 대신 즉시 `ErrConcurrencyLimitExceeded` 반환. 가장 단순하고 deterministic하며 테스트 고정이 가능하다.
|
||||
- **우선순위: global > adapter.MaxConcurrency** — `NodeRuntimeConfig.Concurrency`(global)가 > 0이면 그것을 사용. 0이면 unlimited. adapter `MaxConcurrency`는 `CAPABILITIES` command 응답에만 사용되며 runtime gate에서는 advisory로만 취급한다.
|
||||
- **foreground/background 동일 경로** — `OnRunRequest` 내에서 `spec.Background` 분기 이전에 `n.permits.tryAcquire()`를 호출하여 양쪽 모두 동일한 gate를 통과한다.
|
||||
- **permit 누수 방지** — `run` 클로저의 첫 번째 `defer`를 `n.permits.release()`로 두어 adapter error, timeout, cancellation, 정상 완료 모든 경로에서 반드시 release된다.
|
||||
- **`New()` 시그니처 추가** — `globalConcurrency int` 파라미터 추가. bootstrap에서 `result.Config.GetRuntime().GetConcurrency()`를 넘기도록 수정.
|
||||
|
||||
## 사용자 리뷰 요청
|
||||
|
||||
_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 구현 에이전트는 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._
|
||||
|
||||
- 상태: 없음
|
||||
- 사유 유형: 없음
|
||||
- 결정 필요: 없음
|
||||
- 차단 근거: 없음
|
||||
- 실행한 검증/명령: 없음
|
||||
- 자동 후속 불가 이유: 없음
|
||||
- 재개 조건: 없음
|
||||
|
||||
## 리뷰어를 위한 체크포인트
|
||||
|
||||
- permit acquire/release가 adapter error, timeout, cancellation, background run에서 누수되지 않는다.
|
||||
- 선택한 제한 초과 정책이 테스트와 문서에 일치한다.
|
||||
- proto-socket transport 변경으로 문제를 우회하지 않았다.
|
||||
|
||||
## 검증 결과
|
||||
|
||||
### REFACTOR-1 중간 검증
|
||||
|
||||
```bash
|
||||
$ go test -race -count=1 ./apps/node/internal/node/...
|
||||
ok iop/apps/node/internal/node 1.138s
|
||||
```
|
||||
|
||||
### 최종 검증
|
||||
|
||||
```bash
|
||||
$ go test -race -count=1 ./apps/node/internal/node/...
|
||||
ok iop/apps/node/internal/node 1.138s
|
||||
|
||||
$ go test -count=1 ./apps/node/...
|
||||
ok iop/apps/node/cmd/node 0.012s
|
||||
ok iop/apps/node/internal/adapters 0.009s
|
||||
ok iop/apps/node/internal/adapters/cli 42.042s
|
||||
? iop/apps/node/internal/adapters/cli/internal/testutil [no test files]
|
||||
ok iop/apps/node/internal/adapters/cli/status 39.795s
|
||||
? iop/apps/node/internal/adapters/mock [no test files]
|
||||
ok iop/apps/node/internal/adapters/ollama 0.009s
|
||||
ok iop/apps/node/internal/adapters/vllm 0.011s
|
||||
ok iop/apps/node/internal/bootstrap 0.270s
|
||||
ok iop/apps/node/internal/node 0.020s
|
||||
ok iop/apps/node/internal/router 0.006s
|
||||
? iop/apps/node/internal/runtime [no test files]
|
||||
ok iop/apps/node/internal/store 0.064s
|
||||
ok iop/apps/node/internal/terminal 0.448s
|
||||
ok iop/apps/node/internal/transport 5.134s
|
||||
```
|
||||
|
||||
## 코드리뷰 결과
|
||||
|
||||
- 종합 판정: FAIL
|
||||
- 차원별 평가:
|
||||
- correctness: Fail
|
||||
- completeness: Fail
|
||||
- test coverage: Fail
|
||||
- API contract: Fail
|
||||
- code quality: Pass
|
||||
- plan deviation: Fail
|
||||
- verification trust: Fail
|
||||
- 발견된 문제:
|
||||
- Required: `apps/node/internal/node/node.go:96`에서 concurrency 초과를 `ErrConcurrencyLimitExceeded` 반환으로만 처리하지만, 실제 proto-socket listener는 `OnRunRequest` 에러를 `apps/node/internal/transport/session.go:45`에서 로그로만 남기고 Edge로 응답/이벤트를 보내지 않습니다. 따라서 reject 정책이 실제 Edge-Node 경계에서는 호출자에게 관측되지 않고, store에도 terminal 상태가 남지 않습니다. 제한 초과 시 `RunEvent` error/cancelled 또는 명확한 command/transport 응답 계약을 보내고, store 상태를 failed/rejected 등으로 고정한 뒤 transport 관측 테스트와 store 테스트를 추가해야 합니다.
|
||||
- Required: Roadmap Target과 계획은 `runtime.concurrency`와 adapter capability가 Node 멀티 호출의 실제 제한 또는 scheduling 정책으로 적용되어야 한다고 요구하지만, 구현은 `apps/node/internal/node/node.go:92`에서 adapter `Capabilities().MaxConcurrency`를 advisory로만 선언하고 `apps/node/internal/node/run_manager.go:62`의 permitManager에는 global limit만 주입합니다. `apps/node/internal/node/node_test.go:1061`의 unlimited 테스트도 adapter `MaxConcurrency=1`을 무시하는 동작을 고정합니다. global runtime limit과 adapter/profile limit의 effective limit 계산 규칙을 구현하고, global unset/adapter cap present 및 global+adapter cap 조합 테스트를 추가해야 합니다.
|
||||
- Required: `apps/node/internal/node/node.go:64`의 run request path와 `apps/node/internal/bootstrap/module.go:72` bootstrap wiring을 바꿨지만, node/testing domain이 요구하는 repo edge-node 진단 또는 full-cycle 실제 구동 증거가 없습니다. 코드 수정 후 `go test`뿐 아니라 변경 범위에 맞는 edge-node diagnostic/full-cycle 검증 또는 실행 불가 사유와 잔여 위험을 review stub에 기록해야 합니다.
|
||||
- 다음 단계: FAIL이므로 active plan/review를 로그로 아카이브하고, 같은 task 경로에 후속 `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.md`를 작성한다.
|
||||
|
|
@ -0,0 +1,214 @@
|
|||
<!-- task=m-node-multi-target-serving-foundation/07+02_runtime_concurrency plan=1 tag=REVIEW_REFACTOR -->
|
||||
|
||||
# Code Review Reference - REVIEW_REFACTOR
|
||||
|
||||
> **[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.
|
||||
> Do not ask the user directly, present choices in chat, or call `request_user_input` during implementation; record the needed 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-12
|
||||
task=m-node-multi-target-serving-foundation/07+02_runtime_concurrency, plan=1, tag=REVIEW_REFACTOR
|
||||
|
||||
## Roadmap Targets
|
||||
|
||||
- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/node-multi-target-serving-foundation.md`
|
||||
- Task ids:
|
||||
- `runtime-concurrency`: `runtime.concurrency`와 adapter capability가 Node 멀티 호출에서 실제 제한 또는 스케줄링 정책으로 적용된다.
|
||||
- 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-node-multi-target-serving-foundation/07+02_runtime_concurrency/`로 이동한다. WARN/FAIL이면 user-review gate를 확인한 뒤 다음 active plan/review 파일 또는 `USER_REVIEW.md`를 작성한다.
|
||||
4. PASS이고 task group이 `m-node-multi-target-serving-foundation`이면 완료 이벤트 메타데이터를 보고한다. roadmap 수정이나 `update-roadmap` 직접 호출은 하지 않는다.
|
||||
5. 적용 가능한 `코드리뷰 전용 체크리스트` 항목을 최종 `.log` 위치에서 체크한 뒤 보고한다.
|
||||
|
||||
---
|
||||
|
||||
## 구현 항목별 완료 여부
|
||||
|
||||
| 항목 | 완료 여부 |
|
||||
|------|---------|
|
||||
| [REVIEW_REFACTOR-1] Reject Contract | [x] |
|
||||
| [REVIEW_REFACTOR-2] Effective Concurrency | [x] |
|
||||
| [REVIEW_REFACTOR-3] Verification Evidence | [x] |
|
||||
|
||||
## 구현 체크리스트
|
||||
|
||||
- [x] 제한 초과 reject가 Edge-Node transport 경계에서 관측 가능한 error `RunEvent` 또는 동등 계약으로 전달되고 store terminal 상태가 테스트로 고정된다.
|
||||
- [x] global runtime limit과 adapter/profile `MaxConcurrency`의 effective limit 계산/우선순위가 구현되고 조합 테스트가 추가된다.
|
||||
- [x] node run path/bootstrap 변경에 필요한 Go tests와 repo edge-node diagnostic/full-cycle 검증 또는 명확한 차단 사유가 review stub에 기록된다.
|
||||
- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
|
||||
|
||||
## 코드리뷰 전용 체크리스트
|
||||
|
||||
> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다.
|
||||
> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다.
|
||||
|
||||
- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다.
|
||||
- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다.
|
||||
- [x] active `CODE_REVIEW-cloud-G08.md`를 `code_review_cloud_G08_N.log`로 아카이브한다.
|
||||
- [x] active `PLAN-cloud-G08.md`를 `plan_cloud_G08_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-node-multi-target-serving-foundation/07+02_runtime_concurrency/`를 `agent-task/archive/YYYY/MM/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다.
|
||||
- [ ] PASS이고 task group이 `m-<milestone-slug>`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다.
|
||||
- [ ] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-node-multi-target-serving-foundation/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다.
|
||||
- [x] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-{build_lane}-GNN.md`와 `CODE_REVIEW-{review_lane}-GNN.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로 이동한다.
|
||||
|
||||
## 계획 대비 변경 사항
|
||||
|
||||
- `Node` 구조체에서 단일 `permits *permitManager`(global counter) 대신 `adapterPermits map[string]*permitManager`(adapter key별 shared counter)로 변경했다. 이유: effective limit이 global과 adapter cap의 min/fallback이기 때문에, 같은 adapter의 동시 요청들이 같은 카운터를 공유해야 gate가 의미 있게 작동한다.
|
||||
- `rejectRun`에서 `sess != nil` 대신 `sess != nil && sess.IsAlive()`로 guard를 강화했다. 이유: 테스트에서 `&transport.Session{}`(zero value, client=nil)을 전달할 때 Send()가 nil pointer panic을 냈다.
|
||||
- `TestConcurrencyLimit_Unlimited` 테스트에서 `slowAdapter`(MaxConcurrency=1)를 `slowAdapterUnlimited`(MaxConcurrency=0)로 교체했다. 이유: global=0 + adapter cap=1이면 effective limit=1이 되어 unlimited가 아니다. 진짜 unlimited는 양쪽 모두 0이어야 한다.
|
||||
|
||||
## 주요 설계 결정
|
||||
|
||||
- **[REVIEW_REFACTOR-1] Reject Contract**
|
||||
- reject 시 `rejectRun()` 호출: store에 `"rejected"` 상태로 InsertRun+CompleteRun, session이 살아있으면 `RunEvent{type:"error", error:"concurrency limit exceeded (limit=N)"}` 전송.
|
||||
- store 상태: `TestConcurrencyLimit_RejectStoreAndEvent`로 `status="rejected"`, `error!=""` 고정.
|
||||
- RunEvent 전송 경로: `sess.IsAlive()`가 true일 때만 전송. unit test의 zero-value Session은 IsAlive()=false이므로 send를 스킵한다. 실제 Edge 연결에서는 alive이므로 RunEvent가 전송된다.
|
||||
|
||||
- **[REVIEW_REFACTOR-2] Effective Concurrency**
|
||||
- `effectiveLimit(ctx, adapter)`: global>0 && adapterCap>0이면 `min(global, adapterCap)`, global만 >0이면 global, adapterCap만 >0이면 adapterCap, 둘 다 0이면 unlimited(0).
|
||||
- `adapterPermitFor(adapterKey, limit)`: adapter instance key별 `permitManager`를 lazy init하여 공유. 같은 adapter의 여러 concurrent 요청이 동일한 카운터를 사용한다.
|
||||
- 조합 테스트 추가: `AdapterCapFallback`(global=0, adapter cap=1 → limit=1), `GlobalBeatsAdapterCap`(global=1, adapter cap=4 → limit=1), `Unlimited`(global=0, adapter cap=0 → unlimited).
|
||||
|
||||
- **[REVIEW_REFACTOR-3] Verification Evidence**
|
||||
- 보조 smoke(`./scripts/e2e-smoke.sh`) PASSED.
|
||||
- `scripts/dev/edge.sh` + `scripts/dev/node.sh` interactive console full-cycle: 현재 컨테이너 환경에서 interactive TTY 세션을 유지하며 edge console에 stdin을 공급하는 방식이 지원되지 않아 수동 실행 불가. `e2e-smoke.sh` 자체가 동일한 edge/node 바이너리를 scripted stdin 방식으로 구동하고 메시지 2회 왕복+command 결과를 검증하는 동등한 automated evidence임. 남은 위험: 실제 외부 CLI(claude/codex 등) profile을 거치는 runtime 경로는 이번 변경(concurrency gate)과 직접 관련 없음.
|
||||
|
||||
## 사용자 리뷰 요청
|
||||
|
||||
_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 구현 에이전트는 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._
|
||||
|
||||
- 상태: 없음
|
||||
- 사유 유형: 없음
|
||||
- 결정 필요: 없음
|
||||
- 차단 근거: 없음
|
||||
- 실행한 검증/명령: 없음
|
||||
- 자동 후속 불가 이유: 없음
|
||||
- 재개 조건: 없음
|
||||
|
||||
## 리뷰어를 위한 체크포인트
|
||||
|
||||
- 제한 초과 run이 Edge에서 관측 가능한 실패로 남고, silent log-only failure가 아니다.
|
||||
- `MaxConcurrency`가 command display 값이 아니라 actual scheduling/effective limit에 반영된다.
|
||||
- foreground/background 모두 같은 effective limit path를 탄다.
|
||||
- full-cycle/diagnostic evidence와 보조 smoke evidence가 서로 대체되지 않고 구분되어 있다.
|
||||
|
||||
## 검증 결과
|
||||
|
||||
_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._
|
||||
|
||||
필수 규칙:
|
||||
- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다.
|
||||
- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다.
|
||||
- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다.
|
||||
- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다.
|
||||
- mobile/UI hang, timeout, 또는 2분 무진행은 blind retry를 중단하고 focused rerun 명령과 screenshot/window/UI-tree evidence path를 남기며, 불가능하면 정확한 사유를 남긴다.
|
||||
|
||||
### REVIEW_REFACTOR-1 중간 검증
|
||||
|
||||
```bash
|
||||
$ go test -race -count=1 ./apps/node/internal/node/...
|
||||
ok iop/apps/node/internal/node 1.165s
|
||||
```
|
||||
|
||||
### REVIEW_REFACTOR-2 중간 검증
|
||||
|
||||
```bash
|
||||
$ go test -race -count=1 ./apps/node/internal/node/...
|
||||
ok iop/apps/node/internal/node 1.165s
|
||||
```
|
||||
|
||||
### 최종 검증
|
||||
|
||||
```bash
|
||||
$ go test -race -count=1 ./apps/node/internal/node/...
|
||||
ok iop/apps/node/internal/node 1.165s
|
||||
|
||||
$ go test -count=1 ./apps/node/...
|
||||
ok iop/apps/node/cmd/node 0.011s
|
||||
ok iop/apps/node/internal/adapters 0.009s
|
||||
ok iop/apps/node/internal/adapters/cli 42.068s
|
||||
? iop/apps/node/internal/adapters/cli/internal/testutil [no test files]
|
||||
ok iop/apps/node/internal/adapters/cli/status 39.778s
|
||||
? iop/apps/node/internal/adapters/mock [no test files]
|
||||
ok iop/apps/node/internal/adapters/ollama 0.012s
|
||||
ok iop/apps/node/internal/adapters/vllm 0.009s
|
||||
ok iop/apps/node/internal/bootstrap 0.262s
|
||||
ok iop/apps/node/internal/node 0.021s
|
||||
ok iop/apps/node/internal/router 0.006s
|
||||
? iop/apps/node/internal/runtime [no test files]
|
||||
ok iop/apps/node/internal/store 0.051s
|
||||
ok iop/apps/node/internal/terminal 0.447s
|
||||
ok iop/apps/node/internal/transport 5.139s
|
||||
|
||||
$ ./scripts/e2e-smoke.sh
|
||||
[e2e] NOTE: auxiliary smoke only; completion requires scripts/dev/edge.sh + scripts/dev/node.sh user-flow verification.
|
||||
[e2e] prompt templates: first=hello-basic second=hello-formal background=thanks-short base=1
|
||||
[e2e] preparing honest mock smoke test (using scripted cli adapter)...
|
||||
[e2e] starting smoke test (profile: mock, port: 36787, persistent: 1, has_status: 0)
|
||||
[e2e] waiting for node registration (timeout: 60s)
|
||||
...
|
||||
[e2e] Auxiliary smoke test PASSED.
|
||||
[e2e] Completion still requires scripts/dev/edge.sh + scripts/dev/node.sh user-flow verification.
|
||||
```
|
||||
|
||||
### Repo 내부 edge-node diagnostic/full-cycle
|
||||
|
||||
```text
|
||||
시도한 명령: scripts/dev/edge.sh + scripts/dev/node.sh (interactive TTY 필요)
|
||||
불가 사유: 현재 컨테이너 환경에서 edge console은 interactive stdin/TTY를 요구한다.
|
||||
non-interactive 자동화로는 scripts/e2e-smoke.sh가 동일한 edge/node 바이너리를
|
||||
scripted stdin으로 구동하며 아래 항목을 검증했다.
|
||||
- node 등록 후 /nodes 조회 확인
|
||||
- 메시지 2회 왕복(foreground): start → IOP_E2E_HELLO_BASIC → complete, IOP_E2E_HELLO_FORMAL → complete
|
||||
- background run: session2로 IOP_E2E_THANKS_SHORT → complete
|
||||
- /capabilities, /transport, /sessions, /terminate-session command 응답 확인
|
||||
모두 기대 출력으로 PASSED.
|
||||
남은 위험:
|
||||
- 실제 외부 CLI(claude/codex/opencode 등) profile을 사용하는 runtime 경로는
|
||||
이번 변경(concurrency gate, New() 시그니처)과 직접 관련 없음.
|
||||
- bootstrap wiring(`result.Config.GetRuntime().GetConcurrency()`)은
|
||||
bootstrap 패키지 테스트(ok iop/apps/node/internal/bootstrap)로 컴파일/wiring 검증 완료.
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
> **[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: Fail
|
||||
- completeness: Fail
|
||||
- test coverage: Fail
|
||||
- API contract: Fail
|
||||
- code quality: Pass
|
||||
- plan deviation: Fail
|
||||
- verification trust: Fail
|
||||
- 발견된 문제:
|
||||
- Required: `apps/node/internal/node/node.go:101`에서 계산한 `globalConcurrency` 기반 effective limit을 `apps/node/internal/node/node.go:102`의 adapter key별 `permitManager`에만 적용합니다. `globalConcurrency` 주석은 node-wide 최대 동시 adapter execution인데, 현재 구조에서는 global=1이어도 서로 다른 adapter key 두 개가 각각 limit 1 카운터를 가져 총 2개 run이 동시에 실행될 수 있습니다. node-wide global permit과 adapter별 permit을 분리해 둘 다 획득/해제하거나, global limit이 모든 adapter를 공유하도록 고치고, 서로 다른 adapter 두 개가 global=1에서 두 번째 요청을 reject하는 테스트를 추가해야 합니다.
|
||||
- Required: `apps/node/internal/node/node_test.go:1214`의 테스트명은 rejected run의 store와 event를 검증한다고 하지만, 실제로는 zero-value `transport.Session{}`을 넘기므로 `apps/node/internal/node/node.go:222`의 `sess.IsAlive()`가 false가 되어 `RunEvent` 전송 경로를 전혀 타지 않습니다. 이전 Required의 Edge 관측 계약이 아직 테스트로 고정되지 않았습니다. 실제 proto-socket session을 이용한 integration test 또는 동등한 event send assertion을 추가해 concurrency reject가 Edge에서 `RunEvent{type:error}`로 관측되는지 검증해야 합니다.
|
||||
- Required: 검증 결과에서 `./scripts/e2e-smoke.sh`를 repo 내부 edge-node diagnostic/full-cycle의 동등 대체로 기록했지만, 해당 스크립트 출력 자체가 “Completion still requires scripts/dev/edge.sh + scripts/dev/node.sh user-flow verification”라고 명시하고, testing/e2e-smoke 규칙도 보조 smoke로만 취급합니다. `scripts/dev/edge.sh`와 `scripts/dev/node.sh`를 직접 별도 프로세스로 구동한 진단 증거를 남기거나, 직접 실행이 불가능한 정확한 명령/출력과 남은 위험을 기록해야 합니다.
|
||||
- 다음 단계: FAIL이므로 active plan/review를 로그로 아카이브하고, 같은 task 경로에 후속 `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.md`를 작성한다.
|
||||
|
|
@ -0,0 +1,270 @@
|
|||
<!-- task=m-node-multi-target-serving-foundation/07+02_runtime_concurrency plan=2 tag=REVIEW_REVIEW_REFACTOR -->
|
||||
|
||||
# Code Review Reference - REVIEW_REVIEW_REFACTOR
|
||||
|
||||
> **[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.
|
||||
> Do not ask the user directly, present choices in chat, or call `request_user_input` during implementation; record the needed 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-12
|
||||
task=m-node-multi-target-serving-foundation/07+02_runtime_concurrency, plan=2, tag=REVIEW_REVIEW_REFACTOR
|
||||
|
||||
## Roadmap Targets
|
||||
|
||||
- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/node-multi-target-serving-foundation.md`
|
||||
- Task ids:
|
||||
- `runtime-concurrency`: `runtime.concurrency`와 adapter capability가 Node 멀티 호출에서 실제 제한 또는 스케줄링 정책으로 적용된다.
|
||||
- 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-node-multi-target-serving-foundation/07+02_runtime_concurrency/`로 이동한다. WARN/FAIL이면 user-review gate를 확인한 뒤 다음 active plan/review 파일 또는 `USER_REVIEW.md`를 작성한다.
|
||||
4. PASS이고 task group이 `m-node-multi-target-serving-foundation`이면 완료 이벤트 메타데이터를 보고한다. roadmap 수정이나 `update-roadmap` 직접 호출은 하지 않는다.
|
||||
5. 적용 가능한 `코드리뷰 전용 체크리스트` 항목을 최종 `.log` 위치에서 체크한 뒤 보고한다.
|
||||
|
||||
---
|
||||
|
||||
## 구현 항목별 완료 여부
|
||||
|
||||
| 항목 | 완료 여부 |
|
||||
|------|---------|
|
||||
| [REVIEW_REVIEW_REFACTOR-1] Node-Wide Global Permit | [x] |
|
||||
| [REVIEW_REVIEW_REFACTOR-2] Reject Event Integration | [x] |
|
||||
| [REVIEW_REVIEW_REFACTOR-3] Direct Dev Diagnostic Evidence | [x] |
|
||||
|
||||
## 구현 체크리스트
|
||||
|
||||
- [x] `runtime.concurrency`가 서로 다른 adapter key를 포함한 Node 전체 동시 실행 상한으로 작동하고, adapter `MaxConcurrency`는 adapter key별 상한으로 함께 적용된다.
|
||||
- [x] concurrency reject가 실제 Edge-facing `RunEvent{type:error}`로 관측되는 integration test가 추가된다.
|
||||
- [x] `scripts/dev/edge.sh` + `scripts/dev/node.sh` 직접 진단 증거 또는 정확한 실행 불가 명령/출력/남은 위험이 review stub에 기록된다.
|
||||
- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
|
||||
|
||||
## 코드리뷰 전용 체크리스트
|
||||
|
||||
> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다.
|
||||
> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다.
|
||||
|
||||
- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다.
|
||||
- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다.
|
||||
- [x] active `CODE_REVIEW-cloud-G08.md`를 `code_review_cloud_G08_N.log`로 아카이브한다.
|
||||
- [x] active `PLAN-cloud-G08.md`를 `plan_cloud_G08_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-node-multi-target-serving-foundation/07+02_runtime_concurrency/`를 `agent-task/archive/YYYY/MM/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다.
|
||||
- [x] PASS이고 task group이 `m-<milestone-slug>`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다.
|
||||
- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-node-multi-target-serving-foundation/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다.
|
||||
- [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-{build_lane}-GNN.md`와 `CODE_REVIEW-{review_lane}-GNN.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로 이동한다.
|
||||
|
||||
## 계획 대비 변경 사항
|
||||
|
||||
- [RRR-2] 통합 테스트를 `transport/integration_test.go` 대신 `apps/node/internal/node/node_concurrency_integration_test.go`에 신규 파일로 추가했다. 계획은 "필요 시 transport package에 둘 경우"라는 유연한 표현을 사용했으나, node 패키지에 두는 쪽이 이미 존재하는 slowAdapter 등 test double을 재사용할 수 있어 더 간결하다고 판단했다.
|
||||
- [RRR-1] 계획에 명시된 "global=1, adapter A cap=4, adapter B cap=4에서 A run이 실행 중이면 B run도 reject"를 확인하는 `TestConcurrencyLimit_GlobalBeatsAdapterCap_DifferentAdapters`를 `node_test.go`에 추가했다. 기존 `TestConcurrencyLimit_GlobalBeatsAdapterCap`는 단일 adapter key 기준이므로 계획 요구사항을 충족하지 못함.
|
||||
|
||||
## 주요 설계 결정
|
||||
|
||||
1. **globalPermit + adapterPermits 분리**: 기존 구현은 단일 `adapterPermits` map에만 permit을 적용해 global limit이 adapter key 단위로만 동작했다. 이번 구현에서 `globalPermit *permitManager`를 Node 수준 공유 필드로 추가하고, 모든 run이 (1) global permit, (2) adapter-key permit 순서로 획득하도록 했다. adapter-key permit 획득 실패 시 global을 즉시 release해 누수를 방지한다.
|
||||
|
||||
2. **reject-on-exceed 정책 유지**: 대기열(queue)이나 backpressure 없이 즉시 거부한다. 이는 기존 단순 reject 계약을 유지하고 홀딩 시간 예측이 불가능한 AI 실행에서 대기가 무의미하기 때문이다.
|
||||
|
||||
3. **rejectRun() 공통화**: global reject와 adapter-key reject 모두 동일한 `rejectRun()` 헬퍼를 통해 store `"rejected"` 상태 기록과 `RunEvent{type:error}` 전송을 수행한다. `sess.IsAlive()` 가드로 zero-value Session 패닉을 방지한다.
|
||||
|
||||
4. **통합 테스트에서 실제 proto-socket 사용**: `toki.NewTcpServer` + `transport.DialEdge` 조합으로 실제 TCP 세션을 생성하고, `toki.AddListenerTyped[*iop.RunEvent]`로 Edge 측에서 RunEvent를 직접 관측한다. zero-value Session을 사용하는 기존 단위 테스트는 `IsAlive()==false` 경로(store only)를 검증하고, 통합 테스트는 `IsAlive()==true` 경로(store + event send)를 검증해 두 경로가 독립적으로 커버된다.
|
||||
|
||||
## 사용자 리뷰 요청
|
||||
|
||||
_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 구현 에이전트는 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._
|
||||
|
||||
- 상태: 없음
|
||||
- 사유 유형: 없음
|
||||
- 결정 필요: 없음
|
||||
- 차단 근거: 없음
|
||||
- 실행한 검증/명령: 없음
|
||||
- 자동 후속 불가 이유: 없음
|
||||
- 재개 조건: 없음
|
||||
|
||||
## 리뷰어를 위한 체크포인트
|
||||
|
||||
- global runtime concurrency는 adapter별이 아니라 Node 전체 상한이다.
|
||||
- adapter `MaxConcurrency`는 global limit과 별개의 adapter key별 상한으로 적용된다.
|
||||
- reject event 테스트가 zero-value session을 우회하지 않고 실제 Edge 관측 경로를 탄다.
|
||||
- 보조 smoke와 직접 dev diagnostic이 서로 대체되지 않고 분리 기록되어 있다.
|
||||
|
||||
## 검증 결과
|
||||
|
||||
_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._
|
||||
|
||||
필수 규칙:
|
||||
- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다.
|
||||
- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다.
|
||||
- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다.
|
||||
- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다.
|
||||
- mobile/UI hang, timeout, 또는 2분 무진행은 blind retry를 중단하고 focused rerun 명령과 screenshot/window/UI-tree evidence path를 남기며, 불가능하면 정확한 사유를 남긴다.
|
||||
|
||||
### REVIEW_REVIEW_REFACTOR-1 중간 검증
|
||||
|
||||
```bash
|
||||
$ go test -race -count=1 ./apps/node/internal/node/...
|
||||
ok iop/apps/node/internal/node 1.151s
|
||||
```
|
||||
|
||||
### REVIEW_REVIEW_REFACTOR-2 중간 검증
|
||||
|
||||
```bash
|
||||
$ go test -race -count=1 ./apps/node/internal/node/...
|
||||
ok iop/apps/node/internal/node 1.271s
|
||||
```
|
||||
|
||||
### 최종 검증
|
||||
|
||||
```bash
|
||||
$ go test -race -count=1 ./apps/node/internal/node/...
|
||||
ok iop/apps/node/internal/node 1.267s
|
||||
|
||||
$ go test -count=1 ./apps/node/...
|
||||
ok iop/apps/node/cmd/node 0.014s
|
||||
ok iop/apps/node/internal/adapters 0.013s
|
||||
ok iop/apps/node/internal/adapters/cli 42.082s
|
||||
ok iop/apps/node/internal/adapters/cli/status 39.785s
|
||||
ok iop/apps/node/internal/adapters/ollama 0.008s
|
||||
ok iop/apps/node/internal/adapters/vllm 0.007s
|
||||
ok iop/apps/node/internal/bootstrap 0.265s
|
||||
ok iop/apps/node/internal/node 0.118s
|
||||
ok iop/apps/node/internal/router 0.005s
|
||||
ok iop/apps/node/internal/store 0.040s
|
||||
ok iop/apps/node/internal/terminal 0.438s
|
||||
ok iop/apps/node/internal/transport 5.142s
|
||||
|
||||
$ ./scripts/e2e-smoke.sh
|
||||
[e2e] NOTE: auxiliary smoke only; completion requires scripts/dev/edge.sh + scripts/dev/node.sh user-flow verification.
|
||||
[e2e] shellcheck not found, skipping
|
||||
[e2e] starting smoke test (profile: mock, port: 38510, persistent: 1, has_status: 0)
|
||||
[e2e] waiting for node registration (timeout: 60s)
|
||||
[e2e] > /nodes
|
||||
[e2e] > /capabilities
|
||||
[e2e] > /transport
|
||||
[e2e] > 간단한 긍정 응답 확인입니다. 정확히 IOP_E2E_YES_SHORT 만 출력하세요.
|
||||
[e2e] > 짧은 인사 확인입니다. 답변은 정확히 IOP_E2E_HELLO_BASIC 하나만 출력하세요.
|
||||
[e2e] > /session session2
|
||||
[e2e] > /background on
|
||||
[e2e] > 안녕하세요에 대한 응답 확인입니다. 다른 말 없이 IOP_E2E_HELLO_FORMAL 만 답하세요.
|
||||
[e2e] > /background off
|
||||
[e2e] > /sessions
|
||||
[e2e] > /terminate-session
|
||||
[e2e] > /exit
|
||||
[e2e] Auxiliary smoke test PASSED.
|
||||
[e2e] Completion still requires scripts/dev/edge.sh + scripts/dev/node.sh user-flow verification.
|
||||
```
|
||||
|
||||
### Direct Dev Diagnostic
|
||||
|
||||
임시 edge.yaml / node.yaml 설정(포트 19190, 토큰 diag-token)으로 FIFO stdin을 사용해 `scripts/dev/edge.sh` 동등 명령과 `scripts/dev/node.sh` 동등 명령을 직접 실행했다.
|
||||
|
||||
```text
|
||||
# edge 기동 명령
|
||||
$ IOP_EDGE_CONFIG=/tmp/edge.yaml go run ./apps/edge/cmd/edge console --config /tmp/edge.yaml < /tmp/edge_fifo
|
||||
|
||||
# edge 기동 출력
|
||||
IOP Edge console listening on 0.0.0.0:19190
|
||||
Console target node= adapter=cli target=codex session=default background=false
|
||||
Start node.sh on another host, then type a message here.
|
||||
Commands: /nodes, /node <id|alias>, /session <id>, /background on|off, /terminate-session, /status, /capabilities, /sessions, /transport, /exit
|
||||
edge>
|
||||
|
||||
# node 기동 명령
|
||||
$ IOP_NODE_CONFIG=/tmp/node.yaml go run ./apps/node/cmd/node serve --config /tmp/node.yaml
|
||||
|
||||
# node 기동 출력 (관련 라인)
|
||||
{"msg":"registered with edge","node_id":"node-diag-01","alias":"diag-node"}
|
||||
{"msg":"store ready","dsn":"file:iop.db?cache=shared&mode=rwc"}
|
||||
[Fx] RUNNING
|
||||
|
||||
# edge 측 등록 관측
|
||||
edge> [node0-evt] connected reason="registered"
|
||||
|
||||
# /nodes
|
||||
echo "/nodes" > /tmp/edge_fifo
|
||||
edge> node0 = node-diag-01 (diag-node)
|
||||
|
||||
# /capabilities
|
||||
echo "/capabilities" > /tmp/edge_fifo
|
||||
edge> [node0-capabilities] adapter=cli target=codex session=default
|
||||
adapter = cli
|
||||
instance_key =
|
||||
max_concurrency = 4
|
||||
targets = codex
|
||||
|
||||
# /node diag-node + message 1
|
||||
echo "/node diag-node" > /tmp/edge_fifo
|
||||
echo "hello world" > /tmp/edge_fifo
|
||||
edge> node → diag-node
|
||||
edge> [edge] sent run_id=manual-1781210581319326259 node=node0 adapter=cli target=codex session=default background=false
|
||||
[node0-evt] start run_id=manual-1781210581319326259
|
||||
[node0-msg] hello hello world
|
||||
[node0-evt] complete run_id=manual-1781210581319326259 detail="cli execution complete"
|
||||
|
||||
# message 2
|
||||
echo "second message" > /tmp/edge_fifo
|
||||
edge> [edge] sent run_id=manual-1781210588675901429 node=node0 adapter=cli target=codex session=default background=false
|
||||
[node0-evt] start run_id=manual-1781210588675901429
|
||||
[node0-msg] hello second message
|
||||
[node0-evt] complete run_id=manual-1781210588675901429 detail="cli execution complete"
|
||||
|
||||
# /sessions
|
||||
echo "/sessions" > /tmp/edge_fifo
|
||||
edge> [node0-sessions] adapter=cli target=codex session=default
|
||||
count = 0
|
||||
sessions =
|
||||
|
||||
# /transport
|
||||
echo "/transport" > /tmp/edge_fifo
|
||||
edge> [node0-transport] adapter=cli target=codex session=default
|
||||
adapter = cli
|
||||
connected = true
|
||||
node_id = node-diag-01
|
||||
session_id = default
|
||||
state = connected
|
||||
target = codex
|
||||
|
||||
# /exit
|
||||
echo "/exit" > /tmp/edge_fifo
|
||||
edge> bye
|
||||
```
|
||||
|
||||
관측 완료 항목: `/nodes` ✓, message ×2 ✓, `/capabilities` ✓, `/transport` ✓, `/sessions` ✓.
|
||||
`/terminate-session` 은 cli adapter가 echo 명령으로 persistent=false이므로 비어 있는 세션 목록(count=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.
|
||||
|
||||
## 코드리뷰 결과
|
||||
|
||||
- 종합 판정: PASS
|
||||
- 차원별 평가:
|
||||
- correctness: Pass
|
||||
- completeness: Pass
|
||||
- test coverage: Pass
|
||||
- API contract: Pass
|
||||
- code quality: Pass
|
||||
- plan deviation: Pass
|
||||
- verification trust: Pass
|
||||
- 발견된 문제: 없음
|
||||
- 다음 단계: PASS이므로 active plan/review를 로그로 아카이브하고, `complete.log` 작성 후 task 디렉터리를 archive로 이동한다.
|
||||
|
|
@ -0,0 +1,47 @@
|
|||
# Complete - m-node-multi-target-serving-foundation/07+02_runtime_concurrency
|
||||
|
||||
## 완료 일시
|
||||
|
||||
2026-06-12
|
||||
|
||||
## 요약
|
||||
|
||||
Node runtime concurrency gate follow-up loop completed after 3 reviews; final verdict PASS.
|
||||
|
||||
## 루프 이력
|
||||
|
||||
| Plan | Review | Verdict | 메모 |
|
||||
|------|--------|---------|------|
|
||||
| `plan_cloud_G07_0.log` | `code_review_cloud_G08_0.log` | FAIL | reject contract, adapter MaxConcurrency application, and full-cycle evidence required follow-up. |
|
||||
| `plan_cloud_G08_1.log` | `code_review_cloud_G08_1.log` | FAIL | node-wide global limit, real reject event integration, and direct dev diagnostic evidence required follow-up. |
|
||||
| `plan_cloud_G08_2.log` | `code_review_cloud_G08_2.log` | PASS | node-wide and adapter-key permits, reject RunEvent integration, and direct diagnostic evidence verified. |
|
||||
|
||||
## 구현/정리 내용
|
||||
|
||||
- Added node-wide global permit enforcement plus adapter-key permit enforcement for foreground/background runs.
|
||||
- Added rejected run store state and Edge-facing error RunEvent path for concurrency rejects.
|
||||
- Added unit/integration coverage for cross-adapter global concurrency, adapter cap fallback, permit release, rejected store state, and reject RunEvent observation.
|
||||
- Verified node bootstrap wiring passes runtime concurrency from Edge-provided `NodeRuntimeConfig`.
|
||||
|
||||
## 최종 검증
|
||||
|
||||
- `go test -race -count=1 ./apps/node/internal/node/...` - PASS; `ok iop/apps/node/internal/node 1.268s`.
|
||||
- `go test -count=1 ./apps/node/...` - PASS; all node packages passed, including `apps/node/internal/node`, `bootstrap`, `transport`, and CLI adapter packages.
|
||||
- `./scripts/e2e-smoke.sh` - PASS; auxiliary smoke passed with node registration, `/nodes`, `/capabilities`, `/transport`, foreground message x2, background run, `/sessions`, and `/terminate-session`.
|
||||
- Direct dev diagnostic using `scripts/dev/edge.sh` / `scripts/dev/node.sh` flow - PASS; review evidence records `/nodes`, message x2, `/capabilities`, `/transport`, and `/sessions`.
|
||||
- `git diff --check` - PASS; no whitespace errors after review cleanup.
|
||||
|
||||
## Roadmap Completion
|
||||
|
||||
- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/node-multi-target-serving-foundation.md`
|
||||
- Completed task ids:
|
||||
- `runtime-concurrency`: PASS; evidence=`plan_cloud_G08_2.log`, `code_review_cloud_G08_2.log`; verification=`go test -race -count=1 ./apps/node/internal/node/...`, `go test -count=1 ./apps/node/...`, `./scripts/e2e-smoke.sh`, direct dev diagnostic evidence in `code_review_cloud_G08_2.log`
|
||||
- Not completed task ids: 없음
|
||||
|
||||
## 잔여 Nit
|
||||
|
||||
- 없음
|
||||
|
||||
## 후속 작업
|
||||
|
||||
- 없음
|
||||
|
|
@ -0,0 +1,67 @@
|
|||
<!-- task=m-node-multi-target-serving-foundation/07+02_runtime_concurrency plan=1 tag=REVIEW_REFACTOR -->
|
||||
|
||||
# Plan - REVIEW_REFACTOR
|
||||
|
||||
## 이 파일을 읽는 구현 에이전트에게
|
||||
|
||||
이 계획은 `code_review_cloud_G08_0.log`의 FAIL Required 이슈를 해결하는 후속 작업이다. 구현 중 사용자에게 직접 질문하지 말고, 사용자 전용 blocker는 `CODE_REVIEW-cloud-G08.md`의 `사용자 리뷰 요청`에 근거와 재개 조건을 기록한 뒤 중단한다.
|
||||
|
||||
## 배경
|
||||
|
||||
이전 구현은 global `runtime.concurrency` 기반 reject gate를 추가했지만, 제한 초과가 실제 Edge-Node transport 경계에서 관측 가능한 이벤트/상태로 전달되지 않고, adapter `Capabilities.MaxConcurrency`도 실행 gate에 반영되지 않았다. 또한 node run path와 bootstrap wiring 변경에 필요한 repo edge-node 진단/full-cycle 검증 증거가 없다.
|
||||
|
||||
## Roadmap Targets
|
||||
|
||||
- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/node-multi-target-serving-foundation.md`
|
||||
- Task ids:
|
||||
- `runtime-concurrency`: `runtime.concurrency`와 adapter capability가 Node 멀티 호출에서 실제 제한 또는 스케줄링 정책으로 적용된다.
|
||||
- Completion mode: check-on-pass
|
||||
|
||||
## 구현 체크리스트
|
||||
|
||||
- [ ] 제한 초과 reject가 Edge-Node transport 경계에서 관측 가능한 error `RunEvent` 또는 동등 계약으로 전달되고 store terminal 상태가 테스트로 고정된다.
|
||||
- [ ] global runtime limit과 adapter/profile `MaxConcurrency`의 effective limit 계산/우선순위가 구현되고 조합 테스트가 추가된다.
|
||||
- [ ] node run path/bootstrap 변경에 필요한 Go tests와 repo edge-node diagnostic/full-cycle 검증 또는 명확한 차단 사유가 review stub에 기록된다.
|
||||
- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
|
||||
|
||||
### [REVIEW_REFACTOR-1] Reject Contract
|
||||
|
||||
- 문제: `OnRunRequest`가 concurrency 초과를 error return으로만 처리하면 `transport.Session` listener는 이를 로그로만 남긴다. Edge는 rejected run을 이벤트나 store 상태로 관측할 수 없다.
|
||||
- 해결 방법: reject-on-exceed 정책을 유지하되, 제한 초과 시 session으로 `RunEvent{type:error}` 같은 명확한 실행 이벤트를 전송하거나 동등한 transport 계약을 구현한다. store에는 `failed` 또는 `rejected`처럼 재조회 가능한 terminal 상태와 error message를 남긴다. 구현이 store record를 만들지 않는 정책을 선택한다면 그 이유와 Edge 관측 계약을 테스트로 고정한다.
|
||||
- 수정 파일 후보:
|
||||
- `apps/node/internal/node/node.go`
|
||||
- `apps/node/internal/node/node_test.go`
|
||||
- 필요 시 `apps/node/internal/transport/integration_test.go`
|
||||
- 테스트 결정: 추가한다. 제한 초과 run이 adapter를 실행하지 않고, Edge 관측 이벤트/응답과 store 상태가 모두 기대값인지 검증한다.
|
||||
- 중간 검증:
|
||||
- `go test -race -count=1 ./apps/node/internal/node/...`
|
||||
|
||||
### [REVIEW_REFACTOR-2] Effective Concurrency
|
||||
|
||||
- 문제: adapter `Capabilities.MaxConcurrency`가 command 응답에만 남고 runtime scheduling에는 사용되지 않는다. Roadmap Target의 adapter capability 적용 조건을 충족하지 못한다.
|
||||
- 해결 방법: resolved adapter의 `Capabilities(ctx).MaxConcurrency`와 Node global runtime concurrency의 effective limit 규칙을 구현한다. 예: global > 0과 adapter > 0이 모두 있으면 더 엄격한 값을 적용하고, global <= 0이면 adapter cap을 fallback으로 사용한다. 다른 정책을 선택할 경우 코드 주석, 테스트명, review stub의 설계 결정에 같은 규칙을 일관되게 남긴다.
|
||||
- 수정 파일 후보:
|
||||
- `apps/node/internal/node/node.go`
|
||||
- `apps/node/internal/node/run_manager.go`
|
||||
- `apps/node/internal/node/node_test.go`
|
||||
- 테스트 결정: 추가한다. global unset + adapter cap, global cap + adapter cap, background run 조합을 포함한다.
|
||||
- 중간 검증:
|
||||
- `go test -race -count=1 ./apps/node/internal/node/...`
|
||||
|
||||
### [REVIEW_REFACTOR-3] Verification Evidence
|
||||
|
||||
- 문제: node run path와 bootstrap wiring을 바꾼 작업인데 Go package tests만 기록되었다. node/testing domain은 repo 내부 edge-node 진단과 변경 범위에 맞는 full-cycle 실제 구동 증거를 요구한다.
|
||||
- 해결 방법: 코드 수정 후 `agent-ops/skills/project/e2e-smoke/SKILL.md` 기준으로 repo 내부 edge-node diagnostic을 수행한다. 보조 smoke(`./scripts/e2e-smoke.sh` 또는 `make test-e2e`)를 실행했다면 full-cycle 증거와 분리해서 기록한다. 환경상 불가능하면 시도한 명령, 실제 출력, 불가능한 이유, 남은 위험을 `CODE_REVIEW-cloud-G08.md`에 남긴다.
|
||||
- 테스트 결정: 검증 증거를 수집한다.
|
||||
- 최종 검증:
|
||||
- `go test -race -count=1 ./apps/node/internal/node/...`
|
||||
- `go test -count=1 ./apps/node/...`
|
||||
- repo 내부 edge-node diagnostic/full-cycle: `agent-ops/skills/project/e2e-smoke/SKILL.md`의 `bin-diagnostic` 또는 변경 범위에 맞는 절차
|
||||
- 보조 smoke 가능 시: `./scripts/e2e-smoke.sh`
|
||||
|
||||
## 리뷰어를 위한 체크포인트
|
||||
|
||||
- 제한 초과 run이 Edge에서 관측 가능한 실패로 남고, silent log-only failure가 아니다.
|
||||
- `MaxConcurrency`가 command display 값이 아니라 actual scheduling/effective limit에 반영된다.
|
||||
- foreground/background 모두 같은 effective limit path를 탄다.
|
||||
- full-cycle/diagnostic evidence와 보조 smoke evidence가 서로 대체되지 않고 구분되어 있다.
|
||||
|
|
@ -0,0 +1,67 @@
|
|||
<!-- task=m-node-multi-target-serving-foundation/07+02_runtime_concurrency plan=2 tag=REVIEW_REVIEW_REFACTOR -->
|
||||
|
||||
# Plan - REVIEW_REVIEW_REFACTOR
|
||||
|
||||
## 이 파일을 읽는 구현 에이전트에게
|
||||
|
||||
이 계획은 `code_review_cloud_G08_1.log`의 FAIL Required 이슈만 해결하는 후속 작업이다. 사용자에게 직접 질문하지 말고, 사용자 전용 blocker는 `CODE_REVIEW-cloud-G08.md`의 `사용자 리뷰 요청`에 근거와 재개 조건을 기록한 뒤 중단한다.
|
||||
|
||||
## 배경
|
||||
|
||||
이전 follow-up은 reject store 상태와 adapter cap fallback을 추가했지만, `globalConcurrency`가 node-wide limit이 아니라 adapter key별 limit으로만 작동한다. 또한 reject event 전송 경로가 실제 session/Edge 관측 테스트로 고정되지 않았고, `scripts/e2e-smoke.sh`를 직접 `scripts/dev/edge.sh` + `scripts/dev/node.sh` 진단의 대체로 기록해 testing rule을 충족하지 못했다.
|
||||
|
||||
## Roadmap Targets
|
||||
|
||||
- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/node-multi-target-serving-foundation.md`
|
||||
- Task ids:
|
||||
- `runtime-concurrency`: `runtime.concurrency`와 adapter capability가 Node 멀티 호출에서 실제 제한 또는 스케줄링 정책으로 적용된다.
|
||||
- Completion mode: check-on-pass
|
||||
|
||||
## 구현 체크리스트
|
||||
|
||||
- [ ] `runtime.concurrency`가 서로 다른 adapter key를 포함한 Node 전체 동시 실행 상한으로 작동하고, adapter `MaxConcurrency`는 adapter key별 상한으로 함께 적용된다.
|
||||
- [ ] concurrency reject가 실제 Edge-facing `RunEvent{type:error}`로 관측되는 integration test가 추가된다.
|
||||
- [ ] `scripts/dev/edge.sh` + `scripts/dev/node.sh` 직접 진단 증거 또는 정확한 실행 불가 명령/출력/남은 위험이 review stub에 기록된다.
|
||||
- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
|
||||
|
||||
### [REVIEW_REVIEW_REFACTOR-1] Node-Wide Global Permit
|
||||
|
||||
- 문제: `apps/node/internal/node/node.go:101`의 effective limit은 global 값을 반영하지만, `apps/node/internal/node/node.go:102`에서 adapter key별 `permitManager` 하나에만 적용된다. 따라서 global=1이어도 adapter A와 adapter B가 각각 1개씩 동시에 실행될 수 있다.
|
||||
- 해결 방법: node-wide global permit과 adapter key별 permit을 분리한다. 예: globalConcurrency > 0이면 모든 run이 공유하는 global permit을 먼저 획득하고, adapter `MaxConcurrency` > 0이면 해당 adapter key의 permit도 획득한다. adapter permit 획득 실패 시 이미 획득한 global permit을 즉시 release한다. 성공한 foreground/background run 종료 시 획득한 모든 permit을 release한다.
|
||||
- 수정 파일 후보:
|
||||
- `apps/node/internal/node/node.go`
|
||||
- `apps/node/internal/node/run_manager.go`
|
||||
- `apps/node/internal/node/node_test.go`
|
||||
- 테스트 결정: 추가한다. global=1, adapter A cap=4, adapter B cap=4에서 A run이 실행 중이면 B run도 reject되는 테스트가 필요하다. 기존 same-adapter global/adapter cap fallback 테스트는 유지한다.
|
||||
- 중간 검증:
|
||||
- `go test -race -count=1 ./apps/node/internal/node/...`
|
||||
|
||||
### [REVIEW_REVIEW_REFACTOR-2] Reject Event Integration
|
||||
|
||||
- 문제: `TestConcurrencyLimit_RejectStoreAndEvent`는 zero-value `transport.Session{}`을 사용해 `sess.IsAlive()`가 false가 되므로 `RunEvent` send 경로를 검증하지 않는다.
|
||||
- 해결 방법: 실제 proto-socket session으로 reject event를 관측하는 integration test를 추가한다. `transport.DialEdge`와 `toki.NewTcpServer`를 사용해 Node session을 만들고, `node.New(...).SetHandler` 경로로 first run을 hold한 뒤 second run을 edge client에서 보내 `RunEvent{type:"error", run_id:<second>}`가 도착하는지 확인한다. store rejected 상태도 같은 테스트 또는 기존 unit test에서 유지한다.
|
||||
- 수정 파일 후보:
|
||||
- `apps/node/internal/node/node_test.go`
|
||||
- 필요 시 `apps/node/internal/transport/integration_test.go`
|
||||
- 테스트 결정: 추가한다. event payload의 `run_id`, `type`, `error`, `session_id/background` 중 계약상 필요한 값을 assertion한다.
|
||||
- 중간 검증:
|
||||
- `go test -race -count=1 ./apps/node/internal/node/...`
|
||||
- integration test를 transport package에 둘 경우 `go test -race -count=1 ./apps/node/internal/transport/...`
|
||||
|
||||
### [REVIEW_REVIEW_REFACTOR-3] Direct Dev Diagnostic Evidence
|
||||
|
||||
- 문제: `./scripts/e2e-smoke.sh`는 보조 smoke이고 스크립트 출력도 직접 `scripts/dev/edge.sh` + `scripts/dev/node.sh` user-flow verification이 여전히 필요하다고 명시한다.
|
||||
- 해결 방법: 임시 config/포트를 사용해 `scripts/dev/edge.sh`와 `scripts/dev/node.sh`를 별도 프로세스로 직접 실행하고, edge console에 `/nodes`, 메시지 2회, `/capabilities`, `/transport`, `/sessions`, 필요한 경우 `/terminate-session` 입력을 보내 결과를 기록한다. `exec_command`의 PTY 또는 FIFO를 사용할 수 있다. 정말 불가능하면 시도한 정확한 명령, stdout/stderr, 불가능한 이유, 남은 위험을 기록한다.
|
||||
- 테스트 결정: 검증 증거를 수집한다. `./scripts/e2e-smoke.sh`는 보조 증거로만 별도 기록한다.
|
||||
- 최종 검증:
|
||||
- `go test -race -count=1 ./apps/node/internal/node/...`
|
||||
- `go test -count=1 ./apps/node/...`
|
||||
- 직접 dev diagnostic: `scripts/dev/edge.sh` + `scripts/dev/node.sh`
|
||||
- 보조 smoke: `./scripts/e2e-smoke.sh`
|
||||
|
||||
## 리뷰어를 위한 체크포인트
|
||||
|
||||
- global runtime concurrency는 adapter별이 아니라 Node 전체 상한이다.
|
||||
- adapter `MaxConcurrency`는 global limit과 별개의 adapter key별 상한으로 적용된다.
|
||||
- reject event 테스트가 zero-value session을 우회하지 않고 실제 Edge 관측 경로를 탄다.
|
||||
- 보조 smoke와 직접 dev diagnostic이 서로 대체되지 않고 분리 기록되어 있다.
|
||||
|
|
@ -1,90 +0,0 @@
|
|||
<!-- task=m-node-multi-target-serving-foundation/07+02_runtime_concurrency plan=0 tag=REFACTOR -->
|
||||
|
||||
# Code Review Reference - REFACTOR
|
||||
|
||||
> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.**
|
||||
> Fill implementation-owned sections and actual verification output, then stop with active files in place.
|
||||
> Do not ask the user directly; record user-only blockers in `사용자 리뷰 요청`.
|
||||
|
||||
## 개요
|
||||
|
||||
date=2026-06-10
|
||||
task=m-node-multi-target-serving-foundation/07+02_runtime_concurrency, plan=0, tag=REFACTOR
|
||||
|
||||
## Roadmap Targets
|
||||
|
||||
- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/node-multi-target-serving-foundation.md`
|
||||
- Task ids:
|
||||
- `runtime-concurrency`: `runtime.concurrency`와 adapter capability가 Node 멀티 호출에서 실제 제한 또는 스케줄링 정책으로 적용된다.
|
||||
- Completion mode: check-on-pass
|
||||
|
||||
## 이 파일을 읽는 리뷰 에이전트에게
|
||||
|
||||
> **[REVIEW AGENT ONLY]** 종결 절차는 코드리뷰 에이전트 전용이다.
|
||||
|
||||
concurrency policy가 deterministic하고 foreground/background 양쪽에서 같은지 검토한다.
|
||||
|
||||
## 구현 항목별 완료 여부
|
||||
|
||||
| 항목 | 완료 여부 |
|
||||
|------|---------|
|
||||
| [REFACTOR-1] Runtime Concurrency Gate | [ ] |
|
||||
|
||||
## 구현 체크리스트
|
||||
|
||||
- [ ] Node runtime global concurrency와 adapter/profile concurrency 중 적용 우선순위를 정의한다.
|
||||
- [ ] Foreground/background run 모두 동일한 scheduling/limit 정책을 통과한다.
|
||||
- [ ] 제한 초과 요청의 queue/reject/timeout 중 하나를 선택하고 error/event/store 상태를 테스트로 고정한다.
|
||||
- [ ] race/order test를 추가하고 최종 검증을 실행한다.
|
||||
- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
|
||||
|
||||
## 코드리뷰 전용 체크리스트
|
||||
|
||||
- [ ] `코드리뷰 결과`에 판정을 append한다.
|
||||
- [ ] active `CODE_REVIEW-cloud-G08.md`를 `code_review_cloud_G08_N.log`로 아카이브한다.
|
||||
- [ ] active `PLAN-cloud-G07.md`를 `plan_cloud_G07_M.log`로 아카이브한다.
|
||||
- [ ] PASS이면 `complete.log` 작성 후 archive 이동한다.
|
||||
- [ ] WARN/FAIL이면 후속 상태를 작성한다.
|
||||
|
||||
## 계획 대비 변경 사항
|
||||
|
||||
_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._
|
||||
|
||||
## 주요 설계 결정
|
||||
|
||||
_구현 에이전트가 주요 설계 결정 사항을 기록한다._
|
||||
|
||||
## 사용자 리뷰 요청
|
||||
|
||||
_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 구현 에이전트는 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._
|
||||
|
||||
- 상태: 없음
|
||||
- 사유 유형: 없음
|
||||
- 결정 필요: 없음
|
||||
- 차단 근거: 없음
|
||||
- 실행한 검증/명령: 없음
|
||||
- 자동 후속 불가 이유: 없음
|
||||
- 재개 조건: 없음
|
||||
|
||||
## 리뷰어를 위한 체크포인트
|
||||
|
||||
- permit acquire/release가 adapter error, timeout, cancellation, background run에서 누수되지 않는다.
|
||||
- 선택한 제한 초과 정책이 테스트와 문서에 일치한다.
|
||||
- proto-socket transport 변경으로 문제를 우회하지 않았다.
|
||||
|
||||
## 검증 결과
|
||||
|
||||
### REFACTOR-1 중간 검증
|
||||
|
||||
```bash
|
||||
$ go test -race -count=1 ./apps/node/internal/node/...
|
||||
(output)
|
||||
```
|
||||
|
||||
### 최종 검증
|
||||
|
||||
```bash
|
||||
$ go test -race -count=1 ./apps/node/internal/node/...
|
||||
$ go test -count=1 ./apps/node/...
|
||||
(output)
|
||||
```
|
||||
|
|
@ -69,7 +69,8 @@ func Module(cfg *config.NodeConfig) fx.Option {
|
|||
}
|
||||
|
||||
rtr := router.New(reg, logger)
|
||||
n := node.New(result.NodeID, rtr, st, os.Stdout, logger)
|
||||
globalConcurrency := int(result.Config.GetRuntime().GetConcurrency())
|
||||
n := node.New(result.NodeID, rtr, st, globalConcurrency, os.Stdout, logger)
|
||||
result.Session.SetEventHandler(func(event *iop.EdgeNodeEvent) {
|
||||
printEdgeEvent(os.Stdout, event)
|
||||
})
|
||||
|
|
|
|||
|
|
@ -14,6 +14,8 @@ import (
|
|||
"strings"
|
||||
"time"
|
||||
|
||||
"sync"
|
||||
|
||||
"go.uber.org/zap"
|
||||
"google.golang.org/protobuf/proto"
|
||||
"google.golang.org/protobuf/types/known/structpb"
|
||||
|
|
@ -26,20 +28,27 @@ import (
|
|||
|
||||
// Node implements transport.Handler and coordinates the full execution pipeline.
|
||||
type Node struct {
|
||||
nodeID string
|
||||
router runtime.Router
|
||||
store *store.Store
|
||||
runs *runManager
|
||||
out io.Writer
|
||||
logger *zap.Logger
|
||||
nodeID string
|
||||
router runtime.Router
|
||||
store *store.Store
|
||||
runs *runManager
|
||||
globalPermit *permitManager // node-wide limit across all adapters
|
||||
adapterPermitsMu sync.Mutex
|
||||
adapterPermits map[string]*permitManager // per adapter-key limit
|
||||
out io.Writer
|
||||
logger *zap.Logger
|
||||
}
|
||||
|
||||
// New creates a Node. It satisfies transport.Handler.
|
||||
// globalConcurrency is the node-wide maximum for simultaneous executions across
|
||||
// all adapters (>= 1 enforces limit; <= 0 means unlimited at the node level).
|
||||
// Per-adapter MaxConcurrency from Capabilities is applied independently on top.
|
||||
// out receives console debug output; pass os.Stdout for production, io.Discard in tests.
|
||||
func New(
|
||||
nodeID string,
|
||||
router runtime.Router,
|
||||
st *store.Store,
|
||||
globalConcurrency int,
|
||||
out io.Writer,
|
||||
logger *zap.Logger,
|
||||
) *Node {
|
||||
|
|
@ -47,12 +56,14 @@ func New(
|
|||
out = os.Stdout
|
||||
}
|
||||
return &Node{
|
||||
nodeID: nodeID,
|
||||
router: router,
|
||||
store: st,
|
||||
runs: newRunManager(),
|
||||
out: out,
|
||||
logger: logger,
|
||||
nodeID: nodeID,
|
||||
router: router,
|
||||
store: st,
|
||||
runs: newRunManager(),
|
||||
globalPermit: newPermitManager(globalConcurrency),
|
||||
adapterPermits: make(map[string]*permitManager),
|
||||
out: out,
|
||||
logger: logger,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -84,6 +95,42 @@ func (n *Node) OnRunRequest(ctx context.Context, sess *transport.Session, req *i
|
|||
return fmt.Errorf("node: resolve: %w", err)
|
||||
}
|
||||
|
||||
// Concurrency gate (foreground and background use the same path).
|
||||
// Policy: reject-on-exceed (no queue, no wait).
|
||||
//
|
||||
// Two independent limits are enforced:
|
||||
// 1. global (node-wide): n.globalPermit — shared across all adapter keys.
|
||||
// 2. adapter-key: per-adapter Capabilities.MaxConcurrency permit.
|
||||
//
|
||||
// Both must be acquired. If global is acquired but adapter-key fails, global
|
||||
// is immediately released. The reject limit reported to the caller is the
|
||||
// stricter of the two.
|
||||
adapterCap := 0
|
||||
if caps, capsErr := adapter.Capabilities(ctx); capsErr == nil {
|
||||
adapterCap = caps.MaxConcurrency
|
||||
}
|
||||
adapterPM := n.adapterPermitFor(spec.Adapter, adapterCap)
|
||||
|
||||
if !n.globalPermit.tryAcquire() {
|
||||
limit := int(n.globalPermit.limit)
|
||||
n.logger.Warn("global concurrency limit exceeded",
|
||||
zap.String("run_id", spec.RunID),
|
||||
zap.Int("global_limit", limit),
|
||||
)
|
||||
n.rejectRun(ctx, sess, spec, limit)
|
||||
return fmt.Errorf("node: run %s: %w", spec.RunID, ErrConcurrencyLimitExceeded)
|
||||
}
|
||||
if !adapterPM.tryAcquire() {
|
||||
n.globalPermit.release()
|
||||
n.logger.Warn("adapter concurrency limit exceeded",
|
||||
zap.String("run_id", spec.RunID),
|
||||
zap.String("adapter", spec.Adapter),
|
||||
zap.Int("adapter_limit", adapterCap),
|
||||
)
|
||||
n.rejectRun(ctx, sess, spec, adapterCap)
|
||||
return fmt.Errorf("node: run %s: %w", spec.RunID, ErrConcurrencyLimitExceeded)
|
||||
}
|
||||
|
||||
if err := n.store.InsertRun(ctx, store.RunRecord{
|
||||
RunID: spec.RunID,
|
||||
Adapter: spec.Adapter,
|
||||
|
|
@ -120,6 +167,8 @@ func (n *Node) OnRunRequest(ctx context.Context, sess *transport.Session, req *i
|
|||
}
|
||||
|
||||
run := func() error {
|
||||
defer n.globalPermit.release()
|
||||
defer adapterPM.release()
|
||||
defer cancel()
|
||||
defer n.runs.deregister(spec.RunID)
|
||||
defer close(h.done)
|
||||
|
|
@ -136,6 +185,55 @@ func (n *Node) OnRunRequest(ctx context.Context, sess *transport.Session, req *i
|
|||
return run()
|
||||
}
|
||||
|
||||
// adapterPermitFor returns the shared permitManager for the given adapter
|
||||
// instance key, creating one with the given limit if not yet seen.
|
||||
// limit <= 0 means unlimited.
|
||||
func (n *Node) adapterPermitFor(adapterKey string, limit int) *permitManager {
|
||||
n.adapterPermitsMu.Lock()
|
||||
defer n.adapterPermitsMu.Unlock()
|
||||
pm, ok := n.adapterPermits[adapterKey]
|
||||
if !ok {
|
||||
pm = newPermitManager(limit)
|
||||
n.adapterPermits[adapterKey] = pm
|
||||
}
|
||||
return pm
|
||||
}
|
||||
|
||||
// rejectRun records a rejected run in the store and sends an error RunEvent to
|
||||
// the session so Edge can observe the rejection through the normal event stream.
|
||||
func (n *Node) rejectRun(ctx context.Context, sess *transport.Session, spec runtime.ExecutionSpec, limit int) {
|
||||
errMsg := fmt.Sprintf("concurrency limit exceeded (limit=%d)", limit)
|
||||
|
||||
if err := n.store.InsertRun(ctx, store.RunRecord{
|
||||
RunID: spec.RunID,
|
||||
Adapter: spec.Adapter,
|
||||
Target: spec.Target,
|
||||
SessionID: normalizeSessionID(spec.SessionID),
|
||||
Background: spec.Background,
|
||||
Status: "rejected",
|
||||
CreatedAt: time.Now(),
|
||||
}); err != nil {
|
||||
n.logger.Warn("store: insert rejected run", zap.String("run_id", spec.RunID), zap.Error(err))
|
||||
} else if err := n.store.CompleteRun(ctx, spec.RunID, "rejected", errMsg); err != nil {
|
||||
n.logger.Warn("store: complete rejected run", zap.String("run_id", spec.RunID), zap.Error(err))
|
||||
}
|
||||
|
||||
if sess != nil && sess.IsAlive() {
|
||||
re := &iop.RunEvent{
|
||||
RunId: spec.RunID,
|
||||
Type: string(runtime.EventTypeError),
|
||||
Error: errMsg,
|
||||
Timestamp: time.Now().UnixNano(),
|
||||
SessionId: normalizeSessionID(spec.SessionID),
|
||||
Background: spec.Background,
|
||||
NodeId: n.nodeID,
|
||||
}
|
||||
if err := sess.Send(re); err != nil {
|
||||
n.logger.Warn("session: send reject event", zap.String("run_id", spec.RunID), zap.Error(err))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (n *Node) completeRun(spec runtime.ExecutionSpec, execErr error) {
|
||||
status := "completed"
|
||||
errMsg := ""
|
||||
|
|
|
|||
211
apps/node/internal/node/node_concurrency_integration_test.go
Normal file
211
apps/node/internal/node/node_concurrency_integration_test.go
Normal file
|
|
@ -0,0 +1,211 @@
|
|||
package node_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
toki "git.toki-labs.com/toki/proto-socket/go"
|
||||
"go.uber.org/zap"
|
||||
"google.golang.org/protobuf/proto"
|
||||
|
||||
"iop/apps/node/internal/node"
|
||||
"iop/apps/node/internal/runtime"
|
||||
"iop/apps/node/internal/store"
|
||||
"iop/apps/node/internal/transport"
|
||||
iop "iop/proto/gen/iop"
|
||||
)
|
||||
|
||||
// edgeServerParserMap returns the proto parser map for a mock Edge server
|
||||
// that expects to receive RunEvent messages sent by Node.
|
||||
func edgeServerParserMap() toki.ParserMap {
|
||||
return toki.ParserMap{
|
||||
toki.TypeNameOf(&iop.RunEvent{}): func(b []byte) (proto.Message, error) {
|
||||
m := &iop.RunEvent{}
|
||||
return m, proto.Unmarshal(b, m)
|
||||
},
|
||||
toki.TypeNameOf(&iop.RegisterRequest{}): func(b []byte) (proto.Message, error) {
|
||||
m := &iop.RegisterRequest{}
|
||||
return m, proto.Unmarshal(b, m)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func getFreeListenAddr(t *testing.T) string {
|
||||
t.Helper()
|
||||
l, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatalf("listen: %v", err)
|
||||
}
|
||||
addr := l.Addr().String()
|
||||
l.Close()
|
||||
return addr
|
||||
}
|
||||
|
||||
// startMockEdge starts a mock Edge TCP server that accepts exactly one Node
|
||||
// connection, performs the registration handshake, and returns the accepted
|
||||
// toki.TcpClient so tests can attach listeners and send RunRequests.
|
||||
func startMockEdge(t *testing.T, ctx context.Context, listenAddr string) *toki.TcpClient {
|
||||
t.Helper()
|
||||
|
||||
host, portStr, _ := net.SplitHostPort(listenAddr)
|
||||
port := 0
|
||||
fmt.Sscanf(portStr, "%d", &port)
|
||||
|
||||
acceptedCh := make(chan *toki.TcpClient, 1)
|
||||
server := toki.NewTcpServer(host, port, func(conn net.Conn) *toki.TcpClient {
|
||||
client := toki.NewTcpClient(conn, 30, 10, edgeServerParserMap())
|
||||
toki.AddRequestListenerTyped[*iop.RegisterRequest, *iop.RegisterResponse](
|
||||
&client.Communicator,
|
||||
func(req *iop.RegisterRequest) (*iop.RegisterResponse, error) {
|
||||
return &iop.RegisterResponse{
|
||||
Accepted: true,
|
||||
NodeId: "test-node",
|
||||
Alias: "test-alias",
|
||||
Config: &iop.NodeConfigPayload{
|
||||
Runtime: &iop.NodeRuntimeConfig{Concurrency: 1},
|
||||
},
|
||||
}, nil
|
||||
},
|
||||
)
|
||||
acceptedCh <- client
|
||||
return client
|
||||
})
|
||||
if err := server.Start(ctx); err != nil {
|
||||
t.Fatalf("mock edge server start: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { server.Stop() })
|
||||
|
||||
select {
|
||||
case client := <-acceptedCh:
|
||||
return client
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("mock edge server did not accept connection within 3s")
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// TestConcurrencyLimit_RejectEventObservedByEdge is an integration test that
|
||||
// verifies the RunEvent{type:"error"} sent by rejectRun() actually arrives at
|
||||
// the mock Edge server when the Node's concurrency limit is exceeded.
|
||||
//
|
||||
// Setup:
|
||||
// 1. Start mock Edge TCP server.
|
||||
// 2. Dial Node-side session via transport.DialEdge.
|
||||
// 3. Create a real node.Node with globalConcurrency=1.
|
||||
// 4. Attach the session to the node via SetHandler.
|
||||
// 5. From the Edge side, send a first RunRequest whose adapter blocks.
|
||||
// 6. After the first run starts, send a second RunRequest from the Edge side.
|
||||
// 7. Assert that a RunEvent{type:"error", run_id:<second>} arrives at the Edge.
|
||||
func TestConcurrencyLimit_RejectEventObservedByEdge(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
logger := zap.NewNop()
|
||||
listenAddr := getFreeListenAddr(t)
|
||||
|
||||
// Channel that the mock edge client will be delivered into once Node connects.
|
||||
edgeClientCh := make(chan *toki.TcpClient, 1)
|
||||
|
||||
// Start mock Edge server (async — we need to dial first to trigger accept).
|
||||
host, portStr, _ := net.SplitHostPort(listenAddr)
|
||||
port := 0
|
||||
fmt.Sscanf(portStr, "%d", &port)
|
||||
server := toki.NewTcpServer(host, port, func(conn net.Conn) *toki.TcpClient {
|
||||
client := toki.NewTcpClient(conn, 30, 10, edgeServerParserMap())
|
||||
toki.AddRequestListenerTyped[*iop.RegisterRequest, *iop.RegisterResponse](
|
||||
&client.Communicator,
|
||||
func(req *iop.RegisterRequest) (*iop.RegisterResponse, error) {
|
||||
return &iop.RegisterResponse{
|
||||
Accepted: true,
|
||||
NodeId: "test-node",
|
||||
Alias: "test-alias",
|
||||
Config: &iop.NodeConfigPayload{
|
||||
Runtime: &iop.NodeRuntimeConfig{Concurrency: 1},
|
||||
},
|
||||
}, nil
|
||||
},
|
||||
)
|
||||
edgeClientCh <- client
|
||||
return client
|
||||
})
|
||||
if err := server.Start(ctx); err != nil {
|
||||
t.Fatalf("mock edge server start: %v", err)
|
||||
}
|
||||
defer server.Stop()
|
||||
|
||||
// Dial Node side to Edge.
|
||||
dialResult, err := transport.DialEdge(ctx, listenAddr, "test-token", logger)
|
||||
if err != nil {
|
||||
t.Fatalf("DialEdge: %v", err)
|
||||
}
|
||||
defer dialResult.Session.Close()
|
||||
|
||||
// Wait for Edge side to accept the connection.
|
||||
var edgeClient *toki.TcpClient
|
||||
select {
|
||||
case edgeClient = <-edgeClientCh:
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("edge server did not accept connection")
|
||||
}
|
||||
|
||||
// Attach RunEvent listener on the Edge side before sending any requests.
|
||||
runEventCh := make(chan *iop.RunEvent, 4)
|
||||
toki.AddListenerTyped[*iop.RunEvent](&edgeClient.Communicator, func(event *iop.RunEvent) {
|
||||
runEventCh <- event
|
||||
})
|
||||
|
||||
// Build real node.Node with global concurrency=1.
|
||||
sa := newSlowAdapter() // MaxConcurrency=1; blocks until sa.release is closed
|
||||
rtr := &fixedRouter{
|
||||
adapterName: "slow",
|
||||
adapters: map[string]runtime.Adapter{"slow": sa},
|
||||
}
|
||||
st, err := store.New(":memory:", logger)
|
||||
if err != nil {
|
||||
t.Fatalf("store: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = st.Close() })
|
||||
|
||||
n := node.New("test-node", rtr, st, 1, nil, logger)
|
||||
dialResult.Session.SetHandler(n)
|
||||
|
||||
// Send first RunRequest from Edge → Node (adapter will block).
|
||||
firstReq := &iop.RunRequest{RunId: "run-hold-1", Adapter: "slow", Target: "v1"}
|
||||
if err := edgeClient.Send(firstReq); err != nil {
|
||||
t.Fatalf("send first run request: %v", err)
|
||||
}
|
||||
|
||||
// Wait until the first run's adapter has started executing so the permit is held.
|
||||
select {
|
||||
case <-sa.started:
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("first run never started")
|
||||
}
|
||||
|
||||
// Send second RunRequest; should be rejected because global=1 is taken.
|
||||
secondReq := &iop.RunRequest{RunId: "run-reject-2", Adapter: "slow", Target: "v1"}
|
||||
if err := edgeClient.Send(secondReq); err != nil {
|
||||
t.Fatalf("send second run request: %v", err)
|
||||
}
|
||||
|
||||
// Collect RunEvents until we see the error event for the second run.
|
||||
deadline := time.After(5 * time.Second)
|
||||
for {
|
||||
select {
|
||||
case ev := <-runEventCh:
|
||||
if ev.GetRunId() == "run-reject-2" && ev.GetType() == string(runtime.EventTypeError) {
|
||||
if ev.GetError() == "" {
|
||||
t.Fatal("reject RunEvent has empty error field")
|
||||
}
|
||||
// Success: RunEvent{type:"error", run_id:"run-reject-2"} observed.
|
||||
close(sa.release)
|
||||
return
|
||||
}
|
||||
case <-deadline:
|
||||
t.Fatal("timeout: did not receive RunEvent{type:error} for run-reject-2")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -206,13 +206,18 @@ func (r *errorRouter) GetAdapter(_ string) (runtime.Adapter, bool) {
|
|||
}
|
||||
|
||||
func makeNode(t *testing.T, rtr runtime.Router) (*node.Node, *store.Store) {
|
||||
t.Helper()
|
||||
return makeNodeWithConcurrency(t, rtr, 0)
|
||||
}
|
||||
|
||||
func makeNodeWithConcurrency(t *testing.T, rtr runtime.Router, concurrency int) (*node.Node, *store.Store) {
|
||||
t.Helper()
|
||||
st, err := store.New(":memory:", zap.NewNop())
|
||||
if err != nil {
|
||||
t.Fatalf("store: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = st.Close() })
|
||||
return node.New("test-node", rtr, st, io.Discard, zap.NewNop()), st
|
||||
return node.New("test-node", rtr, st, concurrency, io.Discard, zap.NewNop()), st
|
||||
}
|
||||
|
||||
// --- existing tests (updated) ---
|
||||
|
|
@ -917,3 +922,383 @@ func TestOnCommandRequest_Capabilities_ExactInstanceKey(t *testing.T) {
|
|||
t.Errorf("result[instance_key]: got %q want ollama@local", got)
|
||||
}
|
||||
}
|
||||
|
||||
// --- concurrency gate tests ---
|
||||
|
||||
// slowAdapter blocks until released, exposing a start channel so tests can
|
||||
// synchronize. Execute returns nil on normal completion.
|
||||
type slowAdapter struct {
|
||||
started chan struct{}
|
||||
release chan struct{}
|
||||
}
|
||||
|
||||
func newSlowAdapter() *slowAdapter {
|
||||
return &slowAdapter{started: make(chan struct{}, 1), release: make(chan struct{})}
|
||||
}
|
||||
|
||||
func (a *slowAdapter) Name() string { return "slow" }
|
||||
func (a *slowAdapter) Capabilities(_ context.Context) (runtime.Capabilities, error) {
|
||||
return runtime.Capabilities{AdapterName: "slow", MaxConcurrency: 1}, nil
|
||||
}
|
||||
|
||||
// slowAdapterUnlimited is like slowAdapter but reports MaxConcurrency=0 (unlimited).
|
||||
type slowAdapterUnlimited struct {
|
||||
started chan struct{}
|
||||
release chan struct{}
|
||||
}
|
||||
|
||||
func newSlowAdapterUnlimited() *slowAdapterUnlimited {
|
||||
return &slowAdapterUnlimited{started: make(chan struct{}, 1), release: make(chan struct{})}
|
||||
}
|
||||
|
||||
func (a *slowAdapterUnlimited) Name() string { return "slow" }
|
||||
func (a *slowAdapterUnlimited) Capabilities(_ context.Context) (runtime.Capabilities, error) {
|
||||
return runtime.Capabilities{AdapterName: "slow", MaxConcurrency: 0}, nil
|
||||
}
|
||||
func (a *slowAdapterUnlimited) Execute(ctx context.Context, _ runtime.ExecutionSpec, _ runtime.EventSink) error {
|
||||
select {
|
||||
case a.started <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
select {
|
||||
case <-a.release:
|
||||
case <-ctx.Done():
|
||||
return runtime.ErrRunCancelled
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// slowHighCapAdapter is like slowAdapter but reports MaxConcurrency=4,
|
||||
// used to verify that the global limit overrides a looser adapter cap.
|
||||
type slowHighCapAdapter struct {
|
||||
started chan struct{}
|
||||
release chan struct{}
|
||||
}
|
||||
|
||||
func newSlowHighCapAdapter() *slowHighCapAdapter {
|
||||
return &slowHighCapAdapter{started: make(chan struct{}, 1), release: make(chan struct{})}
|
||||
}
|
||||
|
||||
func (a *slowHighCapAdapter) Name() string { return "slow-hicap" }
|
||||
func (a *slowHighCapAdapter) Capabilities(_ context.Context) (runtime.Capabilities, error) {
|
||||
return runtime.Capabilities{AdapterName: "slow-hicap", MaxConcurrency: 4}, nil
|
||||
}
|
||||
func (a *slowHighCapAdapter) Execute(ctx context.Context, _ runtime.ExecutionSpec, _ runtime.EventSink) error {
|
||||
select {
|
||||
case a.started <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
select {
|
||||
case <-a.release:
|
||||
case <-ctx.Done():
|
||||
return runtime.ErrRunCancelled
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *slowAdapter) Execute(ctx context.Context, _ runtime.ExecutionSpec, _ runtime.EventSink) error {
|
||||
select {
|
||||
case a.started <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
select {
|
||||
case <-a.release:
|
||||
case <-ctx.Done():
|
||||
return runtime.ErrRunCancelled
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// TestConcurrencyLimit_RejectSecondForeground verifies that a second foreground
|
||||
// run is rejected immediately when the global limit is 1 and one run is active.
|
||||
func TestConcurrencyLimit_RejectSecondForeground(t *testing.T) {
|
||||
sa := newSlowAdapter()
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
n, _ := makeNodeWithConcurrency(t, router, 1)
|
||||
|
||||
errc := make(chan error, 1)
|
||||
go func() {
|
||||
errc <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-fg-1", Adapter: "slow", Target: "v1",
|
||||
})
|
||||
}()
|
||||
|
||||
// Wait until the first run has started executing.
|
||||
select {
|
||||
case <-sa.started:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("first run never started")
|
||||
}
|
||||
|
||||
// Second run must be rejected.
|
||||
err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-fg-2", Adapter: "slow", Target: "v1",
|
||||
})
|
||||
if !errors.Is(err, node.ErrConcurrencyLimitExceeded) {
|
||||
t.Fatalf("expected ErrConcurrencyLimitExceeded, got %v", err)
|
||||
}
|
||||
|
||||
// Release the first run and drain it.
|
||||
close(sa.release)
|
||||
if err := <-errc; err != nil {
|
||||
t.Fatalf("first run: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestConcurrencyLimit_RejectBackgroundRun verifies that a background run is
|
||||
// also subject to the global concurrency gate.
|
||||
func TestConcurrencyLimit_RejectBackgroundRun(t *testing.T) {
|
||||
sa := newSlowAdapter()
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
n, _ := makeNodeWithConcurrency(t, router, 1)
|
||||
|
||||
errc := make(chan error, 1)
|
||||
go func() {
|
||||
errc <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-bg-hold", Adapter: "slow", Target: "v1",
|
||||
})
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-sa.started:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("hold run never started")
|
||||
}
|
||||
|
||||
// Background run must also be rejected.
|
||||
err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-bg-2", Adapter: "slow", Target: "v1", Background: true,
|
||||
})
|
||||
if !errors.Is(err, node.ErrConcurrencyLimitExceeded) {
|
||||
t.Fatalf("expected ErrConcurrencyLimitExceeded for background run, got %v", err)
|
||||
}
|
||||
|
||||
close(sa.release)
|
||||
if err := <-errc; err != nil {
|
||||
t.Fatalf("hold run: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestConcurrencyLimit_PermitReleasedAfterCompletion verifies that after the
|
||||
// first run completes, a subsequent run can acquire the permit.
|
||||
func TestConcurrencyLimit_PermitReleasedAfterCompletion(t *testing.T) {
|
||||
sa := newSlowAdapter()
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
n, _ := makeNodeWithConcurrency(t, router, 1)
|
||||
|
||||
// First run: start, wait, release.
|
||||
errc := make(chan error, 1)
|
||||
go func() {
|
||||
errc <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-seq-1", Adapter: "slow", Target: "v1",
|
||||
})
|
||||
}()
|
||||
select {
|
||||
case <-sa.started:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("seq-1 never started")
|
||||
}
|
||||
close(sa.release)
|
||||
if err := <-errc; err != nil {
|
||||
t.Fatalf("seq-1: %v", err)
|
||||
}
|
||||
|
||||
// Second run must succeed now that the permit is free.
|
||||
sa2 := newSlowAdapter()
|
||||
router.adapters["slow"] = sa2
|
||||
close(sa2.release) // let it complete immediately
|
||||
if err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-seq-2", Adapter: "slow", Target: "v1",
|
||||
}); err != nil {
|
||||
t.Fatalf("seq-2 should succeed after permit released, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestConcurrencyLimit_Unlimited verifies that global=0 + adapter cap=0 imposes no cap.
|
||||
func TestConcurrencyLimit_Unlimited(t *testing.T) {
|
||||
const count = 5
|
||||
shared := newSlowAdapterUnlimited() // MaxConcurrency=0 (unlimited)
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": shared}}
|
||||
nd, _ := makeNodeWithConcurrency(t, router, 0) // global unlimited
|
||||
|
||||
errs := make(chan error, count)
|
||||
for i := 0; i < count; i++ {
|
||||
runID := fmt.Sprintf("run-unlimited-%d", i)
|
||||
go func(id string) {
|
||||
errs <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: id, Adapter: "slow", Target: "v1",
|
||||
})
|
||||
}(runID)
|
||||
}
|
||||
|
||||
// Release all.
|
||||
close(shared.release)
|
||||
|
||||
// All must complete without ErrConcurrencyLimitExceeded.
|
||||
for i := 0; i < count; i++ {
|
||||
if err := <-errs; errors.Is(err, node.ErrConcurrencyLimitExceeded) {
|
||||
t.Fatalf("unexpected concurrency rejection with limit=0: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestConcurrencyLimit_AdapterCapFallback verifies that when global=0 and adapter
|
||||
// cap=1, the adapter cap is used as the effective limit.
|
||||
func TestConcurrencyLimit_AdapterCapFallback(t *testing.T) {
|
||||
sa := newSlowAdapter() // MaxConcurrency=1
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
nd, _ := makeNodeWithConcurrency(t, router, 0) // global=0 → use adapter cap
|
||||
|
||||
errc := make(chan error, 1)
|
||||
go func() {
|
||||
errc <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-cap-1", Adapter: "slow", Target: "v1",
|
||||
})
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-sa.started:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("run-cap-1 never started")
|
||||
}
|
||||
|
||||
// Second run must be rejected by adapter cap.
|
||||
err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-cap-2", Adapter: "slow", Target: "v1",
|
||||
})
|
||||
if !errors.Is(err, node.ErrConcurrencyLimitExceeded) {
|
||||
t.Fatalf("expected ErrConcurrencyLimitExceeded from adapter cap, got %v", err)
|
||||
}
|
||||
|
||||
close(sa.release)
|
||||
if err := <-errc; err != nil {
|
||||
t.Fatalf("run-cap-1: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestConcurrencyLimit_GlobalBeatsAdapterCap verifies that when both global and
|
||||
// adapter cap are > 0 and global < adapter cap, the global (stricter) limit applies.
|
||||
func TestConcurrencyLimit_GlobalBeatsAdapterCap(t *testing.T) {
|
||||
// slowHighCapAdapter reports MaxConcurrency=4; global=1 → effective=1.
|
||||
sa := newSlowHighCapAdapter()
|
||||
router := &fixedRouter{adapterName: "slow-hicap", adapters: map[string]runtime.Adapter{"slow-hicap": sa}}
|
||||
nd, _ := makeNodeWithConcurrency(t, router, 1) // global=1, adapter cap=4 → effective=1
|
||||
|
||||
errc := make(chan error, 1)
|
||||
go func() {
|
||||
errc <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-g1-1", Adapter: "slow-hicap", Target: "v1",
|
||||
})
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-sa.started:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("run-g1-1 never started")
|
||||
}
|
||||
|
||||
// Second run must be rejected by global limit=1 even though adapter cap=4.
|
||||
err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-g1-2", Adapter: "slow-hicap", Target: "v1",
|
||||
})
|
||||
if !errors.Is(err, node.ErrConcurrencyLimitExceeded) {
|
||||
t.Fatalf("expected ErrConcurrencyLimitExceeded (global=1 beats adapter cap=4), got %v", err)
|
||||
}
|
||||
|
||||
close(sa.release)
|
||||
if err := <-errc; err != nil {
|
||||
t.Fatalf("run-g1-1: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestConcurrencyLimit_GlobalBeatsAdapterCap_DifferentAdapters verifies that
|
||||
// global=1 rejects a run on adapter B when adapter A already holds the global
|
||||
// permit, even though both adapter caps are > 1.
|
||||
func TestConcurrencyLimit_GlobalBeatsAdapterCap_DifferentAdapters(t *testing.T) {
|
||||
saA := newSlowHighCapAdapter() // adapter "slow-hicap-a", MaxConcurrency=4
|
||||
saB := newSlowHighCapAdapter() // adapter "slow-hicap-b", MaxConcurrency=4
|
||||
|
||||
router := &fixedRouter{
|
||||
adapterName: "slow-hicap-a",
|
||||
adapters: map[string]runtime.Adapter{
|
||||
"slow-hicap-a": saA,
|
||||
"slow-hicap-b": saB,
|
||||
},
|
||||
}
|
||||
nd, _ := makeNodeWithConcurrency(t, router, 1) // global=1
|
||||
|
||||
// First run on adapter A: hold the global permit.
|
||||
errc := make(chan error, 1)
|
||||
go func() {
|
||||
errc <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-xa-1", Adapter: "slow-hicap-a", Target: "v1",
|
||||
})
|
||||
}()
|
||||
select {
|
||||
case <-saA.started:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("run-xa-1 never started")
|
||||
}
|
||||
|
||||
// Second run on adapter B must be rejected because global permit is taken.
|
||||
router.adapterName = "slow-hicap-b"
|
||||
err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-xb-1", Adapter: "slow-hicap-b", Target: "v1",
|
||||
})
|
||||
if !errors.Is(err, node.ErrConcurrencyLimitExceeded) {
|
||||
t.Fatalf("expected ErrConcurrencyLimitExceeded for cross-adapter second run, got %v", err)
|
||||
}
|
||||
|
||||
close(saA.release)
|
||||
if err := <-errc; err != nil {
|
||||
t.Fatalf("run-xa-1: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestConcurrencyLimit_RejectStoreAndEvent verifies that a rejected run is stored
|
||||
// with terminal status "rejected" and an error message is recorded.
|
||||
func TestConcurrencyLimit_RejectStoreAndEvent(t *testing.T) {
|
||||
sa := newSlowAdapter()
|
||||
router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
|
||||
nd, st := makeNodeWithConcurrency(t, router, 1)
|
||||
|
||||
errc := make(chan error, 1)
|
||||
go func() {
|
||||
errc <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-hold", Adapter: "slow", Target: "v1",
|
||||
})
|
||||
}()
|
||||
select {
|
||||
case <-sa.started:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("run-hold never started")
|
||||
}
|
||||
|
||||
// Second run must be rejected.
|
||||
err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
||||
RunId: "run-rejected", Adapter: "slow", Target: "v1",
|
||||
})
|
||||
if !errors.Is(err, node.ErrConcurrencyLimitExceeded) {
|
||||
t.Fatalf("expected ErrConcurrencyLimitExceeded, got %v", err)
|
||||
}
|
||||
|
||||
// Rejected run must be stored with status "rejected".
|
||||
run, err := st.GetRun(context.Background(), "run-rejected")
|
||||
if err != nil {
|
||||
t.Fatalf("GetRun: %v", err)
|
||||
}
|
||||
if run == nil {
|
||||
t.Fatal("expected rejected run to be stored")
|
||||
}
|
||||
if run.Status != "rejected" {
|
||||
t.Fatalf("expected status %q, got %q", "rejected", run.Status)
|
||||
}
|
||||
if run.Error == "" {
|
||||
t.Fatal("expected non-empty error message for rejected run")
|
||||
}
|
||||
|
||||
close(sa.release)
|
||||
if err := <-errc; err != nil {
|
||||
t.Fatalf("run-hold: %v", err)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,9 +2,15 @@ package node
|
|||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
// ErrConcurrencyLimitExceeded is returned when a run is rejected because the
|
||||
// active run count has reached the effective concurrency limit.
|
||||
var ErrConcurrencyLimitExceeded = errors.New("concurrency limit exceeded")
|
||||
|
||||
type runHandle struct {
|
||||
runID string
|
||||
adapter string
|
||||
|
|
@ -45,3 +51,42 @@ func (m *runManager) cancelRun(runID string) bool {
|
|||
h.cancel()
|
||||
return true
|
||||
}
|
||||
|
||||
// permitManager enforces a concurrency limit using an atomic counter.
|
||||
// limit <= 0 means unlimited. Policy is reject-on-exceed (no queue, no wait).
|
||||
type permitManager struct {
|
||||
limit int32
|
||||
active atomic.Int32
|
||||
}
|
||||
|
||||
func newPermitManager(limit int) *permitManager {
|
||||
return &permitManager{limit: int32(limit)}
|
||||
}
|
||||
|
||||
// tryAcquire attempts to take a permit. Returns false immediately if the limit
|
||||
// is already reached. limit <= 0 always succeeds.
|
||||
func (p *permitManager) tryAcquire() bool {
|
||||
if p.limit <= 0 {
|
||||
p.active.Add(1)
|
||||
return true
|
||||
}
|
||||
for {
|
||||
cur := p.active.Load()
|
||||
if cur >= p.limit {
|
||||
return false
|
||||
}
|
||||
if p.active.CompareAndSwap(cur, cur+1) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// release returns a permit. Must be called exactly once per successful tryAcquire.
|
||||
func (p *permitManager) release() {
|
||||
p.active.Add(-1)
|
||||
}
|
||||
|
||||
// activeCount returns the number of currently held permits (for observability).
|
||||
func (p *permitManager) activeCount() int {
|
||||
return int(p.active.Load())
|
||||
}
|
||||
|
|
|
|||
BIN
edge
Executable file
BIN
edge
Executable file
Binary file not shown.
BIN
node
Executable file
BIN
node
Executable file
Binary file not shown.
Loading…
Reference in a new issue