toki
dc9a9a8c59
승인된 execution preset을 Edge 조정 경계와 Node workspace/tool 실행 경계로 연결해 단일 요청 수명주기와 관측 계약을 일관되게 처리한다.
92 lines
3 KiB
Go
92 lines
3 KiB
Go
// Package node is the core IOP Node service. It implements
|
|
// transport.Handler and orchestrates routing → adapter execution.
|
|
package node
|
|
|
|
import (
|
|
"io"
|
|
"os"
|
|
"sync"
|
|
|
|
"go.uber.org/zap"
|
|
|
|
"iop/apps/node/internal/adapters"
|
|
"iop/apps/node/internal/store"
|
|
"iop/apps/node/internal/workspace"
|
|
"iop/packages/go/credentiallease"
|
|
runtime "iop/packages/go/execution"
|
|
)
|
|
|
|
// Node implements transport.Handler and coordinates the full execution pipeline.
|
|
type Node struct {
|
|
nodeID string
|
|
router runtime.Router
|
|
store *store.Store
|
|
runs *runManager
|
|
globalGate *fifoGate // node-wide concurrency safety guard across all adapters (retained for compatibility; not used for admission)
|
|
adapterGatesMu sync.Mutex
|
|
adapterGates map[string]*fifoGate // per adapter-key concurrency safety guard
|
|
out io.Writer
|
|
logger *zap.Logger
|
|
currentConfigSet *adapters.ConfigSet
|
|
configSetMu sync.RWMutex
|
|
credentialConsumer *credentiallease.Consumer
|
|
workspaceMu sync.RWMutex
|
|
workspaceRuntime *workspace.Runtime
|
|
watchdogClock attemptClock
|
|
|
|
// liveness is the bounded stall-observability observer. Production Nodes
|
|
// share one process-global collector set; tests inject an isolated registry
|
|
// via the test-only constructor path in liveness_observability.go.
|
|
liveness *nodeLivenessObserver
|
|
}
|
|
|
|
func (n *Node) SetCredentialConsumer(consumer *credentiallease.Consumer) {
|
|
n.credentialConsumer = consumer
|
|
}
|
|
|
|
// SetWorkspaceRuntime installs the Node-private workspace authority after the
|
|
// bootstrap catalog has been fully validated. It intentionally does not change
|
|
// the constructor so existing provider-only callers remain compatible.
|
|
func (n *Node) SetWorkspaceRuntime(workspaceRuntime *workspace.Runtime) {
|
|
n.workspaceMu.Lock()
|
|
n.workspaceRuntime = workspaceRuntime
|
|
n.workspaceMu.Unlock()
|
|
}
|
|
|
|
func (n *Node) getWorkspaceRuntime() *workspace.Runtime {
|
|
n.workspaceMu.RLock()
|
|
defer n.workspaceMu.RUnlock()
|
|
return n.workspaceRuntime
|
|
}
|
|
|
|
// New creates a Node. It satisfies transport.Handler.
|
|
// globalConcurrency is retained as a compatibility argument but is no longer
|
|
// used for admission. Node-wide concurrency limits have been removed; per-
|
|
// adapter MaxConcurrency from Capabilities is the sole admission gate.
|
|
// 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,
|
|
initialConfigSet *adapters.ConfigSet,
|
|
) *Node {
|
|
if out == nil {
|
|
out = os.Stdout
|
|
}
|
|
return &Node{
|
|
nodeID: nodeID,
|
|
router: router,
|
|
store: st,
|
|
runs: newRunManager(),
|
|
globalGate: newFifoGate(globalConcurrency),
|
|
adapterGates: make(map[string]*fifoGate),
|
|
out: out,
|
|
logger: logger,
|
|
currentConfigSet: initialConfigSet,
|
|
watchdogClock: realAttemptClock{},
|
|
liveness: newProductionNodeLivenessObserver(logger),
|
|
}
|
|
}
|