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") }