diff --git a/go-backend/internal/http/handler/handler.go b/go-backend/internal/http/handler/handler.go index a5a814a..0ddad66 100644 --- a/go-backend/internal/http/handler/handler.go +++ b/go-backend/internal/http/handler/handler.go @@ -228,6 +228,7 @@ func (h *Handler) Register(mux *http.ServeMux) { mux.HandleFunc("/api/v1/tunnel/update-order", h.tunnelUpdateOrder) mux.HandleFunc("/api/v1/tunnel/batch-delete", h.tunnelBatchDelete) mux.HandleFunc("/api/v1/tunnel/batch-redeploy", h.tunnelBatchRedeploy) + mux.HandleFunc("/api/v1/tunnel/batch-redeploy/stream", h.tunnelBatchRedeployStream) mux.HandleFunc("/api/v1/tunnel/user/assign", h.userTunnelAssign) mux.HandleFunc("/api/v1/tunnel/user/batch-assign", h.userTunnelBatchAssign) mux.HandleFunc("/api/v1/tunnel/user/remove", h.userTunnelRemove) diff --git a/go-backend/internal/http/handler/mutations.go b/go-backend/internal/http/handler/mutations.go index 75947a2..23162b1 100644 --- a/go-backend/internal/http/handler/mutations.go +++ b/go-backend/internal/http/handler/mutations.go @@ -2046,51 +2046,37 @@ func normalizeBatchFailureReason(reason string) string { return trimmed } +func (h *Handler) redeployTunnelConfiguration(tunnelID int64) error { + if _, err := h.getTunnelRecord(tunnelID); err != nil { + return err + } + if nftMode, entryNodeIDs, err := h.tunnelUsesNftables(tunnelID); err != nil { + return err + } else if nftMode { + if len(entryNodeIDs) == 0 { + return errors.New("nftables 转发缺少入口节点") + } + return h.reconcileNftablesNodeByRequest(entryNodeIDs[0]) + } + return h.redeployTunnelAndForwards(tunnelID) +} + func (h *Handler) tunnelBatchRedeploy(w http.ResponseWriter, r *http.Request) { ids := idsFromBody(r, w) if ids == nil { return } - success := 0 - fail := 0 - failures := make([]batchFailureDetail, 0) + result := batchOperationResult{} for _, tunnelID := range ids { - tunnel, tunnelErr := h.getTunnelRecord(tunnelID) - tunnelName := "" - if tunnelErr == nil && tunnel != nil { - tunnelName, _ = h.repo.GetTunnelName(tunnelID) + name, _ := h.repo.GetTunnelName(tunnelID) + if err := h.redeployTunnelConfiguration(tunnelID); err != nil { + result.FailCount++ + result.Failures = appendBatchFailure(result.Failures, tunnelID, name, err) + } else { + result.SuccessCount++ } - if tunnelErr != nil { - fail++ - failures = appendBatchFailure(failures, tunnelID, tunnelName, tunnelErr) - continue - } - if nftMode, entryNodeIDs, modeErr := h.tunnelUsesNftables(tunnelID); modeErr != nil { - fail++ - failures = appendBatchFailure(failures, tunnelID, tunnelName, modeErr) - continue - } else if nftMode { - if len(entryNodeIDs) == 0 { - fail++ - failures = appendBatchFailureReason(failures, tunnelID, tunnelName, "nftables 转发缺少入口节点") - continue - } - if reconcileErr := h.reconcileNftablesNodeByRequest(entryNodeIDs[0]); reconcileErr != nil { - fail++ - failures = appendBatchFailure(failures, tunnelID, tunnelName, reconcileErr) - continue - } - success++ - continue - } - if err := h.redeployTunnelAndForwards(tunnelID); err != nil { - fail++ - failures = appendBatchFailure(failures, tunnelID, tunnelName, err) - continue - } - success++ } - response.WriteJSON(w, response.OK(batchOperationResult{SuccessCount: success, FailCount: fail, Failures: failures})) + response.WriteJSON(w, response.OK(result)) } func (h *Handler) userTunnelAssign(w http.ResponseWriter, r *http.Request) { diff --git a/go-backend/internal/http/handler/tunnel_redeploy_stream.go b/go-backend/internal/http/handler/tunnel_redeploy_stream.go new file mode 100644 index 0000000..3dbaa0f --- /dev/null +++ b/go-backend/internal/http/handler/tunnel_redeploy_stream.go @@ -0,0 +1,105 @@ +package handler + +import ( + "encoding/json" + "errors" + "net/http" + "time" + + "go-backend/internal/http/response" +) + +type redeployProgress struct { + Total int `json:"total"` + Completed int `json:"completed"` + Success int `json:"success"` + Failed int `json:"failed"` +} + +type redeployItemResult struct { + ID int64 `json:"id"` + Name string `json:"name"` + Success bool `json:"success"` + Reason string `json:"reason,omitempty"` +} + +// tunnelBatchRedeployStream reports completed tunnels, not estimated runtime. +// A disconnected caller may have an in-flight mutation whose result it did not +// receive. Finish that mutation, but never start another or automatically retry. +func (h *Handler) tunnelBatchRedeployStream(w http.ResponseWriter, r *http.Request) { + ids := idsFromBody(r, w) + if ids == nil { + return + } + // idsFromBody sorts IDs; ignore duplicate selections so each tunnel runs once. + unique := ids[:0] + for _, id := range ids { + if len(unique) == 0 || unique[len(unique)-1] != id { + unique = append(unique, id) + } + } + ids = unique + if len(ids) == 0 { + response.WriteJSON(w, response.ErrDefault("ids不能为空")) + return + } + _, err := prepareDiagnosisStreamResponse(w) + if err != nil { + response.WriteJSON(w, response.ErrDefault(err.Error())) + return + } + encoder := json.NewEncoder(w) + controller := http.NewResponseController(w) + emit := func(kind string, data interface{}) bool { + if r.Context().Err() != nil { + return false + } + // The server's WriteTimeout is an absolute request deadline. Renew only + // this stream's bounded write window as progress is emitted, so a long + // batch is not cut off despite making steady progress. Recorders and + // other non-network writers may not support deadlines. + if err := controller.SetWriteDeadline(time.Now().Add(2 * time.Minute)); err != nil && !errors.Is(err, http.ErrNotSupported) { + return false + } + event := diagnosisStreamEvent{Type: kind, Data: data, TS: time.Now().UnixMilli()} + if err := encoder.Encode(event); err != nil { + return false + } + // Buffered writes can succeed while the actual socket flush fails. + // Do not start a mutation after its processing event failed to flush. + return controller.Flush() == nil + } + progress := redeployProgress{Total: len(ids)} + result := batchOperationResult{} + if !emit("start", progress) { + return + } + for _, id := range ids { + if r.Context().Err() != nil { + return + } + name, _ := h.repo.GetTunnelName(id) + if !emit("processing", map[string]interface{}{"id": id, "name": name, "progress": progress}) { + return + } + if r.Context().Err() != nil { + return + } + item := redeployItemResult{ID: id, Name: name} + if err := h.redeployTunnelConfiguration(id); err != nil { + result.FailCount++ + result.Failures = appendBatchFailure(result.Failures, id, name, err) + item.Reason = result.Failures[len(result.Failures)-1].Reason + progress.Failed++ + } else { + result.SuccessCount++ + item.Success = true + progress.Success++ + } + progress.Completed++ + if !emit("item", map[string]interface{}{"result": item, "progress": progress}) { + return + } + } + emit("done", map[string]interface{}{"progress": progress, "result": result}) +} diff --git a/go-backend/internal/http/handler/tunnel_redeploy_stream_test.go b/go-backend/internal/http/handler/tunnel_redeploy_stream_test.go new file mode 100644 index 0000000..90a664e --- /dev/null +++ b/go-backend/internal/http/handler/tunnel_redeploy_stream_test.go @@ -0,0 +1,248 @@ +package handler + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "go-backend/internal/http/middleware" + runtimenft "go-backend/internal/runtime/nftables" +) + +type blockingRedeployManager struct { + fakeNftablesManager + entered chan struct{} + release chan struct{} + once sync.Once +} + +func (m *blockingRedeployManager) Reconcile(ctx context.Context, cfg runtimenft.SSHConfig, plan runtimenft.NodePlan) (runtimenft.ApplyResult, error) { + m.once.Do(func() { close(m.entered); <-m.release }) + return m.fakeNftablesManager.Reconcile(ctx, cfg, plan) +} + +func TestTunnelRedeployStreamFlushesBeforeSlowMutationAndReportsMixedResults(t *testing.T) { + fixture := setupNftablesHandler(t) + h := fixture.handler + seedNftablesSSHConfig(t, h, fixture.nodeID) + id := seedTunnelForNftables(t, h, "stream-tunnel", fixture.nodeID) + manager := &blockingRedeployManager{entered: make(chan struct{}), release: make(chan struct{})} + h.nftablesManager = manager + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(manager.release) }) } + defer release() + server := httptest.NewServer(http.HandlerFunc(h.tunnelBatchRedeployStream)) + defer server.Close() + client := &http.Client{Timeout: 5 * time.Second} + response, err := client.Post(server.URL, "application/json", strings.NewReader(fmt.Sprintf(`{"ids":[%d,999999,%d]}`, id, id))) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + decoder := json.NewDecoder(response.Body) + read := func(want string) map[string]interface{} { + t.Helper() + var event struct { + Type string + Data map[string]interface{} + } + if err := decoder.Decode(&event); err != nil { + t.Fatal(err) + } + if event.Type != want { + t.Fatalf("expected %s, got %s", want, event.Type) + } + return event.Data + } + if start := read("start"); start["total"] != float64(2) { + t.Fatalf("duplicate IDs not deduplicated: %+v", start) + } + processing := read("processing") + if processing["name"] != "stream-tunnel" || processing["id"] != float64(id) { + t.Fatalf("wrong current tunnel: %+v", processing) + } + select { + case <-manager.entered: + case <-time.After(2 * time.Second): + t.Fatal("did not start mutation") + } + // Both events arrived over HTTP while the actual mutation is still blocked. + release() + item := read("item") + if item["result"].(map[string]interface{})["success"] != true || item["progress"].(map[string]interface{})["completed"] != float64(1) { + t.Fatalf("wrong success result: %+v", item) + } + read("processing") + item = read("item") + if item["result"].(map[string]interface{})["success"] != false || item["result"].(map[string]interface{})["reason"] == "" { + t.Fatalf("wrong failure result: %+v", item) + } + done := read("done") + progress := done["progress"].(map[string]interface{}) + result := done["result"].(map[string]interface{}) + if progress["completed"] != float64(2) || progress["success"] != float64(1) || progress["failed"] != float64(1) || result["successCount"] != float64(1) || result["failCount"] != float64(1) { + t.Fatalf("incorrect final result: %+v", done) + } + if manager.reconcileCount() != 1 { + t.Fatalf("expected one execution, got %d", manager.reconcileCount()) + } + if err := decoder.Decode(new(interface{})); err != io.EOF { + t.Fatalf("expected EOF after done, got %v", err) + } +} + +func TestTunnelRedeployStreamStopsAfterDisconnect(t *testing.T) { + fixture := setupNftablesHandler(t) + h := fixture.handler + seedNftablesSSHConfig(t, h, fixture.nodeID) + first := seedTunnelForNftables(t, h, "first", fixture.nodeID) + second := seedTunnelForNftables(t, h, "second", fixture.nodeID) + manager := &blockingRedeployManager{entered: make(chan struct{}), release: make(chan struct{})} + h.nftablesManager = manager + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + req := httptest.NewRequest(http.MethodPost, "/api/v1/tunnel/batch-redeploy/stream", strings.NewReader(fmt.Sprintf(`{"ids":[%d,%d]}`, first, second))).WithContext(ctx) + res := httptest.NewRecorder() + finished := make(chan struct{}) + go func() { defer close(finished); h.tunnelBatchRedeployStream(res, req) }() + select { + case <-manager.entered: + case <-time.After(2 * time.Second): + close(manager.release) + t.Fatal("mutation did not start") + } + cancel() + close(manager.release) + select { + case <-finished: + case <-time.After(2 * time.Second): + t.Fatal("handler did not stop") + } + if manager.reconcileCount() != 1 { + t.Fatalf("started another mutation after disconnect: %d", manager.reconcileCount()) + } + if strings.Contains(res.Body.String(), `"type":"done"`) { + t.Fatal("disconnected stream claimed completion") + } +} + +type failedRedeployWriter struct{ *httptest.ResponseRecorder } + +func (w failedRedeployWriter) Write([]byte) (int, error) { return 0, io.ErrClosedPipe } + +func TestTunnelRedeployStreamDoesNotMutateWhenInitialWriteFails(t *testing.T) { + fixture := setupNftablesHandler(t) + h := fixture.handler + seedNftablesSSHConfig(t, h, fixture.nodeID) + id := seedTunnelForNftables(t, h, "stream-tunnel", fixture.nodeID) + manager := &fakeNftablesManager{} + h.nftablesManager = manager + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(fmt.Sprintf(`{"ids":[%d]}`, id))) + h.tunnelBatchRedeployStream(failedRedeployWriter{httptest.NewRecorder()}, req) + if manager.reconcileCount() != 0 { + t.Fatal("mutation ran despite failed progress stream") + } +} + +type delayedRedeployManager struct{ fakeNftablesManager } + +func (m *delayedRedeployManager) Reconcile(ctx context.Context, cfg runtimenft.SSHConfig, plan runtimenft.NodePlan) (runtimenft.ApplyResult, error) { + time.Sleep(100 * time.Millisecond) + return m.fakeNftablesManager.Reconcile(ctx, cfg, plan) +} + +type shortRedeployDeadlineWriter struct { + http.ResponseWriter + deadlineCalls *atomic.Int32 +} + +func (w shortRedeployDeadlineWriter) Unwrap() http.ResponseWriter { return w.ResponseWriter } +func (w shortRedeployDeadlineWriter) SetWriteDeadline(deadline time.Time) error { + remaining := time.Until(deadline) + if remaining < time.Minute || remaining > 2*time.Minute { + return fmt.Errorf("expected bounded two-minute write window, got %v", remaining) + } + w.deadlineCalls.Add(1) + // Shorten the production idle window for a real, fast timeout regression. + return http.NewResponseController(w.ResponseWriter).SetWriteDeadline(time.Now().Add(200 * time.Millisecond)) +} + +func TestTunnelRedeployStreamRenewsDeadlineThroughRequestLog(t *testing.T) { + fixture := setupNftablesHandler(t) + h := fixture.handler + seedNftablesSSHConfig(t, h, fixture.nodeID) + ids := make([]int64, 3) + for i := range ids { + ids[i] = seedTunnelForNftables(t, h, fmt.Sprintf("slow-%d", i), fixture.nodeID) + } + manager := &delayedRedeployManager{} + h.nftablesManager = manager + var deadlineCalls atomic.Int32 + logged := middleware.RequestLog(http.HandlerFunc(h.tunnelBatchRedeployStream)) + server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + logged.ServeHTTP(shortRedeployDeadlineWriter{w, &deadlineCalls}, r) + })) + server.Config.WriteTimeout = 50 * time.Millisecond + server.Start() + defer server.Close() + body, _ := json.Marshal(map[string]interface{}{"ids": ids}) + client := &http.Client{Timeout: 5 * time.Second} + response, err := client.Post(server.URL, "application/json", strings.NewReader(string(body))) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + payload, err := io.ReadAll(response.Body) + if err != nil { + t.Fatalf("stream timed out across batch: %v (body %s)", err, payload) + } + if !strings.Contains(string(payload), `"type":"done"`) { + t.Fatalf("batch did not finish: %s", payload) + } + if manager.reconcileCount() != 3 { + t.Fatalf("wrong execution count: %d", manager.reconcileCount()) + } + if deadlineCalls.Load() != 8 { + t.Fatalf("each of 8 events must renew its deadline, got %d", deadlineCalls.Load()) + } +} + +type flushFailureRedeployWriter struct { + *httptest.ResponseRecorder + flushes int +} + +func (w *flushFailureRedeployWriter) FlushError() error { + w.flushes++ + if w.flushes == 2 { + return io.ErrClosedPipe + } + w.ResponseRecorder.Flush() + return nil +} + +func TestTunnelRedeployStreamStopsBeforeMutationOnProcessingFlushError(t *testing.T) { + fixture := setupNftablesHandler(t) + h := fixture.handler + seedNftablesSSHConfig(t, h, fixture.nodeID) + id := seedTunnelForNftables(t, h, "flush-error", fixture.nodeID) + manager := &fakeNftablesManager{} + h.nftablesManager = manager + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(fmt.Sprintf(`{"ids":[%d]}`, id))) + writer := &flushFailureRedeployWriter{ResponseRecorder: httptest.NewRecorder()} + middleware.RequestLog(http.HandlerFunc(h.tunnelBatchRedeployStream)).ServeHTTP(writer, req) + if writer.flushes != 2 { + t.Fatalf("expected start and processing flush, got %d", writer.flushes) + } + if manager.reconcileCount() != 0 { + t.Fatal("mutation ran after failed processing flush") + } +} diff --git a/go-backend/internal/http/middleware/request_log.go b/go-backend/internal/http/middleware/request_log.go index 506f325..8ca53b9 100644 --- a/go-backend/internal/http/middleware/request_log.go +++ b/go-backend/internal/http/middleware/request_log.go @@ -14,6 +14,11 @@ type statusWriter struct { status int } +// Unwrap preserves ResponseController features such as per-response deadlines. +func (w *statusWriter) Unwrap() http.ResponseWriter { + return w.ResponseWriter +} + func (w *statusWriter) WriteHeader(code int) { w.status = code w.ResponseWriter.WriteHeader(code) @@ -27,10 +32,12 @@ func (w *statusWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) { return hj.Hijack() } +func (w *statusWriter) FlushError() error { + return http.NewResponseController(w.ResponseWriter).Flush() +} + func (w *statusWriter) Flush() { - if f, ok := w.ResponseWriter.(http.Flusher); ok { - f.Flush() - } + _ = w.FlushError() } func (w *statusWriter) ReadFrom(r io.Reader) (int64, error) { diff --git a/go-backend/tests/contract/tunnel_redeploy_stream_contract_test.go b/go-backend/tests/contract/tunnel_redeploy_stream_contract_test.go new file mode 100644 index 0000000..fe0c62f --- /dev/null +++ b/go-backend/tests/contract/tunnel_redeploy_stream_contract_test.go @@ -0,0 +1,98 @@ +package contract_test + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "go-backend/internal/auth" +) + +func TestTunnelRedeployStreamContract(t *testing.T) { + const secret = "contract-jwt-secret" + router, _ := setupContractRouter(t, secret) + token, err := auth.GenerateToken(1, "admin_user", 0, secret) + if err != nil { + t.Fatal(err) + } + req := httptest.NewRequest(http.MethodPost, "/api/v1/tunnel/batch-redeploy/stream", strings.NewReader(`{"ids":[999998,999999]}`)) + req.Header.Set("Authorization", token) + res := httptest.NewRecorder() + router.ServeHTTP(res, req) + if !strings.HasPrefix(res.Header().Get("Content-Type"), "application/x-ndjson") { + t.Fatalf("expected streamed progress, got %s: %s", res.Header().Get("Content-Type"), res.Body.String()) + } + decoder := json.NewDecoder(res.Body) + for _, want := range []string{"start", "processing", "item", "processing", "item", "done"} { + var event struct { + Type string `json:"type"` + Data json.RawMessage `json:"data"` + } + if err := decoder.Decode(&event); err != nil { + t.Fatal(err) + } + if event.Type != want { + t.Fatalf("got %s, want %s", event.Type, want) + } + if want == "item" { + var data struct{ Result map[string]interface{} } + if err := json.Unmarshal(event.Data, &data); err != nil { + t.Fatal(err) + } + if _, ok := data.Result["name"].(string); !ok { + t.Fatalf("item name must be a string, including missing tunnels: %s", event.Data) + } + } + if want == "done" { + var data struct { + Progress struct{ Total, Completed, Success, Failed int } + Result struct { + SuccessCount, FailCount int + Failures []struct { + ID int64 + Reason string + } + } + } + if err := json.Unmarshal(event.Data, &data); err != nil { + t.Fatal(err) + } + if data.Progress.Total != 2 || data.Progress.Completed != 2 || data.Progress.Failed != 2 || data.Progress.Success != 0 || data.Result.FailCount != 2 || len(data.Result.Failures) != 2 { + t.Fatalf("incorrect completion: %+v", data) + } + if data.Result.Failures[0].Reason == "" { + t.Fatal("missing failure detail") + } + } + } + if !res.Flushed { + t.Fatal("stream did not flush") + } +} + +func TestTunnelRedeployStreamAuthorizationContract(t *testing.T) { + const secret = "contract-jwt-secret" + router, repository := setupContractRouter(t, secret) + seedLegacyUser(t, repository, 9111, "stream-normal-user", "password") + token, err := auth.GenerateToken(9111, "stream-normal-user", 1, secret) + if err != nil { + t.Fatal(err) + } + for _, tc := range []struct { + name, token string + code int + }{{"anonymous", "", 401}, {"non-admin", token, 403}} { + t.Run(tc.name, func(t *testing.T) { + req := httptest.NewRequest(http.MethodPost, "/api/v1/tunnel/batch-redeploy/stream", strings.NewReader(`{"ids":[999999]}`)) + req.Header.Set("Authorization", tc.token) + res := httptest.NewRecorder() + router.ServeHTTP(res, req) + assertCode(t, res, tc.code) + if strings.Contains(res.Body.String(), `"type":"start"`) { + t.Fatal("unauthorized request started stream") + } + }) + } +} diff --git a/vite-frontend/nginx.conf b/vite-frontend/nginx.conf index dff20e4..060adfe 100644 --- a/vite-frontend/nginx.conf +++ b/vite-frontend/nginx.conf @@ -71,6 +71,20 @@ http { } # API代理到后端 + location = /api/v1/tunnel/batch-redeploy/stream { + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_http_version 1.1; + proxy_buffering off; + proxy_cache off; + chunked_transfer_encoding on; + proxy_read_timeout 120s; + proxy_send_timeout 120s; + proxy_pass http://backend:6365/api/v1/tunnel/batch-redeploy/stream; + } + location = /api/v1/tunnel/diagnose/stream { proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; diff --git a/vite-frontend/src/api/tunnel-redeploy-stream.ts b/vite-frontend/src/api/tunnel-redeploy-stream.ts new file mode 100644 index 0000000..c215c37 --- /dev/null +++ b/vite-frontend/src/api/tunnel-redeploy-stream.ts @@ -0,0 +1,287 @@ +import type { BatchOperationResult } from "@/api/types"; + +import axios from "axios"; + +import { clearSession, getToken } from "@/utils/session"; + +const STREAM_IDLE_TIMEOUT_MS = 2 * 60 * 1000; +const MAX_EVENT_LENGTH = 1024 * 1024; +const UNCERTAIN_RESULT = + "下发进度连接中断，部分配置可能已执行，结果尚未确认。请核对节点与隧道状态后再决定是否重新下发，系统不会自动重试。"; + +export interface TunnelRedeployProgress { + total: number; + completed: number; + success: number; + failed: number; +} + +export interface TunnelRedeployProcessing { + id: number; + name: string; + progress: TunnelRedeployProgress; +} + +export interface TunnelRedeployItem { + result: { + id: number; + name: string; + success: boolean; + reason?: string; + }; + progress: TunnelRedeployProgress; +} + +interface TunnelRedeployCallbacks { + onStart?: (progress: TunnelRedeployProgress) => void; + onProcessing?: (payload: TunnelRedeployProcessing) => void; + onItem?: (payload: TunnelRedeployItem) => void; +} + +type RawObject = Record; + +class RequestRejectedError extends Error {} + +const object = (value: unknown): RawObject => { + if (!value || typeof value !== "object" || Array.isArray(value)) { + throw new Error(UNCERTAIN_RESULT); + } + + return value as RawObject; +}; + +const readProgress = (value: unknown): TunnelRedeployProgress => { + const candidate = object(value); + const { total, completed, success, failed } = candidate; + + if ( + ![total, completed, success, failed].every( + (count) => + typeof count === "number" && Number.isSafeInteger(count) && count >= 0, + ) || + (completed as number) > (total as number) || + (success as number) + (failed as number) !== completed + ) { + throw new Error(UNCERTAIN_RESULT); + } + + return { total, completed, success, failed } as TunnelRedeployProgress; +}; + +const expireSession = () => { + clearSession(); + if (window.location.pathname !== "/") window.location.href = "/"; +}; + +/** A mutation stream must never be retried or replayed through the legacy API. */ +export const batchRedeployTunnelsStream = async ( + ids: number[], + callbacks: TunnelRedeployCallbacks, + signal?: AbortSignal, +): Promise => { + if (signal?.aborted) throw new DOMException("Aborted", "AbortError"); + if (!ids.length || ids.some((id) => !Number.isSafeInteger(id) || id <= 0)) { + throw new Error("请选择有效的隧道"); + } + if (typeof fetch !== "function" || typeof TextDecoder === "undefined") { + throw new Error("当前浏览器不支持实时下发进度，请使用支持流式响应的浏览器"); + } + + const requestedIds = new Set(ids); + const controller = new AbortController(); + const abort = () => controller.abort(); + let timer: ReturnType | undefined; + let reader: ReadableStreamDefaultReader | undefined; + const resetTimeout = () => { + clearTimeout(timer); + timer = setTimeout(abort, STREAM_IDLE_TIMEOUT_MS); + }; + + signal?.addEventListener("abort", abort, { once: true }); + resetTimeout(); + + try { + const base = axios.defaults.baseURL || "/api/v1/"; + const response = await fetch( + `${base.endsWith("/") ? base : `${base}/`}tunnel/batch-redeploy/stream`, + { + method: "POST", + headers: { + "Content-Type": "application/json", + Accept: "application/x-ndjson, application/json", + Authorization: getToken() || "", + }, + body: JSON.stringify({ ids: Array.from(requestedIds) }), + signal: controller.signal, + }, + ); + + if (response.status === 401) { + expireSession(); + throw new RequestRejectedError("登录已过期，请重新登录"); + } + if (response.status === 404 || response.status === 405) { + throw new RequestRejectedError( + "后端尚不支持实时下发进度，请升级后端后再操作；未自动改用旧接口下发", + ); + } + if ( + !response.headers.get("content-type")?.includes("application/x-ndjson") + ) { + let payload: RawObject | undefined; + + try { + payload = object(await response.json()); + } catch { + throw new Error(UNCERTAIN_RESULT); + } + if (payload.code === 401) { + expireSession(); + throw new RequestRejectedError("登录已过期，请重新登录"); + } + if (typeof payload.code === "number" && payload.code !== 0) { + throw new RequestRejectedError( + typeof payload.msg === "string" ? payload.msg : "下发请求失败", + ); + } + throw new Error(UNCERTAIN_RESULT); + } + if (!response.ok || !response.body) throw new Error(UNCERTAIN_RESULT); + + reader = response.body.getReader(); + const decoder = new TextDecoder("utf-8", { fatal: true }); + let buffer = ""; + let progress: TunnelRedeployProgress | undefined; + let currentId: number | undefined; + const completedIds = new Set(); + let finalResult: BatchOperationResult | undefined; + + const processLine = (line: string) => { + if (!line.trim()) return; + const event = object(JSON.parse(line)); + const data = object(event.data); + const next = readProgress(event.type === "start" ? data : data.progress); + + if ( + next.total !== requestedIds.size || + (progress && + (next.success < progress.success || next.failed < progress.failed)) + ) + throw new Error(UNCERTAIN_RESULT); + + if (event.type === "start") { + if (progress || next.completed !== 0) throw new Error(UNCERTAIN_RESULT); + progress = next; + callbacks.onStart?.(next); + + return; + } + if (!progress) throw new Error(UNCERTAIN_RESULT); + if (event.type === "processing") { + if ( + typeof data.id !== "number" || + !requestedIds.has(data.id) || + typeof data.name !== "string" || + completedIds.has(data.id) || + currentId !== undefined || + next.completed !== progress.completed + ) + throw new Error(UNCERTAIN_RESULT); + currentId = data.id; + callbacks.onProcessing?.({ + id: data.id, + name: data.name, + progress: next, + }); + } else if (event.type === "item") { + const result = object(data.result); + + if ( + result.id !== currentId || + currentId === undefined || + typeof result.name !== "string" || + typeof result.success !== "boolean" || + (result.reason !== undefined && typeof result.reason !== "string") || + next.completed !== progress.completed + 1 || + next.success !== progress.success + (result.success ? 1 : 0) || + next.failed !== progress.failed + (result.success ? 0 : 1) + ) + throw new Error(UNCERTAIN_RESULT); + completedIds.add(currentId); + currentId = undefined; + callbacks.onItem?.({ + result: result as unknown as TunnelRedeployItem["result"], + progress: next, + }); + } else if (event.type === "done") { + const result = object(data.result); + const failures = result.failures ?? []; + + if ( + currentId !== undefined || + next.completed !== next.total || + next.completed !== completedIds.size || + next.completed !== progress.completed || + result.successCount !== next.success || + result.failCount !== next.failed || + !Array.isArray(failures) || + failures.length !== next.failed || + failures.some((failure) => { + const item = object(failure); + + return ( + typeof item.id !== "number" || + !completedIds.has(item.id) || + typeof item.reason !== "string" || + (item.name !== undefined && typeof item.name !== "string") + ); + }) + ) + throw new Error(UNCERTAIN_RESULT); + finalResult = result as unknown as BatchOperationResult; + } else { + throw new Error(UNCERTAIN_RESULT); + } + progress = next; + }; + + while (!finalResult) { + const { value, done } = await reader.read(); + + if (done) { + buffer += decoder.decode(); + if (buffer.trim()) processLine(buffer); + break; + } + resetTimeout(); + buffer += decoder.decode(value, { stream: true }); + let newline: number; + + while ((newline = buffer.indexOf("\n")) >= 0) { + const line = buffer.slice(0, newline); + + buffer = buffer.slice(newline + 1); + if (line.length > MAX_EVENT_LENGTH) throw new Error(UNCERTAIN_RESULT); + processLine(line); + if (finalResult) break; + } + if (buffer.length > MAX_EVENT_LENGTH) throw new Error(UNCERTAIN_RESULT); + } + if (!finalResult) throw new Error(UNCERTAIN_RESULT); + + return finalResult; + } catch (error) { + if (signal?.aborted) throw new DOMException("Aborted", "AbortError"); + if (error instanceof RequestRejectedError) throw error; + throw new Error(UNCERTAIN_RESULT); + } finally { + clearTimeout(timer); + controller.abort(); + signal?.removeEventListener("abort", abort); + if (reader) { + // Cancel only the response reader, never resend an interrupted mutation. + await reader.cancel().catch(() => undefined); + reader.releaseLock(); + } + } +}; diff --git a/vite-frontend/src/components/server-order.tsx b/vite-frontend/src/components/server-order.tsx index cbe96aa..97d45ff 100644 --- a/vite-frontend/src/components/server-order.tsx +++ b/vite-frontend/src/components/server-order.tsx @@ -58,7 +58,7 @@ export function OrderItem({ {...attributes} {...listeners} aria-label={`拖动排序：${label}`} - className="shrink-0 rounded p-1 text-default-500 hover:bg-default-100 focus-visible:outline focus-visible:outline-2 focus-visible:outline-primary disabled:opacity-40 disabled:cursor-not-allowed cursor-grab active:cursor-grabbing" + className={`inline-flex items-center justify-center shrink-0 rounded text-default-500 hover:bg-default-100 focus-visible:outline focus-visible:outline-2 focus-visible:outline-primary disabled:opacity-40 disabled:cursor-not-allowed cursor-grab active:cursor-grabbing ${row ? "p-1" : "h-8 w-8 [@media(pointer:coarse)]:h-11 [@media(pointer:coarse)]:w-11"}`} disabled={disabled} style={{ touchAction: "none" }} type="button" diff --git a/vite-frontend/src/pages/limit.tsx b/vite-frontend/src/pages/limit.tsx index 5e45779..574976a 100644 --- a/vite-frontend/src/pages/limit.tsx +++ b/vite-frontend/src/pages/limit.tsx @@ -358,14 +358,18 @@ export default function LimitPage() { > - -
-
-

+
+
+ +

{rule.name}

+ Record >({}); const [infoPopoverPlacement, setInfoPopoverPlacement] = useState< @@ -1120,6 +1123,13 @@ export default function NodePage() { const node = nodeList.find((n) => n.id === upgradeTargetNodeId); if (!node) return; + setUpgradeProgress((prev) => { + const next = { ...prev }; + + delete next[upgradeTargetNodeId]; + + return next; + }); setNodeList((prev) => prev.map((n) => n.id === upgradeTargetNodeId ? { ...n, upgradeLoading: true } : n, @@ -1160,6 +1170,13 @@ export default function NodePage() { return; } + setUpgradeProgress((prev) => { + const next = { ...prev }; + + selectedLocalIds.forEach((id) => delete next[id]); + + return next; + }); setBatchUpgradeLoading(true); setUpgradeModalOpen(false); try { @@ -1956,6 +1973,11 @@ export default function NodePage() { {node.version || "-"} + {!isRemoteNode && ( + + )}
@@ -2261,19 +2283,9 @@ export default function NodePage() { {node.version || "未知"}
- {upgradeProgress[node.id] && - upgradeProgress[node.id].percent < 100 && ( -
- -
- )} +
开机时间 diff --git a/vite-frontend/src/pages/node/upgrade-progress.tsx b/vite-frontend/src/pages/node/upgrade-progress.tsx new file mode 100644 index 0000000..78fc52e --- /dev/null +++ b/vite-frontend/src/pages/node/upgrade-progress.tsx @@ -0,0 +1,46 @@ +import { Progress } from "@/shadcn-bridge/heroui/progress"; + +export interface NodeUpgradeProgressState { + stage: string; + percent: number; + message: string; +} + +const STAGE_LABELS: Record = { + downloading: "下载", + verifying: "校验", + installing: "安装 / 重启", +}; + +export function NodeUpgradeProgress({ + progress, +}: { + progress?: NodeUpgradeProgressState; +}) { + // The final installing event means the agent is restarting, not a durable + // job state. Do not leave a stale "重启中 100%" banner after it reconnects. + if ( + !progress || + (progress.stage === "installing" && progress.percent >= 100) + ) { + return null; + } + + const percent = Number.isFinite(progress.percent) + ? Math.min(100, Math.max(0, progress.percent)) + : 0; + const stage = STAGE_LABELS[progress.stage]; + + return ( +
+ +
+ ); +} diff --git a/vite-frontend/src/pages/tunnel.tsx b/vite-frontend/src/pages/tunnel.tsx index 393695b..bafbc13 100644 --- a/vite-frontend/src/pages/tunnel.tsx +++ b/vite-frontend/src/pages/tunnel.tsx @@ -64,7 +64,6 @@ import { getNodeList, diagnoseTunnel, updateTunnelOrder, - batchRedeployTunnels, previewBatchTunnelDelete, previewTunnelDelete, } from "@/api"; @@ -74,6 +73,11 @@ import { getDiagnosisQualityDisplay, type DiagnosisResult, } from "@/pages/tunnel/diagnosis"; +import { useTunnelRedeploy } from "@/pages/tunnel/use-tunnel-redeploy"; +import { + TunnelRedeployStatus, + TunnelRedeploySummary, +} from "@/pages/tunnel/redeploy-progress"; import { diagnoseTunnelStream } from "@/api/diagnosis-stream"; import { createTunnelFormDefaults, @@ -455,6 +459,7 @@ export default function TunnelPage() { number | null >(null); const [batchLoading, setBatchLoading] = useState(false); + const redeploy = useTunnelRedeploy(); const [batchProgress, setBatchProgress] = useState({ active: false, label: "", @@ -1351,55 +1356,39 @@ export default function TunnelPage() { }; const handleBatchRedeploy = async () => { - if (selectedIds.size === 0) return; - setBatchLoading(true); - setBatchProgress({ - active: true, - label: `正在重新下发 ${selectedIds.size} 条隧道...`, - percent: 30, - }); + if (selectedIds.size === 0 || batchLoading || redeploy.state?.running) + return; try { - const res = await batchRedeployTunnels(Array.from(selectedIds)); + const result = await redeploy.run(Array.from(selectedIds)); - if (res.code === 0) { - const result = res.data; - - if (result.failCount === 0) { - toast.success(`成功重新下发 ${result.successCount} 项`); - } else { - const failures = extractBatchFailures(result); - - if (failures.length > 0) { - openBatchResultModal( - "批量下发结果", - `成功 ${result.successCount} 项，失败 ${result.failCount} 项`, - failures, - ); - } else { - toast.error( - buildBatchFailureMessage( - result, - `成功 ${result.successCount} 项，失败 ${result.failCount} 项`, - ), - ); - } - } - setSelectedIds(new Set()); - setSelectMode(false); - setBatchProgress({ - active: true, - label: `重新下发完成：成功 ${result.successCount} 项，正在刷新列表...`, - percent: 100, - }); - await refreshTunnelList(false); + if (!result) return; + if (result.failCount === 0) { + toast.success(`成功重新下发 ${result.successCount} 项`); } else { - toast.error(res.msg || "下发失败"); + const failures = extractBatchFailures(result); + + if (failures.length > 0) { + openBatchResultModal( + "批量下发结果", + `成功 ${result.successCount} 项，失败 ${result.failCount} 项`, + failures, + ); + } else { + toast.error( + buildBatchFailureMessage( + result, + `成功 ${result.successCount} 项，失败 ${result.failCount} 项`, + ), + ); + } } + setSelectedIds(new Set()); + setSelectMode(false); + await refreshTunnelList(false); } catch (error) { - toast.error(extractApiErrorMessage(error, "下发失败")); - } finally { - setBatchProgress({ active: false, label: "", percent: 0 }); - setBatchLoading(false); + toast.error( + extractApiErrorMessage(error, "下发进度中断，请检查结果后再重试"), + ); } }; @@ -1707,7 +1696,11 @@ export default function TunnelPage() {
+
+ ); +} + +export function TunnelRedeploySummary({ + state, + onDismiss, +}: { + state: TunnelRedeployState | null; + onDismiss: () => void; +}) { + if (!state) return null; + + const { progress, running, error, currentName } = state; + const label = `${error ? "已确认" : "已完成"} ${progress.completed}/${progress.total} 条，成功 ${progress.success} 条，失败 ${progress.failed} 条`; + const percent = + progress.total > 0 ? (progress.completed / progress.total) * 100 : 0; + + return ( +
+
+
+

+ {running + ? "正在下发隧道配置" + : error + ? "下发进度已中断" + : "隧道配置下发结束"} +

+

{label}

+ {currentName &&

当前：{currentName}

} + {running && ( +

+ 进度按已完成的隧道统计，当前隧道内部步骤不计百分比。请保留此页面。 +

+ )} + {error &&

{error}

} +
+ {!running && ( + + )} +
+ 0 ? "warning" : "primary"} + size="sm" + value={percent} + /> +
+ ); +} diff --git a/vite-frontend/src/pages/tunnel/use-tunnel-redeploy.ts b/vite-frontend/src/pages/tunnel/use-tunnel-redeploy.ts new file mode 100644 index 0000000..5882d4d --- /dev/null +++ b/vite-frontend/src/pages/tunnel/use-tunnel-redeploy.ts @@ -0,0 +1,118 @@ +import { useEffect, useRef, useState } from "react"; + +import { + batchRedeployTunnelsStream, + type TunnelRedeployProgress, +} from "@/api/tunnel-redeploy-stream"; +import { extractApiErrorMessage } from "@/api/error-message"; + +export interface TunnelRedeployRow { + status: "queued" | "running" | "success" | "failed" | "unknown"; + reason?: string; +} + +export interface TunnelRedeployState { + running: boolean; + progress: TunnelRedeployProgress; + rows: Record; + currentName?: string; + error?: string; +} + +export function useTunnelRedeploy() { + const [state, setState] = useState(null); + const abortRef = useRef(null); + + useEffect(() => () => abortRef.current?.abort(), []); + + const run = async (ids: number[]) => { + // Guard synchronously, before React has rendered a disabled button. + if (abortRef.current || ids.length === 0) return null; + + const controller = new AbortController(); + + abortRef.current = controller; + setState({ + running: true, + progress: { total: ids.length, completed: 0, success: 0, failed: 0 }, + rows: Object.fromEntries(ids.map((id) => [id, { status: "queued" }])), + }); + try { + const result = await batchRedeployTunnelsStream( + ids, + { + onStart: (progress) => { + setState((prev) => prev && { ...prev, progress }); + }, + onProcessing: ({ id, name, progress }) => { + setState( + (prev) => + prev && { + ...prev, + progress, + currentName: name || `隧道 #${id}`, + rows: { ...prev.rows, [id]: { status: "running" } }, + }, + ); + }, + onItem: ({ result: item, progress }) => { + setState( + (prev) => + prev && { + ...prev, + progress, + currentName: undefined, + rows: { + ...prev.rows, + [item.id]: { + status: item.success ? "success" : "failed", + reason: item.reason, + }, + }, + }, + ); + }, + }, + controller.signal, + ); + + if (controller.signal.aborted) return null; + setState( + (prev) => prev && { ...prev, running: false, currentName: undefined }, + ); + + return result; + } catch (error) { + if (controller.signal.aborted) return null; + setState( + (prev) => + prev && { + ...prev, + running: false, + currentName: undefined, + error: extractApiErrorMessage( + error, + "下发进度中断，请检查未确认的结果后再重试", + ), + rows: Object.fromEntries( + Object.entries(prev.rows).map(([id, row]) => [ + id, + row.status === "queued" || row.status === "running" + ? { status: "unknown" } + : row, + ]), + ), + }, + ); + throw error; + } finally { + if (abortRef.current === controller) abortRef.current = null; + } + }; + + const clear = () => { + if (!abortRef.current) setState(null); + }; + + return { state, run, clear }; +} diff --git a/vite-frontend/src/pages/user.tsx b/vite-frontend/src/pages/user.tsx index aa62139..a148eb2 100644 --- a/vite-frontend/src/pages/user.tsx +++ b/vite-frontend/src/pages/user.tsx @@ -1316,25 +1316,30 @@ export default function UserPage() { > - -
-
+
+
{(user.name || user.user).slice(0, 2).toUpperCase()}
-
-

- {user.name || user.user} -

+
+
+ +

+ {user.name || user.user} +

+

@{user.user}

-
+