perf(sub): cache each inbound's settings without clients for subscription bodies (#6758)

Every subscription fetch read and scanned each inbound's whole settings
blob, about 25 MB at 100k clients, although subscription bodies resolve
their clients from the clients table and never use its clients array.

Each inbound's settings minus the clients array are now cached, keyed by
inbound and shared by all its subscribers; the inbound row, clients,
stats and hosts are still read per fetch, and WireGuard and AmneziaWG keep
full settings for their peers. Panel edits purge the cache through an
invalidator registry the subscription server releases on stop, with a
10 s TTL as the backstop. At 100k clients a hit takes 2.1 ms on SQLite
(72.5 ms before) and 20.7 ms on PostgreSQL (332 ms).

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

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

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
+	// settingsCache is shared by this service's requests (ForRequest copies the pointer); only
-	hostGroupOrders   map[string]hostEntryOrder
+	// subscription bodies read through it, never panel link rendering.
-	hostsFirst        bool
+	settingsCache   *inboundSettingsCache
-	sharedGroupSni    map[string]string
+	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
+	cached := s.settingsCache != nil && s.subscriptionBody
-	err := db.Model(model.Inbound{}).Where(`id in (
+	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
+	)`, subId, true).Order("sub_sort_index ASC").Order("id ASC")
-	if err != nil {
+	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,

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)
+}

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())
+	}
+}

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

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)

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)

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

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
 }
 

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.

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", &current).Error; err != nil {
 		return err

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()
+	}
+}