From 58d2e891479edbd18d2071318fb60be524e28a4c Mon Sep 17 00:00:00 2001 From: sagit <36596628+Sagit-chu@users.noreply.github.com> Date: Sun, 26 Apr 2026 23:35:35 +0800 Subject: [PATCH] fix: reduce reconnect redeploy and metrics load (#476) * fix: reduce reconnect redeploy and metrics load Throttle node-online redeploy retries and lower the agent metric cadence so brief reconnect churn no longer fans out into repeated runtime syncs and backend connection pressure. * docs: add follow-up implementation design notes Document the planned flow upload batching work and the local remote-address toggle so the next changesets can implement them against an agreed design. --- ...26-04-26-flow-upload-batch-optimization.md | 849 ++++++++++++++++++ ...26-04-26-allow-local-remote-addr-design.md | 133 +++ go-backend/internal/http/handler/handler.go | 24 +- go-backend/internal/http/handler/upgrade.go | 124 ++- .../internal/http/handler/upgrade_test.go | 111 +++ go-backend/internal/store/repo/repository.go | 92 +- .../store/repo/repository_pool_test.go | 32 + go-gost/x/socket/websocket_reporter.go | 17 +- go-gost/x/socket/websocket_reporter_test.go | 8 + 9 files changed, 1324 insertions(+), 66 deletions(-) create mode 100644 docs/superpowers/plans/2026-04-26-flow-upload-batch-optimization.md create mode 100644 docs/superpowers/specs/2026-04-26-allow-local-remote-addr-design.md create mode 100644 go-backend/internal/http/handler/upgrade_test.go create mode 100644 go-backend/internal/store/repo/repository_pool_test.go diff --git a/docs/superpowers/plans/2026-04-26-flow-upload-batch-optimization.md b/docs/superpowers/plans/2026-04-26-flow-upload-batch-optimization.md new file mode 100644 index 0000000..e875570 --- /dev/null +++ b/docs/superpowers/plans/2026-04-26-flow-upload-batch-optimization.md @@ -0,0 +1,849 @@ +# flow/upload Batch Optimization Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Reduce `POST /flow/upload` database pressure by converting the hot path from per-item queries and per-item transactions to per-request aggregation, batched metadata reads, and batched writes, while preserving immediate quota disable / forward pause behavior inside the same upload. + +**Architecture:** Parse one upload into a batch object in the handler layer, fetch one shared `forward+tunnel` metadata map, then reuse that map for flow accounting and tunnel metric aggregation. Replace `AddFlow` and `AddUserQuotaUsage` per-item transactions with one batched flow transaction and one batched quota transaction; run policy enforcement, orphan cleanup, and peer-share flow handling once per affected target instead of once per item. + +**Tech Stack:** Go, net/http, GORM, SQLite/PostgreSQL, existing backend contract tests. + +--- + +## File Map + +- Create: `go-backend/internal/http/handler/flow_upload_batch.go` + Responsibility: request-scoped parsing, aggregation, and application of one `/flow/upload` batch. +- Create: `go-backend/internal/http/handler/flow_upload_batch_test.go` + Responsibility: unit coverage for batch aggregation semantics. +- Create: `go-backend/internal/store/repo/repository_flow_batch_test.go` + Responsibility: unit coverage for batched flow and quota persistence. +- Create: `go-backend/tests/contract/flow_upload_batch_contract_test.go` + Responsibility: contract coverage that repeated items still accumulate correctly and still disable quota immediately. +- Modify: `go-backend/internal/http/handler/handler.go` + Responsibility: switch `/flow/upload` entrypoint to the new batch pipeline. +- Modify: `go-backend/internal/http/handler/tunnel_metrics_ingestion.go` + Responsibility: accept pre-aggregated forward deltas plus shared forward metadata instead of reparsing the raw items. +- Modify: `go-backend/internal/store/repo/repository.go` + Responsibility: add batched flow persistence primitives near the existing flow update code. +- Modify: `go-backend/internal/store/repo/repository_flow.go` + Responsibility: add shared flow-upload metadata query helpers. +- Modify: `go-backend/internal/store/repo/repository_user_quota.go` + Responsibility: add batched quota usage persistence that still returns normalized quota views for immediate enforcement. + +--- + +### Task 1: Add Failing Tests For Batched flow/upload Semantics + +**Files:** +- Create: `go-backend/internal/http/handler/flow_upload_batch_test.go` +- Create: `go-backend/tests/contract/flow_upload_batch_contract_test.go` + +- [ ] **Step 1: Write the failing handler unit test** + +Create `go-backend/internal/http/handler/flow_upload_batch_test.go` with a unit test that locks in the new aggregation contract. + +```go +package handler + +import ( + "testing" + + "go-backend/internal/store/repo" +) + +func TestBuildFlowUploadBatchAggregatesForwardQuotaPeerShareAndCleanupTargets(t *testing.T) { + h := &Handler{} + metas := map[int64]repo.FlowUploadForwardMeta{ + 20: { + ForwardID: 20, + TunnelID: 1, + TrafficRatio: 2, + TunnelFlow: 3, + }, + } + + batch := h.buildFlowUploadBatch([]flowItem{ + {N: "20_2_10", U: 70, D: 50}, + {N: "20_2_10_tcp", U: 40, D: 30}, + {N: "99_2_10", U: 12, D: 8}, + {N: "fed_svc_17", U: 9, D: 1}, + }, metas) + + if len(batch.flowDeltas) != 1 { + t.Fatalf("expected 1 flow delta, got %d", len(batch.flowDeltas)) + } + delta := batch.flowDeltas[0] + if delta.ForwardID != 20 || delta.UserID != 2 || delta.UserTunnelID != 10 { + t.Fatalf("unexpected flow delta identity: %#v", delta) + } + if delta.InFlow != 480 || delta.OutFlow != 660 { + t.Fatalf("expected scaled flow in=480 out=660, got in=%d out=%d", delta.InFlow, delta.OutFlow) + } + if batch.quotaUsage[2] != 1140 { + t.Fatalf("expected quota usage 1140, got %d", batch.quotaUsage[2]) + } + if len(batch.policyTargets) != 1 { + t.Fatalf("expected 1 policy target, got %d", len(batch.policyTargets)) + } + if batch.policyTargets[0].UserID != 2 || batch.policyTargets[0].UserTunnelID != 10 { + t.Fatalf("unexpected policy target: %#v", batch.policyTargets[0]) + } + traffic := batch.forwardTraffic[20] + if traffic.bytesIn != 80 || traffic.bytesOut != 110 { + t.Fatalf("expected raw traffic in=80 out=110, got in=%d out=%d", traffic.bytesIn, traffic.bytesOut) + } + if _, ok := batch.orphanServices["99_2_10"]; !ok { + t.Fatalf("expected orphan service cleanup target for 99_2_10") + } + if item, ok := batch.peerShareForwardItems["20_2_10"]; !ok || item.U != 110 || item.D != 80 { + t.Fatalf("expected merged peer-share forward item, got %#v ok=%v", item, ok) + } + if item, ok := batch.peerShareRuntimeItems[17]; !ok || item.U != 9 || item.D != 1 { + t.Fatalf("expected merged peer-share runtime item, got %#v ok=%v", item, ok) + } +} +``` + +- [ ] **Step 2: Run the handler unit test to verify RED** + +Run: + +```bash +go test ./internal/http/handler -run TestBuildFlowUploadBatchAggregatesForwardQuotaPeerShareAndCleanupTargets -v +``` + +Expected: FAIL because `FlowUploadForwardMeta`, `buildFlowUploadBatch`, and the new batch fields do not exist yet. + +- [ ] **Step 3: Write the contract test that guards current behavior** + +Create `go-backend/tests/contract/flow_upload_batch_contract_test.go` so the optimization cannot weaken same-request quota enforcement. + +```go +package contract_test + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "go-backend/internal/store/model" +) + +func TestFlowUploadAggregatesRepeatedItemsAndDisablesQuotaImmediately(t *testing.T) { + secret := "monitoring-jwt-secret" + router, repo := setupContractRouter(t, secret) + now := time.Now() + nowMs := now.UnixMilli() + dayKey := int64(now.Year()*10000 + int(now.Month())*100 + now.Day()) + monthKey := int64(now.Year()*100 + int(now.Month())) + const bytesPerGB = int64(1024 * 1024 * 1024) + + node := &model.Node{Name: "node-1", Secret: "node-secret", ServerIP: "127.0.0.1", Port: "10000-10010", TCPListenAddr: "[::]", UDPListenAddr: "[::]", CreatedTime: nowMs, Status: 1} + if err := repo.DB().Create(node).Error; err != nil { + t.Fatalf("seed node: %v", err) + } + if err := repo.DB().Exec(`INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status) VALUES(2, 'flow_user', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)`, nowMs, nowMs).Error; err != nil { + t.Fatalf("insert user: %v", err) + } + tunnel := &model.Tunnel{Name: "tunnel-1", TrafficRatio: 1.0, Type: 1, Protocol: "tls", Flow: 1, CreatedTime: nowMs, UpdatedTime: nowMs, Status: 1} + if err := repo.DB().Create(tunnel).Error; err != nil { + t.Fatalf("seed tunnel: %v", err) + } + if err := repo.DB().Exec(`INSERT INTO user_tunnel(id, user_id, tunnel_id, speed_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status) VALUES(10, 2, ?, NULL, 99999, 99999, 0, 0, 1, 2727251700000, 1)`, tunnel.ID).Error; err != nil { + t.Fatalf("insert user_tunnel: %v", err) + } + forward := &model.Forward{ID: 20, UserID: 2, UserName: "flow_user", Name: "forward-20", TunnelID: tunnel.ID, RemoteAddr: "1.1.1.1:80", Strategy: "fifo", CreatedTime: nowMs, UpdatedTime: nowMs, Status: 1} + if err := repo.DB().Create(forward).Error; err != nil { + t.Fatalf("seed forward: %v", err) + } + if err := repo.DB().Exec(`INSERT INTO user_quota(user_id, daily_limit_gb, monthly_limit_gb, daily_used_bytes, monthly_used_bytes, day_key, month_key, disabled_by_quota, disabled_at, paused_forward_ids, created_time, updated_time) VALUES(2, 1, 0, ?, ?, ?, ?, 0, 0, '', ?, ?)`, bytesPerGB-100, bytesPerGB-100, dayKey, monthKey, nowMs, nowMs).Error; err != nil { + t.Fatalf("insert user_quota: %v", err) + } + + body, err := json.Marshal([]map[string]interface{}{ + {"n": "20_2_10", "u": 70, "d": 50}, + {"n": "20_2_10_tcp", "u": 40, "d": 30}, + }) + if err != nil { + t.Fatalf("marshal body: %v", err) + } + req := httptest.NewRequest(http.MethodPost, "/flow/upload?secret="+node.Secret, bytes.NewReader(body)) + res := httptest.NewRecorder() + router.ServeHTTP(res, req) + + if res.Code != http.StatusOK { + t.Fatalf("expected status 200, got %d", res.Code) + } + if got := mustQueryInt(t, repo, `SELECT status FROM forward WHERE id = 20`); got != 0 { + t.Fatalf("expected forward paused immediately, got status=%d", got) + } + if got := mustQueryInt(t, repo, `SELECT disabled_by_quota FROM user_quota WHERE user_id = 2`); got != 1 { + t.Fatalf("expected quota disabled flag=1, got %d", got) + } + if got := mustQueryInt(t, repo, `SELECT in_flow FROM forward WHERE id = 20`); got != 80 { + t.Fatalf("expected forward in_flow=80, got %d", got) + } + if got := mustQueryInt(t, repo, `SELECT out_flow FROM forward WHERE id = 20`); got != 110 { + t.Fatalf("expected forward out_flow=110, got %d", got) + } + metrics, err := repo.GetTunnelMetrics(tunnel.ID, 0, nowMs+60_000) + if err != nil { + t.Fatalf("get tunnel metrics: %v", err) + } + if len(metrics) != 1 || metrics[0].BytesIn != 80 || metrics[0].BytesOut != 110 { + t.Fatalf("expected one aggregated metric row, got %#v", metrics) + } +} +``` + +- [ ] **Step 4: Run the contract test to verify the same-request guard stays green or reveals an existing regression** + +Run: + +```bash +go test ./tests/contract/... -run TestFlowUploadAggregatesRepeatedItemsAndDisablesQuotaImmediately -v +``` + +Expected: this test may already PASS before the refactor because it locks in existing external behavior. Keep it either way; it is the guardrail for the optimization. + +- [ ] **Step 5: Optional commit if the user explicitly requested commits** + +```bash +git add go-backend/internal/http/handler/flow_upload_batch_test.go go-backend/tests/contract/flow_upload_batch_contract_test.go +git commit -m "test: cover flow upload batch semantics" +``` + +--- + +### Task 2: Add Batched Repository Primitives + +**Files:** +- Modify: `go-backend/internal/store/repo/repository_flow.go` +- Modify: `go-backend/internal/store/repo/repository.go` +- Modify: `go-backend/internal/store/repo/repository_user_quota.go` +- Create: `go-backend/internal/store/repo/repository_flow_batch_test.go` + +- [ ] **Step 1: Write the failing repository tests** + +Create `go-backend/internal/store/repo/repository_flow_batch_test.go` with coverage for both the shared metadata query and the batched counter/quota writes. + +```go +package repo + +import ( + "path/filepath" + "testing" + "time" +) + +func TestGetFlowUploadForwardMetasAndApplyFlowUploadDeltasBatch(t *testing.T) { + r, err := Open(filepath.Join(t.TempDir(), "flow-batch.db")) + if err != nil { + t.Fatalf("open repo: %v", err) + } + defer r.Close() + + now := time.Now().UnixMilli() + if err := r.DB().Exec(`INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status) VALUES(2, 'u2', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)`, now, now).Error; err != nil { + t.Fatalf("insert user: %v", err) + } + if err := r.DB().Exec(`INSERT INTO tunnel(id, name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx) VALUES(1, 't1', 2.0, 1, 'tls', 3, ?, ?, 1, NULL, 0)`, now, now).Error; err != nil { + t.Fatalf("insert tunnel: %v", err) + } + if err := r.DB().Exec(`INSERT INTO user_tunnel(id, user_id, tunnel_id, speed_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status) VALUES(10, 2, 1, NULL, 99999, 99999, 0, 0, 1, 2727251700000, 1)`).Error; err != nil { + t.Fatalf("insert user_tunnel: %v", err) + } + if err := r.DB().Exec(`INSERT INTO forward(id, user_id, user_name, name, tunnel_id, remote_addr, strategy, in_flow, out_flow, created_time, updated_time, status, inx) VALUES(20, 2, 'u2', 'f20', 1, '1.1.1.1:80', 'fifo', 0, 0, ?, ?, 1, 0)`, now, now).Error; err != nil { + t.Fatalf("insert forward: %v", err) + } + + metas, err := r.GetFlowUploadForwardMetas([]int64{20, 99}) + if err != nil { + t.Fatalf("get metas: %v", err) + } + if metas[20].TunnelID != 1 || metas[20].TrafficRatio != 2 || metas[20].TunnelFlow != 3 { + t.Fatalf("unexpected meta for forward 20: %#v", metas[20]) + } + if _, ok := metas[99]; ok { + t.Fatalf("did not expect meta for missing forward 99") + } + + err = r.ApplyFlowUploadDeltasBatch([]FlowUploadCounterDelta{{ForwardID: 20, UserID: 2, UserTunnelID: 10, InFlow: 480, OutFlow: 660}}) + if err != nil { + t.Fatalf("apply flow batch: %v", err) + } + if got := mustFlowBatchCount(t, r, `SELECT in_flow FROM forward WHERE id = 20`); got != 480 { + t.Fatalf("expected forward in_flow=480, got %d", got) + } + if got := mustFlowBatchCount(t, r, `SELECT out_flow FROM user WHERE id = 2`); got != 660 { + t.Fatalf("expected user out_flow=660, got %d", got) + } + if got := mustFlowBatchCount(t, r, `SELECT in_flow FROM user_tunnel WHERE id = 10`); got != 480 { + t.Fatalf("expected user_tunnel in_flow=480, got %d", got) + } +} + +func TestAddUserQuotaUsageBatchReturnsNormalizedViews(t *testing.T) { + r, err := Open(filepath.Join(t.TempDir(), "quota-batch.db")) + if err != nil { + t.Fatalf("open repo: %v", err) + } + defer r.Close() + + now := time.Now() + nowMs := now.UnixMilli() + if err := r.DB().Exec(`INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status) VALUES(2, 'u2', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)`, nowMs, nowMs).Error; err != nil { + t.Fatalf("insert user: %v", err) + } + views, err := r.AddUserQuotaUsageBatch(map[int64]int64{2: 1140}, now) + if err != nil { + t.Fatalf("batch quota update: %v", err) + } + if views[2] == nil || views[2].DailyUsedBytes != 1140 || views[2].MonthlyUsedBytes != 1140 { + t.Fatalf("unexpected quota view: %#v", views[2]) + } +} + +func mustFlowBatchCount(t *testing.T, r *Repository, query string, args ...interface{}) int64 { + t.Helper() + var value int64 + if err := r.DB().Raw(query, args...).Row().Scan(&value); err != nil { + t.Fatalf("query %q failed: %v", query, err) + } + return value +} +``` + +- [ ] **Step 2: Run the repository tests to verify RED** + +Run: + +```bash +go test ./internal/store/repo -run 'TestGetFlowUploadForwardMetasAndApplyFlowUploadDeltasBatch|TestAddUserQuotaUsageBatchReturnsNormalizedViews' -v +``` + +Expected: FAIL because `GetFlowUploadForwardMetas`, `ApplyFlowUploadDeltasBatch`, `FlowUploadCounterDelta`, and `AddUserQuotaUsageBatch` do not exist yet. + +- [ ] **Step 3: Implement shared flow-upload metadata and batched persistence** + +Update `go-backend/internal/store/repo/repository_flow.go`, `repository.go`, and `repository_user_quota.go` with the following concrete APIs. Add `sort` to the `repository_user_quota.go` import list. + +```go +// repository_flow.go +type FlowUploadForwardMeta struct { + ForwardID int64 + TunnelID int64 + TrafficRatio float64 + TunnelFlow int64 +} + +func (r *Repository) GetFlowUploadForwardMetas(forwardIDs []int64) (map[int64]FlowUploadForwardMeta, error) { + if r == nil || r.db == nil { + return nil, errors.New("repository not initialized") + } + if len(forwardIDs) == 0 { + return map[int64]FlowUploadForwardMeta{}, nil + } + ids := make([]int64, 0, len(forwardIDs)) + seen := make(map[int64]struct{}, len(forwardIDs)) + for _, id := range forwardIDs { + if id <= 0 { + continue + } + if _, ok := seen[id]; ok { + continue + } + seen[id] = struct{}{} + ids = append(ids, id) + } + type row struct { + ForwardID int64 `gorm:"column:forward_id"` + TunnelID int64 `gorm:"column:tunnel_id"` + TrafficRatio float64 `gorm:"column:traffic_ratio"` + TunnelFlow int64 `gorm:"column:tunnel_flow"` + } + var rows []row + err := r.db.Table("forward AS f"). + Select("f.id AS forward_id, f.tunnel_id AS tunnel_id, t.traffic_ratio AS traffic_ratio, t.flow AS tunnel_flow"). + Joins("JOIN tunnel t ON t.id = f.tunnel_id"). + Where("f.id IN ?", ids). + Scan(&rows).Error + if err != nil { + return nil, err + } + out := make(map[int64]FlowUploadForwardMeta, len(rows)) + for _, row := range rows { + if row.TunnelFlow <= 0 { + row.TunnelFlow = 1 + } + if row.TrafficRatio <= 0 { + row.TrafficRatio = 1 + } + out[row.ForwardID] = FlowUploadForwardMeta{ForwardID: row.ForwardID, TunnelID: row.TunnelID, TrafficRatio: row.TrafficRatio, TunnelFlow: row.TunnelFlow} + } + return out, nil +} +``` + +```go +// repository.go +type FlowUploadCounterDelta struct { + ForwardID int64 + UserID int64 + UserTunnelID int64 + InFlow int64 + OutFlow int64 +} + +func (r *Repository) ApplyFlowUploadDeltasBatch(deltas []FlowUploadCounterDelta) error { + if r == nil || r.db == nil { + return errors.New("repository not initialized") + } + if len(deltas) == 0 { + return nil + } + forwardTotals := make(map[int64][2]int64, len(deltas)) + userTotals := make(map[int64][2]int64, len(deltas)) + userTunnelTotals := make(map[int64][2]int64, len(deltas)) + for _, delta := range deltas { + if delta.ForwardID > 0 { + current := forwardTotals[delta.ForwardID] + current[0] += delta.InFlow + current[1] += delta.OutFlow + forwardTotals[delta.ForwardID] = current + } + if delta.UserID > 0 { + current := userTotals[delta.UserID] + current[0] += delta.InFlow + current[1] += delta.OutFlow + userTotals[delta.UserID] = current + } + if delta.UserTunnelID > 0 { + current := userTunnelTotals[delta.UserTunnelID] + current[0] += delta.InFlow + current[1] += delta.OutFlow + userTunnelTotals[delta.UserTunnelID] = current + } + } + return r.db.Transaction(func(tx *gorm.DB) error { + for forwardID, total := range forwardTotals { + if err := tx.Model(&model.Forward{}).Where("id = ?", forwardID).UpdateColumns(map[string]interface{}{"in_flow": gorm.Expr("in_flow + ?", total[0]), "out_flow": gorm.Expr("out_flow + ?", total[1])}).Error; err != nil { + return err + } + } + for userID, total := range userTotals { + if err := tx.Model(&model.User{}).Where("id = ?", userID).UpdateColumns(map[string]interface{}{"in_flow": gorm.Expr("in_flow + ?", total[0]), "out_flow": gorm.Expr("out_flow + ?", total[1])}).Error; err != nil { + return err + } + } + for userTunnelID, total := range userTunnelTotals { + if err := tx.Model(&model.UserTunnel{}).Where("id = ?", userTunnelID).UpdateColumns(map[string]interface{}{"in_flow": gorm.Expr("in_flow + ?", total[0]), "out_flow": gorm.Expr("out_flow + ?", total[1])}).Error; err != nil { + return err + } + } + return nil + }) +} +``` + +```go +// repository_user_quota.go +func (r *Repository) AddUserQuotaUsageBatch(usages map[int64]int64, now time.Time) (map[int64]*model.UserQuotaView, error) { + if r == nil || r.db == nil { + return nil, errors.New("repository not initialized") + } + if len(usages) == 0 { + return map[int64]*model.UserQuotaView{}, nil + } + result := make(map[int64]*model.UserQuotaView, len(usages)) + err := r.db.Transaction(func(tx *gorm.DB) error { + userIDs := make([]int64, 0, len(usages)) + for userID := range usages { + if userID > 0 { + userIDs = append(userIDs, userID) + } + } + sort.Slice(userIDs, func(i, j int) bool { return userIDs[i] < userIDs[j] }) + for _, userID := range userIDs { + q, err := r.loadOrCreateUserQuotaTx(tx, userID, now) + if err != nil { + return err + } + applyUserQuotaWindowRoll(q, now) + if usages[userID] > 0 { + q.DailyUsedBytes += usages[userID] + q.MonthlyUsedBytes += usages[userID] + } + q.UpdatedTime = now.UnixMilli() + if err := tx.Model(&model.UserQuota{}).Where("user_id = ?", userID).Updates(map[string]interface{}{"daily_used_bytes": q.DailyUsedBytes, "monthly_used_bytes": q.MonthlyUsedBytes, "day_key": q.DayKey, "month_key": q.MonthKey, "updated_time": q.UpdatedTime}).Error; err != nil { + return err + } + result[userID] = normalizeUserQuotaView(cloneUserQuotaView(*q), now) + } + return nil + }) + if err != nil { + return nil, err + } + return result, nil +} +``` + +- [ ] **Step 4: Run the repository tests to verify GREEN** + +Run: + +```bash +go test ./internal/store/repo -run 'TestGetFlowUploadForwardMetasAndApplyFlowUploadDeltasBatch|TestAddUserQuotaUsageBatchReturnsNormalizedViews' -v +``` + +Expected: PASS. + +- [ ] **Step 5: Optional commit if the user explicitly requested commits** + +```bash +git add go-backend/internal/store/repo/repository.go go-backend/internal/store/repo/repository_flow.go go-backend/internal/store/repo/repository_user_quota.go go-backend/internal/store/repo/repository_flow_batch_test.go +git commit -m "refactor: batch flow upload persistence" +``` + +--- + +### Task 3: Refactor flow/upload To Use One Parsed Batch + +**Files:** +- Create: `go-backend/internal/http/handler/flow_upload_batch.go` +- Modify: `go-backend/internal/http/handler/handler.go` +- Modify: `go-backend/internal/http/handler/tunnel_metrics_ingestion.go` +- Modify: `go-backend/internal/http/handler/flow_upload_batch_test.go` +- Modify: `go-backend/tests/contract/flow_upload_batch_contract_test.go` + +- [ ] **Step 1: Write the new handler batch implementation** + +Create `go-backend/internal/http/handler/flow_upload_batch.go` and move the request-scoped aggregation there. + +```go +package handler + +import ( + "log" + "sort" + "strings" + "time" + + "go-backend/internal/store/repo" +) + +type flowPolicyTarget struct { + UserID int64 + UserTunnelID int64 +} + +type flowUploadBatch struct { + flowDeltas []repo.FlowUploadCounterDelta + quotaUsage map[int64]int64 + policyTargets []flowPolicyTarget + forwardTraffic map[int64]tunnelTrafficDelta + orphanServices map[string]struct{} + peerShareForwardItems map[string]flowItem + peerShareRuntimeItems map[int64]flowItem +} + +func (h *Handler) buildFlowUploadBatch(items []flowItem, metas map[int64]repo.FlowUploadForwardMeta) flowUploadBatch { + batch := flowUploadBatch{ + quotaUsage: make(map[int64]int64), + forwardTraffic: make(map[int64]tunnelTrafficDelta), + orphanServices: make(map[string]struct{}), + peerShareForwardItems: make(map[string]flowItem), + peerShareRuntimeItems: make(map[int64]flowItem), + } + policySeen := map[flowPolicyTarget]struct{}{} + flowSeen := map[int64]int{} + + for _, item := range items { + serviceName := strings.TrimSpace(item.N) + if serviceName == "" || serviceName == "web_api" { + continue + } + if runtimeID, ok := parsePeerShareRuntimeServiceID(serviceName); ok { + merged := batch.peerShareRuntimeItems[runtimeID] + merged.N = serviceName + merged.U += item.U + merged.D += item.D + batch.peerShareRuntimeItems[runtimeID] = merged + continue + } + forwardID, userID, userTunnelID, ok := parseFlowServiceIDs(serviceName) + if !ok { + continue + } + meta, exists := metas[forwardID] + if !exists { + batch.orphanServices[serviceName] = struct{}{} + continue + } + raw := batch.forwardTraffic[forwardID] + raw.bytesIn += item.D + raw.bytesOut += item.U + batch.forwardTraffic[forwardID] = raw + + scaledIn := int64(float64(item.D)*meta.TrafficRatio) * meta.TunnelFlow + scaledOut := int64(float64(item.U)*meta.TrafficRatio) * meta.TunnelFlow + if idx, ok := flowSeen[forwardID]; ok { + batch.flowDeltas[idx].InFlow += scaledIn + batch.flowDeltas[idx].OutFlow += scaledOut + } else { + flowSeen[forwardID] = len(batch.flowDeltas) + batch.flowDeltas = append(batch.flowDeltas, repo.FlowUploadCounterDelta{ForwardID: forwardID, UserID: userID, UserTunnelID: userTunnelID, InFlow: scaledIn, OutFlow: scaledOut}) + } + batch.quotaUsage[userID] += scaledIn + scaledOut + target := flowPolicyTarget{UserID: userID, UserTunnelID: userTunnelID} + if _, seen := policySeen[target]; !seen { + policySeen[target] = struct{}{} + batch.policyTargets = append(batch.policyTargets, target) + } + merged := batch.peerShareForwardItems[normalizeForwardRuntimeServiceName(serviceName)] + merged.N = normalizeForwardRuntimeServiceName(serviceName) + merged.U += item.U + merged.D += item.D + batch.peerShareForwardItems[normalizeForwardRuntimeServiceName(serviceName)] = merged + } + + sort.Slice(batch.policyTargets, func(i, j int) bool { + if batch.policyTargets[i].UserID == batch.policyTargets[j].UserID { + return batch.policyTargets[i].UserTunnelID < batch.policyTargets[j].UserTunnelID + } + return batch.policyTargets[i].UserID < batch.policyTargets[j].UserID + }) + return batch +} + +func (h *Handler) applyFlowUploadBatch(nodeID int64, batch flowUploadBatch, now time.Time) { + if h == nil || h.repo == nil { + return + } + if err := h.repo.ApplyFlowUploadDeltasBatch(batch.flowDeltas); err != nil { + log.Printf("flow upload write failed op=flow.batch_apply node_id=%d err=%v", nodeID, err) + return + } + quotaViews, err := h.repo.AddUserQuotaUsageBatch(batch.quotaUsage, now) + if err != nil { + log.Printf("flow upload write failed op=quota.batch_apply node_id=%d err=%v", nodeID, err) + return + } + for userID, quota := range quotaViews { + h.enforceUserQuotaIfNeeded(userID, quota) + } + for _, target := range batch.policyTargets { + if target.UserID <= 0 || target.UserTunnelID <= 0 { + continue + } + h.enforceFlowPolicies(target.UserID, target.UserTunnelID) + } + for serviceName := range batch.orphanServices { + h.sendDeleteOrphanedForwardService(nodeID, serviceName) + } + for serviceName, item := range batch.peerShareForwardItems { + forwardID, _, _, ok := parseFlowServiceIDs(serviceName) + if ok { + h.processPeerShareFlowFromForward(forwardID, nodeID, serviceName, item) + } + } + for runtimeID, item := range batch.peerShareRuntimeItems { + h.processPeerShareFlow(runtimeID, item) + } +} +``` + +- [ ] **Step 2: Switch the `/flow/upload` entrypoint and tunnel metric ingestion to the shared batch** + +Modify `handler.go` and `tunnel_metrics_ingestion.go` so the raw JSON is parsed once and the same forward metadata powers both flow counters and tunnel metrics. + +```go +// handler.go +func (h *Handler) flowUpload(w http.ResponseWriter, r *http.Request) { + secret := r.URL.Query().Get("secret") + node, _ := h.repo.GetNodeBySecret(secret) + if node == nil { + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + _, _ = w.Write([]byte("ok")) + return + } + + raw, err := readAndDecryptFlowBody(r.Body, secret) + if err == nil && strings.TrimSpace(raw) != "" { + var items []flowItem + if json.Unmarshal([]byte(raw), &items) == nil { + now := time.Now() + forwardIDs := collectFlowUploadForwardIDs(items) + metas, metaErr := h.repo.GetFlowUploadForwardMetas(forwardIDs) + if metaErr != nil { + log.Printf("flow upload metadata lookup failed node_id=%d err=%v", node.ID, metaErr) + metas = map[int64]repo.FlowUploadForwardMeta{} + } + batch := h.buildFlowUploadBatch(items, metas) + h.recordTunnelMetricsFromForwardBatch(node.ID, batch.forwardTraffic, metas, now.UnixMilli()) + h.applyFlowUploadBatch(node.ID, batch, now) + } + } + + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + _, _ = w.Write([]byte("ok")) +} +``` + +```go +// tunnel_metrics_ingestion.go +func collectFlowUploadForwardIDs(items []flowItem) []int64 { + ids := make([]int64, 0, len(items)) + seen := make(map[int64]struct{}, len(items)) + for _, item := range items { + forwardID, _, _, ok := parseFlowServiceIDs(strings.TrimSpace(item.N)) + if !ok || forwardID <= 0 { + continue + } + if _, exists := seen[forwardID]; exists { + continue + } + seen[forwardID] = struct{}{} + ids = append(ids, forwardID) + } + return ids +} + +func (h *Handler) recordTunnelMetricsFromForwardBatch(nodeID int64, forwardDeltas map[int64]tunnelTrafficDelta, metas map[int64]repo.FlowUploadForwardMeta, nowMs int64) { + if h == nil || h.repo == nil || nodeID <= 0 || len(forwardDeltas) == 0 { + return + } + bucketTs := unixMilliBucketMinute(nowMs) + if bucketTs <= 0 { + return + } + tunnelAgg := make(map[int64]tunnelTrafficDelta) + for forwardID, delta := range forwardDeltas { + meta, ok := metas[forwardID] + if !ok || meta.TunnelID <= 0 { + continue + } + current := tunnelAgg[meta.TunnelID] + current.bytesIn += delta.bytesIn + current.bytesOut += delta.bytesOut + tunnelAgg[meta.TunnelID] = current + } + metrics := make([]*model.TunnelMetric, 0, len(tunnelAgg)) + for tunnelID, delta := range tunnelAgg { + if delta.bytesIn == 0 && delta.bytesOut == 0 { + continue + } + metrics = append(metrics, &model.TunnelMetric{TunnelID: tunnelID, NodeID: nodeID, Timestamp: bucketTs, BytesIn: delta.bytesIn, BytesOut: delta.bytesOut}) + } + if len(metrics) == 0 { + return + } + if err := h.repo.UpsertTunnelMetricBuckets(metrics); err != nil { + log.Printf("monitoring write failed op=tunnel_metric.upsert_buckets node_id=%d bucket_ts=%d count=%d err=%v", nodeID, bucketTs, len(metrics), err) + return + } + log.Printf("monitoring ok op=tunnel_metric.upsert_buckets node_id=%d bucket_ts=%d count=%d", nodeID, bucketTs, len(metrics)) +} +``` + +- [ ] **Step 3: Run focused handler and contract tests to verify GREEN** + +Run: + +```bash +go test ./internal/http/handler -run TestBuildFlowUploadBatchAggregatesForwardQuotaPeerShareAndCleanupTargets -v +go test ./tests/contract/... -run TestFlowUploadAggregatesRepeatedItemsAndDisablesQuotaImmediately -v +``` + +Expected: PASS. + +- [ ] **Step 4: Run the full backend suite** + +Run: + +```bash +go test ./... +``` + +Expected: PASS across the backend module. + +- [ ] **Step 5: Optional commit if the user explicitly requested commits** + +```bash +git add go-backend/internal/http/handler/handler.go go-backend/internal/http/handler/tunnel_metrics_ingestion.go go-backend/internal/http/handler/flow_upload_batch.go go-backend/internal/http/handler/flow_upload_batch_test.go go-backend/tests/contract/flow_upload_batch_contract_test.go +git commit -m "refactor: batch flow upload processing" +``` + +--- + +### Task 4: Final Verification And Performance Sanity Check + +**Files:** +- Modify: `go-backend/tests/contract/flow_upload_batch_contract_test.go` + +- [ ] **Step 1: Add a same-batch duplicate-item stress assertion** + +Extend the contract test with a second request that repeats the same service name multiple times and assert the counters advance by exactly the summed amount. + +```go +body, err = json.Marshal([]map[string]interface{}{ + {"n": "20_2_10", "u": 10, "d": 20}, + {"n": "20_2_10", "u": 10, "d": 20}, + {"n": "20_2_10_tcp", "u": 10, "d": 20}, +}) +if err != nil { + t.Fatalf("marshal body: %v", err) +} +req = httptest.NewRequest(http.MethodPost, "/flow/upload?secret="+node.Secret, bytes.NewReader(body)) +res = httptest.NewRecorder() +router.ServeHTTP(res, req) + +if got := mustQueryInt(t, repo, `SELECT in_flow FROM forward WHERE id = 20`); got != 140 { + t.Fatalf("expected forward in_flow=140 after second request, got %d", got) +} +if got := mustQueryInt(t, repo, `SELECT out_flow FROM forward WHERE id = 20`); got != 140 { + t.Fatalf("expected forward out_flow=140 after second request, got %d", got) +} +``` + +- [ ] **Step 2: Run the targeted contract test again** + +Run: + +```bash +go test ./tests/contract/... -run TestFlowUploadAggregatesRepeatedItemsAndDisablesQuotaImmediately -v +``` + +Expected: PASS. + +- [ ] **Step 3: Re-run the full backend suite before claiming completion** + +Run: + +```bash +go test ./... +``` + +Expected: PASS. + +- [ ] **Step 4: Optional local profiling sanity check** + +Run a short local comparison before and after the change with the same repeated flow payload. + +```bash +go test ./tests/contract/... -run TestFlowUploadAggregatesRepeatedItemsAndDisablesQuotaImmediately -count=10 +``` + +Expected: the test remains stable across repeated runs and does not introduce flakiness. + +- [ ] **Step 5: Optional commit if the user explicitly requested commits** + +```bash +git add go-backend/tests/contract/flow_upload_batch_contract_test.go +git commit -m "test: harden flow upload batch regression coverage" +``` diff --git a/docs/superpowers/specs/2026-04-26-allow-local-remote-addr-design.md b/docs/superpowers/specs/2026-04-26-allow-local-remote-addr-design.md new file mode 100644 index 0000000..a6566a4 --- /dev/null +++ b/docs/superpowers/specs/2026-04-26-allow-local-remote-addr-design.md @@ -0,0 +1,133 @@ +# 允许转发到本地地址开关设计 + +**日期**: 2026-04-26 +**状态**: 待审核 +**作者**: AI Assistant + +## 概述 + +新增一个全局设置开关，控制规则目标地址是否允许指向本地/内网地址。默认关闭，保持当前安全策略不变；开启后，规则创建和编辑时允许将目标地址设置为 `127.0.0.1`、`10.x.x.x`、`172.16-31.x.x`、`192.168.x.x` 等本地或私网地址。 + +## 背景 + +当前后端在规则创建和编辑时会调用 `IsSafeRemoteAddr()`，统一禁止目标地址指向本地/内网地址，用来降低 SSRF / 开放代理风险。这一行为是全局硬编码的，无法按部署场景调整。 + +有些用户需要把规则转发到本机或内网服务，因此需要一个显式、全局的开关来放宽这条限制。 + +## 目标 + +1. 在设置页提供一个全局开关控制该行为。 +2. 默认关闭，不改变现有安全默认值。 +3. 开启后，规则创建和编辑允许本地/内网目标地址。 +4. 不影响其他安全校验和其他业务流程。 + +## 影响范围 + +### 后端 +- `go-backend/internal/http/handler/security_utils.go` +- `go-backend/internal/http/handler/mutations.go` +- `go-backend/internal/http/handler/handler.go` + +### 前端 +- `vite-frontend/src/pages/config.tsx` + +### 测试 +- `go-backend/tests/contract/forward_contract_test.go` 或新增独立 contract test + +## 详细设计 + +### 1. 配置存储 + +使用现有 `vite_config` 表新增一个配置项： + +| name | value | 说明 | +|------|-------|------| +| `allow_local_remote_addr` | `"1"` / `"0"` | 是否允许规则目标地址指向本地/内网地址 | + +约定： +- 未配置时按 `"0"` 处理 +- `"1"` 表示允许 +- 其他值一律按关闭处理 + +### 2. 后端行为 + +新增一个轻量辅助函数，用于读取该配置开关： + +```go +func (h *Handler) allowLocalRemoteAddr() bool { + if h == nil || h.repo == nil { + return false + } + cfg, err := h.repo.GetConfigByName("allow_local_remote_addr") + if err != nil || cfg == nil { + return false + } + return strings.TrimSpace(cfg.Value) == "1" +} +``` + +在以下路径中应用： +- `forwardCreate` +- `forwardUpdate` + +行为改为： +- 当开关关闭时，继续执行 `IsSafeRemoteAddr(remoteAddr)` +- 当开关开启时，跳过这条“本地/内网地址禁止”校验 + +这样可以把改动范围限定在规则创建/编辑，不改变其他依赖 `IsSafeRemoteAddr()` 的场景。 + +### 3. 前端设置页 + +在 `vite-frontend/src/pages/config.tsx` 增加一个全局开关配置项。 + +建议文案： + +- 标签：`允许转发到本地地址` +- 描述：`开启后，规则目标地址可指向 127.0.0.1、10.x.x.x、172.16-31.x.x、192.168.x.x 等本地或内网地址。默认关闭以降低开放代理风险。` + +控件类型： +- 使用现有设置页的布尔开关模式 + +默认显示策略： +- 不依赖其他配置项 +- 直接显示在设置页的网络/安全相关区域；若现有页面没有单独分区，则先按现有配置项组织方式加入即可 + +### 4. 错误与兼容性 + +关闭开关时： +- 保持现有错误行为，继续阻止本地/内网地址 + +开启开关时： +- 仅放开“本地/内网地址禁止”这条限制 +- 仍保留地址格式解析失败等其他错误 + +### 5. 测试 + +需要补两类后端契约测试： + +1. 开关关闭时拒绝本地/内网地址 +- 创建规则时使用本地/内网地址 +- 断言接口返回非 0 code + +2. 开关开启时允许本地/内网地址 +- 先写入 `vite_config(name=allow_local_remote_addr, value=1)` +- 创建或更新规则时使用相同地址 +- 断言接口成功 + +建议至少覆盖： +- create 路径 +- update 路径 +- 多目标地址输入（逗号或换行分隔）中包含本地地址时的行为 + +## 风险与约束 + +1. 该开关会降低默认安全防护，应明确标注风险。 +2. 这是全局开关，不做用户级或规则级细分控制。 +3. 该开关只影响规则目标地址校验，不影响其他独立的安全策略。 + +## 推荐实施顺序 + +1. 先补失败的后端契约测试 +2. 实现后端配置读取与创建/更新分支控制 +3. 在设置页增加开关 +4. 跑后端测试与前端构建验证 diff --git a/go-backend/internal/http/handler/handler.go b/go-backend/internal/http/handler/handler.go index 3720603..5931b9c 100644 --- a/go-backend/internal/http/handler/handler.go +++ b/go-backend/internal/http/handler/handler.go @@ -44,8 +44,11 @@ type Handler struct { jobsStarted bool jobsWG sync.WaitGroup - upgradeMu sync.Mutex - pendingUpgradeRedeploy map[int64]struct{} + upgradeMu sync.Mutex + pendingUpgradeRedeploy map[int64]struct{} + nodeOnlineRedeployAt map[int64]time.Time + nodeOnlineRedeployQueued map[int64]struct{} + nodeOnlineRedeploying map[int64]struct{} qualityProber *tunnelQualityProber } @@ -97,13 +100,16 @@ const ( func New(repo *repo.Repository, jwtSecret string) *Handler { h := &Handler{ - repo: repo, - jwtSecret: jwtSecret, - wsServer: ws.NewServer(repo, jwtSecret), - metrics: metrics.NewIngestionService(repo), - healthCheck: nil, - captchaTokens: make(map[string]int64), - pendingUpgradeRedeploy: make(map[int64]struct{}), + repo: repo, + jwtSecret: jwtSecret, + wsServer: ws.NewServer(repo, jwtSecret), + metrics: metrics.NewIngestionService(repo), + healthCheck: nil, + captchaTokens: make(map[string]int64), + pendingUpgradeRedeploy: make(map[int64]struct{}), + nodeOnlineRedeployAt: make(map[int64]time.Time), + nodeOnlineRedeployQueued: make(map[int64]struct{}), + nodeOnlineRedeploying: make(map[int64]struct{}), } h.healthCheck = health.NewChecker(repo, h.wsServer) h.qualityProber = newTunnelQualityProber(h) diff --git a/go-backend/internal/http/handler/upgrade.go b/go-backend/internal/http/handler/upgrade.go index bf203f1..f7164b0 100644 --- a/go-backend/internal/http/handler/upgrade.go +++ b/go-backend/internal/http/handler/upgrade.go @@ -39,6 +39,8 @@ var ( testKeywordPattern = regexp.MustCompile(`(?i)(alpha|beta|rc)`) ) +const nodeOnlineRedeployCooldown = 30 * time.Second + type githubRelease struct { TagName string `json:"tag_name"` Name string `json:"name"` @@ -396,23 +398,120 @@ func (h *Handler) consumeNodePendingUpgradeRedeploy(nodeID int64) bool { } func (h *Handler) onNodeOnline(nodeID int64) { - h.consumeNodePendingUpgradeRedeploy(nodeID) - // Always redeploy rules on reconnection, not just for pending upgrade nodes. - // This handles cases where the node restarted and lost its in-memory config - // before persistence had time to flush, or if the panel also restarted. - h.redeployNodeRuntimeAfterUpgrade(nodeID) + if !h.startNodeOnlineRedeploy(nodeID, time.Now()) { + return + } + defer h.finishNodeOnlineRedeploy(nodeID) + + // Reconcile node runtime on the first reconnect, but suppress rapid flapping + // so websocket churn does not trigger repeated full redeploy storms. + if !h.redeployNodeRuntimeAfterUpgrade(nodeID) { + h.markNodePendingUpgradeRedeploy(nodeID) + } } -func (h *Handler) redeployNodeRuntimeAfterUpgrade(nodeID int64) { +func (h *Handler) startNodeOnlineRedeploy(nodeID int64, now time.Time) bool { + if h == nil || nodeID <= 0 { + return false + } + if now.IsZero() { + now = time.Now() + } + + h.upgradeMu.Lock() + defer h.upgradeMu.Unlock() + if h.pendingUpgradeRedeploy == nil { + h.pendingUpgradeRedeploy = make(map[int64]struct{}) + } + if h.nodeOnlineRedeployAt == nil { + h.nodeOnlineRedeployAt = make(map[int64]time.Time) + } + if h.nodeOnlineRedeployQueued == nil { + h.nodeOnlineRedeployQueued = make(map[int64]struct{}) + } + if h.nodeOnlineRedeploying == nil { + h.nodeOnlineRedeploying = make(map[int64]struct{}) + } + + _, pendingUpgrade := h.pendingUpgradeRedeploy[nodeID] + lastRedeployAt := h.nodeOnlineRedeployAt[nodeID] + _, inFlight := h.nodeOnlineRedeploying[nodeID] + if fireAt, start := nextNodeOnlineRedeployFireAt(lastRedeployAt, now, pendingUpgrade, inFlight); !start { + h.queueNodeOnlineRedeployLocked(nodeID, fireAt) + return false + } + + delete(h.pendingUpgradeRedeploy, nodeID) + h.nodeOnlineRedeployAt[nodeID] = now + h.nodeOnlineRedeploying[nodeID] = struct{}{} + return true +} + +func nextNodeOnlineRedeployFireAt(lastRedeployAt, now time.Time, pendingUpgrade bool, inFlight bool) (time.Time, bool) { + if now.IsZero() { + now = time.Now() + } + if inFlight { + fireAt := now.Add(nodeOnlineRedeployCooldown) + if !lastRedeployAt.IsZero() { + cooldownAt := lastRedeployAt.Add(nodeOnlineRedeployCooldown) + if cooldownAt.After(now) { + fireAt = cooldownAt + } + } + return fireAt, false + } + if !pendingUpgrade && !lastRedeployAt.IsZero() && now.Sub(lastRedeployAt) < nodeOnlineRedeployCooldown { + return lastRedeployAt.Add(nodeOnlineRedeployCooldown), false + } + return time.Time{}, true +} + +func (h *Handler) queueNodeOnlineRedeployLocked(nodeID int64, fireAt time.Time) { + if h == nil || nodeID <= 0 { + return + } + if h.nodeOnlineRedeployQueued == nil { + h.nodeOnlineRedeployQueued = make(map[int64]struct{}) + } + if _, queued := h.nodeOnlineRedeployQueued[nodeID]; queued { + return + } + if fireAt.IsZero() { + fireAt = time.Now().Add(nodeOnlineRedeployCooldown) + } + delay := time.Until(fireAt) + if delay < 0 { + delay = 0 + } + h.nodeOnlineRedeployQueued[nodeID] = struct{}{} + time.AfterFunc(delay, func() { + h.upgradeMu.Lock() + delete(h.nodeOnlineRedeployQueued, nodeID) + h.upgradeMu.Unlock() + h.onNodeOnline(nodeID) + }) +} + +func (h *Handler) finishNodeOnlineRedeploy(nodeID int64) { + if h == nil || nodeID <= 0 { + return + } + h.upgradeMu.Lock() + delete(h.nodeOnlineRedeploying, nodeID) + h.upgradeMu.Unlock() +} + +func (h *Handler) redeployNodeRuntimeAfterUpgrade(nodeID int64) bool { tunnelIDs, err := h.repo.ListActiveTunnelIDsByNode(nodeID) if err != nil { fmt.Printf("post-upgrade redeploy: list tunnels for node %d failed: %v\n", nodeID, err) - return + return false } forwardIDs, err := h.repo.ListForwardIDsByNode(nodeID) if err != nil { fmt.Printf("post-upgrade redeploy: list forwards for node %d failed: %v\n", nodeID, err) - return + return false } // First pass: deploy everything @@ -442,7 +541,7 @@ func (h *Handler) redeployNodeRuntimeAfterUpgrade(nodeID int64) { } // Retry failed items with exponential backoff (max 3 attempts) - h.retryFailedRedeploys(nodeID, tunnelFailed, failedForwards) + return h.retryFailedRedeploys(nodeID, tunnelFailed, failedForwards) } // isRetryableError returns true if the error looks transient and worth retrying. @@ -463,9 +562,9 @@ func isRetryableError(err error) bool { } // retryFailedRedeploys retries failed tunnels and forwards with exponential backoff. -func (h *Handler) retryFailedRedeploys(nodeID int64, tunnelFailed map[int64]struct{}, failedForwards []failedForward) { +func (h *Handler) retryFailedRedeploys(nodeID int64, tunnelFailed map[int64]struct{}, failedForwards []failedForward) bool { if len(tunnelFailed) == 0 && len(failedForwards) == 0 { - return + return true } const maxRetries = 3 @@ -507,7 +606,7 @@ func (h *Handler) retryFailedRedeploys(nodeID int64, tunnelFailed map[int64]stru if len(tunnelFailed) == 0 && len(failedForwards) == 0 { fmt.Printf("post-upgrade redeploy retry: all items recovered on node %d\n", nodeID) - return + return true } } @@ -518,4 +617,5 @@ func (h *Handler) retryFailedRedeploys(nodeID int64, tunnelFailed map[int64]stru for _, ff := range failedForwards { fmt.Printf("post-upgrade redeploy: forward %d permanently failed on node %d after retries\n", ff.id, nodeID) } + return false } diff --git a/go-backend/internal/http/handler/upgrade_test.go b/go-backend/internal/http/handler/upgrade_test.go new file mode 100644 index 0000000..abe8d33 --- /dev/null +++ b/go-backend/internal/http/handler/upgrade_test.go @@ -0,0 +1,111 @@ +package handler + +import ( + "testing" + "time" +) + +func TestStartNodeOnlineRedeploySkipsRecentReconnects(t *testing.T) { + h := &Handler{ + pendingUpgradeRedeploy: map[int64]struct{}{}, + nodeOnlineRedeployAt: map[int64]time.Time{}, + nodeOnlineRedeployQueued: map[int64]struct{}{}, + nodeOnlineRedeploying: map[int64]struct{}{}, + } + now := time.Unix(1_777_176_720, 0) + + if !h.startNodeOnlineRedeploy(54, now) { + t.Fatalf("expected first reconnect to redeploy") + } + h.finishNodeOnlineRedeploy(54) + + if h.startNodeOnlineRedeploy(54, now.Add(5*time.Second)) { + t.Fatalf("expected recent reconnect to skip redeploy") + } + if h.consumeNodePendingUpgradeRedeploy(54) { + t.Fatalf("did not expect pending upgrade marker to be consumed") + } +} + +func TestStartNodeOnlineRedeployAllowsPendingUpgradeDuringCooldown(t *testing.T) { + h := &Handler{ + pendingUpgradeRedeploy: map[int64]struct{}{}, + nodeOnlineRedeployAt: map[int64]time.Time{}, + nodeOnlineRedeployQueued: map[int64]struct{}{}, + nodeOnlineRedeploying: map[int64]struct{}{}, + } + now := time.Unix(1_777_176_720, 0) + + if !h.startNodeOnlineRedeploy(54, now) { + t.Fatalf("expected first reconnect to redeploy") + } + h.finishNodeOnlineRedeploy(54) + h.markNodePendingUpgradeRedeploy(54) + + if !h.startNodeOnlineRedeploy(54, now.Add(5*time.Second)) { + t.Fatalf("expected pending upgrade reconnect to bypass cooldown") + } + if h.consumeNodePendingUpgradeRedeploy(54) { + t.Fatalf("expected pending upgrade marker to be consumed during redeploy") + } +} + +func TestStartNodeOnlineRedeployQueuesCooldownReconnect(t *testing.T) { + h := &Handler{ + pendingUpgradeRedeploy: map[int64]struct{}{}, + nodeOnlineRedeployAt: map[int64]time.Time{}, + nodeOnlineRedeployQueued: map[int64]struct{}{}, + nodeOnlineRedeploying: map[int64]struct{}{}, + } + now := time.Unix(1_777_176_720, 0) + + if !h.startNodeOnlineRedeploy(54, now) { + t.Fatalf("expected first reconnect to redeploy") + } + h.finishNodeOnlineRedeploy(54) + + if h.startNodeOnlineRedeploy(54, now.Add(5*time.Second)) { + t.Fatalf("expected cooldown reconnect to skip immediate redeploy") + } + if _, queued := h.nodeOnlineRedeployQueued[54]; !queued { + t.Fatalf("expected cooldown reconnect to queue a follow-up redeploy") + } +} + +func TestStartNodeOnlineRedeployKeepsPendingUpgradeWhileInFlight(t *testing.T) { + h := &Handler{ + pendingUpgradeRedeploy: map[int64]struct{}{}, + nodeOnlineRedeployAt: map[int64]time.Time{}, + nodeOnlineRedeployQueued: map[int64]struct{}{}, + nodeOnlineRedeploying: map[int64]struct{}{}, + } + now := time.Unix(1_777_176_720, 0) + + if !h.startNodeOnlineRedeploy(54, now) { + t.Fatalf("expected first reconnect to redeploy") + } + h.markNodePendingUpgradeRedeploy(54) + + if h.startNodeOnlineRedeploy(54, now.Add(time.Second)) { + t.Fatalf("expected in-flight redeploy to suppress parallel restart") + } + if !h.consumeNodePendingUpgradeRedeploy(54) { + t.Fatalf("expected pending upgrade marker to remain for the next retry") + } + h.finishNodeOnlineRedeploy(54) +} + +func TestNextNodeOnlineRedeployFireAtDefersExpiredInFlightReconnect(t *testing.T) { + now := time.Unix(1_777_176_720, 0) + last := now.Add(-nodeOnlineRedeployCooldown - 5*time.Second) + + fireAt, start := nextNodeOnlineRedeployFireAt(last, now, false, true) + if start { + t.Fatalf("expected in-flight reconnect to queue instead of starting immediately") + } + + want := now.Add(nodeOnlineRedeployCooldown) + if !fireAt.Equal(want) { + t.Fatalf("expected queued reconnect at %s, got %s", want, fireAt) + } +} diff --git a/go-backend/internal/store/repo/repository.go b/go-backend/internal/store/repo/repository.go index b027dc9..92cafc8 100644 --- a/go-backend/internal/store/repo/repository.go +++ b/go-backend/internal/store/repo/repository.go @@ -22,6 +22,13 @@ import ( "go-backend/internal/store/model" ) +const ( + defaultPostgresMaxOpenConns = 32 + defaultPostgresMaxIdleConns = 8 + defaultPostgresConnMaxIdle = 5 * time.Minute + defaultPostgresConnMaxLife = 30 * time.Minute +) + // ─── Type aliases for backward compatibility ───────────────────────── // Handlers still reference repo.User, repo.BackupData, etc. @@ -210,6 +217,7 @@ func OpenPostgres(dsn string) (*Repository, error) { if err != nil { return nil, err } + configurePostgresPool(sqlDB) if err := sqlDB.Ping(); err != nil { _ = sqlDB.Close() return nil, err @@ -235,6 +243,16 @@ func OpenPostgres(dsn string) (*Repository, error) { return &Repository{db: db}, nil } +func configurePostgresPool(sqlDB *sql.DB) { + if sqlDB == nil { + return + } + sqlDB.SetMaxOpenConns(defaultPostgresMaxOpenConns) + sqlDB.SetMaxIdleConns(defaultPostgresMaxIdleConns) + sqlDB.SetConnMaxIdleTime(defaultPostgresConnMaxIdle) + sqlDB.SetConnMaxLifetime(defaultPostgresConnMaxLife) +} + func (r *Repository) Close() error { if r == nil || r.db == nil { return nil @@ -771,11 +789,11 @@ func (r *Repository) ListNodes() ([]map[string]interface{}, error) { "version": nullableString(n.Version), "http": n.HTTP, "tls": n.TLS, "socks": n.Socks, "status": n.Status, "isRemote": n.IsRemote, - "remoteUrl": nullableString(n.RemoteURL), - "remoteToken": nullableString(n.RemoteToken), - "remoteConfig": nullableString(n.RemoteConfig), - "expiryReminderDismissed": n.ExpiryReminderDismissed, - "interfaceName": nullableString(n.InterfaceName), + "remoteUrl": nullableString(n.RemoteURL), + "remoteToken": nullableString(n.RemoteToken), + "remoteConfig": nullableString(n.RemoteConfig), + "expiryReminderDismissed": n.ExpiryReminderDismissed, + "interfaceName": nullableString(n.InterfaceName), }) } return items, nil @@ -806,7 +824,7 @@ func (r *Repository) ListUsers() ([]map[string]interface{}, error) { "flowResetTime": u.FlowResetTime, "createdTime": u.CreatedTime, "updatedTime": nullableInt64(u.UpdatedTime), "inFlow": u.InFlow, "outFlow": u.OutFlow, - "maxConn": u.MaxConn, + "maxConn": u.MaxConn, } if quota := quotaMap[u.ID]; quota != nil { item["dailyQuotaGB"] = quota.DailyLimitGB @@ -847,22 +865,22 @@ func (r *Repository) ListForwards() ([]map[string]interface{}, error) { } type fwdRow struct { - ID int64 - UserID int64 - UserName string - Name string - TunnelID int64 - TunnelName string - TrafficRatio float64 - RemoteAddr string - Strategy string - InFlow int64 - OutFlow int64 - CreatedTime int64 - Status int - Inx int - SpeedID sql.NullInt64 - MaxConn int + ID int64 + UserID int64 + UserName string + Name string + TunnelID int64 + TunnelName string + TrafficRatio float64 + RemoteAddr string + Strategy string + InFlow int64 + OutFlow int64 + CreatedTime int64 + Status int + Inx int + SpeedID sql.NullInt64 + MaxConn int ProxyProtocol int } @@ -890,7 +908,7 @@ func (r *Repository) ListForwards() ([]map[string]interface{}, error) { "remoteAddr": row.RemoteAddr, "strategy": row.Strategy, "inFlow": row.InFlow, "outFlow": row.OutFlow, "createdTime": row.CreatedTime, "status": row.Status, "inx": int64(row.Inx), - "maxConn": row.MaxConn, + "maxConn": row.MaxConn, "proxyProtocol": row.ProxyProtocol, } if row.SpeedID.Valid { @@ -2469,19 +2487,19 @@ func importForwards(tx *gorm.DB, forwards []model.ForwardBackup, now int64) (int count := 0 for _, f := range forwards { item := model.Forward{ - ID: f.ID, - UserID: f.UserID, - UserName: f.UserName, - Name: f.Name, - TunnelID: f.TunnelID, - RemoteAddr: f.RemoteAddr, - Strategy: f.Strategy, - InFlow: f.InFlow, - OutFlow: f.OutFlow, - CreatedTime: f.CreatedTime, - UpdatedTime: now, - Status: f.Status, - Inx: f.Inx, + ID: f.ID, + UserID: f.UserID, + UserName: f.UserName, + Name: f.Name, + TunnelID: f.TunnelID, + RemoteAddr: f.RemoteAddr, + Strategy: f.Strategy, + InFlow: f.InFlow, + OutFlow: f.OutFlow, + CreatedTime: f.CreatedTime, + UpdatedTime: now, + Status: f.Status, + Inx: f.Inx, ProxyProtocol: f.ProxyProtocol, } err := tx.Clauses(clause.OnConflict{ @@ -3436,7 +3454,7 @@ func (r *Repository) GetNodeMetrics(nodeID int64, startMs, endMs int64) ([]model rangeMs := endMs - startMs const maxRawRangeMs = int64(60 * 60 * 1000) // 1 hour — return raw data for short ranges - const targetPoints = 500 // target number of chart points for downsampled data + const targetPoints = 500 // target number of chart points for downsampled data // For short ranges, return raw data (full resolution). if rangeMs <= maxRawRangeMs { diff --git a/go-backend/internal/store/repo/repository_pool_test.go b/go-backend/internal/store/repo/repository_pool_test.go new file mode 100644 index 0000000..b734531 --- /dev/null +++ b/go-backend/internal/store/repo/repository_pool_test.go @@ -0,0 +1,32 @@ +package repo + +import ( + "testing" + + gsqlite "github.com/glebarez/sqlite" + "gorm.io/gorm" + "gorm.io/gorm/logger" +) + +func TestConfigurePostgresPoolSetsMaxOpenConnections(t *testing.T) { + db, err := gorm.Open(gsqlite.Open(":memory:"), &gorm.Config{ + Logger: logger.Default.LogMode(logger.Silent), + }) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + + sqlDB, err := db.DB() + if err != nil { + t.Fatalf("db handle: %v", err) + } + t.Cleanup(func() { + _ = sqlDB.Close() + }) + + configurePostgresPool(sqlDB) + + if got := sqlDB.Stats().MaxOpenConnections; got != defaultPostgresMaxOpenConns { + t.Fatalf("expected max open conns %d, got %d", defaultPostgresMaxOpenConns, got) + } +} diff --git a/go-gost/x/socket/websocket_reporter.go b/go-gost/x/socket/websocket_reporter.go index 7aba916..c213eca 100644 --- a/go-gost/x/socket/websocket_reporter.go +++ b/go-gost/x/socket/websocket_reporter.go @@ -145,11 +145,12 @@ type ServiceMonitorCheckResult struct { } const ( - reporterReadWait = 60 * time.Second - reporterWriteWait = 5 * time.Second - wsPingInterval = 20 * time.Second // 独立 WebSocket ping 间隔 - initialBackoff = 2 * time.Second // 重连初始退避 - maxBackoff = 2 * time.Minute // 重连最大退避 + reporterReadWait = 60 * time.Second + reporterWriteWait = 5 * time.Second + wsPingInterval = 20 * time.Second // 独立 WebSocket ping 间隔 + initialBackoff = 2 * time.Second // 重连初始退避 + maxBackoff = 2 * time.Minute // 重连最大退避 + defaultMetricReportInterval = 5 * time.Second ) type WebSocketReporter struct { @@ -189,9 +190,9 @@ func NewWebSocketReporter(serverURL string, secret string) *WebSocketReporter { return &WebSocketReporter{ url: serverURL, - curBackoff: initialBackoff, // 当前退避间隔 - pingInterval: 1 * time.Second, // 指标上报间隔（每秒采集） - configInterval: 10 * time.Minute, // 配置上报间隔 + curBackoff: initialBackoff, // 当前退避间隔 + pingInterval: defaultMetricReportInterval, // 指标上报间隔 + configInterval: 10 * time.Minute, // 配置上报间隔 ctx: ctx, cancel: cancel, connected: false, diff --git a/go-gost/x/socket/websocket_reporter_test.go b/go-gost/x/socket/websocket_reporter_test.go index aa2639e..1587510 100644 --- a/go-gost/x/socket/websocket_reporter_test.go +++ b/go-gost/x/socket/websocket_reporter_test.go @@ -110,6 +110,14 @@ func TestSanitizeWebSocketURL(t *testing.T) { } } +func TestNewWebSocketReporterUsesReducedMetricInterval(t *testing.T) { + reporter := NewWebSocketReporter("panel.example.com:443", "abc") + + if reporter.pingInterval != defaultMetricReportInterval { + t.Fatalf("expected metric interval %s, got %s", defaultMetricReportInterval, reporter.pingInterval) + } +} + func TestFormatWebSocketDialErrorIncludesHTTPStatus(t *testing.T) { err := errors.New("websocket: bad handshake") resp := &http.Response{