Function Run

internal/nodeagent/agent.go:116–174

Run 阻塞运行 agent，直到 ctx 取消。期间任何错误（连接失败、stream 出错、 Recv EOF 等）都会触发指数退避重连。

(ctx context.Context, cfg Config)

// Run 阻塞运行 agent，直到 ctx 取消。期间任何错误（连接失败、stream 出错、
// Recv EOF 等）都会触发指数退避重连。
func Run(ctx context.Context, cfg Config) error {
	if cfg.Dispatcher == nil {
		return errors.New("nodeagent: Config.Dispatcher is required")
	}
	if cfg.HelloProvider == nil {
		return errors.New("nodeagent: Config.HelloProvider is required")
	}
	if cfg.ServerAddr == "" && cfg.Dialer == nil {
		return errors.New("nodeagent: Config.ServerAddr or Config.Dialer is required")
	}
	if cfg.Logger == nil {
		cfg.Logger = slog.Default()
	}
	if len(cfg.ReconnectBackoff) == 0 {
		cfg.ReconnectBackoff = []time.Duration{2 * time.Second, 5 * time.Second, 15 * time.Second, 60 * time.Second}
	}
	if cfg.KeepaliveTime == 0 {
		cfg.KeepaliveTime = 30 * time.Second
	}
	if cfg.KeepaliveTimeout == 0 {
		cfg.KeepaliveTimeout = 10 * time.Second
	}

	// 默认 Dialer：从 Cert/Key/CA 加载 mTLS，按 ServerAddr 建立 gRPC 连接。
	if cfg.Dialer == nil {
		creds, err := loadTLSCreds(cfg)
		if err != nil {
			return fmt.Errorf("nodeagent: load TLS: %w", err)
		}
		cfg.Dialer = func(_ context.Context) (*grpc.ClientConn, error) {
			return grpc.NewClient(cfg.ServerAddr,
				grpc.WithTransportCredentials(creds),
				keepaliveParams(cfg),
			)
		}
	}

	attempts := 0
	for {
		err := runSession(ctx, cfg)
		if ctx.Err() != nil {
			return ctx.Err()
		}
		// 选取等待时长：min(attempts, len-1)。
		idx := attempts
		if idx >= len(cfg.ReconnectBackoff) {
			idx = len(cfg.ReconnectBackoff) - 1
		}
		wait := cfg.ReconnectBackoff[idx]
		cfg.Logger.Warn("nodeagent: session ended, reconnecting",
			"node_id", cfg.NodeID, "wait", wait, "attempt", attempts, "err", err)
		select {
		case <-time.After(wait):
			attempts++
		case <-ctx.Done():
			return ctx.Err()
		}
	}