fix(ip-limit): CAS-retry inbound_client_ips merges under Postgres (#6612)

* fix(ip-limit): CAS-retry inbound_client_ips merges under Postgres Two writers RMW the same ips JSON blob; on PostgreSQL a lost update drops remote IPs that partitionLiveIps only sees through that blob (#6587). Compare-and-set on the previous blob with re-merge on miss, matching the repo's conditional Where+RowsAffected pattern. Fixes #6587 * ci: retrigger frontend after npm registry maintenance The frontend job failed solely on `npm audit` while registry.npmjs.org returned 503 (Service Under Maintenance). Lint, typecheck, vitest, vite build, and storybook all passed. Local `npm audit --omit=dev --audit-level=high` now reports 0 vulnerabilities. * test(ip-limit): cover the scan's CAS against a mid-scan node sync The job-side compare-and-set had no test of its own. A write injected between the scan's read and its update now has to keep the node's remote IP; main's blind Save drops it. Also keeps the new comments to two lines, as CLAUDE.md requires. --------- Co-authored-by: mrchatam <mrchatam@users.noreply.github.com> Co-authored-by: MHSanaei <ho3ein.sanaei@gmail.com>
2026-09-27 01:32:08 +03:00 · 2026-09-26 23:30:40 +03:30
parent d42e2133c7
commit 8f1201553e
4 changed files with 266 additions and 44 deletions
@@ -0,0 +1,64 @@
+package job
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/mhsanaei/3x-ui/v3/internal/database"
+	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
+
+	"gorm.io/gorm"
+)
+
+// 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)
+	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}})
+
+	nodeBlob, err := json.Marshal([]IPWithTimestamp{
+		{IP: "198.51.100.1", Timestamp: now - 60},
+		{IP: "203.0.113.77", Timestamp: now - 5},
+	})
+	if err != nil {
+		t.Fatalf("marshal node blob: %v", err)
+	}
+	const callback = "test:client_ip_scan_cas_inject"
+	injected := false
+	if err := db.Callback().Update().Before("gorm:update").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{SkipHooks: true, NewDB: true}).
+			Model(&model.InboundClientIps{}).
+			Where("client_email = ?", email).
+			Update("ips", string(nodeBlob)).Error; err != nil {
+			_ = tx.AddError(err)
+		}
+	}); err != nil {
+		t.Fatalf("register callback: %v", err)
+	}
+	t.Cleanup(func() { _ = db.Callback().Update().Remove(callback) })
+
+	NewCheckClientIpJob().processObserved(map[string]map[string]int64{
+		email: {"198.51.100.1": now},
+	}, true, true)
+	if !injected {
+		t.Fatal("inject callback never fired; the scan's write path is untested")
+	}
+
+	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)
+	}
+}
@@ -510,43 +510,58 @@ func (j *CheckClientIpJob) updateInboundClientIps(tx *gorm.DB, inboundClientIps
 		return nil, 0
 	}

-	// Parse old IPs from database
-	var oldIpsWithTime []IPWithTimestamp
-	if inboundClientIps.Ips != "" {
-		_ = json.Unmarshal([]byte(inboundClientIps.Ips), &oldIpsWithTime)
-	}
-
-	ipMap := mergeClientIps(oldIpsWithTime, newIpsWithTime, time.Now().Unix()-ipStaleAfterSeconds, observedAreLive)
-
 	// only ips seen in this scan count toward the limit. see
 	// partitionLiveIps.
 	observedThisScan := make(map[string]bool, len(newIpsWithTime))
 	for _, ipTime := range newIpsWithTime {
 		observedThisScan[ipTime.IP] = true
 	}
-	liveIps, historicalIps := partitionLiveIps(ipMap, observedThisScan)
+	staleCutoff := time.Now().Unix() - ipStaleAfterSeconds

-	// historical db-only ips are excluded from this count on purpose.
-	limitedIps, allowedIps := j.allowlist.split(liveIps)
-	keptLive, bannedLive := selectIpsToBan(limitedIps, limitIp)
-	// Allowlisted addresses stay connected and out of the count: charging them
-	// against the limit would still cut the shared network the entry protects.
-	keptLive = append(keptLive, allowedIps...)
+	// 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
+		if inboundClientIps.Ips != "" {
+			_ = json.Unmarshal([]byte(inboundClientIps.Ips), &oldIpsWithTime)
+		}
+		ipMap := mergeClientIps(oldIpsWithTime, newIpsWithTime, staleCutoff, observedAreLive)
+		liveIps, historicalIps := partitionLiveIps(ipMap, observedThisScan)

-	// keep kept-live + historical in the blob so the panel keeps showing recently
-	// seen ips; banned live ips reappear in the next scan if they reconnect.
-	dbIps := make([]IPWithTimestamp, 0, len(keptLive)+len(historicalIps))
-	dbIps = append(dbIps, keptLive...)
-	dbIps = append(dbIps, historicalIps...)
-	jsonIps, _ := json.Marshal(dbIps)
-	inboundClientIps.Ips = string(jsonIps)
+		// historical db-only ips are excluded from this count on purpose.
+		limitedIps, allowedIps := j.allowlist.split(liveIps)
+		keptLive, bannedLive := selectIpsToBan(limitedIps, limitIp)
+		// Allowlisted addresses stay connected and out of the count: charging them
+		// against the limit would still cut the shared network the entry protects.
+		keptLive = append(keptLive, allowedIps...)

-	if err := tx.Save(inboundClientIps).Error; err != nil {
-		logger.Error("failed to save inboundClientIps:", err)
-		return nil, 0
+		// keep kept-live + historical in the blob so the panel keeps showing recently
+		// seen ips; banned live ips reappear in the next scan if they reconnect.
+		dbIps := make([]IPWithTimestamp, 0, len(keptLive)+len(historicalIps))
+		dbIps = append(dbIps, keptLive...)
+		dbIps = append(dbIps, historicalIps...)
+		jsonIps, _ := json.Marshal(dbIps)
+		newIps := string(jsonIps)
+		if newIps == inboundClientIps.Ips {
+			return bannedLive, len(keptLive)
+		}
+
+		updated, err := service.CasUpdateInboundClientIps(tx, inboundClientIps.Id, inboundClientIps.Ips, newIps)
+		if err != nil {
+			logger.Error("failed to save inboundClientIps:", err)
+			return nil, 0
+		}
+		if updated {
+			inboundClientIps.Ips = newIps
+			return bannedLive, len(keptLive)
+		}
+		if err := tx.Where("id = ?", inboundClientIps.Id).First(inboundClientIps).Error; err != nil {
+			logger.Error("failed to re-read inboundClientIps after a concurrent write:", err)
+			return nil, 0
+		}
 	}
-
-	return bannedLive, len(keptLive)
+	logger.Error("failed to save inboundClientIps: exhausted CAS retries")
+	return nil, 0
 }

 // pendingBan carries one client's enforcement outcome from inside the scan's
@@ -2,6 +2,7 @@ package service

 import (
 	"encoding/json"
+	"fmt"
 	"sort"
 	"time"

@@ -156,28 +157,61 @@ func (s *InboundService) MergeInboundClientIps(incomingIps []model.InboundClient
 			continue
 		}

-		var oldEntries []clientIpEntry
-		if current.Ips != "" {
-			_ = json.Unmarshal([]byte(current.Ips), &oldEntries)
-		}
-
-		merged := mergeClientIpEntries(oldEntries, incomingEntries, cutoff)
-		b, _ := json.Marshal(merged)
-		mergedStr := string(b)
-
-		// A concurrent check_client_ip_job db.Save on the same row can interleave
-		// with this update (benign last-writer-wins; any dropped IP reappears on the
-		// next scan/sync), so only write when the blob actually changed.
-		if current.Ips != mergedStr {
-			if err := tx.Model(&model.InboundClientIps{}).Where("id = ?", current.Id).Update("ips", mergedStr).Error; err != nil {
-				tx.Rollback()
-				return err
-			}
+		// 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 {
+			tx.Rollback()
+			return err
 		}
 	}
 	return tx.Commit().Error
 }

+// ClientIpCasRetries bounds the re-reads after losing a compare-and-set. Running
+// out is an error, so the caller retries on its next schedule instead of dropping IPs.
+const ClientIpCasRetries = 8
+
+// CasUpdateInboundClientIps writes newIps only while the row still holds expectedIps;
+// updated=false with a nil error means another writer won and the caller must re-read.
+func CasUpdateInboundClientIps(tx *gorm.DB, id int, expectedIps, newIps string) (updated bool, err error) {
+	res := tx.Model(&model.InboundClientIps{}).
+		Where("id = ? AND ips = ?", id, expectedIps).
+		Update("ips", newIps)
+	if res.Error != nil {
+		return false, res.Error
+	}
+	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)
+		}
+		merged := mergeClientIpEntries(oldEntries, incoming, cutoff)
+		b, _ := json.Marshal(merged)
+		mergedStr := string(b)
+		if row.Ips == mergedStr {
+			return nil
+		}
+		ok, err := CasUpdateInboundClientIps(tx, id, row.Ips, mergedStr)
+		if err != nil {
+			return err
+		}
+		if ok {
+			return nil
+		}
+	}
+	return fmt.Errorf("inbound_client_ips id=%d: exhausted CAS retries merging client IPs", id)
+}
+
 func (s *InboundService) UpdateClientIPs(tx *gorm.DB, oldEmail string, newEmail string) error {
 	// The caller only renames onto a free identity, so a row already sitting on
 	// newEmail is stale tracking data — drop it instead of failing the edit.
@@ -8,6 +8,8 @@ import (

 	"github.com/mhsanaei/3x-ui/v3/internal/database"
 	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
+
+	"gorm.io/gorm"
 )

 // setupClientIpTestDB spins up a throwaway SQLite database (migrations + seeders)
@@ -209,3 +211,110 @@ func TestMergeInboundClientIps_SkipsBlankRows(t *testing.T) {
 		t.Fatalf("blank rows should be skipped, but %d row(s) created", count)
 	}
 }
+
+func TestCasUpdateInboundClientIps_MatchAndMismatch(t *testing.T) {
+	setupClientIpTestDB(t)
+	db := database.GetDB()
+	now := time.Now().Unix()
+
+	seed := &model.InboundClientIps{
+		ClientEmail: "cas@x",
+		Ips:         marshalIps(t, 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})
+	ok, err := CasUpdateInboundClientIps(db, seed.Id, "not-the-blob", next)
+	if err != nil {
+		t.Fatalf("stale CAS: %v", err)
+	}
+	if ok {
+		t.Fatalf("CAS with wrong expected must not update")
+	}
+	ips, _ := readClientIps(t, "cas@x")
+	if ips["1.1.1.1"] != now || len(ips) != 1 {
+		t.Fatalf("row changed on stale CAS: %v", ips)
+	}
+
+	ok, err = CasUpdateInboundClientIps(db, seed.Id, seed.Ips, next)
+	if err != nil {
+		t.Fatalf("fresh CAS: %v", err)
+	}
+	if !ok {
+		t.Fatalf("CAS with matching expected must update")
+	}
+	ips, _ = readClientIps(t, "cas@x")
+	if ips["2.2.2.2"] != now || len(ips) != 1 {
+		t.Fatalf("fresh CAS did not land: %v", ips)
+	}
+}
+
+// A job write landing between the merge's read and its Update must not drop the
+// node's report (#6587); a Before(update) hook injects it, since SQLite serializes writers.
+func TestMergeInboundClientIps_RetriesAfterConcurrentWriter(t *testing.T) {
+	setupClientIpTestDB(t)
+	db := database.GetDB()
+	now := time.Now().Unix()
+
+	seed := &model.InboundClientIps{
+		ClientEmail: "race@x",
+		Ips:         marshalIps(t, 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})
+	const callback = "test:inbound_client_ips_cas_inject"
+	injected := false
+	if err := db.Callback().Update().Before("gorm:update").Register(callback, func(tx *gorm.DB) {
+		if injected {
+			return
+		}
+		table := tx.Statement.Table
+		if table == "" && tx.Statement.Schema != nil {
+			table = tx.Statement.Schema.Table
+		}
+		if table != "inbound_client_ips" {
+			return
+		}
+		injected = true
+		// Same connection, SkipHooks: simulate the job committing a different
+		// blob before this merge's CAS Update runs.
+		if err := tx.Session(&gorm.Session{SkipHooks: true}).
+			Model(&model.InboundClientIps{}).
+			Where("id = ?", seed.Id).
+			Update("ips", jobBlob).Error; err != nil {
+			tx.AddError(err)
+		}
+	}); err != nil {
+		t.Fatalf("register callback: %v", err)
+	}
+	t.Cleanup(func() { _ = db.Callback().Update().Remove(callback) })
+
+	incoming := []model.InboundClientIps{{
+		ClientEmail: "race@x",
+		Ips:         marshalIps(t, clientIpEntry{IP: "10.0.0.3", Timestamp: now}),
+	}}
+	if err := (&InboundService{}).MergeInboundClientIps(incoming); err != nil {
+		t.Fatalf("merge: %v", err)
+	}
+	if !injected {
+		t.Fatalf("inject callback never fired; CAS path untested")
+	}
+
+	ips, _ := readClientIps(t, "race@x")
+	// After the injected job write (only .2) and the node's .3 report, both
+	// must survive. .1 was only in the pre-job snapshot and is correctly gone.
+	if _, ok := ips["10.0.0.2"]; !ok {
+		t.Fatalf("job IP lost after merge retry: %v", ips)
+	}
+	if _, ok := ips["10.0.0.3"]; !ok {
+		t.Fatalf("node IP lost (the #6587 failure mode): %v", ips)
+	}
+	if _, ok := ips["10.0.0.1"]; ok {
+		t.Fatalf("pre-job IP should not resurrect after job replaced the blob: %v", ips)
+	}
+}