From 173d2586e88b5d489dcb9351a6c9fae3a37fc5d3 Mon Sep 17 00:00:00 2001 From: toki Date: Sat, 2 May 2026 17:54:58 +0900 Subject: [PATCH] =?UTF-8?q?=EB=A6=AC=ED=8C=A9=ED=84=B0:=20node=20=EC=84=9C?= =?UTF-8?q?=EB=B2=84=20=EC=BD=94=EB=93=9C=20=EC=A0=9C=EA=B1=B0=20=EB=B0=8F?= =?UTF-8?q?=20=ED=81=B4=EB=9D=BC=EC=9D=B4=EC=96=B8=ED=8A=B8=20=EA=B5=AC?= =?UTF-8?q?=EC=A1=B0=EB=A1=9C=20=EC=A0=84=ED=99=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - transport/{server,session,frame,codec,tls}.go 삭제 - packages/protocol/protocol.go 삭제 (JSON 기반 메시지 타입) - bootstrap/module.go에서 transport.Server 와이어링 제거 - TransportConf.Listen → EdgeAddr (node는 서버가 아닌 클라이언트) - configs/node.yaml transport.listen → transport.edge_addr Co-Authored-By: Claude Sonnet 4.6 --- apps/node/cmd/iop-node/main.go | 2 +- apps/node/internal/bootstrap/module.go | 21 +-- apps/node/internal/transport/codec.go | 32 ---- apps/node/internal/transport/frame.go | 63 ------- apps/node/internal/transport/server.go | 135 -------------- apps/node/internal/transport/session.go | 228 ------------------------ apps/node/internal/transport/tls.go | 17 -- configs/node.yaml | 2 +- packages/config/config.go | 4 +- packages/protocol/protocol.go | 92 ---------- 10 files changed, 6 insertions(+), 590 deletions(-) delete mode 100644 apps/node/internal/transport/codec.go delete mode 100644 apps/node/internal/transport/frame.go delete mode 100644 apps/node/internal/transport/server.go delete mode 100644 apps/node/internal/transport/session.go delete mode 100644 apps/node/internal/transport/tls.go delete mode 100644 packages/protocol/protocol.go diff --git a/apps/node/cmd/iop-node/main.go b/apps/node/cmd/iop-node/main.go index ae06bc8..daca3f6 100644 --- a/apps/node/cmd/iop-node/main.go +++ b/apps/node/cmd/iop-node/main.go @@ -36,7 +36,7 @@ func rootCmd() *cobra.Command { func serveCmd() *cobra.Command { return &cobra.Command{ Use: "serve", - Short: "Start the IOP node server", + Short: "Start the IOP node (connects to edge)", RunE: func(cmd *cobra.Command, _ []string) error { cfg, err := config.Load(cfgFile) if err != nil { diff --git a/apps/node/internal/bootstrap/module.go b/apps/node/internal/bootstrap/module.go index 16c7d6d..229f355 100644 --- a/apps/node/internal/bootstrap/module.go +++ b/apps/node/internal/bootstrap/module.go @@ -16,7 +16,6 @@ import ( "iop/apps/node/internal/router" "iop/apps/node/internal/runtime" "iop/apps/node/internal/store" - "iop/apps/node/internal/transport" "iop/packages/config" "iop/packages/observability" ) @@ -66,25 +65,11 @@ func Module(cfg *config.NodeConfig) fx.Option { ) *node.Node { return node.New(cfg, r, reg, st, logger) }, - - func(n *node.Node, logger *zap.Logger) *transport.Server { - return transport.NewServer(n, logger) - }, ), - fx.Invoke(func(lc fx.Lifecycle, srv *transport.Server, cfg *config.NodeConfig, logger *zap.Logger) { + fx.Invoke(func(lc fx.Lifecycle, cfg *config.NodeConfig, logger *zap.Logger) { lc.Append(fx.Hook{ OnStart: func(ctx context.Context) error { - tlsCfg, err := transport.TLSFromConfig(cfg.TLS) - if err != nil { - return err - } - if tlsCfg != nil { - srv.SetTLS(tlsCfg) - } - if err := srv.Start(ctx, cfg.Transport.Listen); err != nil { - return err - } go func() { if err := observability.ServeMetrics(cfg.Metrics.Port); err != nil { logger.Warn("metrics server exited", zap.Error(err)) @@ -92,9 +77,7 @@ func Module(cfg *config.NodeConfig) fx.Option { }() return nil }, - OnStop: func(_ context.Context) error { - return srv.Stop() - }, + OnStop: func(_ context.Context) error { return nil }, }) }), ) diff --git a/apps/node/internal/transport/codec.go b/apps/node/internal/transport/codec.go deleted file mode 100644 index 8376e77..0000000 --- a/apps/node/internal/transport/codec.go +++ /dev/null @@ -1,32 +0,0 @@ -package transport - -import ( - "encoding/json" - "fmt" - - "iop/packages/protocol" -) - -// encodeEnvelope marshals env to JSON bytes (the frame payload). -func encodeEnvelope(env *protocol.Envelope) ([]byte, error) { - return json.Marshal(env) -} - -// decodeEnvelope unmarshals frame payload bytes into an Envelope. -func decodeEnvelope(data []byte) (*protocol.Envelope, error) { - var env protocol.Envelope - if err := json.Unmarshal(data, &env); err != nil { - return nil, fmt.Errorf("decode envelope: %w", err) - } - return &env, nil -} - -// encodePayload marshals an inner message to JSON bytes for Envelope.Payload. -func encodePayload(v any) ([]byte, error) { - return json.Marshal(v) -} - -// decodePayload unmarshals Envelope.Payload into dst. -func decodePayload(payload []byte, dst any) error { - return json.Unmarshal(payload, dst) -} diff --git a/apps/node/internal/transport/frame.go b/apps/node/internal/transport/frame.go deleted file mode 100644 index 36d0f2b..0000000 --- a/apps/node/internal/transport/frame.go +++ /dev/null @@ -1,63 +0,0 @@ -// Package transport implements IOP's TCP framing layer. -// -// Wire format (same endianness as proto-socket): -// [4 bytes big-endian uint32: payload length] [payload bytes] -// -// Payload is a JSON-encoded Envelope (see codec.go). -// TODO: switch payload encoding to protobuf once proto/gen is available. -package transport - -import ( - "encoding/binary" - "fmt" - "io" -) - -const ( - frameLenBytes = 4 - maxFrameSize = 64 << 20 // 64 MiB — same limit as proto-socket -) - -// readFrame reads one length-prefixed frame from r. -func readFrame(r io.Reader) ([]byte, error) { - hdr := make([]byte, frameLenBytes) - if _, err := io.ReadFull(r, hdr); err != nil { - return nil, err - } - length := binary.BigEndian.Uint32(hdr) - if length == 0 { - return []byte{}, nil - } - if length > maxFrameSize { - return nil, fmt.Errorf("frame too large: %d bytes (max %d)", length, maxFrameSize) - } - buf := make([]byte, length) - if _, err := io.ReadFull(r, buf); err != nil { - return nil, err - } - return buf, nil -} - -// writeFrame writes payload as a length-prefixed frame to w. -func writeFrame(w io.Writer, payload []byte) error { - hdr := make([]byte, frameLenBytes) - binary.BigEndian.PutUint32(hdr, uint32(len(payload))) - if err := writeFull(w, hdr); err != nil { - return err - } - return writeFull(w, payload) -} - -func writeFull(w io.Writer, b []byte) error { - for len(b) > 0 { - n, err := w.Write(b) - if err != nil { - return err - } - if n == 0 { - return io.ErrShortWrite - } - b = b[n:] - } - return nil -} diff --git a/apps/node/internal/transport/server.go b/apps/node/internal/transport/server.go deleted file mode 100644 index a883677..0000000 --- a/apps/node/internal/transport/server.go +++ /dev/null @@ -1,135 +0,0 @@ -package transport - -import ( - "context" - "crypto/tls" - "errors" - "fmt" - "net" - "sync" - "sync/atomic" - - "go.uber.org/zap" -) - -// Server is the IOP TCP transport server. -type Server struct { - handler Handler - logger *zap.Logger - tlsConfig *tls.Config - - listener net.Listener - started atomic.Bool - mu sync.Mutex - sessions map[string]*Session - nextID atomic.Uint64 -} - -func NewServer(handler Handler, logger *zap.Logger) *Server { - return &Server{ - handler: handler, - logger: logger, - sessions: make(map[string]*Session), - } -} - -// SetTLS configures mTLS for the server. Must be called before Start. -func (s *Server) SetTLS(cfg *tls.Config) { - s.tlsConfig = cfg -} - -// Start begins accepting connections on addr. -func (s *Server) Start(ctx context.Context, addr string) error { - ln, err := net.Listen("tcp", addr) - if err != nil { - return fmt.Errorf("transport: listen %s: %w", addr, err) - } - if s.tlsConfig != nil { - ln = tls.NewListener(ln, s.tlsConfig) - } - - s.mu.Lock() - s.listener = ln - s.started.Store(true) - s.mu.Unlock() - - s.logger.Info("transport listening", - zap.String("addr", addr), - zap.Bool("tls", s.tlsConfig != nil), - ) - - go func() { - <-ctx.Done() - _ = s.Stop() - }() - go s.acceptLoop(ctx) - return nil -} - -// Stop closes the listener and all active sessions. -func (s *Server) Stop() error { - if !s.started.CompareAndSwap(true, false) { - return nil - } - - s.mu.Lock() - ln := s.listener - sessions := make([]*Session, 0, len(s.sessions)) - for _, sess := range s.sessions { - sessions = append(sessions, sess) - } - s.mu.Unlock() - - var err error - if ln != nil { - err = ln.Close() - } - for _, sess := range sessions { - sess.Close() - } - s.logger.Info("transport stopped") - return err -} - -// SessionCount returns the number of active sessions. -func (s *Server) SessionCount() int { - s.mu.Lock() - defer s.mu.Unlock() - return len(s.sessions) -} - -func (s *Server) acceptLoop(ctx context.Context) { - for { - conn, err := s.listener.Accept() - if err != nil { - if !s.started.Load() || errors.Is(err, net.ErrClosed) { - return - } - s.logger.Warn("accept error", zap.Error(err)) - continue - } - go s.handleConn(ctx, conn) - } -} - -func (s *Server) handleConn(ctx context.Context, conn net.Conn) { - id := fmt.Sprintf("sess-%d", s.nextID.Add(1)) - sess := newSession(id, conn, s.handler, s.logger) - - s.mu.Lock() - s.sessions[id] = sess - s.mu.Unlock() - - s.logger.Info("client connected", - zap.String("session_id", id), - zap.String("remote", conn.RemoteAddr().String()), - ) - - sess.readLoop(ctx) - - s.mu.Lock() - delete(s.sessions, id) - s.mu.Unlock() - - s.logger.Info("client disconnected", zap.String("session_id", id)) -} diff --git a/apps/node/internal/transport/session.go b/apps/node/internal/transport/session.go deleted file mode 100644 index 3a3837b..0000000 --- a/apps/node/internal/transport/session.go +++ /dev/null @@ -1,228 +0,0 @@ -package transport - -import ( - "context" - "fmt" - "net" - "sync" - "sync/atomic" - "time" - - "go.uber.org/zap" - - "iop/packages/protocol" -) - -const ( - heartbeatInterval = 30 * time.Second - heartbeatTimeout = 10 * time.Second -) - -// Handler handles decoded messages arriving on a Session. -type Handler interface { - OnRunRequest(ctx context.Context, sess *Session, req *protocol.RunRequest) error - OnCapabilityRequest(ctx context.Context, sess *Session) (*protocol.CapabilityResponse, error) - OnCancel(ctx context.Context, sess *Session, req *protocol.CancelRequest) error -} - -// Session represents one active TCP connection from a client. -type Session struct { - id string - conn net.Conn - handler Handler - logger *zap.Logger - - writeMu sync.Mutex - closed atomic.Bool - - cancelFns sync.Map // run_id → context.CancelFunc -} - -func newSession(id string, conn net.Conn, handler Handler, logger *zap.Logger) *Session { - return &Session{ - id: id, - conn: conn, - handler: handler, - logger: logger.With(zap.String("session_id", id)), - } -} - -// ID returns the session identifier. -func (s *Session) ID() string { return s.id } - -// Send encodes env as a frame and writes it to the connection. -func (s *Session) Send(env *protocol.Envelope) error { - if s.closed.Load() { - return fmt.Errorf("session %s: closed", s.id) - } - payload, err := encodeEnvelope(env) - if err != nil { - return err - } - s.writeMu.Lock() - defer s.writeMu.Unlock() - return writeFrame(s.conn, payload) -} - -// Close terminates the session. -func (s *Session) Close() { - if s.closed.CompareAndSwap(false, true) { - s.conn.Close() - s.logger.Info("session closed") - } -} - -// RegisterCancel stores a cancel function for a run, enabling CancelRequest handling. -func (s *Session) RegisterCancel(runID string, cancel context.CancelFunc) { - s.cancelFns.Store(runID, cancel) -} - -// DeregisterCancel removes a run's cancel function after completion. -func (s *Session) DeregisterCancel(runID string) { - s.cancelFns.Delete(runID) -} - -// CancelRun invokes the cancel function for a run if one is registered. -func (s *Session) CancelRun(runID string) { - if v, ok := s.cancelFns.Load(runID); ok { - v.(context.CancelFunc)() - } -} - -// readLoop drives the session until the connection closes or an error occurs. -func (s *Session) readLoop(ctx context.Context) { - defer s.Close() - - // Perform capability handshake: server sends capability request, - // client responds (optional — here we accept any client proactively). - go s.heartbeatLoop(ctx) - - for { - if s.closed.Load() { - return - } - - raw, err := readFrame(s.conn) - if err != nil { - if !s.closed.Load() { - s.logger.Debug("read frame error", zap.Error(err)) - } - return - } - if len(raw) == 0 { - continue - } - - env, err := decodeEnvelope(raw) - if err != nil { - s.logger.Warn("decode envelope failed", zap.Error(err)) - continue - } - - if err := s.dispatch(ctx, env); err != nil { - s.logger.Warn("dispatch error", zap.String("type", string(env.Type)), zap.Error(err)) - } - } -} - -func (s *Session) dispatch(ctx context.Context, env *protocol.Envelope) error { - switch env.Type { - case protocol.TypeHeartbeat: - return s.handleHeartbeat(env) - - case protocol.TypeRunRequest: - var req protocol.RunRequest - if err := decodePayload(env.Payload, &req); err != nil { - return s.sendError(env.RequestID, "DECODE_ERROR", err.Error()) - } - go func() { - if err := s.handler.OnRunRequest(ctx, s, &req); err != nil { - _ = s.sendError(env.RequestID, "RUN_ERROR", err.Error()) - } - }() - - case protocol.TypeCapabilityRequest: - resp, err := s.handler.OnCapabilityRequest(ctx, s) - if err != nil { - return s.sendError(env.RequestID, "CAPABILITY_ERROR", err.Error()) - } - return s.sendCapabilityResponse(env.RequestID, resp) - - case protocol.TypeCancelRequest: - var req protocol.CancelRequest - if err := decodePayload(env.Payload, &req); err != nil { - return s.sendError(env.RequestID, "DECODE_ERROR", err.Error()) - } - return s.handler.OnCancel(ctx, s, &req) - - default: - s.logger.Warn("unknown message type", zap.String("type", string(env.Type))) - } - return nil -} - -func (s *Session) handleHeartbeat(env *protocol.Envelope) error { - // Echo heartbeat back - resp, err := encodePayload(protocol.Heartbeat{Timestamp: time.Now().UnixNano()}) - if err != nil { - return err - } - return s.Send(&protocol.Envelope{ - ProtocolVersion: protocol.ProtocolVersion, - RequestID: env.RequestID, - Type: protocol.TypeHeartbeat, - Payload: resp, - }) -} - -func (s *Session) sendError(requestID, code, message string) error { - payload, err := encodePayload(protocol.ErrorMsg{Code: code, Message: message}) - if err != nil { - return err - } - return s.Send(&protocol.Envelope{ - ProtocolVersion: protocol.ProtocolVersion, - RequestID: requestID, - Type: protocol.TypeError, - Payload: payload, - }) -} - -func (s *Session) sendCapabilityResponse(requestID string, resp *protocol.CapabilityResponse) error { - payload, err := encodePayload(resp) - if err != nil { - return err - } - return s.Send(&protocol.Envelope{ - ProtocolVersion: protocol.ProtocolVersion, - RequestID: requestID, - Type: protocol.TypeCapabilityResponse, - Payload: payload, - }) -} - -// heartbeatLoop periodically sends heartbeat frames to detect dead connections. -func (s *Session) heartbeatLoop(ctx context.Context) { - ticker := time.NewTicker(heartbeatInterval) - defer ticker.Stop() - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - if s.closed.Load() { - return - } - payload, _ := encodePayload(protocol.Heartbeat{Timestamp: time.Now().UnixNano()}) - if err := s.Send(&protocol.Envelope{ - ProtocolVersion: protocol.ProtocolVersion, - Type: protocol.TypeHeartbeat, - Payload: payload, - }); err != nil { - s.logger.Debug("heartbeat send failed", zap.Error(err)) - s.Close() - return - } - } - } -} diff --git a/apps/node/internal/transport/tls.go b/apps/node/internal/transport/tls.go deleted file mode 100644 index 4d2eb32..0000000 --- a/apps/node/internal/transport/tls.go +++ /dev/null @@ -1,17 +0,0 @@ -package transport - -import ( - "crypto/tls" - - "iop/packages/auth" - "iop/packages/config" -) - -// TLSFromConfig builds a server *tls.Config from NodeConfig.TLS. -// Returns nil if TLS is disabled. -func TLSFromConfig(cfg config.TLSConf) (*tls.Config, error) { - if !cfg.Enabled { - return nil, nil - } - return auth.LoadServerTLS(cfg.Cert, cfg.Key, cfg.CA) -} diff --git a/configs/node.yaml b/configs/node.yaml index 3429def..88b6085 100644 --- a/configs/node.yaml +++ b/configs/node.yaml @@ -3,7 +3,7 @@ node: name: "iop-node-local" transport: - listen: "0.0.0.0:9090" + edge_addr: "localhost:9090" tls: enabled: false diff --git a/packages/config/config.go b/packages/config/config.go index 59b27fd..e902db9 100644 --- a/packages/config/config.go +++ b/packages/config/config.go @@ -21,7 +21,7 @@ type NodeInfo struct { } type TransportConf struct { - Listen string `mapstructure:"listen" yaml:"listen"` + EdgeAddr string `mapstructure:"edge_addr" yaml:"edge_addr"` } type TLSConf struct { @@ -90,7 +90,7 @@ func Load(cfgFile string) (*NodeConfig, error) { } func setDefaults(v *viper.Viper) { - v.SetDefault("transport.listen", "0.0.0.0:9090") + v.SetDefault("transport.edge_addr", "localhost:9090") v.SetDefault("runtime.concurrency", 4) v.SetDefault("runtime.workspace_root", "/tmp/iop/workspace") v.SetDefault("sqlite.dsn", "file:iop.db?cache=shared&mode=rwc") diff --git a/packages/protocol/protocol.go b/packages/protocol/protocol.go deleted file mode 100644 index 5573372..0000000 --- a/packages/protocol/protocol.go +++ /dev/null @@ -1,92 +0,0 @@ -// Package protocol defines the IOP wire-level message types. -// The transport layer uses 4-byte big-endian length-prefixed frames. -// Each frame carries a JSON-encoded Envelope; Envelope.Payload holds -// the JSON-encoded inner message (RunRequest, RunEvent, etc.). -// -// TODO: replace JSON encoding with protobuf once proto/iop/runtime.proto -// is compiled with protoc-gen-go. -package protocol - -const ProtocolVersion uint32 = 1 - -type MessageType string - -const ( - TypeRunRequest MessageType = "run.request" - TypeRunEvent MessageType = "run.event" - TypeHeartbeat MessageType = "heartbeat" - TypeCancelRequest MessageType = "cancel.request" - TypeError MessageType = "error" - TypeCapabilityRequest MessageType = "capability.request" - TypeCapabilityResponse MessageType = "capability.response" -) - -// Envelope is the outer wrapper for every IOP TCP message. -type Envelope struct { - ProtocolVersion uint32 `json:"protocol_version"` - RequestID string `json:"request_id"` - RunID string `json:"run_id,omitempty"` - StreamID string `json:"stream_id,omitempty"` - Type MessageType `json:"type"` - Payload []byte `json:"payload"` -} - -// RunRequest initiates a model execution on the node. -type RunRequest struct { - RunID string `json:"run_id"` - Adapter string `json:"adapter,omitempty"` - Model string `json:"model,omitempty"` - Workspace string `json:"workspace,omitempty"` - Policy map[string]any `json:"policy,omitempty"` - Input map[string]any `json:"input"` - TimeoutSec int `json:"timeout_sec,omitempty"` - Metadata map[string]string `json:"metadata,omitempty"` -} - -// RunEvent is a streaming execution event sent from node to client. -type RunEvent struct { - RunID string `json:"run_id"` - Type string `json:"type"` // start | delta | complete | error - Delta string `json:"delta,omitempty"` - Message string `json:"message,omitempty"` - Error string `json:"error,omitempty"` - Usage *UsageInfo `json:"usage,omitempty"` - Metadata map[string]string `json:"metadata,omitempty"` - Timestamp int64 `json:"timestamp"` // unix nano -} - -type UsageInfo struct { - InputTokens int `json:"input_tokens"` - OutputTokens int `json:"output_tokens"` -} - -// Heartbeat keeps the TCP connection alive. -type Heartbeat struct { - Timestamp int64 `json:"timestamp"` -} - -// CancelRequest asks the node to cancel a running execution. -type CancelRequest struct { - RunID string `json:"run_id"` -} - -// ErrorMsg is returned when a request fails at the transport layer. -type ErrorMsg struct { - Code string `json:"code"` - Message string `json:"message"` -} - -// CapabilityRequest asks the node what adapters/models it supports. -type CapabilityRequest struct{} - -// CapabilityResponse lists the node's available adapters. -type CapabilityResponse struct { - NodeID string `json:"node_id"` - Adapters []AdapterInfo `json:"adapters"` -} - -type AdapterInfo struct { - Name string `json:"name"` - Models []string `json:"models"` - MaxConcurrency int `json:"max_concurrency"` -}