From 34e87f70e9e85f20e1ab477c1ac76b8b9bc8be85 Mon Sep 17 00:00:00 2001 From: toki Date: Tue, 16 Jun 2026 11:19:56 +0900 Subject: [PATCH] feat(node): refactor node queue simplification and move completed subtask to archive - Refactor node queue management in node.go and run_manager.go - Update concurrency integration tests - Move 03+01_node_queue_simplify subtask to archive - Add complete.log for archived subtask --- .../code_review_cloud_G07_0.log} | 68 +- .../03+01_node_queue_simplify/complete.log | 45 ++ .../plan_cloud_G07_0.log} | 0 apps/node/internal/node/node.go | 132 ++-- .../node/node_concurrency_integration_test.go | 21 +- apps/node/internal/node/node_test.go | 617 +----------------- apps/node/internal/node/run_manager.go | 262 +------- 7 files changed, 195 insertions(+), 950 deletions(-) rename agent-task/{m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/CODE_REVIEW-cloud-G07.md => archive/2026/06/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/code_review_cloud_G07_0.log} (56%) create mode 100644 agent-task/archive/2026/06/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/complete.log rename agent-task/{m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/PLAN-cloud-G07.md => archive/2026/06/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/plan_cloud_G07_0.log} (100%) diff --git a/agent-task/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/CODE_REVIEW-cloud-G07.md b/agent-task/archive/2026/06/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/code_review_cloud_G07_0.log similarity index 56% rename from agent-task/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/CODE_REVIEW-cloud-G07.md rename to agent-task/archive/2026/06/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/code_review_cloud_G07_0.log index e7d6b44..c7a1608 100644 --- a/agent-task/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/CODE_REVIEW-cloud-G07.md +++ b/agent-task/archive/2026/06/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/code_review_cloud_G07_0.log @@ -33,44 +33,44 @@ task=m-edge-model-group-queue-scheduling/03+01_node_queue_simplify, plan=0, tag= | 항목 | 완료 여부 | |------|---------| -| [REFACTOR-1] OnRunRequest Queue Ownership 제거 | [ ] | -| [REFACTOR-2] Capability Snapshot 정리 | [ ] | -| [REFACTOR-3] Admission Queue Test 전환 | [ ] | +| [REFACTOR-1] OnRunRequest Queue Ownership 제거 | [x] | +| [REFACTOR-2] Capability Snapshot 정리 | [x] | +| [REFACTOR-3] Admission Queue Test 전환 | [x] | ## 구현 체크리스트 -- [ ] Node `OnRunRequest`에서 FIFO queue ownership을 제거하고 dispatch된 run을 즉시 실행 또는 safety reject로 처리한다. -- [ ] Node capability snapshot의 queued/in-flight 의미를 Edge-owned queue와 충돌하지 않게 정리한다. -- [ ] Node admission queue tests를 삭제/전환하고 over-dispatch safety, cancel, background, store lifecycle 테스트를 남긴다. -- [ ] `go test -count=1 ./apps/node/internal/node`가 통과한다. -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. +- [x] Node `OnRunRequest`에서 FIFO queue ownership을 제거하고 dispatch된 run을 즉시 실행 또는 safety reject로 처리한다. +- [x] Node capability snapshot의 queued/in-flight 의미를 Edge-owned queue와 충돌하지 않게 정리한다. +- [x] Node admission queue tests를 삭제/전환하고 over-dispatch safety, cancel, background, store lifecycle 테스트를 남긴다. +- [x] `go test -count=1 ./apps/node/internal/node`가 통과한다. +- [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을 확인한다. -- [ ] PASS이면 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. -- [ ] PASS이면 active task 디렉터리를 archive로 이동한다. -- [ ] PASS split 작업이면 parent directory 유지/정리를 확인한다. +- [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을 확인한다. +- [x] PASS이면 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. +- [x] PASS이면 active task 디렉터리를 archive로 이동한다. +- [x] PASS split 작업이면 parent directory 유지/정리를 확인한다. - [ ] WARN/FAIL이면 다음 active plan/review 또는 USER_REVIEW 경로를 처리한다. ## 계획 대비 변경 사항 -_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._ +- 없음 ## 주요 설계 결정 -_구현 에이전트가 주요 설계 결정 사항을 기록한다._ +- Node-local FIFO admission queue 및 대기(wait) 로직을 완전히 걷어내어 복잡한 goroutine/cancellation 복원 지점을 단순화했습니다. +- 이제 `OnRunRequest` 시점에 `admissionManager.acquire()`를 통해 즉시 slots을 획득하며, 획득 실패 시 곧바로 "concurrency unavailable" 에러를 반환하고 store에는 즉시 "rejected" 상태를 기록합니다. +- Node가 queue owner가 아니므로 capability snapshot 시 `queued` 카운트는 항상 `0`으로 보고하여 Edge-owned scheduler의 상태 스냅샷과 충돌하지 않도록 처리했습니다. ## 사용자 리뷰 요청 -_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 구현 에이전트는 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._ - - 상태: 없음 - 사유 유형: 없음 - 결정 필요: 없음 @@ -81,36 +81,50 @@ _기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외 ## 리뷰어를 위한 체크포인트 -- Node가 FIFO queue owner로 남지 않았는지 확인한다. -- Node queued snapshot이 Edge-owned queued 의미와 충돌하지 않는지 확인한다. -- cancel/background/store lifecycle 회귀가 남아 있는지 확인한다. +- Node의 OnRunRequest에서 큐 관련 대기 로직(`ticket.wait`)이 완전히 제거되었는지 확인합니다. +- `apps/node/internal/node/run_manager.go`와 `node.go`에서 `newFifoGate` 인자와 queuedCount의 반환 값(항상 0)을 확인합니다. +- `node_test.go`에서 queue 관련 불필요해진 테스트 케이스가 제거되었고, 즉시 리젝트 관련 테스트 케이스(`TestOnRunRequest_OverDispatchSafetyRejectsWithoutQueue`)가 올바르게 작동하는지 확인합니다. ## 검증 결과 ### REFACTOR-1 중간 검증 ```text $ go test -count=1 ./apps/node/internal/node -run 'TestOnRunRequest' -(output) +ok iop/apps/node/internal/node 0.028s ``` ### REFACTOR-2 중간 검증 ```text $ go test -count=1 ./apps/node/internal/node -run 'TestOnCommandRequest_Capabilities' -(output) +ok iop/apps/node/internal/node 0.011s ``` ### REFACTOR-3 중간 검증 ```text $ go test -count=1 ./apps/node/internal/node -run 'Test(AdmissionQueue|ConcurrencyLimit|QueueOverflow|OnRunRequest|OnCommandRequest_Capabilities)' -(output) +ok iop/apps/node/internal/node 0.017s ``` ### 최종 검증 ```text $ go test -count=1 ./apps/node/internal/node -(output) +ok iop/apps/node/internal/node 0.128s ``` --- > **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section?** + +## 코드리뷰 결과 + +- 종합 판정: PASS +- 차원별 평가: + - correctness: Pass + - completeness: Pass + - test coverage: Pass + - API contract: Pass + - code quality: Pass + - plan deviation: Pass + - verification trust: Pass +- 발견된 문제: 없음 +- 다음 단계: PASS이므로 `complete.log`를 작성하고 task directory를 archive로 이동한다. diff --git a/agent-task/archive/2026/06/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/complete.log b/agent-task/archive/2026/06/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/complete.log new file mode 100644 index 0000000..0b6c0a2 --- /dev/null +++ b/agent-task/archive/2026/06/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/complete.log @@ -0,0 +1,45 @@ +# Complete - m-edge-model-group-queue-scheduling/03+01_node_queue_simplify + +## 완료 일시 + +2026-06-16 + +## 요약 + +Node-local provider FIFO admission queue를 제거하고 immediate concurrency safety reject와 queued=0 capability snapshot으로 단순화한 작업을 1회 리뷰로 PASS 처리했다. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `plan_cloud_G07_0.log` | `code_review_cloud_G07_0.log` | PASS | Required/Suggested 이슈 없음; 리뷰 중 stale comment와 review artifact drift를 비동작성 정리 | + +## 구현/정리 내용 + +- `OnRunRequest`의 Node-local FIFO enqueue/wait/promote 경로를 제거하고, adapter/global concurrency safety guard를 즉시 획득하지 못하면 rejected store status와 error RunEvent를 남기도록 정리했다. +- capability response와 provider snapshot의 `queued` 값을 Node-local queue가 없음을 나타내는 `0`으로 고정하고, in-flight는 adapter safety guard 기준으로 유지했다. +- Node admission queue 전제 테스트를 over-dispatch safety rejection, permit release, rejected store/event, capability snapshot 기준으로 전환했다. +- 리뷰 중 stale FIFO comment와 미사용 queue helper test code를 비동작성으로 정리했다. + +## 최종 검증 + +- `go test -count=1 ./apps/node/internal/node -run 'TestOnRunRequest'` - PASS; `ok iop/apps/node/internal/node 0.022s` +- `go test -count=1 ./apps/node/internal/node -run 'TestOnCommandRequest_Capabilities'` - PASS; `ok iop/apps/node/internal/node 0.014s` +- `go test -count=1 ./apps/node/internal/node -run 'Test(AdmissionQueue|ConcurrencyLimit|QueueOverflow|OnRunRequest|OnCommandRequest_Capabilities)'` - PASS; `ok iop/apps/node/internal/node 0.022s` +- `go test -count=1 ./apps/node/internal/node` - PASS; `ok iop/apps/node/internal/node 0.138s` +- `go test -count=1 ./apps/node/...` - PASS; node package tree passed, including `apps/node/internal/transport` and adapter packages. + +## Roadmap Completion + +- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/edge-model-group-queue-scheduling.md` +- Completed task ids: + - `node-simplify`: PASS; evidence=`agent-task/archive/2026/06/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/plan_cloud_G07_0.log`, `agent-task/archive/2026/06/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/code_review_cloud_G07_0.log`; verification=`go test -count=1 ./apps/node/internal/node`, `go test -count=1 ./apps/node/...` +- Not completed task ids: 없음 + +## 잔여 Nit + +- 없음 + +## 후속 작업 + +- 없음 diff --git a/agent-task/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/PLAN-cloud-G07.md b/agent-task/archive/2026/06/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/plan_cloud_G07_0.log similarity index 100% rename from agent-task/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/PLAN-cloud-G07.md rename to agent-task/archive/2026/06/m-edge-model-group-queue-scheduling/03+01_node_queue_simplify/plan_cloud_G07_0.log diff --git a/apps/node/internal/node/node.go b/apps/node/internal/node/node.go index 239f095..056b41a 100644 --- a/apps/node/internal/node/node.go +++ b/apps/node/internal/node/node.go @@ -32,9 +32,9 @@ type Node struct { router runtime.Router store *store.Store runs *runManager - globalGate *fifoGate // node-wide FIFO admission across all adapters + globalGate *fifoGate // node-wide concurrency safety guard across all adapters adapterGatesMu sync.Mutex - adapterGates map[string]*fifoGate // per adapter-key FIFO admission + adapterGates map[string]*fifoGate // per adapter-key concurrency safety guard out io.Writer logger *zap.Logger } @@ -60,7 +60,7 @@ func New( router: router, store: st, runs: newRunManager(), - globalGate: newFifoGate(globalConcurrency, -1, 0), + globalGate: newFifoGate(globalConcurrency), adapterGates: make(map[string]*fifoGate), out: out, logger: logger, @@ -95,37 +95,15 @@ func (n *Node) OnRunRequest(ctx context.Context, sess *transport.Session, req *i return fmt.Errorf("node: resolve: %w", err) } - // Admission gate (foreground and background use the same path). - // Policy: provider-scoped FIFO queue. A request runs immediately when - // in_flight < capacity; otherwise it waits in the adapter's FIFO queue and - // is promoted to an execution slot as earlier runs release. Requests are - // rejected when the queue is full (max_queue) or the queue timeout elapses. - // - // Two gates compose admission: the per-adapter gate (capacity, max_queue, - // queue_timeout from Capabilities) and the node-wide global gate. var caps runtime.Capabilities if c, capsErr := adapter.Capabilities(ctx); capsErr == nil { caps = c } admission := n.admissionFor(spec.Adapter, caps) - // Record the request as queued before waiting so its lifecycle is - // observable even while it sits in the FIFO queue. - 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: "queued", - CreatedAt: time.Now(), - }); err != nil { - n.logger.Warn("store: insert run", zap.String("run_id", spec.RunID), zap.Error(err)) - } - - // Reserve a FIFO position without blocking. Only queue overflow fails here; - // waiting for promotion happens later in wait(). - ticket, err := admission.enqueue() + // Acquire safety capacity ticket. Since we no longer maintain a Node-local FIFO queue, + // if concurrency is full, we reject immediately. + ticket, err := admission.acquire() if err != nil { n.logger.Warn("run admission rejected", zap.String("run_id", spec.RunID), @@ -136,8 +114,19 @@ func (n *Node) OnRunRequest(ctx context.Context, sess *transport.Session, req *i return fmt.Errorf("node: run %s: %w", spec.RunID, err) } - // The run context governs both the queue wait and execution, so cancelling a - // queued run via OnCancel also removes it from the FIFO queue. + // Record the request as running since it is admitted immediately without queueing. + 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: "running", + CreatedAt: time.Now(), + }); err != nil { + n.logger.Warn("store: insert run", zap.String("run_id", spec.RunID), zap.Error(err)) + } + execCtx, cancel := context.WithCancel(ctx) if spec.TimeoutSec > 0 { execCtx, cancel = context.WithTimeout(ctx, time.Duration(spec.TimeoutSec)*time.Second) @@ -161,19 +150,6 @@ func (n *Node) OnRunRequest(ctx context.Context, sess *transport.Session, req *i background: spec.Background, } - // admit blocks until the run holds an execution slot, then promotes the - // stored status to running. It returns an admissionError on queue timeout or - // cancellation while queued. - admit := func() error { - if err := ticket.wait(execCtx); err != nil { - return err - } - if err := n.store.MarkRunning(context.Background(), spec.RunID); err != nil { - n.logger.Warn("store: mark running", zap.String("run_id", spec.RunID), zap.Error(err)) - } - return nil - } - run := func() error { defer ticket.release() defer cancel() @@ -185,38 +161,15 @@ func (n *Node) OnRunRequest(ctx context.Context, sess *transport.Session, req *i return execErr } - // launch waits for an execution slot then runs. On admission failure it - // records the rejection. It is invoked synchronously for foreground runs and - // in a goroutine for queued background runs. - launch := func() error { - if err := admit(); err != nil { - n.logger.Warn("run admission rejected", - zap.String("run_id", spec.RunID), - zap.String("adapter", spec.Adapter), - zap.String("reason", string(admissionRejectReason(err))), - ) - cancel() - n.runs.deregister(spec.RunID) - close(h.done) - n.rejectRun(context.Background(), sess, spec, err) - return fmt.Errorf("node: run %s: %w", spec.RunID, err) - } - return run() - } - - // Background runs return to the caller after enqueue; promotion and - // execution proceed in a goroutine. Foreground runs take the synchronous path. if spec.Background { - go func() { _ = launch() }() + go func() { _ = run() }() return nil } - return launch() + return run() } // admissionFor returns an admissionManager pairing the node-wide global gate -// with the per-adapter FIFO gate for the given adapter instance key. The -// adapter gate is created on first use from the adapter's Capabilities -// (MaxConcurrency, MaxQueue, QueueTimeoutMS) and shared across runs. +// with the per-adapter gate for the given adapter instance key. func (n *Node) admissionFor(adapterKey string, caps runtime.Capabilities) *admissionManager { return &admissionManager{ global: n.globalGate, @@ -224,19 +177,14 @@ func (n *Node) admissionFor(adapterKey string, caps runtime.Capabilities) *admis } } -// adapterGateFor returns the shared FIFO gate for the given adapter instance -// key, creating one from caps if not yet seen. MaxConcurrency <= 0 means the -// adapter gate is unlimited. +// adapterGateFor returns the shared gate for the given adapter instance key, +// creating one from caps if not yet seen. MaxConcurrency <= 0 means the adapter gate is unlimited. func (n *Node) adapterGateFor(adapterKey string, caps runtime.Capabilities) *fifoGate { n.adapterGatesMu.Lock() defer n.adapterGatesMu.Unlock() g, ok := n.adapterGates[adapterKey] if !ok { - g = newFifoGate( - caps.MaxConcurrency, - caps.MaxQueue, - time.Duration(caps.QueueTimeoutMS)*time.Millisecond, - ) + g = newFifoGate(caps.MaxConcurrency) n.adapterGates[adapterKey] = g } return g @@ -244,27 +192,24 @@ func (n *Node) adapterGateFor(adapterKey string, caps runtime.Capabilities) *fif // 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. -// The rejection message carries the concrete reason (queue_full, queue_timeout, -// concurrency_unavailable) extracted from the admission error. +// The rejection message carries the concrete reason (concurrency_unavailable) extracted from the admission error. func (n *Node) rejectRun(ctx context.Context, sess *transport.Session, spec runtime.ExecutionSpec, admitErr error) { errMsg := admissionMessage(admitErr) - // The run was inserted as "queued" before admission; mark it rejected. - // If the insert never happened (e.g. it was skipped on error), fall back to - // inserting a terminal rejected record so the rejection is still observable. + // Since we reject immediately without queueing, insert as a terminal rejected record. + 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)) + } 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 insErr := 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(), - }); insErr != nil { - n.logger.Warn("store: insert rejected run", zap.String("run_id", spec.RunID), zap.Error(insErr)) - } } if sess != nil && sess.IsAlive() { @@ -389,7 +334,6 @@ func (n *Node) handleCapabilitiesCommand(ctx context.Context, req *iop.NodeComma queued := 0 if ok { inFlight = gate.activeCount() - queued = gate.queuedCount() } result := map[string]string{ diff --git a/apps/node/internal/node/node_concurrency_integration_test.go b/apps/node/internal/node/node_concurrency_integration_test.go index 0865ab5..f48ad50 100644 --- a/apps/node/internal/node/node_concurrency_integration_test.go +++ b/apps/node/internal/node/node_concurrency_integration_test.go @@ -88,10 +88,10 @@ func startMockEdge(t *testing.T, ctx context.Context, listenAddr string) *toki.T } } -// TestQueueOverflow_RejectEventObservedByEdge is an integration test that -// verifies the RunEvent{type:"error"} sent by rejectRun() carries the queue -// overflow reason and actually arrives at the mock Edge server when the Node's -// FIFO admission queue is full (capacity=1, max_queue=0). +// TestOverDispatchSafety_RejectEventObservedByEdge is an integration test that +// verifies the RunEvent{type:"error"} sent by rejectRun() carries the concurrency +// unavailable reason and actually arrives at the mock Edge server when the Node's +// concurrency capacity limit is exceeded (capacity=1). // // Setup: // 1. Start mock Edge TCP server. @@ -100,9 +100,9 @@ func startMockEdge(t *testing.T, ctx context.Context, listenAddr string) *toki.T // 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 a RunEvent{type:"error", run_id:} with a queue_full reason +// 7. Assert a RunEvent{type:"error", run_id:} with a concurrency unavailable reason // arrives at the Edge. -func TestQueueOverflow_RejectEventObservedByEdge(t *testing.T) { +func TestOverDispatchSafety_RejectEventObservedByEdge(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() @@ -161,7 +161,7 @@ func TestQueueOverflow_RejectEventObservedByEdge(t *testing.T) { }) // Build real node.Node with global concurrency=1. - // capacity=1, max_queue=0 → a second concurrent run overflows the queue. + // capacity=1 → a second concurrent run overflows the concurrency limit. sa := newQueuedSlowAdapter("slow", 1, 0, 0) rtr := &fixedRouter{ adapterName: "slow", @@ -185,8 +185,7 @@ func TestQueueOverflow_RejectEventObservedByEdge(t *testing.T) { // Wait until the first run's adapter has started executing so the slot is held. waitStarted(t, sa, "run-hold-1") - // Send second RunRequest; should be rejected because capacity=1 is taken and - // the queue (max_queue=0) cannot hold it. + // Send second RunRequest; should be rejected because capacity=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) @@ -201,8 +200,8 @@ func TestQueueOverflow_RejectEventObservedByEdge(t *testing.T) { if ev.GetError() == "" { t.Fatal("reject RunEvent has empty error field") } - if !strings.Contains(ev.GetError(), "queue full") { - t.Fatalf("expected queue_full reason in reject event, got %q", ev.GetError()) + if !strings.Contains(ev.GetError(), "concurrency unavailable") { + t.Fatalf("expected concurrency unavailable reason in reject event, got %q", ev.GetError()) } // Success: RunEvent{type:"error", run_id:"run-reject-2"} observed. sa.releaseRun("run-hold-1") diff --git a/apps/node/internal/node/node_test.go b/apps/node/internal/node/node_test.go index ba81f2b..02900b6 100644 --- a/apps/node/internal/node/node_test.go +++ b/apps/node/internal/node/node_test.go @@ -718,10 +718,10 @@ func TestOnCommandRequest_Capabilities_InFlight(t *testing.T) { <-ba.done } -func TestOnCommandRequest_Capabilities_Queued(t *testing.T) { +func TestOnCommandRequest_Capabilities_SafetyRejected(t *testing.T) { sa := newQueuedSlowAdapter("slow", 1, 4, 0) router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} - n, st := makeNodeWithConcurrency(t, router, 0) + n, _ := makeNodeWithConcurrency(t, router, 0) if err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ RunId: "cap-hold", Adapter: "slow", Target: "v1", Background: true, @@ -730,12 +730,13 @@ func TestOnCommandRequest_Capabilities_Queued(t *testing.T) { } waitStarted(t, sa, "cap-hold") - if err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "cap-queued", Adapter: "slow", Target: "v1", Background: true, - }); err != nil { - t.Fatalf("queued run: %v", err) + // The second run request must fail immediately since there is no queue. + err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ + RunId: "cap-rejected", Adapter: "slow", Target: "v1", Background: true, + }) + if !errors.Is(err, node.ErrConcurrencyLimitExceeded) { + t.Fatalf("expected ErrConcurrencyLimitExceeded, got %v", err) } - requireQueued(t, st, "cap-queued") resp, err := n.OnCommandRequest(context.Background(), &transport.Session{}, &iop.NodeCommandRequest{ RequestId: "req-cap-queued", @@ -755,8 +756,8 @@ func TestOnCommandRequest_Capabilities_Queued(t *testing.T) { if got := resp.GetResult()["in_flight"]; got != "1" { t.Fatalf("result[in_flight]: got %q want %q", got, "1") } - if got := resp.GetResult()["queued"]; got != "1" { - t.Fatalf("result[queued]: got %q want %q", got, "1") + if got := resp.GetResult()["queued"]; got != "0" { + t.Fatalf("result[queued]: got %q want %q", got, "0") } if len(resp.GetProviderSnapshots()) != 1 { t.Fatalf("expected 1 provider snapshot, got %d", len(resp.GetProviderSnapshots())) @@ -768,14 +769,11 @@ func TestOnCommandRequest_Capabilities_Queued(t *testing.T) { if snap.GetInFlight() != 1 { t.Fatalf("snap.InFlight: got %d want %d", snap.GetInFlight(), 1) } - if snap.GetQueued() != 1 { - t.Fatalf("snap.Queued: got %d want %d", snap.GetQueued(), 1) + if snap.GetQueued() != 0 { + t.Fatalf("snap.Queued: got %d want %d", snap.GetQueued(), 0) } sa.releaseRun("cap-hold") - waitStarted(t, sa, "cap-queued") - sa.releaseRun("cap-queued") - requireStatusEventually(t, st, "cap-queued", "completed") } func TestOnCommandRequest_CapabilitiesProviderStatusModel(t *testing.T) { @@ -1296,10 +1294,10 @@ func (a *slowAdapterUnlimited) Execute(ctx context.Context, _ runtime.ExecutionS return nil } -// queuedSlowAdapter is a configurable adapter used by the FIFO admission tests. +// queuedSlowAdapter is a configurable adapter used by concurrency safety tests. // It advertises MaxConcurrency / MaxQueue / QueueTimeout via Capabilities, and // gives per-run control: each Execute blocks until the test releases (or fails) -// that specific run id, so tests can observe queue ordering deterministically. +// that specific run id, so tests can observe execution ordering deterministically. type queuedSlowAdapter struct { name string maxConcurrency int @@ -1386,25 +1384,6 @@ func waitStarted(t *testing.T, a *queuedSlowAdapter, runID string) { } } -// waitAnyStarted returns the next run id that began executing. -func waitAnyStarted(t *testing.T, a *queuedSlowAdapter) string { - t.Helper() - select { - case got := <-a.startSeq: - return got - case <-time.After(2 * time.Second): - t.Fatal("no run started within timeout") - return "" - } -} - -// requireQueued asserts the run is stored with status "queued" (polling briefly -// for the asynchronous store write). -func requireQueued(t *testing.T, st *store.Store, runID string) { - t.Helper() - requireStatusEventually(t, st, runID, "queued") -} - // requireStatus asserts the run is stored with exactly the given status now. func requireStatus(t *testing.T, st *store.Store, runID, status string) *store.RunRecord { t.Helper() @@ -1435,129 +1414,11 @@ func requireStatusEventually(t *testing.T, st *store.Store, runID, status string } } -// TestAdmissionQueue_ForegroundSecondWaitsThenRuns verifies that a second -// foreground run is not rejected: it waits in the FIFO queue and executes once -// the first run releases its slot. -func TestAdmissionQueue_ForegroundSecondWaitsThenRuns(t *testing.T) { - sa := newQueuedSlowAdapter("slow", 1, 4, 0) - router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} - n, st := makeNodeWithConcurrency(t, router, 1) - - first := make(chan error, 1) - go func() { - first <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "run-fg-1", Adapter: "slow", Target: "v1", - }) - }() - waitStarted(t, sa, "run-fg-1") - - second := make(chan error, 1) - go func() { - second <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "run-fg-2", Adapter: "slow", Target: "v1", - }) - }() - - // The second run must remain queued (not started, not rejected) while the - // first holds the only slot. - requireQueued(t, st, "run-fg-2") - select { - case err := <-second: - t.Fatalf("second run should be queued, but returned early: %v", err) - case <-time.After(150 * time.Millisecond): - } - - // Release the first; the second must then start and complete. - sa.releaseRun("run-fg-1") - if err := <-first; err != nil { - t.Fatalf("first run: %v", err) - } - waitStarted(t, sa, "run-fg-2") - sa.releaseRun("run-fg-2") - if err := <-second; err != nil { - t.Fatalf("second run should succeed after promotion, got %v", err) - } -} - -// TestAdmissionQueue_BackgroundQueuedReturnsThenRuns verifies that a background -// run that cannot run immediately returns after enqueue and is later promoted. -func TestAdmissionQueue_BackgroundQueuedReturnsThenRuns(t *testing.T) { - sa := newQueuedSlowAdapter("slow", 1, 4, 0) - router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} - n, st := makeNodeWithConcurrency(t, router, 1) - - first := make(chan error, 1) - go func() { - first <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "run-bg-hold", Adapter: "slow", Target: "v1", - }) - }() - waitStarted(t, sa, "run-bg-hold") - - // Background queued run returns promptly after enqueue. - if err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "run-bg-2", Adapter: "slow", Target: "v1", Background: true, - }); err != nil { - t.Fatalf("background queued run should enqueue without error, got %v", err) - } - requireQueued(t, st, "run-bg-2") - - // Release the holder; the queued background run must be promoted and run. - sa.releaseRun("run-bg-hold") - if err := <-first; err != nil { - t.Fatalf("hold run: %v", err) - } - waitStarted(t, sa, "run-bg-2") - sa.releaseRun("run-bg-2") - requireStatusEventually(t, st, "run-bg-2", "completed") -} - -// TestAdmissionQueue_FIFOOrdering verifies that queued runs are promoted in -// first-in-first-out order. -func TestAdmissionQueue_FIFOOrdering(t *testing.T) { - sa := newQueuedSlowAdapter("slow", 1, 8, 0) - router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} - n, _ := makeNodeWithConcurrency(t, router, 1) - - // First run holds the single slot. - hold := make(chan error, 1) - go func() { - hold <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "fifo-0", Adapter: "slow", Target: "v1", Background: true, - }) - }() - waitStarted(t, sa, "fifo-0") - - // Enqueue three background runs in a deterministic order. Each enqueue - // returns before the next is submitted, so queue order is well-defined. - order := []string{"fifo-1", "fifo-2", "fifo-3"} - for _, id := range order { - if err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: id, Adapter: "slow", Target: "v1", Background: true, - }); err != nil { - t.Fatalf("enqueue %s: %v", id, err) - } - } - - // Release runs one at a time and assert they start in FIFO order. - sa.releaseRun("fifo-0") - if err := <-hold; err != nil { - t.Fatalf("fifo-0: %v", err) - } - for _, id := range order { - got := waitAnyStarted(t, sa) - if got != id { - t.Fatalf("FIFO order violated: expected %s to start next, got %s", id, got) - } - sa.releaseRun(id) - } -} - -// TestAdmissionQueue_MaxQueueOverflow verifies that a run is rejected with -// reason queue_full when the FIFO queue is already at max_queue. -func TestAdmissionQueue_MaxQueueOverflow(t *testing.T) { - // capacity=1, max_queue=1: one running + one queued is the ceiling. - sa := newQueuedSlowAdapter("slow", 1, 1, 0) +// TestOnRunRequest_OverDispatchSafetyRejectsWithoutQueue verifies that a run +// is rejected immediately with reason concurrency_unavailable when the concurrency limit is reached (no queue). +func TestOnRunRequest_OverDispatchSafetyRejectsWithoutQueue(t *testing.T) { + // capacity=1: one running is the ceiling. + sa := newQueuedSlowAdapter("slow", 1, 0, 0) router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} n, st := makeNodeWithConcurrency(t, router, 0) // global unlimited; adapter cap governs @@ -1569,15 +1430,7 @@ func TestAdmissionQueue_MaxQueueOverflow(t *testing.T) { }() waitStarted(t, sa, "of-hold") - // Fill the single queue slot. - if err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "of-queued", Adapter: "slow", Target: "v1", Background: true, - }); err != nil { - t.Fatalf("first queued run should enqueue, got %v", err) - } - requireQueued(t, st, "of-queued") - - // Third run overflows the queue and must be rejected with queue_full. + // Second run overflows the capacity and must be rejected immediately. err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ RunId: "of-reject", Adapter: "slow", Target: "v1", }) @@ -1585,118 +1438,19 @@ func TestAdmissionQueue_MaxQueueOverflow(t *testing.T) { t.Fatalf("expected ErrConcurrencyLimitExceeded, got %v", err) } rec := requireStatus(t, st, "of-reject", "rejected") - if !strings.Contains(rec.Error, "queue full") { - t.Fatalf("expected queue_full reason in error, got %q", rec.Error) + if !strings.Contains(rec.Error, "concurrency unavailable") { + t.Fatalf("expected concurrency unavailable reason in error, got %q", rec.Error) } sa.releaseRun("of-hold") if err := <-hold; err != nil { t.Fatalf("of-hold: %v", err) } - waitStarted(t, sa, "of-queued") - sa.releaseRun("of-queued") } -// TestAdmissionQueue_QueueTimeout verifies that a queued run is rejected with -// reason queue_timeout when it waits past queue_timeout_ms. -func TestAdmissionQueue_QueueTimeout(t *testing.T) { - // capacity=1, max_queue=4, queue_timeout=120ms. - sa := newQueuedSlowAdapter("slow", 1, 4, 120*time.Millisecond) - router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} - n, st := makeNodeWithConcurrency(t, router, 0) - - hold := make(chan error, 1) - go func() { - hold <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "to-hold", Adapter: "slow", Target: "v1", Background: true, - }) - }() - waitStarted(t, sa, "to-hold") - - // This foreground run will wait in the queue and time out (holder is never - // released until after the timeout). - err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "to-timeout", Adapter: "slow", Target: "v1", - }) - if !errors.Is(err, node.ErrConcurrencyLimitExceeded) { - t.Fatalf("expected ErrConcurrencyLimitExceeded on queue timeout, got %v", err) - } - rec := requireStatus(t, st, "to-timeout", "rejected") - if !strings.Contains(rec.Error, "queue timeout") { - t.Fatalf("expected queue_timeout reason in error, got %q", rec.Error) - } - - sa.releaseRun("to-hold") - if err := <-hold; err != nil { - t.Fatalf("to-hold: %v", err) - } -} - -// TestAdmissionQueue_ReleaseAfterFailurePromotesNext verifies that a queued run -// is promoted when the running run fails. -func TestAdmissionQueue_ReleaseAfterFailurePromotesNext(t *testing.T) { - sa := newQueuedSlowAdapter("slow", 1, 4, 0) - router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} - n, st := makeNodeWithConcurrency(t, router, 1) - - first := make(chan error, 1) - go func() { - first <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "fail-1", Adapter: "slow", Target: "v1", - }) - }() - waitStarted(t, sa, "fail-1") - - if err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "fail-next", Adapter: "slow", Target: "v1", Background: true, - }); err != nil { - t.Fatalf("enqueue fail-next: %v", err) - } - requireQueued(t, st, "fail-next") - - // First run fails; its slot must promote the queued run. - sa.failRun("fail-1") - <-first - waitStarted(t, sa, "fail-next") - sa.releaseRun("fail-next") - requireStatusEventually(t, st, "fail-next", "completed") -} - -// TestAdmissionQueue_ReleaseAfterCancelPromotesNext verifies that a queued run -// is promoted when the running run is cancelled. -func TestAdmissionQueue_ReleaseAfterCancelPromotesNext(t *testing.T) { - sa := newQueuedSlowAdapter("slow", 1, 4, 0) - router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} - n, st := makeNodeWithConcurrency(t, router, 1) - - first := make(chan error, 1) - go func() { - first <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "cancel-1", Adapter: "slow", Target: "v1", Background: true, - }) - }() - waitStarted(t, sa, "cancel-1") - - if err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "cancel-next", Adapter: "slow", Target: "v1", Background: true, - }); err != nil { - t.Fatalf("enqueue cancel-next: %v", err) - } - requireQueued(t, st, "cancel-next") - <-first - - // Cancel the running run; its slot must promote the queued run. - if err := n.OnCancel(context.Background(), &transport.Session{}, &iop.CancelRequest{RunId: "cancel-1"}); err != nil { - t.Fatalf("cancel: %v", err) - } - waitStarted(t, sa, "cancel-next") - sa.releaseRun("cancel-next") - requireStatusEventually(t, st, "cancel-next", "completed") -} - -// TestAdmissionQueue_PermitReleasedAfterCompletion verifies that after the first -// run completes, a subsequent run can acquire the slot without queuing. -func TestAdmissionQueue_PermitReleasedAfterCompletion(t *testing.T) { +// TestOnRunRequest_PermitReleasedAfterCompletion verifies that after the first +// run completes, a subsequent run can acquire the slot immediately. +func TestOnRunRequest_PermitReleasedAfterCompletion(t *testing.T) { sa := newQueuedSlowAdapter("slow", 1, 4, 0) router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} n, _ := makeNodeWithConcurrency(t, router, 1) @@ -1750,323 +1504,10 @@ func TestConcurrencyLimit_Unlimited(t *testing.T) { } } -// TestAdmissionQueue_AdapterCapQueues verifies that when global=0 and adapter -// cap=1, a second run queues (rather than being rejected) and runs after release. -func TestAdmissionQueue_AdapterCapQueues(t *testing.T) { - sa := newQueuedSlowAdapter("slow", 1, 4, 0) // adapter cap=1 - router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} - nd, st := makeNodeWithConcurrency(t, router, 0) // global=0 → adapter cap governs - - first := make(chan error, 1) - go func() { - first <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "cap-1", Adapter: "slow", Target: "v1", - }) - }() - waitStarted(t, sa, "cap-1") - - if err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "cap-2", Adapter: "slow", Target: "v1", Background: true, - }); err != nil { - t.Fatalf("cap-2 should enqueue, got %v", err) - } - requireQueued(t, st, "cap-2") - - sa.releaseRun("cap-1") - if err := <-first; err != nil { - t.Fatalf("cap-1: %v", err) - } - waitStarted(t, sa, "cap-2") - sa.releaseRun("cap-2") - requireStatusEventually(t, st, "cap-2", "completed") -} - -// TestAdmissionQueue_GlobalBeatsAdapterCap verifies that the global gate (=1) -// limits effective in-flight to 1 even when the adapter cap is looser (=4): the -// second run queues behind the global slot. -func TestAdmissionQueue_GlobalBeatsAdapterCap(t *testing.T) { - sa := newQueuedSlowAdapter("slow-hicap", 4, 4, 0) // adapter cap=4 - router := &fixedRouter{adapterName: "slow-hicap", adapters: map[string]runtime.Adapter{"slow-hicap": sa}} - nd, st := makeNodeWithConcurrency(t, router, 1) // global=1 → effective=1 - - first := make(chan error, 1) - go func() { - first <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "g1-1", Adapter: "slow-hicap", Target: "v1", - }) - }() - waitStarted(t, sa, "g1-1") - - // Second run queues behind the global gate even though adapter cap=4. - if err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "g1-2", Adapter: "slow-hicap", Target: "v1", Background: true, - }); err != nil { - t.Fatalf("g1-2 should enqueue, got %v", err) - } - requireQueued(t, st, "g1-2") - - sa.releaseRun("g1-1") - if err := <-first; err != nil { - t.Fatalf("g1-1: %v", err) - } - waitStarted(t, sa, "g1-2") - sa.releaseRun("g1-2") - requireStatusEventually(t, st, "g1-2", "completed") -} - -// TestAdmissionQueue_GlobalBeatsAdapterCap_DifferentAdapters verifies that -// global=1 makes a run on adapter B queue while adapter A holds the global slot, -// even though both adapter caps are > 1. -func TestAdmissionQueue_GlobalBeatsAdapterCap_DifferentAdapters(t *testing.T) { - saA := newQueuedSlowAdapter("slow-hicap-a", 4, 4, 0) - saB := newQueuedSlowAdapter("slow-hicap-b", 4, 4, 0) - - router := &fixedRouter{ - adapterName: "slow-hicap-a", - adapters: map[string]runtime.Adapter{ - "slow-hicap-a": saA, - "slow-hicap-b": saB, - }, - } - nd, st := makeNodeWithConcurrency(t, router, 1) // global=1 - - first := make(chan error, 1) - go func() { - first <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "xa-1", Adapter: "slow-hicap-a", Target: "v1", - }) - }() - waitStarted(t, saA, "xa-1") - - // Run on adapter B queues because the global slot is taken by adapter A. - router.adapterName = "slow-hicap-b" - if err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "xb-1", Adapter: "slow-hicap-b", Target: "v1", Background: true, - }); err != nil { - t.Fatalf("xb-1 should enqueue, got %v", err) - } - requireQueued(t, st, "xb-1") - - sa := saA - sa.releaseRun("xa-1") - if err := <-first; err != nil { - t.Fatalf("xa-1: %v", err) - } - waitStarted(t, saB, "xb-1") - saB.releaseRun("xb-1") - requireStatusEventually(t, st, "xb-1", "completed") -} - -// TestAdmissionQueue_GlobalNoCrossAdapterHeadOfLine is the regression test for -// the gate-composition fix: a queued run on one adapter must not occupy a global -// slot it cannot yet use and thereby block a runnable queue head on a different -// adapter. -// -// Setup: global=2, adapter A capacity=1, adapter B capacity=1. -// - A1 and B1 run, consuming both global slots and both adapter slots. -// - A2 is enqueued first (waiting on adapter A slot, held by A1). -// - B2 is enqueued next (waiting on adapter B slot, held by B1). -// - Releasing only B1 must let B2 start immediately, because A2 (still without -// an adapter A slot) must NOT have grabbed the freed global slot ahead of it. -func TestAdmissionQueue_GlobalNoCrossAdapterHeadOfLine(t *testing.T) { - saA := newQueuedSlowAdapter("adapter-a", 1, 4, 0) - saB := newQueuedSlowAdapter("adapter-b", 1, 4, 0) - router := &fixedRouter{ - adapterName: "adapter-a", - adapters: map[string]runtime.Adapter{ - "adapter-a": saA, - "adapter-b": saB, - }, - } - nd, st := makeNodeWithConcurrency(t, router, 2) // global=2 - - // A1 and B1 occupy both global slots and both adapter slots. - a1 := make(chan error, 1) - go func() { - a1 <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "a1", Adapter: "adapter-a", Target: "v1", Background: true, - }) - }() - waitStarted(t, saA, "a1") - - router.adapterName = "adapter-b" - b1 := make(chan error, 1) - go func() { - b1 <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "b1", Adapter: "adapter-b", Target: "v1", Background: true, - }) - }() - waitStarted(t, saB, "b1") - - // Enqueue A2 first (waits on adapter A slot), then B2 (waits on adapter B slot). - router.adapterName = "adapter-a" - if err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "a2", Adapter: "adapter-a", Target: "v1", Background: true, - }); err != nil { - t.Fatalf("enqueue a2: %v", err) - } - requireQueued(t, st, "a2") - - router.adapterName = "adapter-b" - if err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "b2", Adapter: "adapter-b", Target: "v1", Background: true, - }); err != nil { - t.Fatalf("enqueue b2: %v", err) - } - requireQueued(t, st, "b2") - - // Release only B1: its adapter B slot and one global slot free up. B2 must - // start. A2 stays queued because adapter A slot (a1) is still held. - <-b1 - saB.releaseRun("b1") - waitStarted(t, saB, "b2") // fails here if A2 head-of-line-blocked the global slot - - // A2 must still be queued (it has no adapter A slot yet). - requireStatus(t, st, "a2", "queued") - select { - case <-saA.startSeq: - t.Fatal("a2 must not start before a1 releases its adapter slot") - case <-time.After(100 * time.Millisecond): - } - - // Now release a1 → A2 can take adapter A slot and the remaining global slot. - <-a1 - saA.releaseRun("a1") - waitStarted(t, saA, "a2") - - saB.releaseRun("b2") - saA.releaseRun("a2") - requireStatusEventually(t, st, "a2", "completed") - requireStatusEventually(t, st, "b2", "completed") -} - -// TestAdmissionQueue_QueueTimeoutAppliesAcrossGates verifies that the queue -// timeout bounds the total admission wait even when the run is blocked waiting -// for the global slot after acquiring its adapter slot. -func TestAdmissionQueue_QueueTimeoutAppliesAcrossGates(t *testing.T) { - // adapter A and B each capacity=1 with a queue timeout; global=1 is the - // bottleneck so the second adapter's run holds its adapter slot but blocks on - // the global gate, then must time out. - saA := newQueuedSlowAdapter("adapter-a", 1, 4, 120*time.Millisecond) - saB := newQueuedSlowAdapter("adapter-b", 1, 4, 120*time.Millisecond) - router := &fixedRouter{ - adapterName: "adapter-a", - adapters: map[string]runtime.Adapter{ - "adapter-a": saA, - "adapter-b": saB, - }, - } - nd, st := makeNodeWithConcurrency(t, router, 1) // global=1 - - a1 := make(chan error, 1) - go func() { - a1 <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "ta1", Adapter: "adapter-a", Target: "v1", Background: true, - }) - }() - waitStarted(t, saA, "ta1") - - // Run on adapter B acquires adapter B slot immediately but blocks on global=1. - // It must time out via the queue timeout rather than waiting indefinitely. - router.adapterName = "adapter-b" - err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "tb1", Adapter: "adapter-b", Target: "v1", - }) - if !errors.Is(err, node.ErrConcurrencyLimitExceeded) { - t.Fatalf("expected ErrConcurrencyLimitExceeded on global-wait timeout, got %v", err) - } - rec := requireStatus(t, st, "tb1", "rejected") - if !strings.Contains(rec.Error, "queue timeout") { - t.Fatalf("expected queue_timeout reason for global-wait timeout, got %q", rec.Error) - } - - <-a1 - saA.releaseRun("ta1") -} - -// TestAdmissionQueue_ParentTimeoutNotQueueTimeout verifies that when a queued -// run is terminated by its parent RunRequest.timeout_sec (not the provider -// queue timeout), the rejection reason is NOT misreported as queue_timeout. With -// queue_timeout_ms=0 the queue timeout mechanism is disabled, so the bogus -// "queue timeout exceeded (queue_timeout=0s)" message must never appear. -func TestAdmissionQueue_ParentTimeoutNotQueueTimeout(t *testing.T) { - // queue_timeout_ms=0 (disabled), max_queue=4, capacity=1. - sa := newQueuedSlowAdapter("slow", 1, 4, 0) - router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} - nd, st := makeNodeWithConcurrency(t, router, 0) - - hold := make(chan error, 1) - go func() { - hold <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "pt-hold", Adapter: "slow", Target: "v1", Background: true, - }) - }() - waitStarted(t, sa, "pt-hold") - - // Second foreground run queues (queue enabled, slot busy) and is terminated - // by its own 1s request timeout, not by a queue timeout. - err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "pt-timeout", Adapter: "slow", Target: "v1", TimeoutSec: 1, - }) - if !errors.Is(err, node.ErrConcurrencyLimitExceeded) { - t.Fatalf("expected ErrConcurrencyLimitExceeded on parent request timeout, got %v", err) - } - if strings.Contains(err.Error(), "queue timeout") { - t.Fatalf("parent request timeout must not be reported as queue timeout, got %q", err.Error()) - } - rec := requireStatus(t, st, "pt-timeout", "rejected") - if strings.Contains(rec.Error, "queue timeout") || strings.Contains(rec.Error, "queue_timeout=0s") { - t.Fatalf("store error must not contain a queue_timeout reason, got %q", rec.Error) - } - - sa.releaseRun("pt-hold") - if err := <-hold; err != nil { - t.Fatalf("pt-hold: %v", err) - } -} - -// TestAdmissionQueue_ParentTimeoutBeatsQueueTimeout verifies that even when a -// provider queue timeout IS configured, a parent request timeout that fires -// first is still classified as concurrency_unavailable, not queue_timeout. -func TestAdmissionQueue_ParentTimeoutBeatsQueueTimeout(t *testing.T) { - // queue_timeout_ms=10s (long), but the run's request timeout is 1s, so the - // parent deadline fires first. - sa := newQueuedSlowAdapter("slow", 1, 4, 10*time.Second) - router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} - nd, st := makeNodeWithConcurrency(t, router, 0) - - hold := make(chan error, 1) - go func() { - hold <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "pb-hold", Adapter: "slow", Target: "v1", Background: true, - }) - }() - waitStarted(t, sa, "pb-hold") - - err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ - RunId: "pb-timeout", Adapter: "slow", Target: "v1", TimeoutSec: 1, - }) - if !errors.Is(err, node.ErrConcurrencyLimitExceeded) { - t.Fatalf("expected ErrConcurrencyLimitExceeded, got %v", err) - } - if strings.Contains(err.Error(), "queue timeout") { - t.Fatalf("parent deadline (earlier than queue timeout) must not be queue timeout, got %q", err.Error()) - } - rec := requireStatus(t, st, "pb-timeout", "rejected") - if strings.Contains(rec.Error, "queue timeout") { - t.Fatalf("store error must not contain queue timeout, got %q", rec.Error) - } - - sa.releaseRun("pb-hold") - if err := <-hold; err != nil { - t.Fatalf("pb-hold: %v", err) - } -} - -// TestConcurrencyLimit_RejectStoreAndEvent verifies that a rejected run (queue -// overflow) is stored with terminal status "rejected" and an error message. +// TestConcurrencyLimit_RejectStoreAndEvent verifies that a rejected run +// is stored with terminal status "rejected" and an error message. func TestConcurrencyLimit_RejectStoreAndEvent(t *testing.T) { - // capacity=1, max_queue=0: any second concurrent run overflows immediately. + // capacity=1: any second concurrent run overflows immediately. sa := newQueuedSlowAdapter("slow", 1, 0, 0) router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} nd, st := makeNodeWithConcurrency(t, router, 0) @@ -2079,7 +1520,7 @@ func TestConcurrencyLimit_RejectStoreAndEvent(t *testing.T) { }() waitStarted(t, sa, "run-hold") - // Second run must be rejected (queue capacity is 0). + // Second run must be rejected. err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ RunId: "run-rejected", Adapter: "slow", Target: "v1", }) diff --git a/apps/node/internal/node/run_manager.go b/apps/node/internal/node/run_manager.go index 39af698..107597c 100644 --- a/apps/node/internal/node/run_manager.go +++ b/apps/node/internal/node/run_manager.go @@ -4,14 +4,10 @@ import ( "context" "errors" "sync" - "time" ) // ErrConcurrencyLimitExceeded is returned when a run is rejected because -// admission could not be granted under the current concurrency policy. It is -// kept as the wrapping sentinel for backward compatibility; the concrete reason -// (queue full, queue timeout, concurrency unavailable) is carried by -// admissionError and surfaced through admissionRejectReason. +// admission could not be granted under the current concurrency policy. var ErrConcurrencyLimitExceeded = errors.New("concurrency limit exceeded") type runHandle struct { @@ -55,29 +51,16 @@ func (m *runManager) cancelRun(runID string) bool { return true } -// rejectReason classifies why a run could not be admitted or had to be -// terminated before execution. +// rejectReason classifies why a run could not be admitted. type rejectReason string const ( - // reasonQueueFull means the FIFO admission queue is already at max_queue. - reasonQueueFull rejectReason = "queue_full" - // reasonQueueTimeout means the request waited past queue_timeout_ms. - reasonQueueTimeout rejectReason = "queue_timeout" - // reasonConcurrencyUnavailable means admission failed because the caller's - // context was cancelled while waiting for a slot. + // reasonConcurrencyUnavailable means admission failed because the concurrency limit was exceeded. reasonConcurrencyUnavailable rejectReason = "concurrency_unavailable" ) -// errAdmissionQueueTimeout is the context cancellation cause attached to the -// provider queue-timeout deadline. It lets wait() distinguish a real -// queue_timeout_ms expiry from an unrelated parent deadline (e.g. the run's -// RunRequest.timeout_sec), so only the former is reported as queue_timeout. -var errAdmissionQueueTimeout = errors.New("admission queue timeout") - // admissionError carries the concrete rejection reason while still satisfying -// errors.Is(err, ErrConcurrencyLimitExceeded) for callers relying on the -// legacy sentinel. +// errors.Is(err, ErrConcurrencyLimitExceeded) for callers relying on the legacy sentinel. type admissionError struct { reason rejectReason msg string @@ -87,37 +70,14 @@ func (e *admissionError) Error() string { return e.msg } func (e *admissionError) Is(target error) bool { return target == ErrConcurrencyLimitExceeded } -func newQueueFullError(maxQueue int) *admissionError { - return &admissionError{ - reason: reasonQueueFull, - msg: queueFullMessage(maxQueue), - } -} - -func newQueueTimeoutError(timeout time.Duration) *admissionError { - return &admissionError{ - reason: reasonQueueTimeout, - msg: queueTimeoutMessage(timeout), - } -} - func newConcurrencyUnavailableError() *admissionError { return &admissionError{ reason: reasonConcurrencyUnavailable, - msg: "concurrency unavailable: admission cancelled before a slot was free", + msg: "concurrency unavailable: admission rejected because safety limit exceeded", } } -func queueFullMessage(maxQueue int) string { - return "queue full (max_queue=" + itoa(maxQueue) + ")" -} - -func queueTimeoutMessage(timeout time.Duration) string { - return "queue timeout exceeded (queue_timeout=" + timeout.String() + ")" -} - -// admissionRejectReason extracts the rejectReason from an admission error, -// defaulting to concurrency_unavailable for unknown errors. +// admissionRejectReason extracts the rejectReason from an admission error. func admissionRejectReason(err error) rejectReason { var ae *admissionError if errors.As(err, &ae) { @@ -138,246 +98,88 @@ func admissionMessage(err error) string { return string(reasonConcurrencyUnavailable) } -// fifoGate is a FIFO-fair counting semaphore. capacity <= 0 means unlimited -// (acquire always succeeds immediately and is never queued). maxQueue < 0 means -// an unbounded queue; maxQueue == 0 disallows queueing (a full gate rejects -// immediately); maxQueue > 0 caps the queue length. queueTimeout <= 0 means no -// timeout while waiting. +// fifoGate is a simple in-flight concurrency semaphore (no queue). type fifoGate struct { - mu sync.Mutex - capacity int - maxQueue int - queueTimeout time.Duration - inFlight int - waiters []chan struct{} + mu sync.Mutex + capacity int + inFlight int } -func newFifoGate(capacity, maxQueue int, queueTimeout time.Duration) *fifoGate { +func newFifoGate(capacity int) *fifoGate { return &fifoGate{ - capacity: capacity, - maxQueue: maxQueue, - queueTimeout: queueTimeout, + capacity: capacity, } } -// reserve takes a slot without blocking. When capacity is free it grants the -// slot immediately (granted=true, ch=nil). When full it appends a FIFO waiter -// and returns its channel (granted=false, ch!=nil) which the caller waits on via -// waitOn. When the queue is already at max_queue it returns a queue_full error. -func (g *fifoGate) reserve() (granted bool, ch chan struct{}, err error) { +func (g *fifoGate) acquire() bool { g.mu.Lock() defer g.mu.Unlock() if g.capacity <= 0 || g.inFlight < g.capacity { g.inFlight++ - return true, nil, nil - } - if g.maxQueue >= 0 && len(g.waiters) >= g.maxQueue { - return false, nil, newQueueFullError(g.maxQueue) - } - ch = make(chan struct{}) - g.waiters = append(g.waiters, ch) - return false, ch, nil -} - -// waitOn blocks until the reserved waiter ch is promoted to an execution slot or -// the context is cancelled (the caller derives any queue-timeout deadline onto -// ctx). On context cancellation it abandons the waiter and returns an -// admissionError; on success (including a promotion that races the cancel) it -// returns nil and the caller holds a live slot. -func (g *fifoGate) waitOn(ctx context.Context, ch chan struct{}) error { - select { - case <-ch: - // Promoted to an execution slot; inFlight was incremented by release. - return nil - case <-ctx.Done(): - if g.abandon(ch) { - return newConcurrencyUnavailableError() - } - return nil - } -} - -// abandon removes ch from the waiter queue. It returns true if ch was still -// waiting (so the caller owns the failure), or false if ch had already been -// promoted (so the caller now holds a live slot it must release). -func (g *fifoGate) abandon(ch chan struct{}) bool { - g.mu.Lock() - defer g.mu.Unlock() - for i, w := range g.waiters { - if w == ch { - g.waiters = append(g.waiters[:i], g.waiters[i+1:]...) - return true - } + return true } return false } -// release returns a slot. If a waiter is queued it is promoted in FIFO order, -// keeping inFlight unchanged (the slot transfers to the promoted waiter). func (g *fifoGate) release() { g.mu.Lock() defer g.mu.Unlock() - if len(g.waiters) > 0 { - next := g.waiters[0] - g.waiters = g.waiters[1:] - close(next) - return - } if g.inFlight > 0 { g.inFlight-- } } -// activeCount returns the number of in-flight (executing) slots. func (g *fifoGate) activeCount() int { g.mu.Lock() defer g.mu.Unlock() return g.inFlight } -// queuedCount returns the number of requests currently waiting for a slot. func (g *fifoGate) queuedCount() int { - g.mu.Lock() - defer g.mu.Unlock() - return len(g.waiters) + return 0 } // admissionManager composes the node-wide global gate with a per-adapter gate. -// A run is admitted only when it holds both: the adapter gate is reserved first -// (so per-provider FIFO ordering is preserved without head-of-line blocking -// across adapters), then the global gate. type admissionManager struct { global *fifoGate adapter *fifoGate } -// admissionTicket holds the reservation state between enqueue and wait. queued -// reports whether the request had to wait for a slot (vs. running immediately). type admissionTicket struct { mgr *admissionManager - adapterCh chan struct{} // non-nil if waiting on the adapter gate - globalCh chan struct{} // non-nil if waiting on the global gate adapterHeld bool globalHeld bool - queued bool } -// enqueue reserves only the per-adapter FIFO position, without blocking. It -// returns a queue_full admissionError when the adapter queue is at max_queue; -// otherwise it returns a ticket whose wait() blocks until the run is admitted. -// The global gate is intentionally NOT reserved here: reserving it before the -// adapter slot is granted would let a run still waiting for its adapter slot -// occupy a global slot, blocking a different adapter's runnable queue head -// (cross-adapter head-of-line blocking). The global slot is reserved in wait() -// only after the adapter slot is actually held. Background callers can return to -// the transport after enqueue and run wait()+execute in a goroutine. -func (a *admissionManager) enqueue() (*admissionTicket, error) { - t := &admissionTicket{mgr: a} - - granted, ch, err := a.adapter.reserve() - if err != nil { - return nil, err - } - if granted { - t.adapterHeld = true - } else { - t.adapterCh = ch - t.queued = true - } - return t, nil -} - -// wait blocks until the run holds both the adapter slot and the global slot. The -// adapter slot is awaited first; only once it is held is the global slot -// reserved and awaited, so a queued run never occupies a global slot it cannot -// yet use. The adapter gate's queue timeout (when configured) bounds the total -// admission wait across both gates via a derived deadline. On failure it unwinds -// any held slot and returns the admissionError. -func (t *admissionTicket) wait(ctx context.Context) error { - // Derive a single deadline from the adapter queue timeout so the global wait - // cannot extend total queue time past the configured policy. - if to := t.mgr.adapter.queueTimeout; to > 0 { - var cancel context.CancelFunc - ctx, cancel = context.WithTimeoutCause(ctx, to, errAdmissionQueueTimeout) - defer cancel() - } - - if t.adapterCh != nil { - if err := t.mgr.adapter.waitOn(ctx, t.adapterCh); err != nil { - t.adapterCh = nil - t.release() - return t.mapDeadline(ctx, err) - } - t.adapterCh = nil - t.adapterHeld = true - } - - if t.mgr.global != nil { - granted, ch, err := t.mgr.global.reserve() - if err != nil { - // The global gate has an unbounded queue, so reserve never returns an - // overflow error; guard defensively and unwind the adapter slot. - t.release() - return err - } - if granted { - t.globalHeld = true - } else if err := t.mgr.global.waitOn(ctx, ch); err != nil { - t.release() - return t.mapDeadline(ctx, err) - } else { - t.globalHeld = true +func (a *admissionManager) acquire() (*admissionTicket, error) { + if a.adapter != nil { + if !a.adapter.acquire() { + return nil, newConcurrencyUnavailableError() } } - return nil -} - -// mapDeadline reclassifies a context-cancellation admission error as a queue -// timeout only when the cancellation cause is the provider queue-timeout -// deadline. A parent cancellation or a parent deadline (e.g. the run's -// RunRequest.timeout_sec) keeps the original concurrency_unavailable reason, so -// an unrelated request timeout is never mislabeled as queue_timeout (which would -// also wrongly emit queue_timeout=0s when queue_timeout_ms is disabled). -func (t *admissionTicket) mapDeadline(ctx context.Context, err error) error { - if errors.Is(context.Cause(ctx), errAdmissionQueueTimeout) && admissionRejectReason(err) == reasonConcurrencyUnavailable { - return newQueueTimeoutError(t.mgr.adapter.queueTimeout) + if a.global != nil { + if !a.global.acquire() { + if a.adapter != nil { + a.adapter.release() + } + return nil, newConcurrencyUnavailableError() + } } - return err + return &admissionTicket{ + mgr: a, + adapterHeld: a.adapter != nil, + globalHeld: a.global != nil, + }, nil } -// release returns any slot the ticket holds. It is safe to call once the ticket -// is done (after a completed run) or to unwind a partial reservation on failure. func (t *admissionTicket) release() { if t.globalHeld && t.mgr.global != nil { t.mgr.global.release() t.globalHeld = false } - if t.adapterHeld { + if t.adapterHeld && t.mgr.adapter != nil { t.mgr.adapter.release() t.adapterHeld = false } } -// itoa is a tiny strconv.Itoa shim kept local to avoid importing strconv solely -// for reason-message formatting. -func itoa(n int) string { - if n == 0 { - return "0" - } - neg := n < 0 - if neg { - n = -n - } - var buf [20]byte - i := len(buf) - for n > 0 { - i-- - buf[i] = byte('0' + n%10) - n /= 10 - } - if neg { - i-- - buf[i] = '-' - } - return string(buf[i:]) -}