From 8f1201553ea003476fe0384bb0c37484a64845f1 Mon Sep 17 00:00:00 2001 From: mrchatam <287639636+mrchatam@users.noreply.github.com> Date: Sat, 26 Sep 2026 23:30:40 +0330 Subject: [PATCH] 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 Co-authored-by: MHSanaei --- internal/web/job/check_client_ip_cas_test.go | 64 ++++++++++ internal/web/job/check_client_ip_job.go | 69 ++++++----- internal/web/service/inbound_client_ips.go | 68 ++++++++--- .../service/inbound_client_ips_merge_test.go | 109 ++++++++++++++++++ 4 files changed, 266 insertions(+), 44 deletions(-) create mode 100644 internal/web/job/check_client_ip_cas_test.go diff --git a/internal/web/job/check_client_ip_cas_test.go b/internal/web/job/check_client_ip_cas_test.go new file mode 100644 index 000000000..7b5378b3d --- /dev/null +++ b/internal/web/job/check_client_ip_cas_test.go @@ -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) + } +} diff --git a/internal/web/job/check_client_ip_job.go b/internal/web/job/check_client_ip_job.go index 471d75c8a..5e802550e 100644 --- a/internal/web/job/check_client_ip_job.go +++ b/internal/web/job/check_client_ip_job.go @@ -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 diff --git a/internal/web/service/inbound_client_ips.go b/internal/web/service/inbound_client_ips.go index bb281deb1..ad4daa374 100644 --- a/internal/web/service/inbound_client_ips.go +++ b/internal/web/service/inbound_client_ips.go @@ -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. diff --git a/internal/web/service/inbound_client_ips_merge_test.go b/internal/web/service/inbound_client_ips_merge_test.go index 06a30d19c..304dfc7f3 100644 --- a/internal/web/service/inbound_client_ips_merge_test.go +++ b/internal/web/service/inbound_client_ips_merge_test.go @@ -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) + } +}