diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8a45f8287..350a85e74 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -90,6 +90,16 @@ jobs: exit 1 fi + - name: PostgreSQL client IP sync regressions + run: | + set -o pipefail + go test -p=1 ./internal/web/service ./internal/web/job -run '^TestClientIpSyncPostgres(Scanner)?$' -count=1 -v | tee /tmp/postgres-client-ip-sync.log + passed=$(grep -c -- '^--- PASS' /tmp/postgres-client-ip-sync.log || true) + if [ "$passed" -ne 2 ]; then + echo "expected 2 passing PostgreSQL client IP sync tests, got $passed" >&2 + exit 1 + fi + - name: PostgreSQL global-traffic heartbeat regressions run: | set -o pipefail diff --git a/docs/real-client-ip.md b/docs/real-client-ip.md index cb2337c2f..324e0f6ef 100644 --- a/docs/real-client-ip.md +++ b/docs/real-client-ip.md @@ -86,6 +86,12 @@ fields) is pushed to child nodes verbatim, so the node's Xray records the real I parent panel pulls each node's per-client IPs roughly every 10 seconds. The real visitor IP shows up on the parent automatically. +Reports retain each address's newest timestamp within the 30-minute retention window. +Flat and per-node IP blobs use a stable timestamp-ascending order, with IP text breaking +ties, so equivalent reports preserve the same bytes and skip redundant writes. Flat-row +merges use compare-and-set retries to preserve concurrent observations; advancing a +timestamp remains a real change. GUID attribution stays separate from the flat union. + ## Security note Both `acceptProxyProtocol` and `trustedXForwardedFor` are **server-side only** — they are diff --git a/internal/database/model/node_client_ip.go b/internal/database/model/node_client_ip.go index 11922f86e..e7a6718d5 100644 --- a/internal/database/model/node_client_ip.go +++ b/internal/database/model/node_client_ip.go @@ -1,14 +1,35 @@ package model -// ClientIpEntry is the wire/JSON shape of a single observed client IP with the -// last time it was seen (unix seconds). It mirrors job.IPWithTimestamp and the -// service-internal clientIpEntry so the per-node attribution blob round-trips -// with the existing inbound_client_ips storage. +import ( + "cmp" + "encoding/json" + "slices" +) + +// ClientIpEntry records an observed client IP and its last-seen Unix timestamp. type ClientIpEntry struct { IP string `json:"ip"` Timestamp int64 `json:"timestamp"` } +func SortClientIpEntries(entries []ClientIpEntry) { + slices.SortFunc(entries, func(a, b ClientIpEntry) int { + if order := cmp.Compare(a.Timestamp, b.Timestamp); order != 0 { + return order + } + return cmp.Compare(a.IP, b.IP) + }) +} + +func MarshalClientIpEntries(entries []ClientIpEntry) (string, error) { + if len(entries) == 0 { + return "[]", nil + } + SortClientIpEntries(entries) + encoded, err := json.Marshal(entries) + return string(encoded), err +} + // NodeClientIp records which panel (identified by its stable panelGuid) observed // a client's IPs on its own Xray. Unlike InboundClientIps (a flattened, // cluster-wide union used for IP-limit counting and that is pushed back to every diff --git a/internal/nodee2e/harness_test.go b/internal/nodee2e/harness_test.go index 2aa28ec36..7541b7058 100644 --- a/internal/nodee2e/harness_test.go +++ b/internal/nodee2e/harness_test.go @@ -20,6 +20,7 @@ import ( "time" "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" ) const settleTimeout = 30 * time.Second @@ -196,6 +197,32 @@ func (p *panel) deleteInboundRow(id int) { } } +func (p *panel) seedClientIpAttribution(email, ips string) string { + p.t.Helper() + if p.cmd != nil { + p.t.Fatalf("%s: seedClientIpAttribution on a running panel", p.name) + } + sharedDBMu.Lock() + defer sharedDBMu.Unlock() + if err := database.InitDB(filepath.Join(p.dir, "db", "x-ui.db")); err != nil { + p.t.Fatalf("%s: open db: %v", p.name, err) + } + defer func() { _ = database.CloseDB() }() + db := database.GetDB() + var setting model.Setting + if err := db.Where("key = ?", "panelGuid").First(&setting).Error; err != nil { + p.t.Fatalf("%s: read panelGuid: %v", p.name, err) + } + if setting.Value == "" { + p.t.Fatalf("%s: empty panelGuid", p.name) + } + row := model.NodeClientIp{NodeGuid: setting.Value, Email: email, Ips: ips} + if err := db.Create(&row).Error; err != nil { + p.t.Fatalf("%s: seed attribution: %v", p.name, err) + } + return setting.Value +} + func (p *panel) url() string { return "http://127.0.0.1:" + strconv.Itoa(p.port) } type envelope struct { diff --git a/internal/nodee2e/node_sync_test.go b/internal/nodee2e/node_sync_test.go index 22fa26a29..d62a8aea4 100644 --- a/internal/nodee2e/node_sync_test.go +++ b/internal/nodee2e/node_sync_test.go @@ -3,11 +3,14 @@ package nodee2e import ( "encoding/json" "fmt" + "maps" "net/http" "slices" "strconv" "testing" "time" + + "github.com/mhsanaei/3x-ui/v3/internal/database/model" ) // TestNodeSync walks one master/node pair per enrollment scope through every @@ -91,6 +94,35 @@ func hasAll(have []string, want ...string) bool { return true } +func clientIpSnapshot(t *testing.T, p *panel, email string) (string, map[string]int64) { + t.Helper() + var rows []struct { + ClientEmail string `json:"clientEmail"` + Ips json.RawMessage `json:"ips"` + } + if err := json.Unmarshal(p.call(http.MethodGet, "/panel/api/server/clientIps", nil), &rows); err != nil { + t.Fatalf("decode %s client IP rows: %v", p.name, err) + } + for _, row := range rows { + if row.ClientEmail != email { + continue + } + var entries []model.ClientIpEntry + if err := json.Unmarshal(row.Ips, &entries); err != nil { + t.Fatalf("decode %s IP array: %v (%s)", p.name, err, row.Ips) + } + got := make(map[string]int64, len(entries)) + for _, entry := range entries { + if _, duplicate := got[entry.IP]; duplicate { + t.Fatalf("%s contains duplicate IP %s: %s", p.name, entry.IP, row.Ips) + } + got[entry.IP] = entry.Timestamp + } + return string(row.Ips), got + } + return "", nil +} + func runNodeSyncScenarios(t *testing.T, bin, scope string) { master := newPanel(t, bin, "master") node := newPanel(t, bin, "node") @@ -243,6 +275,86 @@ func runNodeSyncScenarios(t *testing.T, bin, scope string) { } }) + t.Run("client IP union keeps maximum timestamps stable across sync and an offline node", func(t *testing.T) { + now := time.Now().Unix() + post := func(p *panel, ips any) { + p.call(http.MethodPost, "/panel/api/server/clientIps", []map[string]any{{"clientEmail": "p5", "ips": ips}}) + } + post(master, []model.ClientIpEntry{ + {IP: "198.51.100.11", Timestamp: now - 30}, + {IP: "198.51.100.12", Timestamp: now - 20}, + }) + post(node, fmt.Sprintf(`[{"ip":"198.51.100.13","timestamp":%d},{"ip":"198.51.100.12","timestamp":%d}]`, now-10, now-5)) + want := map[string]int64{"198.51.100.11": now - 30, "198.51.100.12": now - 5, "198.51.100.13": now - 10} + waitUnion := func() { + t.Helper() + eventually(t, settleTimeout, "both panels retain the exact IP union and maximum timestamps", func() (bool, string) { + masterRaw, masterIPs := clientIpSnapshot(t, master, "p5") + nodeRaw, nodeIPs := clientIpSnapshot(t, node, "p5") + return maps.Equal(masterIPs, want) && maps.Equal(nodeIPs, want) && masterRaw == nodeRaw, + fmt.Sprintf("master=%s node=%s want=%v", masterRaw, nodeRaw, want) + }) + } + waitUnion() + stableRaw, _ := clientIpSnapshot(t, master, "p5") + time.Sleep(12 * time.Second) + for _, p := range []*panel{master, node} { + raw, got := clientIpSnapshot(t, p, "p5") + if raw != stableRaw || !maps.Equal(got, want) { + t.Fatalf("%s IPs drifted after another sync cycle: %s, want %s", p.name, raw, stableRaw) + } + } + node.stop() + post(master, []model.ClientIpEntry{{IP: "198.51.100.14", Timestamp: now - 1}}) + node.start() + want["198.51.100.14"] = now - 1 + waitUnion() + }) + + t.Run("stopped database observations retain separate GUID attribution buckets", func(t *testing.T) { + now := time.Now().Unix() + master.stop() + node.stop() + masterIPs := fmt.Sprintf(`[{"ip":"203.0.113.21","timestamp":%d}]`, now-20) + nodeIPs := fmt.Sprintf(`[{"ip":"203.0.113.22","timestamp":%d}]`, now-10) + masterGuid := master.seedClientIpAttribution("p5", masterIPs) + nodeGuid := node.seedClientIpAttribution("p5", nodeIPs) + if masterGuid == nodeGuid { + t.Fatalf("independent panels share GUID %q", masterGuid) + } + master.start() + node.start() + want := map[string]map[string][]model.ClientIpEntry{ + masterGuid: {"p5": {{IP: "203.0.113.21", Timestamp: now - 20}}}, + nodeGuid: {"p5": {{IP: "203.0.113.22", Timestamp: now - 10}}}, + } + eventually(t, settleTimeout, "master retains the exact independent GUID observation buckets", func() (bool, string) { + var got map[string]map[string][]model.ClientIpEntry + raw := master.call(http.MethodPost, "/panel/api/clients/clientIpsByGuid", nil) + if err := json.Unmarshal(raw, &got); err != nil { + t.Fatalf("decode attribution: %v", err) + } + expected, err := json.Marshal(want) + if err != nil { + t.Fatal(err) + } + actual, err := json.Marshal(got) + if err != nil { + t.Fatal(err) + } + return string(actual) == string(expected), fmt.Sprintf("attribution=%s want=%s", actual, expected) + }) + var nodeTree map[string]map[string][]model.ClientIpEntry + raw := node.call(http.MethodPost, "/panel/api/clients/clientIpsByGuid", nil) + if err := json.Unmarshal(raw, &nodeTree); err != nil { + t.Fatalf("decode node attribution: %v", err) + } + if len(nodeTree) != 1 || len(nodeTree[nodeGuid]) != 1 || + !slices.Equal(nodeTree[nodeGuid]["p5"], want[nodeGuid]["p5"]) { + t.Fatalf("node attribution mixed parent observations into its own bucket: %s", raw) + } + }) + t.Run("an inbound deleted on the node is removed from the master too", func(t *testing.T) { pr.t = t master.call(http.MethodPost, "/panel/api/inbounds/add", vlessInbound("deleted-on-node", lostPort, &pr.nodeID, vlessClient("l1"))) diff --git a/internal/web/job/check_client_ip_ban_commit_test.go b/internal/web/job/check_client_ip_ban_commit_test.go index 0ce329632..164287928 100644 --- a/internal/web/job/check_client_ip_ban_commit_test.go +++ b/internal/web/job/check_client_ip_ban_commit_test.go @@ -7,6 +7,7 @@ import ( "time" "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" "gorm.io/gorm" ) @@ -86,7 +87,7 @@ func TestProcessObserved_CommitFailureDoesNotPublishBan(t *testing.T) { const email = "rollback-must-not-ban@x" seedLinkedInboundWithClient(t, "rollback-must-not-ban", email, 1) now := time.Now().Unix() - seedClientIps(t, email, []IPWithTimestamp{{IP: "198.51.100.10", Timestamp: now - 2}}) + seedClientIps(t, email, []model.ClientIpEntry{{IP: "198.51.100.10", Timestamp: now - 2}}) installClientIpCommitFailure(t) @@ -146,7 +147,7 @@ func TestProcessObserved_ForgetsBannedSeenWhenClientReturnsUnderLimit(t *testing const email = "prune-banned-seen@x" seedLinkedInboundWithClient(t, "prune-banned-seen", email, 1) now := time.Now().Unix() - seedClientIps(t, email, []IPWithTimestamp{{IP: "203.0.113.1", Timestamp: now - 500}}) + seedClientIps(t, email, []model.ClientIpEntry{{IP: "203.0.113.1", Timestamp: now - 500}}) j := NewCheckClientIpJob() j.processObserved(map[string]map[string]int64{ @@ -180,7 +181,7 @@ func (f failingWriter) Write([]byte) (int, error) { return 0, f.err } // sees, so it has to reach the caller. func TestWriteBanLinesSurfacesWriteFailure(t *testing.T) { want := errors.New("no space left on device") - err := writeBanLines(failingWriter{err: want}, "write-failure@x", []IPWithTimestamp{ + err := writeBanLines(failingWriter{err: want}, "write-failure@x", []model.ClientIpEntry{ {IP: "203.0.113.9", Timestamp: time.Now().Unix()}, }) if !errors.Is(err, want) { @@ -195,7 +196,7 @@ func TestProcessObserved_PublishesBanForCommittedScan(t *testing.T) { const email = "published-ban@x" seedLinkedInboundWithClient(t, "published-ban", email, 1) now := time.Now().Unix() - seedClientIps(t, email, []IPWithTimestamp{{IP: "203.0.113.50", Timestamp: now - 500}}) + seedClientIps(t, email, []model.ClientIpEntry{{IP: "203.0.113.50", Timestamp: now - 500}}) j := NewCheckClientIpJob() if cleaned := j.processObserved(map[string]map[string]int64{ diff --git a/internal/web/job/check_client_ip_cas_test.go b/internal/web/job/check_client_ip_cas_test.go index 7b5378b3d..6165fa317 100644 --- a/internal/web/job/check_client_ip_cas_test.go +++ b/internal/web/job/check_client_ip_cas_test.go @@ -1,7 +1,7 @@ package job import ( - "encoding/json" + "maps" "testing" "time" @@ -9,20 +9,26 @@ import ( "github.com/mhsanaei/3x-ui/v3/internal/database/model" "gorm.io/gorm" + "gorm.io/gorm/clause" ) // A node sync landing between the scan's read and its write must keep its remote // IP (#6587); the hook injects that write, since SQLite serializes real writers. func TestProcessObserved_KeepsNodeSyncWriteThatLandsMidScan(t *testing.T) { setupIntegrationDB(t) + processObservedKeepsNodeSyncWrite(t) +} + +func processObservedKeepsNodeSyncWrite(t *testing.T) { + t.Helper() db := database.GetDB() const email = "cas-scan@x" seedLinkedInboundWithClient(t, "cas-scan", email, 3) now := time.Now().Unix() - seedClientIps(t, email, []IPWithTimestamp{{IP: "198.51.100.1", Timestamp: now - 60}}) + seed := seedClientIps(t, email, []model.ClientIpEntry{{IP: "198.51.100.1", Timestamp: now - 60}}) - nodeBlob, err := json.Marshal([]IPWithTimestamp{ + nodeBlob, err := model.MarshalClientIpEntries([]model.ClientIpEntry{ {IP: "198.51.100.1", Timestamp: now - 60}, {IP: "203.0.113.77", Timestamp: now - 5}, }) @@ -35,11 +41,19 @@ func TestProcessObserved_KeepsNodeSyncWriteThatLandsMidScan(t *testing.T) { if injected || tx.Statement.Schema == nil || tx.Statement.Schema.Table != "inbound_client_ips" { return } + where, ok := tx.Statement.Clauses["WHERE"].Expression.(clause.Where) + if !ok || len(where.Exprs) != 1 { + return + } + predicate, ok := where.Exprs[0].(clause.Expr) + if !ok || len(predicate.Vars) != 2 || predicate.Vars[0] != seed.Id { + return + } injected = true if err := tx.Session(&gorm.Session{SkipHooks: true, NewDB: true}). Model(&model.InboundClientIps{}). Where("client_email = ?", email). - Update("ips", string(nodeBlob)).Error; err != nil { + Update("ips", nodeBlob).Error; err != nil { _ = tx.AddError(err) } }); err != nil { @@ -55,10 +69,8 @@ func TestProcessObserved_KeepsNodeSyncWriteThatLandsMidScan(t *testing.T) { } got := ipSet(readClientIps(t, email)) - if _, ok := got["203.0.113.77"]; !ok { - t.Fatalf("scan overwrote the node's remote IP (the #6587 lost update): %v", got) - } - if got["198.51.100.1"] != now { - t.Fatalf("scan's own observation missing or stale: %v", got) + want := map[string]int64{"203.0.113.77": now - 5, "198.51.100.1": now} + if !maps.Equal(got, want) { + t.Fatalf("scan/node union = %v, want %v", got, want) } } diff --git a/internal/web/job/check_client_ip_frozen_ban_test.go b/internal/web/job/check_client_ip_frozen_ban_test.go index 79daccb1f..4dfa22572 100644 --- a/internal/web/job/check_client_ip_frozen_ban_test.go +++ b/internal/web/job/check_client_ip_frozen_ban_test.go @@ -5,6 +5,8 @@ import ( "strings" "testing" "time" + + "github.com/mhsanaei/3x-ui/v3/internal/database/model" ) func banLineCount(t *testing.T, email string) int { @@ -27,7 +29,7 @@ func TestUpdateInboundClientIps_FrozenLastSeenBannedOnce(t *testing.T) { now := time.Now().Unix() deadStart := now - 300 - live := []IPWithTimestamp{ + live := []model.ClientIpEntry{ {IP: "10.2.0.1", Timestamp: deadStart}, {IP: "192.0.2.7", Timestamp: now}, } @@ -53,7 +55,7 @@ func TestUpdateInboundClientIps_FrozenLastSeenBannedOnce(t *testing.T) { t.Fatalf("ban lines after frozen rescan = %d, want still 1", got) } - reconnected := []IPWithTimestamp{ + reconnected := []model.ClientIpEntry{ {IP: "10.2.0.1", Timestamp: now + 30}, {IP: "192.0.2.7", Timestamp: now + 60}, } diff --git a/internal/web/job/check_client_ip_job.go b/internal/web/job/check_client_ip_job.go index 74649a924..eec1278ef 100644 --- a/internal/web/job/check_client_ip_job.go +++ b/internal/web/job/check_client_ip_job.go @@ -21,14 +21,9 @@ import ( "github.com/mhsanaei/3x-ui/v3/internal/xray" "gorm.io/gorm" + "gorm.io/gorm/clause" ) -// IPWithTimestamp tracks an IP address with its last seen timestamp -type IPWithTimestamp struct { - IP string `json:"ip"` - Timestamp int64 `json:"timestamp"` -} - // onlineUsersSource is the one service.XrayService method the scan needs, so // tests can drive Run end to end without a live xray core. type onlineUsersSource interface { @@ -352,14 +347,12 @@ func (j *CheckClientIpJob) processObserved(observed map[string]map[string]int64, } } - // Convert to IPWithTimestamp slice - ipsWithTime := make([]IPWithTimestamp, 0, len(ipTimestamps)) + ipsWithTime := make([]model.ClientIpEntry, 0, len(ipTimestamps)) attrEntries := make([]model.ClientIpEntry, 0, len(ipTimestamps)) for ip, timestamp := range ipTimestamps { - ipsWithTime = append(ipsWithTime, IPWithTimestamp{IP: ip, Timestamp: timestamp}) - // Live API observations may carry an old lastSeen (connection start), - // so stamp attribution with now; otherwise the stale cutoff would evict - // an IP that is connected right now. + ipsWithTime = append(ipsWithTime, model.ClientIpEntry{IP: ip, Timestamp: timestamp}) + // A live connection's lastSeen may be its old start time. + // Stamp attribution now so retention cannot evict a connected IP. attrTs := timestamp if observedAreLive { attrTs = now @@ -372,15 +365,27 @@ func (j *CheckClientIpJob) processObserved(observed map[string]map[string]int64, clientIpsRecord, ok := ipRowByEmail[email] if !ok { - jsonIps, err := json.Marshal(ipsWithTime) + jsonIps, err := model.MarshalClientIpEntries(ipsWithTime) if err != nil { j.checkError(err) continue } - if err := tx.Save(&model.InboundClientIps{ClientEmail: email, Ips: string(jsonIps)}).Error; err != nil { + clientIpsRecord = &model.InboundClientIps{ClientEmail: email, Ips: jsonIps} + result := tx.Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "client_email"}}, + DoNothing: true, + }).Create(clientIpsRecord) + if result.Error != nil { + j.checkError(result.Error) + continue + } + if result.RowsAffected > 0 { + continue + } + if err := tx.Where("client_email = ?", email).First(clientIpsRecord).Error; err != nil { j.checkError(err) + continue } - continue } candidates, keptLive := j.updateInboundClientIps(tx, clientIpsRecord, inbound, email, limitByEmail[email], ipsWithTime, enforce, observedAreLive) @@ -429,11 +434,9 @@ func (j *CheckClientIpJob) recordLocalAttribution(attribution map[string][]model } } -// mergeClientIps folds this scan's observations into the persisted set, -// dropping entries older than staleCutoff. newAlwaysLive exempts the new -// entries from that cutoff: an API-observed IP is a live connection by -// definition, even when its lastSeen (set at dispatch time) is hours old. -func mergeClientIps(old, new []IPWithTimestamp, staleCutoff int64, newAlwaysLive bool) map[string]int64 { +// mergeClientIps retains fresh entries and exempts currently observed live +// connections from expiry even when their dispatch timestamps are old. +func mergeClientIps(old, new []model.ClientIpEntry, staleCutoff int64, newAlwaysLive bool) map[string]int64 { ipMap := make(map[string]int64, len(old)+len(new)) for _, ipTime := range old { if ipTime.Timestamp < staleCutoff { @@ -454,7 +457,7 @@ func mergeClientIps(old, new []IPWithTimestamp, staleCutoff int64, newAlwaysLive // selectIpsToBan splits the live IPs (sorted oldest-first by partitionLiveIps) // into the newest `limit` entries to keep and the older remainder to ban. -func selectIpsToBan(live []IPWithTimestamp, limit int) (kept, banned []IPWithTimestamp) { +func selectIpsToBan(live []model.ClientIpEntry, limit int) (kept, banned []model.ClientIpEntry) { if limit <= 0 || len(live) <= limit { return live, nil } @@ -462,33 +465,21 @@ func selectIpsToBan(live []IPWithTimestamp, limit int) (kept, banned []IPWithTim return live[cutoff:], live[:cutoff] } -func partitionLiveIps(ipMap map[string]int64, observedThisScan map[string]bool) (live, historical []IPWithTimestamp) { - live = make([]IPWithTimestamp, 0, len(observedThisScan)) - historical = make([]IPWithTimestamp, 0, len(ipMap)) +func partitionLiveIps(ipMap map[string]int64, observedThisScan map[string]bool) (live, historical []model.ClientIpEntry) { + live = make([]model.ClientIpEntry, 0, len(observedThisScan)) + historical = make([]model.ClientIpEntry, 0, len(ipMap)) now := time.Now().Unix() for ip, ts := range ipMap { - entry := IPWithTimestamp{IP: ip, Timestamp: ts} - // Consider an IP "live" if it was seen locally in this scan, OR if its - // timestamp from the synced database is very recent (e.g. within 2 minutes). - // This ensures cluster-wide limits work even if the IP was seen on another node. + entry := model.ClientIpEntry{IP: ip, Timestamp: ts} + // Recent remote observations count toward the cluster-wide limit. if observedThisScan[ip] || now-ts < 120 { live = append(live, entry) } else { historical = append(historical, entry) } } - sort.Slice(live, func(i, j int) bool { - if live[i].Timestamp != live[j].Timestamp { - return live[i].Timestamp < live[j].Timestamp - } - return live[i].IP < live[j].IP - }) - sort.Slice(historical, func(i, j int) bool { - if historical[i].Timestamp != historical[j].Timestamp { - return historical[i].Timestamp < historical[j].Timestamp - } - return historical[i].IP < historical[j].IP - }) + model.SortClientIpEntries(live) + model.SortClientIpEntries(historical) return live, historical } @@ -523,10 +514,8 @@ func (j *CheckClientIpJob) delInboundClientIps(tx *gorm.DB, clientEmail string) } } -// updateInboundClientIps merges one email's observed IPs into its tracking row -// and applies the IP limit. Ban candidates are returned, not written: the -// fail2ban log is the point of no return and must wait for the commit. -func (j *CheckClientIpJob) updateInboundClientIps(tx *gorm.DB, inboundClientIps *model.InboundClientIps, inbound *model.Inbound, clientEmail string, limitIp int, newIpsWithTime []IPWithTimestamp, enforce, observedAreLive bool) (banCandidates []IPWithTimestamp, keptLiveCount int) { +// updateInboundClientIps returns ban candidates; publication must wait for commit. +func (j *CheckClientIpJob) updateInboundClientIps(tx *gorm.DB, inboundClientIps *model.InboundClientIps, inbound *model.Inbound, clientEmail string, limitIp int, newIpsWithTime []model.ClientIpEntry, enforce, observedAreLive bool) (banCandidates []model.ClientIpEntry, keptLiveCount int) { if inbound.Settings == "" { logger.Debug("wrong data:", inbound) return nil, 0 @@ -545,14 +534,14 @@ func (j *CheckClientIpJob) updateInboundClientIps(tx *gorm.DB, inboundClientIps // Node sync merges into the same blob (#6587): compare-and-set it, and on a // miss re-read and re-merge so neither writer drops the other's IPs. for range service.ClientIpCasRetries { - var oldIpsWithTime []IPWithTimestamp + var oldIpsWithTime []model.ClientIpEntry if inboundClientIps.Ips != "" { _ = json.Unmarshal([]byte(inboundClientIps.Ips), &oldIpsWithTime) } ipMap := mergeClientIps(oldIpsWithTime, newIpsWithTime, staleCutoff, observedAreLive) - dbIps := make([]IPWithTimestamp, 0, len(ipMap)) - var bannedLive []IPWithTimestamp + dbIps := make([]model.ClientIpEntry, 0, len(ipMap)) + var bannedLive []model.ClientIpEntry keptLive := 0 if enforceLimit { liveIps, historicalIps := partitionLiveIps(ipMap, observedThisScan) @@ -569,12 +558,16 @@ func (j *CheckClientIpJob) updateInboundClientIps(tx *gorm.DB, inboundClientIps dbIps = append(dbIps, kept...) dbIps = append(dbIps, historicalIps...) } else { - // Collection-only run (#6722): no ban selection; keep every IP in - // the 30-minute window, sorted so unchanged scans skip the write. - dbIps = sortedIpsFromMap(ipMap) + // Collection-only run (#6722): no ban selection, keep every IP in the window. + for ip, timestamp := range ipMap { + dbIps = append(dbIps, model.ClientIpEntry{IP: ip, Timestamp: timestamp}) + } + } + newIps, err := model.MarshalClientIpEntries(dbIps) + if err != nil { + j.checkError(err) + return nil, 0 } - jsonIps, _ := json.Marshal(dbIps) - newIps := string(jsonIps) if newIps == inboundClientIps.Ips { return bannedLive, keptLive } @@ -597,28 +590,12 @@ func (j *CheckClientIpJob) updateInboundClientIps(tx *gorm.DB, inboundClientIps return nil, 0 } -// sortedIpsFromMap orders the merged IP set by timestamp with the IP as -// tie-breaker, giving a stable blob for the skip-when-unchanged compare. -func sortedIpsFromMap(ipMap map[string]int64) []IPWithTimestamp { - out := make([]IPWithTimestamp, 0, len(ipMap)) - for ip, ts := range ipMap { - out = append(out, IPWithTimestamp{IP: ip, Timestamp: ts}) - } - sort.Slice(out, func(i, j int) bool { - if out[i].Timestamp != out[j].Timestamp { - return out[i].Timestamp < out[j].Timestamp - } - return out[i].IP < out[j].IP - }) - return out -} - // pendingBan carries one client's enforcement outcome from inside the scan's // transaction to the publication that may only follow a successful commit. type pendingBan struct { inbound *model.Inbound email string - candidates []IPWithTimestamp + candidates []model.ClientIpEntry keptLive int } @@ -662,7 +639,7 @@ func (j *CheckClientIpJob) publishBans(bans []pendingBan) []pendingBan { // writeBanLines emits one line per address; the wording is load-bearing, since // x-ui.sh create_iplimit_jails builds filter.d/3x-ipl.conf failregex from it. -func writeBanLines(w io.Writer, clientEmail string, actionable []IPWithTimestamp) error { +func writeBanLines(w io.Writer, clientEmail string, actionable []model.ClientIpEntry) error { stamp := time.Now().Format("2006/01/02 15:04:05") for _, ipTime := range actionable { if _, err := fmt.Fprintf(w, "%s [LIMIT_IP] Email = %s || Disconnecting OLD IP = %s || Timestamp = %d\n", @@ -675,8 +652,8 @@ func writeBanLines(w io.Writer, clientEmail string, actionable []IPWithTimestamp // selectAdvancedSinceLastBan drops pairs with a frozen lastSeen: the core // refreshes it only on a new dispatch, so those are unreaped dead connections. -func (j *CheckClientIpJob) selectAdvancedSinceLastBan(email string, banned []IPWithTimestamp) []IPWithTimestamp { - actionable := make([]IPWithTimestamp, 0, len(banned)) +func (j *CheckClientIpJob) selectAdvancedSinceLastBan(email string, banned []model.ClientIpEntry) []model.ClientIpEntry { + actionable := make([]model.ClientIpEntry, 0, len(banned)) for _, ipTime := range banned { if last, ok := j.bannedSeen[email+"|"+ipTime.IP]; ok && ipTime.Timestamp <= last { continue @@ -688,7 +665,7 @@ func (j *CheckClientIpJob) selectAdvancedSinceLastBan(email string, banned []IPW // recordBannedSeen marks published pairs and forgets addresses this scan no // longer bans; it runs for every enforced client, which is what prunes the map. -func (j *CheckClientIpJob) recordBannedSeen(email string, banned, published []IPWithTimestamp) { +func (j *CheckClientIpJob) recordBannedSeen(email string, banned, published []model.ClientIpEntry) { if j.bannedSeen == nil { j.bannedSeen = make(map[string]int64) } diff --git a/internal/web/job/check_client_ip_job_integration_test.go b/internal/web/job/check_client_ip_job_integration_test.go index 6b5c66934..136630543 100644 --- a/internal/web/job/check_client_ip_job_integration_test.go +++ b/internal/web/job/check_client_ip_job_integration_test.go @@ -22,10 +22,7 @@ import ( // log a warning. otherwise log.Warningf panics on a nil logger. var loggerInitOnce sync.Once -// setupIntegrationDB wires a temp sqlite db and log folder so -// updateInboundClientIps can run end to end. closes the db before -// TempDir cleanup so windows doesn't complain about the file being in -// use. +// setupIntegrationDB closes its temporary SQLite database before directory cleanup. func setupIntegrationDB(t *testing.T) { t.Helper() @@ -38,6 +35,8 @@ func setupIntegrationDB(t *testing.T) { logDir := t.TempDir() t.Setenv("XUI_DB_FOLDER", dbDir) + t.Setenv("XUI_DB_TYPE", "sqlite") + t.Setenv("XUI_DB_DSN", "") t.Setenv("XUI_LOG_FOLDER", logDir) // updateInboundClientIps calls log.SetOutput on the package global, @@ -54,7 +53,7 @@ func setupIntegrationDB(t *testing.T) { // enforceIpLimitForTest runs the same two steps processObserved does: select // inside the transaction, publish only once it would have committed. -func (j *CheckClientIpJob) enforceIpLimitForTest(t *testing.T, row *model.InboundClientIps, inbound *model.Inbound, email string, limit int, live []IPWithTimestamp, observedAreLive bool) (banned bool, published []IPWithTimestamp) { +func (j *CheckClientIpJob) enforceIpLimitForTest(t *testing.T, row *model.InboundClientIps, inbound *model.Inbound, email string, limit int, live []model.ClientIpEntry, observedAreLive bool) (banned bool, published []model.ClientIpEntry) { t.Helper() candidates, keptLive := j.updateInboundClientIps(database.GetDB(), row, inbound, email, limit, live, true, observedAreLive) actionable := j.selectAdvancedSinceLastBan(email, candidates) @@ -115,15 +114,15 @@ func seedLinkedInboundWithClient(t *testing.T, tag, email string, limitIp int) * } // seed an InboundClientIps row with the given blob. -func seedClientIps(t *testing.T, email string, ips []IPWithTimestamp) *model.InboundClientIps { +func seedClientIps(t *testing.T, email string, ips []model.ClientIpEntry) *model.InboundClientIps { t.Helper() - blob, err := json.Marshal(ips) + blob, err := model.MarshalClientIpEntries(ips) if err != nil { t.Fatalf("marshal ips: %v", err) } row := &model.InboundClientIps{ ClientEmail: email, - Ips: string(blob), + Ips: blob, } if err := database.GetDB().Create(row).Error; err != nil { t.Fatalf("seed InboundClientIps: %v", err) @@ -132,7 +131,7 @@ func seedClientIps(t *testing.T, email string, ips []IPWithTimestamp) *model.Inb } // read the persisted blob and parse it back. -func readClientIps(t *testing.T, email string) []IPWithTimestamp { +func readClientIps(t *testing.T, email string) []model.ClientIpEntry { t.Helper() row := &model.InboundClientIps{} if err := database.GetDB().Where("client_email = ?", email).First(row).Error; err != nil { @@ -141,7 +140,7 @@ func readClientIps(t *testing.T, email string) []IPWithTimestamp { if row.Ips == "" { return nil } - var out []IPWithTimestamp + var out []model.ClientIpEntry if err := json.Unmarshal([]byte(row.Ips), &out); err != nil { t.Fatalf("unmarshal Ips blob %q: %v", row.Ips, err) } @@ -149,7 +148,7 @@ func readClientIps(t *testing.T, email string) []IPWithTimestamp { } // make a lookup map so asserts don't depend on slice order. -func ipSet(entries []IPWithTimestamp) map[string]int64 { +func ipSet(entries []model.ClientIpEntry) map[string]int64 { out := make(map[string]int64, len(entries)) for _, e := range entries { out[e.IP] = e.Timestamp @@ -198,7 +197,7 @@ func TestUpdateInboundClientIps_LiveIpNotBannedByStillFreshHistoricals(t *testin now := time.Now().Unix() // idle but still within the 30min staleness window. - row := seedClientIps(t, email, []IPWithTimestamp{ + row := seedClientIps(t, email, []model.ClientIpEntry{ {IP: "10.0.0.1", Timestamp: now - 20*60}, {IP: "10.0.0.2", Timestamp: now - 15*60}, {IP: "10.0.0.3", Timestamp: now - 10*60}, @@ -206,7 +205,7 @@ func TestUpdateInboundClientIps_LiveIpNotBannedByStillFreshHistoricals(t *testin j := NewCheckClientIpJob() // the one that's actually connecting (user's 128.71.x.x). - live := []IPWithTimestamp{ + live := []model.ClientIpEntry{ {IP: "128.71.1.1", Timestamp: now}, } @@ -250,7 +249,7 @@ func TestUpdateInboundClientIps_ExcessLiveIpIsStillBanned(t *testing.T) { seedInboundWithClient(t, "inbound-pr4091-abuse", email, 1) now := time.Now().Unix() - row := seedClientIps(t, email, []IPWithTimestamp{ + row := seedClientIps(t, email, []model.ClientIpEntry{ {IP: "10.1.0.1", Timestamp: now - 60}, // original connection }) @@ -258,7 +257,7 @@ func TestUpdateInboundClientIps_ExcessLiveIpIsStillBanned(t *testing.T) { // both live, limit=1. use distinct timestamps so sort-by-timestamp // is deterministic: 10.1.0.1 is the original (older) and must get // banned; 192.0.2.9 joined later and keeps the slot (last IP wins). - live := []IPWithTimestamp{ + live := []model.ClientIpEntry{ {IP: "10.1.0.1", Timestamp: now - 5}, {IP: "192.0.2.9", Timestamp: now}, } @@ -329,7 +328,7 @@ func TestProcessObserved_CollectionOnlyKeepsRecentWindow(t *testing.T) { const email = "window-user" seedInboundWithClient(t, "inbound-window", email, 0) fiveMinAgo := time.Now().Unix() - 5*60 - seedClientIps(t, email, []IPWithTimestamp{{IP: "198.51.100.1", Timestamp: fiveMinAgo}}) + seedClientIps(t, email, []model.ClientIpEntry{{IP: "198.51.100.1", Timestamp: fiveMinAgo}}) observed := map[string]map[string]int64{ email: {"203.0.113.30": time.Now().Unix()}, @@ -391,7 +390,7 @@ func TestProcessObserved_StaleEmailIsSkippedAndOrphanDropped(t *testing.T) { const staleEmail = "renamed-away" // No inbound references staleEmail. Pre-seed an orphan tracking row to // confirm the job removes it rather than leaving it to error forever. - seedClientIps(t, staleEmail, []IPWithTimestamp{{IP: "203.0.113.5", Timestamp: time.Now().Unix()}}) + seedClientIps(t, staleEmail, []model.ClientIpEntry{{IP: "203.0.113.5", Timestamp: time.Now().Unix()}}) observed := map[string]map[string]int64{ staleEmail: {"203.0.113.5": time.Now().Unix()}, @@ -493,14 +492,14 @@ func TestUpdateInboundClientIps_AllowlistedIpIsNeitherCountedNorBanned(t *testin seedInboundWithClient(t, "inbound-issue5378", email, 1) now := time.Now().Unix() - row := seedClientIps(t, email, []IPWithTimestamp{ + row := seedClientIps(t, email, []model.ClientIpEntry{ {IP: "203.0.113.10", Timestamp: now - 60}, }) j := NewCheckClientIpJob() j.allowlist = parseIpLimitAllowlist("203.0.113.0/24") - live := []IPWithTimestamp{ + live := []model.ClientIpEntry{ {IP: "203.0.113.10", Timestamp: now - 5}, {IP: "192.0.2.9", Timestamp: now}, } diff --git a/internal/web/job/check_client_ip_job_test.go b/internal/web/job/check_client_ip_job_test.go index 22d50e0af..ef9ac776a 100644 --- a/internal/web/job/check_client_ip_job_test.go +++ b/internal/web/job/check_client_ip_job_test.go @@ -18,11 +18,11 @@ func TestMergeClientIps_EvictsStaleOldEntries(t *testing.T) { // banned again immediately because a long-disconnected IP stayed in the // DB with an ancient timestamp and kept "protecting" itself against // eviction. Guard against that regression here. - old := []IPWithTimestamp{ + old := []model.ClientIpEntry{ {IP: "1.1.1.1", Timestamp: 100}, // stale — client disconnected long ago {IP: "2.2.2.2", Timestamp: 1900}, // fresh — still connecting } - new := []IPWithTimestamp{ + new := []model.ClientIpEntry{ {IP: "2.2.2.2", Timestamp: 2000}, // same IP, newer log line } @@ -37,7 +37,7 @@ func TestMergeClientIps_EvictsStaleOldEntries(t *testing.T) { func TestMergeClientIps_KeepsFreshOldEntriesUnchanged(t *testing.T) { // Backwards-compat: entries that aren't stale are still carried forward, // so enforcement survives access-log rotation. - old := []IPWithTimestamp{ + old := []model.ClientIpEntry{ {IP: "1.1.1.1", Timestamp: 1500}, } got := mergeClientIps(old, nil, 1000, false) @@ -49,8 +49,8 @@ func TestMergeClientIps_KeepsFreshOldEntriesUnchanged(t *testing.T) { } func TestMergeClientIps_PrefersLaterTimestampForSameIp(t *testing.T) { - old := []IPWithTimestamp{{IP: "1.1.1.1", Timestamp: 1500}} - new := []IPWithTimestamp{{IP: "1.1.1.1", Timestamp: 1700}} + old := []model.ClientIpEntry{{IP: "1.1.1.1", Timestamp: 1500}} + new := []model.ClientIpEntry{{IP: "1.1.1.1", Timestamp: 1700}} got := mergeClientIps(old, new, 1000, false) @@ -62,7 +62,7 @@ func TestMergeClientIps_PrefersLaterTimestampForSameIp(t *testing.T) { func TestMergeClientIps_DropsStaleNewEntries(t *testing.T) { // A log line with a clock-skewed old timestamp must not resurrect a // stale IP past the cutoff. - new := []IPWithTimestamp{{IP: "1.1.1.1", Timestamp: 500}} + new := []model.ClientIpEntry{{IP: "1.1.1.1", Timestamp: 500}} got := mergeClientIps(nil, new, 1000, false) if len(got) != 0 { @@ -73,8 +73,8 @@ func TestMergeClientIps_DropsStaleNewEntries(t *testing.T) { func TestMergeClientIps_NoStaleCutoffStillWorks(t *testing.T) { // Defensive: a zero cutoff (e.g. during very first run on a fresh // install) must not over-evict. - old := []IPWithTimestamp{{IP: "1.1.1.1", Timestamp: 100}} - new := []IPWithTimestamp{{IP: "2.2.2.2", Timestamp: 200}} + old := []model.ClientIpEntry{{IP: "1.1.1.1", Timestamp: 100}} + new := []model.ClientIpEntry{{IP: "2.2.2.2", Timestamp: 200}} got := mergeClientIps(old, new, 0, false) @@ -88,7 +88,7 @@ func TestMergeClientIps_LiveObservationsBypassStaleCutoff(t *testing.T) { // online-API mode: lastSeen is set when the connection was dispatched, so // a connection held open for hours has an "old" timestamp while being live // by definition. It must survive the stale cutoff. - new := []IPWithTimestamp{{IP: "1.1.1.1", Timestamp: 500}} // opened long ago, still connected + new := []model.ClientIpEntry{{IP: "1.1.1.1", Timestamp: 500}} // opened long ago, still connected got := mergeClientIps(nil, new, 1000, true) want := map[string]int64{"1.1.1.1": 500} @@ -100,8 +100,8 @@ func TestMergeClientIps_LiveObservationsBypassStaleCutoff(t *testing.T) { func TestMergeClientIps_LiveModeStillEvictsStaleOldEntries(t *testing.T) { // the bypass applies only to this scan's observations — persisted entries // from past scans still age out as before. - old := []IPWithTimestamp{{IP: "2.2.2.2", Timestamp: 100}} - new := []IPWithTimestamp{{IP: "1.1.1.1", Timestamp: 2000}} + old := []model.ClientIpEntry{{IP: "2.2.2.2", Timestamp: 100}} + new := []model.ClientIpEntry{{IP: "1.1.1.1", Timestamp: 2000}} got := mergeClientIps(old, new, 1000, true) want := map[string]int64{"1.1.1.1": 2000} @@ -111,7 +111,7 @@ func TestMergeClientIps_LiveModeStillEvictsStaleOldEntries(t *testing.T) { } func TestSelectIpsToBan(t *testing.T) { - live := []IPWithTimestamp{ // sorted oldest-first, as partitionLiveIps returns + live := []model.ClientIpEntry{ // sorted oldest-first, as partitionLiveIps returns {IP: "A", Timestamp: 100}, {IP: "B", Timestamp: 200}, {IP: "C", Timestamp: 300}, @@ -144,7 +144,7 @@ func TestSelectIpsToBan(t *testing.T) { } } -func collectIps(entries []IPWithTimestamp) []string { +func collectIps(entries []model.ClientIpEntry) []string { out := make([]string, 0, len(entries)) for _, e := range entries { out = append(out, e.IP) @@ -294,7 +294,7 @@ func TestCheckFail2BanInstalled_EmptyEnvSkipsClientProbe(t *testing.T) { func TestPartitionLiveIps_TiedTimestampsKeepStableIpOrder(t *testing.T) { ipMap := map[string]int64{"203.0.113.3": 100, "203.0.113.1": 100, "203.0.113.2": 100} observed := map[string]bool{"203.0.113.1": true, "203.0.113.2": true, "203.0.113.3": true} - var first []IPWithTimestamp + var first []model.ClientIpEntry for i := range 50 { live, _ := partitionLiveIps(ipMap, observed) if i == 0 { diff --git a/internal/web/job/check_client_ip_scale_test.go b/internal/web/job/check_client_ip_scale_test.go index a6d62cf27..9b70ca9de 100644 --- a/internal/web/job/check_client_ip_scale_test.go +++ b/internal/web/job/check_client_ip_scale_test.go @@ -5,8 +5,11 @@ import ( "fmt" "os" "path/filepath" + "reflect" + "runtime" "strconv" "strings" + "sync/atomic" "testing" "time" @@ -18,6 +21,7 @@ import ( "github.com/mhsanaei/3x-ui/v3/internal/database/dbtest" "github.com/mhsanaei/3x-ui/v3/internal/database/model" xuilogger "github.com/mhsanaei/3x-ui/v3/internal/logger" + "github.com/mhsanaei/3x-ui/v3/internal/testpg" ) // setupScaleJobDB mirrors the service package's scale gating: Postgres via @@ -28,10 +32,15 @@ func setupScaleJobDB(t *testing.T) { t.Setenv("XUI_LOG_FOLDER", t.TempDir()) if os.Getenv("XUI_DB_TYPE") == "postgres" && strings.TrimSpace(os.Getenv("XUI_DB_DSN")) != "" { + cleanup, err := testpg.IsolatePackage("job_ip_scale") + if err != nil { + t.Fatal(err) + } + t.Cleanup(cleanup) + t.Cleanup(func() { _ = database.CloseDB() }) if err := database.InitDB(""); err != nil { t.Fatalf("InitDB(postgres): %v", err) } - t.Cleanup(func() { _ = database.CloseDB() }) return } switch strings.ToLower(strings.TrimSpace(os.Getenv("XUI_SCALE_TEST"))) { @@ -164,11 +173,7 @@ func seedScaleIPDataset(t *testing.T, n, numInbounds int) []string { return limitedEmails } -// TestCheckClientIpScale measures the @every 10s ip-limit job pieces: the -// hasLimitIp gate (settings LIKE scan + full JSON parse of every matching -// inbound) and processObserved with M online users (per-email inbound lookup, -// settings parse and autocommit save). Run twice: first scan half add / half -// update, second scan all update path. +// TestCheckClientIpScale measures the gate and two scans over an identical fixture. func TestCheckClientIpScale(t *testing.T) { shapes := []struct { name string @@ -201,15 +206,27 @@ func TestCheckClientIpScale(t *testing.T) { } } for i := range m / 2 { - seedClientIps(t, limited[i], []IPWithTimestamp{{IP: "10.99.0.1", Timestamp: now - 60}}) + seedClientIps(t, limited[i], []model.ClientIpEntry{{IP: "10.99.0.1", Timestamp: now - 60}}) } + counts := countScannerScaleSQL(t) + before := captureScannerScaleStats(t) + var memoryBefore, memoryAfter runtime.MemStats + runtime.ReadMemStats(&memoryBefore) start = time.Now() j.processObserved(observed, true, true) firstScan := time.Since(start) + firstCounts := counts.snapshot() start = time.Now() j.processObserved(observed, true, true) secondScan := time.Since(start) + runtime.ReadMemStats(&memoryAfter) + finalCounts := counts.snapshot() + t.Logf("first scan SQL flat_SELECTs=%d per_ID_SELECTs=%d CAS_attempts=%d CAS_misses=%d flat_creates=%d flat_affected=%d attribution_SELECTs=%d upsert_attempts=%d attribution_affected=%d", firstCounts[0], firstCounts[1], firstCounts[2], firstCounts[2]-firstCounts[3], firstCounts[4], firstCounts[3]+firstCounts[5], firstCounts[6], firstCounts[7], firstCounts[8]) + t.Logf("second scan SQL flat_SELECTs=%d per_ID_SELECTs=%d CAS_attempts=%d CAS_misses=%d flat_creates=%d flat_affected=%d attribution_SELECTs=%d upsert_attempts=%d attribution_affected=%d", finalCounts[0]-firstCounts[0], finalCounts[1]-firstCounts[1], finalCounts[2]-firstCounts[2], finalCounts[2]-firstCounts[2]-(finalCounts[3]-firstCounts[3]), finalCounts[4]-firstCounts[4], finalCounts[3]-firstCounts[3]+finalCounts[5]-firstCounts[5], finalCounts[6]-firstCounts[6], finalCounts[7]-firstCounts[7], finalCounts[8]-firstCounts[8]) + after := captureScannerScaleStats(t) + t.Logf("scan allocations=%d bytes=%d", memoryAfter.Mallocs-memoryBefore.Mallocs, memoryAfter.TotalAlloc-memoryBefore.TotalAlloc) + logScannerScaleStats(t, before, after) t.Logf("N=%-7d shape=%-8s processObserved M=%-5d first=%-10v second=%-10v (%.1fms/email)", n, shape.name, m, firstScan.Round(time.Millisecond), secondScan.Round(time.Millisecond), float64(secondScan.Milliseconds())/float64(m)) @@ -221,7 +238,120 @@ func TestCheckClientIpScale(t *testing.T) { if rows != int64(m) { t.Fatalf("inbound_client_ips rows = %d, want %d", rows, m) } + for i := range m { + want := map[string]int64{fmt.Sprintf("10.0.%d.%d", i/250, i%250+1): now} + if i < m/2 { + want["10.99.0.1"] = now - 60 + } + if got := ipSet(readClientIps(t, limited[i])); !reflect.DeepEqual(got, want) { + t.Fatalf("retained %s = %v, want %v", limited[i], got, want) + } + } }) } } } + +type scannerScaleStats struct { + Calls, Rows, WALRecords, WALFpi, WALBytes, SharedBlksHit, SharedBlksRead, SharedBlksDirtied int64 + TotalExecTime float64 + StatsReset string + Dealloc int64 +} + +func captureScannerScaleStats(t *testing.T) scannerScaleStats { + t.Helper() + var result scannerScaleStats + if config.GetDBKind() != "postgres" { + return result + } + db := database.GetDB() + if err := db.Raw(`SELECT COALESCE(sum(calls),0) AS calls, COALESCE(sum(rows),0) AS rows, + COALESCE(sum(total_exec_time),0) AS total_exec_time, + COALESCE(sum(wal_records),0) AS wal_records, COALESCE(sum(wal_fpi),0) AS wal_fpi, COALESCE(sum(wal_bytes),0) AS wal_bytes, + COALESCE(sum(shared_blks_hit),0) AS shared_blks_hit, COALESCE(sum(shared_blks_read),0) AS shared_blks_read, + COALESCE(sum(shared_blks_dirtied),0) AS shared_blks_dirtied + FROM public.pg_stat_statements + WHERE dbid = (SELECT oid FROM pg_database WHERE datname = current_database()) + AND (query ILIKE '%inbound_client_ips%' OR query ILIKE '%node_client_ips%') + AND query NOT ILIKE '%pg_stat_statements%'`).Scan(&result).Error; err != nil { + t.Fatal(err) + } + var info struct { + StatsReset string + Dealloc int64 + } + if err := db.Raw("SELECT stats_reset::text AS stats_reset, dealloc FROM public.pg_stat_statements_info").Scan(&info).Error; err != nil { + t.Fatal(err) + } + result.StatsReset, result.Dealloc = info.StatsReset, info.Dealloc + return result +} + +func logScannerScaleStats(t *testing.T, before, after scannerScaleStats) { + t.Helper() + if config.GetDBKind() != "postgres" { + return + } + t.Logf("PG own-db IP statements delta calls=%d rows=%d total_exec_time_ms=%.3f wal_records=%d wal_fpi=%d wal_bytes=%d shared_blks_hit=%d shared_blks_read=%d shared_blks_dirtied=%d stats_reset=%q->%q dealloc=%d->%d", + after.Calls-before.Calls, after.Rows-before.Rows, after.TotalExecTime-before.TotalExecTime, + after.WALRecords-before.WALRecords, after.WALFpi-before.WALFpi, after.WALBytes-before.WALBytes, + after.SharedBlksHit-before.SharedBlksHit, after.SharedBlksRead-before.SharedBlksRead, after.SharedBlksDirtied-before.SharedBlksDirtied, + before.StatsReset, after.StatsReset, before.Dealloc, after.Dealloc) +} + +type scannerScaleCounts [9]atomic.Int64 + +func (c *scannerScaleCounts) snapshot() [9]int64 { + var out [9]int64 + for i := range out { + out[i] = c[i].Load() + } + return out +} + +func countScannerScaleSQL(t *testing.T) *scannerScaleCounts { + t.Helper() + db := database.GetDB() + counts := &scannerScaleCounts{} + const name = "test:scanner_scale_sql" + if err := db.Callback().Query().After("gorm:query").Register(name, func(tx *gorm.DB) { + switch tx.Statement.Table { + case "inbound_client_ips": + counts[0].Add(1) + if strings.Contains(tx.Statement.SQL.String(), "WHERE id =") { + counts[1].Add(1) + } + case "node_client_ips": + counts[6].Add(1) + } + }); err != nil { + t.Fatal(err) + } + if err := db.Callback().Update().After("gorm:update").Register(name, func(tx *gorm.DB) { + if tx.Statement.Table == "inbound_client_ips" { + counts[2].Add(1) + counts[3].Add(tx.RowsAffected) + } + }); err != nil { + t.Fatal(err) + } + if err := db.Callback().Create().After("gorm:create").Register(name, func(tx *gorm.DB) { + switch tx.Statement.Table { + case "inbound_client_ips": + counts[4].Add(1) + counts[5].Add(tx.RowsAffected) + case "node_client_ips": + counts[7].Add(1) + counts[8].Add(tx.RowsAffected) + } + }); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = db.Callback().Query().Remove(name) + _ = db.Callback().Update().Remove(name) + _ = db.Callback().Create().Remove(name) + }) + return counts +} diff --git a/internal/web/job/check_client_ip_sync_postgres_test.go b/internal/web/job/check_client_ip_sync_postgres_test.go new file mode 100644 index 000000000..bbfed2c40 --- /dev/null +++ b/internal/web/job/check_client_ip_sync_postgres_test.go @@ -0,0 +1,188 @@ +package job + +import ( + "encoding/json" + "fmt" + "os" + "testing" + "time" + + "github.com/op/go-logging" + "gorm.io/gorm" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" + xuilogger "github.com/mhsanaei/3x-ui/v3/internal/logger" + "github.com/mhsanaei/3x-ui/v3/internal/testpg" + "github.com/mhsanaei/3x-ui/v3/internal/web/service" +) + +func TestClientIpWritersPreserveCanonicalBytes(t *testing.T) { + setupIntegrationDB(t) + clientIpWritersPreserveCanonicalBytes(t) +} + +func clientIpWritersPreserveCanonicalBytes(t *testing.T) { + t.Helper() + const email = "canonical-writers@x" + seedLinkedInboundWithClient(t, "canonical-writers", email, 3) + now := time.Now().Unix() + entries := []model.ClientIpEntry{ + {IP: "192.0.2.9", Timestamp: now}, + {IP: "10.0.0.1", Timestamp: now}, + {IP: "203.0.113.8", Timestamp: now - 300}, + {IP: "192.0.2.1", Timestamp: now}, + } + want := fmt.Sprintf(`[{"ip":"203.0.113.8","timestamp":%d},{"ip":"10.0.0.1","timestamp":%d},{"ip":"192.0.2.1","timestamp":%d},{"ip":"192.0.2.9","timestamp":%d}]`, now-300, now, now, now) + row := model.InboundClientIps{ClientEmail: email, Ips: want} + db := database.GetDB() + if err := db.Create(&row).Error; err != nil { + t.Fatal(err) + } + flatWrites, attributionWrites := 0, 0 + const callback = "test:canonical_writer_updates" + if err := db.Callback().Update().After("gorm:update").Register(callback, func(tx *gorm.DB) { + if tx.Statement.Schema == nil || tx.RowsAffected == 0 { + return + } + if tx.Statement.Schema.Table == "inbound_client_ips" { + flatWrites++ + } + }); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Callback().Update().Remove(callback) }) + if err := db.Callback().Create().After("gorm:create").Register(callback, func(tx *gorm.DB) { + if tx.Statement.Schema != nil && tx.Statement.Schema.Table == "node_client_ips" && tx.RowsAffected > 0 { + attributionWrites++ + } + }); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Callback().Create().Remove(callback) }) + j := NewCheckClientIpJob() + j.allowlist = parseIpLimitAllowlist("10.0.0.1") + observed := map[string]map[string]int64{email: {"192.0.2.9": now, "10.0.0.1": now, "192.0.2.1": now}} + for iteration := range 4 { + blob, err := json.Marshal(entries) + if err != nil { + t.Fatal(err) + } + if err := (&service.InboundService{}).MergeInboundClientIps([]model.InboundClientIps{{ClientEmail: email, Ips: string(blob)}}); err != nil { + t.Fatal(err) + } + j.processObserved(observed, true, false) + var got model.InboundClientIps + if err := db.Where("client_email = ?", email).First(&got).Error; err != nil { + t.Fatal(err) + } + if got.Ips != want { + t.Fatalf("iteration %d canonical bytes = %s, want %s", iteration, got.Ips, want) + } + } + if flatWrites != 0 { + t.Fatalf("unchanged alternating writers performed %d flat updates, want 0", flatWrites) + } + beforeAttribution := attributionWrites + j.processObserved(map[string]map[string]int64{email: {"192.0.2.9": now + 1}}, true, false) + if flatWrites != 1 { + t.Fatalf("advancing observation flat updates = %d, want 1", flatWrites) + } + advancedWant := fmt.Sprintf(`[{"ip":"203.0.113.8","timestamp":%d},{"ip":"10.0.0.1","timestamp":%d},{"ip":"192.0.2.1","timestamp":%d},{"ip":"192.0.2.9","timestamp":%d}]`, now-300, now, now, now+1) + var advanced model.InboundClientIps + if err := db.Where("client_email = ?", email).First(&advanced).Error; err != nil { + t.Fatal(err) + } + if advanced.Ips != advancedWant { + t.Fatalf("advancing flat bytes = %s, want %s", advanced.Ips, advancedWant) + } + if attributionWrites != beforeAttribution+1 { + t.Fatalf("advancing attribution writes = %d, want %d", attributionWrites, beforeAttribution+1) + } + var attribution model.NodeClientIp + if err := db.Where("email = ?", email).First(&attribution).Error; err != nil { + t.Fatal(err) + } + var attributed []model.ClientIpEntry + if err := json.Unmarshal([]byte(attribution.Ips), &attributed); err != nil { + t.Fatal(err) + } + found := false + for _, entry := range attributed { + if entry.IP == "192.0.2.9" { + found = entry.Timestamp == now+1 + } + } + if !found { + t.Fatalf("advancing attribution = %v, want 192.0.2.9 at %d", attributed, now+1) + } +} + +func TestProcessObserved_CreateConflictPreservesUnion(t *testing.T) { + setupIntegrationDB(t) + processObservedCreateConflictPreservesUnion(t) +} + +func processObservedCreateConflictPreservesUnion(t *testing.T) { + t.Helper() + const email = "create-conflict@x" + seedLinkedInboundWithClient(t, "create-conflict", email, 3) + now := time.Now().Unix() + winning := fmt.Sprintf(`[{"ip":"192.0.2.1","timestamp":%d},{"ip":"203.0.113.8","timestamp":%d}]`, now-30, now) + db := database.GetDB() + injected := false + const callback = "test:scanner_create_conflict" + if err := db.Callback().Create().Before("gorm:create").Register(callback, func(tx *gorm.DB) { + if injected || tx.Statement.Schema == nil || tx.Statement.Schema.Table != "inbound_client_ips" { + return + } + injected = true + if err := tx.Session(&gorm.Session{NewDB: true, SkipHooks: true}).Exec("INSERT INTO inbound_client_ips (client_email, ips) VALUES (?, ?)", email, winning).Error; err != nil { + _ = tx.AddError(err) + } + }); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Callback().Create().Remove(callback) }) + NewCheckClientIpJob().processObserved(map[string]map[string]int64{email: {"192.0.2.1": now, "198.51.100.2": now - 5}}, true, true) + if !injected { + t.Fatal("missing-row insert was not exercised") + } + var row model.InboundClientIps + if err := db.Where("client_email = ?", email).First(&row).Error; err != nil { + t.Fatal(err) + } + want := fmt.Sprintf(`[{"ip":"198.51.100.2","timestamp":%d},{"ip":"192.0.2.1","timestamp":%d},{"ip":"203.0.113.8","timestamp":%d}]`, now-5, now, now) + if row.Ips != want { + t.Fatalf("conflict union bytes = %s, want %s", row.Ips, want) + } +} + +func TestClientIpSyncPostgresScanner(t *testing.T) { + if os.Getenv("XUI_DB_TYPE") != "postgres" || os.Getenv("XUI_DB_DSN") == "" { + t.Skip("requires PostgreSQL") + } + for _, tc := range []struct { + name string + run func(*testing.T) + }{ + {"CanonicalWriters", clientIpWritersPreserveCanonicalBytes}, + {"CreateConflict", processObservedCreateConflictPreservesUnion}, + {"ConcurrentWriter", processObservedKeepsNodeSyncWrite}, + } { + t.Run(tc.name, func(t *testing.T) { + cleanup, err := testpg.IsolatePackage("job_ip_sync") + if err != nil { + t.Fatal(err) + } + t.Cleanup(cleanup) + t.Setenv("XUI_LOG_FOLDER", t.TempDir()) + loggerInitOnce.Do(func() { xuilogger.InitLogger(logging.ERROR) }) + t.Cleanup(func() { _ = database.CloseDB() }) + if err := database.InitDB(""); err != nil { + t.Fatal(err) + } + tc.run(t) + }) + } +} diff --git a/internal/web/job/ip_limit_allowlist.go b/internal/web/job/ip_limit_allowlist.go index d463deb32..aab43aec0 100644 --- a/internal/web/job/ip_limit_allowlist.go +++ b/internal/web/job/ip_limit_allowlist.go @@ -4,6 +4,8 @@ import ( "net/netip" "slices" "strings" + + "github.com/mhsanaei/3x-ui/v3/internal/database/model" ) // An address that matches is neither counted towards a client's IP limit nor @@ -66,11 +68,11 @@ func (l ipLimitAllowlist) contains(ip string) bool { // split separates the entries an allowlist protects from the ones the limit // still applies to, preserving the caller's ordering in both. -func (l ipLimitAllowlist) split(entries []IPWithTimestamp) (limited, allowed []IPWithTimestamp) { +func (l ipLimitAllowlist) split(entries []model.ClientIpEntry) (limited, allowed []model.ClientIpEntry) { if l.empty() { return entries, nil } - limited = make([]IPWithTimestamp, 0, len(entries)) + limited = make([]model.ClientIpEntry, 0, len(entries)) for _, entry := range entries { if l.contains(entry.IP) { allowed = append(allowed, entry) diff --git a/internal/web/job/ip_limit_allowlist_test.go b/internal/web/job/ip_limit_allowlist_test.go index fc1c9c86a..9464644e9 100644 --- a/internal/web/job/ip_limit_allowlist_test.go +++ b/internal/web/job/ip_limit_allowlist_test.go @@ -1,6 +1,10 @@ package job -import "testing" +import ( + "testing" + + "github.com/mhsanaei/3x-ui/v3/internal/database/model" +) // Addresses in the examples below come from the documentation ranges reserved // by RFC 5737 and RFC 3849. @@ -37,7 +41,7 @@ func TestIpLimitAllowlistIgnoresUnparsableEntries(t *testing.T) { // The point of the setting: a shared address is neither banned nor counted, so // the office NAT it protects does not consume the client's limit either. func TestIpLimitAllowlistSplitKeepsAllowedOutOfTheCount(t *testing.T) { - live := []IPWithTimestamp{ + live := []model.ClientIpEntry{ {IP: "203.0.113.10", Timestamp: 1}, {IP: "192.0.2.1", Timestamp: 2}, {IP: "192.0.2.2", Timestamp: 3}, @@ -62,7 +66,7 @@ func TestIpLimitAllowlistSplitKeepsAllowedOutOfTheCount(t *testing.T) { } func TestIpLimitAllowlistEmptyListChangesNothing(t *testing.T) { - live := []IPWithTimestamp{{IP: "192.0.2.1", Timestamp: 1}, {IP: "192.0.2.2", Timestamp: 2}} + live := []model.ClientIpEntry{{IP: "192.0.2.1", Timestamp: 1}, {IP: "192.0.2.2", Timestamp: 2}} limited, allowed := parseIpLimitAllowlist("").split(live) if allowed != nil || len(limited) != 2 { t.Fatalf("empty allowlist changed the input: limited=%v allowed=%v", limited, allowed) diff --git a/internal/web/service/client_ip_sync_postgres_test.go b/internal/web/service/client_ip_sync_postgres_test.go new file mode 100644 index 000000000..7fc575313 --- /dev/null +++ b/internal/web/service/client_ip_sync_postgres_test.go @@ -0,0 +1,36 @@ +package service + +import ( + "os" + "strings" + "testing" +) + +func TestClientIpSyncPostgres(t *testing.T) { + if os.Getenv("XUI_DB_TYPE") != "postgres" || strings.TrimSpace(os.Getenv("XUI_DB_DSN")) == "" { + t.Skip("set XUI_DB_TYPE=postgres and XUI_DB_DSN to run PostgreSQL IP sync regressions") + } + for _, tc := range []struct { + name string + run func(*testing.T) + }{ + {"seeded snapshot", TestClientIpSyncSeededSnapshot}, + {"duplicates", TestClientIpSyncDuplicatePayloadsKeepUnion}, + {"raw snapshot retry", TestClientIpSyncRawSnapshotAndSingleRetry}, + {"create conflict", TestClientIpSyncCreateConflictMergesWinner}, + {"exhaustion rollback", TestClientIpSyncCASExhaustionRollsBack}, + {"concurrent writer", TestMergeInboundClientIps_RetriesAfterConcurrentWriter}, + {"attribution unchanged", TestNodeClientIpCanonicalUnchangedSkipsUpsert}, + {"remote ids", TestMergeInboundClientIps_CreatesNodeOnlyRowIgnoringRemoteId}, + {"attribution round trip", TestRecordLocalClientIps_RoundTripByGuid}, + {"attribution expiry", TestRecordLocalClientIps_MergesAndDropsStale}, + {"attribution empty", TestUpsertNodeClientIps_EmptyMergeDeletesRow}, + {"retention", TestPruneStaleClientIpsExpiresUnobservedRows}, + {"cloned GUID", TestMergeClientIpsByGuid_RemapsClonedNodeSubtree}, + {"maximum timestamps", TestMergeInboundClientIps_DedupKeepsMaxTimestamp}, + {"stale merge", TestMergeInboundClientIps_DropsStaleIps}, + {"stale insert", TestMergeInboundClientIps_SkipsAllStaleCreate}, + } { + t.Run(tc.name, tc.run) + } +} diff --git a/internal/web/service/client_ip_sync_scale_test.go b/internal/web/service/client_ip_sync_scale_test.go new file mode 100644 index 000000000..fcbc022eb --- /dev/null +++ b/internal/web/service/client_ip_sync_scale_test.go @@ -0,0 +1,206 @@ +package service + +import ( + "encoding/json" + "fmt" + "reflect" + "sync" + "testing" + "time" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" + "github.com/mhsanaei/3x-ui/v3/internal/testpg" +) + +type clientIpPGStats struct { + Calls float64 + Rows float64 + TotalExecTime float64 + WalRecords float64 + WalFpi float64 + WalBytes float64 + SharedBlksHit float64 + SharedBlksRead float64 + SharedBlksDirtied float64 +} + +func clientIpPGSnapshot(t *testing.T) clientIpPGStats { + t.Helper() + var out clientIpPGStats + db := database.GetDB() + if db.Name() != "postgres" { + return out + } + err := db.Raw(`SELECT COALESCE(sum(calls),0) AS calls, COALESCE(sum(rows),0) AS rows, + COALESCE(sum(total_exec_time),0) AS total_exec_time, COALESCE(sum(wal_records),0) AS wal_records, + COALESCE(sum(wal_fpi),0) AS wal_fpi, COALESCE(sum(wal_bytes),0) AS wal_bytes, + COALESCE(sum(shared_blks_hit),0) AS shared_blks_hit, COALESCE(sum(shared_blks_read),0) AS shared_blks_read, + COALESCE(sum(shared_blks_dirtied),0) AS shared_blks_dirtied + FROM public.pg_stat_statements WHERE dbid = (SELECT oid FROM pg_database WHERE datname = current_database()) + AND (query ILIKE '%inbound_client_ips%' OR query ILIKE '%node_client_ips%') + AND query NOT ILIKE '%pg_stat_statements%'`).Scan(&out).Error + if err != nil { + t.Fatal(err) + } + return out +} + +func logClientIpPGDelta(t *testing.T, before, after clientIpPGStats) { + t.Helper() + if database.GetDB().Name() != "postgres" { + return + } + t.Logf("PG delta calls=%.0f rows=%.0f total_exec_time_ms=%.3f wal_records=%.0f wal_fpi=%.0f wal_bytes=%.0f shared_blks_hit=%.0f read=%.0f dirtied=%.0f", + after.Calls-before.Calls, after.Rows-before.Rows, after.TotalExecTime-before.TotalExecTime, + after.WalRecords-before.WalRecords, after.WalFpi-before.WalFpi, after.WalBytes-before.WalBytes, + after.SharedBlksHit-before.SharedBlksHit, after.SharedBlksRead-before.SharedBlksRead, after.SharedBlksDirtied-before.SharedBlksDirtied) +} + +func logClientIpPGContext(t *testing.T) { + t.Helper() + db := database.GetDB() + if db.Name() != "postgres" { + return + } + var info struct { + StatsReset string + Dealloc int64 + } + if err := db.Raw("SELECT stats_reset::text, dealloc FROM public.pg_stat_statements_info").Scan(&info).Error; err != nil { + t.Fatal(err) + } + var version string + if err := db.Raw("SELECT version()").Scan(&version).Error; err != nil { + t.Fatal(err) + } + var settings struct { + FullPageWrites string + CheckpointTimeout string + } + if err := db.Raw("SELECT current_setting('full_page_writes') AS full_page_writes, current_setting('checkpoint_timeout') AS checkpoint_timeout").Scan(&settings).Error; err != nil { + t.Fatal(err) + } + t.Logf("PG %s stats_reset=%s dealloc=%d full_page_writes=%s checkpoint_timeout=%s", version, info.StatsReset, info.Dealloc, settings.FullPageWrites, settings.CheckpointTimeout) +} + +func TestClientIpSyncScale(t *testing.T) { + for _, n := range scaleSizes(t, 1000, 10000, 20000) { + t.Run(fmt.Sprintf("N=%d", n), func(t *testing.T) { + cleanup, err := testpg.IsolatePackage("client_ip_sync_scale") + if err != nil { + t.Fatal(err) + } + t.Cleanup(cleanup) + setupScaleDB(t) + db := database.GetDB() + resetScaleTables(t, db, "inbound_client_ips", "node_client_ips") + logClientIpPGContext(t) + now := time.Now().Unix() + makeReport := func(node int, advance int64) ([]model.InboundClientIps, map[string][]model.ClientIpEntry) { + rows := make([]model.InboundClientIps, n) + attribution := make(map[string][]model.ClientIpEntry, n) + for i := range n { + email := fmt.Sprintf("sync-%07d@scale", i) + entries := []model.ClientIpEntry{{IP: fmt.Sprintf("2001:db8:%x::%x", node+1, i+1), Timestamp: now + int64(node) + advance}} + b, err := json.Marshal(entries) + if err != nil { + t.Fatal(err) + } + rows[i] = model.InboundClientIps{ClientEmail: email, Ips: string(b)} + attribution[email] = entries + } + return rows, attribution + } + apply := func(node int, rows []model.InboundClientIps, attrs map[string][]model.ClientIpEntry) error { + if err := (&InboundService{}).MergeInboundClientIps(rows); err != nil { + return err + } + return (&InboundService{}).RecordLocalClientIps(fmt.Sprintf("bench-node-%d", node), attrs) + } + for node := range 4 { + rows, attrs := makeReport(node, 0) + if err := apply(node, rows, attrs); err != nil { + t.Fatal(err) + } + } + flat := countClientIpSQL(t, "inbound_client_ips") + attr := countClientIpSQL(t, "node_client_ips") + for phase, advance := range []int64{0, 10, 20} { + t.Run([]string{"unchanged", "advancing", "contention"}[phase], func(t *testing.T) { + reports := make([][]model.InboundClientIps, 4) + attrs := make([]map[string][]model.ClientIpEntry, 4) + for node := range 4 { + reports[node], attrs[node] = makeReport(node, advance) + } + before := clientIpPGSnapshot(t) + fs, fu, fa := flat.selects.Load(), flat.updates.Load(), flat.affected.Load() + as, ac, aa := attr.selects.Load(), attr.creates.Load(), attr.affected.Load() + start := time.Now() + if phase == 2 { + var wg sync.WaitGroup + errs := make(chan error, 4) + for node := range 4 { + wg.Go(func() { errs <- apply(node, reports[node], attrs[node]) }) + } + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatal(err) + } + } + } else { + for node := range 4 { + if err := apply(node, reports[node], attrs[node]); err != nil { + t.Fatal(err) + } + } + } + elapsed := time.Since(start) + after := clientIpPGSnapshot(t) + t.Logf("N=%d reports=4 phase=%s elapsed=%v flat_SELECTs=%d CAS_attempts=%d CAS_misses=%d flat_affected=%d attribution_SELECTs=%d upsert_attempts=%d attribution_affected=%d", n, t.Name(), elapsed, + flat.selects.Load()-fs, flat.updates.Load()-fu, flat.updates.Load()-fu-(flat.affected.Load()-fa), flat.affected.Load()-fa, + attr.selects.Load()-as, attr.creates.Load()-ac, attr.affected.Load()-aa) + logClientIpPGDelta(t, before, after) + var actual []model.InboundClientIps + if err := db.Find(&actual).Error; err != nil { + t.Fatal(err) + } + if len(actual) != n { + t.Fatalf("flat rows=%d, want %d", len(actual), n) + } + for _, row := range actual { + var entries []model.ClientIpEntry + if err := json.Unmarshal([]byte(row.Ips), &entries); err != nil { + t.Fatal(err) + } + got := make(map[string]int64, len(entries)) + for _, entry := range entries { + got[entry.IP] = entry.Timestamp + } + want := make(map[string]int64, 4) + for node := range 4 { + for _, entry := range attrs[node][row.ClientEmail] { + want[entry.IP] = entry.Timestamp + } + } + if len(entries) != 4 || !reflect.DeepEqual(got, want) { + t.Fatalf("%s entries=%v, want %v", row.ClientEmail, entries, want) + } + } + trees, err := (&InboundService{}).GetClientIpsByGuid() + if err != nil { + t.Fatal(err) + } + for node := range 4 { + if !reflect.DeepEqual(trees[fmt.Sprintf("bench-node-%d", node)], attrs[node]) { + t.Fatalf("node %d attribution does not match exact report", node) + } + } + }) + } + logClientIpPGContext(t) + }) + } +} diff --git a/internal/web/service/client_ips_prune_test.go b/internal/web/service/client_ips_prune_test.go index d6a2c33e1..213cce064 100644 --- a/internal/web/service/client_ips_prune_test.go +++ b/internal/web/service/client_ips_prune_test.go @@ -29,10 +29,10 @@ func TestPruneStaleClientIpsExpiresUnobservedRows(t *testing.T) { } seed := []any{ - &model.InboundClientIps{ClientEmail: "offline", Ips: marshalIps(t, clientIpEntry{IP: "198.51.100.7", Timestamp: stale})}, + &model.InboundClientIps{ClientEmail: "offline", Ips: marshalIps(t, model.ClientIpEntry{IP: "198.51.100.7", Timestamp: stale})}, &model.InboundClientIps{ClientEmail: "mixed", Ips: marshalIps(t, - clientIpEntry{IP: "198.51.100.8", Timestamp: stale}, - clientIpEntry{IP: "203.0.113.9", Timestamp: fresh})}, + model.ClientIpEntry{IP: "198.51.100.8", Timestamp: stale}, + model.ClientIpEntry{IP: "203.0.113.9", Timestamp: fresh})}, &model.NodeClientIp{NodeGuid: "g1", Email: "node-offline", Ips: mkNode(model.ClientIpEntry{IP: "198.51.100.9", Timestamp: stale})}, &model.NodeClientIp{NodeGuid: "g1", Email: "node-fresh", Ips: mkNode(model.ClientIpEntry{IP: "203.0.113.10", Timestamp: fresh})}, } diff --git a/internal/web/service/inbound_client_ips.go b/internal/web/service/inbound_client_ips.go index ad4daa374..31039af4c 100644 --- a/internal/web/service/inbound_client_ips.go +++ b/internal/web/service/inbound_client_ips.go @@ -47,23 +47,11 @@ func (s *InboundService) GetNodeInboundClientIps(nodeID int) ([]model.InboundCli return ips, nil } -// clientIpStaleAfterSeconds mirrors job.ipStaleAfterSeconds: client IPs older than -// 30 minutes are evicted. Applying the same cutoff inside the cross-node merge keeps -// the synced blob bounded and stops the master's push-back from resurrecting IPs that -// a node has already pruned (otherwise the merge defeats the eviction cluster-wide). +// Match the scanner's 30-minute retention so sync cannot resurrect pruned IPs. const clientIpStaleAfterSeconds = int64(30 * 60) -// clientIpEntry is the on-disk shape of each element of InboundClientIps.Ips. Tags -// match job.IPWithTimestamp so the blob round-trips with the access.log scanner. -type clientIpEntry struct { - IP string `json:"ip"` - Timestamp int64 `json:"timestamp"` -} - -// mergeClientIpEntries unions old and incoming IP observations, dropping anything -// older than cutoff, keeping the most recent timestamp per IP, and returning the -// result sorted newest-first. -func mergeClientIpEntries(old, incoming []clientIpEntry, cutoff int64) []clientIpEntry { +// mergeClientIpEntries keeps each fresh IP's maximum timestamp. +func mergeClientIpEntries(old, incoming []model.ClientIpEntry, cutoff int64) []model.ClientIpEntry { ipMap := make(map[string]int64, len(old)+len(incoming)) for _, e := range old { if e.Timestamp < cutoff { @@ -79,19 +67,14 @@ func mergeClientIpEntries(old, incoming []clientIpEntry, cutoff int64) []clientI ipMap[e.IP] = e.Timestamp } } - out := make([]clientIpEntry, 0, len(ipMap)) + out := make([]model.ClientIpEntry, 0, len(ipMap)) for ip, ts := range ipMap { - out = append(out, clientIpEntry{IP: ip, Timestamp: ts}) + out = append(out, model.ClientIpEntry{IP: ip, Timestamp: ts}) } - sort.Slice(out, func(i, j int) bool { return out[i].Timestamp > out[j].Timestamp }) return out } -// MergeInboundClientIps folds client IPs synced from another node into the local -// inbound_client_ips table without double-counting an IP seen on multiple nodes and -// without resurrecting stale entries. Existing rows are updated in place; brand-new -// clients (typically node-only clients with no local row) are created with a fresh -// local id. +// MergeInboundClientIps unions remote observations without losing concurrent scanner writes. func (s *InboundService) MergeInboundClientIps(incomingIps []model.InboundClientIps) error { db := database.GetDB() var currentIps []model.InboundClientIps @@ -107,10 +90,7 @@ 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. + // All sync writers take row locks in email order to avoid PostgreSQL deadlocks. sort.Slice(incomingIps, func(i, j int) bool { return incomingIps[i].ClientEmail < incomingIps[j].ClientEmail }) @@ -127,39 +107,42 @@ func (s *InboundService) MergeInboundClientIps(incomingIps []model.InboundClient continue } - var incomingEntries []clientIpEntry + var incomingEntries []model.ClientIpEntry _ = json.Unmarshal([]byte(incoming.Ips), &incomingEntries) current, exists := currentMap[incoming.ClientEmail] if !exists { - // New client we've never seen locally. Drop stale entries up front and - // skip the row entirely if nothing is fresh, so we don't persist a row - // that is dead on arrival. + // A new row must contain at least one fresh observation. fresh := mergeClientIpEntries(nil, incomingEntries, cutoff) if len(fresh) == 0 { continue } - b, _ := json.Marshal(fresh) - incoming.Ips = string(b) - // Never carry the remote node's primary key into the local table: id - // spaces are independent across nodes and the remote id would collide - // with an unrelated local row. OnConflict guards the race where - // check_client_ip_job creates the same brand-new email between the - // snapshot above and this insert. + incoming.Ips, _ = model.MarshalClientIpEntries(fresh) + // Remote primary keys belong to an independent database. incoming.Id = 0 - if err := tx.Clauses(clause.OnConflict{ + result := tx.Clauses(clause.OnConflict{ Columns: []clause.Column{{Name: "client_email"}}, DoNothing: true, - }).Create(&incoming).Error; err != nil { + }).Create(&incoming) + if result.Error != nil { + tx.Rollback() + return result.Error + } + if result.RowsAffected == 1 { + currentMap[incoming.ClientEmail] = &incoming + continue + } + current = &model.InboundClientIps{} + if err := tx.Where("client_email = ?", incoming.ClientEmail).First(current).Error; err != nil { tx.Rollback() return err } - continue + currentMap[incoming.ClientEmail] = current } // check_client_ip_job rewrites this blob too; an unconditional Update loses // whichever writer commits second and its remote IPs with it (#6587). - if err := mergeExistingClientIps(tx, current.Id, incomingEntries, cutoff); err != nil { + if err := mergeExistingClientIps(tx, current, incomingEntries, cutoff); err != nil { tx.Rollback() return err } @@ -183,33 +166,33 @@ func CasUpdateInboundClientIps(tx *gorm.DB, id int, expectedIps, newIps string) return res.RowsAffected == 1, nil } -// mergeExistingClientIps folds incoming into the row at id under a CAS loop so -// a concurrent check_client_ip_job write cannot erase the merge (#6587). -func mergeExistingClientIps(tx *gorm.DB, id int, incoming []clientIpEntry, cutoff int64) error { - for attempt := 0; attempt < ClientIpCasRetries; attempt++ { - var row model.InboundClientIps - if err := tx.Where("id = ?", id).First(&row).Error; err != nil { - return err - } - var oldEntries []clientIpEntry - if row.Ips != "" { - _ = json.Unmarshal([]byte(row.Ips), &oldEntries) +// Seed the first CAS from the preload; only a competing write requires a re-read. +func mergeExistingClientIps(tx *gorm.DB, current *model.InboundClientIps, incoming []model.ClientIpEntry, cutoff int64) error { + for attempt := range ClientIpCasRetries { + var oldEntries []model.ClientIpEntry + if current.Ips != "" { + _ = json.Unmarshal([]byte(current.Ips), &oldEntries) } merged := mergeClientIpEntries(oldEntries, incoming, cutoff) - b, _ := json.Marshal(merged) - mergedStr := string(b) - if row.Ips == mergedStr { + mergedStr, _ := model.MarshalClientIpEntries(merged) + if current.Ips == mergedStr { return nil } - ok, err := CasUpdateInboundClientIps(tx, id, row.Ips, mergedStr) + ok, err := CasUpdateInboundClientIps(tx, current.Id, current.Ips, mergedStr) if err != nil { return err } if ok { + current.Ips = mergedStr return nil } + if attempt+1 < ClientIpCasRetries { + if err := tx.Where("id = ?", current.Id).First(current).Error; err != nil { + return err + } + } } - return fmt.Errorf("inbound_client_ips id=%d: exhausted CAS retries merging client IPs", id) + return fmt.Errorf("inbound_client_ips id=%d: exhausted CAS retries merging client IPs", current.Id) } func (s *InboundService) UpdateClientIPs(tx *gorm.DB, oldEmail string, newEmail string) error { @@ -250,13 +233,7 @@ func (s *InboundService) GetInboundClientIps(clientEmail string) (string, error) return "", nil } - // Try to parse as new format (with timestamps) - type IPWithTimestamp struct { - IP string `json:"ip"` - Timestamp int64 `json:"timestamp"` - } - - var ipsWithTime []IPWithTimestamp + var ipsWithTime []model.ClientIpEntry err = json.Unmarshal([]byte(InboundClientIps.Ips), &ipsWithTime) // If successfully parsed as new format, return with timestamps @@ -270,12 +247,9 @@ func (s *InboundService) GetInboundClientIps(clientEmail string) (string, error) err = json.Unmarshal([]byte(InboundClientIps.Ips), &oldIps) if err == nil && len(oldIps) > 0 { // Convert old format to new format with current timestamp - newIpsWithTime := make([]IPWithTimestamp, len(oldIps)) + newIpsWithTime := make([]model.ClientIpEntry, len(oldIps)) for i, ip := range oldIps { - newIpsWithTime[i] = IPWithTimestamp{ - IP: ip, - Timestamp: time.Now().Unix(), - } + newIpsWithTime[i] = model.ClientIpEntry{IP: ip, Timestamp: time.Now().Unix()} } result, _ := json.Marshal(newIpsWithTime) return string(result), nil @@ -309,7 +283,7 @@ func (s *InboundService) PruneStaleClientIps() error { return err } for _, row := range rows { - var entries []clientIpEntry + var entries []model.ClientIpEntry if row.Ips != "" { // Legacy blobs without timestamps stay untouched; the next scan rewrites them. if err := json.Unmarshal([]byte(row.Ips), &entries); err != nil { @@ -326,8 +300,8 @@ func (s *InboundService) PruneStaleClientIps() error { if len(kept) == len(entries) { continue } - b, _ := json.Marshal(kept) - if err := db.Model(&model.InboundClientIps{}).Where("id = ?", row.Id).Update("ips", string(b)).Error; err != nil { + encoded, _ := model.MarshalClientIpEntries(kept) + if err := db.Model(&model.InboundClientIps{}).Where("id = ?", row.Id).Update("ips", encoded).Error; err != nil { return err } } diff --git a/internal/web/service/inbound_client_ips_merge_test.go b/internal/web/service/inbound_client_ips_merge_test.go index 3470fac49..28a5ad868 100644 --- a/internal/web/service/inbound_client_ips_merge_test.go +++ b/internal/web/service/inbound_client_ips_merge_test.go @@ -2,27 +2,37 @@ package service import ( "encoding/json" + "fmt" "path/filepath" + "reflect" + "sync/atomic" "testing" "time" "github.com/mhsanaei/3x-ui/v3/internal/database" "github.com/mhsanaei/3x-ui/v3/internal/database/dbtest" "github.com/mhsanaei/3x-ui/v3/internal/database/model" + "github.com/mhsanaei/3x-ui/v3/internal/testpg" "gorm.io/gorm" + "gorm.io/gorm/clause" ) // setupClientIpTestDB spins up a throwaway SQLite database (migrations + seeders) // for a single test, mirroring the harness used by the other service tests. func setupClientIpTestDB(t *testing.T) { t.Helper() + cleanup, err := testpg.IsolatePackage("client_ip_sync") + if err != nil { + t.Fatal(err) + } + t.Cleanup(cleanup) dbDir := t.TempDir() t.Setenv("XUI_DB_FOLDER", dbDir) dbtest.InitDB(t, filepath.Join(dbDir, "x-ui.db")) } -func marshalIps(t *testing.T, entries ...clientIpEntry) string { +func marshalIps(t *testing.T, entries ...model.ClientIpEntry) string { t.Helper() b, err := json.Marshal(entries) if err != nil { @@ -43,7 +53,7 @@ func readClientIps(t *testing.T, email string) (map[string]int64, bool) { if err != nil { t.Fatalf("read client ips for %s: %v", email, err) } - var entries []clientIpEntry + var entries []model.ClientIpEntry if row.Ips != "" { if err := json.Unmarshal([]byte(row.Ips), &entries); err != nil { t.Fatalf("unmarshal stored ips for %s: %v", email, err) @@ -62,7 +72,7 @@ func TestMergeInboundClientIps_CreatesNodeOnlyRowIgnoringRemoteId(t *testing.T) now := time.Now().Unix() // Local client occupies id 1. - local := &model.InboundClientIps{ClientEmail: "local@x", Ips: marshalIps(t, clientIpEntry{IP: "1.1.1.1", Timestamp: now})} + local := &model.InboundClientIps{ClientEmail: "local@x", Ips: marshalIps(t, model.ClientIpEntry{IP: "1.1.1.1", Timestamp: now})} if err := db.Create(local).Error; err != nil { t.Fatalf("seed local row: %v", err) } @@ -72,7 +82,7 @@ func TestMergeInboundClientIps_CreatesNodeOnlyRowIgnoringRemoteId(t *testing.T) incoming := []model.InboundClientIps{{ Id: 1, ClientEmail: "node@x", - Ips: marshalIps(t, clientIpEntry{IP: "2.2.2.2", Timestamp: now}), + Ips: marshalIps(t, model.ClientIpEntry{IP: "2.2.2.2", Timestamp: now}), }} if err := (&InboundService{}).MergeInboundClientIps(incoming); err != nil { t.Fatalf("merge: %v", err) @@ -103,7 +113,7 @@ func TestMergeInboundClientIps_DedupKeepsMaxTimestamp(t *testing.T) { if err := db.Create(&model.InboundClientIps{ ClientEmail: "a@x", - Ips: marshalIps(t, clientIpEntry{IP: "1.1.1.1", Timestamp: now - 100}), + Ips: marshalIps(t, model.ClientIpEntry{IP: "1.1.1.1", Timestamp: now - 100}), }).Error; err != nil { t.Fatalf("seed: %v", err) } @@ -111,8 +121,8 @@ func TestMergeInboundClientIps_DedupKeepsMaxTimestamp(t *testing.T) { incoming := []model.InboundClientIps{{ ClientEmail: "a@x", Ips: marshalIps(t, - clientIpEntry{IP: "1.1.1.1", Timestamp: now - 50}, // newer than stored -> wins - clientIpEntry{IP: "2.2.2.2", Timestamp: now - 10}, + model.ClientIpEntry{IP: "1.1.1.1", Timestamp: now - 50}, // newer than stored -> wins + model.ClientIpEntry{IP: "2.2.2.2", Timestamp: now - 10}, ), }} if err := (&InboundService{}).MergeInboundClientIps(incoming); err != nil { @@ -139,8 +149,8 @@ func TestMergeInboundClientIps_DropsStaleIps(t *testing.T) { if err := db.Create(&model.InboundClientIps{ ClientEmail: "a@x", Ips: marshalIps(t, - clientIpEntry{IP: "old", Timestamp: now - 3600}, // > 30m -> stale - clientIpEntry{IP: "fresh", Timestamp: now - 60}, + model.ClientIpEntry{IP: "old", Timestamp: now - 3600}, // > 30m -> stale + model.ClientIpEntry{IP: "fresh", Timestamp: now - 60}, ), }).Error; err != nil { t.Fatalf("seed: %v", err) @@ -149,8 +159,8 @@ func TestMergeInboundClientIps_DropsStaleIps(t *testing.T) { incoming := []model.InboundClientIps{{ ClientEmail: "a@x", Ips: marshalIps(t, - clientIpEntry{IP: "incStale", Timestamp: now - 4000}, // > 30m -> stale - clientIpEntry{IP: "incFresh", Timestamp: now - 10}, + model.ClientIpEntry{IP: "incStale", Timestamp: now - 4000}, // > 30m -> stale + model.ClientIpEntry{IP: "incFresh", Timestamp: now - 10}, ), }} if err := (&InboundService{}).MergeInboundClientIps(incoming); err != nil { @@ -178,7 +188,7 @@ func TestMergeInboundClientIps_SkipsAllStaleCreate(t *testing.T) { incoming := []model.InboundClientIps{{ ClientEmail: "b@x", - Ips: marshalIps(t, clientIpEntry{IP: "1.1.1.1", Timestamp: now - 9999}), + Ips: marshalIps(t, model.ClientIpEntry{IP: "1.1.1.1", Timestamp: now - 9999}), }} if err := (&InboundService{}).MergeInboundClientIps(incoming); err != nil { t.Fatalf("merge: %v", err) @@ -194,7 +204,7 @@ func TestMergeInboundClientIps_SkipsBlankRows(t *testing.T) { now := time.Now().Unix() incoming := []model.InboundClientIps{ - {ClientEmail: "", Ips: marshalIps(t, clientIpEntry{IP: "1.1.1.1", Timestamp: now})}, + {ClientEmail: "", Ips: marshalIps(t, model.ClientIpEntry{IP: "1.1.1.1", Timestamp: now})}, {ClientEmail: "c@x", Ips: ""}, } if err := (&InboundService{}).MergeInboundClientIps(incoming); err != nil { @@ -217,13 +227,13 @@ func TestCasUpdateInboundClientIps_MatchAndMismatch(t *testing.T) { seed := &model.InboundClientIps{ ClientEmail: "cas@x", - Ips: marshalIps(t, clientIpEntry{IP: "1.1.1.1", Timestamp: now}), + Ips: marshalIps(t, model.ClientIpEntry{IP: "1.1.1.1", Timestamp: now}), } if err := db.Create(seed).Error; err != nil { t.Fatalf("seed: %v", err) } - next := marshalIps(t, clientIpEntry{IP: "2.2.2.2", Timestamp: now}) + next := marshalIps(t, model.ClientIpEntry{IP: "2.2.2.2", Timestamp: now}) ok, err := CasUpdateInboundClientIps(db, seed.Id, "not-the-blob", next) if err != nil { t.Fatalf("stale CAS: %v", err) @@ -258,13 +268,13 @@ func TestMergeInboundClientIps_RetriesAfterConcurrentWriter(t *testing.T) { seed := &model.InboundClientIps{ ClientEmail: "race@x", - Ips: marshalIps(t, clientIpEntry{IP: "10.0.0.1", Timestamp: now - 30}), + Ips: marshalIps(t, model.ClientIpEntry{IP: "10.0.0.1", Timestamp: now - 30}), } if err := db.Create(seed).Error; err != nil { t.Fatalf("seed: %v", err) } - jobBlob := marshalIps(t, clientIpEntry{IP: "10.0.0.2", Timestamp: now - 10}) + jobBlob := marshalIps(t, model.ClientIpEntry{IP: "10.0.0.2", Timestamp: now - 10}) const callback = "test:inbound_client_ips_cas_inject" injected := false if err := db.Callback().Update().Before("gorm:update").Register(callback, func(tx *gorm.DB) { @@ -294,7 +304,7 @@ func TestMergeInboundClientIps_RetriesAfterConcurrentWriter(t *testing.T) { incoming := []model.InboundClientIps{{ ClientEmail: "race@x", - Ips: marshalIps(t, clientIpEntry{IP: "10.0.0.3", Timestamp: now}), + Ips: marshalIps(t, model.ClientIpEntry{IP: "10.0.0.3", Timestamp: now}), }} if err := (&InboundService{}).MergeInboundClientIps(incoming); err != nil { t.Fatalf("merge: %v", err) @@ -316,3 +326,314 @@ func TestMergeInboundClientIps_RetriesAfterConcurrentWriter(t *testing.T) { t.Fatalf("pre-job IP should not resurrect after job replaced the blob: %v", ips) } } + +type clientIpSQLCounts struct { + selects atomic.Int64 + updates atomic.Int64 + creates atomic.Int64 + affected atomic.Int64 +} + +func countClientIpSQL(t *testing.T, table string) *clientIpSQLCounts { + t.Helper() + db := database.GetDB() + counts := &clientIpSQLCounts{} + name := "test:ip_sql_" + table + query := func(tx *gorm.DB) { + if tx.Statement.Table == table { + counts.selects.Add(1) + } + } + update := func(tx *gorm.DB) { + if tx.Statement.Table == table { + counts.updates.Add(1) + counts.affected.Add(tx.RowsAffected) + } + } + create := func(tx *gorm.DB) { + if tx.Statement.Table == table { + counts.creates.Add(1) + counts.affected.Add(tx.RowsAffected) + } + } + if err := db.Callback().Query().After("gorm:query").Register(name, query); err != nil { + t.Fatal(err) + } + if err := db.Callback().Update().After("gorm:update").Register(name, update); err != nil { + t.Fatal(err) + } + if err := db.Callback().Create().After("gorm:create").Register(name, create); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = db.Callback().Query().Remove(name) + _ = db.Callback().Update().Remove(name) + _ = db.Callback().Create().Remove(name) + }) + return counts +} + +func rawClientIpBlob(t *testing.T, entries ...model.ClientIpEntry) string { + t.Helper() + b, err := json.Marshal(entries) + if err != nil { + t.Fatal(err) + } + return string(b) +} + +func requireClientIpSet(t *testing.T, email string, want map[string]int64) { + t.Helper() + got, exists := readClientIps(t, email) + if !exists || !reflect.DeepEqual(got, want) { + t.Fatalf("%s IPs = %v (exists=%v), want %v", email, got, exists, want) + } +} + +func TestClientIpSyncSeededSnapshot(t *testing.T) { + for _, changed := range []bool{false, true} { + t.Run(fmt.Sprintf("changed=%v", changed), func(t *testing.T) { + setupClientIpTestDB(t) + now := time.Now().Unix() + rows := make([]model.InboundClientIps, 3) + incoming := make([]model.InboundClientIps, 3) + for i := range rows { + rows[i] = model.InboundClientIps{ClientEmail: fmt.Sprintf("snapshot-%d@x", i), Ips: rawClientIpBlob(t, + model.ClientIpEntry{IP: "192.0.2.1", Timestamp: now - 10}, model.ClientIpEntry{IP: "192.0.2.2", Timestamp: now})} + incoming[i] = rows[i] + if changed { + incoming[i].Ips = rawClientIpBlob(t, model.ClientIpEntry{IP: "192.0.2.2", Timestamp: now + 1}) + } + } + if err := database.GetDB().Create(&rows).Error; err != nil { + t.Fatal(err) + } + counts := countClientIpSQL(t, "inbound_client_ips") + if err := (&InboundService{}).MergeInboundClientIps(incoming); err != nil { + t.Fatal(err) + } + wantUpdates := int64(0) + if changed { + wantUpdates = 3 + } + if counts.selects.Load() != 1 || counts.updates.Load() != wantUpdates { + t.Fatalf("SELECTs/CAS attempts = %d/%d, want 1/%d", counts.selects.Load(), counts.updates.Load(), wantUpdates) + } + for _, row := range rows { + ts := now + if changed { + ts++ + } + requireClientIpSet(t, row.ClientEmail, map[string]int64{"192.0.2.1": now - 10, "192.0.2.2": ts}) + } + }) + } +} + +func TestClientIpSyncDuplicatePayloadsKeepUnion(t *testing.T) { + setupClientIpTestDB(t) + now := time.Now().Unix() + seed := model.InboundClientIps{ClientEmail: "existing@x", Ips: rawClientIpBlob(t, model.ClientIpEntry{IP: "192.0.2.1", Timestamp: now})} + if err := database.GetDB().Create(&seed).Error; err != nil { + t.Fatal(err) + } + var incoming []model.InboundClientIps + for _, email := range []string{"new@x", "existing@x"} { + for i := range 3 { + incoming = append(incoming, model.InboundClientIps{Id: 99, ClientEmail: email, Ips: rawClientIpBlob(t, model.ClientIpEntry{IP: fmt.Sprintf("192.0.2.%d", i+1), Timestamp: now + int64(i)})}) + } + } + if err := (&InboundService{}).MergeInboundClientIps(incoming); err != nil { + t.Fatal(err) + } + for _, email := range []string{"new@x", "existing@x"} { + requireClientIpSet(t, email, map[string]int64{"192.0.2.1": now, "192.0.2.2": now + 1, "192.0.2.3": now + 2}) + } +} + +func TestClientIpSyncCreateConflictMergesWinner(t *testing.T) { + setupClientIpTestDB(t) + db := database.GetDB() + now := time.Now().Unix() + winner := rawClientIpBlob(t, model.ClientIpEntry{IP: "198.51.100.1", Timestamp: now}) + const callback = "test:ip_create_winner" + injected := false + if err := db.Callback().Create().Before("gorm:create").Register(callback, func(tx *gorm.DB) { + row, ok := tx.Statement.Dest.(*model.InboundClientIps) + if injected || tx.Statement.Table != "inbound_client_ips" || !ok || row.ClientEmail != "winner@x" { + return + } + injected = true + if err := tx.Exec("INSERT INTO inbound_client_ips (client_email, ips) VALUES (?, ?)", row.ClientEmail, winner).Error; err != nil { + tx.AddError(err) + } + }); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Callback().Create().Remove(callback) }) + if err := (&InboundService{}).MergeInboundClientIps([]model.InboundClientIps{{ClientEmail: "winner@x", Ips: rawClientIpBlob(t, model.ClientIpEntry{IP: "203.0.113.1", Timestamp: now + 1})}}); err != nil { + t.Fatal(err) + } + if !injected { + t.Fatal("create conflict was not injected") + } + requireClientIpSet(t, "winner@x", map[string]int64{"198.51.100.1": now, "203.0.113.1": now + 1}) +} + +func TestClientIpSyncCASExhaustionRollsBack(t *testing.T) { + setupClientIpTestDB(t) + db := database.GetDB() + now := time.Now().Unix() + rows := []model.InboundClientIps{ + {ClientEmail: "a@x", Ips: rawClientIpBlob(t, model.ClientIpEntry{IP: "192.0.2.1", Timestamp: now})}, + {ClientEmail: "z@x", Ips: rawClientIpBlob(t, model.ClientIpEntry{IP: "192.0.2.2", Timestamp: now})}, + } + if err := db.Create(&rows).Error; err != nil { + t.Fatal(err) + } + counts := countClientIpSQL(t, "inbound_client_ips") + const callback = "test:ip_exhaustion" + attempts := 0 + if err := db.Callback().Update().Before("gorm:update").Register(callback, func(tx *gorm.DB) { + if tx.Statement.Table != "inbound_client_ips" { + return + } + where, ok := tx.Statement.Clauses["WHERE"].Expression.(clause.Where) + if !ok || len(where.Exprs) != 1 { + return + } + predicate, ok := where.Exprs[0].(clause.Expr) + if !ok || len(predicate.Vars) != 2 || predicate.Vars[0] != rows[1].Id { + return + } + attempts++ + blob := rawClientIpBlob(t, model.ClientIpEntry{IP: "198.51.100.9", Timestamp: now + int64(attempts)}) + if err := tx.Exec("UPDATE inbound_client_ips SET ips = ? WHERE id = ?", blob, rows[1].Id).Error; err != nil { + tx.AddError(err) + } + }); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Callback().Update().Remove(callback) }) + err := (&InboundService{}).MergeInboundClientIps([]model.InboundClientIps{ + {ClientEmail: "z@x", Ips: rawClientIpBlob(t, model.ClientIpEntry{IP: "203.0.113.8", Timestamp: now})}, + {ClientEmail: "a@x", Ips: rawClientIpBlob(t, model.ClientIpEntry{IP: "203.0.113.7", Timestamp: now})}, + }) + want := fmt.Sprintf("inbound_client_ips id=%d: exhausted CAS retries merging client IPs", rows[1].Id) + if err == nil || err.Error() != want { + t.Fatalf("error = %v, want %q", err, want) + } + if attempts != 8 || counts.selects.Load() != 8 || counts.updates.Load() != 9 { + t.Fatalf("attempts/selects/updates = %d/%d/%d, want 8/8/9", attempts, counts.selects.Load(), counts.updates.Load()) + } + for _, row := range rows { + var got model.InboundClientIps + if err := db.First(&got, row.Id).Error; err != nil { + t.Fatal(err) + } + if got.Ips != row.Ips { + t.Fatalf("rollback changed %s: %s, want %s", row.ClientEmail, got.Ips, row.Ips) + } + } +} + +func TestNodeClientIpCanonicalUnchangedSkipsUpsert(t *testing.T) { + setupClientIpTestDB(t) + now := time.Now().Unix() + report := map[string][]model.ClientIpEntry{"u@x": {{IP: "192.0.2.1", Timestamp: now - 1}, {IP: "192.0.2.2", Timestamp: now}}} + row := model.NodeClientIp{NodeGuid: "guid", Email: "u@x", Ips: rawClientIpBlob(t, report["u@x"]...)} + if err := database.GetDB().Create(&row).Error; err != nil { + t.Fatal(err) + } + counts := countClientIpSQL(t, "node_client_ips") + if err := (&InboundService{}).RecordLocalClientIps("guid", report); err != nil { + t.Fatal(err) + } + if counts.selects.Load() != 1 || counts.creates.Load() != 0 || counts.updates.Load() != 0 { + t.Fatalf("unchanged attribution SELECT/create/update = %d/%d/%d, want 1/0/0", counts.selects.Load(), counts.creates.Load(), counts.updates.Load()) + } + report["u@x"][1].Timestamp++ + if err := (&InboundService{}).RecordLocalClientIps("guid", report); err != nil { + t.Fatal(err) + } + if counts.creates.Load() != 1 || counts.affected.Load() != 1 { + t.Fatalf("advancing attribution attempts/affected = %d/%d, want 1/1", counts.creates.Load(), counts.affected.Load()) + } + trees, err := (&InboundService{}).GetClientIpsByGuid() + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(trees["guid"]["u@x"], report["u@x"]) { + t.Fatalf("attribution = %v, want %v", trees, report) + } +} + +func TestClientIpSyncRawSnapshotAndSingleRetry(t *testing.T) { + for _, contend := range []bool{false, true} { + t.Run(fmt.Sprintf("contention=%v", contend), func(t *testing.T) { + setupClientIpTestDB(t) + db := database.GetDB() + now := time.Now().Unix() + raw := fmt.Sprintf("[ {\"ip\":\"192.0.2.2\",\"timestamp\":%d}, {\"ip\":\"192.0.2.1\",\"timestamp\":%d} ]", now, now-1) + row := model.InboundClientIps{ClientEmail: "raw@x", Ips: raw} + if err := db.Create(&row).Error; err != nil { + t.Fatal(err) + } + counts := countClientIpSQL(t, "inbound_client_ips") + injected := false + if contend { + const callback = "test:ip_one_miss" + if err := db.Callback().Update().Before("gorm:update").Register(callback, func(tx *gorm.DB) { + if injected || tx.Statement.Table != "inbound_client_ips" { + return + } + where, ok := tx.Statement.Clauses["WHERE"].Expression.(clause.Where) + if !ok || len(where.Exprs) != 1 { + return + } + predicate, ok := where.Exprs[0].(clause.Expr) + if !ok || len(predicate.Vars) != 2 || predicate.Vars[0] != row.Id { + return + } + injected = true + winner := rawClientIpBlob(t, model.ClientIpEntry{IP: "198.51.100.1", Timestamp: now + 2}) + if err := tx.Exec("UPDATE inbound_client_ips SET ips = ? WHERE id = ?", winner, row.Id).Error; err != nil { + tx.AddError(err) + } + }); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Callback().Update().Remove(callback) }) + } + if err := (&InboundService{}).MergeInboundClientIps([]model.InboundClientIps{{ClientEmail: row.ClientEmail, Ips: rawClientIpBlob(t, model.ClientIpEntry{IP: "203.0.113.1", Timestamp: now + 1})}}); err != nil { + t.Fatal(err) + } + wantCount := int64(1) + want := map[string]int64{"192.0.2.1": now - 1, "192.0.2.2": now, "203.0.113.1": now + 1} + if contend { + wantCount = 2 + want = map[string]int64{"198.51.100.1": now + 2, "203.0.113.1": now + 1} + if !injected { + t.Fatal("competing SQL update was not injected") + } + } + if counts.selects.Load() != wantCount || counts.updates.Load() != wantCount { + t.Fatalf("SELECTs/CAS=%d/%d, want %d/%d", counts.selects.Load(), counts.updates.Load(), wantCount, wantCount) + } + requireClientIpSet(t, row.ClientEmail, want) + var got model.InboundClientIps + if err := db.First(&got, row.Id).Error; err != nil { + t.Fatal(err) + } + var entries []model.ClientIpEntry + if err := json.Unmarshal([]byte(got.Ips), &entries); err != nil { + t.Fatal(err) + } + for i := 1; i < len(entries); i++ { + if entries[i-1].Timestamp > entries[i].Timestamp || (entries[i-1].Timestamp == entries[i].Timestamp && entries[i-1].IP > entries[i].IP) { + t.Fatalf("noncanonical persisted blob %s", got.Ips) + } + } + }) + } +} diff --git a/internal/web/service/inbound_node_ips.go b/internal/web/service/inbound_node_ips.go index 2384046c1..43882d55d 100644 --- a/internal/web/service/inbound_node_ips.go +++ b/internal/web/service/inbound_node_ips.go @@ -18,9 +18,7 @@ import ( // observations under its panelGuid, and the master merges every node's // guid-keyed report — never mixing in IPs a parent pushed down. -// mergeModelClientIpEntries unions old and incoming observations, drops anything -// older than cutoff, keeps the newest timestamp per IP, and sorts newest-first. -// It mirrors mergeClientIpEntries but operates on the exported wire type. +// Attribution excludes empty addresses and keeps each fresh IP's maximum timestamp. func mergeModelClientIpEntries(old, incoming []model.ClientIpEntry, cutoff int64) []model.ClientIpEntry { ipMap := make(map[string]int64, len(old)+len(incoming)) for _, e := range old { @@ -41,13 +39,11 @@ func mergeModelClientIpEntries(old, incoming []model.ClientIpEntry, cutoff int64 for ip, ts := range ipMap { out = append(out, model.ClientIpEntry{IP: ip, Timestamp: ts}) } - sort.Slice(out, func(i, j int) bool { return out[i].Timestamp > out[j].Timestamp }) return out } -// upsertNodeClientIps folds a guid's per-email observations into NodeClientIp, -// merging with whatever is already stored for that (guid, email) and dropping -// stale entries. Empty merged results delete the row so the table stays bounded. +// upsertNodeClientIps merges fresh observations per (guid, email). +// Empty results delete attribution instead of retaining expired rows. func upsertNodeClientIps(guid string, perEmail map[string][]model.ClientIpEntry) error { if guid == "" || len(perEmail) == 0 { return nil @@ -99,8 +95,11 @@ func upsertNodeClientIps(guid string, perEmail map[string][]model.ClientIpEntry) } continue } - b, _ := json.Marshal(merged) - row := model.NodeClientIp{NodeGuid: guid, Email: email, Ips: string(b)} + encoded, _ := model.MarshalClientIpEntries(merged) + if current := existingByEmail[email]; current != nil && current.Ips == encoded { + continue + } + row := model.NodeClientIp{NodeGuid: guid, Email: email, Ips: encoded} if err := tx.Clauses(clause.OnConflict{ Columns: []clause.Column{{Name: "node_guid"}, {Name: "email"}}, DoUpdates: clause.AssignmentColumns([]string{"ips"}), @@ -112,9 +111,7 @@ func upsertNodeClientIps(guid string, perEmail map[string][]model.ClientIpEntry) return tx.Commit().Error } -// RecordLocalClientIps stores this panel's own Xray observations under its -// panelGuid. Called by check_client_ip_job each scan with the live per-email IPs -// the local core reported. +// RecordLocalClientIps stores the scanner's local observations under this panel's GUID. func (s *InboundService) RecordLocalClientIps(panelGuid string, observed map[string][]model.ClientIpEntry) error { return upsertNodeClientIps(panelGuid, observed) } @@ -172,6 +169,7 @@ func (s *InboundService) GetClientIpsByGuid() (map[string]map[string][]model.Cli if len(fresh) == 0 { continue } + model.SortClientIpEntries(fresh) if out[row.NodeGuid] == nil { out[row.NodeGuid] = make(map[string][]model.ClientIpEntry) } @@ -325,8 +323,8 @@ func pruneStaleNodeClientIps(cutoff int64) error { if len(kept) == len(entries) { continue } - b, _ := json.Marshal(kept) - if err := db.Model(&model.NodeClientIp{}).Where("id = ?", row.Id).Update("ips", string(b)).Error; err != nil { + encoded, _ := model.MarshalClientIpEntries(kept) + if err := db.Model(&model.NodeClientIp{}).Where("id = ?", row.Id).Update("ips", encoded).Error; err != nil { return err } } diff --git a/internal/web/service/inbound_node_ips_test.go b/internal/web/service/inbound_node_ips_test.go index c534b14a1..e6c331a43 100644 --- a/internal/web/service/inbound_node_ips_test.go +++ b/internal/web/service/inbound_node_ips_test.go @@ -28,8 +28,8 @@ func TestRecordLocalClientIps_RoundTripByGuid(t *testing.T) { if len(got) != 2 { t.Fatalf("want 2 entries, got %v", got) } - if got[0].IP != "1.1.1.1" { // newest-first ordering - t.Fatalf("want newest first, got %v", got) + if got[0].IP != "2.2.2.2" || got[0].Timestamp != now-10 || got[1].IP != "1.1.1.1" || got[1].Timestamp != now { + t.Fatalf("want ascending timestamps, got %v", got) } }