759 lines
22 KiB
Go
759 lines
22 KiB
Go
package node_test
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"io"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strings"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"go.uber.org/zap"
|
|
"google.golang.org/protobuf/types/known/structpb"
|
|
"iop/apps/node/internal/adapters"
|
|
"iop/apps/node/internal/node"
|
|
"iop/apps/node/internal/router"
|
|
"iop/apps/node/internal/store"
|
|
"iop/apps/node/internal/transport"
|
|
runtime "iop/packages/go/agentruntime"
|
|
iop "iop/proto/gen/iop"
|
|
)
|
|
|
|
// --- admission gate / config refresh tests ---
|
|
|
|
// TestNodeConfigRefreshWithoutApplyManagerReportsRestartRequired verifies that
|
|
// a Node without a live apply manager responds restart_required for any refresh
|
|
// request that carries changed paths, and applied for a no-op (empty) request.
|
|
func TestNodeConfigRefreshWithoutApplyManagerReportsRestartRequired(t *testing.T) {
|
|
router := &fixedRouter{adapterName: "test", adapters: map[string]runtime.Provider{"test": &countingAdapter{}}}
|
|
n, _ := makeNode(t, router)
|
|
|
|
// Non-empty changed_paths → restart_required.
|
|
resp, err := n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
|
|
RequestId: "req-restart",
|
|
ChangedPaths: []string{"nodes.0.providers.0.capacity"},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("OnConfigRefresh: %v", err)
|
|
}
|
|
if resp.GetStatus() != iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_RESTART_REQUIRED {
|
|
t.Fatalf("expected restart_required, got %v", resp.GetStatus())
|
|
}
|
|
if resp.GetRequestId() != "req-restart" {
|
|
t.Fatalf("expected request_id=req-restart, got %q", resp.GetRequestId())
|
|
}
|
|
if len(resp.GetRestartRequiredPaths()) == 0 {
|
|
t.Fatal("expected restart_required_paths to be populated")
|
|
}
|
|
|
|
// No config and no changed_paths → applied (no-op).
|
|
resp2, err := n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
|
|
RequestId: "req-noop",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("OnConfigRefresh noop: %v", err)
|
|
}
|
|
if resp2.GetStatus() != iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_APPLIED {
|
|
t.Fatalf("expected applied for no-op, got %v", resp2.GetStatus())
|
|
}
|
|
}
|
|
|
|
func TestNodeConfigRefreshAppliesOpenAICompatCapacity(t *testing.T) {
|
|
initialPayload := &iop.NodeConfigPayload{
|
|
Adapters: []*iop.AdapterConfig{
|
|
{
|
|
Name: "openai",
|
|
Type: "openai_compat",
|
|
Enabled: true,
|
|
Config: &iop.AdapterConfig_OpenaiCompat{
|
|
OpenaiCompat: &iop.OpenAICompatAdapterConfig{
|
|
Provider: "openai",
|
|
Endpoint: "http://localhost:8080",
|
|
Capacity: 2,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
set, err := adapters.BuildConfigSet(initialPayload, zap.NewNop())
|
|
if err != nil {
|
|
t.Fatalf("BuildConfigSet: %v", err)
|
|
}
|
|
|
|
rtr := router.New(set.Registry, zap.NewNop())
|
|
st, err := store.New(":memory:", zap.NewNop())
|
|
if err != nil {
|
|
t.Fatalf("store: %v", err)
|
|
}
|
|
t.Cleanup(func() { _ = st.Close() })
|
|
|
|
n := node.New("test-node", rtr, st, 1, io.Discard, zap.NewNop(), set)
|
|
|
|
// Check initial capabilities
|
|
resp, err := n.OnCommandRequest(context.Background(), nil, &iop.NodeCommandRequest{
|
|
RequestId: "cmd-1",
|
|
Type: iop.NodeCommandType_NODE_COMMAND_TYPE_CAPABILITIES,
|
|
Adapter: "openai",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("OnCommandRequest: %v", err)
|
|
}
|
|
if resp.GetResult()["capacity"] != "2" {
|
|
t.Fatalf("expected capacity 2, got %s", resp.GetResult()["capacity"])
|
|
}
|
|
|
|
// Refresh payload with new capacity
|
|
refreshPayload := &iop.NodeConfigPayload{
|
|
Adapters: []*iop.AdapterConfig{
|
|
{
|
|
Name: "openai",
|
|
Type: "openai_compat",
|
|
Enabled: true,
|
|
Config: &iop.AdapterConfig_OpenaiCompat{
|
|
OpenaiCompat: &iop.OpenAICompatAdapterConfig{
|
|
Provider: "openai",
|
|
Endpoint: "http://localhost:8080",
|
|
Capacity: 5,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
refreshResp, err := n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
|
|
RequestId: "refresh-1",
|
|
Config: refreshPayload,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("OnConfigRefresh: %v", err)
|
|
}
|
|
if refreshResp.GetStatus() != iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_APPLIED {
|
|
t.Fatalf("expected applied status, got %v", refreshResp.GetStatus())
|
|
}
|
|
|
|
// Check updated capabilities
|
|
resp2, err := n.OnCommandRequest(context.Background(), nil, &iop.NodeCommandRequest{
|
|
RequestId: "cmd-2",
|
|
Type: iop.NodeCommandType_NODE_COMMAND_TYPE_CAPABILITIES,
|
|
Adapter: "openai",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("OnCommandRequest: %v", err)
|
|
}
|
|
if resp2.GetResult()["capacity"] != "5" {
|
|
t.Fatalf("expected capacity 5, got %s", resp2.GetResult()["capacity"])
|
|
}
|
|
}
|
|
|
|
func TestNodeConfigRefreshUsesUpdatedOpenAICompatEndpoint(t *testing.T) {
|
|
var oldCalls int32
|
|
var newCalls int32
|
|
|
|
serverOld := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
atomic.AddInt32(&oldCalls, 1)
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_, _ = w.Write([]byte(`{"data": [{"id": "gpt-4o"}]}`))
|
|
}))
|
|
defer serverOld.Close()
|
|
|
|
serverNew := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
atomic.AddInt32(&newCalls, 1)
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_, _ = w.Write([]byte(`{"data": [{"id": "gpt-4o"}]}`))
|
|
}))
|
|
defer serverNew.Close()
|
|
|
|
initialPayload := &iop.NodeConfigPayload{
|
|
Adapters: []*iop.AdapterConfig{
|
|
{
|
|
Name: "openai",
|
|
Type: "openai_compat",
|
|
Enabled: true,
|
|
Config: &iop.AdapterConfig_OpenaiCompat{
|
|
OpenaiCompat: &iop.OpenAICompatAdapterConfig{
|
|
Provider: "openai",
|
|
Endpoint: serverOld.URL,
|
|
Capacity: 2,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
set, err := adapters.BuildConfigSet(initialPayload, zap.NewNop())
|
|
if err != nil {
|
|
t.Fatalf("BuildConfigSet: %v", err)
|
|
}
|
|
|
|
rtr := router.New(set.Registry, zap.NewNop())
|
|
st, err := store.New(":memory:", zap.NewNop())
|
|
if err != nil {
|
|
t.Fatalf("store: %v", err)
|
|
}
|
|
t.Cleanup(func() { _ = st.Close() })
|
|
|
|
n := node.New("test-node", rtr, st, 1, io.Discard, zap.NewNop(), set)
|
|
|
|
// Call capabilities command -> triggers probe (requests serverOld)
|
|
_, err = n.OnCommandRequest(context.Background(), nil, &iop.NodeCommandRequest{
|
|
RequestId: "cmd-1",
|
|
Type: iop.NodeCommandType_NODE_COMMAND_TYPE_CAPABILITIES,
|
|
Adapter: "openai",
|
|
Target: "gpt-4o",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("OnCommandRequest: %v", err)
|
|
}
|
|
|
|
if atomic.LoadInt32(&oldCalls) != 2 {
|
|
t.Fatalf("expected 2 calls to serverOld, got %d", oldCalls)
|
|
}
|
|
if atomic.LoadInt32(&newCalls) != 0 {
|
|
t.Fatalf("expected 0 calls to serverNew, got %d", newCalls)
|
|
}
|
|
|
|
// Refresh with new endpoint
|
|
refreshPayload := &iop.NodeConfigPayload{
|
|
Adapters: []*iop.AdapterConfig{
|
|
{
|
|
Name: "openai",
|
|
Type: "openai_compat",
|
|
Enabled: true,
|
|
Config: &iop.AdapterConfig_OpenaiCompat{
|
|
OpenaiCompat: &iop.OpenAICompatAdapterConfig{
|
|
Provider: "openai",
|
|
Endpoint: serverNew.URL,
|
|
Capacity: 2,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
refreshResp, err := n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
|
|
RequestId: "refresh-1",
|
|
Config: refreshPayload,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("OnConfigRefresh: %v", err)
|
|
}
|
|
if refreshResp.GetStatus() != iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_APPLIED {
|
|
t.Fatalf("expected applied status, got %v", refreshResp.GetStatus())
|
|
}
|
|
|
|
// Call capabilities command again -> triggers probe (requests serverNew)
|
|
_, err = n.OnCommandRequest(context.Background(), nil, &iop.NodeCommandRequest{
|
|
RequestId: "cmd-2",
|
|
Type: iop.NodeCommandType_NODE_COMMAND_TYPE_CAPABILITIES,
|
|
Adapter: "openai",
|
|
Target: "gpt-4o",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("OnCommandRequest 2: %v", err)
|
|
}
|
|
|
|
if atomic.LoadInt32(&oldCalls) != 2 {
|
|
t.Fatalf("expected still 2 calls to serverOld, got %d", oldCalls)
|
|
}
|
|
if atomic.LoadInt32(&newCalls) != 2 {
|
|
t.Fatalf("expected 2 calls to serverNew, got %d", newCalls)
|
|
}
|
|
}
|
|
|
|
func TestNodeConfigRefreshUsesUpdatedOpenAICompatHeaders(t *testing.T) {
|
|
var observedHeader string
|
|
|
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
observedHeader = r.Header.Get("X-Test-Key")
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_, _ = w.Write([]byte(`{"data": [{"id": "gpt-4o"}]}`))
|
|
}))
|
|
defer server.Close()
|
|
|
|
initialPayload := &iop.NodeConfigPayload{
|
|
Adapters: []*iop.AdapterConfig{
|
|
{
|
|
Name: "openai",
|
|
Type: "openai_compat",
|
|
Enabled: true,
|
|
Config: &iop.AdapterConfig_OpenaiCompat{
|
|
OpenaiCompat: &iop.OpenAICompatAdapterConfig{
|
|
Provider: "openai",
|
|
Endpoint: server.URL,
|
|
Headers: map[string]string{"X-Test-Key": "initial-value"},
|
|
Capacity: 2,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
set, err := adapters.BuildConfigSet(initialPayload, zap.NewNop())
|
|
if err != nil {
|
|
t.Fatalf("BuildConfigSet: %v", err)
|
|
}
|
|
|
|
rtr := router.New(set.Registry, zap.NewNop())
|
|
st, err := store.New(":memory:", zap.NewNop())
|
|
if err != nil {
|
|
t.Fatalf("store: %v", err)
|
|
}
|
|
t.Cleanup(func() { _ = st.Close() })
|
|
|
|
n := node.New("test-node", rtr, st, 1, io.Discard, zap.NewNop(), set)
|
|
|
|
// Trigger request -> should see initial header
|
|
_, _ = n.OnCommandRequest(context.Background(), nil, &iop.NodeCommandRequest{
|
|
RequestId: "cmd-1",
|
|
Type: iop.NodeCommandType_NODE_COMMAND_TYPE_CAPABILITIES,
|
|
Adapter: "openai",
|
|
Target: "gpt-4o",
|
|
})
|
|
if observedHeader != "initial-value" {
|
|
t.Fatalf("expected header 'initial-value', got %q", observedHeader)
|
|
}
|
|
|
|
// Refresh with new header
|
|
refreshPayload := &iop.NodeConfigPayload{
|
|
Adapters: []*iop.AdapterConfig{
|
|
{
|
|
Name: "openai",
|
|
Type: "openai_compat",
|
|
Enabled: true,
|
|
Config: &iop.AdapterConfig_OpenaiCompat{
|
|
OpenaiCompat: &iop.OpenAICompatAdapterConfig{
|
|
Provider: "openai",
|
|
Endpoint: server.URL,
|
|
Headers: map[string]string{"X-Test-Key": "updated-value"},
|
|
Capacity: 2,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
_, _ = n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
|
|
RequestId: "refresh-1",
|
|
Config: refreshPayload,
|
|
})
|
|
|
|
// Trigger request again -> should see updated header
|
|
_, _ = n.OnCommandRequest(context.Background(), nil, &iop.NodeCommandRequest{
|
|
RequestId: "cmd-2",
|
|
Type: iop.NodeCommandType_NODE_COMMAND_TYPE_CAPABILITIES,
|
|
Adapter: "openai",
|
|
Target: "gpt-4o",
|
|
})
|
|
if observedHeader != "updated-value" {
|
|
t.Fatalf("expected header 'updated-value', got %q", observedHeader)
|
|
}
|
|
}
|
|
|
|
func TestNodeConfigRefreshUpdatesExistingAdapterGateCapacity(t *testing.T) {
|
|
// mock HTTP server that can be blocked or unblocked
|
|
blockChan := make(chan struct{})
|
|
defer func() {
|
|
select {
|
|
case <-blockChan:
|
|
default:
|
|
close(blockChan)
|
|
}
|
|
}()
|
|
|
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
if strings.Contains(r.URL.Path, "/v1/models") {
|
|
_, _ = w.Write([]byte(`{"data": [{"id": "gpt-4o"}]}`))
|
|
return
|
|
}
|
|
// Block completion requests until blockChan is closed or read
|
|
<-blockChan
|
|
// Write a dummy SSE chat completion response
|
|
_, _ = w.Write([]byte("data: {\"choices\": [{\"delta\": {\"content\": \"hello\"}}]}\n\ndata: [DONE]\n\n"))
|
|
}))
|
|
defer server.Close()
|
|
|
|
initialPayload := &iop.NodeConfigPayload{
|
|
Adapters: []*iop.AdapterConfig{
|
|
{
|
|
Name: "openai",
|
|
Type: "openai_compat",
|
|
Enabled: true,
|
|
Config: &iop.AdapterConfig_OpenaiCompat{
|
|
OpenaiCompat: &iop.OpenAICompatAdapterConfig{
|
|
Provider: "openai",
|
|
Endpoint: server.URL,
|
|
Capacity: 1,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
set, err := adapters.BuildConfigSet(initialPayload, zap.NewNop())
|
|
if err != nil {
|
|
t.Fatalf("BuildConfigSet: %v", err)
|
|
}
|
|
|
|
rtr := router.New(set.Registry, zap.NewNop())
|
|
st, err := store.New(":memory:", zap.NewNop())
|
|
if err != nil {
|
|
t.Fatalf("store: %v", err)
|
|
}
|
|
t.Cleanup(func() { _ = st.Close() })
|
|
|
|
n := node.New("test-node", rtr, st, 0, io.Discard, zap.NewNop(), set)
|
|
|
|
inputMap := map[string]interface{}{
|
|
"prompt": "hello",
|
|
}
|
|
inputStruct, err := structpb.NewStruct(inputMap)
|
|
if err != nil {
|
|
t.Fatalf("failed to create input struct: %v", err)
|
|
}
|
|
|
|
// First request: should run and block inside the mock server
|
|
errChan1 := make(chan error, 1)
|
|
go func() {
|
|
errChan1 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
|
RunId: "run-1",
|
|
Adapter: "openai",
|
|
Target: "gpt-4o",
|
|
Background: false,
|
|
Input: inputStruct,
|
|
})
|
|
}()
|
|
|
|
// Wait a bit to ensure the first request is indeed running and blocked
|
|
time.Sleep(50 * time.Millisecond)
|
|
|
|
// Second request: should fail because capacity is 1
|
|
err2 := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
|
RunId: "run-2",
|
|
Adapter: "openai",
|
|
Target: "gpt-4o",
|
|
Background: false,
|
|
Input: inputStruct,
|
|
})
|
|
if err2 == nil {
|
|
t.Fatal("expected second request to fail due to concurrency limit")
|
|
}
|
|
if !errors.Is(err2, node.ErrConcurrencyLimitExceeded) {
|
|
t.Fatalf("expected ErrConcurrencyLimitExceeded, got: %v", err2)
|
|
}
|
|
|
|
// Refresh config to increase capacity to 2
|
|
refreshPayload := &iop.NodeConfigPayload{
|
|
Adapters: []*iop.AdapterConfig{
|
|
{
|
|
Name: "openai",
|
|
Type: "openai_compat",
|
|
Enabled: true,
|
|
Config: &iop.AdapterConfig_OpenaiCompat{
|
|
OpenaiCompat: &iop.OpenAICompatAdapterConfig{
|
|
Provider: "openai",
|
|
Endpoint: server.URL,
|
|
Capacity: 2,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
refreshResp, refreshErr := n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
|
|
RequestId: "refresh-1",
|
|
Config: refreshPayload,
|
|
})
|
|
if refreshErr != nil {
|
|
t.Fatalf("OnConfigRefresh failed: %v", refreshErr)
|
|
}
|
|
if refreshResp.Status != iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_APPLIED {
|
|
t.Fatalf("expected refresh status APPLIED, got %v", refreshResp.Status)
|
|
}
|
|
|
|
// Third request: should now succeed (and block) instead of getting rejected
|
|
errChan3 := make(chan error, 1)
|
|
go func() {
|
|
errChan3 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
|
RunId: "run-3",
|
|
Adapter: "openai",
|
|
Target: "gpt-4o",
|
|
Background: false,
|
|
Input: inputStruct,
|
|
})
|
|
}()
|
|
|
|
// Wait a bit and check that third request hasn't errored immediately (it should be blocking)
|
|
time.Sleep(50 * time.Millisecond)
|
|
select {
|
|
case err3 := <-errChan3:
|
|
t.Fatalf("expected third request to block, but it failed immediately: %v", err3)
|
|
default:
|
|
// It is blocking as expected!
|
|
}
|
|
|
|
// Unblock mock server for all requests
|
|
close(blockChan)
|
|
|
|
// Wait for run-1 and run-3 to finish successfully
|
|
if err1 := <-errChan1; err1 != nil {
|
|
t.Fatalf("run-1 failed: %v", err1)
|
|
}
|
|
if err3 := <-errChan3; err3 != nil {
|
|
t.Fatalf("run-3 failed: %v", err3)
|
|
}
|
|
}
|
|
|
|
func TestNodeConfigRefreshDecreasesExistingAdapterGateCapacity(t *testing.T) {
|
|
blockChan := make(chan struct{})
|
|
defer func() {
|
|
select {
|
|
case <-blockChan:
|
|
default:
|
|
close(blockChan)
|
|
}
|
|
}()
|
|
|
|
startedChan := make(chan struct{}, 1)
|
|
|
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
if strings.Contains(r.URL.Path, "/v1/models") {
|
|
_, _ = w.Write([]byte(`{"data": [{"id": "gpt-4o"}]}`))
|
|
return
|
|
}
|
|
select {
|
|
case startedChan <- struct{}{}:
|
|
default:
|
|
}
|
|
<-blockChan
|
|
_, _ = w.Write([]byte("data: {\"choices\": [{\"delta\": {\"content\": \"hello\"}}]}\n\ndata: [DONE]\n\n"))
|
|
}))
|
|
defer server.Close()
|
|
|
|
initialPayload := &iop.NodeConfigPayload{
|
|
Adapters: []*iop.AdapterConfig{
|
|
{
|
|
Name: "openai",
|
|
Type: "openai_compat",
|
|
Enabled: true,
|
|
Config: &iop.AdapterConfig_OpenaiCompat{
|
|
OpenaiCompat: &iop.OpenAICompatAdapterConfig{
|
|
Provider: "openai",
|
|
Endpoint: server.URL,
|
|
Capacity: 2,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
set, err := adapters.BuildConfigSet(initialPayload, zap.NewNop())
|
|
if err != nil {
|
|
t.Fatalf("BuildConfigSet: %v", err)
|
|
}
|
|
|
|
rtr := router.New(set.Registry, zap.NewNop())
|
|
st, err := store.New(":memory:", zap.NewNop())
|
|
if err != nil {
|
|
t.Fatalf("store: %v", err)
|
|
}
|
|
t.Cleanup(func() { _ = st.Close() })
|
|
|
|
n := node.New("test-node", rtr, st, 0, io.Discard, zap.NewNop(), set)
|
|
|
|
inputMap := map[string]interface{}{
|
|
"prompt": "hello",
|
|
}
|
|
inputStruct, err := structpb.NewStruct(inputMap)
|
|
if err != nil {
|
|
t.Fatalf("failed to create input struct: %v", err)
|
|
}
|
|
|
|
errChan1 := make(chan error, 1)
|
|
go func() {
|
|
errChan1 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
|
|
RunId: "run-1",
|
|
Adapter: "openai",
|
|
Target: "gpt-4o",
|
|
Background: false,
|
|
Input: inputStruct,
|
|
})
|
|
}()
|
|
|
|
<-startedChan
|
|
|
|
refreshPayload := &iop.NodeConfigPayload{
|
|
Adapters: []*iop.AdapterConfig{
|
|
{
|
|
Name: "openai",
|
|
Type: "openai_compat",
|
|
Enabled: true,
|
|
Config: &iop.AdapterConfig_OpenaiCompat{
|
|
OpenaiCompat: &iop.OpenAICompatAdapterConfig{
|
|
Provider: "openai",
|
|
Endpoint: server.URL,
|
|
Capacity: 1,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
refreshResp, refreshErr := n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
|
|
RequestId: "refresh-1",
|
|
Config: refreshPayload,
|
|
})
|
|
if refreshErr != nil {
|
|
t.Fatalf("OnConfigRefresh failed: %v", refreshErr)
|
|
}
|
|
if refreshResp.Status != iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_APPLIED {
|
|
t.Fatalf("expected refresh status APPLIED, got %v", refreshResp.Status)
|
|
}
|
|
|
|
ctx2, cancel2 := context.WithTimeout(context.Background(), 200*time.Millisecond)
|
|
defer cancel2()
|
|
err2 := n.OnRunRequest(ctx2, &transport.Session{}, &iop.RunRequest{
|
|
RunId: "run-2",
|
|
Adapter: "openai",
|
|
Target: "gpt-4o",
|
|
Background: false,
|
|
Input: inputStruct,
|
|
})
|
|
if err2 == nil {
|
|
t.Fatal("expected second request to fail due to concurrency limit")
|
|
}
|
|
if !errors.Is(err2, node.ErrConcurrencyLimitExceeded) {
|
|
t.Fatalf("expected ErrConcurrencyLimitExceeded, got: %v", err2)
|
|
}
|
|
|
|
close(blockChan)
|
|
|
|
if err1 := <-errChan1; err1 != nil {
|
|
t.Fatalf("run-1 failed: %v", err1)
|
|
}
|
|
}
|
|
|
|
// --- runtime concurrency metadata tests (admission is no-op) ---
|
|
|
|
func TestConfigRefreshRuntimeConcurrencyDoesNotAffectAdmission(t *testing.T) {
|
|
// Adapter is unlimited (MaxConcurrency=0), so the adapter gate imposes no limit.
|
|
sa := newQueuedSlowAdapter("slow", 0, 0, 0)
|
|
rtr := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Provider{"slow": sa}}
|
|
st, err := store.New(":memory:", zap.NewNop())
|
|
if err != nil {
|
|
t.Fatalf("store: %v", err)
|
|
}
|
|
t.Cleanup(func() { _ = st.Close() })
|
|
|
|
// Global concurrency is set to 1, but it must NOT affect admission.
|
|
// All concurrent runs should succeed because the adapter gate is unlimited.
|
|
n := node.New("test-node", rtr, st, 1, io.Discard, zap.NewNop(), nil)
|
|
|
|
errCh1 := make(chan error, 1)
|
|
go func() {
|
|
errCh1 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{RunId: "run-1", Adapter: "slow"})
|
|
}()
|
|
waitStarted(t, sa, "run-1")
|
|
|
|
// Second run should NOT be rejected: global concurrency is no longer used for admission.
|
|
errCh2 := make(chan error, 1)
|
|
go func() {
|
|
errCh2 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{RunId: "run-2", Adapter: "slow"})
|
|
}()
|
|
waitStarted(t, sa, "run-2")
|
|
|
|
// Config refresh with runtime concurrency must succeed and store metadata.
|
|
resp, refreshErr := n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
|
|
RequestId: "refresh-conc",
|
|
Config: &iop.NodeConfigPayload{Runtime: &iop.NodeRuntimeConfig{Concurrency: 2}},
|
|
})
|
|
if refreshErr != nil {
|
|
t.Fatalf("OnConfigRefresh failed: %v", refreshErr)
|
|
}
|
|
if resp.Status != iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_APPLIED {
|
|
t.Fatalf("expected refresh status APPLIED, got %v", resp.Status)
|
|
}
|
|
|
|
// Third run must also succeed: runtime concurrency is metadata only.
|
|
errCh3 := make(chan error, 1)
|
|
go func() {
|
|
errCh3 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{RunId: "run-3", Adapter: "slow"})
|
|
}()
|
|
waitStarted(t, sa, "run-3")
|
|
|
|
sa.releaseRun("run-1")
|
|
sa.releaseRun("run-2")
|
|
sa.releaseRun("run-3")
|
|
if err := <-errCh1; err != nil {
|
|
t.Fatalf("run-1 failed: %v", err)
|
|
}
|
|
if err := <-errCh2; err != nil {
|
|
t.Fatalf("run-2 failed: %v", err)
|
|
}
|
|
if err := <-errCh3; err != nil {
|
|
t.Fatalf("run-3 failed: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestConfigRefreshConcurrencyDecreaseDoesNotAffectAdmission(t *testing.T) {
|
|
sa := newQueuedSlowAdapter("slow", 0, 0, 0)
|
|
rtr := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Provider{"slow": sa}}
|
|
st, err := store.New(":memory:", zap.NewNop())
|
|
if err != nil {
|
|
t.Fatalf("store: %v", err)
|
|
}
|
|
t.Cleanup(func() { _ = st.Close() })
|
|
|
|
n := node.New("test-node", rtr, st, 2, io.Discard, zap.NewNop(), nil)
|
|
|
|
errCh1 := make(chan error, 1)
|
|
go func() {
|
|
errCh1 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{RunId: "run-1", Adapter: "slow"})
|
|
}()
|
|
waitStarted(t, sa, "run-1")
|
|
|
|
errCh2 := make(chan error, 1)
|
|
go func() {
|
|
errCh2 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{RunId: "run-2", Adapter: "slow"})
|
|
}()
|
|
waitStarted(t, sa, "run-2")
|
|
|
|
// Config refresh decreases runtime concurrency from 2 to 1.
|
|
// This must NOT affect admission: the adapter gate is unlimited.
|
|
resp, refreshErr := n.OnConfigRefresh(context.Background(), nil, &iop.NodeConfigRefreshRequest{
|
|
RequestId: "refresh-dec",
|
|
Config: &iop.NodeConfigPayload{Runtime: &iop.NodeRuntimeConfig{Concurrency: 1}},
|
|
})
|
|
if refreshErr != nil {
|
|
t.Fatalf("OnConfigRefresh failed: %v", refreshErr)
|
|
}
|
|
if resp.Status != iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_APPLIED {
|
|
t.Fatalf("expected refresh status APPLIED, got %v", resp.Status)
|
|
}
|
|
|
|
// Third run must be admitted: global concurrency is no longer used for admission.
|
|
errCh3 := make(chan error, 1)
|
|
go func() {
|
|
errCh3 <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{RunId: "run-3", Adapter: "slow"})
|
|
}()
|
|
waitStarted(t, sa, "run-3")
|
|
|
|
// All three runs complete normally.
|
|
sa.releaseRun("run-1")
|
|
sa.releaseRun("run-2")
|
|
sa.releaseRun("run-3")
|
|
if err := <-errCh1; err != nil {
|
|
t.Fatalf("run-1 should complete normally, got %v", err)
|
|
}
|
|
if err := <-errCh2; err != nil {
|
|
t.Fatalf("run-2 should complete normally, got %v", err)
|
|
}
|
|
if err := <-errCh3; err != nil {
|
|
t.Fatalf("run-3 should complete normally, got %v", err)
|
|
}
|
|
}
|