feat: wake backgrounded accounts via MTProto internal push connection
This commit is contained in:
onysd
parent
ddd26e51b5
commit
32d128e496
4 changed files with 138 additions and 4 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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 无意义。
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue