diff --git a/go.mod b/go.mod index 99f0f25e0..77e23ff91 100644 --- a/go.mod +++ b/go.mod @@ -32,6 +32,7 @@ require ( go.uber.org/atomic v1.12.0 golang.org/x/crypto v0.58.0 golang.org/x/net v0.61.0 + golang.org/x/sync v0.24.0 golang.org/x/sys v0.49.0 golang.org/x/text v0.43.0 google.golang.org/grpc v1.86.0-dev @@ -110,7 +111,6 @@ require ( golang.org/x/arch v0.32.0 // indirect golang.org/x/crypto/x509roots/fallback v0.0.0-20261020153306-9f2b28d4331a // indirect golang.org/x/exp v0.0.0-20261014143641-f0251eb4aa3e // indirect - golang.org/x/sync v0.24.0 // indirect golang.org/x/time v0.16.0 // indirect golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2 // indirect golang.zx2c4.com/wireguard v0.0.0-20261006164505-2631ce99a06f // indirect diff --git a/internal/sub/controller.go b/internal/sub/controller.go index e495bcb19..e2de1f750 100644 --- a/internal/sub/controller.go +++ b/internal/sub/controller.go @@ -88,6 +88,8 @@ type SUBController struct { subTemplateMu sync.RWMutex subTemplateCache map[string]*cachedSubTemplate + + releaseSettingsCache func() } type subControllerConfig struct { @@ -282,6 +284,11 @@ func defaultSUBControllerConfig() subControllerConfig { } } +// Close unhooks the controller's settings cache from panel mutations. +func (a *SUBController) Close() { + a.releaseSettingsCache() +} + // NewSUBController creates a new subscription controller with the given configuration. func NewSUBController(g *gin.RouterGroup, options ...SUBControllerOption) *SUBController { config := defaultSUBControllerConfig() @@ -290,6 +297,7 @@ func NewSUBController(g *gin.RouterGroup, options ...SUBControllerOption) *SUBCo } sub := NewSubService(config.remarkTemplate) + releaseSettingsCache := sub.enableSettingsCache() subJsonSvc := NewSubJsonService(config.subJsonMux, config.subJsonRules, config.subJsonFinalMask, config.subJsonRoutingRules, sub) subJsonSvc.SetObservatoryConfig(config.subJsonObservatory) subJsonSvc.SetDnsConfig(config.subJsonDns) @@ -326,7 +334,8 @@ func NewSUBController(g *gin.RouterGroup, options ...SUBControllerOption) *SUBCo subJsonService: subJsonSvc, subClashService: NewSubClashService(config.subClashEnableRouting, config.subClashRules, config.subClashRemoteDNS, sub), - subTemplateCache: map[string]*cachedSubTemplate{}, + subTemplateCache: map[string]*cachedSubTemplate{}, + releaseSettingsCache: releaseSettingsCache, } a.initRouter(g) return a diff --git a/internal/sub/service.go b/internal/sub/service.go index 893356ab7..d507fac63 100644 --- a/internal/sub/service.go +++ b/internal/sub/service.go @@ -77,10 +77,13 @@ type SubService struct { // with the clients array left out; generators read only inbound-level // fields (encryption, method, version, …) from it. settingsByInbound map[int]map[string]any - hostsByInbound map[int][]*model.Host - hostGroupOrders map[string]hostEntryOrder - hostsFirst bool - sharedGroupSni map[string]string + // settingsCache is shared by this service's requests (ForRequest copies the pointer); only + // subscription bodies read through it, never panel link rendering. + settingsCache *inboundSettingsCache + hostsByInbound map[int][]*model.Host + hostGroupOrders map[string]hostEntryOrder + hostsFirst bool + sharedGroupSni map[string]string } // NewSubService creates a new subscription service with the given configuration. @@ -90,6 +93,13 @@ func NewSubService(remarkTemplate string) *SubService { } } +// enableSettingsCache gives the subscription server's service its inbound settings cache, purged +// on every panel mutation; release unhooks it when the server stops. +func (s *SubService) enableSettingsCache() (release func()) { + s.settingsCache = newInboundSettingsCache(settingsCacheTTL) + return service.RegisterSubDataInvalidator(s.settingsCache.purgeAll) +} + // ForRequest returns a shallow copy with request-scoped state populated. // Subscription controllers share one base SubService, so request-specific // fields such as address and nodesByID must live on a per-request copy. @@ -725,10 +735,22 @@ func subscriptionExpiryFromClient(nowMs, expiryTime int64) int64 { return 0 } +// getInboundsBySubId loads the subscriber's inbounds; hosts are read per request, never cached. func (s *SubService) getInboundsBySubId(subId string) ([]*model.Inbound, error) { + inbounds, err := s.loadInboundsBySubId(subId) + if err != nil { + return nil, err + } + if err := s.loadSubscriptionHosts(inbounds); err != nil { + return nil, err + } + return inbounds, nil +} + +func (s *SubService) loadInboundsBySubId(subId string) ([]*model.Inbound, error) { db := database.GetDB() - var inbounds []*model.Inbound - err := db.Model(model.Inbound{}).Where(`id in ( + cached := s.settingsCache != nil && s.subscriptionBody + query := db.Model(model.Inbound{}).Where(`id in ( SELECT DISTINCT inbounds.id FROM inbounds JOIN client_inbounds ON client_inbounds.inbound_id = inbounds.id @@ -736,17 +758,41 @@ func (s *SubService) getInboundsBySubId(subId string) ([]*model.Inbound, error) WHERE inbounds.protocol in ('vmess','vless','trojan','shadowsocks','hysteria','wireguard','amneziawg','mtproto','tuic','masque') AND clients.sub_id = ? AND inbounds.enable = ? - )`, subId, true).Order("sub_sort_index ASC").Order("id ASC").Find(&inbounds).Error - if err != nil { + )`, subId, true).Order("sub_sort_index ASC").Order("id ASC") + if cached { + query = query.Omit("settings") + } + var inbounds []*model.Inbound + if err := query.Find(&inbounds).Error; err != nil { return nil, err } + if cached { + for _, inbound := range inbounds { + settings, err := s.subscriptionSettings(inbound) + if err != nil { + return nil, err + } + inbound.Settings = settings + } + } s.indexStatsBySubId(subId) - if err := s.loadSubscriptionHosts(inbounds); err != nil { - return nil, err - } return inbounds, nil } +// subscriptionSettings is an inbound's settings for a subscription body: WireGuard and AmneziaWG +// keep their peers, every other protocol gets the shared copy without the clients array. +func (s *SubService) subscriptionSettings(inbound *model.Inbound) (string, error) { + load := func() (string, error) { + var settings string + err := database.GetDB().Model(&model.Inbound{}).Select("settings").Where("id = ?", inbound.Id).Row().Scan(&settings) + return settings, err + } + if inbound.Protocol == model.WireGuard || inbound.Protocol == model.AmneziaWG { + return load() + } + return s.settingsCache.settingsFor(inbound.Id, load) +} + // indexStatsBySubId loads the traffic rows for just this subscriber's clients // into statsByEmail so statsForClient can resolve a client's usage on any of // its inbounds. It replaces preloading every matched inbound's ClientStats, diff --git a/internal/sub/settings_cache.go b/internal/sub/settings_cache.go new file mode 100644 index 000000000..b79dab557 --- /dev/null +++ b/internal/sub/settings_cache.go @@ -0,0 +1,103 @@ +package sub + +import ( + "strconv" + "sync" + "time" + + "github.com/goccy/go-json" + "golang.org/x/sync/singleflight" +) + +// Subscription apps poll on multi-minute intervals, so inbound settings at most this old read the +// same as fresh ones; panel edits purge them at once. +const settingsCacheTTL = 10 * time.Second + +// inboundSettingsCache keeps each inbound's settings without its clients array for subscription +// bodies, so every subscriber of a large inbound shares one small copy instead of the whole blob. +type inboundSettingsCache struct { + ttl time.Duration + now func() time.Time + group singleflight.Group + + mu sync.Mutex + generation uint64 + entries map[int]settingsEntry +} + +type settingsEntry struct { + settings string + expiresAt time.Time +} + +func newInboundSettingsCache(ttl time.Duration) *inboundSettingsCache { + return &inboundSettingsCache{ttl: ttl, now: time.Now, entries: map[int]settingsEntry{}} +} + +// settingsFor returns the inbound's client-less settings, loading them at most once per TTL window +// even when concurrent requests miss together. +func (c *inboundSettingsCache) settingsFor(inboundId int, load func() (string, error)) (string, error) { + if settings, ok := c.lookup(inboundId); ok { + return settings, nil + } + v, err, _ := c.group.Do(strconv.Itoa(inboundId), func() (any, error) { + if settings, ok := c.lookup(inboundId); ok { + return settings, nil + } + c.mu.Lock() + generation := c.generation + c.mu.Unlock() + raw, err := load() + if err != nil { + return "", err + } + settings := withoutClients(raw) + c.mu.Lock() + defer c.mu.Unlock() + // A purge during the load means the row may have changed under it; serve it, never keep it. + if generation == c.generation { + c.entries[inboundId] = settingsEntry{settings: settings, expiresAt: c.now().Add(c.ttl)} + } + return settings, nil + }) + if err != nil { + return "", err + } + return v.(string), nil +} + +func (c *inboundSettingsCache) lookup(inboundId int) (string, bool) { + c.mu.Lock() + defer c.mu.Unlock() + entry, ok := c.entries[inboundId] + if !ok || !c.now().Before(entry.expiresAt) { + return "", false + } + return entry.settings, true +} + +// purgeAll drops every entry after a panel mutation, so an edit reaches the next fetch. +func (c *inboundSettingsCache) purgeAll() { + c.mu.Lock() + defer c.mu.Unlock() + c.generation++ + clear(c.entries) +} + +// withoutClients drops the clients array, nearly all of a large inbound's settings, which +// subscription bodies resolve from the clients table instead; malformed settings stay as they are. +func withoutClients(settings string) string { + var shallow map[string]json.RawMessage + if err := json.Unmarshal([]byte(settings), &shallow); err != nil { + return settings + } + if _, ok := shallow["clients"]; !ok { + return settings + } + delete(shallow, "clients") + out, err := json.Marshal(shallow) + if err != nil { + return settings + } + return string(out) +} diff --git a/internal/sub/settings_cache_test.go b/internal/sub/settings_cache_test.go new file mode 100644 index 000000000..57b8ee98f --- /dev/null +++ b/internal/sub/settings_cache_test.go @@ -0,0 +1,244 @@ +package sub + +import ( + "fmt" + "strings" + "sync" + "sync/atomic" + "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/web/service" +) + +func TestInboundSettingsCacheHitExpiryAndPurge(t *testing.T) { + now := time.Unix(1_000_000, 0) + cache := newInboundSettingsCache(10 * time.Second) + cache.now = func() time.Time { return now } + var loads atomic.Int32 + load := func() (string, error) { + loads.Add(1) + return `{"clients":[{"email":"a@e"}],"decryption":"none"}`, nil + } + + got, err := cache.settingsFor(1, load) + if err != nil { + t.Fatalf("miss: %v", err) + } + if got != `{"decryption":"none"}` { + t.Fatalf("settings = %s, want the clients array dropped", got) + } + if _, err := cache.settingsFor(1, load); err != nil || loads.Load() != 1 { + t.Fatalf("hit reloaded: loads = %d, err = %v", loads.Load(), err) + } + now = now.Add(11 * time.Second) + if _, err := cache.settingsFor(1, load); err != nil || loads.Load() != 2 { + t.Fatalf("expired entry must reload: loads = %d, err = %v", loads.Load(), err) + } + cache.purgeAll() + if _, err := cache.settingsFor(1, load); err != nil || loads.Load() != 3 { + t.Fatalf("purged entry must reload: loads = %d, err = %v", loads.Load(), err) + } +} + +func TestInboundSettingsCacheSingleFlight(t *testing.T) { + cache := newInboundSettingsCache(time.Minute) + release := make(chan struct{}) + var loads atomic.Int32 + load := func() (string, error) { + loads.Add(1) + <-release + return `{"method":"aes-128-gcm"}`, nil + } + + const callers = 8 + var wg sync.WaitGroup + for range callers { + wg.Go(func() { + if got, err := cache.settingsFor(7, load); err != nil || got != `{"method":"aes-128-gcm"}` { + t.Errorf("settingsFor = %q, %v", got, err) + } + }) + } + // Let every caller reach the shared load before releasing it. + time.Sleep(100 * time.Millisecond) + close(release) + wg.Wait() + if loads.Load() != 1 { + t.Fatalf("%d concurrent misses ran %d loads, want 1", callers, loads.Load()) + } +} + +// A load that a panel edit overlaps may have read the row before the edit; it is served once but +// must not be kept, or the edit would stay invisible for a whole TTL. +func TestInboundSettingsCacheDropsALoadOverlappingAPurge(t *testing.T) { + cache := newInboundSettingsCache(time.Minute) + var loads atomic.Int32 + load := func() (string, error) { + if loads.Add(1) == 1 { + cache.purgeAll() + } + return `{"decryption":"none"}`, nil + } + if _, err := cache.settingsFor(3, load); err != nil { + t.Fatalf("first read: %v", err) + } + if _, err := cache.settingsFor(3, load); err != nil { + t.Fatalf("second read: %v", err) + } + if loads.Load() != 2 { + t.Fatalf("loads = %d, want 2: a load that overlapped a purge was kept", loads.Load()) + } +} + +const ( + parityServerKey = "c2VydmVya2V5MTIzNDU2Nw==" + parityClientKey = "Y2xpZW50a2V5MTIzNDU2Nw==" +) + +// seedParitySubscription gives subscriber "par" a VLESS, a Shadowsocks 2022 and a WireGuard +// inbound, whose links read the top-level settings, the server key and the peers respectively. +func seedParitySubscription(t *testing.T) { + t.Helper() + db := database.GetDB() + seedSubProtocolInbound(t, "par", "pv", 4701, 1, `{"network":"tcp","security":"tls","tlsSettings":{"serverName":"base.sni"}}`, model.VLESS) + ss := seedSubProtocolInbound(t, "par", "ps", 4702, 2, `{"network":"tcp","security":"none"}`, model.Shadowsocks) + ssSettings := fmt.Sprintf(`{"method":"2022-blake3-aes-128-gcm","password":%q,"network":"tcp,udp","clients":[{"email":"ps@e","password":%q,"subId":"par","enable":true}]}`, parityServerKey, parityClientKey) + if err := db.Model(ss).Update("settings", ssSettings).Error; err != nil { + t.Fatalf("shadowsocks settings: %v", err) + } + if err := db.Model(&model.ClientRecord{}).Where("email = ?", "ps@e").Update("password", parityClientKey).Error; err != nil { + t.Fatalf("shadowsocks client password: %v", err) + } + serverPriv, _ := mustWireguardKeypair(t) + clientPriv, _ := mustWireguardKeypair(t) + wgSettings := fmt.Sprintf(`{"secretKey":%q,"clients":[{"email":"pw@e","privateKey":%q,"allowedIPs":["10.0.0.2/32"],"enable":true}]}`, serverPriv, clientPriv) + seedTunnelSubInbound(t, model.WireGuard, "pw", "par", "pw@e", wgSettings, 4703) +} + +// renderSubscription is everything a subscriber can fetch: the raw links, the JSON and the Clash body. +func renderSubscription(t *testing.T, svc *SubService) []string { + t.Helper() + req := svc.ForRequest("req.example.com") + req.subscriptionBody = true + links, _, _, _, err := req.getSubs("par") + if err != nil { + t.Fatalf("getSubs: %v", err) + } + if len(links) != 3 { + t.Fatalf("links = %d, want one per inbound: %v", len(links), links) + } + jsonBody, _, err := NewSubJsonService("", "", "", "", svc).GetJson("par", "req.example.com", false) + if err != nil { + t.Fatalf("GetJson: %v", err) + } + clashBody, _, err := NewSubClashService(false, "", false, svc).GetClash("par", "req.example.com") + if err != nil { + t.Fatalf("GetClash: %v", err) + } + return []string{strings.Join(links, "\n"), jsonBody, clashBody} +} + +// The cache hands generators settings without the clients array; every format must still render +// exactly what the full settings render, from the miss and from the hit alike. +func TestSettingsCacheRendersTheSameSubscription(t *testing.T) { + seedSubDB(t) + seedParitySubscription(t) + + want := renderSubscription(t, NewSubService("")) + cached := NewSubService("") + t.Cleanup(cached.enableSettingsCache()) + for round := range 2 { + got := renderSubscription(t, cached) + for i, format := range []string{"raw", "json", "clash"} { + if got[i] != want[i] { + t.Fatalf("round %d %s differs with the settings cache:\n got: %s\nwant: %s", round, format, got[i], want[i]) + } + } + } + if !strings.Contains(want[0], "ss://") { + t.Fatalf("raw subscription lost its shadowsocks link: %s", want[0]) + } + if n := len(cached.settingsCache.entries); n != 2 { + t.Fatalf("cached %d inbounds, want 2: VLESS and Shadowsocks, never the WireGuard peers", n) + } +} + +// Only the settings blob is cached: the inbound row is read on every fetch, so a port change or a +// disabled inbound reaches the next fetch, while settings wait for a panel edit's purge. +func TestSettingsCacheReadsTheInboundRowFresh(t *testing.T) { + seedSubDB(t) + ib := seedSubInbound(t, "s1", "fresh", 4611, 1, `{"network":"tcp","security":"none"}`) + cached := NewSubService("") + t.Cleanup(cached.enableSettingsCache()) + cached.subscriptionBody = true + + if _, err := cached.getInboundsBySubId("s1"); err != nil { + t.Fatalf("priming read: %v", err) + } + db := database.GetDB() + if err := db.Model(ib).Updates(map[string]any{"port": 4612, "settings": `{"clients":[],"decryption":"changed"}`}).Error; err != nil { + t.Fatalf("edit inbound: %v", err) + } + inbounds, err := cached.getInboundsBySubId("s1") + if err != nil { + t.Fatalf("hit read: %v", err) + } + if len(inbounds) != 1 || inbounds[0].Port != 4612 { + t.Fatalf("inbounds = %+v, want the new port on the next fetch", inbounds) + } + if strings.Contains(inbounds[0].Settings, "changed") { + t.Fatal("settings changed without a purge must still come from the cache") + } + service.InvalidateSubData() + inbounds, err = cached.getInboundsBySubId("s1") + if err != nil { + t.Fatalf("post-purge read: %v", err) + } + if len(inbounds) != 1 || !strings.Contains(inbounds[0].Settings, "changed") { + t.Fatalf("settings = %+v, want the edit after a purge", inbounds) + } + + if err := db.Model(ib).Update("enable", false).Error; err != nil { + t.Fatalf("disable inbound: %v", err) + } + if inbounds, err = cached.getInboundsBySubId("s1"); err != nil || len(inbounds) != 0 { + t.Fatalf("disabled inbound still served: %d inbounds, err %v", len(inbounds), err) + } +} + +// Host rows have no invalidation hooks, so a cached fetch must still read them fresh: a host added +// between two reads has to show up on the second. +func TestGetInboundsBySubIdLoadsHostsOnCacheHit(t *testing.T) { + seedSubDB(t) + ib := seedSubInbound(t, "s1", "hh", 4603, 1, `{"network":"tcp","security":"none"}`) + cached := NewSubService("") + t.Cleanup(cached.enableSettingsCache()) + cached.subscriptionBody = true + + if _, err := cached.getInboundsBySubId("s1"); err != nil { + t.Fatalf("priming read: %v", err) + } + seedHost(t, &model.Host{InboundId: ib.Id, Remark: "edge", Address: "edge.example.com", SortOrder: 1}) + if _, err := cached.getInboundsBySubId("s1"); err != nil { + t.Fatalf("hit read: %v", err) + } + if len(cached.hostsByInbound[ib.Id]) != 1 { + t.Fatalf("cache hit did not load fresh host rows: %+v", cached.hostsByInbound) + } +} + +// Panel link rendering registered an invalidator per render and nothing ever unregistered one; a +// released invalidator must stop running. +func TestSubDataInvalidatorRelease(t *testing.T) { + var calls atomic.Int32 + release := service.RegisterSubDataInvalidator(func() { calls.Add(1) }) + service.InvalidateSubData() + release() + service.InvalidateSubData() + if calls.Load() != 1 { + t.Fatalf("invalidator ran %d times, want 1: a released invalidator kept running", calls.Load()) + } +} diff --git a/internal/sub/sub.go b/internal/sub/sub.go index d6ff2edf4..be8f793c1 100644 --- a/internal/sub/sub.go +++ b/internal/sub/sub.go @@ -466,6 +466,9 @@ func (s *Server) Start() (err error) { // Stop gracefully shuts down the subscription server and closes the listener. func (s *Server) Stop() error { s.cancel() + if s.sub != nil { + s.sub.Close() + } var err1 error var err2 error diff --git a/internal/web/service/client_bulk.go b/internal/web/service/client_bulk.go index 3d0a6bbfd..a7d5733fe 100644 --- a/internal/web/service/client_bulk.go +++ b/internal/web/service/client_bulk.go @@ -1047,6 +1047,7 @@ func (s *ClientService) bulkDelInboundClients( ) bulkInboundDeleteResult { res := bulkInboundDeleteResult{perEmailSkipped: map[string]string{}} + defer InvalidateSubData() defer lockInbound(inboundId).Unlock() oldInbound, err := inboundSvc.GetInbound(inboundId) diff --git a/internal/web/service/client_inbound_apply.go b/internal/web/service/client_inbound_apply.go index c342d8383..d1204ae2b 100644 --- a/internal/web/service/client_inbound_apply.go +++ b/internal/web/service/client_inbound_apply.go @@ -59,6 +59,7 @@ func advancePushedInbound(rt runtime.Runtime, prevSettings, pushedSettings strin // semantics of DelInboundClientByEmail for each removed client. needRestart is // the OR across all removals. func (s *ClientService) delInboundClients(inboundSvc *InboundService, inboundId int, recs []*model.ClientRecord, keepTraffic bool) (bool, error) { + defer InvalidateSubData() if len(recs) == 0 { return false, nil } @@ -325,6 +326,7 @@ func (s *ClientService) checkEmailsExistForClients(inboundSvc *InboundService, c } func (s *ClientService) AddInboundClient(inboundSvc *InboundService, data *model.Inbound) (bool, error) { + defer InvalidateSubData() defer lockInbound(data.Id).Unlock() clients, err := inboundSvc.GetClients(data) @@ -667,6 +669,7 @@ func (s *ClientService) AddInboundClient(inboundSvc *InboundService, data *model } func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *model.Inbound, oldEmail string) (bool, error) { + defer InvalidateSubData() defer lockInbound(data.Id).Unlock() clients, err := inboundSvc.GetClients(data) @@ -1112,6 +1115,7 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo } func (s *ClientService) DelInboundClientByEmail(inboundSvc *InboundService, inboundId int, email string, keepTraffic bool, fullDelete bool) (bool, error) { + defer InvalidateSubData() defer lockInbound(inboundId).Unlock() oldInbound, err := inboundSvc.GetInbound(inboundId) diff --git a/internal/web/service/inbound.go b/internal/web/service/inbound.go index 31d9dd4d6..85f48322f 100644 --- a/internal/web/service/inbound.go +++ b/internal/web/service/inbound.go @@ -1276,6 +1276,7 @@ func (s *InboundService) normalizeMtprotoXrayPort(inbound *model.Inbound, oldSet // then saves the inbound to the database and optionally adds it to the running Xray instance. // Returns the created inbound, whether Xray needs restart, and any error. func (s *InboundService) AddInbound(inbound *model.Inbound) (*model.Inbound, bool, error) { + defer InvalidateSubData() inbound.Id = 0 if err := normalizeTuicSettings(inbound); err != nil { return inbound, false, err @@ -1602,6 +1603,7 @@ func (s *InboundService) AddInbound(inbound *model.Inbound) (*model.Inbound, boo } func (s *InboundService) DelInbound(id int) (bool, error) { + defer InvalidateSubData() needRestart, nodePush, err := s.delInbound(id) if nodePush != nil { nodePush() @@ -1734,6 +1736,7 @@ type BulkDelInboundReport struct { // the rest; the aggregated needRestart is returned so the caller restarts // xray at most once. func (s *InboundService) DelInbounds(ids []int) (BulkDelInboundResult, bool, error) { + defer InvalidateSubData() result := BulkDelInboundResult{} needRestart := false var pushIDs []int @@ -1786,6 +1789,7 @@ func (s *InboundService) GetInboundDetail(id int) (*model.Inbound, error) { // SetInboundSubSortIndex changes only the subscription sort order, so a // reorder cannot carry a stale settings/client payload over another edit. func (s *InboundService) SetInboundSubSortIndex(id int, index int) error { + defer InvalidateSubData() index = normalizeSubSortIndex(index) inbound, err := s.GetInbound(id) if err != nil { @@ -1832,6 +1836,7 @@ func (s *InboundService) SetInboundSubSortIndex(id int, index int) error { } func (s *InboundService) SetInboundEnable(id int, enable bool) (bool, error) { + defer InvalidateSubData() inbound, err := s.GetInbound(id) if err != nil { return false, err @@ -1963,6 +1968,7 @@ func (s *InboundService) validateUpdatedInboundClients(inbound, existing *model. } func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound, bool, error) { + defer InvalidateSubData() legacyShareAddr := legacyMtprotoShareAddr(inbound) inbound.TrafficResetDay = normalizeTrafficResetDay(inbound.TrafficResetDay) // Normalize streamSettings based on protocol diff --git a/internal/web/service/inbound_disable.go b/internal/web/service/inbound_disable.go index 06f890462..8e1c7fb98 100644 --- a/internal/web/service/inbound_disable.go +++ b/internal/web/service/inbound_disable.go @@ -232,6 +232,8 @@ func (s *InboundService) markClientsDisabledInSettings(tx *gorm.DB, inboundID in Update("settings", ib.Settings).Error; err != nil { return nil, nil, err } + // A just-disabled client must leave subscriptions within this tick. + InvalidateSubData() return &snapshot, &ib, nil } diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index d2ba6bbe8..848a5a011 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -1266,6 +1266,9 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi return false, err } structuralChange = true + // Adopted wire settings feed subscription bodies; drop snapshots + // so the adopted rows are served this tick, not after the TTL. + InvalidateSubData() } // The fingerprint stamps the un-lifted wire blob on purpose: a lift must // leave reconcile a mismatch to re-push against. diff --git a/internal/web/service/inbound_settings_commit.go b/internal/web/service/inbound_settings_commit.go index 5ad8e48dd..2c902f2ec 100644 --- a/internal/web/service/inbound_settings_commit.go +++ b/internal/web/service/inbound_settings_commit.go @@ -12,6 +12,7 @@ import ( // commitInboundClientSettings writes a client op's edit (base → ib.Settings) onto // the settings committed now: a traffic tick after the op's read must survive. func commitInboundClientSettings(tx *gorm.DB, ib *model.Inbound, base string) error { + defer InvalidateSubData() var current []string if err := tx.Model(&model.Inbound{}).Where("id = ?", ib.Id).Pluck("settings", ¤t).Error; err != nil { return err diff --git a/internal/web/service/sub_data_invalidation.go b/internal/web/service/sub_data_invalidation.go new file mode 100644 index 000000000..c55726353 --- /dev/null +++ b/internal/web/service/sub_data_invalidation.go @@ -0,0 +1,34 @@ +package service + +import "sync" + +var ( + subDataInvalidatorsMu sync.Mutex + subDataInvalidators = map[int]func(){} + nextSubDataInvalidator int +) + +// RegisterSubDataInvalidator registers a callback run after any mutation that can change what a +// subscription serves; release unregisters it, so a stopped subscription server's cache goes. +func RegisterSubDataInvalidator(f func()) (release func()) { + subDataInvalidatorsMu.Lock() + defer subDataInvalidatorsMu.Unlock() + id := nextSubDataInvalidator + nextSubDataInvalidator++ + subDataInvalidators[id] = f + return func() { + subDataInvalidatorsMu.Lock() + defer subDataInvalidatorsMu.Unlock() + delete(subDataInvalidators, id) + } +} + +// InvalidateSubData drops every registered subscription snapshot. Purging is +// clearing a small map, so callers never need to compute affected subIds. +func InvalidateSubData() { + subDataInvalidatorsMu.Lock() + defer subDataInvalidatorsMu.Unlock() + for _, invalidate := range subDataInvalidators { + invalidate() + } +}