fix(upgrade): stabilize batch node upgrades under long-running operations

This commit is contained in:
sagit
2026-02-12 04:57:49 +00:00
parent bdfc704f95
commit cdb2914dbf
4 changed files with 38 additions and 20 deletions
+26 -12
View File
@@ -6,6 +6,7 @@ import (
"io" "io"
"net/http" "net/http"
"strings" "strings"
"sync"
"time" "time"
"go-backend/internal/http/response" "go-backend/internal/http/response"
@@ -16,6 +17,8 @@ const (
githubProxy = "https://gcode.hostcentral.cc" githubProxy = "https://gcode.hostcentral.cc"
githubAPIBase = "https://api.github.com" githubAPIBase = "https://api.github.com"
githubHTMLBase = "https://github.com" githubHTMLBase = "https://github.com"
upgradeTimeout = 5 * time.Minute
batchWorkers = 5
) )
func (h *Handler) nodeUpgrade(w http.ResponseWriter, r *http.Request) { 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{}{ result, err := h.wsServer.SendCommand(req.ID, "UpgradeAgent", map[string]interface{}{
"downloadUrl": downloadURL, "downloadUrl": downloadURL,
"checksumUrl": checksumURL, "checksumUrl": checksumURL,
}, 120*time.Second) }, upgradeTimeout)
if err != nil { if err != nil {
response.WriteJSON(w, response.Err(-2, fmt.Sprintf("升级失败: %v", err))) response.WriteJSON(w, response.Err(-2, fmt.Sprintf("升级失败: %v", err)))
return return
@@ -173,18 +176,29 @@ func (h *Handler) nodeBatchUpgrade(w http.ResponseWriter, r *http.Request) {
Message string `json:"message"` Message string `json:"message"`
} }
results := make([]upgradeResult, 0, len(req.IDs)) results := make([]upgradeResult, len(req.IDs))
for _, id := range req.IDs { sem := make(chan struct{}, batchWorkers)
result, err := h.wsServer.SendCommand(id, "UpgradeAgent", map[string]interface{}{ var wg sync.WaitGroup
"downloadUrl": downloadURL,
"checksumUrl": checksumURL, for i, id := range req.IDs {
}, 120*time.Second) wg.Add(1)
if err != nil { go func(index int, nodeID int64) {
results = append(results, upgradeResult{ID: id, Success: false, Message: err.Error()}) defer wg.Done()
} else { sem <- struct{}{}
results = append(results, upgradeResult{ID: id, Success: true, Message: result.Message}) 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{}{ response.WriteJSON(w, response.OK(map[string]interface{}{
"version": version, "version": version,
+2 -4
View File
@@ -485,9 +485,8 @@ func (w *WebSocketReporter) handleReceivedMessage(messageType int, message []byt
} }
if cmdMsg.Type != "call" { if cmdMsg.Type != "call" {
// TcpPing 诊断命令异步执行,避免阻塞其他命令
// 其他状态变更命令保持同步,确保顺序执行 // 其他状态变更命令保持同步,确保顺序执行
if cmdMsg.Type == "TcpPing" { if cmdMsg.Type == "TcpPing" || cmdMsg.Type == "UpgradeAgent" || cmdMsg.Type == "RollbackAgent" {
go w.routeCommand(cmdMsg) go w.routeCommand(cmdMsg)
} else { } else {
w.routeCommand(cmdMsg) w.routeCommand(cmdMsg)
@@ -502,9 +501,8 @@ func (w *WebSocketReporter) handleReceivedMessage(messageType int, message []byt
return return
} }
if cmdMsg.Type != "call" { if cmdMsg.Type != "call" {
// TcpPing 诊断命令异步执行,避免阻塞其他命令
// 其他状态变更命令保持同步,确保顺序执行 // 其他状态变更命令保持同步,确保顺序执行
if cmdMsg.Type == "TcpPing" { if cmdMsg.Type == "TcpPing" || cmdMsg.Type == "UpgradeAgent" || cmdMsg.Type == "RollbackAgent" {
go w.routeCommand(cmdMsg) go w.routeCommand(cmdMsg)
} else { } else {
w.routeCommand(cmdMsg) w.routeCommand(cmdMsg)
+2 -2
View File
@@ -42,9 +42,9 @@ export const checkNodeStatus = (nodeId?: number) => {
}; };
export const upgradeNode = (id: number, version?: string) => 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) => 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 getNodeReleases = () => Network.post("/node/releases");
export const rollbackNode = (id: number) => export const rollbackNode = (id: number) =>
Network.post("/node/rollback", { id }); Network.post("/node/rollback", { id });
+8 -2
View File
@@ -43,6 +43,10 @@ interface ApiResponse<T = any> {
data: T; data: T;
} }
interface RequestOptions {
timeout?: number;
}
// 处理token失效的逻辑 // 处理token失效的逻辑
function handleTokenExpired() { function handleTokenExpired() {
// 清除localStorage中的token // 清除localStorage中的token
@@ -71,6 +75,7 @@ const Network = {
get: function <T = any>( get: function <T = any>(
path: string = "", path: string = "",
data: any = {}, data: any = {},
options: RequestOptions = {},
): Promise<ApiResponse<T>> { ): Promise<ApiResponse<T>> {
return new Promise(function (resolve) { return new Promise(function (resolve) {
// 如果baseURL是默认值且是WebView环境,说明没有设置面板地址 // 如果baseURL是默认值且是WebView环境,说明没有设置面板地址
@@ -83,7 +88,7 @@ const Network = {
axios axios
.get(path, { .get(path, {
params: data, params: data,
timeout: 30000, timeout: options.timeout ?? 30000,
headers: { headers: {
Authorization: window.localStorage.getItem("token"), Authorization: window.localStorage.getItem("token"),
}, },
@@ -117,6 +122,7 @@ const Network = {
post: function <T = any>( post: function <T = any>(
path: string = "", path: string = "",
data: any = {}, data: any = {},
options: RequestOptions = {},
): Promise<ApiResponse<T>> { ): Promise<ApiResponse<T>> {
return new Promise(function (resolve) { return new Promise(function (resolve) {
// 如果baseURL是默认值且是WebView环境,说明没有设置面板地址 // 如果baseURL是默认值且是WebView环境,说明没有设置面板地址
@@ -128,7 +134,7 @@ const Network = {
axios axios
.post(path, data, { .post(path, data, {
timeout: 30000, timeout: options.timeout ?? 30000,
headers: { headers: {
Authorization: window.localStorage.getItem("token"), Authorization: window.localStorage.getItem("token"),
"Content-Type": "application/json", "Content-Type": "application/json",