fix(realtime): evict fanout 增加时效校验,防止延迟重放误踢新登录
deliverRealtimeSessionControl 分发前校验 Ctrl.Timestamp,超过 10s 的 session-control(evict)消息直接丢弃并记 skip_stale 指标,避免 consumer 重启重放 PEL / stream 积压时旧消息的 skipSid 与新会话不符导致新登录被误踢。无时间戳消息放行以兼容旧发布方。 验证:TestRealtime 3/3、regression_kick_notification 14/14、regression_mobile_kick 21/21。
This commit is contained in:
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/json"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/tinode/chat/server/logs"
|
||||
immetrics "github.com/tinode/chat/server/metrics"
|
||||
@@ -223,11 +224,33 @@ func deliverRealtimeTopicMessage(envelope *store.RealtimeFanoutEnvelope, msg *Se
|
||||
recordRealtimeFanoutDeliverResult(envelope.Kind, realtimeTargetTopic, delivered)
|
||||
}
|
||||
|
||||
// fanoutSessionControlStaleLimit caps how long after publication a
|
||||
// session-control (evict) fanout message remains actionable. Delayed replays
|
||||
// (consumer backlog, pending-entry redelivery after restart) carry the original
|
||||
// skipSid, so a replayed evict would otherwise terminate freshly established
|
||||
// sessions whose sid differs from the stale skipSid (新登录被旧消息误踢).
|
||||
const fanoutSessionControlStaleLimit = 10 * time.Second
|
||||
|
||||
// fanoutSessionControlStale reports whether the session-control fanout message
|
||||
// is too old to act on. Messages without a timestamp (unknown origin) are
|
||||
// delivered to stay compatible with legacy publishers.
|
||||
func fanoutSessionControlStale(msg *ServerComMessage) bool {
|
||||
if msg == nil || msg.Ctrl == nil || msg.Ctrl.Timestamp.IsZero() {
|
||||
return false
|
||||
}
|
||||
return time.Since(msg.Ctrl.Timestamp) > fanoutSessionControlStaleLimit
|
||||
}
|
||||
|
||||
func deliverRealtimeSessionControl(envelope *store.RealtimeFanoutEnvelope, msg *ServerComMessage) {
|
||||
if globals.sessionStore == nil {
|
||||
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_no_session_store")
|
||||
return
|
||||
}
|
||||
if fanoutSessionControlStale(msg) {
|
||||
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_stale")
|
||||
logs.Warn.Printf("redis realtime fanout: dropped stale session-control message target=%s older=%s", envelope.Target, fanoutSessionControlStaleLimit)
|
||||
return
|
||||
}
|
||||
switch envelope.TargetType {
|
||||
case realtimeTargetDevice:
|
||||
uidText, deviceID, ok := strings.Cut(envelope.Target, ":")
|
||||
|
||||
@@ -103,6 +103,52 @@ func TestRealtimeP2PTopicFallbackRewritesVisibleTopic(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRealtimeSessionControlStaleEvictDropped(t *testing.T) {
|
||||
oldStore := globals.sessionStore
|
||||
defer func() { globals.sessionStore = oldStore }()
|
||||
|
||||
tenantID := types.TenantID(1)
|
||||
uid := types.Uid(1001)
|
||||
globals.sessionStore = realtimeTestSessionStore(
|
||||
realtimeTestPlatformSession("sid-live", tenantID, uid, "web"),
|
||||
)
|
||||
|
||||
envelope := func(ts time.Time) *store.RealtimeFanoutEnvelope {
|
||||
msg := &ServerComMessage{
|
||||
Ctrl: &MsgServerCtrl{
|
||||
Code: 205,
|
||||
Text: "evicted",
|
||||
Timestamp: ts,
|
||||
Params: map[string]any{"reason": "replaced_by_new_login"},
|
||||
},
|
||||
}
|
||||
raw, err := json.Marshal(msg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return &store.RealtimeFanoutEnvelope{
|
||||
TenantID: tenantID,
|
||||
Kind: realtimeKindSessionControl,
|
||||
TargetType: realtimeTargetPlatf,
|
||||
Target: uid.UserId() + ":web",
|
||||
SkipSid: "sid-new",
|
||||
Message: raw,
|
||||
}
|
||||
}
|
||||
|
||||
// A stale replay (old skipSid != current session sid) must not evict.
|
||||
deliverRealtimeFanoutEnvelope(envelope(time.Now().Add(-2 * fanoutSessionControlStaleLimit)))
|
||||
if globals.sessionStore.Get("sid-live") == nil {
|
||||
t.Fatal("stale evict must not terminate the session")
|
||||
}
|
||||
|
||||
// A fresh evict must evict as usual.
|
||||
deliverRealtimeFanoutEnvelope(envelope(time.Now()))
|
||||
if globals.sessionStore.Get("sid-live") != nil {
|
||||
t.Fatal("fresh evict must terminate the session")
|
||||
}
|
||||
}
|
||||
|
||||
func realtimeTestSessionStore(sessions ...*Session) *SessionStore {
|
||||
ss := NewSessionStore(time.Minute)
|
||||
for _, sess := range sessions {
|
||||
@@ -117,6 +163,7 @@ func realtimeTestSession(sid string, tenantID types.TenantID, uid types.Uid, sub
|
||||
tenantID: tenantID,
|
||||
uid: uid,
|
||||
send: make(chan any, 4),
|
||||
stop: make(chan any, 1),
|
||||
subs: make(map[string]*Subscription),
|
||||
}
|
||||
if sub != "" {
|
||||
@@ -124,3 +171,9 @@ func realtimeTestSession(sid string, tenantID types.TenantID, uid types.Uid, sub
|
||||
}
|
||||
return sess
|
||||
}
|
||||
|
||||
func realtimeTestPlatformSession(sid string, tenantID types.TenantID, uid types.Uid, platf string) *Session {
|
||||
sess := realtimeTestSession(sid, tenantID, uid, "")
|
||||
sess.platf = platf
|
||||
return sess
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user