feat(session): 实现同平台抢登互踢功能,优化会话踢下线逻辑

新增同平台会话踢下线能力,将iOS/Android/mobile归为同一移动分组,web仅匹配web;扩展会话存储接口支持按平台、带原因踢除会话,同步更新集群广播逻辑与各调用点,统一传递踢下线原因参数
This commit is contained in:
Austin
2026-09-16 10:23:27 +08:00
parent dbb0a06db0
commit 4280b535c2
6 changed files with 124 additions and 9 deletions
+2 -2
View File
@@ -437,7 +437,7 @@ func executeKafkaUserGovernanceCommand(tenantID types.TenantID, envelope kafkaEn
}
evicted := 0
if globals.sessionStore != nil {
evicted = globals.sessionStore.EvictUserDevice(target, deviceID)
evicted = globals.sessionStore.EvictUserDevice(target, deviceID, "admin_device_kick")
}
publishRealtimeEvictDevice(tenantID, target, deviceID, "admin_device_kick")
if store.Devices != nil {
@@ -500,7 +500,7 @@ func executeKafkaUserPasswordResetCommand(tenantID types.TenantID, payload admin
func evictUserSessions(tenantID types.TenantID, uid types.Uid) {
if globals.sessionStore != nil {
globals.sessionStore.EvictUser(uid, "")
globals.sessionStore.EvictUser(uid, "", "admin_user_evict")
}
publishRealtimeEvictUser(tenantID, uid, "admin_user_evict", "")
}
+50 -2
View File
@@ -19,6 +19,7 @@ const (
realtimeTargetTopic = "topic"
realtimeTargetUser = "user"
realtimeTargetDevice = "device"
realtimeTargetPlatf = "platform"
)
func startRedisRealtimeFanout(ctx context.Context) {
@@ -105,6 +106,45 @@ func publishRealtimeEvictDevice(tenantID types.TenantID, uid types.Uid, deviceID
immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetDevice, "ok")
}
// publishRealtimeEvictUserPlatform evicts all sessions of the user on the given
// platform across cluster nodes. skipSid (the new login session) is part of the
// dedup key because repeated logins on the same platform within the dedup window
// must each reach the subscriber nodes.
func publishRealtimeEvictUserPlatform(tenantID types.TenantID, uid types.Uid, platf, reason, skipSid string) {
if !tenantID.IsValid() || uid.IsZero() || strings.TrimSpace(platf) == "" {
immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetPlatf, "skip_invalid")
return
}
msg := NoErrEvicted("", "", types.TimeNow())
msg.TenantID = tenantID
msg.AsUser = uid.UserId()
if reason != "" {
msg.Ctrl.Params = map[string]any{"reason": reason}
}
target := uid.UserId() + ":" + strings.TrimSpace(platf)
dedupKey := "evict:platform:" + target + ":" + skipSid
if err := store.PublishRealtimeFanout(currentNodeID(), tenantID, realtimeKindSessionControl, realtimeTargetPlatf, target, skipSid, dedupKey, msg); err != nil {
immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetPlatf, "error")
logs.Warn.Printf("redis realtime fanout: publish evict platform=%s failed: %v", target, err)
return
}
immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetPlatf, "ok")
}
// fanoutEvictReason extracts the machine-readable eviction reason from a
// session-control fanout message, if present.
func fanoutEvictReason(msg *ServerComMessage) string {
if msg == nil || msg.Ctrl == nil {
return ""
}
params, ok := msg.Ctrl.Params.(map[string]any)
if !ok {
return ""
}
reason, _ := params["reason"].(string)
return reason
}
func deliverRealtimeFanoutEnvelope(envelope *store.RealtimeFanoutEnvelope) {
if envelope == nil || !envelope.TenantID.IsValid() || len(envelope.Message) == 0 {
immetrics.IncRealtimeFanout("receive", realtimeKindMessage, realtimeTargetTopic, "skip_invalid")
@@ -195,7 +235,15 @@ func deliverRealtimeSessionControl(envelope *store.RealtimeFanoutEnvelope, msg *
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_invalid_target")
return
}
globals.sessionStore.EvictUserDevice(types.ParseUserId(uidText), deviceID)
globals.sessionStore.EvictUserDevice(types.ParseUserId(uidText), deviceID, fanoutEvictReason(msg))
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "evicted")
case realtimeTargetPlatf:
uidText, platf, ok := strings.Cut(envelope.Target, ":")
if !ok {
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_invalid_target")
return
}
globals.sessionStore.EvictUserPlatform(types.ParseUserId(uidText), platf, envelope.SkipSid, fanoutEvictReason(msg))
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "evicted")
case realtimeTargetUser:
uid := types.ParseUserId(envelope.Target)
@@ -203,7 +251,7 @@ func deliverRealtimeSessionControl(envelope *store.RealtimeFanoutEnvelope, msg *
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_invalid_target")
return
}
globals.sessionStore.EvictUser(uid, envelope.SkipSid)
globals.sessionStore.EvictUser(uid, envelope.SkipSid, fanoutEvictReason(msg))
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "evicted")
}
_ = msg
+5
View File
@@ -1293,6 +1293,11 @@ func (s *Session) onLogin(msgID string, timestamp time.Time, rec *auth.Rec, miss
if err := store.MarkUserOnline(s.tenantID, s.uid, s.sid, currentNodeID(), s.platf, s.userAgent); err != nil {
logs.Warn.Println("s.onLogin: failed to mark user online", s.uid, err, s.sid)
}
// 同平台抢登互踢:新登录踢掉同平台旧会话,异平台共存。
if globals.sessionStore != nil {
globals.sessionStore.EvictUserPlatform(rec.Uid, s.platf, s.sid, "replaced_by_new_login")
}
publishRealtimeEvictUserPlatform(s.tenantID, rec.Uid, s.platf, "replaced_by_new_login", s.sid)
// Reset expiration time.
rec.Lifetime = 0
}
+41 -2
View File
@@ -304,13 +304,16 @@ func (ss *SessionStore) Shutdown() {
}
// EvictUser terminates all sessions of a given user.
func (ss *SessionStore) EvictUser(uid types.Uid, skipSid string) {
func (ss *SessionStore) EvictUser(uid types.Uid, skipSid string, reason ...string) {
ss.lock.Lock()
defer ss.lock.Unlock()
// FIXME: this probably needs to be optimized. This may take very long time if the node hosts 100000 sessions.
evicted := NoErrEvicted("", "", types.TimeNow())
evicted.AsUser = uid.UserId()
if len(reason) > 0 && reason[0] != "" {
evicted.Ctrl.Params = map[string]any{"reason": reason[0]}
}
for _, s := range ss.sessCache {
if s.uid == uid && !s.isMultiplex() && s.sid != skipSid {
_, data := s.serialize(evicted)
@@ -326,7 +329,7 @@ func (ss *SessionStore) EvictUser(uid types.Uid, skipSid string) {
}
// EvictUserDevice terminates all sessions of a given user connected with a device.
func (ss *SessionStore) EvictUserDevice(uid types.Uid, deviceID string) int {
func (ss *SessionStore) EvictUserDevice(uid types.Uid, deviceID string, reason ...string) int {
ss.lock.Lock()
defer ss.lock.Unlock()
@@ -335,6 +338,9 @@ func (ss *SessionStore) EvictUserDevice(uid types.Uid, deviceID string) int {
}
evicted := NoErrEvicted("", "", types.TimeNow())
evicted.AsUser = uid.UserId()
if len(reason) > 0 && reason[0] != "" {
evicted.Ctrl.Params = map[string]any{"reason": reason[0]}
}
count := 0
for _, s := range ss.sessCache {
if s.uid == uid && s.deviceID == deviceID && !s.isMultiplex() {
@@ -352,6 +358,39 @@ func (ss *SessionStore) EvictUserDevice(uid types.Uid, deviceID string) int {
return count
}
// EvictUserPlatform terminates all sessions of a given user in the same eviction
// platform group, keeping the session identified by skipSid (e.g. the new login
// which triggers the eviction). ios/android/mobile are treated as one mobile
// group (顶号互踢); web only matches web.
func (ss *SessionStore) EvictUserPlatform(uid types.Uid, platf, skipSid string, reason ...string) int {
ss.lock.Lock()
defer ss.lock.Unlock()
if uid.IsZero() || platf == "" {
return 0
}
evicted := NoErrEvicted("", "", types.TimeNow())
evicted.AsUser = uid.UserId()
if len(reason) > 0 && reason[0] != "" {
evicted.Ctrl.Params = map[string]any{"reason": reason[0]}
}
count := 0
for _, s := range ss.sessCache {
if s.uid == uid && sameEvictPlatform(s.platf, platf) && !s.isMultiplex() && s.sid != skipSid {
_, data := s.serialize(evicted)
s.stopSession(data)
delete(ss.sessCache, s.sid)
if s.proto == LPOLL {
ss.lru.Remove(s.lpTracker)
}
count++
}
}
statsSet("LiveSessions", int64(len(ss.sessCache)))
return count
}
// NodeRestarted removes stale sessions from a restarted cluster node.
// - nodeName is the name of affected node
// - fingerprint is the new fingerprint of the node.
+5 -3
View File
@@ -824,7 +824,7 @@ func changeUserState(s *Session, uid types.Uid, user *types.User, msg *ClientCom
if state != types.StateOK {
// Terminate all sessions.
globals.sessionStore.EvictUser(uid, "")
globals.sessionStore.EvictUser(uid, "", "user_state_changed")
publishRealtimeEvictUser(s.tenantID, uid, "user_state_changed", "")
}
@@ -900,7 +900,7 @@ func replyDelUser(s *Session, msg *ClientComMessage) {
}
// Terminate all sessions. Skip the current session so the requester gets a response.
globals.sessionStore.EvictUser(uid, s.sid)
globals.sessionStore.EvictUser(uid, s.sid, "user_deleted")
publishRealtimeEvictUser(s.tenantID, uid, "user_deleted", s.sid)
// Remove user from cache and announce to cluster that the user is deleted.
usersRemoveUser(msg.TenantID, uid)
@@ -944,7 +944,9 @@ func replyDelUser(s *Session, msg *ClientComMessage) {
if s.uid == uid && s.multi == nil {
// Evict the current session if it belongs to the deleted user.
// No need to send it to multiplexing session: remote node will be notified separately.
_, data := s.serialize(NoErrEvicted("", "", msg.Timestamp))
evicted := NoErrEvicted("", "", msg.Timestamp)
evicted.Ctrl.Params = map[string]any{"reason": "user_deleted"}
_, data := s.serialize(evicted)
s.stopSession(data)
}
}
+21
View File
@@ -741,6 +741,27 @@ func platformFromUA(ua string) string {
return ""
}
// sameEvictPlatform reports whether two session platforms belong to the same
// eviction group for same-platform login eviction (顶号互踢): identical
// identifiers match, and ios/android/mobile are all treated as one mobile group
// (an iOS login evicts Android sessions and vice versa). Web only matches web.
func sameEvictPlatform(a, b string) bool {
if a == "" || b == "" {
return false
}
if strings.EqualFold(a, b) {
return true
}
switch strings.ToLower(a) {
case "ios", "android", "mobile":
switch strings.ToLower(b) {
case "ios", "android", "mobile":
return true
}
}
return false
}
func parseTLSConfig(tlsEnabled bool, jsconfig json.RawMessage) (*tls.Config, error) {
type tlsAutocertConfig struct {
// Domains to support by autocert