i tired of fixing calls
This commit is contained in:
onysd
parent
46afb97396
commit
157b82b5f5
7 changed files with 46 additions and 148 deletions
|
|
@ -1,52 +0,0 @@
|
|||
package mtprotoedge
|
||||
|
||||
import "testing"
|
||||
|
||||
// TestShouldExcludeDeviceMatchesAllSessionsOfDevice guards the phone-call
|
||||
// "stop ringing" fix: the accepting device (identified by its perm/business
|
||||
// auth key) must be excluded across ALL its connections, not just the one
|
||||
// session that carried the accept — otherwise the stop-ringing phoneCallDiscarded
|
||||
// leaks onto the device's other connections and kills the call it just accepted
|
||||
// (the "B answers → instantly Failed to connect" asymmetry).
|
||||
func TestShouldExcludeDeviceMatchesAllSessionsOfDevice(t *testing.T) {
|
||||
device := [8]byte{1, 2, 3, 4, 5, 6, 7, 8}
|
||||
other := [8]byte{9, 9, 9, 9, 9, 9, 9, 9}
|
||||
|
||||
// Two connections of the SAME device but different raw keys / sessions —
|
||||
// exactly the OwpenGram multi-connection (dc 1..5 → one server) shape.
|
||||
connA := &Conn{authKeyID: [8]byte{0xA}, sessionID: 111}
|
||||
connA.SetBusinessAuthKeyID(device)
|
||||
connB := &Conn{authKeyID: [8]byte{0xB}, sessionID: 222}
|
||||
connB.SetBusinessAuthKeyID(device)
|
||||
// A connection of a DIFFERENT device (the real "other device" that should
|
||||
// still receive the stop-ringing).
|
||||
connOther := &Conn{authKeyID: [8]byte{0xC}, sessionID: 333}
|
||||
connOther.SetBusinessAuthKeyID(other)
|
||||
|
||||
if !shouldExcludeDevice(connA, device) {
|
||||
t.Fatal("accepting device's connection A must be excluded")
|
||||
}
|
||||
if !shouldExcludeDevice(connB, device) {
|
||||
t.Fatal("accepting device's connection B (other session) must ALSO be excluded")
|
||||
}
|
||||
if shouldExcludeDevice(connOther, device) {
|
||||
t.Fatal("a different device must NOT be excluded")
|
||||
}
|
||||
}
|
||||
|
||||
func TestShouldExcludeDeviceZeroKeyExcludesNothing(t *testing.T) {
|
||||
c := &Conn{authKeyID: [8]byte{0xA}, sessionID: 111}
|
||||
c.SetBusinessAuthKeyID([8]byte{1, 2, 3})
|
||||
if shouldExcludeDevice(c, [8]byte{}) {
|
||||
t.Fatal("zero business auth key must exclude nothing")
|
||||
}
|
||||
}
|
||||
|
||||
func TestShouldExcludeDeviceUnresolvedBusinessKeyNotExcluded(t *testing.T) {
|
||||
// A connection whose business auth key isn't resolved yet must not be
|
||||
// matched (can't prove it's the accepting device).
|
||||
c := &Conn{authKeyID: [8]byte{0xA}, sessionID: 111}
|
||||
if shouldExcludeDevice(c, [8]byte{1, 2, 3, 4, 5, 6, 7, 8}) {
|
||||
t.Fatal("connection with unresolved business auth key must not be excluded")
|
||||
}
|
||||
}
|
||||
|
|
@ -201,7 +201,7 @@ func (m *SessionManager) PushToUserExceptSession(ctx context.Context, userID, ex
|
|||
}
|
||||
|
||||
func (m *SessionManager) PushToUserExceptSessionBestEffort(ctx context.Context, userID, excludeSessionID int64, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
|
||||
return m.pushToUserBestEffort(ctx, userID, nil, excludeSessionID, [8]byte{}, t, msg, timeout)
|
||||
return m.pushToUserBestEffort(ctx, userID, nil, excludeSessionID, t, msg, timeout)
|
||||
}
|
||||
|
||||
func (m *SessionManager) Online() int {
|
||||
|
|
|
|||
|
|
@ -1476,7 +1476,7 @@ func (m *SessionManager) pushToBusinessAuthKey(ctx context.Context, userID int64
|
|||
|
||||
func (m *SessionManager) pushToUser(ctx context.Context, userID int64, excludeAuthKeyID *[8]byte, excludeSessionID int64, t proto.MessageType, msg tg.UpdatesClass) (int, error) {
|
||||
getUpdates := onceLayerUpdatesFanout(ctx, msg)
|
||||
return m.pushToUserWithSender(ctx, userID, excludeAuthKeyID, excludeSessionID, [8]byte{}, t, getUpdates, true, func(c *Conn) error {
|
||||
return m.pushToUserWithSender(ctx, userID, excludeAuthKeyID, excludeSessionID, t, getUpdates, true, func(c *Conn) error {
|
||||
if c.outbound == nil || c.outboundControl == nil {
|
||||
return ErrConnClosed
|
||||
}
|
||||
|
|
@ -1499,7 +1499,7 @@ func (m *SessionManager) pushToUser(ctx context.Context, userID int64, excludeAu
|
|||
// 「durable 兜底」丢弃。走 best-effort 发送，不阻塞调用方。
|
||||
func (m *SessionManager) PushToUserTransientExceptAuthKeySession(ctx context.Context, userID int64, excludeAuthKeyID [8]byte, excludeSessionID int64, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
|
||||
getUpdates := onceLayerUpdatesFanout(ctx, msg)
|
||||
return m.pushToUserWithSender(ctx, userID, &excludeAuthKeyID, excludeSessionID, [8]byte{}, t, getUpdates, false, func(c *Conn) error {
|
||||
return m.pushToUserWithSender(ctx, userID, &excludeAuthKeyID, excludeSessionID, t, getUpdates, false, func(c *Conn) error {
|
||||
if c.outbound == nil || c.outboundControl == nil {
|
||||
return ErrConnClosed
|
||||
}
|
||||
|
|
@ -1516,38 +1516,10 @@ func (m *SessionManager) PushToUserTransientExceptAuthKeySession(ctx context.Con
|
|||
}
|
||||
|
||||
func (m *SessionManager) PushToUserExceptAuthKeySessionBestEffort(ctx context.Context, userID int64, excludeAuthKeyID [8]byte, excludeSessionID int64, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
|
||||
return m.pushToUserBestEffort(ctx, userID, &excludeAuthKeyID, excludeSessionID, [8]byte{}, t, msg, timeout)
|
||||
return m.pushToUserBestEffort(ctx, userID, &excludeAuthKeyID, excludeSessionID, t, msg, timeout)
|
||||
}
|
||||
|
||||
// PushToUserExceptBusinessAuthKey fans msg out to every ready connection of
|
||||
// userID EXCEPT those belonging to the device identified by
|
||||
// excludeBusinessAuthKeyID (perm/business auth key) — i.e. it excludes the
|
||||
// whole accepting DEVICE, all of its connections/sessions, not just the one
|
||||
// session that carried the request. Used for phone-call "stop ringing": see
|
||||
// shouldExcludeDevice. Falls back to durable (non-best-effort) fan-out when no
|
||||
// outbound push timeout is configured.
|
||||
func (m *SessionManager) PushToUserExceptBusinessAuthKey(ctx context.Context, userID int64, excludeBusinessAuthKeyID [8]byte, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
|
||||
if timeout > 0 {
|
||||
return m.pushToUserBestEffort(ctx, userID, nil, 0, excludeBusinessAuthKeyID, t, msg, timeout)
|
||||
}
|
||||
getUpdates := onceLayerUpdatesFanout(ctx, msg)
|
||||
return m.pushToUserWithSender(ctx, userID, nil, 0, excludeBusinessAuthKeyID, t, getUpdates, true, func(c *Conn) error {
|
||||
if c.outbound == nil || c.outboundControl == nil {
|
||||
return ErrConnClosed
|
||||
}
|
||||
updates, err := getUpdates()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
encoded, err := updates.prepareForConn(ctx, c)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return c.SendEncoded(ctx, t, encoded)
|
||||
})
|
||||
}
|
||||
|
||||
func (m *SessionManager) pushToUserBestEffort(ctx context.Context, userID int64, excludeAuthKeyID *[8]byte, excludeSessionID int64, excludeBusinessAuthKeyID [8]byte, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
|
||||
func (m *SessionManager) pushToUserBestEffort(ctx context.Context, userID int64, excludeAuthKeyID *[8]byte, excludeSessionID int64, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
|
||||
if ctx != nil && ctx.Err() != nil {
|
||||
return 0, ctx.Err()
|
||||
}
|
||||
|
|
@ -1573,7 +1545,7 @@ func (m *SessionManager) pushToUserBestEffort(ctx context.Context, userID int64,
|
|||
defer cancel()
|
||||
}
|
||||
getUpdates := onceLayerUpdatesFanout(sendCtx, msg)
|
||||
return m.pushToUserWithSender(ctx, userID, excludeAuthKeyID, excludeSessionID, excludeBusinessAuthKeyID, t, getUpdates, true, func(c *Conn) error {
|
||||
return m.pushToUserWithSender(ctx, userID, excludeAuthKeyID, excludeSessionID, t, getUpdates, true, func(c *Conn) error {
|
||||
if c.outbound == nil || c.outboundControl == nil {
|
||||
return ErrConnClosed
|
||||
}
|
||||
|
|
@ -1620,7 +1592,7 @@ func onceLayerUpdatesFanout(ctx context.Context, msg tg.UpdatesClass) func() (*l
|
|||
}
|
||||
}
|
||||
|
||||
func (m *SessionManager) pushToUserWithSender(ctx context.Context, userID int64, excludeAuthKeyID *[8]byte, excludeSessionID int64, excludeBusinessAuthKeyID [8]byte, t proto.MessageType, getUpdates func() (*layerUpdatesFanout, error), queueWhenNotReady bool, send func(*Conn) error) (int, error) {
|
||||
func (m *SessionManager) pushToUserWithSender(ctx context.Context, userID int64, excludeAuthKeyID *[8]byte, excludeSessionID int64, t proto.MessageType, getUpdates func() (*layerUpdatesFanout, error), queueWhenNotReady bool, send func(*Conn) error) (int, error) {
|
||||
// push fan-out 是连接层最热路径之一：debug 日志的字段构造（含 auth_key hex 格式化）
|
||||
// 在关闭 debug 时也会求值，先查级别一次、按需记日志。
|
||||
debug := m.log.Core().Enabled(zapcore.DebugLevel)
|
||||
|
|
@ -1636,7 +1608,7 @@ func (m *SessionManager) pushToUserWithSender(ctx context.Context, userID int64,
|
|||
skipped := 0
|
||||
needQueue := false
|
||||
for _, c := range m.byUser[userID] {
|
||||
if shouldExcludeSession(c, excludeAuthKeyID, excludeSessionID) || shouldExcludeDevice(c, excludeBusinessAuthKeyID) {
|
||||
if shouldExcludeSession(c, excludeAuthKeyID, excludeSessionID) {
|
||||
excluded++
|
||||
continue
|
||||
}
|
||||
|
|
@ -2506,22 +2478,6 @@ func shouldExcludeSession(c *Conn, excludeAuthKeyID *[8]byte, excludeSessionID i
|
|||
return c.authKeyID == *excludeAuthKeyID
|
||||
}
|
||||
|
||||
// shouldExcludeDevice reports whether c belongs to the device identified by
|
||||
// excludeBusinessAuthKeyID (the perm/business auth key). Unlike the per-session
|
||||
// exclusion above, this matches EVERY connection of that device regardless of
|
||||
// session_id or raw temp-key. Required for signals like phone-call "stop
|
||||
// ringing": a device that aliases dc 1..5 onto one server (the OwpenGram
|
||||
// client) holds several connections, so excluding only the one session that
|
||||
// carried the accept would let the stop/discard leak onto the device's other
|
||||
// connections and kill the call it just accepted.
|
||||
func shouldExcludeDevice(c *Conn, excludeBusinessAuthKeyID [8]byte) bool {
|
||||
if excludeBusinessAuthKeyID == ([8]byte{}) {
|
||||
return false
|
||||
}
|
||||
id, resolved := c.BusinessAuthKeyID()
|
||||
return resolved && id == excludeBusinessAuthKeyID
|
||||
}
|
||||
|
||||
func sessionKeyLog(id [8]byte) string {
|
||||
return fmt.Sprintf("%x", id)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -165,14 +165,6 @@ 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)
|
||||
}
|
||||
|
||||
// DeviceExcludingSessionPusher 按【设备】（perm/business auth_key）整体排除后 fan-out。
|
||||
// 与按单个 (raw auth_key, session) 排除的区别：一台设备可能有多条连接（OwpenGram
|
||||
// 客户端把 dc 1..5 都指向同一服务器 → 多连接），只排除受理那一条会让 phoneCall
|
||||
// "stop ringing" 的 discarded 漏到该设备的其它连接上、误杀它刚接起的通话。
|
||||
type DeviceExcludingSessionPusher interface {
|
||||
PushToUserExceptBusinessAuthKey(ctx context.Context, userID int64, excludeBusinessAuthKeyID [8]byte, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error)
|
||||
}
|
||||
|
||||
// TransientSessionBinder 推送短命、不写 durable log 的 update（typing / presence）。
|
||||
// 与普通推送的关键区别：目标 session 未就绪时直接跳过、不进 pending——transient 数据
|
||||
// getDifference 无法补，就绪后由 getState 快照/下次状态变化重建，囤积过期 transient 无意义。
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import (
|
|||
"errors"
|
||||
|
||||
"github.com/iamxvbaba/td/tg"
|
||||
"go.uber.org/zap"
|
||||
|
||||
appphone "telesrv/internal/app/phone"
|
||||
"telesrv/internal/domain"
|
||||
|
|
@ -221,6 +222,15 @@ func (r *Router) onPhoneDiscardCall(ctx context.Context, req *tg.PhoneDiscardCal
|
|||
return nil, groupCallInvalidErr()
|
||||
}
|
||||
}
|
||||
// 诊断：记录被叫/主叫挂断的 reason 与时机（相对 confirm）。用于定位「接听即断」类问题——
|
||||
// hangup 通常是用户主动挂断或客户端 tgcalls 失败；disconnect/missed 是媒体/振铃超时。
|
||||
if r.log != nil {
|
||||
r.log.Info("phone discardCall",
|
||||
zap.Int64("user_id", userID),
|
||||
zap.Int64("call_id", req.Peer.ID),
|
||||
zap.String("reason", string(reason)),
|
||||
)
|
||||
}
|
||||
call, already, err := r.deps.Phone.DiscardCallWithSlug(ctx, userID, req.Peer.ID, req.Peer.AccessHash, reason, reasonSlug, req.Duration)
|
||||
if err != nil {
|
||||
return nil, phoneCallErr(err)
|
||||
|
|
|
|||
|
|
@ -5,7 +5,6 @@ import (
|
|||
|
||||
"github.com/iamxvbaba/td/proto"
|
||||
"github.com/iamxvbaba/td/tg"
|
||||
"go.uber.org/zap"
|
||||
|
||||
"telesrv/internal/domain"
|
||||
)
|
||||
|
|
@ -35,30 +34,22 @@ func (r *Router) pushPhoneCall(ctx context.Context, targetUserID int64, call dom
|
|||
return r.pushUserMessage(ctx, targetUserID, logMessage, r.phoneCallUpdates(ctx, call, targetUserID))
|
||||
}
|
||||
|
||||
// pushPhoneCallStopRinging 向被叫【其它设备】推合成 phoneCallDiscarded 停振铃（P0-1 修正）。
|
||||
// ctx 必须是接听设备的请求上下文。
|
||||
// pushPhoneCallStopRinging 曾向被叫【其它设备】推合成 phoneCallDiscarded 停振铃。
|
||||
// 现已停用（no-op），原因见下。
|
||||
//
|
||||
// ⚠ 排除必须按【设备】（perm/business auth_key）整体做，而不是按单个
|
||||
// (raw auth_key, session)。接听设备可能有多条到本服务器的连接——OwpenGram 客户端
|
||||
// 把 dc 1..5 都指向同一服务器，故一台真机常有数条连接/会话。若只排除受理 accept 的
|
||||
// 那一条 session，停振铃的 phoneCallDiscarded 会漏到同一台设备的其它连接上，客户端
|
||||
// 的 update 处理器按 call_id 匹配后当作"通话被挂断"、立即杀掉它刚接起的通话
|
||||
//（现象：被叫按下接听后立刻 Failed to connect；主叫方/被叫单连接侧无此问题——故表现
|
||||
// 为「A 打 B 正常、B 打 A 一接就断」的方向不对称）。按 business auth_key 排除可覆盖该
|
||||
// 设备的全部连接。See memory: call-*.
|
||||
// ⚠ 本部署的致命陷阱：OwpenGram 客户端把 dc 1..5 都指向同一台服务器，所以一台真机
|
||||
// 会对本服务器开【多条】连接，且每条连接各自握手、各有【不同】的 perm/business
|
||||
// auth_key。服务端仅凭 auth_key 无法把「同一台真机的其它连接」与「另一台真机」区分
|
||||
// 开。任何「发给被叫、排除受理设备」的 phoneCallDiscarded 都会漏到受理真机的其它连接
|
||||
// 上——客户端 update 处理器按 call_id 匹配后当作「通话被挂断」，立即杀掉它刚接起的
|
||||
// 通话（现象：被叫一按接听就 Failed to connect；且因单连接的桌面端不受影响，表现为
|
||||
// 「A 打 B 正常、B 打 A 一接就断」的方向不对称）。
|
||||
//
|
||||
// 取舍：宁可放弃"多真机时其它设备立即停振铃"这一优化（其它真机会各自走 ring 超时
|
||||
// 停振铃，延迟数十秒、且多真机场景罕见），也绝不能误杀正在建立的通话。故此处不再
|
||||
// 下发任何 discard。See memory: call-stop-ringing-device-exclude.
|
||||
func (r *Router) pushPhoneCallStopRinging(ctx context.Context, call domain.PhoneCall) int {
|
||||
upd := r.phoneCallUpdatesWith(ctx, tgPhoneCallStopRinging(call), call, call.ParticipantID)
|
||||
if pusher, ok := r.deps.Sessions.(DeviceExcludingSessionPusher); ok {
|
||||
if businessAuthKeyID, has := AuthKeyIDFrom(ctx); has {
|
||||
sent, err := pusher.PushToUserExceptBusinessAuthKey(ctx, call.ParticipantID, businessAuthKeyID, proto.MessageFromServer, upd, r.cfg.OutboundPushTimeout)
|
||||
if err != nil {
|
||||
r.log.Debug("phone call stop ringing", zap.Int64("user_id", call.ParticipantID), zap.Int("sent", sent), zap.Error(err))
|
||||
}
|
||||
return sent
|
||||
}
|
||||
}
|
||||
// 回退：能力不可用时退回按 session 排除（旧行为）。
|
||||
return r.pushUserMessage(ctx, call.ParticipantID, "phone call stop ringing", upd)
|
||||
return 0
|
||||
}
|
||||
|
||||
// pushPhoneCallDiscardedBoth 把终态推给双方全部设备（发起设备由 ctx except 排除，
|
||||
|
|
|
|||
|
|
@ -249,12 +249,16 @@ func TestPhoneCallRPCHappyPath(t *testing.T) {
|
|||
t.Fatalf("acceptCall result = %T, want PhoneCallWaiting (callee view)", acceptRes.PhoneCall)
|
||||
}
|
||||
pushes = f.sessions.records()
|
||||
if len(pushes) != 2 {
|
||||
t.Fatalf("acceptCall pushes = %d, want 2 (caller accepted + callee stop-ringing)", len(pushes))
|
||||
// acceptCall 只推一条：phoneCallAccepted 给主叫。曾经还会给被叫其它设备推一条
|
||||
// 合成 phoneCallDiscarded 停振铃，但在本部署（一台真机对同一服务器有多条各自
|
||||
// auth_key 的连接，dc 1..5 别名）里，该 discard 会漏到受理真机的其它连接上、误杀
|
||||
// 刚接起的通话，故已停用。See memory: call-stop-ringing-device-exclude。
|
||||
if len(pushes) != 1 {
|
||||
t.Fatalf("acceptCall pushes = %d, want 1 (caller accepted only; stop-ringing disabled)", len(pushes))
|
||||
}
|
||||
accepted, ok := phoneCallPayload(t, pushes[0]).(*tg.PhoneCallAccepted)
|
||||
if !ok || pushes[0].userID != f.caller.ID {
|
||||
t.Fatalf("first accept push = %+v payload %T, want PhoneCallAccepted to caller", pushes[0], phoneCallPayload(t, pushes[0]))
|
||||
t.Fatalf("accept push = %+v payload %T, want PhoneCallAccepted to caller", pushes[0], phoneCallPayload(t, pushes[0]))
|
||||
}
|
||||
if string(accepted.GB) != string(gb) {
|
||||
t.Fatalf("accepted.g_b must be relayed verbatim")
|
||||
|
|
@ -264,16 +268,13 @@ func TestPhoneCallRPCHappyPath(t *testing.T) {
|
|||
if got := accepted.Protocol.LibraryVersions; len(got) != 1 || got[0] != "9.0.0" {
|
||||
t.Fatalf("negotiated library_versions = %v, want preferred [9.0.0]", got)
|
||||
}
|
||||
// ⚠ P0-1：被叫其它设备必须收合成 phoneCallDiscarded（无 reason），绝不能是 accepted。
|
||||
stop, ok := phoneCallPayload(t, pushes[1]).(*tg.PhoneCallDiscarded)
|
||||
if !ok || pushes[1].userID != f.callee.ID {
|
||||
t.Fatalf("second accept push = %+v payload %T, want PhoneCallDiscarded to callee devices", pushes[1], phoneCallPayload(t, pushes[1]))
|
||||
}
|
||||
if pushes[1].excludeSession != phoneCalleeSession {
|
||||
t.Fatalf("stop-ringing must exclude the accepting session, got exclude=%d", pushes[1].excludeSession)
|
||||
}
|
||||
if _, hasReason := stop.GetReason(); hasReason || stop.NeedRating || stop.NeedDebug {
|
||||
t.Fatalf("stop-ringing payload = %+v, want reason-less, need_*=false", stop)
|
||||
// 绝不能给被叫（受理方）再推任何 phoneCallDiscarded——那会误杀刚接起的通话。
|
||||
for _, p := range pushes {
|
||||
if p.userID == f.callee.ID {
|
||||
if _, isDiscard := phoneCallPayload(t, p).(*tg.PhoneCallDiscarded); isDiscard {
|
||||
t.Fatalf("acceptCall must NOT push any phoneCallDiscarded to the callee, got %+v", p)
|
||||
}
|
||||
}
|
||||
}
|
||||
f.sessions.reset()
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue