fix p2p permission refresh during load tests

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 对档位。

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"
 ```
 

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) {

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

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)

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

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

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
 		}

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