From f9f7321b116e6f7f013677c645841a47f97cd022 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Sun, 27 Sep 2026 12:53:09 +0800 Subject: [PATCH] fix p2p permission refresh during load tests --- ...压测Pod日志根因分析与改进方案-20260927.md | 38 +++++++++++ loadtest/im-pressure/README.md | 2 +- loadtest/im-pressure/agent.go | 43 ++++++++---- loadtest/im-pressure/agent_history_test.go | 41 +++++++++++ server/im_http.go | 1 + server/im_p2p_events.go | 56 +++++++++++++++ server/im_p2p_events_test.go | 68 +++++++++++++++++++ server/topic.go | 31 ++++++++- server/topic_test.go | 4 ++ 9 files changed, 269 insertions(+), 15 deletions(-) create mode 100644 loadtest/im-pressure/agent_history_test.go create mode 100644 server/im_p2p_events.go create mode 100644 server/im_p2p_events_test.go diff --git a/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md b/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md index 3f1e757..be04d86 100644 --- a/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md +++ b/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md @@ -259,3 +259,41 @@ Snowflake ID 超过有符号 32 位 `INT` 上限，批量插入立即报 1264。 - 压测程序增加全量 P2P 预检、48 worker 历史回查、3 分钟总超时、publish 错误分类、指标快照阶段标记、P95/ACK 成功率门禁及 MySQL 独立精确核验。 当前调试 kubeconfig 对 StatefulSet/Service/Ingress 没有 patch 权限，因此现存 Pod 资源尚未被本地会话直接修改；以上部署项会在新提交触发有权限的 Gitea deploy runner 后生效。生效后必须按第 8 节执行 200/500/1000 对阶梯复测，不能用单元测试替代容量验收。 + +## 11. 200 对阶梯复测结果 + +Run ID：`p2p-200-20260927-01`，A/B 两台压测机各模拟 200 个用户，共 200 对、400 条 WebSocket 连接。 + +| 指标 | 结果 | +|---|---:| +| WS Ready | 400/400 | +| Publish 尝试 | 4000 | +| 服务端 ACK | 3890 | +| 403 拒绝 | 110 | +| 502 / 503 / ctrl timeout / socket error | 0 / 0 / 0 / 0 | +| ACK 后应到达 / 实际到达 | 3890 / 3890 | +| ACK 后到达率 | 100% | +| 实时丢失 / 重复 / 乱序 / 断连 | 0 / 0 / 0 / 0 | +| Kafka P2P `batch_ok` | 3890 | +| Kafka consumer lag / backlog age | 0 / 0 | +| MySQL 精确匹配 | 3890/3890 | +| MySQL 缺失 / 错配 / 重复 / 非 ACK 落库 | 0 / 0 / 0 / 0 | + +本轮没有再出现 `cluster: load exceeded`、`server busy`、502 或 503，说明 Cluster P0 改造在 200 对档位已生效。Kafka 消费指标和 MySQL 对 ACK ledger 的独立精确查询均确认 3890 条消息已全部落库。 + +压测报告当时的 `history_found=2213`、`persistence_lost=1677` 不是真实落库丢失。旧实现首轮查询后仅等待 2 秒，再进行两次 2 秒重试，并未持续到配置的 3 分钟截止时间。已改为在 `history_timeout` 内持续重试，并按 Topic 合并缺失区间查询，避免过早判定和大量单条请求。 + +### 11.1 403 根因与修复 + +110 次 403 稳定集中在用户序号 `0018`、`0043`、`0065`、`0082`、`0175`、`0183` 的 6 个 P2P Topic，与 Cluster 转发无关。`EnsureP2P` 已将 MySQL 中双方订阅修复为 `JRWPA`，但已加载 Topic 的 `perUser` 仍保留旧权限，因此预检通过后 WebSocket publish 仍被内存态拒绝。 + +已增加两层修复： + +1. HTTP `EnsureP2P` 成功后向 Topic master 发布内部权限刷新事件，原生 Cluster 由 `routeSrv` 路由，非原生多 Pod 模式由 Redis realtime fanout 路由。 +2. P2P publish 发现内存态无写权限时，先从 MySQL 重载该用户订阅再判定，作为刷新事件延迟或丢失时的自愈保障。数据库仍无写权限时依然返回 403。 + +### 11.2 当前未达标项 + +- ACK P95 和到达 P95 均落在 2 秒直方图上界，未达 500ms / 1s 门槛。 +- ArgoCD 已将 StatefulSet 资源回滚为 `request 100m/128Mi`、`limit 500m/256Mi`，且 liveness 仍使用 `/healthz`。压测期间存在 CPU throttling 累计，部署配置必须在 GitOps 声明源中固化，否则应用后仍会被回滚。 +- 403 修复发布后应重跑 200 对；只有 ACK 成功率、实时到达率、MySQL 精确落库均达标，才进入 500 对档位。 diff --git a/loadtest/im-pressure/README.md b/loadtest/im-pressure/README.md index d1cd82c..f5c3c9f 100644 --- a/loadtest/im-pressure/README.md +++ b/loadtest/im-pressure/README.md @@ -244,7 +244,7 @@ export MYSQL_DSN='readonly:password@tcp(mysql-host:3306)/meeting_im?parseTime=tr -coordinator "loadtest/im-pressure/results/${RUN_ID}/coordinator.json" \ -tenant allrich \ -dsn-env MYSQL_DSN \ - -registry-db meeting_im \ + -registry-db hotgo \ -output "loadtest/im-pressure/results/${RUN_ID}/db-verification.json" ``` diff --git a/loadtest/im-pressure/agent.go b/loadtest/im-pressure/agent.go index 712f28d..3fcc9e7 100644 --- a/loadtest/im-pressure/agent.go +++ b/loadtest/im-pressure/agent.go @@ -259,24 +259,14 @@ func (r *agentRun) reconcileHistory(ctx context.Context, ledger []AcceptedMessag defer cancel() r.issueHistoryQueries(historyCtx, queries) _ = waitContext(historyCtx, 2*time.Second) - for round := 0; round < 2; round++ { + for { missing := missingHistory(r.recorder, expected) if len(missing) == 0 || historyCtx.Err() != nil { break } - retryQueries := make([]historyQuery, 0, len(missing)) - for _, message := range missing { - client := r.clients[message.PeerIndex] - if message.Kind == "group" { - client = r.clients[1] - } - if client != nil { - retryQueries = append(retryQueries, historyQuery{client: client, topic: message.Topic, - since: message.Seq, before: message.Seq + 1, limit: 1}) - } - } + retryQueries := r.historyQueriesForMessages(missing) r.issueHistoryQueries(historyCtx, retryQueries) - _ = waitContext(historyCtx, 2*time.Second) + _ = waitContext(historyCtx, 5*time.Second) } missing := missingHistory(r.recorder, expected) missingIDs := make([]string, 0, len(missing)) @@ -287,6 +277,33 @@ func (r *agentRun) reconcileHistory(ctx context.Context, ledger []AcceptedMessag return len(expected), len(expected) - len(missing), missingIDs } +func (r *agentRun) historyQueriesForMessages(messages []AcceptedMessage) []historyQuery { + byTopic := make(map[string][]AcceptedMessage) + for _, message := range messages { + byTopic[message.Topic] = append(byTopic[message.Topic], message) + } + queries := make([]historyQuery, 0, len(byTopic)) + for topic, topicMessages := range byTopic { + sort.Slice(topicMessages, func(i, j int) bool { return topicMessages[i].Seq < topicMessages[j].Seq }) + client := r.clients[topicMessages[0].PeerIndex] + if topicMessages[0].Kind == "group" { + client = r.clients[1] + } + if client == nil { + continue + } + for start := 0; start < len(topicMessages); start += 200 { + end := start + 200 + if end > len(topicMessages) { + end = len(topicMessages) + } + queries = append(queries, historyQuery{client: client, topic: topic, + since: topicMessages[start].Seq, before: topicMessages[end-1].Seq + 1, limit: 256}) + } + } + return queries +} + func (r *agentRun) issueHistoryQueries(ctx context.Context, queries []historyQuery) { workers := r.cfg.HistoryWorkers if workers > len(queries) { diff --git a/loadtest/im-pressure/agent_history_test.go b/loadtest/im-pressure/agent_history_test.go new file mode 100644 index 0000000..c4795a7 --- /dev/null +++ b/loadtest/im-pressure/agent_history_test.go @@ -0,0 +1,41 @@ +package main + +import "testing" + +func TestHistoryQueriesForMessagesGroupsRangesByTopic(t *testing.T) { + p2pClient := &WSClient{} + groupClient := &WSClient{} + run := &agentRun{clients: map[int]*WSClient{1: groupClient, 7: p2pClient}} + messages := make([]AcceptedMessage, 0, 207) + for seq := int64(205); seq >= 1; seq-- { + messages = append(messages, AcceptedMessage{Kind: "p2p", Topic: "p2p-a", PeerIndex: 7, Seq: seq}) + } + messages = append(messages, + AcceptedMessage{Kind: "group", Topic: "grp-a", PeerIndex: 99, Seq: 11}, + AcceptedMessage{Kind: "group", Topic: "grp-a", PeerIndex: 99, Seq: 12}, + ) + + queries := run.historyQueriesForMessages(messages) + if len(queries) != 3 { + t.Fatalf("queries=%d want=3", len(queries)) + } + p2pQueries := 0 + groupQueries := 0 + for _, query := range queries { + switch query.topic { + case "p2p-a": + p2pQueries++ + if query.client != p2pClient || query.limit != 256 { + t.Fatalf("unexpected p2p query: %#v", query) + } + case "grp-a": + groupQueries++ + if query.client != groupClient || query.since != 11 || query.before != 13 { + t.Fatalf("unexpected group query: %#v", query) + } + } + } + if p2pQueries != 2 || groupQueries != 1 { + t.Fatalf("p2p queries=%d group queries=%d", p2pQueries, groupQueries) + } +} diff --git a/server/im_http.go b/server/im_http.go index 5034c73..eb93166 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -2367,6 +2367,7 @@ func serveIMP2PConversation(wrt http.ResponseWriter, req *http.Request) { writeIMRelationError(wrt, req, err) return } + publishIMP2PPermissionsRefresh(principal.TenantID, topic, principal.Uid, peer) ctx, err := store.Relations.GetRelationshipContext(principal.TenantID, principal.Uid, peer, topic) if err != nil { writeIMRelationError(wrt, req, err) diff --git a/server/im_p2p_events.go b/server/im_p2p_events.go new file mode 100644 index 0000000..2a5f3b9 --- /dev/null +++ b/server/im_p2p_events.go @@ -0,0 +1,56 @@ +package main + +import ( + "encoding/json" + "strings" + "time" + + "github.com/tinode/chat/server/logs" + "github.com/tinode/chat/server/store/types" +) + +const p2pPermissionsRefreshedEvent = "p2p.permissions.refreshed" + +type p2pPermissionsPayload struct { + UserIDs []string `json:"user_ids"` +} + +// publishIMP2PPermissionsRefresh asks the topic master to reload permissions +// after the HTTP API has repaired or created the two P2P subscriptions. +func publishIMP2PPermissionsRefresh(tenantID types.TenantID, topic string, uids ...types.Uid) { + topic = strings.TrimSpace(topic) + if !tenantID.IsValid() || topic == "" { + return + } + payload := p2pPermissionsPayload{UserIDs: make([]string, 0, len(uids))} + for _, uid := range uids { + if !uid.IsZero() { + payload.UserIDs = append(payload.UserIDs, uid.UserId()) + } + } + if len(payload.UserIDs) == 0 { + return + } + encoded, err := json.Marshal(payload) + if err != nil { + logs.Warn.Printf("p2p permissions refresh: encode topic=%s: %v", topic, err) + return + } + message := &ServerComMessage{ + TenantID: tenantID, + RcptTo: topic, + Timestamp: time.Now().UTC().Round(time.Millisecond), + Info: &MsgServerInfo{ + Topic: topic, What: "p2p", Event: p2pPermissionsRefreshedEvent, + Payload: json.RawMessage(encoded), + }, + } + if globals.hub != nil { + select { + case globals.hub.routeSrv <- message: + default: + logs.Warn.Printf("p2p permissions refresh: hub queue full topic=%s", topic) + } + } + publishRealtimeTopicMessage(topic, message) +} diff --git a/server/im_p2p_events_test.go b/server/im_p2p_events_test.go new file mode 100644 index 0000000..826da66 --- /dev/null +++ b/server/im_p2p_events_test.go @@ -0,0 +1,68 @@ +package main + +import ( + "encoding/json" + "testing" + + "github.com/golang/mock/gomock" + "github.com/tinode/chat/server/store/types" +) + +func TestHandleP2PPermissionsRefreshReloadsLoadedTopic(t *testing.T) { + helper := TopicTestHelper{} + helper.setUp(t, 2, types.TopicCatP2P, "p2p-test", true) + defer helper.tearDown() + + uid := helper.uids[0] + stale := types.ModeCP2P &^ types.ModeWrite + pud := helper.topic.perUser[uid] + pud.modeWant = stale + pud.modeGiven = stale + helper.topic.perUser[uid] = pud + helper.ss.EXPECT().Get(types.TenantID(1), "p2p-test", uid, true).Return(&types.Subscription{ + TenantID: 1, User: uid.String(), Topic: "p2p-test", + ModeWant: types.ModeCP2P, ModeGiven: types.ModeCP2P, + }, nil) + + payload, err := json.Marshal(p2pPermissionsPayload{UserIDs: []string{uid.UserId()}}) + if err != nil { + t.Fatal(err) + } + helper.topic.handleServerMsg(&ServerComMessage{ + TenantID: 1, + Info: &MsgServerInfo{What: "p2p", Event: p2pPermissionsRefreshedEvent, Payload: payload}, + }) + helper.finish() + + got := helper.topic.perUser[uid] + if !got.modeWant.IsWriter() || !got.modeGiven.IsWriter() { + t.Fatalf("permissions were not refreshed: want=%s given=%s", got.modeWant, got.modeGiven) + } +} + +func TestP2PPublishReloadsStalePermissions(t *testing.T) { + helper := TopicTestHelper{} + helper.setUp(t, 2, types.TopicCatP2P, "p2p-test", true) + defer helper.tearDown() + + uid := helper.uids[0] + stale := types.ModeCP2P &^ types.ModeWrite + pud := helper.topic.perUser[uid] + pud.modeWant = stale + pud.modeGiven = stale + helper.topic.perUser[uid] = pud + helper.ss.EXPECT().Get(types.TenantID(1), "p2p-test", uid, true).Return(&types.Subscription{ + TenantID: 1, User: uid.String(), Topic: "p2p-test", + ModeWant: types.ModeCP2P, ModeGiven: types.ModeCP2P, + }, nil) + helper.mm.EXPECT().Save(types.TenantID(1), gomock.Any(), gomock.Any(), gomock.Any()). + Return(&types.MessageSaveResult{SeqId: 1, MarkedReadBySender: true}, nil) + + err := helper.topic.saveAndBroadcastMessage(&ClientComMessage{ + Id: "pub-1", Original: "p2p-test", Timestamp: types.TimeNow(), sess: helper.sessions[0], + }, uid, true, nil, nil, "test") + helper.finish() + if err != nil { + t.Fatalf("publish failed after permission reload: %v", err) + } +} diff --git a/server/topic.go b/server/topic.go index 81b4aff..0a4fb51 100644 --- a/server/topic.go +++ b/server/topic.go @@ -678,6 +678,10 @@ func (t *Topic) handleServerMsg(msg *ServerComMessage) { // Ignore message - the topic is paused or being deleted. return } + if msg.Info != nil && msg.Info.What == "p2p" && msg.Info.Event == p2pPermissionsRefreshedEvent { + t.handleP2PPermissionsRefresh(msg.Info) + return + } if msg.Pres != nil { t.handlePresence(msg) } else if msg.Info != nil || msg.Data != nil { @@ -700,6 +704,24 @@ func (t *Topic) handleServerMsg(msg *ServerComMessage) { } } +func (t *Topic) handleP2PPermissionsRefresh(info *MsgServerInfo) { + if t.cat != types.TopicCatP2P || t.isProxy { + return + } + var payload p2pPermissionsPayload + if err := json.Unmarshal(info.Payload, &payload); err != nil { + logs.Warn.Printf("topic[%s]: failed to decode p2p permissions refresh: %v", t.name, err) + return + } + uids := make([]types.Uid, 0, len(payload.UserIDs)) + for _, raw := range payload.UserIDs { + if uid := types.ParseUserId(raw); !uid.IsZero() { + uids = append(uids, uid) + } + } + t.refreshPerUserFromDB(types.ZeroUid, uids...) +} + // flexUID accepts a JSON string or number representation of a user id: the // HTTP interface publishes base32 user id strings, while Kafka admin commands // publish numeric database ids. @@ -1282,8 +1304,15 @@ func (t *Topic) saveAndBroadcastMessage(msg *ClientComMessage, asUid types.Uid, return types.ErrPermissionDenied } } else { + // EnsureP2P may have repaired the subscription in the database while this + // topic was already loaded. Reload once on denied P2P publishes so a lost + // or delayed refresh event cannot leave the conversation permanently stale. + if t.cat == types.TopicCatP2P && userFound && !(pud.modeWant & pud.modeGiven).IsWriter() && store.Subs != nil { + t.refreshPerUserFromDB(asUid, asUid) + pud, userFound = t.perUser[asUid] + } // If it's not 'sys' check write permission. - if !(pud.modeWant & pud.modeGiven).IsWriter() { + if !userFound || !(pud.modeWant & pud.modeGiven).IsWriter() { msg.sess.queueOut(ErrPermissionDenied(msg.Id, t.original(asUid), msg.Timestamp)) return types.ErrPermissionDenied } diff --git a/server/topic_test.go b/server/topic_test.go index 87272e2..bf463bb 100644 --- a/server/topic_test.go +++ b/server/topic_test.go @@ -818,6 +818,10 @@ func TestHandleBroadcastDataMissingWritePermission(t *testing.T) { pud := helper.topic.perUser[uid1] pud.modeGiven = types.ModeRead | types.ModeJoin helper.topic.perUser[uid1] = pud + helper.ss.EXPECT().Get(types.TenantID(1), topicName, uid1, true).Return(&types.Subscription{ + TenantID: 1, User: uid1.String(), Topic: topicName, + ModeWant: pud.modeWant, ModeGiven: pud.modeGiven, + }, nil) // Make test message. from := helper.uids[0].UserId()