diff --git a/internal/mtprotoedge/layer_seed_test.go b/internal/mtprotoedge/layer_seed_test.go new file mode 100644 index 00000000..7fe4e909 --- /dev/null +++ b/internal/mtprotoedge/layer_seed_test.go @@ -0,0 +1,73 @@ +package mtprotoedge + +import ( + "context" + "testing" + "time" + + "go.uber.org/zap/zaptest" + + "github.com/gotd/td/bin" + "github.com/gotd/td/mt" + "github.com/gotd/td/proto" + "github.com/gotd/td/tg" +) + +func TestSessionManagerSetClientLayerForAuthKey(t *testing.T) { + sm := NewSessionManager(zaptest.NewLogger(t)) + c := &Conn{sessionID: 42, authKeyID: [8]byte{1, 2, 3}} + sm.Register(c) + defer sm.Unregister(c) + + sm.SetClientLayerForAuthKey([8]byte{1, 2, 3}, 42, 225) + if got := c.ClientLayer(); got != 225 { + t.Fatalf("ClientLayer = %d, want 225", got) + } + // 未注册 session：no-op，不 panic。 + sm.SetClientLayerForAuthKey([8]byte{9}, 1, 220) + // 非法 layer：忽略，保留已有值。 + sm.SetClientLayerForAuthKey([8]byte{1, 2, 3}, 42, 0) + if got := c.ClientLayer(); got != 225 { + t.Fatalf("ClientLayer after layer=0 = %d, want 225", got) + } +} + +type seededLayerRPC struct{} + +func (seededLayerRPC) Dispatch(context.Context, [8]byte, int64, *bin.Buffer) (bin.Encoder, error) { + return &tg.Config{ThisDC: 2}, nil +} + +func (seededLayerRPC) NegotiatedLayer([8]byte, int64) (int, bool) { return 225, true } + +// TestRegisterSeedsNegotiatedLayerBeforeFirstRPC 验证连接注册即从 rpc 层播种协商 layer： +// 只发一条 ping（服务消息，不经 RPC Dispatch），连接的 ClientLayer 就必须是协商值， +// 而不是等首条 RPC 的 Dispatch 返回后才刷新——否则重连老客户端在首条 RPC handler +// 执行期间收到的 pending flush / 并发 push 会按 canonical 227 漏降级。 +func TestRegisterSeedsNegotiatedLayerBeforeFirstRPC(t *testing.T) { + addr, pub, srv := startTestServer(t, Options{DC: 2, RPC: seededLayerRPC{}}) + conn, auth, cipher := dialHandshake(t, addr, 2, pub) + + clientMsgID := proto.NewMessageIDGen(time.Now) + sendEncrypted(t, conn, cipher, auth, clientMsgID.New(proto.MessageFromClient), &mt.PingRequest{PingID: 7}) + + // 等 pong 回来，确保携带注册动作的那一帧已处理完成。 + gotPong := false + for i := 0; i < 12 && !gotPong; i++ { + _, id, _ := readServerMessage(t, conn, cipher, auth.AuthKey) + gotPong = id == mt.PongTypeID + } + if !gotPong { + t.Fatal("missing pong for ping") + } + + srv.conns.mu.RLock() + c := srv.conns.bySession[sessionKey{authKeyID: auth.AuthKey.ID, sessionID: auth.SessionID}] + srv.conns.mu.RUnlock() + if c == nil { + t.Fatal("connection not registered") + } + if got := c.ClientLayer(); got != 225 { + t.Fatalf("ClientLayer after registration = %d, want seeded 225", got) + } +} diff --git a/internal/mtprotoedge/pending_flush_test.go b/internal/mtprotoedge/pending_flush_test.go index 9149ad49..89326c30 100644 --- a/internal/mtprotoedge/pending_flush_test.go +++ b/internal/mtprotoedge/pending_flush_test.go @@ -27,7 +27,7 @@ func TestSetReceivesUpdatesFlushesPendingBeforeActivation(t *testing.T) { // 完全就绪还要求 membership 路由建立（ReceivesUpdatesForAuthKey 的另一半条件）。 srv.Conns().BindUserForAuthKey(raw, auth.SessionID, 100) - srv.Conns().SetSessionChannelMemberships(raw, auth.SessionID, 100, nil) + srv.Conns().SetSessionChannelMemberships(raw, auth.SessionID, 100, nil, srv.Conns().ChannelMembershipGeneration(raw, auth.SessionID)) // 未就绪：推送进 pending 而非直发。 for i := 0; i < 2; i++ { diff --git a/internal/mtprotoedge/session_manager.go b/internal/mtprotoedge/session_manager.go index 0c14532c..bdcf4e0e 100644 --- a/internal/mtprotoedge/session_manager.go +++ b/internal/mtprotoedge/session_manager.go @@ -8,6 +8,7 @@ import ( "time" "go.uber.org/zap" + "go.uber.org/zap/zapcore" "github.com/gotd/td/bin" "github.com/gotd/td/proto" @@ -128,14 +129,16 @@ func (m *SessionManager) Register(c *Conn) { replaced = old m.removeLocked(old, false) } else if existing := m.byAuthKey[c.authKeyID]; len(existing) >= maxSessionsPerAuthKey { - // 同 raw auth_key 的 session 数达上限且本次是新 session：驱逐一个现有 session 让位， + // 同 raw auth_key 的 session 数达上限且本次是新 session：驱逐建连最早的 session 让位， // 防对抗客户端用海量 session_id 撑爆索引。驱逐对象与新连接同属一个设备凭据， - // 触顶基本是该凭据自身异常。被驱逐连接的 serveConn 会在下一帧因 actor 已关而退出。 + // 触顶基本是该凭据自身异常；选最旧而非 map 随机，避免误杀刚建立的活跃下载/主连接。 + // 被驱逐连接的 serveConn 会在下一帧因 actor 已关而退出。O(cap) 扫描仅在触顶时发生。 for _, ec := range existing { - evicted = ec - m.removeLocked(ec, true) - break + if evicted == nil || ec.createdAt.Before(evicted.createdAt) { + evicted = ec + } } + m.removeLocked(evicted, true) m.log.Debug("Evicted oldest session for auth key at cap", zap.String("auth_key_id", sessionKeyLog(c.authKeyID)), zap.Int("cap", maxSessionsPerAuthKey), @@ -280,7 +283,7 @@ func (m *SessionManager) bindUserLocked(c *Conn, key sessionKey, userID int64) { removeUserIndex(m.byUser, old, key) if old != userID { m.clearChannelInterestsLocked(key) - m.clearChannelMembershipsLocked(key) + m.clearChannelMembershipsLocked(c, key) c.membershipsSynced.Store(false) // 身份变化即丢弃暂存推送：它们属于前一个账号，flush 给新账号是跨账号泄露。 // 同时取消进行中的排空（runFlush 还另有 owner 校验做批内兜底）。 @@ -293,7 +296,7 @@ func (m *SessionManager) bindUserLocked(c *Conn, key sessionKey, userID int64) { addUserIndex(m.byUser, userID, key, c) } else { m.clearChannelInterestsLocked(key) - m.clearChannelMembershipsLocked(key) + m.clearChannelMembershipsLocked(c, key) c.membershipsSynced.Store(false) delete(m.pending, key) delete(m.flushing, key) @@ -394,7 +397,7 @@ func (m *SessionManager) bindAuthKeyLocked(c *Conn, key sessionKey, authKeyID [8 removeUserIndex(m.byUser, oldUserID, key) } m.clearChannelInterestsLocked(key) - m.clearChannelMembershipsLocked(key) + m.clearChannelMembershipsLocked(c, key) c.membershipsSynced.Store(false) delete(m.pending, key) delete(m.flushing, key) @@ -514,7 +517,7 @@ func (m *SessionManager) UnbindAuthKey(authKeyID [8]byte) int { removeUserIndex(m.byUser, old, key) } m.clearChannelInterestsLocked(key) - m.clearChannelMembershipsLocked(key) + m.clearChannelMembershipsLocked(c, key) c.membershipsSynced.Store(false) // 授权解除后暂存推送属于已登出的账号，不能等下一个登录者置位时 flush 出去。 delete(m.pending, key) @@ -556,7 +559,7 @@ func (m *SessionManager) setReceivesUpdatesLocked(c *Conn, key sessionKey, recei if !receives { c.receivesUpdates.Store(false) m.clearChannelInterestsLocked(key) - m.clearChannelMembershipsLocked(key) + m.clearChannelMembershipsLocked(c, key) c.membershipsSynced.Store(false) // 取消进行中的排空激活：runFlush 在置位前会复查该标志，标志已删则放弃置位， // 避免把刚置 false 的开关翻回 true。 @@ -679,6 +682,22 @@ func (m *SessionManager) ReceivesUpdatesForAuthKey(authKeyID [8]byte, sessionID return ok && c.receivesUpdates.Load() && c.membershipsSynced.Load() } +// SetClientLayerForAuthKey 把协商的 TL layer 即时写到指定连接。由 rpc 层在 +// invokeWithLayer 观测到新 layer 时（Dispatch 入口，早于鉴权门与 updates 就绪 +// 置位）调用，使同一条请求触发的 pending flush 与并发 push 立即按正确 layer +// 降级，不必等连接层在 Dispatch 返回后的兜底刷新。 +func (m *SessionManager) SetClientLayerForAuthKey(authKeyID [8]byte, sessionID int64, layer int) { + if m == nil || layer <= 0 { + return + } + m.mu.RLock() + c, ok := m.bySession[sessionKey{authKeyID: authKeyID, sessionID: sessionID}] + m.mu.RUnlock() + if ok { + c.SetClientLayer(layer) + } +} + // SetReceivesUpdatesForAuthKey 标记指定 raw auth_key_id + session_id 是否接收主动 updates。 func (m *SessionManager) SetReceivesUpdatesForAuthKey(authKeyID [8]byte, sessionID int64, receives bool) { m.mu.Lock() @@ -905,14 +924,21 @@ func onceEncodedOutbound(msg bin.Encoder) func() (*encodedOutboundMessage, error } func (m *SessionManager) pushToUserWithSender(ctx context.Context, userID int64, excludeAuthKeyID *[8]byte, excludeSessionID int64, t proto.MessageType, msg bin.Encoder, queueWhenNotReady bool, send func(*Conn) error) (int, error) { - m.mu.Lock() + // push fan-out 是连接层最热路径之一：debug 日志的字段构造（含 auth_key hex 格式化） + // 在关闭 debug 时也会求值，先查级别一次、按需记日志。 + debug := m.log.Core().Enabled(zapcore.DebugLevel) + // 快路径：稳态下目标连接全部就绪（或 transient 直接跳过未就绪者），收集连接 + // 只读不写，全程共享读锁即可，避免每次 push 都拿独占写锁串行整个注册表。 + // 仅当 durable 推送遇到未就绪连接（需写 pending）时才回落写锁重扫。 + m.mu.RLock() total := len(m.byUser[userID]) conns := make([]*Conn, 0, total) queued := 0 dropped := 0 excluded := 0 skipped := 0 - for key, c := range m.byUser[userID] { + needQueue := false + for _, c := range m.byUser[userID] { if shouldExcludeSession(c, excludeAuthKeyID, excludeSessionID) { excluded++ continue @@ -924,26 +950,53 @@ func (m *SessionManager) pushToUserWithSender(ctx context.Context, userID int64, skipped++ continue } - if m.queueLocked(key, t, msg) { - queued++ - m.log.Debug("Push queued (session not updates-ready)", - zap.Int64("user_id", userID), - zap.String("auth_key_id", sessionKeyLog(key.authKeyID)), - zap.Int64("session_id", key.sessionID), - ) - } else { - dropped++ - m.log.Debug("Push dropped (stale pending; durable log covers)", - zap.Int64("user_id", userID), - zap.String("auth_key_id", sessionKeyLog(key.authKeyID)), - zap.Int64("session_id", key.sessionID), - ) - } - continue + needQueue = true + break } conns = append(conns, c) } - m.mu.Unlock() + m.mu.RUnlock() + if needQueue { + // 写锁下完整重扫（读锁释放到此之间状态可能变化，以重扫结果为准）。 + conns = conns[:0] + queued, dropped, excluded, skipped = 0, 0, 0, 0 + m.mu.Lock() + total = len(m.byUser[userID]) + for key, c := range m.byUser[userID] { + if shouldExcludeSession(c, excludeAuthKeyID, excludeSessionID) { + excluded++ + continue + } + if !c.receivesUpdates.Load() { + if !queueWhenNotReady { + skipped++ + continue + } + if m.queueLocked(key, t, msg) { + queued++ + if debug { + m.log.Debug("Push queued (session not updates-ready)", + zap.Int64("user_id", userID), + zap.String("auth_key_id", sessionKeyLog(key.authKeyID)), + zap.Int64("session_id", key.sessionID), + ) + } + } else { + dropped++ + if debug { + m.log.Debug("Push dropped (stale pending; durable log covers)", + zap.Int64("user_id", userID), + zap.String("auth_key_id", sessionKeyLog(key.authKeyID)), + zap.Int64("session_id", key.sessionID), + ) + } + } + continue + } + conns = append(conns, c) + } + m.mu.Unlock() + } var firstErr error sent := 0 @@ -958,33 +1011,39 @@ func (m *SessionManager) pushToUserWithSender(ctx context.Context, userID int64, if firstErr == nil { firstErr = err } - m.log.Debug("Push to conn failed", - zap.Int64("user_id", userID), - zap.String("auth_key_id", sessionKeyLog(c.authKeyID)), - zap.Int64("session_id", c.sessionID), - zap.Error(err), - ) + if debug { + m.log.Debug("Push to conn failed", + zap.Int64("user_id", userID), + zap.String("auth_key_id", sessionKeyLog(c.authKeyID)), + zap.Int64("session_id", c.sessionID), + zap.Error(err), + ) + } continue } sent++ - m.log.Debug("Push to conn ok", - zap.Int64("user_id", userID), - zap.String("auth_key_id", sessionKeyLog(c.authKeyID)), - zap.Int64("session_id", c.sessionID), - ) + if debug { + m.log.Debug("Push to conn ok", + zap.Int64("user_id", userID), + zap.String("auth_key_id", sessionKeyLog(c.authKeyID)), + zap.Int64("session_id", c.sessionID), + ) + } } - if total == 0 { - m.log.Debug("Push to user: no active conns", zap.Int64("user_id", userID)) - } else if excluded > 0 || queued > 0 || dropped > 0 || skipped > 0 || sent < len(conns) { - m.log.Debug("Push to user summary", - zap.Int64("user_id", userID), - zap.Int("conns", total), - zap.Int("sent", sent), - zap.Int("queued", queued), - zap.Int("dropped", dropped), - zap.Int("skipped_transient", skipped), - zap.Int("excluded", excluded), - ) + if debug { + if total == 0 { + m.log.Debug("Push to user: no active conns", zap.Int64("user_id", userID)) + } else if excluded > 0 || queued > 0 || dropped > 0 || skipped > 0 || sent < len(conns) { + m.log.Debug("Push to user summary", + zap.Int64("user_id", userID), + zap.Int("conns", total), + zap.Int("sent", sent), + zap.Int("queued", queued), + zap.Int("dropped", dropped), + zap.Int("skipped_transient", skipped), + zap.Int("excluded", excluded), + ) + } } return sent + queued, firstErr } @@ -1014,7 +1073,7 @@ func (m *SessionManager) OnlineUserIDsForCandidates(candidateUserIDs []int64, li } m.mu.RLock() defer m.mu.RUnlock() - out := make([]int64, 0, minInt(len(candidateUserIDs), positiveLimitOrLen(limit, len(candidateUserIDs)))) + out := make([]int64, 0, min(len(candidateUserIDs), positiveLimitOrLen(limit, len(candidateUserIDs)))) seen := make(map[int64]struct{}, len(candidateUserIDs)) for _, userID := range candidateUserIDs { if userID == 0 { @@ -1078,10 +1137,29 @@ func (m *SessionManager) OnlineChannelUserIDs(channelID int64, limit int) []int6 return m.onlineChannelUsers(m.byChannel, channelID, limit) } +// ChannelMembershipGeneration 返回该 session 的 membership 索引修订号。 +// 全量同步方必须在读取持久成员列表【之前】采样，并经 SetSessionChannelMemberships +// 带回比对；session 不在线时返回 0（后续 Set 也会因查不到连接而放弃）。 +func (m *SessionManager) ChannelMembershipGeneration(rawAuthKeyID [8]byte, sessionID int64) int64 { + m.mu.RLock() + c, ok := m.bySession[sessionKey{authKeyID: rawAuthKeyID, sessionID: sessionID}] + m.mu.RUnlock() + if !ok { + return 0 + } + return c.membershipGen.Load() +} + // SetSessionChannelMemberships replaces the joined-channel index for one // updates-ready session. This index is broader than TrackChannelInterest and is // used for durable channel updates such as new/edit/delete message. -func (m *SessionManager) SetSessionChannelMemberships(rawAuthKeyID [8]byte, sessionID, userID int64, channelIDs []int64) { +// +// expectedGen 是调用方在读取持久成员列表前经 ChannelMembershipGeneration 采样的 +// 修订号。若落地时修订号已变（读取窗口内发生了 join/leave/kick 的增量修订或整体 +// 清除），全量替换会覆盖掉窗口内的增量——此时改走并集合并（保留增量 Add；合并回的 +// stale 条目由 fan-out 前的 PG active 复核兜底），并保持 membershipsSynced=false， +// 让下一条 RPC 重新走全量同步收敛。 +func (m *SessionManager) SetSessionChannelMemberships(rawAuthKeyID [8]byte, sessionID, userID int64, channelIDs []int64, expectedGen int64) { key := sessionKey{authKeyID: rawAuthKeyID, sessionID: sessionID} m.mu.Lock() defer m.mu.Unlock() @@ -1089,11 +1167,22 @@ func (m *SessionManager) SetSessionChannelMemberships(rawAuthKeyID [8]byte, sess if !ok { return } - m.clearChannelMembershipsLocked(key) - c.membershipsSynced.Store(false) if userID == 0 || c.userID.Load() != userID { + m.clearChannelMembershipsLocked(c, key) + c.membershipsSynced.Store(false) return } + if c.membershipGen.Load() != expectedGen { + c.membershipsSynced.Store(false) + m.trackChannelIndexLocked(m.byMemberChannel, m.bySessionMembers, key, userID, channelIDs) + m.log.Debug("Channel membership sync raced with incremental updates; merged and kept unsynced", + zap.String("auth_key_id", sessionKeyLog(rawAuthKeyID)), + zap.Int64("session_id", sessionID), + ) + return + } + m.clearChannelMembershipsLocked(c, key) + c.membershipsSynced.Store(false) m.trackChannelIndexLocked(m.byMemberChannel, m.bySessionMembers, key, userID, channelIDs) c.membershipsSynced.Store(true) } @@ -1110,6 +1199,7 @@ func (m *SessionManager) AddUserChannelMembership(userID, channelID int64) { if c == nil || c.userID.Load() != userID { continue } + c.membershipGen.Add(1) m.trackChannelIndexLocked(m.byMemberChannel, m.bySessionMembers, key, userID, []int64{channelID}) } } @@ -1122,7 +1212,10 @@ func (m *SessionManager) RemoveUserChannelMembership(userID, channelID int64) { } m.mu.Lock() defer m.mu.Unlock() - for key := range m.byUser[userID] { + for key, c := range m.byUser[userID] { + if c != nil { + c.membershipGen.Add(1) + } m.removeChannelIndexLocked(m.byMemberChannel, m.bySessionMembers, key, channelID) } } @@ -1151,7 +1244,7 @@ func (m *SessionManager) OnlineChannelMemberUserIDsExcluding(channelID int64, ex if len(sessions) == 0 { return nil } - out := make([]int64, 0) + out := make([]int64, 0, positiveLimitOrLen(limit, len(sessions))) seen := make(map[int64]struct{}, len(sessions)) for key, userID := range sessions { if userID == 0 { @@ -1219,7 +1312,7 @@ func (m *SessionManager) removeLocked(c *Conn, dropPending bool) int64 { removeUserIndex(m.byUser, uid, key) } m.clearChannelInterestsLocked(key) - m.clearChannelMembershipsLocked(key) + m.clearChannelMembershipsLocked(c, key) if dropPending { delete(m.pending, key) } @@ -1247,7 +1340,10 @@ func (m *SessionManager) clearChannelInterestsLocked(key sessionKey) { m.clearChannelIndexLocked(m.byChannel, m.bySessionChannels, key) } -func (m *SessionManager) clearChannelMembershipsLocked(key sessionKey) { +// clearChannelMembershipsLocked 整体清除某连接的 membership 索引并递增其修订号， +// 使在飞的全量同步（SetSessionChannelMemberships）能检测到清除并放弃过期替换。 +func (m *SessionManager) clearChannelMembershipsLocked(c *Conn, key sessionKey) { + c.membershipGen.Add(1) m.clearChannelIndexLocked(m.byMemberChannel, m.bySessionMembers, key) } @@ -1322,13 +1418,6 @@ func positiveLimitOrLen(limit, length int) int { return length } -func minInt(a, b int) int { - if a < b { - return a - } - return b -} - func (m *SessionManager) takePendingLocked(key sessionKey, ready bool) []queuedPush { if !ready || len(m.pending[key]) == 0 { return nil diff --git a/internal/mtprotoedge/session_manager_test.go b/internal/mtprotoedge/session_manager_test.go index c815b1a1..6153355b 100644 --- a/internal/mtprotoedge/session_manager_test.go +++ b/internal/mtprotoedge/session_manager_test.go @@ -260,7 +260,7 @@ func TestSessionManagerChannelInterestIndex(t *testing.T) { if got := sm.OnlineChannelMemberUserIDs(10, 10); len(got) != 0 { t.Fatalf("channel 10 online members before membership sync = %v, want empty", got) } - sm.SetSessionChannelMemberships(raw, 42, 100, []int64{10, 30}) + sm.SetSessionChannelMemberships(raw, 42, 100, []int64{10, 30}, sm.ChannelMembershipGeneration(raw, 42)) if got := sm.OnlineChannelMemberUserIDs(10, 10); len(got) != 1 || got[0] != 100 { t.Fatalf("channel 10 online members = %v, want [100]", got) } @@ -315,7 +315,7 @@ func TestSessionManagerClearsChannelIndexesOnAuthAndReadinessChanges(t *testing. track := func() { sm.TrackChannelInterest(raw, 42, 100, []int64{10}) - sm.SetSessionChannelMemberships(raw, 42, 100, []int64{10}) + sm.SetSessionChannelMemberships(raw, 42, 100, []int64{10}, sm.ChannelMembershipGeneration(raw, 42)) if got := sm.OnlineChannelUserIDs(10, 10); len(got) != 1 || got[0] != 100 { t.Fatalf("channel viewers before cleanup = %v, want [100]", got) } diff --git a/internal/mtprotoedge/session_membership_gen_test.go b/internal/mtprotoedge/session_membership_gen_test.go new file mode 100644 index 00000000..47811cb0 --- /dev/null +++ b/internal/mtprotoedge/session_membership_gen_test.go @@ -0,0 +1,95 @@ +package mtprotoedge + +import ( + "testing" + "time" + + "go.uber.org/zap/zaptest" +) + +// TestSetSessionChannelMembershipsDetectsConcurrentIncrementalUpdates 验证全量 +// membership 同步的丢失更新防护：同步方在读持久成员列表前采样修订号，读取窗口内 +// 若发生增量 join/leave（另一设备操作经 Add/RemoveUserChannelMembership 落索引）， +// 携带过期修订号的全量替换必须改走并集合并（不得覆盖增量），并保持 +// membershipsSynced=false 促使下一条 RPC 重试全量同步收敛。 +func TestSetSessionChannelMembershipsDetectsConcurrentIncrementalUpdates(t *testing.T) { + sm := NewSessionManager(zaptest.NewLogger(t)) + raw := [8]byte{1, 2, 3} + c := &Conn{sessionID: 42, authKeyID: raw} + sm.Register(c) + sm.BindUserForAuthKey(raw, 42, 100) + sm.SetReceivesUpdatesForAuthKey(raw, 42, true) + + // 同步方采样修订号后、全量列表落地前，用户在另一台设备加入了频道 7。 + gen := sm.ChannelMembershipGeneration(raw, 42) + sm.AddUserChannelMembership(100, 7) + // 基于旧快照的全量列表（只有频道 5，不含 7）携带过期修订号落地。 + sm.SetSessionChannelMemberships(raw, 42, 100, []int64{5}, gen) + + if got := sm.OnlineChannelMemberUserIDs(7, 10); len(got) != 1 || got[0] != 100 { + t.Fatalf("channel 7 members = %v, want [100]: full replace overwrote the in-window incremental join", got) + } + if got := sm.OnlineChannelMemberUserIDs(5, 10); len(got) != 1 || got[0] != 100 { + t.Fatalf("channel 5 members = %v, want [100]: merge path must still apply the full list", got) + } + if sm.ReceivesUpdatesForAuthKey(raw, 42) { + t.Fatal("session fully ready despite raced membership sync; retry would never happen") + } + + // 重试：新修订号下的全量同步正常替换并置就绪。 + sm.SetSessionChannelMemberships(raw, 42, 100, []int64{5, 7}, sm.ChannelMembershipGeneration(raw, 42)) + if !sm.ReceivesUpdatesForAuthKey(raw, 42) { + t.Fatal("session not ready after clean resync") + } + + // 反方向：窗口内被移出频道 5，stale 全量含 5 → 合并会短暂保留 stale 条目 + // （fan-out 前的 PG active 复核兜底），但必须保持未就绪等待重试。 + gen = sm.ChannelMembershipGeneration(raw, 42) + sm.RemoveUserChannelMembership(100, 5) + sm.SetSessionChannelMemberships(raw, 42, 100, []int64{5, 7}, gen) + if sm.ReceivesUpdatesForAuthKey(raw, 42) { + t.Fatal("session ready despite raced removal during sync") + } + sm.SetSessionChannelMemberships(raw, 42, 100, []int64{7}, sm.ChannelMembershipGeneration(raw, 42)) + if got := sm.OnlineChannelMemberUserIDs(5, 10); len(got) != 0 { + t.Fatalf("channel 5 members after resync = %v, want empty", got) + } + if !sm.ReceivesUpdatesForAuthKey(raw, 42) { + t.Fatal("session not ready after final resync") + } +} + +// TestRegisterEvictsOldestSessionAtCap 验证同 raw auth_key session 数触顶时驱逐的是 +// 建连最早的连接，而不是 map 迭代顺序下的随机一个（随机可能误杀刚建立的活跃连接）。 +func TestRegisterEvictsOldestSessionAtCap(t *testing.T) { + sm := NewSessionManager(zaptest.NewLogger(t)) + raw := [8]byte{9} + base := time.Unix(1_700_000_000, 0) + + const oldestSession = int64(100) + for i := 0; i < maxSessionsPerAuthKey; i++ { + sid := int64(i + 1) + created := base.Add(time.Duration(i+1) * time.Second) + if sid == oldestSession { + created = base // 唯一早于所有其它连接的时间戳，且故意不在注册顺序首位。 + } + sm.Register(&Conn{sessionID: sid, authKeyID: raw, createdAt: created}) + } + + sm.Register(&Conn{sessionID: 9999, authKeyID: raw, createdAt: base.Add(time.Hour)}) + + sm.mu.RLock() + _, oldestAlive := sm.bySession[sessionKey{authKeyID: raw, sessionID: oldestSession}] + _, newestAlive := sm.bySession[sessionKey{authKeyID: raw, sessionID: 9999}] + total := len(sm.byAuthKey[raw]) + sm.mu.RUnlock() + if oldestAlive { + t.Fatal("oldest session survived eviction at cap") + } + if !newestAlive { + t.Fatal("newly registered session missing after eviction") + } + if total != maxSessionsPerAuthKey { + t.Fatalf("sessions for auth key = %d, want cap %d", total, maxSessionsPerAuthKey) + } +} diff --git a/internal/mtprotoedge/session_ready_test.go b/internal/mtprotoedge/session_ready_test.go index a326121c..93ac817b 100644 --- a/internal/mtprotoedge/session_ready_test.go +++ b/internal/mtprotoedge/session_ready_test.go @@ -23,23 +23,23 @@ func TestReceivesUpdatesForAuthKeyRequiresMembershipSync(t *testing.T) { t.Fatal("ready before membership sync — a failed sync would never be retried") } - sm.SetSessionChannelMemberships(raw, 42, 100, []int64{7}) + sm.SetSessionChannelMemberships(raw, 42, 100, []int64{7}, sm.ChannelMembershipGeneration(raw, 42)) if !sm.ReceivesUpdatesForAuthKey(raw, 42) { t.Fatal("not ready after successful membership sync") } // 没有任何频道的账号：空列表的成功同步同样算就绪。 - sm.SetSessionChannelMemberships(raw, 42, 100, nil) + sm.SetSessionChannelMemberships(raw, 42, 100, nil, sm.ChannelMembershipGeneration(raw, 42)) if !sm.ReceivesUpdatesForAuthKey(raw, 42) { t.Fatal("not ready after successful empty membership sync") } // userID 与连接当前绑定不一致（换号竞态）时不算就绪，等正确身份重试。 - sm.SetSessionChannelMemberships(raw, 42, 999, []int64{7}) + sm.SetSessionChannelMemberships(raw, 42, 999, []int64{7}, sm.ChannelMembershipGeneration(raw, 42)) if sm.ReceivesUpdatesForAuthKey(raw, 42) { t.Fatal("ready after membership sync for mismatched user") } - sm.SetSessionChannelMemberships(raw, 42, 100, []int64{7}) + sm.SetSessionChannelMemberships(raw, 42, 100, []int64{7}, sm.ChannelMembershipGeneration(raw, 42)) // 登出清除就绪标志。 sm.BindUserForAuthKey(raw, 42, 0) diff --git a/internal/rpc/auth_qr_login_test.go b/internal/rpc/auth_qr_login_test.go index 38c79e75..eba2df6d 100644 --- a/internal/rpc/auth_qr_login_test.go +++ b/internal/rpc/auth_qr_login_test.go @@ -101,8 +101,8 @@ func TestAuthLoginTokenAcceptedByAndroidBindsTargetSession(t *testing.T) { } snap := sessions.snapshot() - if sessions.scopedAuthKeyID != targetRawAuthKeyID { - t.Fatalf("scoped raw auth key = %x, want %x", sessions.scopedAuthKeyID, targetRawAuthKeyID) + if got := sessions.scopedAuthKey(); got != targetRawAuthKeyID { + t.Fatalf("scoped raw auth key = %x, want %x", got, targetRawAuthKeyID) } if snap.sessionID != targetSession || snap.userID != scannerUserID || !snap.userResolved { t.Fatalf("target session snapshot = %+v, want session/user/resolved %d/%d/true", snap, targetSession, scannerUserID) @@ -110,7 +110,7 @@ func TestAuthLoginTokenAcceptedByAndroidBindsTargetSession(t *testing.T) { if snap.messageType != proto.MessageFromServer { t.Fatalf("push message type = %v, want MessageFromServer", snap.messageType) } - if !sessions.immediatePush { + if !sessions.immediatePushSeen() { t.Fatal("login token update was not pushed through the immediate pre-auth path") } short, ok := snap.message.(*tg.UpdateShort) diff --git a/internal/rpc/channel_interest.go b/internal/rpc/channel_interest.go index 1ba2a88a..f65ed54c 100644 --- a/internal/rpc/channel_interest.go +++ b/internal/rpc/channel_interest.go @@ -53,6 +53,10 @@ func (r *Router) syncSessionChannelMemberships(ctx context.Context, userID int64 if !ok { return } + // 在读取持久成员列表之前采样修订号：读取窗口内若发生增量 join/leave + // （AddUserChannelMembership/RemoveUserChannelMembership），全量替换会覆盖增量， + // SetSessionChannelMemberships 据此改走合并路径并保持未就绪重试。 + expectedGen := provider.ChannelMembershipGeneration(rawAuthKeyID, sessionID) channelIDs := make([]int64, 0, channelMembershipSyncPageSize) after := int64(0) for { @@ -82,7 +86,7 @@ func (r *Router) syncSessionChannelMemberships(ctx context.Context, userID int64 break } } - provider.SetSessionChannelMemberships(rawAuthKeyID, sessionID, userID, channelIDs) + provider.SetSessionChannelMemberships(rawAuthKeyID, sessionID, userID, channelIDs, expectedGen) } func (r *Router) addOnlineChannelMemberships(channelID int64, userIDs ...int64) { diff --git a/internal/rpc/channels_invites_members_rpc_test.go b/internal/rpc/channels_invites_members_rpc_test.go index 6639c46a..f6f520d2 100644 --- a/internal/rpc/channels_invites_members_rpc_test.go +++ b/internal/rpc/channels_invites_members_rpc_test.go @@ -902,6 +902,70 @@ func TestChannelEditBannedKickNotifiesKickedViewer(t *testing.T) { } } +// TestChannelEditBannedMaintainsOnlineMembershipIndex 验证 editBanned 与 editAdmin 对称 +// 维护在线成员推送路由索引：踢出/封禁后必须立即摘除（否则被踢在线成员保留 stale +// byMemberChannel 条目直到断线，占用实时 fan-out cap 名额，且群通话推送会继续投递）； +// 仅限制权限（仍为 active 成员）时索引保留。 +func TestChannelEditBannedMaintainsOnlineMembershipIndex(t *testing.T) { + ctx := context.Background() + userStore := memory.NewUserStore() + owner, _ := userStore.Create(ctx, domain.User{AccessHash: 61, Phone: "15550002261", FirstName: "Owner"}) + target, _ := userStore.Create(ctx, domain.User{AccessHash: 62, Phone: "15550002262", FirstName: "Target"}) + sessions := &captureSessions{} + r := New(Config{}, Deps{ + Users: appusers.NewService(userStore), + Channels: appchannels.NewService(memory.NewChannelStore()), + Sessions: sessions, + }, zaptest.NewLogger(t), clock.System) + created, err := r.onMessagesCreateChat(WithUserID(ctx, owner.ID), &tg.MessagesCreateChatRequest{ + Users: []tg.InputUserClass{&tg.InputUser{UserID: target.ID, AccessHash: target.AccessHash}}, + Title: "Kick Index", + }) + if err != nil { + t.Fatalf("create chat: %v", err) + } + channel := created.Updates.(*tg.Updates).Chats[0].(*tg.Channel) + input := &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash} + contains := func(ids []int64, id int64) bool { + for _, v := range ids { + if v == id { + return true + } + } + return false + } + if !contains(sessions.onlineChannelMemberIDs(channel.ID), target.ID) { + t.Fatalf("membership index after create = %v, want target %d", sessions.onlineChannelMemberIDs(channel.ID), target.ID) + } + + // 仅限制发言（仍为 active 成员）：索引保留。 + if _, err := r.onChannelsEditBanned(WithUserID(ctx, owner.ID), &tg.ChannelsEditBannedRequest{ + Channel: input, + Participant: &tg.InputPeerUser{UserID: target.ID, AccessHash: target.AccessHash}, + BannedRights: tg.ChatBannedRights{SendMessages: true}, + }); err != nil { + t.Fatalf("restrict member: %v", err) + } + if !contains(sessions.onlineChannelMemberIDs(channel.ID), target.ID) { + t.Fatalf("membership index after restrict = %v, restricted member must stay routed", sessions.onlineChannelMemberIDs(channel.ID)) + } + + // 踢出（view_messages）：索引必须立即摘除，其他成员不受影响。 + if _, err := r.onChannelsEditBanned(WithUserID(ctx, owner.ID), &tg.ChannelsEditBannedRequest{ + Channel: input, + Participant: &tg.InputPeerUser{UserID: target.ID, AccessHash: target.AccessHash}, + BannedRights: tg.ChatBannedRights{ViewMessages: true}, + }); err != nil { + t.Fatalf("kick member: %v", err) + } + if contains(sessions.onlineChannelMemberIDs(channel.ID), target.ID) { + t.Fatalf("membership index after kick = %v, kicked member must be removed", sessions.onlineChannelMemberIDs(channel.ID)) + } + if !contains(sessions.onlineChannelMemberIDs(channel.ID), owner.ID) { + t.Fatalf("membership index after kick = %v, owner must survive", sessions.onlineChannelMemberIDs(channel.ID)) + } +} + func TestChannelInviteKickedMemberRPC(t *testing.T) { ctx := context.Background() userStore := memory.NewUserStore() diff --git a/internal/rpc/channels_members.go b/internal/rpc/channels_members.go index 8c8e2058..2c89a593 100644 --- a/internal/rpc/channels_members.go +++ b/internal/rpc/channels_members.go @@ -450,6 +450,14 @@ func (r *Router) onChannelsEditBanned(ctx context.Context, req *tg.ChannelsEditB return nil, channelAdminErr(err) } r.invalidateChannelFullBotInfoCacheForChannel(res.Channel.ID) + // 与 onChannelsEditAdmin 对称维护在线成员路由索引：踢出/封禁后立即摘除， + // 否则被踢在线成员会留着 stale byMemberChannel 条目直到断线（占用实时 + // fan-out cap 名额，且群通话推送等未过 PG 复核的路径会继续投递）。 + if res.Participant.Status == domain.ChannelMemberActive { + r.addOnlineChannelMemberships(res.Channel.ID, res.Participant.UserID) + } else { + r.removeOnlineChannelMemberships(res.Channel.ID, res.Participant.UserID) + } if res.Participant.Status == domain.ChannelMemberKicked && res.Previous.Status == domain.ChannelMemberActive { r.recordChannelStateForUser(ctx, res.Participant.UserID, res.Channel.ID, false) } diff --git a/internal/rpc/channels_updates_rpc_test.go b/internal/rpc/channels_updates_rpc_test.go index 748a9477..736e7506 100644 --- a/internal/rpc/channels_updates_rpc_test.go +++ b/internal/rpc/channels_updates_rpc_test.go @@ -115,7 +115,7 @@ func TestChannelSendHistoryAndDifferenceRPC(t *testing.T) { if pushed.userID != friend.ID || pushed.sessionID != 77 || pushed.messageType != proto.MessageFromServer { t.Fatalf("pushed channel update = user %d exclude session %d type %v, want friend/exclude/from_server", pushed.userID, pushed.sessionID, pushed.messageType) } - if gotAuthKeyID := sessions.scopedAuthKeyID; gotAuthKeyID != authKeyID { + if gotAuthKeyID := sessions.scopedAuthKey(); gotAuthKeyID != authKeyID { t.Fatalf("exclude auth_key_id = %x, want %x", gotAuthKeyID, authKeyID) } pushedUpdates, ok := pushed.message.(*tg.Updates) @@ -156,7 +156,7 @@ func TestChannelSendHistoryAndDifferenceRPC(t *testing.T) { if contentPush.userID != friend.ID || contentPush.sessionID != 88 || contentPush.messageType != proto.MessageFromServer { t.Fatalf("content-read push = user %d exclude session %d type %v, want friend/exclude/from_server", contentPush.userID, contentPush.sessionID, contentPush.messageType) } - if gotAuthKeyID := sessions.scopedAuthKeyID; gotAuthKeyID != contentAuthKeyID { + if gotAuthKeyID := sessions.scopedAuthKey(); gotAuthKeyID != contentAuthKeyID { t.Fatalf("content-read exclude auth_key_id = %x, want %x", gotAuthKeyID, contentAuthKeyID) } contentUpdates, ok := contentPush.message.(*tg.Updates) @@ -379,7 +379,7 @@ func TestChannelsReadMessageContentsClearsUnreadReactionAndPushesUpdate(t *testi if pushed.userID != owner.ID || pushed.sessionID != 99 || pushed.messageType != proto.MessageFromServer { t.Fatalf("reaction read push = user %d session %d type %v, want owner/exclude/from_server", pushed.userID, pushed.sessionID, pushed.messageType) } - if gotAuthKeyID := sessions.scopedAuthKeyID; gotAuthKeyID != contentAuthKeyID { + if gotAuthKeyID := sessions.scopedAuthKey(); gotAuthKeyID != contentAuthKeyID { t.Fatalf("reaction read exclude auth_key_id = %x, want %x", gotAuthKeyID, contentAuthKeyID) } pushedUpdates, ok := pushed.message.(*tg.Updates) diff --git a/internal/rpc/deps.go b/internal/rpc/deps.go index d8b6da71..bd3152f1 100644 --- a/internal/rpc/deps.go +++ b/internal/rpc/deps.go @@ -84,6 +84,14 @@ type SessionUpdatesStateProvider interface { ReceivesUpdatesForAuthKey(rawAuthKeyID [8]byte, sessionID int64) bool } +// ClientLayerBinder 把协商 TL layer 即时下推到连接（可选能力）。 +// invokeWithLayer 在 Dispatch 入口被观测到时立即调用，使同一请求 handler 执行期间 +// 触发的 pending flush / 并发 push 就已按正确 layer 降级；不实现时连接层只能靠 +// Dispatch 返回后的兜底刷新，重连老客户端首条 RPC 期间的推送会漏降级。 +type ClientLayerBinder interface { + SetClientLayerForAuthKey(rawAuthKeyID [8]byte, sessionID int64, layer int) +} + // SessionTerminator 暴露按业务 auth_key 强制断开活跃连接的能力（可选）。 // 授权撤销（被踢设备）必须断开连接：出站推送用连接持有的密钥加密、不回查授权， // perm-key 连接的授权缓存也只有断开重连才会重新回查授权表。 @@ -130,7 +138,11 @@ type OnlineUserProvider interface { TrackChannelInterest(rawAuthKeyID [8]byte, sessionID, userID int64, channelIDs []int64) ClearChannelInterest(rawAuthKeyID [8]byte, sessionID, userID int64) OnlineChannelUserIDs(channelID int64, limit int) []int64 - SetSessionChannelMemberships(rawAuthKeyID [8]byte, sessionID, userID int64, channelIDs []int64) + // ChannelMembershipGeneration / SetSessionChannelMemberships 配对使用：调用方在读取 + // 持久成员列表前采样修订号，落地时带回；期间发生增量 Add/Remove 时 Set 走合并路径 + // 并保持未就绪，由下一条 RPC 重试全量同步（防全量替换覆盖窗口内的增量 join/leave）。 + ChannelMembershipGeneration(rawAuthKeyID [8]byte, sessionID int64) int64 + SetSessionChannelMemberships(rawAuthKeyID [8]byte, sessionID, userID int64, channelIDs []int64, expectedGen int64) AddUserChannelMembership(userID, channelID int64) RemoveUserChannelMembership(userID, channelID int64) OnlineChannelMemberUserIDs(channelID int64, limit int) []int64 diff --git a/internal/rpc/dialogs_rpc_test.go b/internal/rpc/dialogs_rpc_test.go index c7f326a8..e7f59425 100644 --- a/internal/rpc/dialogs_rpc_test.go +++ b/internal/rpc/dialogs_rpc_test.go @@ -477,7 +477,7 @@ func TestMessagesSaveDraftPushesDraftUpdateToOtherSessions(t *testing.T) { if got.userID != userID || got.sessionID != 66 || got.messageType != proto.MessageFromServer { t.Fatalf("push = user %d exclude session %d type %v, want self/exclude/from_server", got.userID, got.sessionID, got.messageType) } - if gotAuthKeyID := sessions.scopedAuthKeyID; gotAuthKeyID != authKeyID { + if gotAuthKeyID := sessions.scopedAuthKey(); gotAuthKeyID != authKeyID { t.Fatalf("exclude auth_key_id = %x, want %x", gotAuthKeyID, authKeyID) } updates, ok := got.message.(*tg.Updates) diff --git a/internal/rpc/messages_history_rpc_test.go b/internal/rpc/messages_history_rpc_test.go index ae0947a8..a136fd9c 100644 --- a/internal/rpc/messages_history_rpc_test.go +++ b/internal/rpc/messages_history_rpc_test.go @@ -516,7 +516,7 @@ func TestMessagesSetTypingPushesUserTypingUpdate(t *testing.T) { if got.userID != 1000000002 || got.sessionID != 55 || got.messageType != proto.MessageFromServer { t.Fatalf("push = user %d exclude session %d type %v, want target/exclude/from_server", got.userID, got.sessionID, got.messageType) } - if gotAuthKeyID := sessions.scopedAuthKeyID; gotAuthKeyID != authKeyID { + if gotAuthKeyID := sessions.scopedAuthKey(); gotAuthKeyID != authKeyID { t.Fatalf("exclude auth_key_id = %x, want %x", gotAuthKeyID, authKeyID) } updateShort, ok := got.message.(*tg.UpdateShort) @@ -623,7 +623,7 @@ func TestMessagesSetTypingPushesChannelTypingTopMsgID(t *testing.T) { if got.userID != memberID || got.sessionID != 77 || got.messageType != proto.MessageFromServer { t.Fatalf("channel typing push = user %d exclude session %d type %v, want member/exclude/from_server", got.userID, got.sessionID, got.messageType) } - if gotAuthKeyID := sessions.scopedAuthKeyID; gotAuthKeyID != authKeyID { + if gotAuthKeyID := sessions.scopedAuthKey(); gotAuthKeyID != authKeyID { t.Fatalf("exclude auth_key_id = %x, want %x", gotAuthKeyID, authKeyID) } updates, ok := got.message.(*tg.Updates) diff --git a/internal/rpc/outbox_dispatcher_test.go b/internal/rpc/outbox_dispatcher_test.go index 92478731..8e519788 100644 --- a/internal/rpc/outbox_dispatcher_test.go +++ b/internal/rpc/outbox_dispatcher_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "reflect" + "sync" "testing" "time" @@ -97,8 +98,8 @@ func TestOutboxDispatcherUsesScopedAuthKeyExclusion(t *testing.T) { dispatcher := NewOutboxDispatcher(events, outbox, sessions, zaptest.NewLogger(t)) dispatcher.DispatchOnce(context.Background()) - if sessions.scopedAuthKeyID != excludeAuthKeyID || sessions.sessionID != 99 || sessions.userID != 1000000002 { - t.Fatalf("scoped push = auth %x session %d user %d, want precise outbox exclusion", sessions.scopedAuthKeyID, sessions.sessionID, sessions.userID) + if sessions.scopedAuthKey() != excludeAuthKeyID || sessions.sessionID != 99 || sessions.userID != 1000000002 { + t.Fatalf("scoped push = auth %x session %d user %d, want precise outbox exclusion", sessions.scopedAuthKey(), sessions.sessionID, sessions.userID) } } @@ -731,13 +732,34 @@ type captureDispatchOutbox struct { type captureScopedSessions struct { *captureSessions + // scopedMu 保护本层扩展字段：presence 等异步推送 goroutine 会并发写 + // scopedAuthKeyID，测试主 goroutine 并发读（race detector 抓过这里）。 + scopedMu sync.Mutex scopedAuthKeyID [8]byte immediatePush bool } +func (s *captureScopedSessions) setScopedAuthKeyID(rawAuthKeyID [8]byte) { + s.scopedMu.Lock() + s.scopedAuthKeyID = rawAuthKeyID + s.scopedMu.Unlock() +} + +func (s *captureScopedSessions) scopedAuthKey() [8]byte { + s.scopedMu.Lock() + defer s.scopedMu.Unlock() + return s.scopedAuthKeyID +} + +func (s *captureScopedSessions) immediatePushSeen() bool { + s.scopedMu.Lock() + defer s.scopedMu.Unlock() + return s.immediatePush +} + func (s *captureScopedSessions) BindAuthKeyForSession(rawAuthKeyID [8]byte, sessionID int64, authKeyID [8]byte) { s.BindAuthKey(sessionID, authKeyID) - s.scopedAuthKeyID = rawAuthKeyID + s.setScopedAuthKeyID(rawAuthKeyID) } func (s *captureScopedSessions) AuthKeyIDForSession([8]byte, int64) ([8]byte, bool) { @@ -746,7 +768,7 @@ func (s *captureScopedSessions) AuthKeyIDForSession([8]byte, int64) ([8]byte, bo func (s *captureScopedSessions) BindUserForAuthKey(rawAuthKeyID [8]byte, sessionID, userID int64) { s.BindUser(sessionID, userID) - s.scopedAuthKeyID = rawAuthKeyID + s.setScopedAuthKeyID(rawAuthKeyID) } func (s *captureScopedSessions) UserIDForAuthKey([8]byte, int64) (int64, bool) { @@ -760,18 +782,20 @@ func (s *captureScopedSessions) UserIDResolvedForAuthKey([8]byte, int64) (int64, func (s *captureScopedSessions) SetReceivesUpdatesForAuthKey([8]byte, int64, bool) {} func (s *captureScopedSessions) PushToSessionForAuthKey(_ context.Context, rawAuthKeyID [8]byte, sessionID int64, t proto.MessageType, msg bin.Encoder) error { - s.scopedAuthKeyID = rawAuthKeyID + s.setScopedAuthKeyID(rawAuthKeyID) return s.PushToSession(context.Background(), sessionID, t, msg) } func (s *captureScopedSessions) PushToSessionForAuthKeyImmediate(_ context.Context, rawAuthKeyID [8]byte, sessionID int64, t proto.MessageType, msg bin.Encoder) error { + s.scopedMu.Lock() s.immediatePush = true s.scopedAuthKeyID = rawAuthKeyID + s.scopedMu.Unlock() return s.PushToSession(context.Background(), sessionID, t, msg) } func (s *captureScopedSessions) PushToUserExceptAuthKeySession(_ context.Context, userID int64, excludeAuthKeyID [8]byte, excludeSessionID int64, t proto.MessageType, msg bin.Encoder) (int, error) { - s.scopedAuthKeyID = excludeAuthKeyID + s.setScopedAuthKeyID(excludeAuthKeyID) return s.PushToUserExceptSession(context.Background(), userID, excludeSessionID, t, msg) } diff --git a/internal/rpc/phone_group_call_rpc_test.go b/internal/rpc/phone_group_call_rpc_test.go index e07ea8c3..ddeaadfe 100644 --- a/internal/rpc/phone_group_call_rpc_test.go +++ b/internal/rpc/phone_group_call_rpc_test.go @@ -40,12 +40,13 @@ func (s *groupCallSessions) OnlineUserIDsForCandidates(candidateUserIDs []int64, } return out } -func (s *groupCallSessions) TrackChannelInterest([8]byte, int64, int64, []int64) {} -func (s *groupCallSessions) ClearChannelInterest([8]byte, int64, int64) {} -func (s *groupCallSessions) OnlineChannelUserIDs(int64, int) []int64 { return nil } -func (s *groupCallSessions) SetSessionChannelMemberships([8]byte, int64, int64, []int64) {} -func (s *groupCallSessions) AddUserChannelMembership(int64, int64) {} -func (s *groupCallSessions) RemoveUserChannelMembership(int64, int64) {} +func (s *groupCallSessions) TrackChannelInterest([8]byte, int64, int64, []int64) {} +func (s *groupCallSessions) ClearChannelInterest([8]byte, int64, int64) {} +func (s *groupCallSessions) OnlineChannelUserIDs(int64, int) []int64 { return nil } +func (s *groupCallSessions) ChannelMembershipGeneration([8]byte, int64) int64 { return 0 } +func (s *groupCallSessions) SetSessionChannelMemberships([8]byte, int64, int64, []int64, int64) {} +func (s *groupCallSessions) AddUserChannelMembership(int64, int64) {} +func (s *groupCallSessions) RemoveUserChannelMembership(int64, int64) {} func (s *groupCallSessions) OnlineChannelMemberUserIDs(channelID int64, limit int) []int64 { return append([]int64(nil), s.online...) } @@ -148,6 +149,49 @@ func findUpdate[T tg.UpdateClass](t *testing.T, updates tg.UpdatesClass) T { return zero } +// TestGroupCallPushSkipsNonMemberOnlineCandidates 验证群通话推送收件人必须过一次 +// PG active 成员复核：在线成员索引 stale（如踢人窗口内、或索引维护缺口）时， +// 非成员绝不能收到通话状态/参与者名单（信息泄漏），真实成员照常收到。 +func TestGroupCallPushSkipsNonMemberOnlineCandidates(t *testing.T) { + f := newGroupCallFixture(t) + ownerCtx := f.userCtx(f.owner, 11) + // 模拟 stale 在线索引：outsider 不是频道成员却出现在候选里。 + f.sessions.online = []int64{f.owner.ID, f.member.ID, f.outsider.ID} + + createRes, err := f.router.onPhoneCreateGroupCall(ownerCtx, &tg.PhoneCreateGroupCallRequest{ + Peer: &tg.InputPeerChannel{ChannelID: f.channel.ID, AccessHash: f.channel.AccessHash}, + RandomID: 1, + }) + if err != nil { + t.Fatalf("createGroupCall: %v", err) + } + callUpdate := findUpdate[*tg.UpdateGroupCall](t, createRes) + call := callUpdate.Call.(*tg.GroupCall) + f.sessions.reset() + + if _, err := f.router.onPhoneJoinGroupCall(ownerCtx, &tg.PhoneJoinGroupCallRequest{ + Call: &tg.InputGroupCall{ID: call.ID, AccessHash: call.AccessHash}, + JoinAs: &tg.InputPeerSelf{}, + Params: groupCallJoinParams(t, 1001), + }); err != nil { + t.Fatalf("joinGroupCall: %v", err) + } + + pushes := f.sessions.records() + memberGot := false + for _, rec := range pushes { + if rec.userID == f.outsider.ID { + t.Fatalf("group call push leaked to non-member outsider %d: %+v", f.outsider.ID, rec) + } + if rec.userID == f.member.ID { + memberGot = true + } + } + if !memberGot { + t.Fatalf("member %d received no group call push, fan-out broken: %+v", f.member.ID, pushes) + } +} + func TestGroupCallM0Lifecycle(t *testing.T) { f := newGroupCallFixture(t) ownerCtx := f.userCtx(f.owner, 11) diff --git a/internal/rpc/phone_group_call_updates.go b/internal/rpc/phone_group_call_updates.go index db3a60ef..02144fdd 100644 --- a/internal/rpc/phone_group_call_updates.go +++ b/internal/rpc/phone_group_call_updates.go @@ -17,12 +17,24 @@ import ( // groupCallOnlineRecipients 返回在线群成员（含 originUserID 本人，推送时由 // pushUserMessage 的 ctx except 排除发起 session、保留其它设备）。 -func (r *Router) groupCallOnlineRecipients(channelID int64) []int64 { +// 在线索引候选必须过一次 PG active 复核（与 channelFanoutRecipients 同款纵深）： +// 索引 stale（如踢人窗口）时不得把通话状态/参与者名单继续推给已非成员者。 +func (r *Router) groupCallOnlineRecipients(ctx context.Context, channelID int64) []int64 { provider, ok := r.deps.Sessions.(OnlineUserProvider) if !ok { return nil } - return provider.OnlineChannelMemberUserIDs(channelID, domain.MaxChannelRealtimeFanout) + online := provider.OnlineChannelMemberUserIDs(channelID, domain.MaxChannelRealtimeFanout) + if len(online) == 0 || r.deps.Channels == nil { + return online + } + active, err := r.deps.Channels.FilterActiveMemberIDs(ctx, channelID, online) + if err != nil { + // 复核失败宁可漏推不误推：群通话 update 无 pts，客户端靠 checkGroupCall/ + // version 跳号自愈，漏推不丢一致性。 + return nil + } + return active } // groupCallUpdateContainer 把单条 update 包进 viewer 视角的 Updates 容器。 @@ -46,7 +58,7 @@ func (r *Router) pushGroupCallUpdate(ctx context.Context, channel domain.Channel r.pushConferenceGroupCallUpdate(ctx, call) return } - recipients := r.groupCallOnlineRecipients(channel.ID) + recipients := r.groupCallOnlineRecipients(ctx, channel.ID) for _, viewerID := range recipients { update := &tg.UpdateGroupCall{Call: tgGroupCall(call, viewerID, false)} update.SetPeer(&tg.PeerChannel{ChannelID: channel.ID}) @@ -69,7 +81,7 @@ func (r *Router) pushGroupCallParticipantsUpdate(ctx context.Context, channel do for _, p := range rows { userIDs = append(userIDs, p.UserID) } - recipients := r.groupCallOnlineRecipients(channel.ID) + recipients := r.groupCallOnlineRecipients(ctx, channel.ID) for _, viewerID := range recipients { update := &tg.UpdateGroupCallParticipants{ Call: &tg.InputGroupCall{ID: call.ID, AccessHash: call.AccessHash}, diff --git a/internal/rpc/router.go b/internal/rpc/router.go index c3277c11..210e7f05 100644 --- a/internal/rpc/router.go +++ b/internal/rpc/router.go @@ -219,6 +219,16 @@ func (r *Router) Dispatch(ctx context.Context, authKeyID [8]byte, sessionID int6 hasClientMetadata = true r.rememberClientSessionInfo(ctx, info) clientMetadataStored = true + // 冷启动回填：授权表恢复的 layer 即时下推到连接，防止本条 RPC handler + // 执行期间的 push 仍按 canonical 227 发给老客户端（该分支只在元数据 + // 尚未入缓存时走到，稳态 RPC 不经过这里）。 + if info.layer != 0 { + if binder, okBinder := r.deps.Sessions.(ClientLayerBinder); okBinder { + if rawAuthKeyID, okRaw := RawAuthKeyIDFrom(ctx); okRaw { + binder.SetClientLayerForAuthKey(rawAuthKeyID, sessionID, info.layer) + } + } + } } } // 前置鉴权阶段（auth key 解析 / user 重校验 / client info）慢路径告警：超阈值才记，避免刷屏。 @@ -703,6 +713,12 @@ func (r *Router) rememberClientLayer(ctx context.Context, layer int) { } sessionKey := clientInfoSessionKey{rawAuthKeyID: rawAuthKeyID, sessionID: sessionID} sessionInfo, exists := r.clientInfo[sessionKey] + // session 级记录缺失或 layer 变化时把新值即时下推到连接：invokeWithLayer 在 + // Dispatch 入口被处理（早于鉴权门与 updates 就绪置位），此时下推能保证同一请求 + // handler 执行期间触发的 pending flush / 并发 push 已按正确 layer 降级。记录已 + // 存在且相同（如客户端每条请求都带 wrapper）时跳过——连接侧已由注册播种或此前 + // 下推持有同值。 + notifyConn := !exists || sessionInfo.layer != layer sessionInfo.layer = layer if !exists { evictMapEntryIfFullLocked(r.clientInfo, maxClientInfoEntries) @@ -713,6 +729,11 @@ func (r *Router) rememberClientLayer(ctx context.Context, layer int) { r.rememberAuthClientLayerLocked(authKeyID, layer) } r.clientInfoMu.Unlock() + if notifyConn { + if binder, ok := r.deps.Sessions.(ClientLayerBinder); ok { + binder.SetClientLayerForAuthKey(rawAuthKeyID, sessionID, layer) + } + } if persistAuthLayer && r.deps.Auth != nil { if err := r.deps.Auth.UpdateAuthorizationLayer(ctx, authKeyID, layer); err != nil { r.log.Warn("update authorization layer failed", diff --git a/internal/rpc/router_layer_binder_test.go b/internal/rpc/router_layer_binder_test.go new file mode 100644 index 00000000..4f036582 --- /dev/null +++ b/internal/rpc/router_layer_binder_test.go @@ -0,0 +1,82 @@ +package rpc + +import ( + "context" + "sync" + "testing" + + "go.uber.org/zap/zaptest" + + "github.com/gotd/td/bin" + "github.com/gotd/td/clock" + "github.com/gotd/td/tg" +) + +type layerBinderCall struct { + rawAuthKeyID [8]byte + sessionID int64 + layer int +} + +// layerCaptureSessions 在 captureSessions 基础上实现可选的 ClientLayerBinder。 +type layerCaptureSessions struct { + captureSessions + layerMu sync.Mutex + layerCalls []layerBinderCall +} + +func (s *layerCaptureSessions) SetClientLayerForAuthKey(rawAuthKeyID [8]byte, sessionID int64, layer int) { + s.layerMu.Lock() + defer s.layerMu.Unlock() + s.layerCalls = append(s.layerCalls, layerBinderCall{rawAuthKeyID: rawAuthKeyID, sessionID: sessionID, layer: layer}) +} + +func (s *layerCaptureSessions) layerCallsSnapshot() []layerBinderCall { + s.layerMu.Lock() + defer s.layerMu.Unlock() + return append([]layerBinderCall(nil), s.layerCalls...) +} + +// TestDispatchInvokeWithLayerPushesLayerToSessionBinder 验证 invokeWithLayer 在 +// Dispatch 入口把新观测的 layer 即时下推到连接层（ClientLayerBinder），且仅在 +// 首次观测或 layer 变化时下推——每条请求都带 wrapper 的客户端不会造成逐 RPC 下推。 +func TestDispatchInvokeWithLayerPushesLayerToSessionBinder(t *testing.T) { + sessions := &layerCaptureSessions{} + r := New(Config{DC: 2, IP: "127.0.0.1", Port: 2398}, Deps{Sessions: sessions}, zaptest.NewLogger(t), clock.System) + rawAuthKeyID := [8]byte{1, 2, 3, 4, 5, 6, 7, 8} + const sessionID = int64(42) + + dispatchWithLayer := func(layer int) { + t.Helper() + var in bin.Buffer + req := &tg.InvokeWithLayerRequest{Layer: layer, Query: &tg.HelpGetConfigRequest{}} + if err := req.Encode(&in); err != nil { + t.Fatalf("encode: %v", err) + } + if _, err := r.Dispatch(context.Background(), rawAuthKeyID, sessionID, &in); err != nil { + t.Fatalf("dispatch layer %d: %v", layer, err) + } + } + + dispatchWithLayer(225) + calls := sessions.layerCallsSnapshot() + if len(calls) != 1 { + t.Fatalf("layer binder calls = %d, want 1", len(calls)) + } + if calls[0] != (layerBinderCall{rawAuthKeyID: rawAuthKeyID, sessionID: sessionID, layer: 225}) { + t.Fatalf("layer binder call = %+v", calls[0]) + } + + // 同 layer 重复 wrapper：不再下推。 + dispatchWithLayer(225) + if calls := sessions.layerCallsSnapshot(); len(calls) != 1 { + t.Fatalf("layer binder calls after repeat = %d, want 1", len(calls)) + } + + // layer 变化：再次下推。 + dispatchWithLayer(226) + calls = sessions.layerCallsSnapshot() + if len(calls) != 2 || calls[1].layer != 226 { + t.Fatalf("layer binder calls after change = %+v, want second call with layer 226", calls) + } +} diff --git a/internal/rpc/rpc_testkit_sessions_test.go b/internal/rpc/rpc_testkit_sessions_test.go index 948e8a34..3ee01d96 100644 --- a/internal/rpc/rpc_testkit_sessions_test.go +++ b/internal/rpc/rpc_testkit_sessions_test.go @@ -58,6 +58,13 @@ func (s *captureSessions) pushedUserIDs() []int64 { return append([]int64(nil), s.pushUserIDs...) } +// onlineChannelMemberIDs 返回当前登记的频道在线成员索引快照（测试断言用）。 +func (s *captureSessions) onlineChannelMemberIDs(channelID int64) []int64 { + s.mu.Lock() + defer s.mu.Unlock() + return append([]int64(nil), s.channelMembers[channelID]...) +} + // lastUserPush 返回最近一次 PushToUser* 的消息，独立于 message（后者也会被 // pushOnlinePeerStatusesToCurrentSession 经 PushToSession 覆盖成对端状态）。 func (s *captureSessions) lastUserPush() bin.Encoder { @@ -239,7 +246,9 @@ func (s *captureSessions) OnlineChannelUserIDs(channelID int64, limit int) []int return limitIDs(s.channelViewers[channelID], limit) } -func (s *captureSessions) SetSessionChannelMemberships(_ [8]byte, _ int64, userID int64, channelIDs []int64) { +func (s *captureSessions) ChannelMembershipGeneration(_ [8]byte, _ int64) int64 { return 0 } + +func (s *captureSessions) SetSessionChannelMemberships(_ [8]byte, _ int64, userID int64, channelIDs []int64, _ int64) { s.mu.Lock() defer s.mu.Unlock() if s.channelMembers == nil {