fix: cache tenant routing on message hot path

docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md

@@ -472,3 +472,13 @@ Run ID：`p2p500-fix-20260927-02`。A/B 各建立 500 条 WebSocket，共 500
 2. 压测程序把实时发压和持久化核验拆成两个阶段：先观察 Kafka 自然排空，再以每机 8–16 个 worker 执行历史核验，避免核验流量反向干扰被测落库吞吐。同时另设“历史读取并发压测”场景验证读写混合容量。
 3. persistence 与在线历史查询至少增加独立并发预算；若继续共用连接池，应为 persistence 保留连接或限制历史查询并发，避免读请求耗尽全部 24 条连接。
 4. 完成上述改造后仍以 500 对复测，门禁保持 ACK 10000/10000、实时错误为 0、Kafka 60 秒内归零、最终数据库 10000/10000，再决定是否升级到 1000 对。
+
+### 16.1 路由缓存和分阶段核验改造
+
+2026-09-27 已完成下一轮复测所需代码调整：
+
+- MySQL `TenantAdapterFor` 增加 1 分钟租户路由有效期；有效期内直接复用 tenant adapter，不再读取 `hotgo.im_tenant`。
+- 同一租户的缓存刷新使用 singleflight 合并，TTL 到期瞬间最多执行一次注册库查询，避免并发击穿。
+- `tenant.updated`、`tenant.status.changed`、`tenant.limits.updated`、`tenant.features.updated` 和 `cache.tenant.invalidate` 继续使用现有 Kafka 广播事件；`InvalidateTenantConfigCache` 现在同时使数据库路由缓存失效。失效与刷新并发时通过 generation 校验阻止旧查询重新写入有效缓存。
+- 压测配置新增 `history_settle`，默认 5 分钟。实时 drain 完成后先等待该窗口，再发起历史查询，从而单独测量 Kafka 自然排空时间。
+- 历史查询默认并发从每机 48 降为 12；历史读取高并发能力后续使用独立读写混合场景验证，不再污染消息落库吞吐结果。

go.mod

@@ -36,6 +36,7 @@ require (
 	go.mongodb.org/mongo-driver v1.17.9
 	golang.org/x/crypto v0.53.0
 	golang.org/x/oauth2 v0.36.0
+	golang.org/x/sync v0.21.0
 	golang.org/x/text v0.38.0
 	google.golang.org/api v0.287.0
 	google.golang.org/grpc v1.82.0
@@ -139,7 +140,6 @@ require (
 	github.com/xdg-go/scram v1.2.0 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect
-	golang.org/x/sync v0.21.0 // indirect
 	golang.org/x/sys v0.46.0 // indirect
 	golang.org/x/time v0.15.0 // indirect
 	google.golang.org/appengine v1.6.8 // indirect

loadtest/im-pressure/MACHINE-B.md

@@ -145,7 +145,7 @@ mkdir -p "loadtest/im-pressure/results/${RUN_ID}"
 2. agent 向机器 A coordinator 报告 READY。
 3. 如果机器 A 尚未 READY，当前命令会等待，这是正常现象。
 4. coordinator 下发 `run_id`、场景和统一 UTC 开始时间。
-5. agent 完成发送、WS 到达统计、drain 和历史持久化回查。
+5. agent 完成发送和 WS 到达统计，等待 `history_settle`（默认 5 分钟）后，再以有界并发执行历史持久化回查；等待期间不要停止进程。
 6. agent 写出 summary、上报 coordinator，然后正常退出。
 
 机器 B 不需要传场景参数。`p2p`、`group` 或 `mixed` 由机器 A coordinator 决定。

loadtest/im-pressure/README.md

@@ -47,9 +47,9 @@
 - 发送成功以 `{ctrl code=2xx, params.seq>0}` 为准，形成两台机器共享的 ACK 台账。
 - 实时到达只统计同时匹配本轮 `run_id` 和 `x-im-client-mid` 的 `{data}`。
 - 群消息设置 `noecho=false`，每条服务端接受的群消息期望 2000 个连接各收到一次。
-- 实时 drain 结束后通过 `{get what=data}` 按 ACK 的 Topic/Seq 回查历史，区分跨 Pod 实时漏投与数据库持久化缺失。
+- 实时 drain 结束后先等待 `history_settle`（默认 5 分钟），让 Kafka 在不受核验查询干扰的情况下自然排空，再通过 `{get what=data}` 按 ACK 的 Topic/Seq 回查历史。
 - 每轮建立 WS 后、正式计时前，以 32 个 worker 对本机 1000 个账号做好友关系、P2P Topic 和发布权限预检；任一异常会终止本轮。
-- 历史回查默认使用 48 个 worker、总超时 3 分钟，并对缺失项再并发重试两轮。
+- 历史回查默认每机使用 12 个 worker、总超时 3 分钟，并在超时内持续重试缺失区间。历史读取并发能力应使用独立场景测试，不与落库吞吐测试混在一起。
 - publish 错误完整分类为 `403`、`502`、`503`、`ctrl_timeout`、`socket_error` 和 `other`，不受文本错误样本上限影响。
 - `manifest.json` 含登录 token，文件权限为 `0600`，不得提交到 Git 或放入结果包。
 

loadtest/im-pressure/agent.go

@@ -183,6 +183,11 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo
 	base.Errors = append(base.Errors, run.errors...)
 	run.mu.Unlock()
 	summary := run.recorder.summarize(base, ledger)
+	historySettle, _ := configDuration(cfg.HistorySettle)
+	log.Printf("agent=%s waiting %s before history verification", machine, historySettle)
+	if err = waitContext(ctx, historySettle); err != nil {
+		return err
+	}
 	checked, found, persistenceMissing := run.reconcileHistory(ctx, ledger)
 	summary.HistoryChecked = int64(checked)
 	summary.HistoryFound = int64(found)

loadtest/im-pressure/config.example.json

@@ -16,7 +16,8 @@
   "command_timeout": "20s",
   "preflight_workers": 32,
   "skip_p2p_preflight": false,
-  "history_workers": 48,
+  "history_workers": 12,
+  "history_settle": "5m",
   "history_timeout": "3m",
   "p2p_messages_per_user": 10,
   "p2p_interval": "1s",

loadtest/im-pressure/model.go

@@ -32,6 +32,7 @@ type Config struct {
 	PhonePrefix      string              `json:"registration_phone_prefix"`
 	VerifyCodeEnv    string              `json:"verification_code_env"`
 	HistoryWorkers   int                 `json:"history_workers"`
+	HistorySettle    string              `json:"history_settle"`
 	HistoryTimeout   string              `json:"history_timeout"`
 	PreflightWorkers int                 `json:"preflight_workers"`
 	SkipP2PPreflight bool                `json:"skip_p2p_preflight"`
@@ -74,7 +75,7 @@ func loadConfig(path string) (*Config, error) {
 		cfg.DrainGroup = "60s"
 	}
 	if cfg.HistoryWorkers <= 0 {
-		cfg.HistoryWorkers = 48
+		cfg.HistoryWorkers = 12
 	}
 	if cfg.HistoryWorkers > 128 {
 		return nil, errors.New("history_workers must not exceed 128")
@@ -82,6 +83,12 @@ func loadConfig(path string) (*Config, error) {
 	if cfg.HistoryTimeout == "" {
 		cfg.HistoryTimeout = "3m"
 	}
+	if cfg.HistorySettle == "" {
+		cfg.HistorySettle = "5m"
+	}
+	if _, err = configDuration(cfg.HistorySettle); err != nil {
+		return nil, fmt.Errorf("history_settle: %w", err)
+	}
 	if _, err = configDuration(cfg.HistoryTimeout); err != nil {
 		return nil, fmt.Errorf("history_timeout: %w", err)
 	}

loadtest/im-pressure/model_test.go

@@ -16,7 +16,7 @@ func TestLoadConfigAppliesBoundedHistoryDefaults(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if cfg.HistoryWorkers != 48 || cfg.HistoryTimeout != "3m" || cfg.PreflightWorkers != 32 {
+	if cfg.HistoryWorkers != 12 || cfg.HistorySettle != "5m" || cfg.HistoryTimeout != "3m" || cfg.PreflightWorkers != 32 {
 		t.Fatalf("unexpected defaults: %+v", cfg)
 	}
 }

server/db/adapter.go

@@ -260,6 +260,12 @@ type TenantRoutedAdapter interface {
 	TenantAdapterFor(t.TenantID) (TenantBusinessAdapter, TenantAuthAdapter, error)
 }
 
+// TenantRouteCacheInvalidator clears cached tenant routing metadata. A zero
+// tenant ID clears every cached route.
+type TenantRouteCacheInvalidator interface {
+	InvalidateTenantRoute(t.TenantID)
+}
+
 type TenantMessageBatchAdapter interface {
 	TenantMessageSaveBatch(t.TenantID, []*t.Message) error
 	TenantGroupMessageSaveBatch(t.TenantID, []*t.Message) error

server/db/mysql/adapter.go

@@ -24,6 +24,7 @@ import (
 	immetrics "github.com/tinode/chat/server/metrics"
 	"github.com/tinode/chat/server/store"
 	t "github.com/tinode/chat/server/store/types"
+	"golang.org/x/sync/singleflight"
 )
 
 // adapter holds MySQL connection data.
@@ -43,14 +44,17 @@ type adapter struct {
 	// DB transaction timeout.
 	txTimeout time.Duration
 
-	baseConfig       *configType
-	tenantDBRouting  bool
-	tenantScopedDB   bool
-	tenantID         t.TenantID
-	tenantRegistryDB string
-	tenantAdapters   map[t.TenantID]*adapter
-	tenantMu         sync.Mutex
-	groupShardWindow string
+	baseConfig         *configType
+	tenantDBRouting    bool
+	tenantScopedDB     bool
+	tenantID           t.TenantID
+	tenantRegistryDB   string
+	tenantAdapters     map[t.TenantID]*adapter
+	tenantRouteExpiry  map[t.TenantID]time.Time
+	tenantRouteGen     uint64
+	tenantRouteRefresh singleflight.Group
+	tenantMu           sync.Mutex
+	groupShardWindow   string
 }
 
 const (
@@ -63,6 +67,7 @@ const (
 	defaultMaxResults = 1024
 	// This is capped by the Session's send queue limit (128).
 	defaultMaxMessageResults = 100
+	tenantRouteCacheTTL      = time.Minute
 
 	// If DB request timeout is specified,
 	// we allocate txTimeoutMultiplier times more time for transactions.
@@ -262,6 +267,7 @@ func (a *adapter) Open(jsonconfig json.RawMessage) error {
 	a.tenantID = defaultTenantID
 	a.tenantRegistryDB = strings.TrimSpace(config.TenantRegistryDB)
 	a.tenantAdapters = make(map[t.TenantID]*adapter)
+	a.tenantRouteExpiry = make(map[t.TenantID]time.Time)
 
 	if a.maxResults <= 0 {
 		a.maxResults = defaultMaxResults
@@ -320,6 +326,8 @@ func (a *adapter) Close() error {
 		}
 		delete(a.tenantAdapters, tenantID)
 	}
+	a.tenantRouteExpiry = make(map[t.TenantID]time.Time)
+	a.tenantRouteGen++
 	a.tenantMu.Unlock()
 	if a.registryDB != nil {
 		if closeErr := a.registryDB.Close(); err == nil {
@@ -475,35 +483,62 @@ func (a *adapter) TenantListActive(limit int) ([]t.Tenant, error) {
 }
 
 func (a *adapter) TenantAdapterFor(tenantID t.TenantID) (dbadapter.TenantBusinessAdapter, dbadapter.TenantAuthAdapter, error) {
-	if !a.tenantDBRouting {
-		if a.tenantID != defaultTenantID && a.tenantID != tenantID {
-			return nil, nil, t.ErrMalformed
-		}
-		if a.registryDB != nil {
-			tenant, err := a.tenantByID(tenantID)
-			if err != nil {
-				return nil, nil, err
-			}
-			if tenant.State != t.TenantStateActive {
-				return nil, nil, t.ErrPermissionDenied
-			}
-		}
-		return a, a, nil
+	if !tenantID.IsValid() || (!a.tenantDBRouting && a.tenantID != defaultTenantID && a.tenantID != tenantID) {
+		return nil, nil, t.ErrMalformed
 	}
-	tenant, err := a.tenantByID(tenantID)
+	if tenantAdp, ok := a.cachedTenantAdapter(tenantID, time.Now()); ok {
+		return tenantAdp, tenantAdp, nil
+	}
+
+	resolved, err, _ := a.tenantRouteRefresh.Do(strconv.FormatInt(int64(tenantID), 10), func() (any, error) {
+		if tenantAdp, ok := a.cachedTenantAdapter(tenantID, time.Now()); ok {
+			return tenantAdp, nil
+		}
+		generation := a.tenantRouteGeneration()
+		if a.registryDB != nil {
+			if _, err := a.tenantByID(tenantID); err != nil {
+				return nil, err
+			}
+		}
+		tenantAdp := a.tenantAdapter(tenantID)
+		a.cacheTenantAdapter(tenantID, tenantAdp, generation, time.Now().Add(tenantRouteCacheTTL))
+		return tenantAdp, nil
+	})
 	if err != nil {
 		return nil, nil, err
 	}
-	if tenant.State != t.TenantStateActive {
-		return nil, nil, t.ErrPermissionDenied
+	tenantAdp := resolved.(*adapter)
+	return tenantAdp, tenantAdp, nil
+}
+
+func (a *adapter) cachedTenantAdapter(tenantID t.TenantID, now time.Time) (*adapter, bool) {
+	a.tenantMu.Lock()
+	defer a.tenantMu.Unlock()
+	if expiresAt, ok := a.tenantRouteExpiry[tenantID]; !ok || !now.Before(expiresAt) {
+		return nil, false
+	}
+	if !a.tenantDBRouting {
+		return a, true
+	}
+	tenantAdp := a.tenantAdapters[tenantID]
+	return tenantAdp, tenantAdp != nil
+}
+
+func (a *adapter) tenantRouteGeneration() uint64 {
+	a.tenantMu.Lock()
+	defer a.tenantMu.Unlock()
+	return a.tenantRouteGen
+}
+
+func (a *adapter) tenantAdapter(tenantID t.TenantID) *adapter {
+	if !a.tenantDBRouting {
+		return a
 	}
 	a.tenantMu.Lock()
-	if tenantAdp := a.tenantAdapters[tenantID]; tenantAdp != nil {
-		a.tenantMu.Unlock()
-		return tenantAdp, tenantAdp, nil
+	defer a.tenantMu.Unlock()
+	if existing := a.tenantAdapters[tenantID]; existing != nil {
+		return existing
 	}
-	a.tenantMu.Unlock()
-
 	registryDB := a.registryDB
 	if registryDB == nil {
 		registryDB = a.db
@@ -525,14 +560,32 @@ func (a *adapter) TenantAdapterFor(tenantID t.TenantID) (dbadapter.TenantBusines
 		// explicit tenant_id writes enabled for logical isolation.
 		tenantScopedDB: false,
 	}
-	a.tenantMu.Lock()
-	if existing := a.tenantAdapters[tenantID]; existing != nil {
-		a.tenantMu.Unlock()
-		return existing, existing, nil
-	}
 	a.tenantAdapters[tenantID] = tenantAdp
-	a.tenantMu.Unlock()
-	return tenantAdp, tenantAdp, nil
+	return tenantAdp
+}
+
+func (a *adapter) cacheTenantAdapter(tenantID t.TenantID, tenantAdp *adapter, generation uint64, expiresAt time.Time) {
+	a.tenantMu.Lock()
+	defer a.tenantMu.Unlock()
+	if generation != a.tenantRouteGen || tenantAdp == nil {
+		return
+	}
+	if a.tenantRouteExpiry == nil {
+		a.tenantRouteExpiry = make(map[t.TenantID]time.Time)
+	}
+	a.tenantRouteExpiry[tenantID] = expiresAt
+}
+
+// InvalidateTenantRoute forces the next operation to revalidate tenant state.
+func (a *adapter) InvalidateTenantRoute(tenantID t.TenantID) {
+	a.tenantMu.Lock()
+	defer a.tenantMu.Unlock()
+	a.tenantRouteGen++
+	if tenantID.IsValid() {
+		delete(a.tenantRouteExpiry, tenantID)
+		return
+	}
+	a.tenantRouteExpiry = make(map[t.TenantID]time.Time)
 }
 
 func (a *adapter) tenantByID(tenantID t.TenantID) (*t.Tenant, error) {

server/db/mysql/adapter_tenant_route_cache_test.go

@@ -0,0 +1,66 @@
+//go:build mysql
+
+package mysql
+
+import (
+	"testing"
+	"time"
+
+	"github.com/tinode/chat/server/store/types"
+)
+
+func TestTenantRouteCacheExpiresAndInvalidates(t *testing.T) {
+	now := time.Now()
+	tenantID := types.TenantID(8)
+	tenantAdp := &adapter{tenantID: tenantID}
+	root := &adapter{
+		tenantDBRouting:   true,
+		tenantAdapters:    map[types.TenantID]*adapter{tenantID: tenantAdp},
+		tenantRouteExpiry: make(map[types.TenantID]time.Time),
+	}
+
+	root.cacheTenantAdapter(tenantID, tenantAdp, root.tenantRouteGeneration(), now.Add(time.Minute))
+	if got, ok := root.cachedTenantAdapter(tenantID, now); !ok || got != tenantAdp {
+		t.Fatalf("cached route = %p, %v; want %p, true", got, ok, tenantAdp)
+	}
+	if got, _, err := root.TenantAdapterFor(tenantID); err != nil || got != tenantAdp {
+		t.Fatalf("TenantAdapterFor cached route = %p, %v; want %p, nil", got, err, tenantAdp)
+	}
+
+	staleGeneration := root.tenantRouteGeneration()
+	root.InvalidateTenantRoute(tenantID)
+	if _, ok := root.cachedTenantAdapter(tenantID, now); ok {
+		t.Fatal("invalidated tenant route remained cached")
+	}
+	root.cacheTenantAdapter(tenantID, tenantAdp, staleGeneration, now.Add(time.Minute))
+	if _, ok := root.cachedTenantAdapter(tenantID, now); ok {
+		t.Fatal("an in-flight stale refresh repopulated the route cache")
+	}
+
+	root.cacheTenantAdapter(tenantID, tenantAdp, root.tenantRouteGeneration(), now.Add(-time.Second))
+	if _, ok := root.cachedTenantAdapter(tenantID, now); ok {
+		t.Fatal("expired tenant route remained cached")
+	}
+}
+
+func TestInvalidateAllTenantRoutes(t *testing.T) {
+	now := time.Now()
+	root := &adapter{
+		tenantDBRouting: true,
+		tenantAdapters: map[types.TenantID]*adapter{
+			8: {tenantID: 8},
+			9: {tenantID: 9},
+		},
+		tenantRouteExpiry: map[types.TenantID]time.Time{
+			8: now.Add(time.Minute),
+			9: now.Add(time.Minute),
+		},
+	}
+
+	root.InvalidateTenantRoute(0)
+	for _, tenantID := range []types.TenantID{8, 9} {
+		if _, ok := root.cachedTenantAdapter(tenantID, now); ok {
+			t.Fatalf("tenant %d route remained cached after global invalidation", tenantID)
+		}
+	}
+}

server/store/config_cache.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 	"time"
 
+	adapter "github.com/tinode/chat/server/db"
 	"github.com/tinode/chat/server/store/types"
 )
 
@@ -152,6 +153,9 @@ func InvalidateTenantConfigCache(code string, id types.TenantID) {
 		tenantByIDCache.clear()
 		realNamePolicyCache.clear()
 	}
+	if invalidator, ok := adp.(adapter.TenantRouteCacheInvalidator); ok {
+		invalidator.InvalidateTenantRoute(id)
+	}
 }
 
 // InvalidateRegistrationConfigCache clears cached global registration config.