toki
fef1f7a9dc
Node의 provider progress 기반 stall timeout, watchdog fencing과 bounded health probe evidence를 실행 경로에 반영한다. Edge-Node 계약과 구현 스펙, 테스트 및 Milestone 완료 evidence를 현재 상태와 맞춘다.
108 lines
2.6 KiB
Go
108 lines
2.6 KiB
Go
package router
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sync"
|
|
|
|
"go.uber.org/zap"
|
|
|
|
runtime "iop/packages/go/execution"
|
|
)
|
|
|
|
type MutableRouter interface {
|
|
runtime.Router
|
|
SetRegistry(*runtime.Registry)
|
|
}
|
|
|
|
type defaultRouter struct {
|
|
mu sync.RWMutex
|
|
registry *runtime.Registry
|
|
logger *zap.Logger
|
|
}
|
|
|
|
// New returns a MutableRouter that resolves requests to registered adapters.
|
|
func New(reg *runtime.Registry, logger *zap.Logger) MutableRouter {
|
|
return &defaultRouter{registry: reg, logger: logger}
|
|
}
|
|
|
|
func (r *defaultRouter) SetRegistry(reg *runtime.Registry) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
r.registry = reg
|
|
}
|
|
|
|
func (r *defaultRouter) resolveWithRegistry(req runtime.RunRequest, reg *runtime.Registry) (runtime.ExecutionSpec, error) {
|
|
adapterName := req.Adapter
|
|
if adapterName == "" {
|
|
return runtime.ExecutionSpec{}, fmt.Errorf("router: adapter is required")
|
|
}
|
|
|
|
if _, err := reg.Lookup(adapterName); err != nil {
|
|
return runtime.ExecutionSpec{}, fmt.Errorf("router: %w", err)
|
|
}
|
|
|
|
spec := runtime.ExecutionSpec{
|
|
RunID: req.RunID,
|
|
Adapter: adapterName,
|
|
Target: req.Target,
|
|
SessionID: req.SessionID,
|
|
Background: req.Background,
|
|
Policy: req.Policy,
|
|
Input: req.Input,
|
|
TimeoutSec: req.TimeoutSec,
|
|
Metadata: req.Metadata,
|
|
ResponseStallTimeoutMS: req.ResponseStallTimeoutMS,
|
|
}
|
|
|
|
r.logger.Debug("resolved execution spec",
|
|
zap.String("run_id", req.RunID),
|
|
zap.String("adapter", spec.Adapter),
|
|
zap.String("target", spec.Target),
|
|
)
|
|
return spec, nil
|
|
}
|
|
|
|
func (r *defaultRouter) Resolve(_ context.Context, req runtime.RunRequest) (runtime.ExecutionSpec, error) {
|
|
r.mu.RLock()
|
|
reg := r.registry
|
|
r.mu.RUnlock()
|
|
|
|
return r.resolveWithRegistry(req, reg)
|
|
}
|
|
|
|
func (r *defaultRouter) ResolveAdapter(ctx context.Context, req runtime.RunRequest) (runtime.ExecutionSpec, runtime.Provider, error) {
|
|
r.mu.RLock()
|
|
reg := r.registry
|
|
r.mu.RUnlock()
|
|
|
|
spec, err := r.resolveWithRegistry(req, reg)
|
|
if err != nil {
|
|
return runtime.ExecutionSpec{}, nil, err
|
|
}
|
|
adapter, err := reg.Lookup(spec.Adapter)
|
|
if err != nil {
|
|
return runtime.ExecutionSpec{}, nil, fmt.Errorf("router: %w", err)
|
|
}
|
|
return spec, adapter, nil
|
|
}
|
|
|
|
func (r *defaultRouter) LookupAdapter(adapterName string) (runtime.Provider, error) {
|
|
r.mu.RLock()
|
|
reg := r.registry
|
|
r.mu.RUnlock()
|
|
|
|
return reg.Lookup(adapterName)
|
|
}
|
|
|
|
func (r *defaultRouter) GetAdapter(adapterName string) (runtime.Provider, bool) {
|
|
r.mu.RLock()
|
|
reg := r.registry
|
|
r.mu.RUnlock()
|
|
|
|
a, err := reg.Lookup(adapterName)
|
|
if err != nil {
|
|
return nil, false
|
|
}
|
|
return a, true
|
|
}
|