diff --git a/internal/database/db.go b/internal/database/db.go index 6e82cf41f..956bbeb76 100644 --- a/internal/database/db.go +++ b/internal/database/db.go @@ -140,6 +140,9 @@ func initModels() error { if err := migrateTgIDIndex(); err != nil { return err } + if err := migrateSyncOrphanColumns(); err != nil { + return err + } if IsPostgres() { if err := resyncPostgresSequences(db, models); err != nil { log.Printf("Error resyncing postgres sequences: %v", err) @@ -297,6 +300,15 @@ func rebuildInboundsWithoutInlineUniquePort() error { }) } +// AutoMigrate adds the column; this only backfills the NULLs an older SQLite +// ALTER TABLE leaves behind, so the reaper's predicate never compares to NULL. +func migrateSyncOrphanColumns() error { + if !db.Migrator().HasColumn(&model.ClientRecord{}, "sync_orphaned_at") { + return nil + } + return db.Exec("UPDATE clients SET sync_orphaned_at = 0 WHERE sync_orphaned_at IS NULL").Error +} + func migrateHostVerifyPeerCertByNameColumn() error { if !db.Migrator().HasColumn(&model.Host{}, "verify_peer_cert_by_name") { return nil diff --git a/internal/database/model/model.go b/internal/database/model/model.go index c2036d900..1da404d1f 100644 --- a/internal/database/model/model.go +++ b/internal/database/model/model.go @@ -911,6 +911,9 @@ type ClientRecord struct { Reset int `json:"reset" gorm:"default:0"` CreatedAt int64 `json:"createdAt" gorm:"autoCreateTime:milli"` UpdatedAt int64 `json:"updatedAt" gorm:"autoUpdateTime:milli"` + // Owned solely by the node-snapshot sweep, which soft-orphans instead of + // deleting; orphans from any other cause stay at zero and are never reaped. + SyncOrphanedAt int64 `json:"-" gorm:"column:sync_orphaned_at;default:0"` } func (ClientRecord) TableName() string { return "clients" } diff --git a/internal/web/job/reap_sync_orphans_job.go b/internal/web/job/reap_sync_orphans_job.go new file mode 100644 index 000000000..4071fdf0d --- /dev/null +++ b/internal/web/job/reap_sync_orphans_job.go @@ -0,0 +1,27 @@ +package job + +import ( + "github.com/mhsanaei/3x-ui/v3/internal/logger" + "github.com/mhsanaei/3x-ui/v3/internal/web/service" +) + +// ReapSyncOrphansJob hard-deletes the clients the node-snapshot sweep only +// soft-orphaned, once enough clean merges have agreed they are really gone. +type ReapSyncOrphansJob struct { + clientService service.ClientService +} + +func NewReapSyncOrphansJob() *ReapSyncOrphansJob { + return new(ReapSyncOrphansJob) +} + +func (j *ReapSyncOrphansJob) Run() { + reaped, err := j.clientService.ReapSyncOrphans() + if err != nil { + logger.Warning("reap sync orphans failed:", err) + return + } + if reaped > 0 { + logger.Infof("reap sync orphans: removed %d client(s)", reaped) + } +} diff --git a/internal/web/service/client_bulk.go b/internal/web/service/client_bulk.go index 1aed052e5..23d62c3db 100644 --- a/internal/web/service/client_bulk.go +++ b/internal/web/service/client_bulk.go @@ -835,13 +835,16 @@ func (s *ClientService) BulkDelete(inboundSvc *InboundService, emails []string, successEmails := make([]string, 0, len(recordsByEmail)) successIds := make([]int, 0, len(recordsByEmail)) + failedEmails := make([]string, 0, len(recordsByEmail)) for email, rec := range recordsByEmail { if _, skipped := skippedReasons[email]; skipped { + failedEmails = append(failedEmails, email) continue } successEmails = append(successEmails, email) successIds = append(successIds, rec.Id) } + withdrawClientTombstones(failedEmails...) if len(successIds) > 0 { // Serialize the row cleanup against the traffic poll to avoid the @@ -875,6 +878,7 @@ func (s *ClientService) BulkDelete(inboundSvc *InboundService, emails []string, } return nil }); err != nil { + withdrawClientTombstones(successEmails...) return result, needRestart, err } } diff --git a/internal/web/service/client_crud.go b/internal/web/service/client_crud.go index 3bb3498e6..9f9eaff0b 100644 --- a/internal/web/service/client_crud.go +++ b/internal/web/service/client_crud.go @@ -7,6 +7,7 @@ import ( "fmt" "strings" "time" + "unicode" "github.com/google/uuid" @@ -21,7 +22,7 @@ import ( func hasForbiddenClientChar(s string) bool { for _, r := range s { - if r == '/' || r == '\\' || r == ' ' || r < 0x20 || r == 0x7f { + if r == '/' || r == '\\' || r < 0x20 || r == 0x7f || unicode.IsSpace(r) { return true } } @@ -523,6 +524,7 @@ func (s *ClientService) Delete(inboundSvc *InboundService, id int, keepTraffic b inboundIds, err := s.GetInboundIdsForRecord(id) if err != nil { + withdrawClientTombstones(existing.Email) return false, err } @@ -560,7 +562,9 @@ func (s *ClientService) Delete(inboundSvc *InboundService, id int, keepTraffic b } // A failed inbound still holds the client in its settings JSON: keep the // record so the next delete retries exactly the leftovers, and report it. + // The tombstone lifts with it, or the next node merge finishes the deletion. if len(delErrs) > 0 { + withdrawClientTombstones(existing.Email) return needRestart, errors.Join(delErrs...) } @@ -593,6 +597,7 @@ func (s *ClientService) Delete(inboundSvc *InboundService, id int, keepTraffic b } return tx.Delete(&model.ClientRecord{}, id).Error }); err != nil { + withdrawClientTombstones(existing.Email) return needRestart, err } return needRestart, nil diff --git a/internal/web/service/client_delete_tombstone_test.go b/internal/web/service/client_delete_tombstone_test.go new file mode 100644 index 000000000..be3e08aca --- /dev/null +++ b/internal/web/service/client_delete_tombstone_test.go @@ -0,0 +1,44 @@ +package service + +import ( + "testing" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" +) + +// Delete tombstones up front and keeps the record when an inbound fails. A +// surviving tombstone lets the next node merge finish the refused deletion. +func TestFailedDeleteWithdrawsTombstone(t *testing.T) { + setupBulkDB(t) + svc := &ClientService{} + inboundSvc := &InboundService{} + db := database.GetDB() + + const email = "retry@x" + broken := mkInbound(t, 30401, model.VLESS, `{"clients": [ THIS IS NOT JSON`) + rec := &model.ClientRecord{Email: email, Enable: true, UUID: "33333333-3333-3333-3333-333333333333"} + if err := db.Create(rec).Error; err != nil { + t.Fatalf("create client record: %v", err) + } + if err := db.Create(&model.ClientInbound{ClientId: rec.Id, InboundId: broken.Id}).Error; err != nil { + t.Fatalf("attach client: %v", err) + } + + t.Cleanup(func() { withdrawClientTombstones(email) }) + + if _, err := svc.Delete(inboundSvc, rec.Id, false); err == nil { + t.Fatal("setup: delete was expected to fail on the unparseable inbound settings") + } + + var surviving int64 + if err := db.Model(&model.ClientRecord{}).Where("email = ?", email).Count(&surviving).Error; err != nil { + t.Fatalf("count clients: %v", err) + } + if surviving != 1 { + t.Fatalf("failed delete must keep the record for a retry, got %d rows", surviving) + } + if isClientEmailTombstoned(email) { + t.Fatal("delete kept the record but left a live tombstone: the next node sync would finish the deletion it refused") + } +} diff --git a/internal/web/service/client_link.go b/internal/web/service/client_link.go index 11a10e4d1..95b34e2d3 100644 --- a/internal/web/service/client_link.go +++ b/internal/web/service/client_link.go @@ -115,6 +115,9 @@ func (s *ClientService) SyncInbound(tx *gorm.DB, inboundId int, clients []model. } incoming := clients[i].ToRecord() + // ToRecord copies the raw email; store the trimmed key this function + // looks up by, or a padded email is inserted and never found again. + incoming.Email = email row, ok := existing[email] if !ok { if _, dup := pending[email]; !dup { diff --git a/internal/web/service/client_locks.go b/internal/web/service/client_locks.go index 61f26c0b0..b9d7d10ee 100644 --- a/internal/web/service/client_locks.go +++ b/internal/web/service/client_locks.go @@ -105,6 +105,19 @@ func tombstoneClientEmail(email string) { } } +func withdrawClientTombstones(emails ...string) { + if len(emails) == 0 { + return + } + recentlyDeletedMu.Lock() + defer recentlyDeletedMu.Unlock() + for _, email := range emails { + if email != "" { + delete(recentlyDeleted, email) + } + } +} + func tombstoneClientEmails(emails []string) { if len(emails) == 0 { return diff --git a/internal/web/service/client_sync_orphan.go b/internal/web/service/client_sync_orphan.go new file mode 100644 index 000000000..17b780ff8 --- /dev/null +++ b/internal/web/service/client_sync_orphan.go @@ -0,0 +1,85 @@ +package service + +import ( + "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/logger" + "github.com/mhsanaei/3x-ui/v3/internal/xray" + + "gorm.io/gorm" +) + +// How long a client stays recoverable after a node merge concluded it is gone. +// Any merge that sees it attached again inside this window clears the mark. +const syncOrphanReapGrace = 15 * time.Minute + +// A traffic row with no clients row behind it is drift the orphan sweep will +// never mark, so it stays on the old delete-immediately path. +func clientRecordExists(tx *gorm.DB, email string) bool { + var n int64 + if err := tx.Model(&model.ClientRecord{}).Where("email = ?", email).Count(&n).Error; err != nil { + return false + } + return n > 0 +} + +func markSyncOrphan(tx *gorm.DB, email string, nowMs int64) error { + if email == "" { + return nil + } + return tx.Model(&model.ClientRecord{}). + Where("email = ? AND sync_orphaned_at = 0", email). + Update("sync_orphaned_at", nowMs).Error +} + +// A client that is attached again is not orphaned, whatever an earlier merge +// concluded — this is what makes a bad merge recoverable instead of fatal. +func clearSyncOrphanMarks(tx *gorm.DB) error { + return tx.Model(&model.ClientRecord{}). + Where("sync_orphaned_at > 0 AND EXISTS (SELECT 1 FROM client_inbounds WHERE client_inbounds.client_id = clients.id)"). + Update("sync_orphaned_at", 0).Error +} + +// ReapSyncOrphans deletes the clients the node-snapshot sweep marked and no +// later merge reclaimed. It is the only path that hard-deletes for that sweep. +func (s *ClientService) ReapSyncOrphans() (int, error) { + db := database.GetDB() + cutoff := time.Now().Add(-syncOrphanReapGrace).UnixMilli() + + var emails []string + if err := db.Model(&model.ClientRecord{}). + Where("sync_orphaned_at > 0 AND sync_orphaned_at <= ?", cutoff). + Where("NOT EXISTS (SELECT 1 FROM client_inbounds WHERE client_inbounds.client_id = clients.id)"). + Pluck("email", &emails).Error; err != nil { + return 0, err + } + if len(emails) == 0 { + return 0, nil + } + + reaped := 0 + for _, batch := range chunkStrings(emails, sqlInChunk) { + if err := runSerializedTx(func(tx *gorm.DB) error { + if err := adjustGroupBaselinesForRemovedTraffic(tx, batch); err != nil { + return err + } + if err := tx.Where("email IN ?", batch).Delete(&model.ClientRecord{}).Error; err != nil { + return err + } + if err := tx.Where("email IN ?", batch).Delete(&xray.ClientTraffic{}).Error; err != nil { + return err + } + if err := tx.Where("email IN ?", batch).Delete(&model.NodeClientTraffic{}).Error; err != nil { + return err + } + return tx.Where("client_email IN ?", batch).Delete(&model.InboundClientIps{}).Error + }); err != nil { + return reaped, err + } + reaped += len(batch) + logger.Infof("reaped %d client(s) confirmed removed on their node", len(batch)) + } + return reaped, nil +} diff --git a/internal/web/service/client_sync_orphan_test.go b/internal/web/service/client_sync_orphan_test.go new file mode 100644 index 000000000..57d06de35 --- /dev/null +++ b/internal/web/service/client_sync_orphan_test.go @@ -0,0 +1,148 @@ +package service + +import ( + "fmt" + "testing" + "time" + + "github.com/mhsanaei/3x-ui/v3/internal/database/model" + "github.com/mhsanaei/3x-ui/v3/internal/xray" + + "gorm.io/gorm" +) + +func readOrphanMark(t *testing.T, db *gorm.DB, email string) int64 { + t.Helper() + var row model.ClientRecord + if err := db.Where("email = ?", email).First(&row).Error; err != nil { + t.Fatalf("read client %q: %v", email, err) + } + return row.SyncOrphanedAt +} + +func backdateOrphanMark(t *testing.T, db *gorm.DB, email string) { + t.Helper() + past := time.Now().Add(-2 * syncOrphanReapGrace).UnixMilli() + if err := db.Model(&model.ClientRecord{}). + Where("email = ?", email). + Update("sync_orphaned_at", past).Error; err != nil { + t.Fatalf("backdate orphan mark: %v", err) + } +} + +// The merge must soft-orphan, not delete: everything stays recoverable until +// the grace period has elapsed and the reaper confirms nothing reclaimed it. +func TestSyncOrphanSurvivesMergeUntilGraceElapses(t *testing.T) { + db := initTrafficTestDB(t) + svc := &InboundService{} + clientSvc := &ClientService{} + + seedNodeRow(t, db, &model.Node{Id: 1, Name: "n1", Address: "127.0.0.1", Port: 2096, ApiToken: "tok", Enable: true}) + + const email = "gone@x" + createNodeInboundWithClient(t, db, 1, "n1-in", 41001, email) + settings := fmt.Sprintf(`{"clients":[{"email":%q,"enable":true}]}`, email) + syncNodeWithSettings(t, svc, 1, "n1-in", settings, + xray.ClientTraffic{Email: email, Up: 5, Down: 5, Enable: true}) + + if rec, traf := countClientRows(t, db, email); rec != 1 || traf != 1 { + t.Fatalf("setup: clients=%d client_traffics=%d, want 1/1", rec, traf) + } + + if _, err := svc.setRemoteTrafficLocked(1, snapshotWithoutClients(t, "n1-in"), false); err != nil { + t.Fatalf("orphaning merge: %v", err) + } + if rec, traf := countClientRows(t, db, email); rec != 1 || traf != 1 { + t.Fatalf("merge hard-deleted the client: clients=%d client_traffics=%d, want 1/1", rec, traf) + } + if readOrphanMark(t, db, email) <= 0 { + t.Fatal("merge did not stamp sync_orphaned_at") + } + + reaped, err := clientSvc.ReapSyncOrphans() + if err != nil { + t.Fatalf("reap inside grace: %v", err) + } + if reaped != 0 { + t.Fatalf("reaped %d client(s) inside the grace period, want 0", reaped) + } + if rec, _ := countClientRows(t, db, email); rec != 1 { + t.Fatal("client removed before the grace period elapsed") + } + + backdateOrphanMark(t, db, email) + reaped, err = clientSvc.ReapSyncOrphans() + if err != nil { + t.Fatalf("reap after grace: %v", err) + } + if reaped != 1 { + t.Fatalf("reaped %d client(s) after the grace period, want 1", reaped) + } + rec, traf := countClientRows(t, db, email) + if rec != 0 || traf != 0 { + t.Fatalf("after reap: clients=%d client_traffics=%d, want 0/0", rec, traf) + } +} + +// A client the node reports again was never gone: clearing the mark is what +// turns a bad merge into a recoverable blip instead of a delayed deletion. +func TestSyncOrphanMarkClearedOnReattach(t *testing.T) { + db := initTrafficTestDB(t) + svc := &InboundService{} + clientSvc := &ClientService{} + + seedNodeRow(t, db, &model.Node{Id: 1, Name: "n1", Address: "127.0.0.1", Port: 2096, ApiToken: "tok", Enable: true}) + + const email = "flaky@x" + createNodeInboundWithClient(t, db, 1, "n1-in", 41001, email) + settings := fmt.Sprintf(`{"clients":[{"email":%q,"enable":true}]}`, email) + syncNodeWithSettings(t, svc, 1, "n1-in", settings, + xray.ClientTraffic{Email: email, Up: 5, Down: 5, Enable: true}) + + if _, err := svc.setRemoteTrafficLocked(1, snapshotWithoutClients(t, "n1-in"), false); err != nil { + t.Fatalf("orphaning merge: %v", err) + } + if readOrphanMark(t, db, email) <= 0 { + t.Fatal("setup: expected the merge to mark the client") + } + + syncNodeWithSettings(t, svc, 1, "n1-in", settings, + xray.ClientTraffic{Email: email, Up: 6, Down: 6, Enable: true}) + + if orphanedAt := readOrphanMark(t, db, email); orphanedAt != 0 { + t.Fatalf("re-attached client kept its orphan mark: sync_orphaned_at=%d", orphanedAt) + } + + backdateOrphanMark(t, db, email) + if reaped, err := clientSvc.ReapSyncOrphans(); err != nil || reaped != 0 { + t.Fatalf("reaped %d client(s) (err=%v) that the node still reports, want 0", reaped, err) + } +} + +// The reaper is scoped to the node sweep. Orphans from any other cause carry no +// mark and keep their existing manual-cleanup semantics. +func TestReapSyncOrphansIgnoresUnmarkedOrphans(t *testing.T) { + db := initTrafficTestDB(t) + clientSvc := &ClientService{} + + const email = "manual@x" + rec := &model.ClientRecord{Email: email, Enable: true, UUID: "44444444-4444-4444-4444-444444444444"} + if err := db.Create(rec).Error; err != nil { + t.Fatalf("create client: %v", err) + } + + reaped, err := clientSvc.ReapSyncOrphans() + if err != nil { + t.Fatalf("reap: %v", err) + } + if reaped != 0 { + t.Fatalf("reaped %d unmarked orphan(s), want 0", reaped) + } + var surviving int64 + if err := db.Model(&model.ClientRecord{}).Where("email = ?", email).Count(&surviving).Error; err != nil { + t.Fatalf("count clients: %v", err) + } + if surviving != 1 { + t.Fatalf("unmarked orphan was reaped: %d rows survive, want 1", surviving) + } +} diff --git a/internal/web/service/inbound.go b/internal/web/service/inbound.go index 336c19d26..d4d975196 100644 --- a/internal/web/service/inbound.go +++ b/internal/web/service/inbound.go @@ -1078,7 +1078,7 @@ func (s *InboundService) AddInbound(inbound *model.Inbound) (*model.Inbound, boo payload := inbound pushable := true if inbound.Protocol == model.MTProto { - if built, bErr := s.buildRuntimeInboundForAPI(tx, inbound); bErr == nil { + if built, bErr := s.buildInboundForLocalRuntime(tx, inbound); bErr == nil { payload = built } else { logger.Debug("Unable to prepare runtime inbound config:", bErr) @@ -1326,7 +1326,7 @@ func (s *InboundService) SetInboundEnable(id int, enable bool) (bool, error) { return needRestart, nil } - runtimeInbound, err := s.buildRuntimeInboundForAPI(db, inbound) + runtimeInbound, err := s.buildInboundForLocalRuntime(db, inbound) if err != nil { logger.Debug("SetInboundEnable: build runtime config failed:", err) return true, nil @@ -1511,7 +1511,7 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound, payload := oldInbound pushable := true if inbound.Enable { - if built, err2 := s.buildRuntimeInboundForAPI(tx, oldInbound); err2 == nil { + if built, err2 := s.buildInboundForLocalRuntime(tx, oldInbound); err2 == nil { payload = built } else { logger.Debug("Unable to prepare runtime inbound config:", err2) @@ -1537,7 +1537,7 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound, var runtimeInbound *model.Inbound if inbound.Enable { var err2 error - runtimeInbound, err2 = s.buildRuntimeInboundForAPI(tx, oldInbound) + runtimeInbound, err2 = s.buildInboundForLocalRuntime(tx, oldInbound) if err2 != nil { logger.Debug("Unable to prepare runtime inbound config:", err2) needRestart = true @@ -1608,81 +1608,95 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound, return inbound, needRestart, nil } -func (s *InboundService) buildRuntimeInboundForAPI(tx *gorm.DB, inbound *model.Inbound) (*model.Inbound, error) { +// A node mirrors this payload into its own DB, so every client must survive: +// filtering one out makes the node delete it, and the master then mirrors that. +func (s *InboundService) buildInboundForNodePush(tx *gorm.DB, inbound *model.Inbound) (*model.Inbound, error) { if inbound == nil { return nil, fmt.Errorf("inbound is nil") } - runtimeInbound := *inbound + built := *inbound settings := map[string]any{} if err := json.Unmarshal([]byte(inbound.Settings), &settings); err != nil { return nil, err } - mutated := false - if clients, ok := settings["clients"].([]any); ok { - var clientStats []xray.ClientTraffic - err := tx.Model(xray.ClientTraffic{}). - Where("inbound_id = ?", inbound.Id). - Select("email", "enable"). - Find(&clientStats).Error - if err != nil { - return nil, err - } + if !inboundCanHostFallbacks(inbound) { + return &built, nil + } + fallbacks, err := s.fallbackService.BuildFallbacksJSON(tx, inbound.Id) + if err != nil { + return nil, err + } + if len(fallbacks) == 0 { + return &built, nil + } + generic := make([]any, 0, len(fallbacks)) + for _, f := range fallbacks { + generic = append(generic, f) + } + settings["fallbacks"] = generic - enableMap := make(map[string]bool, len(clientStats)) - for _, clientTraffic := range clientStats { - enableMap[clientTraffic.Email] = clientTraffic.Enable - } + modifiedSettings, mErr := json.MarshalIndent(settings, "", " ") + if mErr != nil { + return nil, mErr + } + built.Settings = string(modifiedSettings) + return &built, nil +} - finalClients := make([]any, 0, len(clients)) - for _, client := range clients { - c, ok := client.(map[string]any) - if !ok { - continue - } - - email, _ := c["email"].(string) - if enable, exists := enableMap[email]; exists && !enable { - continue - } - - if manualEnable, ok := c["enable"].(bool); ok && !manualEnable { - continue - } - - finalClients = append(finalClients, c) - } - - settings["clients"] = finalClients - mutated = true +// Strips disabled clients on top of the node payload. Safe only because the +// target here is an in-memory Xray/mtg config, not another panel's database. +func (s *InboundService) buildInboundForLocalRuntime(tx *gorm.DB, inbound *model.Inbound) (*model.Inbound, error) { + built, err := s.buildInboundForNodePush(tx, inbound) + if err != nil { + return nil, err } - if inboundCanHostFallbacks(inbound) { - fallbacks, fbErr := s.fallbackService.BuildFallbacksJSON(tx, inbound.Id) - if fbErr != nil { - return nil, fbErr - } - if len(fallbacks) > 0 { - generic := make([]any, 0, len(fallbacks)) - for _, f := range fallbacks { - generic = append(generic, f) - } - settings["fallbacks"] = generic - mutated = true - } + settings := map[string]any{} + if err := json.Unmarshal([]byte(built.Settings), &settings); err != nil { + return nil, err + } + clients, ok := settings["clients"].([]any) + if !ok { + return built, nil } - if !mutated { - return &runtimeInbound, nil + var clientStats []xray.ClientTraffic + if err := tx.Model(xray.ClientTraffic{}). + Where("inbound_id = ?", built.Id). + Select("email", "enable"). + Find(&clientStats).Error; err != nil { + return nil, err } + enableMap := make(map[string]bool, len(clientStats)) + for _, clientTraffic := range clientStats { + enableMap[clientTraffic.Email] = clientTraffic.Enable + } + + finalClients := make([]any, 0, len(clients)) + for _, client := range clients { + c, ok := client.(map[string]any) + if !ok { + continue + } + email, _ := c["email"].(string) + if enable, exists := enableMap[email]; exists && !enable { + continue + } + if manualEnable, ok := c["enable"].(bool); ok && !manualEnable { + continue + } + finalClients = append(finalClients, c) + } + settings["clients"] = finalClients modifiedSettings, err := json.MarshalIndent(settings, "", " ") if err != nil { return nil, err } + runtimeInbound := *built runtimeInbound.Settings = string(modifiedSettings) - return &runtimeInbound, nil } diff --git a/internal/web/service/inbound_fallback_runtime_test.go b/internal/web/service/inbound_fallback_runtime_test.go index 68798e890..5ad2e733e 100644 --- a/internal/web/service/inbound_fallback_runtime_test.go +++ b/internal/web/service/inbound_fallback_runtime_test.go @@ -31,9 +31,9 @@ func TestBuildRuntimeInboundForAPI_InjectsFallbacks(t *testing.T) { t.Fatalf("seed fallback: %v", err) } - runtimeIb, err := svc.buildRuntimeInboundForAPI(db, master) + runtimeIb, err := svc.buildInboundForLocalRuntime(db, master) if err != nil { - t.Fatalf("buildRuntimeInboundForAPI: %v", err) + t.Fatalf("buildInboundForLocalRuntime: %v", err) } var settings map[string]any @@ -68,9 +68,9 @@ func TestBuildRuntimeInboundForAPI_NoFallbacksOnWsInbound(t *testing.T) { t.Fatalf("seed fallback: %v", err) } - runtimeIb, err := svc.buildRuntimeInboundForAPI(db, ib) + runtimeIb, err := svc.buildInboundForLocalRuntime(db, ib) if err != nil { - t.Fatalf("buildRuntimeInboundForAPI: %v", err) + t.Fatalf("buildInboundForLocalRuntime: %v", err) } if strings.Contains(runtimeIb.Settings, "fallbacks") { t.Fatalf("ws inbound must not receive fallbacks: %s", runtimeIb.Settings) diff --git a/internal/web/service/inbound_mtproto.go b/internal/web/service/inbound_mtproto.go index 839008b9b..592fafbdc 100644 --- a/internal/web/service/inbound_mtproto.go +++ b/internal/web/service/inbound_mtproto.go @@ -14,7 +14,7 @@ import ( // running: one instance per enabled local mtproto inbound, serving only the // secrets of clients that are both enabled in the inbound settings and not // depletion-disabled in client_traffics. That is the same effective client set -// buildRuntimeInboundForAPI pushes on interactive edits, so the reconcile job +// buildInboundForLocalRuntime pushes on interactive edits, so the reconcile job // and the push paths agree on one fingerprint — a disagreement would surface // as a needless mtg restart, and a job that read only the raw settings would // keep serving depleted clients until an unrelated restart. Inbounds whose @@ -95,7 +95,7 @@ func (s *InboundService) applyLocalMtproto(inboundId int) { } payload := inbound if inbound.Enable { - if built, bErr := s.buildRuntimeInboundForAPI(database.GetDB(), inbound); bErr == nil { + if built, bErr := s.buildInboundForLocalRuntime(database.GetDB(), inbound); bErr == nil { payload = built } } diff --git a/internal/web/service/inbound_mtproto_desired_test.go b/internal/web/service/inbound_mtproto_desired_test.go index c903b3b0f..f0501a920 100644 --- a/internal/web/service/inbound_mtproto_desired_test.go +++ b/internal/web/service/inbound_mtproto_desired_test.go @@ -55,9 +55,9 @@ func TestDesiredMtprotoInstancesFiltersDepleted(t *testing.T) { }) t.Run("matchesInteractivePushFiltering", func(t *testing.T) { - built, err := svc.buildRuntimeInboundForAPI(database.GetDB(), served) + built, err := svc.buildInboundForLocalRuntime(database.GetDB(), served) if err != nil { - t.Fatalf("buildRuntimeInboundForAPI: %v", err) + t.Fatalf("buildInboundForLocalRuntime: %v", err) } pushInst, ok := mtproto.InstanceFromInbound(built) if !ok { diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index c5e7fc4e1..9f69a44d3 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -125,11 +125,8 @@ func (s *InboundService) ReconcileNode(ctx context.Context, rt *runtime.Remote, } } } - // Reconcile with the same runtime-built payload interactive pushes - // send (disabled clients filtered, settings.fallbacks injected) so - // fingerprints line up and fallback edits actually reach the node. runtimeIb := ib - if built, bErr := s.buildRuntimeInboundForAPI(db, ib); bErr == nil { + if built, bErr := s.buildInboundForNodePush(db, ib); bErr == nil { runtimeIb = built } if _, err := rt.ReconcileInbound(ctx, runtimeIb, existsOnNode); err != nil { @@ -352,7 +349,12 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi // origin (an inbound the node forwards from its own sub-node) is kept as-is, // so a chained Node1->Node2->Node3 still attributes Node3's inbounds to Node3. var nodeRow model.Node - db.Select("guid").Where("id = ?", nodeID).First(&nodeRow) + db.Select("guid", "config_dirty", "inbound_sync_mode", "inbound_tags").Where("id = ?", nodeID).First(&nodeRow) + // Re-read inside the serialized writer: a client added while this snapshot + // was in flight marks the node dirty after the caller sampled the flag. + dirty = dirty || nodeRow.ConfigDirty + nodeRow.Id = nodeID + unmanagedTag := unmanagedTagPredicate(&nodeRow) selfKey := effectiveNodeKey(&model.Node{Id: nodeID, Guid: nodeRow.Guid}) guidShared := nodeRow.Guid != "" && selfKey != nodeRow.Guid originGuidFor := func(snapIb *model.Inbound) string { @@ -671,6 +673,9 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi if _, kept := snapTags[c.Tag]; kept { continue } + if unmanagedTag(c.Tag) { + continue + } var goneEmails []string if err := tx.Model(xray.ClientTraffic{}). Where("inbound_id = ?", c.Id). @@ -886,7 +891,9 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi if uErr != nil { return false, uErr } - if !stillUsed { + // Usage, quota and expiry live on this row, so a client the orphan + // sweep will mark keeps it until the reaper confirms the removal. + if !stillUsed && !clientRecordExists(tx, existing.Email) { if err := tx.Where("inbound_id = ? AND email = ?", c.Id, existing.Email). Delete(&xray.ClientTraffic{}).Error; err != nil { return false, err @@ -901,6 +908,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi emails map[string]struct{} } var perInboundOld []oldSet + syncFailedInbounds := map[int]struct{}{} for _, snapIb := range snap.Inbounds { if snapIb == nil { continue @@ -973,10 +981,16 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } if err := s.clientService.SyncInbound(tx, c.Id, filtered); err != nil { logger.Warningf("setRemoteTraffic: sync clients for tag %q failed: %v", snapIb.Tag, err) + syncFailedInbounds[c.Id] = struct{}{} } } for _, old := range perInboundOld { + // The sweep's premise is that links were just rebuilt from the snapshot, + // which is exactly what a failed SyncInbound violates. + if _, failed := syncFailedInbounds[old.inboundID]; failed { + continue + } var stillAttached []string if err := tx.Table("clients"). Joins("JOIN client_inbounds ON client_inbounds.client_id = clients.id"). @@ -1002,19 +1016,20 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi if attachmentCount > 0 { continue } - if err := tx.Where("email = ?", email).Delete(&model.ClientRecord{}).Error; err != nil { - logger.Warningf("setRemoteTraffic: delete ClientRecord %q failed: %v", email, err) - } - if err := tx.Where("email = ?", email).Delete(&xray.ClientTraffic{}).Error; err != nil { - logger.Warningf("setRemoteTraffic: delete ClientTraffic %q failed: %v", email, err) - } - if err := tx.Where("email = ?", email).Delete(&model.NodeClientTraffic{}).Error; err != nil { - logger.Warningf("setRemoteTraffic: delete NodeClientTraffic %q failed: %v", email, err) + // "Ended the merge unattached" is true for a real remote deletion and + // equally true for a bad merge, so record a strike instead of deleting. + if err := markSyncOrphan(tx, email, now); err != nil { + logger.Warningf("setRemoteTraffic: mark orphan %q failed: %v", email, err) + continue } structuralChange = true } } + if err := clearSyncOrphanMarks(tx); err != nil { + logger.Warning("setRemoteTraffic: clear orphan marks failed:", err) + } + if err := liftActivatedClientRecordExpiries(tx); err != nil { logger.Warning("setRemoteTraffic: lift activated expiries failed:", err) } diff --git a/internal/web/service/node.go b/internal/web/service/node.go index 52c93c726..a22a40d7a 100644 --- a/internal/web/service/node.go +++ b/internal/web/service/node.go @@ -652,9 +652,9 @@ func (s *NodeService) EnsureInboundTagAllowedTx(tx *gorm.DB, nodeID int, tag str Updates(map[string]any{"inbound_tags": string(buf)}).Error } -func FilterNodeSnapshot(n *model.Node, snap *runtime.TrafficSnapshot) { - if n == nil || snap == nil || n.InboundSyncMode != "selected" { - return +func nodeSelectedTagSet(n *model.Node) map[string]struct{} { + if n == nil || n.InboundSyncMode != "selected" { + return nil } prefix := nodeTagPrefix(&n.Id) allowed := make(map[string]struct{}, len(n.InboundTags)*2) @@ -668,6 +668,27 @@ func FilterNodeSnapshot(n *model.Node, snap *runtime.TrafficSnapshot) { } } } + return allowed +} + +// A deselected tag is still served by the node — FilterNodeSnapshot just stops +// reporting it — so its absence must never be read as "the node deleted it". +func unmanagedTagPredicate(n *model.Node) func(string) bool { + managed := nodeSelectedTagSet(n) + if managed == nil { + return func(string) bool { return false } + } + return func(tag string) bool { + _, ok := managed[tag] + return !ok + } +} + +func FilterNodeSnapshot(n *model.Node, snap *runtime.TrafficSnapshot) { + if n == nil || snap == nil || n.InboundSyncMode != "selected" { + return + } + allowed := nodeSelectedTagSet(n) filtered := make([]*model.Inbound, 0, len(snap.Inbounds)) for _, inbound := range snap.Inbounds { if inbound == nil { diff --git a/internal/web/service/node_push_payload_test.go b/internal/web/service/node_push_payload_test.go new file mode 100644 index 000000000..b7d4b6e2a --- /dev/null +++ b/internal/web/service/node_push_payload_test.go @@ -0,0 +1,78 @@ +package service + +import ( + "encoding/json" + "testing" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" + "github.com/mhsanaei/3x-ui/v3/internal/xray" +) + +func payloadClientEmails(t *testing.T, settings string) []string { + t.Helper() + var parsed map[string]any + if err := json.Unmarshal([]byte(settings), &parsed); err != nil { + t.Fatalf("settings not valid json: %v", err) + } + raw, _ := parsed["clients"].([]any) + out := make([]string, 0, len(raw)) + for _, c := range raw { + cm, ok := c.(map[string]any) + if !ok { + continue + } + if email, _ := cm["email"].(string); email != "" { + out = append(out, email) + } + } + return out +} + +// Stripping disabled clients from a node push made the node delete the row and +// the master mirror that back: every depleted client destroyed, not disabled. +func TestNodePushKeepsDisabledClientsLocalPushStripsThem(t *testing.T) { + setupBulkDB(t) + svc := &InboundService{} + db := database.GetDB() + + clients := []model.Client{ + {Email: "alive@x", ID: "11111111-1111-1111-1111-111111111111", Enable: true}, + {Email: "quota@x", ID: "22222222-2222-2222-2222-222222222222", Enable: true}, + } + ib := mkInbound(t, 30301, model.VLESS, clientsSettings(t, clients)) + + for _, c := range clients { + row := &xray.ClientTraffic{InboundId: ib.Id, Email: c.Email, Enable: true} + if err := db.Create(row).Error; err != nil { + t.Fatalf("seed traffic %s: %v", c.Email, err) + } + } + if err := db.Model(xray.ClientTraffic{}). + Where("email = ?", "quota@x"). + Update("enable", false).Error; err != nil { + t.Fatalf("deplete quota@x: %v", err) + } + + t.Run("node payload keeps the depletion-disabled client", func(t *testing.T) { + built, err := svc.buildInboundForNodePush(db, ib) + if err != nil { + t.Fatalf("buildInboundForNodePush: %v", err) + } + got := payloadClientEmails(t, built.Settings) + if len(got) != 2 { + t.Fatalf("node push dropped a client: got %v, want both alive@x and quota@x", got) + } + }) + + t.Run("local xray payload still strips it", func(t *testing.T) { + built, err := svc.buildInboundForLocalRuntime(db, ib) + if err != nil { + t.Fatalf("buildInboundForLocalRuntime: %v", err) + } + got := payloadClientEmails(t, built.Settings) + if len(got) != 1 || got[0] != "alive@x" { + t.Fatalf("local runtime payload = %v, want only alive@x", got) + } + }) +} diff --git a/internal/web/service/node_sweep_guard_test.go b/internal/web/service/node_sweep_guard_test.go new file mode 100644 index 000000000..15c66fac1 --- /dev/null +++ b/internal/web/service/node_sweep_guard_test.go @@ -0,0 +1,162 @@ +package service + +import ( + "fmt" + "testing" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" + "github.com/mhsanaei/3x-ui/v3/internal/web/runtime" + "github.com/mhsanaei/3x-ui/v3/internal/xray" + + "gorm.io/gorm" +) + +func countClientRows(t *testing.T, db *gorm.DB, email string) (records, traffics int64) { + t.Helper() + if err := db.Model(&model.ClientRecord{}).Where("email = ?", email).Count(&records).Error; err != nil { + t.Fatalf("count clients %q: %v", email, err) + } + if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email).Count(&traffics).Error; err != nil { + t.Fatalf("count client_traffics %q: %v", email, err) + } + return records, traffics +} + +func seedNodeRow(t *testing.T, db *gorm.DB, n *model.Node) { + t.Helper() + if err := db.Create(n).Error; err != nil { + t.Fatalf("create node: %v", err) + } +} + +func snapshotWithClients(t *testing.T, tag, settings string, stats ...xray.ClientTraffic) *runtime.TrafficSnapshot { + t.Helper() + return &runtime.TrafficSnapshot{ + Inbounds: []*model.Inbound{{Tag: tag, Settings: settings, ClientStats: stats}}, + } +} + +func snapshotWithoutClients(t *testing.T, tag string) *runtime.TrafficSnapshot { + t.Helper() + return snapshotWithClients(t, tag, `{"clients":[]}`) +} + +func snapshotWithTwoInbounds(t *testing.T, tagA, settingsA, emailA, tagB, settingsB, emailB string) *runtime.TrafficSnapshot { + t.Helper() + return &runtime.TrafficSnapshot{ + Inbounds: []*model.Inbound{ + {Tag: tagA, Settings: settingsA, ClientStats: []xray.ClientTraffic{{Email: emailA, Enable: true}}}, + {Tag: tagB, Settings: settingsB, ClientStats: []xray.ClientTraffic{{Email: emailB, Enable: true}}}, + }, + } +} + +// The job samples config_dirty before the snapshot round-trip; a client added +// in that window is deleted again unless the merge re-reads the flag itself. +func TestSetRemoteTrafficRereadsConfigDirty(t *testing.T) { + db := initTrafficTestDB(t) + svc := &InboundService{} + + seedNodeRow(t, db, &model.Node{Id: 1, Name: "n1", Address: "127.0.0.1", Port: 2096, ApiToken: "tok", Enable: true}) + + const email = "carol" + createNodeInboundWithClient(t, db, 1, "n1-in", 41001, email) + settings := fmt.Sprintf(`{"clients":[{"email":%q,"enable":true}]}`, email) + syncNodeWithSettings(t, svc, 1, "n1-in", settings, + xray.ClientTraffic{Email: email, Up: 1, Down: 1, Enable: true}) + + if rec, _ := countClientRows(t, db, email); rec != 1 { + t.Fatalf("setup: client not attached, got %d rows", rec) + } + + if err := db.Model(model.Node{}).Where("id = ?", 1).Update("config_dirty", true).Error; err != nil { + t.Fatalf("mark node dirty: %v", err) + } + + if _, err := svc.setRemoteTrafficLocked(1, snapshotWithoutClients(t, "n1-in"), false); err != nil { + t.Fatalf("setRemoteTrafficLocked: %v", err) + } + + rec, traf := countClientRows(t, db, email) + if rec != 1 || traf != 1 { + t.Fatalf("stale dirty=false wiped a client added mid-flight: clients=%d client_traffics=%d, want 1/1", rec, traf) + } +} + +// FilterNodeSnapshot stops reporting a deselected tag; reading that absence as +// "the node deleted it" wiped the master's copy of an inbound still running. +func TestDeselectedTagIsNotSwept(t *testing.T) { + db := initTrafficTestDB(t) + svc := &InboundService{} + + seedNodeRow(t, db, &model.Node{ + Id: 1, Name: "n1", Address: "127.0.0.1", Port: 2096, ApiToken: "tok", Enable: true, + InboundSyncMode: "selected", InboundTags: []string{"keep"}, + }) + + createNodeInboundWithClient(t, db, 1, "keep", 41001, "kept@x") + createNodeInboundWithClient(t, db, 1, "drop", 41002, "dropped@x") + + keepSettings := `{"clients":[{"email":"kept@x","enable":true}]}` + dropSettings := `{"clients":[{"email":"dropped@x","enable":true}]}` + snap := snapshotWithTwoInbounds(t, "keep", keepSettings, "kept@x", "drop", dropSettings, "dropped@x") + if _, err := svc.setRemoteTrafficLocked(1, snap, false); err != nil { + t.Fatalf("seed sync: %v", err) + } + if rec, traf := countClientRows(t, db, "dropped@x"); rec != 1 || traf != 1 { + t.Fatalf("setup: dropped@x not seeded, clients=%d client_traffics=%d", rec, traf) + } + + keepOnly := snapshotWithClients(t, "keep", keepSettings, xray.ClientTraffic{Email: "kept@x", Enable: true}) + if _, err := svc.setRemoteTrafficLocked(1, keepOnly, false); err != nil { + t.Fatalf("post-deselect sync: %v", err) + } + + rec, traf := countClientRows(t, db, "dropped@x") + if rec != 1 || traf != 1 { + t.Fatalf("deselecting a tag deleted its clients: clients=%d client_traffics=%d, want 1/1", rec, traf) + } + var inbounds int64 + if err := db.Model(model.Inbound{}).Where("tag = ?", "drop").Count(&inbounds).Error; err != nil { + t.Fatalf("count inbounds: %v", err) + } + if inbounds != 1 { + t.Fatalf("deselecting a tag deleted the inbound the node still serves: got %d rows, want 1", inbounds) + } +} + +func TestSyncInboundStoresTrimmedEmail(t *testing.T) { + db := initTrafficTestDB(t) + svc := &ClientService{} + + ib := &model.Inbound{UserId: 1, Tag: "trim-in", Enable: true, Port: 41501, Protocol: model.VLESS} + if err := database.GetDB().Create(ib).Error; err != nil { + t.Fatalf("create inbound: %v", err) + } + + padded := "bob " + clients := []model.Client{{Email: padded, Enable: true}} + if err := svc.SyncInbound(nil, ib.Id, clients); err != nil { + t.Fatalf("first SyncInbound: %v", err) + } + + var stored []string + if err := db.Model(&model.ClientRecord{}).Pluck("email", &stored).Error; err != nil { + t.Fatalf("read clients: %v", err) + } + if len(stored) != 1 || stored[0] != "bob" { + t.Fatalf("stored email = %q, want the trimmed %q — the lookup key must match what is written", stored, "bob") + } + + if err := svc.SyncInbound(nil, ib.Id, clients); err != nil { + t.Fatalf("second SyncInbound must not hit a unique-constraint on the untrimmed row: %v", err) + } + var links int64 + if err := db.Model(&model.ClientInbound{}).Where("inbound_id = ?", ib.Id).Count(&links).Error; err != nil { + t.Fatalf("count links: %v", err) + } + if links != 1 { + t.Fatalf("links after re-sync = %d, want 1", links) + } +} diff --git a/internal/web/web.go b/internal/web/web.go index 51a638b56..dc55ad198 100644 --- a/internal/web/web.go +++ b/internal/web/web.go @@ -291,6 +291,7 @@ const ( cadenceNodeHeartbeat = "@every 5s" cadenceNodeTraffic = "@every 5s" cadenceOutboundSub = "@every 5m" + cadenceReapOrphans = "@every 5m" cadenceXrayLogPrune = "@every 10m" cadenceCheckHash = "@every 2m" // cpu.Percent samples over a full minute (blocking), so a finer cadence just @@ -336,6 +337,8 @@ func (s *Server) startTask(restartXray bool, loc *time.Location) { // Outbound subscription auto-refresh (respects per-sub updateInterval) _, _ = s.cron.AddJob(cadenceOutboundSub, job.NewOutboundSubscriptionJob()) + _, _ = s.cron.AddJob(cadenceReapOrphans, job.NewReapSyncOrphansJob()) + // check client ips from log file every day _, _ = s.cron.AddJob("@daily", job.NewClearLogsJob()) _, _ = s.cron.AddJob(cadenceXrayLogPrune, job.NewPruneXrayLogsJob())