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

fix(upgrade): stabilize batch node upgrades

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

@@ -6,6 +6,7 @@ import (
 	"io"
 	"net/http"
 	"strings"
+	"sync"
 	"time"
 
 	"go-backend/internal/http/response"
@@ -16,6 +17,8 @@ const (
 	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) {
@@ -59,7 +62,7 @@ func (h *Handler) nodeUpgrade(w http.ResponseWriter, r *http.Request) {
 	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
@@ -173,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,

go-backend/internal/ws/server.go

@@ -292,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()
@@ -429,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)
@@ -478,6 +482,7 @@ func startKeepalive(cw *connWrap, done <-chan struct{}) {
 			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()

go-gost/x/socket/websocket_reporter.go

@@ -485,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)
@@ -502,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)

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",