From 157b82b5f588bdac3cc720e3675a4f7c37eea4eb Mon Sep 17 00:00:00 2001 From: onysd Date: Mon, 20 Jul 2026 16:47:21 +0300 Subject: [PATCH] i tired of fixing calls --- .../phone_stop_ringing_exclude_test.go | 52 ----------------- .../mtprotoedge/production_compat_test.go | 2 +- internal/mtprotoedge/session_manager.go | 58 +++---------------- internal/rpc/deps.go | 8 --- internal/rpc/phone_calls.go | 10 ++++ internal/rpc/phone_push.go | 37 +++++------- internal/rpc/phone_rpc_test.go | 27 ++++----- 7 files changed, 46 insertions(+), 148 deletions(-) delete mode 100644 internal/mtprotoedge/phone_stop_ringing_exclude_test.go diff --git a/internal/mtprotoedge/phone_stop_ringing_exclude_test.go b/internal/mtprotoedge/phone_stop_ringing_exclude_test.go deleted file mode 100644 index 4e3f0951..00000000 --- a/internal/mtprotoedge/phone_stop_ringing_exclude_test.go +++ /dev/null @@ -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") - } -} diff --git a/internal/mtprotoedge/production_compat_test.go b/internal/mtprotoedge/production_compat_test.go index 00c67427..5d235c76 100644 --- a/internal/mtprotoedge/production_compat_test.go +++ b/internal/mtprotoedge/production_compat_test.go @@ -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 { diff --git a/internal/mtprotoedge/session_manager.go b/internal/mtprotoedge/session_manager.go index 4deead65..735a4244 100644 --- a/internal/mtprotoedge/session_manager.go +++ b/internal/mtprotoedge/session_manager.go @@ -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) } diff --git a/internal/rpc/deps.go b/internal/rpc/deps.go index c9ed7d24..eefaa89c 100644 --- a/internal/rpc/deps.go +++ b/internal/rpc/deps.go @@ -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 无意义。 diff --git a/internal/rpc/phone_calls.go b/internal/rpc/phone_calls.go index 4f303776..4923a4f1 100644 --- a/internal/rpc/phone_calls.go +++ b/internal/rpc/phone_calls.go @@ -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) diff --git a/internal/rpc/phone_push.go b/internal/rpc/phone_push.go index 9a96dfd6..45f03e75 100644 --- a/internal/rpc/phone_push.go +++ b/internal/rpc/phone_push.go @@ -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 排除， diff --git a/internal/rpc/phone_rpc_test.go b/internal/rpc/phone_rpc_test.go index 97f53331..b114d683 100644 --- a/internal/rpc/phone_rpc_test.go +++ b/internal/rpc/phone_rpc_test.go @@ -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()