From cc6e71db79e9dd706d96ea28db60e5905a0d2b35 Mon Sep 17 00:00:00 2001 From: toki Date: Wed, 22 Jul 2026 18:11:12 +0900 Subject: [PATCH] =?UTF-8?q?feat(proto,config):=20runtime=20ready=20handsha?= =?UTF-8?q?ke=20proto=20=EC=8A=A4=ED=82=A4=EB=A7=88=20=EB=B0=8F=20?= =?UTF-8?q?=EC=9E=AC=EC=97=B0=EA=B2=B0=20=EC=84=A4=EC=A0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - NodeReadyRequest/Response: 노드→엣지 연결 준비 확인 핸드셰이크 추가 - ReconnectConf: interval_sec, max_attempts 명시적 설정 지원 - node_config_test: 재연결 설정 로드 검증 --- go.mod | 3 +- packages/go/config/load.go | 21 ++ packages/go/config/node_config_test.go | 112 ++++++++++ packages/go/config/node_types.go | 11 +- proto/gen/iop/runtime.pb.go | 296 +++++++++++++++++-------- proto/iop/runtime.proto | 20 ++ 6 files changed, 370 insertions(+), 93 deletions(-) diff --git a/go.mod b/go.mod index a4fd42e..d27672a 100644 --- a/go.mod +++ b/go.mod @@ -14,6 +14,7 @@ require ( google.golang.org/protobuf v1.36.5 gopkg.in/yaml.v3 v3.0.1 modernc.org/sqlite v1.33.1 + nhooyr.io/websocket v1.8.17 ) require ( @@ -25,6 +26,7 @@ require ( github.com/hashicorp/hcl v1.0.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/klauspost/compress v1.17.9 // indirect + github.com/kylelemons/godebug v1.1.0 // indirect github.com/magiconair/properties v1.8.7 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/mitchellh/mapstructure v1.5.0 // indirect @@ -54,7 +56,6 @@ require ( modernc.org/memory v1.8.0 // indirect modernc.org/strutil v1.2.0 // indirect modernc.org/token v1.1.0 // indirect - nhooyr.io/websocket v1.8.17 // indirect ) replace git.toki-labs.com/toki/proto-socket/go => ../proto-socket/go diff --git a/packages/go/config/load.go b/packages/go/config/load.go index b82202f..f29c5ab 100644 --- a/packages/go/config/load.go +++ b/packages/go/config/load.go @@ -18,9 +18,30 @@ func Load(cfgFile string) (*NodeConfig, error) { if err := v.Unmarshal(&cfg); err != nil { return nil, err } + if err := validateReconnect(cfg.Reconnect); err != nil { + return nil, err + } return &cfg, nil } +// validateReconnect enforces the reconnect policy contract (SDD S17). setDefaults +// supplies the omitted defaults (interval_sec=10, max_attempts=10) before this +// runs, so an omitted max_attempts is already 10 here while an explicit 0 stays +// 0 (unlimited). Negative max_attempts or interval_sec are rejected, and an +// unlimited policy requires a positive interval so it cannot hot-loop. +func validateReconnect(rc ReconnectConf) error { + if rc.MaxAttempts < 0 { + return fmt.Errorf("reconnect.max_attempts must not be negative") + } + if rc.IntervalSec < 0 { + return fmt.Errorf("reconnect.interval_sec must not be negative") + } + if rc.MaxAttempts == 0 && rc.IntervalSec <= 0 { + return fmt.Errorf("reconnect.interval_sec must be positive when reconnect.max_attempts=0 (unlimited)") + } + return nil +} + func LoadEdge(cfgFile string) (*EdgeConfig, error) { v := viper.New() v.SetConfigFile(cfgFile) diff --git a/packages/go/config/node_config_test.go b/packages/go/config/node_config_test.go index 29e7991..17cfa24 100644 --- a/packages/go/config/node_config_test.go +++ b/packages/go/config/node_config_test.go @@ -275,6 +275,118 @@ func TestLoad_NodeReconnectOverride(t *testing.T) { } } +// TestLoadReconnectPolicy verifies that config load distinguishes an omitted +// max_attempts (default 10), an explicit 0 (unlimited), and a positive finite +// limit, and preserves an explicit interval alongside each. SDD S16/S17. +func TestLoadReconnectPolicy(t *testing.T) { + cases := []struct { + name string + reconnectYAML string + wantInterval int + wantMax int + }{ + { + name: "omitted keeps defaults", + reconnectYAML: "", + wantInterval: 10, + wantMax: 10, + }, + { + name: "explicit zero max is unlimited with default interval", + reconnectYAML: "reconnect:\n max_attempts: 0\n", + wantInterval: 10, + wantMax: 0, + }, + { + name: "explicit zero max with positive interval stays unlimited", + reconnectYAML: "reconnect:\n max_attempts: 0\n interval_sec: 5\n", + wantInterval: 5, + wantMax: 0, + }, + { + name: "positive finite limit is preserved", + reconnectYAML: "reconnect:\n max_attempts: 3\n interval_sec: 30\n", + wantInterval: 30, + wantMax: 3, + }, + { + name: "positive finite limit allows zero interval", + reconnectYAML: "reconnect:\n max_attempts: 3\n interval_sec: 0\n", + wantInterval: 0, + wantMax: 3, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + dir := t.TempDir() + f := filepath.Join(dir, "node.yaml") + yaml := "transport:\n edge_addr: \"localhost:9090\"\n token: \"tok\"\n" + tc.reconnectYAML + if err := os.WriteFile(f, []byte(yaml), 0o600); err != nil { + t.Fatalf("write yaml: %v", err) + } + cfg, err := config.Load(f) + if err != nil { + t.Fatalf("load: %v", err) + } + if cfg.Reconnect.IntervalSec != tc.wantInterval { + t.Errorf("interval_sec: got %d want %d", cfg.Reconnect.IntervalSec, tc.wantInterval) + } + if cfg.Reconnect.MaxAttempts != tc.wantMax { + t.Errorf("max_attempts: got %d want %d", cfg.Reconnect.MaxAttempts, tc.wantMax) + } + }) + } +} + +// TestLoadRejectsInvalidReconnectPolicy verifies that negative attempts/interval +// and an unlimited policy with a non-positive interval are rejected at load with +// a clear config error. SDD S17. +func TestLoadRejectsInvalidReconnectPolicy(t *testing.T) { + cases := []struct { + name string + reconnectYAML string + wantSubstr string + }{ + { + name: "negative max_attempts", + reconnectYAML: "reconnect:\n max_attempts: -1\n", + wantSubstr: "reconnect.max_attempts must not be negative", + }, + { + name: "negative interval_sec", + reconnectYAML: "reconnect:\n max_attempts: 3\n interval_sec: -1\n", + wantSubstr: "reconnect.interval_sec must not be negative", + }, + { + name: "unlimited with zero interval", + reconnectYAML: "reconnect:\n max_attempts: 0\n interval_sec: 0\n", + wantSubstr: "reconnect.interval_sec must be positive", + }, + { + name: "unlimited with negative interval", + reconnectYAML: "reconnect:\n max_attempts: 0\n interval_sec: -5\n", + wantSubstr: "reconnect.interval_sec must not be negative", + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + dir := t.TempDir() + f := filepath.Join(dir, "node.yaml") + yaml := "transport:\n edge_addr: \"localhost:9090\"\n token: \"tok\"\n" + tc.reconnectYAML + if err := os.WriteFile(f, []byte(yaml), 0o600); err != nil { + t.Fatalf("write yaml: %v", err) + } + _, err := config.Load(f) + if err == nil { + t.Fatalf("expected error for %s", tc.name) + } + if !strings.Contains(err.Error(), tc.wantSubstr) { + t.Fatalf("expected error containing %q, got %v", tc.wantSubstr, err) + } + }) + } +} + // TestLoadEdge_NodeProviderPriorityOmittedDefaultsZero verifies that when // priority is omitted from a provider entry, it defaults to zero. func TestLoadEdge_NodeProviderPriorityOmittedDefaultsZero(t *testing.T) { diff --git a/packages/go/config/node_types.go b/packages/go/config/node_types.go index eefe2b5..9b4d4e7 100644 --- a/packages/go/config/node_types.go +++ b/packages/go/config/node_types.go @@ -8,9 +8,18 @@ type NodeConfig struct { Metrics MetricsConf `mapstructure:"metrics" yaml:"metrics"` } -// ReconnectConf controls Edge reconnect behaviour after a disconnect. +// ReconnectConf controls how the Node connectivity supervisor retries the +// initial dial and post-disconnect reconnects to Edge. The same policy applies +// to both the initial connect and established-session reconnect paths. type ReconnectConf struct { + // IntervalSec is the delay between connect attempts. An omitted value keeps + // the default 10s; it must be positive when MaxAttempts is 0 (unlimited) so + // the supervisor cannot hot-loop. Negative values are a config error. IntervalSec int `mapstructure:"interval_sec" yaml:"interval_sec"` + // MaxAttempts bounds retries per connect episode. An omitted value keeps the + // default 10; an explicit 0 means unlimited retries until local shutdown; a + // positive value is a finite limit whose exhaustion terminates the Node with + // a non-zero exit code. Negative values are a config error. MaxAttempts int `mapstructure:"max_attempts" yaml:"max_attempts"` } diff --git a/proto/gen/iop/runtime.pb.go b/proto/gen/iop/runtime.pb.go index 0deb49a..66cabcb 100644 --- a/proto/gen/iop/runtime.pb.go +++ b/proto/gen/iop/runtime.pb.go @@ -1764,6 +1764,113 @@ func (x *RegisterResponse) GetConfig() *NodeConfigPayload { return nil } +// NodeReadyRequest is sent by node to edge after it has applied the config from +// RegisterResponse and installed its message handler, signalling that it can now +// receive run/tunnel/command dispatch. Accepted registration only claims +// ownership and delivers config; edge opens dispatch eligibility and pumps the +// node's stranded waiters only on this ready handshake, never before. node_id +// carries the identity edge assigned in RegisterResponse so the ready transition +// binds to the exact accepted connection. +type NodeReadyRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + NodeId string `protobuf:"bytes,1,opt,name=node_id,json=nodeId,proto3" json:"node_id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *NodeReadyRequest) Reset() { + *x = NodeReadyRequest{} + mi := &file_proto_iop_runtime_proto_msgTypes[15] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *NodeReadyRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*NodeReadyRequest) ProtoMessage() {} + +func (x *NodeReadyRequest) ProtoReflect() protoreflect.Message { + mi := &file_proto_iop_runtime_proto_msgTypes[15] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use NodeReadyRequest.ProtoReflect.Descriptor instead. +func (*NodeReadyRequest) Descriptor() ([]byte, []int) { + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{15} +} + +func (x *NodeReadyRequest) GetNodeId() string { + if x != nil { + return x.NodeId + } + return "" +} + +// NodeReadyResponse acknowledges a NodeReadyRequest. ready is true once edge has +// marked the connection dispatch-ready (idempotent for the current owner); it is +// false when the connection is stale — superseded by a reconnect or already gone +// — in which case the node closes and reconnects. reason describes a rejection. +type NodeReadyResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + Ready bool `protobuf:"varint,1,opt,name=ready,proto3" json:"ready,omitempty"` + Reason string `protobuf:"bytes,2,opt,name=reason,proto3" json:"reason,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *NodeReadyResponse) Reset() { + *x = NodeReadyResponse{} + mi := &file_proto_iop_runtime_proto_msgTypes[16] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *NodeReadyResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*NodeReadyResponse) ProtoMessage() {} + +func (x *NodeReadyResponse) ProtoReflect() protoreflect.Message { + mi := &file_proto_iop_runtime_proto_msgTypes[16] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use NodeReadyResponse.ProtoReflect.Descriptor instead. +func (*NodeReadyResponse) Descriptor() ([]byte, []int) { + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{16} +} + +func (x *NodeReadyResponse) GetReady() bool { + if x != nil { + return x.Ready + } + return false +} + +func (x *NodeReadyResponse) GetReason() string { + if x != nil { + return x.Reason + } + return "" +} + // NodeConfigPayload carries all configuration edge pushes to the node. type NodeConfigPayload struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -1775,7 +1882,7 @@ type NodeConfigPayload struct { func (x *NodeConfigPayload) Reset() { *x = NodeConfigPayload{} - mi := &file_proto_iop_runtime_proto_msgTypes[15] + mi := &file_proto_iop_runtime_proto_msgTypes[17] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1787,7 +1894,7 @@ func (x *NodeConfigPayload) String() string { func (*NodeConfigPayload) ProtoMessage() {} func (x *NodeConfigPayload) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[15] + mi := &file_proto_iop_runtime_proto_msgTypes[17] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1800,7 +1907,7 @@ func (x *NodeConfigPayload) ProtoReflect() protoreflect.Message { // Deprecated: Use NodeConfigPayload.ProtoReflect.Descriptor instead. func (*NodeConfigPayload) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{15} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{17} } func (x *NodeConfigPayload) GetAdapters() []*AdapterConfig { @@ -1842,7 +1949,7 @@ type AdapterConfig struct { func (x *AdapterConfig) Reset() { *x = AdapterConfig{} - mi := &file_proto_iop_runtime_proto_msgTypes[16] + mi := &file_proto_iop_runtime_proto_msgTypes[18] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1854,7 +1961,7 @@ func (x *AdapterConfig) String() string { func (*AdapterConfig) ProtoMessage() {} func (x *AdapterConfig) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[16] + mi := &file_proto_iop_runtime_proto_msgTypes[18] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1867,7 +1974,7 @@ func (x *AdapterConfig) ProtoReflect() protoreflect.Message { // Deprecated: Use AdapterConfig.ProtoReflect.Descriptor instead. func (*AdapterConfig) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{16} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{18} } func (x *AdapterConfig) GetType() string { @@ -1999,7 +2106,7 @@ type MockAdapterConfig struct { func (x *MockAdapterConfig) Reset() { *x = MockAdapterConfig{} - mi := &file_proto_iop_runtime_proto_msgTypes[17] + mi := &file_proto_iop_runtime_proto_msgTypes[19] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2011,7 +2118,7 @@ func (x *MockAdapterConfig) String() string { func (*MockAdapterConfig) ProtoMessage() {} func (x *MockAdapterConfig) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[17] + mi := &file_proto_iop_runtime_proto_msgTypes[19] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2024,7 +2131,7 @@ func (x *MockAdapterConfig) ProtoReflect() protoreflect.Message { // Deprecated: Use MockAdapterConfig.ProtoReflect.Descriptor instead. func (*MockAdapterConfig) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{17} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{19} } type CLIAdapterConfig struct { @@ -2036,7 +2143,7 @@ type CLIAdapterConfig struct { func (x *CLIAdapterConfig) Reset() { *x = CLIAdapterConfig{} - mi := &file_proto_iop_runtime_proto_msgTypes[18] + mi := &file_proto_iop_runtime_proto_msgTypes[20] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2048,7 +2155,7 @@ func (x *CLIAdapterConfig) String() string { func (*CLIAdapterConfig) ProtoMessage() {} func (x *CLIAdapterConfig) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[18] + mi := &file_proto_iop_runtime_proto_msgTypes[20] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2061,7 +2168,7 @@ func (x *CLIAdapterConfig) ProtoReflect() protoreflect.Message { // Deprecated: Use CLIAdapterConfig.ProtoReflect.Descriptor instead. func (*CLIAdapterConfig) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{18} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{20} } func (x *CLIAdapterConfig) GetProfiles() map[string]*CLIProfileConfig { @@ -2090,7 +2197,7 @@ type CLIProfileConfig struct { func (x *CLIProfileConfig) Reset() { *x = CLIProfileConfig{} - mi := &file_proto_iop_runtime_proto_msgTypes[19] + mi := &file_proto_iop_runtime_proto_msgTypes[21] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2102,7 +2209,7 @@ func (x *CLIProfileConfig) String() string { func (*CLIProfileConfig) ProtoMessage() {} func (x *CLIProfileConfig) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[19] + mi := &file_proto_iop_runtime_proto_msgTypes[21] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2115,7 +2222,7 @@ func (x *CLIProfileConfig) ProtoReflect() protoreflect.Message { // Deprecated: Use CLIProfileConfig.ProtoReflect.Descriptor instead. func (*CLIProfileConfig) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{19} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{21} } func (x *CLIProfileConfig) GetCommand() string { @@ -2205,7 +2312,7 @@ type CLICompletionMarker struct { func (x *CLICompletionMarker) Reset() { *x = CLICompletionMarker{} - mi := &file_proto_iop_runtime_proto_msgTypes[20] + mi := &file_proto_iop_runtime_proto_msgTypes[22] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2217,7 +2324,7 @@ func (x *CLICompletionMarker) String() string { func (*CLICompletionMarker) ProtoMessage() {} func (x *CLICompletionMarker) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[20] + mi := &file_proto_iop_runtime_proto_msgTypes[22] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2230,7 +2337,7 @@ func (x *CLICompletionMarker) ProtoReflect() protoreflect.Message { // Deprecated: Use CLICompletionMarker.ProtoReflect.Descriptor instead. func (*CLICompletionMarker) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{20} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{22} } func (x *CLICompletionMarker) GetLine() string { @@ -2261,7 +2368,7 @@ type OllamaAdapterConfig struct { func (x *OllamaAdapterConfig) Reset() { *x = OllamaAdapterConfig{} - mi := &file_proto_iop_runtime_proto_msgTypes[21] + mi := &file_proto_iop_runtime_proto_msgTypes[23] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2273,7 +2380,7 @@ func (x *OllamaAdapterConfig) String() string { func (*OllamaAdapterConfig) ProtoMessage() {} func (x *OllamaAdapterConfig) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[21] + mi := &file_proto_iop_runtime_proto_msgTypes[23] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2286,7 +2393,7 @@ func (x *OllamaAdapterConfig) ProtoReflect() protoreflect.Message { // Deprecated: Use OllamaAdapterConfig.ProtoReflect.Descriptor instead. func (*OllamaAdapterConfig) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{21} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{23} } func (x *OllamaAdapterConfig) GetBaseUrl() string { @@ -2344,7 +2451,7 @@ type VllmAdapterConfig struct { func (x *VllmAdapterConfig) Reset() { *x = VllmAdapterConfig{} - mi := &file_proto_iop_runtime_proto_msgTypes[22] + mi := &file_proto_iop_runtime_proto_msgTypes[24] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2356,7 +2463,7 @@ func (x *VllmAdapterConfig) String() string { func (*VllmAdapterConfig) ProtoMessage() {} func (x *VllmAdapterConfig) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[22] + mi := &file_proto_iop_runtime_proto_msgTypes[24] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2369,7 +2476,7 @@ func (x *VllmAdapterConfig) ProtoReflect() protoreflect.Message { // Deprecated: Use VllmAdapterConfig.ProtoReflect.Descriptor instead. func (*VllmAdapterConfig) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{22} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{24} } func (x *VllmAdapterConfig) GetEndpoint() string { @@ -2422,7 +2529,7 @@ type OpenAICompatAdapterConfig struct { func (x *OpenAICompatAdapterConfig) Reset() { *x = OpenAICompatAdapterConfig{} - mi := &file_proto_iop_runtime_proto_msgTypes[23] + mi := &file_proto_iop_runtime_proto_msgTypes[25] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2434,7 +2541,7 @@ func (x *OpenAICompatAdapterConfig) String() string { func (*OpenAICompatAdapterConfig) ProtoMessage() {} func (x *OpenAICompatAdapterConfig) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[23] + mi := &file_proto_iop_runtime_proto_msgTypes[25] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2447,7 +2554,7 @@ func (x *OpenAICompatAdapterConfig) ProtoReflect() protoreflect.Message { // Deprecated: Use OpenAICompatAdapterConfig.ProtoReflect.Descriptor instead. func (*OpenAICompatAdapterConfig) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{23} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{25} } func (x *OpenAICompatAdapterConfig) GetProvider() string { @@ -2511,7 +2618,7 @@ type NodeRuntimeConfig struct { func (x *NodeRuntimeConfig) Reset() { *x = NodeRuntimeConfig{} - mi := &file_proto_iop_runtime_proto_msgTypes[24] + mi := &file_proto_iop_runtime_proto_msgTypes[26] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2523,7 +2630,7 @@ func (x *NodeRuntimeConfig) String() string { func (*NodeRuntimeConfig) ProtoMessage() {} func (x *NodeRuntimeConfig) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[24] + mi := &file_proto_iop_runtime_proto_msgTypes[26] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2536,7 +2643,7 @@ func (x *NodeRuntimeConfig) ProtoReflect() protoreflect.Message { // Deprecated: Use NodeRuntimeConfig.ProtoReflect.Descriptor instead. func (*NodeRuntimeConfig) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{24} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{26} } func (x *NodeRuntimeConfig) GetConcurrency() int32 { @@ -2558,7 +2665,7 @@ type NodeConfigRefreshRequest struct { func (x *NodeConfigRefreshRequest) Reset() { *x = NodeConfigRefreshRequest{} - mi := &file_proto_iop_runtime_proto_msgTypes[25] + mi := &file_proto_iop_runtime_proto_msgTypes[27] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2570,7 +2677,7 @@ func (x *NodeConfigRefreshRequest) String() string { func (*NodeConfigRefreshRequest) ProtoMessage() {} func (x *NodeConfigRefreshRequest) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[25] + mi := &file_proto_iop_runtime_proto_msgTypes[27] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2583,7 +2690,7 @@ func (x *NodeConfigRefreshRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use NodeConfigRefreshRequest.ProtoReflect.Descriptor instead. func (*NodeConfigRefreshRequest) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{25} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{27} } func (x *NodeConfigRefreshRequest) GetRequestId() string { @@ -2620,7 +2727,7 @@ type NodeConfigRefreshResponse struct { func (x *NodeConfigRefreshResponse) Reset() { *x = NodeConfigRefreshResponse{} - mi := &file_proto_iop_runtime_proto_msgTypes[26] + mi := &file_proto_iop_runtime_proto_msgTypes[28] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2632,7 +2739,7 @@ func (x *NodeConfigRefreshResponse) String() string { func (*NodeConfigRefreshResponse) ProtoMessage() {} func (x *NodeConfigRefreshResponse) ProtoReflect() protoreflect.Message { - mi := &file_proto_iop_runtime_proto_msgTypes[26] + mi := &file_proto_iop_runtime_proto_msgTypes[28] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2645,7 +2752,7 @@ func (x *NodeConfigRefreshResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use NodeConfigRefreshResponse.ProtoReflect.Descriptor instead. func (*NodeConfigRefreshResponse) Descriptor() ([]byte, []int) { - return file_proto_iop_runtime_proto_rawDescGZIP(), []int{26} + return file_proto_iop_runtime_proto_rawDescGZIP(), []int{28} } func (x *NodeConfigRefreshResponse) GetRequestId() string { @@ -2867,7 +2974,12 @@ const file_proto_iop_runtime_proto_rawDesc = "" + "\anode_id\x18\x02 \x01(\tR\x06nodeId\x12\x14\n" + "\x05alias\x18\x03 \x01(\tR\x05alias\x12\x16\n" + "\x06reason\x18\x04 \x01(\tR\x06reason\x12.\n" + - "\x06config\x18\x05 \x01(\v2\x16.iop.NodeConfigPayloadR\x06config\"u\n" + + "\x06config\x18\x05 \x01(\v2\x16.iop.NodeConfigPayloadR\x06config\"+\n" + + "\x10NodeReadyRequest\x12\x17\n" + + "\anode_id\x18\x01 \x01(\tR\x06nodeId\"A\n" + + "\x11NodeReadyResponse\x12\x14\n" + + "\x05ready\x18\x01 \x01(\bR\x05ready\x12\x16\n" + + "\x06reason\x18\x02 \x01(\tR\x06reason\"u\n" + "\x11NodeConfigPayload\x12.\n" + "\badapters\x18\x01 \x03(\v2\x12.iop.AdapterConfigR\badapters\x120\n" + "\aruntime\x18\x02 \x01(\v2\x16.iop.NodeRuntimeConfigR\aruntime\"\xaa\x03\n" + @@ -2988,7 +3100,7 @@ func file_proto_iop_runtime_proto_rawDescGZIP() []byte { } var file_proto_iop_runtime_proto_enumTypes = make([]protoimpl.EnumInfo, 5) -var file_proto_iop_runtime_proto_msgTypes = make([]protoimpl.MessageInfo, 39) +var file_proto_iop_runtime_proto_msgTypes = make([]protoimpl.MessageInfo, 41) var file_proto_iop_runtime_proto_goTypes = []any{ (RunSessionMode)(0), // 0: iop.RunSessionMode (CancelAction)(0), // 1: iop.CancelAction @@ -3010,69 +3122,71 @@ var file_proto_iop_runtime_proto_goTypes = []any{ (*Error)(nil), // 17: iop.Error (*RegisterRequest)(nil), // 18: iop.RegisterRequest (*RegisterResponse)(nil), // 19: iop.RegisterResponse - (*NodeConfigPayload)(nil), // 20: iop.NodeConfigPayload - (*AdapterConfig)(nil), // 21: iop.AdapterConfig - (*MockAdapterConfig)(nil), // 22: iop.MockAdapterConfig - (*CLIAdapterConfig)(nil), // 23: iop.CLIAdapterConfig - (*CLIProfileConfig)(nil), // 24: iop.CLIProfileConfig - (*CLICompletionMarker)(nil), // 25: iop.CLICompletionMarker - (*OllamaAdapterConfig)(nil), // 26: iop.OllamaAdapterConfig - (*VllmAdapterConfig)(nil), // 27: iop.VllmAdapterConfig - (*OpenAICompatAdapterConfig)(nil), // 28: iop.OpenAICompatAdapterConfig - (*NodeRuntimeConfig)(nil), // 29: iop.NodeRuntimeConfig - (*NodeConfigRefreshRequest)(nil), // 30: iop.NodeConfigRefreshRequest - (*NodeConfigRefreshResponse)(nil), // 31: iop.NodeConfigRefreshResponse - nil, // 32: iop.RunRequest.MetadataEntry - nil, // 33: iop.RunEvent.MetadataEntry - nil, // 34: iop.ProviderTunnelRequest.HeadersEntry - nil, // 35: iop.ProviderTunnelRequest.MetadataEntry - nil, // 36: iop.ProviderTunnelFrame.HeadersEntry - nil, // 37: iop.ProviderTunnelFrame.MetadataEntry - nil, // 38: iop.EdgeNodeEvent.MetadataEntry - nil, // 39: iop.NodeCommandRequest.MetadataEntry - nil, // 40: iop.NodeCommandResponse.ResultEntry - nil, // 41: iop.AgentUsageStatus.MetadataEntry - nil, // 42: iop.CLIAdapterConfig.ProfilesEntry - nil, // 43: iop.OpenAICompatAdapterConfig.HeadersEntry - (*structpb.Struct)(nil), // 44: google.protobuf.Struct + (*NodeReadyRequest)(nil), // 20: iop.NodeReadyRequest + (*NodeReadyResponse)(nil), // 21: iop.NodeReadyResponse + (*NodeConfigPayload)(nil), // 22: iop.NodeConfigPayload + (*AdapterConfig)(nil), // 23: iop.AdapterConfig + (*MockAdapterConfig)(nil), // 24: iop.MockAdapterConfig + (*CLIAdapterConfig)(nil), // 25: iop.CLIAdapterConfig + (*CLIProfileConfig)(nil), // 26: iop.CLIProfileConfig + (*CLICompletionMarker)(nil), // 27: iop.CLICompletionMarker + (*OllamaAdapterConfig)(nil), // 28: iop.OllamaAdapterConfig + (*VllmAdapterConfig)(nil), // 29: iop.VllmAdapterConfig + (*OpenAICompatAdapterConfig)(nil), // 30: iop.OpenAICompatAdapterConfig + (*NodeRuntimeConfig)(nil), // 31: iop.NodeRuntimeConfig + (*NodeConfigRefreshRequest)(nil), // 32: iop.NodeConfigRefreshRequest + (*NodeConfigRefreshResponse)(nil), // 33: iop.NodeConfigRefreshResponse + nil, // 34: iop.RunRequest.MetadataEntry + nil, // 35: iop.RunEvent.MetadataEntry + nil, // 36: iop.ProviderTunnelRequest.HeadersEntry + nil, // 37: iop.ProviderTunnelRequest.MetadataEntry + nil, // 38: iop.ProviderTunnelFrame.HeadersEntry + nil, // 39: iop.ProviderTunnelFrame.MetadataEntry + nil, // 40: iop.EdgeNodeEvent.MetadataEntry + nil, // 41: iop.NodeCommandRequest.MetadataEntry + nil, // 42: iop.NodeCommandResponse.ResultEntry + nil, // 43: iop.AgentUsageStatus.MetadataEntry + nil, // 44: iop.CLIAdapterConfig.ProfilesEntry + nil, // 45: iop.OpenAICompatAdapterConfig.HeadersEntry + (*structpb.Struct)(nil), // 46: google.protobuf.Struct } var file_proto_iop_runtime_proto_depIdxs = []int32{ - 44, // 0: iop.RunRequest.policy:type_name -> google.protobuf.Struct - 44, // 1: iop.RunRequest.input:type_name -> google.protobuf.Struct - 32, // 2: iop.RunRequest.metadata:type_name -> iop.RunRequest.MetadataEntry + 46, // 0: iop.RunRequest.policy:type_name -> google.protobuf.Struct + 46, // 1: iop.RunRequest.input:type_name -> google.protobuf.Struct + 34, // 2: iop.RunRequest.metadata:type_name -> iop.RunRequest.MetadataEntry 0, // 3: iop.RunRequest.session_mode:type_name -> iop.RunSessionMode 10, // 4: iop.RunEvent.usage:type_name -> iop.Usage - 33, // 5: iop.RunEvent.metadata:type_name -> iop.RunEvent.MetadataEntry - 34, // 6: iop.ProviderTunnelRequest.headers:type_name -> iop.ProviderTunnelRequest.HeadersEntry - 35, // 7: iop.ProviderTunnelRequest.metadata:type_name -> iop.ProviderTunnelRequest.MetadataEntry + 35, // 5: iop.RunEvent.metadata:type_name -> iop.RunEvent.MetadataEntry + 36, // 6: iop.ProviderTunnelRequest.headers:type_name -> iop.ProviderTunnelRequest.HeadersEntry + 37, // 7: iop.ProviderTunnelRequest.metadata:type_name -> iop.ProviderTunnelRequest.MetadataEntry 2, // 8: iop.ProviderTunnelFrame.kind:type_name -> iop.ProviderTunnelFrameKind - 36, // 9: iop.ProviderTunnelFrame.headers:type_name -> iop.ProviderTunnelFrame.HeadersEntry + 38, // 9: iop.ProviderTunnelFrame.headers:type_name -> iop.ProviderTunnelFrame.HeadersEntry 10, // 10: iop.ProviderTunnelFrame.usage:type_name -> iop.Usage - 37, // 11: iop.ProviderTunnelFrame.metadata:type_name -> iop.ProviderTunnelFrame.MetadataEntry - 38, // 12: iop.EdgeNodeEvent.metadata:type_name -> iop.EdgeNodeEvent.MetadataEntry + 39, // 11: iop.ProviderTunnelFrame.metadata:type_name -> iop.ProviderTunnelFrame.MetadataEntry + 40, // 12: iop.EdgeNodeEvent.metadata:type_name -> iop.EdgeNodeEvent.MetadataEntry 1, // 13: iop.CancelRequest.action:type_name -> iop.CancelAction 3, // 14: iop.NodeCommandRequest.type:type_name -> iop.NodeCommandType - 39, // 15: iop.NodeCommandRequest.metadata:type_name -> iop.NodeCommandRequest.MetadataEntry + 41, // 15: iop.NodeCommandRequest.metadata:type_name -> iop.NodeCommandRequest.MetadataEntry 3, // 16: iop.NodeCommandResponse.type:type_name -> iop.NodeCommandType 16, // 17: iop.NodeCommandResponse.usage_status:type_name -> iop.AgentUsageStatus - 40, // 18: iop.NodeCommandResponse.result:type_name -> iop.NodeCommandResponse.ResultEntry + 42, // 18: iop.NodeCommandResponse.result:type_name -> iop.NodeCommandResponse.ResultEntry 15, // 19: iop.NodeCommandResponse.provider_snapshots:type_name -> iop.ProviderSnapshot - 41, // 20: iop.AgentUsageStatus.metadata:type_name -> iop.AgentUsageStatus.MetadataEntry - 20, // 21: iop.RegisterResponse.config:type_name -> iop.NodeConfigPayload - 21, // 22: iop.NodeConfigPayload.adapters:type_name -> iop.AdapterConfig - 29, // 23: iop.NodeConfigPayload.runtime:type_name -> iop.NodeRuntimeConfig - 44, // 24: iop.AdapterConfig.settings:type_name -> google.protobuf.Struct - 23, // 25: iop.AdapterConfig.cli:type_name -> iop.CLIAdapterConfig - 26, // 26: iop.AdapterConfig.ollama:type_name -> iop.OllamaAdapterConfig - 27, // 27: iop.AdapterConfig.vllm:type_name -> iop.VllmAdapterConfig - 22, // 28: iop.AdapterConfig.mock:type_name -> iop.MockAdapterConfig - 28, // 29: iop.AdapterConfig.openai_compat:type_name -> iop.OpenAICompatAdapterConfig - 42, // 30: iop.CLIAdapterConfig.profiles:type_name -> iop.CLIAdapterConfig.ProfilesEntry - 25, // 31: iop.CLIProfileConfig.completion_marker:type_name -> iop.CLICompletionMarker - 43, // 32: iop.OpenAICompatAdapterConfig.headers:type_name -> iop.OpenAICompatAdapterConfig.HeadersEntry - 20, // 33: iop.NodeConfigRefreshRequest.config:type_name -> iop.NodeConfigPayload + 43, // 20: iop.AgentUsageStatus.metadata:type_name -> iop.AgentUsageStatus.MetadataEntry + 22, // 21: iop.RegisterResponse.config:type_name -> iop.NodeConfigPayload + 23, // 22: iop.NodeConfigPayload.adapters:type_name -> iop.AdapterConfig + 31, // 23: iop.NodeConfigPayload.runtime:type_name -> iop.NodeRuntimeConfig + 46, // 24: iop.AdapterConfig.settings:type_name -> google.protobuf.Struct + 25, // 25: iop.AdapterConfig.cli:type_name -> iop.CLIAdapterConfig + 28, // 26: iop.AdapterConfig.ollama:type_name -> iop.OllamaAdapterConfig + 29, // 27: iop.AdapterConfig.vllm:type_name -> iop.VllmAdapterConfig + 24, // 28: iop.AdapterConfig.mock:type_name -> iop.MockAdapterConfig + 30, // 29: iop.AdapterConfig.openai_compat:type_name -> iop.OpenAICompatAdapterConfig + 44, // 30: iop.CLIAdapterConfig.profiles:type_name -> iop.CLIAdapterConfig.ProfilesEntry + 27, // 31: iop.CLIProfileConfig.completion_marker:type_name -> iop.CLICompletionMarker + 45, // 32: iop.OpenAICompatAdapterConfig.headers:type_name -> iop.OpenAICompatAdapterConfig.HeadersEntry + 22, // 33: iop.NodeConfigRefreshRequest.config:type_name -> iop.NodeConfigPayload 4, // 34: iop.NodeConfigRefreshResponse.status:type_name -> iop.NodeConfigRefreshStatus - 24, // 35: iop.CLIAdapterConfig.ProfilesEntry.value:type_name -> iop.CLIProfileConfig + 26, // 35: iop.CLIAdapterConfig.ProfilesEntry.value:type_name -> iop.CLIProfileConfig 36, // [36:36] is the sub-list for method output_type 36, // [36:36] is the sub-list for method input_type 36, // [36:36] is the sub-list for extension type_name @@ -3085,7 +3199,7 @@ func file_proto_iop_runtime_proto_init() { if File_proto_iop_runtime_proto != nil { return } - file_proto_iop_runtime_proto_msgTypes[16].OneofWrappers = []any{ + file_proto_iop_runtime_proto_msgTypes[18].OneofWrappers = []any{ (*AdapterConfig_Cli)(nil), (*AdapterConfig_Ollama)(nil), (*AdapterConfig_Vllm)(nil), @@ -3098,7 +3212,7 @@ func file_proto_iop_runtime_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_proto_iop_runtime_proto_rawDesc), len(file_proto_iop_runtime_proto_rawDesc)), NumEnums: 5, - NumMessages: 39, + NumMessages: 41, NumExtensions: 0, NumServices: 0, }, diff --git a/proto/iop/runtime.proto b/proto/iop/runtime.proto index c5c060e..b098b62 100644 --- a/proto/iop/runtime.proto +++ b/proto/iop/runtime.proto @@ -233,6 +233,26 @@ message RegisterResponse { NodeConfigPayload config = 5; } +// NodeReadyRequest is sent by node to edge after it has applied the config from +// RegisterResponse and installed its message handler, signalling that it can now +// receive run/tunnel/command dispatch. Accepted registration only claims +// ownership and delivers config; edge opens dispatch eligibility and pumps the +// node's stranded waiters only on this ready handshake, never before. node_id +// carries the identity edge assigned in RegisterResponse so the ready transition +// binds to the exact accepted connection. +message NodeReadyRequest { + string node_id = 1; +} + +// NodeReadyResponse acknowledges a NodeReadyRequest. ready is true once edge has +// marked the connection dispatch-ready (idempotent for the current owner); it is +// false when the connection is stale — superseded by a reconnect or already gone +// — in which case the node closes and reconnects. reason describes a rejection. +message NodeReadyResponse { + bool ready = 1; + string reason = 2; +} + // NodeConfigPayload carries all configuration edge pushes to the node. message NodeConfigPayload { repeated AdapterConfig adapters = 1;