deepseek-tray/supervisor.go

package main

import (
	"bufio"
	"errors"
	"fmt"
	"io"
	"os/exec"
	"strconv"
	"strings"
	"sync"
	"time"
)

// supervisor.go —— dsh web 守护核心：端口预检、隐藏启动、指数退避重启、状态快照。

// healthyUptime 稳定运行判定阈值：超过它则清零失败计数（退避复位）。
const healthyUptime = 30 * time.Second

// Status 服务状态快照。
type Status struct {
	Running   bool
	PID       int
	Restarts  int
	Uptime    time.Duration
	LastError string
	URL       string
	Updating  bool
	External  bool // 端口被外部实例占用（非本程序启动）
}

// Supervisor 守护 dsh web。
type Supervisor struct {
	mu        sync.Mutex
	probeMu   sync.Mutex // 串行化入口探测（见 probe.go）
	cfg       *Config
	log       *Logger
	pid       int
	running   bool
	desired   bool
	restarts  int
	failures  int
	startedAt time.Time
	lastError string
	updating  bool
	external  bool
	binCache  string
	nodeCache string
	webURL    string // dsh web 输出的完整 URL（含访问 token，仅存内存）

	autoMu   sync.Mutex
	autoLoop bool // 定期更新循环是否已启动（进程生命周期内单例）
}

func NewSupervisor(cfg *Config, log *Logger) *Supervisor {
	return &Supervisor{cfg: cfg, log: log}
}

// Start 设置期望运行并拉起守护循环。
func (s *Supervisor) Start() {
	s.mu.Lock()
	if s.desired {
		s.mu.Unlock()
		return
	}
	s.desired = true
	s.failures = 0
	s.mu.Unlock()
	go s.monitor()
}

// Stop 同步停止：结束进程树并等待其真正退出（最多 8 秒）。
func (s *Supervisor) Stop() {
	s.mu.Lock()
	s.desired = false
	pid := s.pid
	s.webURL = ""
	s.mu.Unlock()
	if pid > 0 {
		s.log.Printf("停止 dsh web pid=%d", pid)
		_ = killTree(pid)
		s.waitStopped(pid, 8*time.Second)
	}
}

// Restart 重启并等待端口释放，避免新旧实例争抢端口。
func (s *Supervisor) Restart() {
	s.log.Printf("重启 dsh web")
	s.Stop()
	s.waitPortFree(5 * time.Second)
	s.Start()
}

func (s *Supervisor) Status() Status {
	s.mu.Lock()
	defer s.mu.Unlock()
	url := s.baseURL()
	if s.webURL != "" {
		url = s.webURL // 优先使用带 token 的地址
	}
	var uptime time.Duration
	if s.running && !s.startedAt.IsZero() {
		uptime = time.Since(s.startedAt)
	}
	return Status{
		Running:   s.running,
		PID:       s.pid,
		Restarts:  s.restarts,
		Uptime:    uptime,
		LastError: s.lastError,
		URL:       url,
		Updating:  s.updating,
		External:  s.external,
	}
}

func (s *Supervisor) baseURL() string {
	host := s.cfg.WebHost
	if host == "" {
		host = "127.0.0.1"
	}
	return fmt.Sprintf("http://%s:%d", host, s.cfg.WebPort)
}

func (s *Supervisor) setError(err error) {
	s.mu.Lock()
	s.lastError = err.Error()
	s.mu.Unlock()
	s.log.Printf("守护错误: %v", err)
}

// maxBackoff 重启退避上限。
const maxBackoff = 2 * time.Minute

// backoff 指数退避：base * 2^failures，封顶 maxBackoff。
// 循环内提前 break，既保证封顶语义，也避免超大 failures 造成数值溢出。
func (s *Supervisor) backoff() time.Duration {
	s.mu.Lock()
	failures := s.failures
	base := s.cfg.RestartDelaySec
	s.mu.Unlock()
	if base <= 0 {
		base = 5
	}
	d := time.Duration(base) * time.Second
	for i := 0; i < failures; i++ {
		if d >= maxBackoff {
			break
		}
		d *= 2
	}
	if d > maxBackoff {
		d = maxBackoff
	}
	return d
}

// portBusyBackoff 端口占用时的温和退避（上限 30 秒）：不是故障，只是等待外部实例释放。
func (s *Supervisor) portBusyBackoff() time.Duration {
	d := s.backoff()
	if d > 30*time.Second {
		d = 30 * time.Second
	}
	return d
}

// monitor 守护循环：端口预检 -> 启动 -> 等待退出 -> 期望运行时退避重启。
func (s *Supervisor) monitor() {
	for {
		if !s.isDesired() {
			return
		}

		// 端口预检：已有实例监听时不再启动，避免端口冲突导致的重启风暴
		if isPortOpen(s.cfg.WebHost, s.cfg.WebPort) {
			s.mu.Lock()
			s.running = false
			s.pid = 0
			s.external = true
			s.lastError = "端口已被占用（可能已有 dsh web 在运行）"
			s.mu.Unlock()
			s.log.Printf("端口 %s 已被占用，跳过启动（等待释放）", s.baseURL())
			s.bumpFailure()
			s.sleepInterruptible(s.portBusyBackoff())
			continue
		}
		s.mu.Lock()
		s.external = false
		s.mu.Unlock()

		exe, args, err := s.resolveCommand()
		if err != nil {
			s.setError(err)
			s.bumpFailure()
			s.sleepInterruptible(s.backoff())
			continue
		}

		cmd := exec.Command(exe, args...)
		hideWindow(cmd)
		stdout, errOut := cmd.StdoutPipe()
		stderr, errErr := cmd.StderrPipe()
		if errOut != nil || errErr != nil {
			s.setError(errors.New("无法创建输出管道"))
			s.bumpFailure()
			s.sleepInterruptible(s.backoff())
			continue
		}
		if err := cmd.Start(); err != nil {
			s.setError(err)
			s.bumpFailure()
			s.sleepInterruptible(s.backoff())
			continue
		}

		pid := cmd.Process.Pid
		startedAt := time.Now()
		s.mu.Lock()
		s.pid = pid
		s.running = true
		s.startedAt = startedAt
		s.lastError = ""
		s.webURL = ""
		s.mu.Unlock()
		s.log.Printf("dsh web 已启动 pid=%d: %s %s", pid, exe, strings.Join(args, " "))

		go s.pipeLog("out", stdout)
		go s.pipeLog("err", stderr)

		waitErr := cmd.Wait()
		ranFor := time.Since(startedAt)

		s.mu.Lock()
		s.running = false
		s.pid = 0
		stillDesired := s.desired
		if ranFor >= healthyUptime {
			s.failures = 0 // 稳定运行过，退避复位
		} else {
			s.failures++
		}
		if stillDesired {
			s.restarts++
		}
		s.mu.Unlock()
		s.log.Printf("dsh web 退出（err=%v，运行 %s），期望运行=%v", waitErr, ranFor.Round(time.Second), stillDesired)

		if !stillDesired {
			return
		}
		delay := s.backoff()
		s.log.Printf("%s 后自动重启", delay)
		s.sleepInterruptible(delay)
	}
}

func (s *Supervisor) isDesired() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.desired
}

// bumpFailure 增加失败计数（触发退避放大）。
func (s *Supervisor) bumpFailure() {
	s.mu.Lock()
	s.failures++
	s.mu.Unlock()
}

func (s *Supervisor) sleepInterruptible(d time.Duration) {
	deadline := time.Now().Add(d)
	for time.Now().Before(deadline) {
		if !s.isDesired() {
			return
		}
		time.Sleep(200 * time.Millisecond)
	}
}

func (s *Supervisor) waitStopped(pid int, timeout time.Duration) {
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		if !processAlive(pid) {
			return
		}
		time.Sleep(100 * time.Millisecond)
	}
	s.log.Printf("等待 pid=%d 退出超时（继续）", pid)
}

func (s *Supervisor) waitPortFree(timeout time.Duration) {
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		if !isPortOpen(s.cfg.WebHost, s.cfg.WebPort) {
			return
		}
		time.Sleep(150 * time.Millisecond)
	}
}

// pipeLog 把子进程输出写入日志（token 由 Logger 统一脱敏），并捕获访问地址。
func (s *Supervisor) pipeLog(tag string, r io.ReadCloser) {
	if r == nil {
		return
	}
	sc := bufio.NewScanner(r)
	sc.Buffer(make([]byte, 0, 64*1024), 1024*1024)
	for sc.Scan() {
		line := sc.Text()
		s.log.Printf("[dsh %s] %s", tag, line)
		if url := extractHTTPURL(line); url != "" {
			s.mu.Lock()
			if url != s.webURL {
				s.webURL = url
				s.log.Printf("捕获 Web 访问地址（含 token，仅存内存）")
			}
			s.mu.Unlock()
		}
	}
}

// resolveCommand 解析启动命令：优先 node + dsh 入口 JS（无 shell 包装，避免弹窗与注入面）。
func (s *Supervisor) resolveCommand() (string, []string, error) {
	node := s.cfg.NodePath
	if node == "" {
		node = s.nodePath()
	}
	binJS := s.cfg.DshBinJS
	if binJS == "" {
		binJS = s.dshBinJS()
	}
	port := strconv.Itoa(s.cfg.WebPort)
	if node != "" && binJS != "" {
		return node, []string{binJS, "web", "--no-open", "--host", s.cfg.WebHost, "--port", port}, nil
	}
	if p, err := exec.LookPath("dsh.cmd"); err == nil {
		return "cmd.exe", []string{"/c", p, "web", "--no-open", "--port", port}, nil
	}
	if p, err := exec.LookPath("dsh"); err == nil {
		return p, []string{"web", "--no-open", "--port", port}, nil
	}
	return "", nil, errors.New("未找到 dsh 入口：请确认已全局安装 @deepseek-ai/dsh，或在配置中指定 nodePath / dshBinJs")
}