diff --git a/internal/web/service/client_locks.go b/internal/web/service/client_locks.go index 8a964896e..9a299688f 100644 --- a/internal/web/service/client_locks.go +++ b/internal/web/service/client_locks.go @@ -1,6 +1,7 @@ package service import ( + "encoding/json" "sync" "time" @@ -139,3 +140,39 @@ func isClientEmailTombstoned(email string) bool { } return true } + +// stripTombstonedClients drops just-deleted client entries from a node +// snapshot's settings JSON so adopting a stale snapshot can't re-add them to +// the central inbound while the delete tombstone is live. Returns the filtered +// JSON and whether anything was removed. +func stripTombstonedClients(settings string) (string, bool) { + if settings == "" { + return settings, false + } + var parsed map[string]any + if err := json.Unmarshal([]byte(settings), &parsed); err != nil { + return settings, false + } + clients, _ := parsed["clients"].([]any) + if len(clients) == 0 { + return settings, false + } + kept := make([]any, 0, len(clients)) + for _, c := range clients { + if cm, ok := c.(map[string]any); ok { + if email, _ := cm["email"].(string); email != "" && isClientEmailTombstoned(email) { + continue + } + } + kept = append(kept, c) + } + if len(kept) == len(clients) { + return settings, false + } + parsed["clients"] = kept + b, err := json.MarshalIndent(parsed, "", " ") + if err != nil { + return settings, false + } + return string(b), true +} diff --git a/internal/web/service/inbound_client_ips.go b/internal/web/service/inbound_client_ips.go index 95d12230d..1b54a69e3 100644 --- a/internal/web/service/inbound_client_ips.go +++ b/internal/web/service/inbound_client_ips.go @@ -79,6 +79,14 @@ func (s *InboundService) MergeInboundClientIps(incomingIps []model.InboundClient now := time.Now().Unix() cutoff := now - clientIpStaleAfterSeconds + // Node syncs run concurrently (one goroutine per node) and shared clients + // appear in several nodes' reports. Locking rows in each node's arbitrary + // report order lets two merges grab the same rows in opposite order, which + // Postgres aborts as a deadlock (40P01) — take them in one global order. + sort.Slice(incomingIps, func(i, j int) bool { + return incomingIps[i].ClientEmail < incomingIps[j].ClientEmail + }) + tx := db.Begin() defer func() { if r := recover(); r != nil { diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index 180603b83..af46d001c 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -486,6 +486,14 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi inGrace := c.LastTrafficResetTime > 0 && now-c.LastTrafficResetTime < resetGracePeriodMs + // Adopting the node's settings verbatim would re-add a client the master + // deleted moments ago if this snapshot was fetched before the deletion + // push landed — filter just-deleted emails out while their tombstone lives. + adoptedSettings := snapIb.Settings + if stripped, changed := stripTombstonedClients(adoptedSettings); changed { + adoptedSettings = stripped + } + updates := map[string]any{} if !dirty { updates["enable"] = snapIb.Enable @@ -496,7 +504,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi updates["protocol"] = snapIb.Protocol updates["total"] = snapIb.Total updates["expiry_time"] = snapIb.ExpiryTime - updates["settings"] = snapIb.Settings + updates["settings"] = adoptedSettings updates["stream_settings"] = snapIb.StreamSettings updates["sniffing"] = snapIb.Sniffing updates["traffic_reset"] = snapIb.TrafficReset @@ -513,7 +521,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi updates["origin_node_guid"] = og } - if !dirty && (c.Settings != snapIb.Settings || + if !dirty && (c.Settings != adoptedSettings || c.Remark != snapIb.Remark || c.Listen != snapIb.Listen || c.Port != snapIb.Port || @@ -634,8 +642,17 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi if dirty { continue } + _, isNewInbound := newInboundIDs[c.Id] + // On a known inbound a missing row plus a live tombstone means the + // master just deleted this client and the snapshot predates the + // deletion push — recreating the row (at zero) would resurrect the + // client. A freshly adopted inbound still gets its row (seeded at + // zero) so adoption semantics stay intact. + if !isNewInbound && isClientEmailTombstoned(cs.Email) { + continue + } var seedUp, seedDown int64 - if _, isNewInbound := newInboundIDs[c.Id]; isNewInbound && !isClientEmailTombstoned(cs.Email) { + if isNewInbound && !isClientEmailTombstoned(cs.Email) { seedUp, seedDown = canon.Up, canon.Down } row := &xray.ClientTraffic{ diff --git a/internal/web/service/inbound_node_ips.go b/internal/web/service/inbound_node_ips.go index 13950db15..2064a14e8 100644 --- a/internal/web/service/inbound_node_ips.go +++ b/internal/web/service/inbound_node_ips.go @@ -64,6 +64,16 @@ func upsertNodeClientIps(guid string, perEmail map[string][]model.ClientIpEntry) existingByEmail[existing[i].Email] = &existing[i] } + // Deterministic row order keeps concurrent guid merges from deadlocking on + // Postgres (40P01) — same discipline as MergeInboundClientIps. + emails := make([]string, 0, len(perEmail)) + for email := range perEmail { + if email != "" { + emails = append(emails, email) + } + } + sort.Strings(emails) + tx := db.Begin() defer func() { if r := recover(); r != nil { @@ -71,10 +81,8 @@ func upsertNodeClientIps(guid string, perEmail map[string][]model.ClientIpEntry) } }() - for email, incoming := range perEmail { - if email == "" { - continue - } + for _, email := range emails { + incoming := perEmail[email] var old []model.ClientIpEntry if cur, ok := existingByEmail[email]; ok && cur.Ips != "" { _ = json.Unmarshal([]byte(cur.Ips), &old) diff --git a/internal/web/service/node_client_traffic_sum_test.go b/internal/web/service/node_client_traffic_sum_test.go index e16e6eb36..9477f2e54 100644 --- a/internal/web/service/node_client_traffic_sum_test.go +++ b/internal/web/service/node_client_traffic_sum_test.go @@ -3,6 +3,7 @@ package service import ( "fmt" "path/filepath" + "strings" "testing" "gorm.io/gorm" @@ -135,6 +136,47 @@ func TestNodeAdd_ImportsClientHistoryWithNewInbound(t *testing.T) { assertUpDown(t, readTraffic(t, db, email), histUp+1024, histDown+2048, "post-import delta accrues, no double count") } +// TestStaleSnapshot_DeletedClientNotResurrected reproduces #5739: a snapshot +// fetched just before a client's deletion still names the email. Applying it +// must neither recreate the client_traffics row (at zero) nor re-add the +// client to the central inbound's settings while the delete tombstone lives. +func TestStaleSnapshot_DeletedClientNotResurrected(t *testing.T) { + db := initTrafficTestDB(t) + createNodeInboundWithClient(t, db, 1, "n1-in", 41001, "victim-5739") + svc := &InboundService{} + + const email = "victim-5739" + withClient := fmt.Sprintf(`{"clients": [{"email": %q, "enable": true}]}`, email) + syncNodeWithSettings(t, svc, 1, "n1-in", withClient, xray.ClientTraffic{Email: email, Up: 100, Down: 100, Enable: true}) + + if err := db.Model(&model.Inbound{}).Where("tag = ?", "n1-in"). + Update("settings", `{"clients": []}`).Error; err != nil { + t.Fatalf("clear central settings: %v", err) + } + if err := db.Where("email = ?", email).Delete(&xray.ClientTraffic{}).Error; err != nil { + t.Fatalf("delete stats row: %v", err) + } + tombstoneClientEmail(email) + + syncNodeWithSettings(t, svc, 1, "n1-in", withClient, xray.ClientTraffic{Email: email, Up: 120, Down: 120, Enable: true}) + + var rows int64 + if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email).Count(&rows).Error; err != nil { + t.Fatalf("count stats rows: %v", err) + } + if rows != 0 { + t.Errorf("deleted client's stats row resurrected by stale snapshot (%d rows)", rows) + } + + var ib model.Inbound + if err := db.Where("tag = ?", "n1-in").First(&ib).Error; err != nil { + t.Fatalf("load inbound: %v", err) + } + if strings.Contains(ib.Settings, email) { + t.Errorf("deleted client re-added to central settings: %s", ib.Settings) + } +} + func TestNodeAdd_TombstonedClientNotResurrected(t *testing.T) { db := initTrafficTestDB(t) svc := &InboundService{}