Merge pull request #92 from Sagit-chu/opencode/lucky-otter

fix(upgrade): stabilize batch node upgrades

.gitignore

@@ -177,6 +177,7 @@ build/
 *.dylib
 your_app.exe
 go-backend/paneld
+go-gost/gost
 
 # Go 测试二进制文件
 *.test

go-backend/internal/http/handler/mutations.go

@@ -413,7 +413,7 @@ func (h *Handler) nodeInstall(w http.ResponseWriter, r *http.Request) {
 		response.WriteJSON(w, response.Err(-2, err.Error()))
 		return
 	}
-	cmd := fmt.Sprintf("curl -L https://gcode.hostcentral.cc/https://github.com/Sagit-chu/flux-panel/releases/latest/download/install.sh -o ./install.sh && chmod +x ./install.sh && ./install.sh -a %s -s %s", processServerAddress(panelAddr), secret)
+	cmd := fmt.Sprintf("curl -L https://gcode.hostcentral.cc/https://github.com/Sagit-chu/flvx/releases/latest/download/install.sh -o ./install.sh && chmod +x ./install.sh && ./install.sh -a %s -s %s", processServerAddress(panelAddr), secret)
 	response.WriteJSON(w, response.OK(cmd))
 }
 

go-backend/internal/http/handler/upgrade.go

@@ -3,13 +3,24 @@ package handler
 import (
 	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
 	"strings"
+	"sync"
 	"time"
 
 	"go-backend/internal/http/response"
 )
 
+const (
+	githubRepo     = "Sagit-chu/flvx"
+	githubProxy    = "https://gcode.hostcentral.cc"
+	githubAPIBase  = "https://api.github.com"
+	githubHTMLBase = "https://github.com"
+	upgradeTimeout = 5 * time.Minute
+	batchWorkers   = 5
+)
+
 func (h *Handler) nodeUpgrade(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {
 		response.WriteJSON(w, response.ErrDefault("请求失败"))
@@ -40,18 +51,18 @@ func (h *Handler) nodeUpgrade(w http.ResponseWriter, r *http.Request) {
 	}
 
 	downloadURL := fmt.Sprintf(
-		"https://gcode.hostcentral.cc/https://github.com/Sagit-chu/flux-panel/releases/download/%s/gost-{ARCH}",
-		version,
+		githubProxy+"/%s/%s/releases/download/%s/gost-{ARCH}",
+		githubHTMLBase, githubRepo, version,
 	)
 	checksumURL := fmt.Sprintf(
-		"https://gcode.hostcentral.cc/https://github.com/Sagit-chu/flux-panel/releases/download/%s/gost-{ARCH}.sha256",
-		version,
+		githubProxy+"/%s/%s/releases/download/%s/gost-{ARCH}.sha256",
+		githubHTMLBase, githubRepo, version,
 	)
 
 	result, err := h.wsServer.SendCommand(req.ID, "UpgradeAgent", map[string]interface{}{
 		"downloadUrl": downloadURL,
 		"checksumUrl": checksumURL,
-	}, 120*time.Second)
+	}, upgradeTimeout)
 	if err != nil {
 		response.WriteJSON(w, response.Err(-2, fmt.Sprintf("升级失败: %v", err)))
 		return
@@ -71,7 +82,7 @@ func resolveLatestRelease() (string, error) {
 		Timeout: 10 * time.Second,
 	}
 
-	resp, err := client.Get("https://gcode.hostcentral.cc/https://github.com/Sagit-chu/flux-panel/releases/latest")
+	resp, err := client.Get(githubProxy + "/" + githubHTMLBase + "/" + githubRepo + "/releases/latest")
 	if err != nil {
 		return "", fmt.Errorf("请求GitHub失败: %v", err)
 	}
@@ -97,12 +108,17 @@ func resolveLatestRelease() (string, error) {
 
 func resolveLatestReleaseAPI() (string, error) {
 	client := &http.Client{Timeout: 10 * time.Second}
-	resp, err := client.Get("https://gcode.hostcentral.cc/https://api.github.com/repos/Sagit-chu/flux-panel/releases/latest")
+	resp, err := client.Get(githubAPIBase + "/repos/" + githubRepo + "/releases/latest")
 	if err != nil {
 		return "", fmt.Errorf("请求GitHub API失败: %v", err)
 	}
 	defer resp.Body.Close()
 
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
+		return "", fmt.Errorf("GitHub API返回 %d: %s", resp.StatusCode, string(body))
+	}
+
 	var release struct {
 		TagName string `json:"tag_name"`
 	}
@@ -146,12 +162,12 @@ func (h *Handler) nodeBatchUpgrade(w http.ResponseWriter, r *http.Request) {
 	}
 
 	downloadURL := fmt.Sprintf(
-		"https://gcode.hostcentral.cc/https://github.com/Sagit-chu/flux-panel/releases/download/%s/gost-{ARCH}",
-		version,
+		githubProxy+"/%s/%s/releases/download/%s/gost-{ARCH}",
+		githubHTMLBase, githubRepo, version,
 	)
 	checksumURL := fmt.Sprintf(
-		"https://gcode.hostcentral.cc/https://github.com/Sagit-chu/flux-panel/releases/download/%s/gost-{ARCH}.sha256",
-		version,
+		githubProxy+"/%s/%s/releases/download/%s/gost-{ARCH}.sha256",
+		githubHTMLBase, githubRepo, version,
 	)
 
 	type upgradeResult struct {
@@ -160,18 +176,29 @@ func (h *Handler) nodeBatchUpgrade(w http.ResponseWriter, r *http.Request) {
 		Message string `json:"message"`
 	}
 
-	results := make([]upgradeResult, 0, len(req.IDs))
-	for _, id := range req.IDs {
-		result, err := h.wsServer.SendCommand(id, "UpgradeAgent", map[string]interface{}{
-			"downloadUrl": downloadURL,
-			"checksumUrl": checksumURL,
-		}, 120*time.Second)
-		if err != nil {
-			results = append(results, upgradeResult{ID: id, Success: false, Message: err.Error()})
-		} else {
-			results = append(results, upgradeResult{ID: id, Success: true, Message: result.Message})
-		}
+	results := make([]upgradeResult, len(req.IDs))
+	sem := make(chan struct{}, batchWorkers)
+	var wg sync.WaitGroup
+
+	for i, id := range req.IDs {
+		wg.Add(1)
+		go func(index int, nodeID int64) {
+			defer wg.Done()
+			sem <- struct{}{}
+			defer func() { <-sem }()
+
+			result, err := h.wsServer.SendCommand(nodeID, "UpgradeAgent", map[string]interface{}{
+				"downloadUrl": downloadURL,
+				"checksumUrl": checksumURL,
+			}, upgradeTimeout)
+			if err != nil {
+				results[index] = upgradeResult{ID: nodeID, Success: false, Message: err.Error()}
+				return
+			}
+			results[index] = upgradeResult{ID: nodeID, Success: true, Message: result.Message}
+		}(i, id)
 	}
+	wg.Wait()
 
 	response.WriteJSON(w, response.OK(map[string]interface{}{
 		"version": version,
@@ -186,13 +213,19 @@ func (h *Handler) listReleases(w http.ResponseWriter, r *http.Request) {
 	}
 
 	client := &http.Client{Timeout: 15 * time.Second}
-	resp, err := client.Get("https://gcode.hostcentral.cc/https://api.github.com/repos/Sagit-chu/flux-panel/releases?per_page=20")
+	resp, err := client.Get(githubAPIBase + "/repos/" + githubRepo + "/releases?per_page=20")
 	if err != nil {
 		response.WriteJSON(w, response.Err(-2, fmt.Sprintf("获取版本列表失败: %v", err)))
 		return
 	}
 	defer resp.Body.Close()
 
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
+		response.WriteJSON(w, response.Err(-2, fmt.Sprintf("获取版本列表失败: GitHub API返回 %d: %s", resp.StatusCode, string(body))))
+		return
+	}
+
 	var releases []struct {
 		TagName     string `json:"tag_name"`
 		Name        string `json:"name"`

go-backend/internal/store/sqlite/repository.go

@@ -165,7 +165,13 @@ func Open(path string) (*Repository, error) {
 		return nil, err
 	}
 
-	db, err := sql.Open("sqlite", path)
+	// Use _pragma DSN parameters so every connection from the pool gets
+	// the same settings (busy_timeout and synchronous are per-connection).
+	dsn := "file:" + path +
+		"?_pragma=busy_timeout(5000)" +
+		"&_pragma=journal_mode(WAL)" +
+		"&_pragma=synchronous(NORMAL)"
+	db, err := sql.Open("sqlite", dsn)
 	if err != nil {
 		return nil, err
 	}

go-backend/internal/ws/server.go

@@ -54,6 +54,12 @@ type pendingRequest struct {
 	ch     chan CommandResult
 }
 
+const (
+	wsPingPeriod = 15 * time.Second
+	wsPongWait   = 45 * time.Second
+	wsWriteWait  = 5 * time.Second
+)
+
 type CommandResult struct {
 	Type    string                 `json:"type"`
 	Success bool                   `json:"success"`
@@ -120,12 +126,19 @@ func (s *Server) handleAdmin(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	cw := &connWrap{conn: conn}
+	_ = conn.SetReadDeadline(time.Now().Add(wsPongWait))
+	conn.SetPongHandler(func(string) error {
+		return conn.SetReadDeadline(time.Now().Add(wsPongWait))
+	})
+	done := make(chan struct{})
+	go startKeepalive(cw, done)
 
 	s.mu.Lock()
 	s.admins[cw] = struct{}{}
 	s.mu.Unlock()
 
 	defer func() {
+		close(done)
 		s.mu.Lock()
 		delete(s.admins, cw)
 		s.mu.Unlock()
@@ -145,6 +158,12 @@ func (s *Server) handleNode(w http.ResponseWriter, r *http.Request, nodeID int64
 		return
 	}
 	cw := &connWrap{conn: conn}
+	_ = conn.SetReadDeadline(time.Now().Add(wsPongWait))
+	conn.SetPongHandler(func(string) error {
+		return conn.SetReadDeadline(time.Now().Add(wsPongWait))
+	})
+	done := make(chan struct{})
+	go startKeepalive(cw, done)
 
 	version := r.URL.Query().Get("version")
 	httpVal := parseIntDefault(r.URL.Query().Get("http"), 0)
@@ -165,6 +184,7 @@ func (s *Server) handleNode(w http.ResponseWriter, r *http.Request, nodeID int64
 	s.broadcastStatus(nodeID, 1)
 
 	defer func() {
+		close(done)
 		needOfflineBroadcast := false
 		s.mu.Lock()
 		current, ok := s.nodes[nodeID]
@@ -272,7 +292,9 @@ func (s *Server) SendCommand(nodeID int64, cmdType string, data interface{}, tim
 	}
 
 	ns.conn.mu.Lock()
+	_ = ns.conn.conn.SetWriteDeadline(time.Now().Add(wsWriteWait))
 	err = ns.conn.conn.WriteMessage(websocket.TextMessage, messageData)
+	_ = ns.conn.conn.SetWriteDeadline(time.Time{})
 	ns.conn.mu.Unlock()
 	if err != nil {
 		cleanup()
@@ -409,7 +431,9 @@ func (s *Server) broadcastToAdmins(message string) {
 
 	for _, c := range admins {
 		c.mu.Lock()
+		_ = c.conn.SetWriteDeadline(time.Now().Add(wsWriteWait))
 		err := c.conn.WriteMessage(websocket.TextMessage, []byte(message))
+		_ = c.conn.SetWriteDeadline(time.Time{})
 		c.mu.Unlock()
 		if err != nil {
 			log.Printf("websocket broadcast failed: %v", err)
@@ -442,3 +466,28 @@ func parseIntDefault(v string, fallback int) int {
 	}
 	return x
 }
+
+func startKeepalive(cw *connWrap, done <-chan struct{}) {
+	if cw == nil || cw.conn == nil {
+		return
+	}
+	ticker := time.NewTicker(wsPingPeriod)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-done:
+			return
+		case <-ticker.C:
+			cw.mu.Lock()
+			_ = cw.conn.SetWriteDeadline(time.Now().Add(wsWriteWait))
+			err := cw.conn.WriteMessage(websocket.PingMessage, nil)
+			_ = cw.conn.SetWriteDeadline(time.Time{})
+			cw.mu.Unlock()
+			if err != nil {
+				_ = cw.conn.Close()
+				return
+			}
+		}
+	}
+}

go-gost/x/socket/websocket_reporter.go

@@ -18,7 +18,6 @@ import (
 	"strconv"
 	"strings"
 	"sync" // 新增：用于管理连接状态的互斥锁
-	"syscall"
 	"time"
 
 	"github.com/go-gost/x/config"
@@ -92,6 +91,11 @@ type TcpPingResponse struct {
 	RequestId    string  `json:"requestId,omitempty"`
 }
 
+const (
+	reporterReadWait  = 60 * time.Second
+	reporterWriteWait = 5 * time.Second
+)
+
 type WebSocketReporter struct {
 	url            string
 	addr           string // 保存服务器地址
@@ -244,6 +248,14 @@ func (w *WebSocketReporter) connect() error {
 
 	w.conn = conn
 	w.connected = true
+	_ = conn.SetReadDeadline(time.Now().Add(reporterReadWait))
+	conn.SetPingHandler(func(appData string) error {
+		_ = conn.SetReadDeadline(time.Now().Add(reporterReadWait))
+		return conn.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(reporterWriteWait))
+	})
+	conn.SetPongHandler(func(string) error {
+		return conn.SetReadDeadline(time.Now().Add(reporterReadWait))
+	})
 
 	// 设置关闭处理器来检测连接状态
 	w.conn.SetCloseHandler(func(code int, text string) error {
@@ -384,7 +396,7 @@ func (w *WebSocketReporter) receiveMessages() {
 			}
 
 			// 设置读取超时
-			conn.SetReadDeadline(time.Now().Add(30 * time.Second))
+			conn.SetReadDeadline(time.Now().Add(reporterReadWait))
 
 			messageType, message, err := conn.ReadMessage()
 			if err != nil {
@@ -473,9 +485,8 @@ func (w *WebSocketReporter) handleReceivedMessage(messageType int, message []byt
 			}
 
 			if cmdMsg.Type != "call" {
-				// TcpPing 诊断命令异步执行，避免阻塞其他命令
 				// 其他状态变更命令保持同步，确保顺序执行
-				if cmdMsg.Type == "TcpPing" {
+				if cmdMsg.Type == "TcpPing" || cmdMsg.Type == "UpgradeAgent" || cmdMsg.Type == "RollbackAgent" {
 					go w.routeCommand(cmdMsg)
 				} else {
 					w.routeCommand(cmdMsg)
@@ -490,9 +501,8 @@ func (w *WebSocketReporter) handleReceivedMessage(messageType int, message []byt
 				return
 			}
 			if cmdMsg.Type != "call" {
-				// TcpPing 诊断命令异步执行，避免阻塞其他命令
 				// 其他状态变更命令保持同步，确保顺序执行
-				if cmdMsg.Type == "TcpPing" {
+				if cmdMsg.Type == "TcpPing" || cmdMsg.Type == "UpgradeAgent" || cmdMsg.Type == "RollbackAgent" {
 					go w.routeCommand(cmdMsg)
 				} else {
 					w.routeCommand(cmdMsg)
@@ -1043,10 +1053,11 @@ func (w *WebSocketReporter) handleUpgradeAgent(data interface{}) error {
 	w.sendUpgradeProgress("installing", 80, "准备重启...")
 	fmt.Printf("✅ 升级包下载完成 (%d bytes), 准备重启...\n", downloaded)
 
-	// 执行重启脚本（分离进程，不阻塞当前进程）
+	// 执行重启脚本
+	// 使用 systemd-run 在独立的 transient unit 中运行重启脚本，
+	// 避免 systemctl stop 杀死 flux_agent cgroup 内所有进程（包括此脚本自身）导致 mv 未执行。
 	script := fmt.Sprintf("sleep 1 && systemctl stop flux_agent && mv %s %s && systemctl start flux_agent", tmpPath, binaryPath)
-	cmd := exec.Command("/bin/sh", "-c", script)
-	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	cmd := exec.Command("systemd-run", "--quiet", "/bin/sh", "-c", script)
 	if err := cmd.Start(); err != nil {
 		os.Remove(tmpPath)
 		return fmt.Errorf("启动重启脚本失败: %v", err)
@@ -1068,10 +1079,9 @@ func (w *WebSocketReporter) handleRollbackAgent(data interface{}) error {
 
 	fmt.Println("🔄 开始回退到旧版本...")
 
-	// 执行回退脚本
+	// 执行回退脚本（同升级逻辑，使用 systemd-run 避免 cgroup 问题）
 	script := fmt.Sprintf("sleep 1 && systemctl stop flux_agent && cp %s %s && systemctl start flux_agent", backupPath, binaryPath)
-	cmd := exec.Command("/bin/sh", "-c", script)
-	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	cmd := exec.Command("systemd-run", "--quiet", "/bin/sh", "-c", script)
 	if err := cmd.Start(); err != nil {
 		return fmt.Errorf("启动回退脚本失败: %v", err)
 	}

vite-frontend/nginx.conf

@@ -87,6 +87,8 @@ http {
             proxy_http_version 1.1;
             proxy_set_header Upgrade $http_upgrade;
             proxy_set_header Connection "upgrade";
+            proxy_read_timeout 3600s;
+            proxy_send_timeout 3600s;
             
             proxy_set_header Host $host;
             proxy_set_header X-Real-IP $remote_addr;
@@ -94,4 +96,4 @@ http {
             proxy_set_header X-Forwarded-Proto $scheme;
         }
     }
-} 
+} 

vite-frontend/src/api/index.ts

@@ -42,9 +42,9 @@ export const checkNodeStatus = (nodeId?: number) => {
 };
 
 export const upgradeNode = (id: number, version?: string) =>
-  Network.post("/node/upgrade", { id, version: version || "" });
+  Network.post("/node/upgrade", { id, version: version || "" }, { timeout: 5 * 60 * 1000 });
 export const batchUpgradeNodes = (ids: number[], version?: string) =>
-  Network.post("/node/batch-upgrade", { ids, version: version || "" });
+  Network.post("/node/batch-upgrade", { ids, version: version || "" }, { timeout: 15 * 60 * 1000 });
 export const getNodeReleases = () => Network.post("/node/releases");
 export const rollbackNode = (id: number) =>
   Network.post("/node/rollback", { id });

vite-frontend/src/api/network.ts

@@ -43,6 +43,10 @@ interface ApiResponse<T = any> {
   data: T;
 }
 
+interface RequestOptions {
+  timeout?: number;
+}
+
 // 处理token失效的逻辑
 function handleTokenExpired() {
   // 清除localStorage中的token
@@ -71,6 +75,7 @@ const Network = {
   get: function <T = any>(
     path: string = "",
     data: any = {},
+    options: RequestOptions = {},
   ): Promise<ApiResponse<T>> {
     return new Promise(function (resolve) {
       // 如果baseURL是默认值且是WebView环境，说明没有设置面板地址
@@ -83,7 +88,7 @@ const Network = {
       axios
         .get(path, {
           params: data,
-          timeout: 30000,
+          timeout: options.timeout ?? 30000,
           headers: {
             Authorization: window.localStorage.getItem("token"),
           },
@@ -117,6 +122,7 @@ const Network = {
   post: function <T = any>(
     path: string = "",
     data: any = {},
+    options: RequestOptions = {},
   ): Promise<ApiResponse<T>> {
     return new Promise(function (resolve) {
       // 如果baseURL是默认值且是WebView环境，说明没有设置面板地址
@@ -128,7 +134,7 @@ const Network = {
 
       axios
         .post(path, data, {
-          timeout: 30000,
+          timeout: options.timeout ?? 30000,
           headers: {
             Authorization: window.localStorage.getItem("token"),
             "Content-Type": "application/json",

vite-frontend/src/pages/node.tsx

@@ -125,7 +125,7 @@ const SortableItem = ({
   };
 
   return (
-    <div ref={setNodeRef} style={style} {...attributes}>
+    <div ref={setNodeRef} style={style} {...attributes} className="overflow-hidden">
       {children(listeners)}
     </div>
   );
@@ -1253,7 +1253,7 @@ export default function NodePage() {
                     {(listeners) => (
                     <Card
                       key={node.id}
-                      className="group shadow-sm border border-divider hover:shadow-md transition-shadow duration-200"
+                      className="group shadow-sm border border-divider hover:shadow-md transition-shadow duration-200 overflow-hidden"
                     >
                       <CardHeader className="pb-2">
                         <div className="flex justify-between items-start w-full">
@@ -1514,54 +1514,56 @@ export default function NodePage() {
 
                         {/* 操作按钮 */}
                         <div className="space-y-1.5">
-                          <div className="flex gap-1.5">
+                          {!isRemoteNode && (
+                            <div className="grid grid-cols-3 gap-1.5">
+                              <Button
+                                className="min-h-8"
+                                color="success"
+                                isLoading={node.copyLoading}
+                                size="sm"
+                                variant="flat"
+                                onPress={() => handleCopyInstallCommand(node)}
+                              >
+                                安装
+                              </Button>
+                              <Button
+                                className="min-h-8"
+                                color="warning"
+                                isDisabled={node.connectionStatus !== "online"}
+                                isLoading={node.upgradeLoading}
+                                size="sm"
+                                variant="flat"
+                                onPress={() => openUpgradeModal("single", node.id)}
+                              >
+                                升级
+                              </Button>
+                              <Button
+                                className="min-h-8"
+                                color="secondary"
+                                isDisabled={node.connectionStatus !== "online"}
+                                isLoading={node.rollbackLoading}
+                                size="sm"
+                                variant="flat"
+                                onPress={() => handleRollbackNode(node)}
+                              >
+                                回退
+                              </Button>
+                            </div>
+                          )}
+                          <div className={`grid gap-1.5 ${isRemoteNode ? "grid-cols-1" : "grid-cols-2"}`}>
                             {!isRemoteNode && (
-                              <>
-                                <Button
-                                  className="flex-1 min-h-8"
-                                  color="success"
-                                  isLoading={node.copyLoading}
-                                  size="sm"
-                                  variant="flat"
-                                  onPress={() => handleCopyInstallCommand(node)}
-                                >
-                                  安装
-                                </Button>
-                                <Button
-                                  className="flex-1 min-h-8"
-                                  color="warning"
-                                  isDisabled={node.connectionStatus !== "online"}
-                                  isLoading={node.upgradeLoading}
-                                  size="sm"
-                                  variant="flat"
-                                  onPress={() => openUpgradeModal("single", node.id)}
-                                >
-                                  升级
-                                </Button>
-                                <Button
-                                  className="flex-1 min-h-8"
-                                  color="secondary"
-                                  isDisabled={node.connectionStatus !== "online"}
-                                  isLoading={node.rollbackLoading}
-                                  size="sm"
-                                  variant="flat"
-                                  onPress={() => handleRollbackNode(node)}
-                                >
-                                  回退
-                                </Button>
-                                <Button
-                                  className="flex-1 min-h-8"
-                                  color="primary"
-                                  size="sm"
-                                  variant="flat"
-                                  onPress={() => handleEdit(node)}
-                                >
-                                  编辑
-                                </Button>
-                              </>
+                              <Button
+                                className="min-h-8"
+                                color="primary"
+                                size="sm"
+                                variant="flat"
+                                onPress={() => handleEdit(node)}
+                              >
+                                编辑
+                              </Button>
                             )}
                             <Button
-                              className={`min-h-8 ${isRemoteNode ? "w-full" : "flex-1"}`}
+                              className="min-h-8"
                               color="danger"
                               size="sm"
                               variant="flat"