diff --git a/internal/mtprotoedge/session_manager.go b/internal/mtprotoedge/session_manager.go index 7ff5a463..6179b404 100644 --- a/internal/mtprotoedge/session_manager.go +++ b/internal/mtprotoedge/session_manager.go @@ -176,6 +176,15 @@ type SessionManager struct { pending map[sessionKey][]queuedPush // updates-ready 前暂存的主动推送 flushing map[sessionKey]bool // 置位时暂存正在排空的 session;排空完成前推送继续进 pending 保序 pendingBudget *outboundTrackedBudget // 未就绪 session 暂存 encoded body 的进程级上限 + // pushSessions 记录经 account.registerDevice(token_type=7) 登记的「MTProto 内部 + // 推送通道」session:raw auth_key_id → 该 auth_key 下已登记的 session_id 集合。 + // 这类连接只发 ping,永远不会调 updates.getState(receivesUpdates 恒 false), + // 但它是账号切到后台/未选中(主连接被客户端 setAppPaused 挂起)时唯一还连着 + // 服务器的连接——官方 Telegram 客户端正是靠它,在无 FCM/APNs 的场景(大陆、 + // 去 Google 化设备)下仍能收到来电、消息等实时推送。登记后在 pushToUserWithSender + // 中被视为【永久就绪】,绕过 receivesUpdates 门槛直接投递,而不是排队等一个永远 + // 不会到来的 getState。See memory: call-inactive-account-network-pause。 + pushSessions map[[8]byte]map[int64]struct{} lifecycle SessionLifecycleObserver log *zap.Logger @@ -200,10 +209,50 @@ func NewSessionManager(log *zap.Logger) *SessionManager { pending: make(map[sessionKey][]queuedPush), flushing: make(map[sessionKey]bool), pendingBudget: newOutboundTrackedBudget(defaultPendingPushMaxBytes), + pushSessions: make(map[[8]byte]map[int64]struct{}), log: log, } } +// MarkPushSession 把 (rawAuthKeyID, sessionID) 登记为该 auth_key 的 MTProto 内部推送 +// 通道,使其在 pushToUserWithSender 中跳过 receivesUpdates 门槛、始终被视为可投递。 +// 由 account.registerDevice(token_type=7) 处理器调用;幂等,与是否已有活跃 Conn 无关 +// (连接可能晚于此调用才建立,或断线重连复用同一 session_id)。 +func (m *SessionManager) MarkPushSession(rawAuthKeyID [8]byte, sessionID int64) { + m.mu.Lock() + defer m.mu.Unlock() + set, ok := m.pushSessions[rawAuthKeyID] + if !ok { + set = make(map[int64]struct{}) + m.pushSessions[rawAuthKeyID] = set + } + set[sessionID] = struct{}{} +} + +// UnmarkPushSession 撤销登记(account.unregisterDevice(token_type=7))。 +func (m *SessionManager) UnmarkPushSession(rawAuthKeyID [8]byte, sessionID int64) { + m.mu.Lock() + defer m.mu.Unlock() + set, ok := m.pushSessions[rawAuthKeyID] + if !ok { + return + } + delete(set, sessionID) + if len(set) == 0 { + delete(m.pushSessions, rawAuthKeyID) + } +} + +// isPushSessionLocked 报告 key 是否登记为推送通道。调用方须已持有 m.mu(读锁或写锁均可)。 +func (m *SessionManager) isPushSessionLocked(key sessionKey) bool { + set, ok := m.pushSessions[key.authKeyID] + if !ok { + return false + } + _, ok = set[key.sessionID] + return ok +} + // SetLifecycleObserver installs a best-effort active session lifecycle observer. func (m *SessionManager) SetLifecycleObserver(observer SessionLifecycleObserver) { m.mu.Lock() @@ -1636,7 +1685,7 @@ func (m *SessionManager) pushToUserWithSender(ctx context.Context, userID int64, excluded := 0 skipped := 0 needQueue := false - for _, c := range m.byUser[userID] { + for key, c := range m.byUser[userID] { if shouldExcludeSession(c, excludeAuthKeyID, excludeSessionID) { excluded++ continue @@ -1645,7 +1694,10 @@ func (m *SessionManager) pushToUserWithSender(ctx context.Context, userID int64, skipped++ continue } - if !c.receivesUpdates.Load() { + // 已登记的推送通道(registerDevice token_type=7)永久视为就绪:它只发 ping, + // receivesUpdates 恒 false,但绕过门槛直接投递才是它存在的意义——排队等一个 + // 永远不会到来的 getState 毫无价值。 + if !c.receivesUpdates.Load() && !m.isPushSessionLocked(key) { if !queueWhenNotReady { // transient(typing/presence):未就绪即丢,不进 pending。这些 update 不写 // durable log,getDifference 无法补;就绪后由 getState 快照/下次状态变化重建。 @@ -1677,7 +1729,7 @@ func (m *SessionManager) pushToUserWithSender(ctx context.Context, userID int64, skipped++ continue } - if !c.receivesUpdates.Load() { + if !c.receivesUpdates.Load() && !m.isPushSessionLocked(key) { if !queueWhenNotReady { skipped++ continue diff --git a/internal/rpc/account.go b/internal/rpc/account.go index 9f729fa9..7fb8b366 100644 --- a/internal/rpc/account.go +++ b/internal/rpc/account.go @@ -2,10 +2,13 @@ package rpc import ( "context" + "encoding/hex" "errors" + "strconv" "strings" "github.com/iamxvbaba/td/tg" + "go.uber.org/zap" "github.com/iamxvbaba/td/tlprofile" "telesrv/internal/branding" @@ -14,6 +17,12 @@ import ( "telesrv/internal/domain" ) +// mtprotoPushTokenType 是 account.registerDevice/unregisterDevice 的 token_type=7: +// 客户端在没有 FCM/APNs 的场景下(自建 owpg:// 服务器、大陆/去 Google 化设备)自己 +// 建立的「MTProto 内部推送」通道——一条独立、常驻、只发 ping 的连接,用自己的 +// session_id 作为 token 上报。见 PushSessionRegistrar。 +const mtprotoPushTokenType = 7 + // registerAccount 注册 account.* RPC handler。 func (r *Router) registerAccount(d *tlprofile.Dispatcher) { registerRPC[*tg.AccountDeleteAccountRequest](d, tlprofile.SemanticMethodAccountDeleteAccount, func(ctx context.Context, req *tg.AccountDeleteAccountRequest) (any, error) { @@ -26,9 +35,11 @@ func (r *Router) registerAccount(d *tlprofile.Dispatcher) { return r.onAccountConfirmPhone(ctx, req) }) registerRPC[*tg.AccountRegisterDeviceRequest](d, tlprofile.SemanticMethodAccountRegisterDevice, func(ctx context.Context, req *tg.AccountRegisterDeviceRequest) (any, error) { + r.registerPushSession(ctx, req.TokenType, req.Token) return true, nil }) registerRPC[*tg.AccountUnregisterDeviceRequest](d, tlprofile.SemanticMethodAccountUnregisterDevice, func(ctx context.Context, req *tg.AccountUnregisterDeviceRequest) (any, error) { + r.unregisterPushSession(ctx, req.TokenType, req.Token) return true, nil }) registerRPC[*tg.AccountUpdateDeviceLockedRequest](d, tlprofile.SemanticMethodAccountUpdateDeviceLocked, func(ctx context.Context, layerRequest *tg.AccountUpdateDeviceLockedRequest) (any, error) { @@ -460,6 +471,65 @@ func (r *Router) registerAccount(d *tlprofile.Dispatcher) { } +// registerPushSession 把 token_type=7 的 registerDevice 登记为发起本次 RPC 的连接 +// (raw auth_key_id + session_id)的推送通道身份,见 mtprotoPushTokenType 与 +// PushSessionRegistrar。其它 token_type(FCM/APNs/…)本服务器不发外部推送,忽略。 +func (r *Router) registerPushSession(ctx context.Context, tokenType int, token string) { + if tokenType != mtprotoPushTokenType { + return + } + registrar, ok := r.deps.Sessions.(PushSessionRegistrar) + if !ok { + return + } + rawAuthKeyID, ok := RawAuthKeyIDFrom(ctx) + if !ok { + return + } + sessionID, ok := SessionIDFrom(ctx) + if !ok { + return + } + // token 上报的是客户端自己的 pushSessionId(ConnectionsManager.cpp 的 + // to_string_uint64((uint64_t) pushSessionId)),理论上应与本次 RPC 所在的 + // session_id 一致(registerDevice 走主连接发起,携带 push 连接自己的 session_id + // 作为 token)。以 token 解析出的 session_id 为准登记——若客户端将来在别的连接上 + // 上报,仍能正确登记到「那条」连接,而不是发起 RPC 的这条。 + if parsed, err := strconv.ParseInt(token, 10, 64); err == nil && parsed != 0 { + sessionID = parsed + } + registrar.MarkPushSession(rawAuthKeyID, sessionID) + if r.log != nil { + r.log.Info("registered MTProto push session", + zap.String("auth_key_id", hex.EncodeToString(rawAuthKeyID[:])), + zap.Int64("push_session_id", sessionID), + ) + } +} + +// unregisterPushSession 撤销 registerPushSession 的登记。 +func (r *Router) unregisterPushSession(ctx context.Context, tokenType int, token string) { + if tokenType != mtprotoPushTokenType { + return + } + registrar, ok := r.deps.Sessions.(PushSessionRegistrar) + if !ok { + return + } + rawAuthKeyID, ok := RawAuthKeyIDFrom(ctx) + if !ok { + return + } + sessionID, ok := SessionIDFrom(ctx) + if !ok { + return + } + if parsed, err := strconv.ParseInt(token, 10, 64); err == nil && parsed != 0 { + sessionID = parsed + } + registrar.UnmarkPushSession(rawAuthKeyID, sessionID) +} + func (r *Router) onAccountGetPassword(ctx context.Context) (*tg.AccountPassword, error) { if r.deps.Account == nil { return tgPassword(domain.PasswordSettings{SecureRandom: []byte("telesrv-tdesktop-dev-secure-rand")}), nil diff --git a/internal/rpc/deps.go b/internal/rpc/deps.go index 2341b2bb..09fcdfa9 100644 --- a/internal/rpc/deps.go +++ b/internal/rpc/deps.go @@ -165,6 +165,18 @@ type BestEffortSessionBinder interface { PushToUserExceptAuthKeySessionBestEffort(ctx context.Context, userID int64, excludeAuthKeyID [8]byte, excludeSessionID int64, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) } +// PushSessionRegistrar 登记/撤销 account.registerDevice(token_type=7,MTProto 内部 +// 推送通道) 的 session。登记后该 session 在推送 fan-out 中被视为永久就绪,绕过 +// receivesUpdates 门槛——它只发 ping、永远不会调 updates.getState,但在账号后台/ +// 未选中(客户端把主连接 setAppPaused 挂起)时是唯一还连着服务器的连接,是官方 +// Telegram 在无 FCM/APNs 场景(大陆、去 Google 化设备)下仍能收到来电、消息等实时 +// 推送的机制。SessionManager 实现;未装配时 registerDevice 静默跳过登记(只影响 +// 后台唤醒,不影响功能正确性)。See memory: call-inactive-account-network-pause。 +type PushSessionRegistrar interface { + MarkPushSession(rawAuthKeyID [8]byte, sessionID int64) + UnmarkPushSession(rawAuthKeyID [8]byte, sessionID int64) +} + // TransientSessionBinder 推送短命、不写 durable log 的 update(typing / presence)。 // 与普通推送的关键区别:目标 session 未就绪时直接跳过、不进 pending——transient 数据 // getDifference 无法补,就绪后由 getState 快照/下次状态变化重建,囤积过期 transient 无意义。 diff --git a/internal/rpc/sticker_link_smoke_test.go b/internal/rpc/sticker_link_smoke_test.go index d0386839..b6ed4cfd 100644 --- a/internal/rpc/sticker_link_smoke_test.go +++ b/internal/rpc/sticker_link_smoke_test.go @@ -89,7 +89,7 @@ func TestCustomStickerPackLinkInstallAndSendSmoke(t *testing.T) { t.Fatalf("sticker link status = %d body=%q, want 200", rr.Code, rr.Body.String()) } body := rr.Body.String() - for _, want := range []string{"https://telesrv.net/addstickers/alice_fresh_pack", "telesrv://addstickers?set=alice_fresh_pack"} { + for _, want := range []string{"https://telesrv.net/addstickers/alice_fresh_pack", "telesrv://telesrv.net/addstickers/alice_fresh_pack"} { if !strings.Contains(body, want) { t.Fatalf("sticker link body missing %q:\n%s", want, body) }