astra/owpengram-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

fix for calls

Browse source
This commit is contained in:
onysd 2026-07-20 16:03:33 +03:00
parent 406a8805d8
commit 46afb97396
10 changed files with 151 additions and 101 deletions
Download patch file Download diff file Expand all files Collapse all files

1
cmd/telesrv/main.go
View file

@ -644,7 +644,6 @@ func run(logger *zap.Logger) error {
t, err := turnsrv.New(turnsrv.Config{
UDPPort: cfg.TURNUDPPort,
AdvertiseIP: turnAdvertise,
ExtraIPs: cfg.TURNExtraIPs,
SharedSecret: cfg.TURNSecret,
RelayMinPort: cfg.TURNRelayMinPort,
RelayMaxPort: cfg.TURNRelayMaxPort,

5
internal/config/config.go
View file

@ -353,10 +353,6 @@ type Config struct {
// TURNAdvertiseIP 是写进 phoneConnectionWebrtc 与 relay 分配的客户端可达
// 地址，默认回落 SFUAdvertiseIP → AdvertiseIP。
TURNAdvertiseIP string
// TURNExtraIPs 是额外下发的 TURN/STUN 候选 IP（逗号分隔）。当客户端与服务器
// 同处一个局域网、AdvertiseIP 又是公网 IP 时，把 LAN IP（如 192.168.x.x）列进
// 这里，可修复 LAN 端发起通话「Failed to connect」（NAT hairpin）。详见 turnsrv.Config.ExtraIPs。
TURNExtraIPs []string
// TURNSecret 是 TURN REST 凭据 HMAC 密钥；为空则进程级随机（单实例自洽，
// 多实例/外部 coturn 必须显式配置同一值）。
TURNSecret string
@ -611,7 +607,6 @@ func Load() (Config, error) {
TURNEnable: envBoolOr("TELESRV_TURN_ENABLE", true),
TURNUDPPort: envIntOr("TELESRV_TURN_UDP_PORT", 12400),
TURNAdvertiseIP: envOr("TELESRV_TURN_ADVERTISE_IP", ""),
TURNExtraIPs: envListOr("TELESRV_TURN_EXTRA_IPS", nil),
TURNSecret: envOr("TELESRV_TURN_SECRET", ""),
TURNRelayMinPort: envIntOr("TELESRV_TURN_RELAY_MIN_PORT", 12500),
TURNRelayMaxPort: envIntOr("TELESRV_TURN_RELAY_MAX_PORT", 12999),

52
internal/mtprotoedge/phone_stop_ringing_exclude_test.go Normal file
View file

@ -0,0 +1,52 @@
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")
}
}

2
internal/mtprotoedge/production_compat_test.go
View file

@ -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, t, msg, timeout)
return m.pushToUserBestEffort(ctx, userID, nil, excludeSessionID, [8]byte{}, t, msg, timeout)
}
func (m *SessionManager) Online() int {

58
internal/mtprotoedge/session_manager.go
View file

@ -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, t, getUpdates, true, func(c *Conn) error {
return m.pushToUserWithSender(ctx, userID, excludeAuthKeyID, excludeSessionID, [8]byte{}, 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, t, getUpdates, false, func(c *Conn) error {
return m.pushToUserWithSender(ctx, userID, &excludeAuthKeyID, excludeSessionID, [8]byte{}, t, getUpdates, false, func(c *Conn) error {
if c.outbound == nil || c.outboundControl == nil {
return ErrConnClosed
}
@ -1516,10 +1516,38 @@ 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, t, msg, timeout)
return m.pushToUserBestEffort(ctx, userID, &excludeAuthKeyID, excludeSessionID, [8]byte{}, t, msg, timeout)
}
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) {
// 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) {
if ctx != nil && ctx.Err() != nil {
return 0, ctx.Err()
}
@ -1545,7 +1573,7 @@ func (m *SessionManager) pushToUserBestEffort(ctx context.Context, userID int64,
defer cancel()
}
getUpdates := onceLayerUpdatesFanout(sendCtx, msg)
return m.pushToUserWithSender(ctx, userID, excludeAuthKeyID, excludeSessionID, t, getUpdates, true, func(c *Conn) error {
return m.pushToUserWithSender(ctx, userID, excludeAuthKeyID, excludeSessionID, excludeBusinessAuthKeyID, t, getUpdates, true, func(c *Conn) error {
if c.outbound == nil || c.outboundControl == nil {
return ErrConnClosed
}
@ -1592,7 +1620,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, 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, excludeBusinessAuthKeyID [8]byte, 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)
@ -1608,7 +1636,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) {
if shouldExcludeSession(c, excludeAuthKeyID, excludeSessionID) || shouldExcludeDevice(c, excludeBusinessAuthKeyID) {
excluded++
continue
}
@ -2478,6 +2506,22 @@ 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)
}

8
internal/rpc/deps.go
View file

@ -165,6 +165,14 @@ 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 无意义。

24
internal/rpc/phone_push.go
View file

@ -5,6 +5,7 @@ import (
"github.com/iamxvbaba/td/proto"
"github.com/iamxvbaba/td/tg"
"go.uber.org/zap"
"telesrv/internal/domain"
)
@ -34,10 +35,29 @@ 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 必须是接听设备的请求上下文：except 语义恰好把赢家排除在外。
// pushPhoneCallStopRinging 向被叫【其它设备】推合成 phoneCallDiscarded 停振铃（P0-1 修正）。
// ctx 必须是接听设备的请求上下文。
//
// ⚠ 排除必须按【设备】（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-*.
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)
}

20
internal/rpc/phone_turn.go
View file

@ -50,22 +50,8 @@ func (r *Router) phoneCallConnections(callerID int64) []domain.PhoneCallConnecti
// stun flag——单条目 stun+turn 在 Android 上只会产出 TURN server、丢失 STUN
//（org_telegram_messenger_voip_Instance.cpp:848-884）。TDesktop 两种写法都认。
// TURN username 是 REST 格式 "<expiry>:<uid>"，天然避开 "reflector" 劫持禁区。
//
// 每个可达 IP（AdvertiseIP + ExtraIPs）都下发一对 STUN/TURN 候选，ICE 逐一
// 尝试并选可达者：LAN 客户端走 LAN IP、外网客户端走公网 IP。id 必须全局唯一
// 且从 1 递增（DrKLO 用 id 做 reflector 映射）。凭据与 IP 无关（TURN REST 只
// 校验 HMAC），同一份 username/password 对所有 IP 有效。
ips := append([]string{t.IP()}, t.ExtraIPs()...)
conns := make([]domain.PhoneCallConnection, 0, len(ips)*2)
id := int64(1)
for _, ip := range ips {
if ip == "" {
continue
return []domain.PhoneCallConnection{
{ID: 1, IP: t.IP(), Port: t.Port(), Stun: true},
{ID: 2, IP: t.IP(), Port: t.Port(), Username: username, Password: password, Turn: true},
}
conns = append(conns, domain.PhoneCallConnection{ID: id, IP: ip, Port: t.Port(), Stun: true})
id++
conns = append(conns, domain.PhoneCallConnection{ID: id, IP: ip, Port: t.Port(), Username: username, Password: password, Turn: true})
id++
}
return conns
}

52
internal/rpc/phone_turn_test.go
View file

@ -11,7 +11,6 @@ import (
type fakeTURN struct {
ip string
port int
extra []string
enabled bool
credUser string
credPass string
@ -23,7 +22,6 @@ func (f *fakeTURN) Credentials(string) (string, string, error) {
}
func (f *fakeTURN) IP() string { return f.ip }
func (f *fakeTURN) Port() int { return f.port }
func (f *fakeTURN) ExtraIPs() []string { return f.extra }
func (f *fakeTURN) Close() error { return nil }
func newTURNRouter(t *testing.T, turn *fakeTURN) *Router {
@ -38,61 +36,21 @@ func TestPhoneCallConnectionsDisabledTURNReturnsNil(t *testing.T) {
}
}
func TestPhoneCallConnectionsSingleIP(t *testing.T) {
func TestPhoneCallConnectionsSplitStunAndTurn(t *testing.T) {
r := newTURNRouter(t, &fakeTURN{
enabled: true, ip: "89.28.58.29", port: 12400,
credUser: "u", credPass: "p",
})
conns := r.phoneCallConnections(1)
if len(conns) != 2 {
t.Fatalf("single IP: want 2 conns (stun+turn), got %d: %+v", len(conns), conns)
t.Fatalf("want 2 conns (stun+turn), got %d: %+v", len(conns), conns)
}
// STUN and TURN must be SEPARATE entries (DrKLO's JNI ignores the stun flag
// on a combined entry, dropping STUN).
if !conns[0].Stun || conns[0].Turn || conns[0].ID != 1 || conns[0].IP != "89.28.58.29" {
t.Fatalf("conn[0] should be STUN id1 on public IP, got %+v", conns[0])
t.Fatalf("conn[0] should be STUN id1, got %+v", conns[0])
}
if !conns[1].Turn || conns[1].Stun || conns[1].ID != 2 || conns[1].Username != "u" || conns[1].Password != "p" {
t.Fatalf("conn[1] should be TURN id2 with creds, got %+v", conns[1])
}
}
func TestPhoneCallConnectionsExtraIPsAddCandidatesWithUniqueIDs(t *testing.T) {
r := newTURNRouter(t, &fakeTURN{
enabled: true, ip: "89.28.58.29", port: 12400,
extra: []string{"192.168.0.20", ""}, // empty entry must be skipped
credUser: "u", credPass: "p",
})
conns := r.phoneCallConnections(1)
// public (stun+turn) + LAN (stun+turn) = 4; the empty extra IP is skipped.
if len(conns) != 4 {
t.Fatalf("want 4 conns, got %d: %+v", len(conns), conns)
}
seenIDs := map[int64]bool{}
for _, c := range conns {
if seenIDs[c.ID] {
t.Fatalf("duplicate connection id %d: %+v", c.ID, conns)
}
seenIDs[c.ID] = true
if c.IP == "" {
t.Fatalf("empty IP leaked into connections: %+v", conns)
}
}
// IDs must be a contiguous 1..4 run (DrKLO maps reflectors by id).
for id := int64(1); id <= 4; id++ {
if !seenIDs[id] {
t.Fatalf("missing contiguous id %d: %+v", id, conns)
}
}
// The LAN IP must appear as both a STUN and a TURN candidate.
var lanStun, lanTurn bool
for _, c := range conns {
if c.IP == "192.168.0.20" && c.Stun {
lanStun = true
}
if c.IP == "192.168.0.20" && c.Turn {
lanTurn = true
}
}
if !lanStun || !lanTurn {
t.Fatalf("LAN IP must yield both STUN and TURN candidates, got %+v", conns)
}
}

12
internal/turnsrv/turnsrv.go
View file

@ -29,14 +29,6 @@ type Config struct {
// AdvertiseIP 是写进 phoneConnectionWebrtc 与 relay 分配地址的客户端可达
// 地址。⚠ 127.0.0.1 时真机拿到的 relay candidate 不可达（媒体面静默失败）。
AdvertiseIP string
// ExtraIPs 是额外下发的 TURN/STUN 候选地址（除 AdvertiseIP 外）。典型用途：
// 当客户端与服务器同处一个局域网、而 AdvertiseIP 是公网 IP 时，路由器多半不
// 支持 NAT 回环（hairpin），LAN 内的客户端无法经公网 IP 触达 TURN 控制通道
//（现象：LAN 端发起的通话「Failed to connect」，外网端发起的却正常）。把 LAN
// IP（如 192.168.x.x）列进这里，ICE 会一并尝试，LAN 客户端走 LAN 候选、外网
// 客户端走公网候选。relay 分配地址仍是 AdvertiseIP（LAN-LAN 通话用 host
// 候选直连、不经 relay，故公网 relay 地址不成问题）。
ExtraIPs []string
// Realm 是 TURN long-term credential 的 realm（任意稳定串即可）。
Realm string
// SharedSecret 是 REST 凭据的 HMAC 密钥；为空则进程级随机生成
@ -62,8 +54,6 @@ type Service interface {
// IP/Port 返回客户端可达的服务地址。
IP() string
Port() int
// ExtraIPs 返回除 IP() 外额外下发的候选地址（可空）。
ExtraIPs() []string
Close() error
}
@ -78,7 +68,6 @@ func (disabled) Credentials(string) (string, string, error) {
}
func (disabled) IP() string { return "" }
func (disabled) Port() int { return 0 }
func (disabled) ExtraIPs() []string { return nil }
func (disabled) Close() error { return nil }
type pionTURN struct {
@ -163,7 +152,6 @@ func (t *pionTURN) Credentials(user string) (string, string, error) {
func (t *pionTURN) IP() string { return t.cfg.AdvertiseIP }
func (t *pionTURN) Port() int { return t.cfg.UDPPort }
func (t *pionTURN) ExtraIPs() []string { return t.cfg.ExtraIPs }
func (t *pionTURN) Close() error { return t.server.Close() }
func randomSecret() (string, error) {