diff --git a/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md b/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md index a5f90fb..a624926 100644 --- a/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md +++ b/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;历史读取高并发能力后续使用独立读写混合场景验证,不再污染消息落库吞吐结果。 diff --git a/go.mod b/go.mod index abebad0..62575dd 100644 --- a/go.mod +++ b/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 diff --git a/loadtest/im-pressure/MACHINE-B.md b/loadtest/im-pressure/MACHINE-B.md index 4d64b9c..71bae69 100644 --- a/loadtest/im-pressure/MACHINE-B.md +++ b/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 决定。 diff --git a/loadtest/im-pressure/README.md b/loadtest/im-pressure/README.md index f5c3c9f..724aa47 100644 --- a/loadtest/im-pressure/README.md +++ b/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 或放入结果包。 diff --git a/loadtest/im-pressure/agent.go b/loadtest/im-pressure/agent.go index 3fcc9e7..c187c1d 100644 --- a/loadtest/im-pressure/agent.go +++ b/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) diff --git a/loadtest/im-pressure/config.example.json b/loadtest/im-pressure/config.example.json index 6178d54..da71974 100644 --- a/loadtest/im-pressure/config.example.json +++ b/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", diff --git a/loadtest/im-pressure/model.go b/loadtest/im-pressure/model.go index 5f6578b..862b471 100644 --- a/loadtest/im-pressure/model.go +++ b/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) } diff --git a/loadtest/im-pressure/model_test.go b/loadtest/im-pressure/model_test.go index e032d2f..e15fb30 100644 --- a/loadtest/im-pressure/model_test.go +++ b/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) } } diff --git a/server/db/adapter.go b/server/db/adapter.go index fa3c570..e679f20 100644 --- a/server/db/adapter.go +++ b/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 diff --git a/server/db/mysql/adapter.go b/server/db/mysql/adapter.go index 60be245..9b242b6 100644 --- a/server/db/mysql/adapter.go +++ b/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) { diff --git a/server/db/mysql/adapter_tenant_route_cache_test.go b/server/db/mysql/adapter_tenant_route_cache_test.go new file mode 100644 index 0000000..c0c9221 --- /dev/null +++ b/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) + } + } +} diff --git a/server/store/config_cache.go b/server/store/config_cache.go index 2f5933c..d010696 100644 --- a/server/store/config_cache.go +++ b/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.