diff --git a/server/kafka_admin_commands.go b/server/kafka_admin_commands.go index 653f832..0c2ee95 100644 --- a/server/kafka_admin_commands.go +++ b/server/kafka_admin_commands.go @@ -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", "") } diff --git a/server/realtime_dispatcher.go b/server/realtime_dispatcher.go index abfccde..2668c88 100644 --- a/server/realtime_dispatcher.go +++ b/server/realtime_dispatcher.go @@ -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 diff --git a/server/session.go b/server/session.go index e60c023..983d6ee 100644 --- a/server/session.go +++ b/server/session.go @@ -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 } diff --git a/server/sessionstore.go b/server/sessionstore.go index 20c00de..99edcd0 100644 --- a/server/sessionstore.go +++ b/server/sessionstore.go @@ -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. diff --git a/server/user.go b/server/user.go index 06528fd..c25857b 100644 --- a/server/user.go +++ b/server/user.go @@ -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) } } diff --git a/server/utils.go b/server/utils.go index ab81e8f..29c07b8 100644 --- a/server/utils.go +++ b/server/utils.go @@ -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