mirror of
https://github.com/Sagit-chu/flvx.git
synced 2026-11-13 00:14:22 +08:00
fix: show live node and tunnel deployment progress in list view
This commit is contained in:
@@ -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/update-order", h.tunnelUpdateOrder)
|
||||||
mux.HandleFunc("/api/v1/tunnel/batch-delete", h.tunnelBatchDelete)
|
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", 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/assign", h.userTunnelAssign)
|
||||||
mux.HandleFunc("/api/v1/tunnel/user/batch-assign", h.userTunnelBatchAssign)
|
mux.HandleFunc("/api/v1/tunnel/user/batch-assign", h.userTunnelBatchAssign)
|
||||||
mux.HandleFunc("/api/v1/tunnel/user/remove", h.userTunnelRemove)
|
mux.HandleFunc("/api/v1/tunnel/user/remove", h.userTunnelRemove)
|
||||||
|
|||||||
@@ -2046,51 +2046,37 @@ func normalizeBatchFailureReason(reason string) string {
|
|||||||
return trimmed
|
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) {
|
func (h *Handler) tunnelBatchRedeploy(w http.ResponseWriter, r *http.Request) {
|
||||||
ids := idsFromBody(r, w)
|
ids := idsFromBody(r, w)
|
||||||
if ids == nil {
|
if ids == nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
success := 0
|
result := batchOperationResult{}
|
||||||
fail := 0
|
|
||||||
failures := make([]batchFailureDetail, 0)
|
|
||||||
for _, tunnelID := range ids {
|
for _, tunnelID := range ids {
|
||||||
tunnel, tunnelErr := h.getTunnelRecord(tunnelID)
|
name, _ := h.repo.GetTunnelName(tunnelID)
|
||||||
tunnelName := ""
|
if err := h.redeployTunnelConfiguration(tunnelID); err != nil {
|
||||||
if tunnelErr == nil && tunnel != nil {
|
result.FailCount++
|
||||||
tunnelName, _ = h.repo.GetTunnelName(tunnelID)
|
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) {
|
func (h *Handler) userTunnelAssign(w http.ResponseWriter, r *http.Request) {
|
||||||
|
|||||||
@@ -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})
|
||||||
|
}
|
||||||
@@ -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")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -14,6 +14,11 @@ type statusWriter struct {
|
|||||||
status int
|
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) {
|
func (w *statusWriter) WriteHeader(code int) {
|
||||||
w.status = code
|
w.status = code
|
||||||
w.ResponseWriter.WriteHeader(code)
|
w.ResponseWriter.WriteHeader(code)
|
||||||
@@ -27,10 +32,12 @@ func (w *statusWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
|
|||||||
return hj.Hijack()
|
return hj.Hijack()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (w *statusWriter) FlushError() error {
|
||||||
|
return http.NewResponseController(w.ResponseWriter).Flush()
|
||||||
|
}
|
||||||
|
|
||||||
func (w *statusWriter) Flush() {
|
func (w *statusWriter) Flush() {
|
||||||
if f, ok := w.ResponseWriter.(http.Flusher); ok {
|
_ = w.FlushError()
|
||||||
f.Flush()
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (w *statusWriter) ReadFrom(r io.Reader) (int64, error) {
|
func (w *statusWriter) ReadFrom(r io.Reader) (int64, error) {
|
||||||
|
|||||||
@@ -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")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -71,6 +71,20 @@ http {
|
|||||||
}
|
}
|
||||||
|
|
||||||
# API代理到后端
|
# 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 {
|
location = /api/v1/tunnel/diagnose/stream {
|
||||||
proxy_set_header Host $host;
|
proxy_set_header Host $host;
|
||||||
proxy_set_header X-Real-IP $remote_addr;
|
proxy_set_header X-Real-IP $remote_addr;
|
||||||
|
|||||||
@@ -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<string, unknown>;
|
||||||
|
|
||||||
|
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<BatchOperationResult> => {
|
||||||
|
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<typeof setTimeout> | undefined;
|
||||||
|
let reader: ReadableStreamDefaultReader<Uint8Array> | 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<number>();
|
||||||
|
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();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
};
|
||||||
@@ -45,7 +45,6 @@ import { Chip } from "@/shadcn-bridge/heroui/chip";
|
|||||||
import { Switch } from "@/shadcn-bridge/heroui/switch";
|
import { Switch } from "@/shadcn-bridge/heroui/switch";
|
||||||
import { Spinner } from "@/shadcn-bridge/heroui/spinner";
|
import { Spinner } from "@/shadcn-bridge/heroui/spinner";
|
||||||
import { Alert } from "@/shadcn-bridge/heroui/alert";
|
import { Alert } from "@/shadcn-bridge/heroui/alert";
|
||||||
import { Progress } from "@/shadcn-bridge/heroui/progress";
|
|
||||||
import { Accordion, AccordionItem } from "@/shadcn-bridge/heroui/accordion";
|
import { Accordion, AccordionItem } from "@/shadcn-bridge/heroui/accordion";
|
||||||
import { Select, SelectItem } from "@/shadcn-bridge/heroui/select";
|
import { Select, SelectItem } from "@/shadcn-bridge/heroui/select";
|
||||||
import { Checkbox } from "@/shadcn-bridge/heroui/checkbox";
|
import { Checkbox } from "@/shadcn-bridge/heroui/checkbox";
|
||||||
@@ -70,6 +69,10 @@ import {
|
|||||||
getConnectionStatusMeta,
|
getConnectionStatusMeta,
|
||||||
getRemoteSyncErrorMessage,
|
getRemoteSyncErrorMessage,
|
||||||
} from "@/pages/node/display";
|
} from "@/pages/node/display";
|
||||||
|
import {
|
||||||
|
NodeUpgradeProgress,
|
||||||
|
type NodeUpgradeProgressState,
|
||||||
|
} from "@/pages/node/upgrade-progress";
|
||||||
import { tryCopyInstallCommand } from "@/pages/node/install-command";
|
import { tryCopyInstallCommand } from "@/pages/node/install-command";
|
||||||
import {
|
import {
|
||||||
getNodeRenewalSnapshot,
|
getNodeRenewalSnapshot,
|
||||||
@@ -422,7 +425,7 @@ export default function NodePage() {
|
|||||||
const [selectedVersion, setSelectedVersion] = useState("");
|
const [selectedVersion, setSelectedVersion] = useState("");
|
||||||
const [batchUpgradeLoading, setBatchUpgradeLoading] = useState(false);
|
const [batchUpgradeLoading, setBatchUpgradeLoading] = useState(false);
|
||||||
const [upgradeProgress, setUpgradeProgress] = useState<
|
const [upgradeProgress, setUpgradeProgress] = useState<
|
||||||
Record<number, { stage: string; percent: number; message: string }>
|
Record<number, NodeUpgradeProgressState>
|
||||||
>({});
|
>({});
|
||||||
|
|
||||||
const [infoPopoverPlacement, setInfoPopoverPlacement] = useState<
|
const [infoPopoverPlacement, setInfoPopoverPlacement] = useState<
|
||||||
@@ -1120,6 +1123,13 @@ export default function NodePage() {
|
|||||||
const node = nodeList.find((n) => n.id === upgradeTargetNodeId);
|
const node = nodeList.find((n) => n.id === upgradeTargetNodeId);
|
||||||
|
|
||||||
if (!node) return;
|
if (!node) return;
|
||||||
|
setUpgradeProgress((prev) => {
|
||||||
|
const next = { ...prev };
|
||||||
|
|
||||||
|
delete next[upgradeTargetNodeId];
|
||||||
|
|
||||||
|
return next;
|
||||||
|
});
|
||||||
setNodeList((prev) =>
|
setNodeList((prev) =>
|
||||||
prev.map((n) =>
|
prev.map((n) =>
|
||||||
n.id === upgradeTargetNodeId ? { ...n, upgradeLoading: true } : n,
|
n.id === upgradeTargetNodeId ? { ...n, upgradeLoading: true } : n,
|
||||||
@@ -1160,6 +1170,13 @@ export default function NodePage() {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
setUpgradeProgress((prev) => {
|
||||||
|
const next = { ...prev };
|
||||||
|
|
||||||
|
selectedLocalIds.forEach((id) => delete next[id]);
|
||||||
|
|
||||||
|
return next;
|
||||||
|
});
|
||||||
setBatchUpgradeLoading(true);
|
setBatchUpgradeLoading(true);
|
||||||
setUpgradeModalOpen(false);
|
setUpgradeModalOpen(false);
|
||||||
try {
|
try {
|
||||||
@@ -1956,6 +1973,11 @@ export default function NodePage() {
|
|||||||
<span className="text-sm text-default-500">
|
<span className="text-sm text-default-500">
|
||||||
{node.version || "-"}
|
{node.version || "-"}
|
||||||
</span>
|
</span>
|
||||||
|
{!isRemoteNode && (
|
||||||
|
<NodeUpgradeProgress
|
||||||
|
progress={upgradeProgress[node.id]}
|
||||||
|
/>
|
||||||
|
)}
|
||||||
</TableCell>
|
</TableCell>
|
||||||
<TableCell>
|
<TableCell>
|
||||||
<div className="flex flex-wrap items-center gap-1.5 min-w-max">
|
<div className="flex flex-wrap items-center gap-1.5 min-w-max">
|
||||||
@@ -2261,19 +2283,9 @@ export default function NodePage() {
|
|||||||
{node.version || "未知"}
|
{node.version || "未知"}
|
||||||
</span>
|
</span>
|
||||||
</div>
|
</div>
|
||||||
{upgradeProgress[node.id] &&
|
<NodeUpgradeProgress
|
||||||
upgradeProgress[node.id].percent < 100 && (
|
progress={upgradeProgress[node.id]}
|
||||||
<div className="mt-1">
|
/>
|
||||||
<Progress
|
|
||||||
showValueLabel
|
|
||||||
aria-label="升级进度"
|
|
||||||
color="warning"
|
|
||||||
label={upgradeProgress[node.id].message}
|
|
||||||
size="sm"
|
|
||||||
value={upgradeProgress[node.id].percent}
|
|
||||||
/>
|
|
||||||
</div>
|
|
||||||
)}
|
|
||||||
<div className="flex justify-between text-sm">
|
<div className="flex justify-between text-sm">
|
||||||
<span className="text-default-600">
|
<span className="text-default-600">
|
||||||
开机时间
|
开机时间
|
||||||
|
|||||||
@@ -0,0 +1,46 @@
|
|||||||
|
import { Progress } from "@/shadcn-bridge/heroui/progress";
|
||||||
|
|
||||||
|
export interface NodeUpgradeProgressState {
|
||||||
|
stage: string;
|
||||||
|
percent: number;
|
||||||
|
message: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
const STAGE_LABELS: Record<string, string> = {
|
||||||
|
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 (
|
||||||
|
<div className="mt-2 min-w-40 max-w-xs break-words" role="status">
|
||||||
|
<Progress
|
||||||
|
showValueLabel
|
||||||
|
aria-label={stage ? `升级${stage}进度` : "升级进度"}
|
||||||
|
color="warning"
|
||||||
|
label={progress.message || (stage ? `正在${stage}` : "正在升级")}
|
||||||
|
size="sm"
|
||||||
|
value={percent}
|
||||||
|
/>
|
||||||
|
</div>
|
||||||
|
);
|
||||||
|
}
|
||||||
@@ -64,7 +64,6 @@ import {
|
|||||||
getNodeList,
|
getNodeList,
|
||||||
diagnoseTunnel,
|
diagnoseTunnel,
|
||||||
updateTunnelOrder,
|
updateTunnelOrder,
|
||||||
batchRedeployTunnels,
|
|
||||||
previewBatchTunnelDelete,
|
previewBatchTunnelDelete,
|
||||||
previewTunnelDelete,
|
previewTunnelDelete,
|
||||||
} from "@/api";
|
} from "@/api";
|
||||||
@@ -74,6 +73,11 @@ import {
|
|||||||
getDiagnosisQualityDisplay,
|
getDiagnosisQualityDisplay,
|
||||||
type DiagnosisResult,
|
type DiagnosisResult,
|
||||||
} from "@/pages/tunnel/diagnosis";
|
} 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 { diagnoseTunnelStream } from "@/api/diagnosis-stream";
|
||||||
import {
|
import {
|
||||||
createTunnelFormDefaults,
|
createTunnelFormDefaults,
|
||||||
@@ -455,6 +459,7 @@ export default function TunnelPage() {
|
|||||||
number | null
|
number | null
|
||||||
>(null);
|
>(null);
|
||||||
const [batchLoading, setBatchLoading] = useState(false);
|
const [batchLoading, setBatchLoading] = useState(false);
|
||||||
|
const redeploy = useTunnelRedeploy();
|
||||||
const [batchProgress, setBatchProgress] = useState<BatchProgressState>({
|
const [batchProgress, setBatchProgress] = useState<BatchProgressState>({
|
||||||
active: false,
|
active: false,
|
||||||
label: "",
|
label: "",
|
||||||
@@ -1351,55 +1356,39 @@ export default function TunnelPage() {
|
|||||||
};
|
};
|
||||||
|
|
||||||
const handleBatchRedeploy = async () => {
|
const handleBatchRedeploy = async () => {
|
||||||
if (selectedIds.size === 0) return;
|
if (selectedIds.size === 0 || batchLoading || redeploy.state?.running)
|
||||||
setBatchLoading(true);
|
return;
|
||||||
setBatchProgress({
|
|
||||||
active: true,
|
|
||||||
label: `正在重新下发 ${selectedIds.size} 条隧道...`,
|
|
||||||
percent: 30,
|
|
||||||
});
|
|
||||||
try {
|
try {
|
||||||
const res = await batchRedeployTunnels(Array.from(selectedIds));
|
const result = await redeploy.run(Array.from(selectedIds));
|
||||||
|
|
||||||
if (res.code === 0) {
|
if (!result) return;
|
||||||
const result = res.data;
|
if (result.failCount === 0) {
|
||||||
|
toast.success(`成功重新下发 ${result.successCount} 项`);
|
||||||
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);
|
|
||||||
} else {
|
} 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) {
|
} catch (error) {
|
||||||
toast.error(extractApiErrorMessage(error, "下发失败"));
|
toast.error(
|
||||||
} finally {
|
extractApiErrorMessage(error, "下发进度中断,请检查结果后再重试"),
|
||||||
setBatchProgress({ active: false, label: "", percent: 0 });
|
);
|
||||||
setBatchLoading(false);
|
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -1707,7 +1696,11 @@ export default function TunnelPage() {
|
|||||||
</Button>
|
</Button>
|
||||||
<Button
|
<Button
|
||||||
color="danger"
|
color="danger"
|
||||||
isDisabled={selectedIds.size === 0}
|
isDisabled={
|
||||||
|
selectedIds.size === 0 ||
|
||||||
|
batchLoading ||
|
||||||
|
redeploy.state?.running
|
||||||
|
}
|
||||||
size="sm"
|
size="sm"
|
||||||
variant="flat"
|
variant="flat"
|
||||||
onPress={handleOpenBatchDeleteModal}
|
onPress={handleOpenBatchDeleteModal}
|
||||||
@@ -1716,8 +1709,12 @@ export default function TunnelPage() {
|
|||||||
</Button>
|
</Button>
|
||||||
<Button
|
<Button
|
||||||
color="primary"
|
color="primary"
|
||||||
isDisabled={selectedIds.size === 0}
|
isDisabled={
|
||||||
isLoading={batchLoading}
|
selectedIds.size === 0 ||
|
||||||
|
batchLoading ||
|
||||||
|
redeploy.state?.running
|
||||||
|
}
|
||||||
|
isLoading={batchLoading || redeploy.state?.running}
|
||||||
size="sm"
|
size="sm"
|
||||||
variant="flat"
|
variant="flat"
|
||||||
onPress={handleBatchRedeploy}
|
onPress={handleBatchRedeploy}
|
||||||
@@ -1789,6 +1786,11 @@ export default function TunnelPage() {
|
|||||||
</div>
|
</div>
|
||||||
)}
|
)}
|
||||||
|
|
||||||
|
<TunnelRedeploySummary
|
||||||
|
state={redeploy.state}
|
||||||
|
onDismiss={redeploy.clear}
|
||||||
|
/>
|
||||||
|
|
||||||
{/* 隧道卡片网格 */}
|
{/* 隧道卡片网格 */}
|
||||||
{tunnels.length > 0 ? (
|
{tunnels.length > 0 ? (
|
||||||
viewMode === "list" ? (
|
viewMode === "list" ? (
|
||||||
@@ -1860,6 +1862,9 @@ export default function TunnelPage() {
|
|||||||
{tunnel.name}
|
{tunnel.name}
|
||||||
</span>
|
</span>
|
||||||
</div>
|
</div>
|
||||||
|
<TunnelRedeployStatus
|
||||||
|
row={redeploy.state?.rows[tunnel.id]}
|
||||||
|
/>
|
||||||
</TableCell>
|
</TableCell>
|
||||||
<TableCell>
|
<TableCell>
|
||||||
<Chip
|
<Chip
|
||||||
@@ -1918,6 +1923,7 @@ export default function TunnelPage() {
|
|||||||
<div className="flex flex-wrap items-center gap-1.5 min-w-max">
|
<div className="flex flex-wrap items-center gap-1.5 min-w-max">
|
||||||
<Button
|
<Button
|
||||||
className="h-6 px-2 min-w-0 text-xs bg-indigo-50 text-indigo-600 hover:bg-indigo-100 dark:bg-indigo-950/30 dark:text-indigo-400"
|
className="h-6 px-2 min-w-0 text-xs bg-indigo-50 text-indigo-600 hover:bg-indigo-100 dark:bg-indigo-950/30 dark:text-indigo-400"
|
||||||
|
isDisabled={redeploy.state?.running}
|
||||||
size="sm"
|
size="sm"
|
||||||
variant="flat"
|
variant="flat"
|
||||||
onPress={() => handleEdit(tunnel)}
|
onPress={() => handleEdit(tunnel)}
|
||||||
@@ -1934,6 +1940,7 @@ export default function TunnelPage() {
|
|||||||
</Button>
|
</Button>
|
||||||
<Button
|
<Button
|
||||||
className="h-6 px-2 min-w-0 text-xs bg-rose-50 text-rose-600 hover:bg-rose-100 dark:bg-rose-950/30 dark:text-rose-400"
|
className="h-6 px-2 min-w-0 text-xs bg-rose-50 text-rose-600 hover:bg-rose-100 dark:bg-rose-950/30 dark:text-rose-400"
|
||||||
|
isDisabled={redeploy.state?.running}
|
||||||
size="sm"
|
size="sm"
|
||||||
variant="flat"
|
variant="flat"
|
||||||
onPress={() => handleDelete(tunnel)}
|
onPress={() => handleDelete(tunnel)}
|
||||||
@@ -2015,6 +2022,9 @@ export default function TunnelPage() {
|
|||||||
</CardHeader>
|
</CardHeader>
|
||||||
|
|
||||||
<CardBody className="pt-0 pb-3 md:pt-0 md:pb-3">
|
<CardBody className="pt-0 pb-3 md:pt-0 md:pb-3">
|
||||||
|
<TunnelRedeployStatus
|
||||||
|
row={redeploy.state?.rows[tunnel.id]}
|
||||||
|
/>
|
||||||
<div className="space-y-3">
|
<div className="space-y-3">
|
||||||
{/* 拓扑结构 */}
|
{/* 拓扑结构 */}
|
||||||
<div className="pt-2 border-t border-divider">
|
<div className="pt-2 border-t border-divider">
|
||||||
@@ -2160,6 +2170,7 @@ export default function TunnelPage() {
|
|||||||
<Button
|
<Button
|
||||||
className="flex-1 min-h-8"
|
className="flex-1 min-h-8"
|
||||||
color="primary"
|
color="primary"
|
||||||
|
isDisabled={redeploy.state?.running}
|
||||||
size="sm"
|
size="sm"
|
||||||
startContent={
|
startContent={
|
||||||
<svg
|
<svg
|
||||||
@@ -2202,6 +2213,7 @@ export default function TunnelPage() {
|
|||||||
<Button
|
<Button
|
||||||
className="flex-1 min-h-8"
|
className="flex-1 min-h-8"
|
||||||
color="danger"
|
color="danger"
|
||||||
|
isDisabled={redeploy.state?.running}
|
||||||
size="sm"
|
size="sm"
|
||||||
startContent={
|
startContent={
|
||||||
<svg
|
<svg
|
||||||
|
|||||||
@@ -0,0 +1,97 @@
|
|||||||
|
import type {
|
||||||
|
TunnelRedeployRow,
|
||||||
|
TunnelRedeployState,
|
||||||
|
} from "./use-tunnel-redeploy";
|
||||||
|
|
||||||
|
import { Button } from "@/shadcn-bridge/heroui/button";
|
||||||
|
import { Progress } from "@/shadcn-bridge/heroui/progress";
|
||||||
|
import { Spinner } from "@/shadcn-bridge/heroui/spinner";
|
||||||
|
|
||||||
|
export function TunnelRedeployStatus({ row }: { row?: TunnelRedeployRow }) {
|
||||||
|
if (!row) return null;
|
||||||
|
|
||||||
|
const text = {
|
||||||
|
queued: "等待下发",
|
||||||
|
running: "正在下发配置…",
|
||||||
|
success: "配置下发成功",
|
||||||
|
failed: "配置下发失败",
|
||||||
|
unknown: "下发结果未确认",
|
||||||
|
}[row.status];
|
||||||
|
const color =
|
||||||
|
row.status === "failed"
|
||||||
|
? "text-danger"
|
||||||
|
: row.status === "success"
|
||||||
|
? "text-success"
|
||||||
|
: row.status === "unknown"
|
||||||
|
? "text-warning"
|
||||||
|
: "text-default-500";
|
||||||
|
|
||||||
|
return (
|
||||||
|
<div className={`mt-1 flex items-start gap-1.5 text-xs ${color}`}>
|
||||||
|
{row.status === "running" && <Spinner size="sm" />}
|
||||||
|
<span className="max-w-xs break-words" title={row.reason}>
|
||||||
|
{text}
|
||||||
|
{row.reason ? `:${row.reason}` : ""}
|
||||||
|
</span>
|
||||||
|
</div>
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
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 (
|
||||||
|
<div
|
||||||
|
className="mb-4 space-y-2 rounded-xl border border-divider bg-content1 p-3"
|
||||||
|
role="status"
|
||||||
|
>
|
||||||
|
<div className="flex items-start justify-between gap-3">
|
||||||
|
<div className="min-w-0 space-y-1 break-words text-sm">
|
||||||
|
<p className="font-medium">
|
||||||
|
{running
|
||||||
|
? "正在下发隧道配置"
|
||||||
|
: error
|
||||||
|
? "下发进度已中断"
|
||||||
|
: "隧道配置下发结束"}
|
||||||
|
</p>
|
||||||
|
<p className="text-default-500">{label}</p>
|
||||||
|
{currentName && <p>当前:{currentName}</p>}
|
||||||
|
{running && (
|
||||||
|
<p className="text-xs text-default-500">
|
||||||
|
进度按已完成的隧道统计,当前隧道内部步骤不计百分比。请保留此页面。
|
||||||
|
</p>
|
||||||
|
)}
|
||||||
|
{error && <p className="text-warning">{error}</p>}
|
||||||
|
</div>
|
||||||
|
{!running && (
|
||||||
|
<Button
|
||||||
|
aria-label="关闭下发结果"
|
||||||
|
size="sm"
|
||||||
|
variant="light"
|
||||||
|
onPress={onDismiss}
|
||||||
|
>
|
||||||
|
关闭
|
||||||
|
</Button>
|
||||||
|
)}
|
||||||
|
</div>
|
||||||
|
<Progress
|
||||||
|
showValueLabel
|
||||||
|
aria-label="隧道配置下发进度"
|
||||||
|
color={error || progress.failed > 0 ? "warning" : "primary"}
|
||||||
|
size="sm"
|
||||||
|
value={percent}
|
||||||
|
/>
|
||||||
|
</div>
|
||||||
|
);
|
||||||
|
}
|
||||||
@@ -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<number, TunnelRedeployRow>;
|
||||||
|
currentName?: string;
|
||||||
|
error?: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function useTunnelRedeploy() {
|
||||||
|
const [state, setState] = useState<TunnelRedeployState | null>(null);
|
||||||
|
const abortRef = useRef<AbortController | null>(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 };
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user