package bootstrap import ( "context" "fmt" "sync" "time" "go.uber.org/zap" "iop/apps/edge/internal/configrefresh" edgecontrolplane "iop/apps/edge/internal/controlplane" edgeevents "iop/apps/edge/internal/events" edgeinput "iop/apps/edge/internal/input" edgenode "iop/apps/edge/internal/node" edgeservice "iop/apps/edge/internal/service" "iop/apps/edge/internal/transport" "iop/packages/go/config" "iop/packages/go/observability" "iop/packages/go/version" ) // Runtime is the shared edge runtime assembly used by both `serve` and // `console` entrypoints. NewRuntime wires logger, registry, node store, // event bus, service, and transport server; Start performs handler wiring // and binds the server and metrics endpoint. type Runtime struct { Cfg *config.EdgeConfig Logger *zap.Logger Registry *edgenode.Registry NodeStore *edgenode.NodeStore EventBus *edgeevents.Bus Service *edgeservice.Service Server *transport.Server Input *edgeinput.Manager Artifact *ArtifactServer ControlPlane *edgecontrolplane.Connector RefreshAdmin *RefreshAdminServer refreshMu sync.Mutex cfgMu sync.RWMutex lifetimeMu sync.Mutex lifetimeCancel context.CancelFunc } func NewRuntime(cfg *config.EdgeConfig) (*Runtime, error) { logger, err := observability.NewLoggerWithFile(cfg.Logging.Level, cfg.Logging.Pretty, cfg.Logging.Path) if err != nil { return nil, err } registry := edgenode.NewRegistry() nodeStore, err := edgenode.LoadFromConfig(cfg.Nodes) if err != nil { return nil, fmt.Errorf("edge: seed node store: %w", err) } bus := edgeevents.NewBus() svc := edgeservice.New(registry, bus) svc.SetNodeStore(nodeStore) svc.SetModelCatalog(cfg.Models) inputManager := edgeinput.NewManager(*cfg, svc, logger.Named("input")) artifactServer := NewArtifactServer(cfg.Bootstrap.Listen, cfg.Bootstrap.ArtifactDir, logger.Named("bootstrap")) server, err := transport.NewServer(cfg.Server.Listen, registry, nodeStore, logger) if err != nil { return nil, err } connector := edgecontrolplane.NewConnector( cfg.Edge, cfg.ControlPlane, version.Version, logger.Named("controlplane"), edgecontrolplane.WithStatusProvider(svc), edgecontrolplane.WithNodeEventBus(bus), ) rt := &Runtime{ Cfg: cfg, Logger: logger, Registry: registry, NodeStore: nodeStore, EventBus: bus, Service: svc, Server: server, Input: inputManager, Artifact: artifactServer, ControlPlane: connector, } rt.RefreshAdmin = newRefreshAdminServer(cfg.Refresh.Listen, rt, logger.Named("refresh-admin")) return rt, nil } func (r *Runtime) wireHandlers() { r.Server.SetRunEventHandler(r.EventBus.PublishRun) r.Server.SetNodeEventHandler(r.EventBus.PublishNode) } func (r *Runtime) Start(ctx context.Context) error { if err := ctx.Err(); err != nil { return err } r.wireHandlers() lifetimeCtx := r.newLifetimeContext() if err := r.Server.Start(lifetimeCtx); err != nil { r.cancelLifetime() return err } if err := ctx.Err(); err != nil { r.cancelLifetime() _ = r.Server.Stop() return err } if err := r.Input.Start(lifetimeCtx); err != nil { r.cancelLifetime() _ = r.Server.Stop() return err } if err := r.Artifact.Start(lifetimeCtx); err != nil { r.cancelLifetime() _ = r.Input.Stop(context.Background()) _ = r.Server.Stop() return err } if err := r.ControlPlane.Start(lifetimeCtx); err != nil { r.cancelLifetime() _ = r.Artifact.Stop(context.Background()) _ = r.Input.Stop(context.Background()) _ = r.Server.Stop() return err } if r.Cfg.Refresh.Enabled { if err := r.RefreshAdmin.Start(lifetimeCtx); err != nil { r.cancelLifetime() r.ControlPlane.Stop() _ = r.Artifact.Stop(context.Background()) _ = r.Input.Stop(context.Background()) _ = r.Server.Stop() return err } } r.startMetrics() return nil } func (r *Runtime) Stop() error { r.cancelLifetime() r.ControlPlane.Stop() ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() refreshErr := r.RefreshAdmin.Stop(ctx) inputErr := r.Input.Stop(ctx) artifactErr := r.Artifact.Stop(ctx) err := r.Server.Stop() _ = r.Logger.Sync() if err == nil { err = refreshErr } if err == nil { err = inputErr } if err == nil { err = artifactErr } return err } // RefreshConfig evaluates a config refresh request against the current runtime // config. In apply mode, mutable-only changes are committed to the runtime. func (r *Runtime) RefreshConfig(ctx context.Context, req configrefresh.Request) (configrefresh.Result, error) { r.refreshMu.Lock() defer r.refreshMu.Unlock() r.cfgMu.RLock() current := r.Cfg r.cfgMu.RUnlock() result, candidate, err := configrefresh.Evaluate(ctx, current, req) if err != nil { return result, err } if req.Mode == configrefresh.ModeApply && result.Status == configrefresh.StatusApplied && candidate != nil { if err := r.applyMutableConfig(candidate); err != nil { result.Status = configrefresh.StatusRejected result.Summary = "apply candidate config: " + err.Error() if result.Changes == nil { result.Changes = []configrefresh.Change{} } } } return result, nil } // applyMutableConfig commits a validated mutable-only candidate via copy-on-write // replacement. Readers observe either the previous runtime snapshot or the new // one, and validation/build failures leave the previous snapshot untouched. func (r *Runtime) applyMutableConfig(candidate *config.EdgeConfig) error { nextStore, err := edgenode.LoadFromConfig(candidate.Nodes) if err != nil { return err } r.cfgMu.Lock() r.Cfg = candidate r.NodeStore = nextStore r.cfgMu.Unlock() r.Server.SetNodeStore(nextStore) r.Service.SetRuntimeConfig(nextStore, candidate.Models) r.Input.SetModelCatalog(candidate.Models) return nil } func (r *Runtime) newLifetimeContext() context.Context { r.lifetimeMu.Lock() defer r.lifetimeMu.Unlock() if r.lifetimeCancel != nil { r.lifetimeCancel() } lifetimeCtx, cancel := context.WithCancel(context.Background()) r.lifetimeCancel = cancel return lifetimeCtx } func (r *Runtime) cancelLifetime() { r.lifetimeMu.Lock() cancel := r.lifetimeCancel r.lifetimeCancel = nil r.lifetimeMu.Unlock() if cancel != nil { cancel() } } func (r *Runtime) startMetrics() { if r.Cfg.Metrics.Port <= 0 { return } go func() { if err := observability.ServeMetrics(r.Cfg.Metrics.Port); err != nil { r.Logger.Warn("metrics server exited", zap.Error(err)) } }() }