package main import ( "bufio" "errors" "fmt" "io" "net" "os" "os/exec" "path/filepath" "strconv" "strings" "sync" "time" ) // 稳定运行的判定阈值:超过它则清零失败计数(退避复位)。 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 // 串行化入口探测,避免并发重复探测 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) } // backoff 指数退避:base * 2^failures(上限 2 分钟),避免故障时重启风暴。 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 < 5; i++ { d *= 2 } if d > 2*time.Minute { d = 2 * time.Minute } 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() } } } // extractHTTPURL 从一行输出里取出 http(s) URL(去除尾随分隔符)。 func extractHTTPURL(line string) string { idx := strings.Index(line, "http") if idx < 0 { return "" } url := strings.TrimSpace(line[idx:]) if i := strings.IndexAny(url, " \t"); i > 0 { url = url[:i] } if strings.HasPrefix(url, "http://") || strings.HasPrefix(url, "https://") { return url } return "" } // 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") } func (s *Supervisor) nodePath() string { s.mu.Lock() if s.nodeCache != "" { v := s.nodeCache s.mu.Unlock() return v } s.mu.Unlock() p, err := exec.LookPath("node") if err != nil { return "" } s.mu.Lock() s.nodeCache = p s.mu.Unlock() return p } // dshBinJS 探测 dsh 的 bin.js(APPDATA\npm 与 npm root -g);probeMu 保证并发只探测一次。 func (s *Supervisor) dshBinJS() string { s.probeMu.Lock() defer s.probeMu.Unlock() s.mu.Lock() if s.binCache != "" { v := s.binCache s.mu.Unlock() return v } s.mu.Unlock() var candidates []string if appdata := os.Getenv("APPDATA"); appdata != "" { candidates = append(candidates, filepath.Join(appdata, "npm", "node_modules", "@deepseek-ai", "dsh", "lib", "bin.js")) } if root := npmGlobalRoot(); root != "" { candidates = append(candidates, filepath.Join(root, "@deepseek-ai", "dsh", "lib", "bin.js")) } for _, c := range candidates { if fileExists(c) { s.mu.Lock() s.binCache = c s.mu.Unlock() s.log.Printf("dsh 入口: %s", c) return c } } return "" } // invalidateProbe 使入口缓存失效(更新后调用)。 func (s *Supervisor) invalidateProbe() { s.mu.Lock() s.binCache = "" s.mu.Unlock() } // installedVersion 读取已安装 dsh 版本。 func (s *Supervisor) installedVersion() string { binJS := s.cfg.DshBinJS if binJS == "" { binJS = s.dshBinJS() } if binJS == "" { return "" } return readPackageVersion(filepath.Join(filepath.Dir(filepath.Dir(binJS)), "package.json")) } func fileExists(p string) bool { st, err := os.Stat(p) return err == nil && !st.IsDir() } // isPortOpen 检测端口是否有监听者。 func isPortOpen(host string, port int) bool { if host == "" { host = "127.0.0.1" } conn, err := net.DialTimeout("tcp", fmt.Sprintf("%s:%d", host, port), 1200*time.Millisecond) if err != nil { return false } _ = conn.Close() return true }