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

fix: sync call ringing device scope

Browse source
This commit is contained in:
A 2026-07-21 17:25:53 +08:00
parent 27ec9dce33
commit bf72c246b6
3 changed files with 103 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

5
internal/rpc/phone_calls.go
View file

@ -135,7 +135,10 @@ func (r *Router) onPhoneReceivedCall(ctx context.Context, peer tg.InputPhoneCall
if transitioned {
// ⚠ P1-2：receiveDate 推送必须在 P1 就位。主叫只有收到带 receive_date 的
// phoneCallWaiting 才会把 20s receive 定时器换成 90s ring 定时器。
r.pushPhoneCall(ctx, call.AdminID, call, "phone call ringing")
// 这条主叫视角更新只属于 requestCall 的来源设备；账号级广播会在 DrKLO
// 多账号同机时按相同 call_id 覆盖被叫的 pending phoneCallRequested，丢失
// g_a_hash 并在接听时触发 Ga hash mismatch。
r.pushPhoneCallToDevice(ctx, call.AdminID, call.CallerDevice, call, "phone call ringing")
}
return true, nil
}

30
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,6 +35,35 @@ func (r *Router) pushPhoneCall(ctx context.Context, targetUserID int64, call dom
return r.pushUserMessage(ctx, targetUserID, logMessage, r.phoneCallUpdates(ctx, call, targetUserID))
}
// pushPhoneCallToDevice 只把 phoneCall 状态推给一台精确的物理 session。
//
// 这是 fail-closed 路径：目标锚点缺失、session 已断开或编码/发送失败时都不得
// 回退为 user 级广播。呼出通话的 ringing 更新只属于 requestCall 来源设备；
// 扩大投递范围会在 DrKLO 多账号同机时污染另一账号的全局 pending 来电对象。
func (r *Router) pushPhoneCallToDevice(ctx context.Context, targetUserID int64, device domain.SessionRef, call domain.PhoneCall, logMessage string) {
if targetUserID == 0 || device.RawAuthKeyID == ([8]byte{}) || device.SessionID == 0 || r.deps.Sessions == nil {
if r.log != nil {
r.log.Debug(logMessage,
zap.Int64("target_user_id", targetUserID),
zap.Int64("call_id", call.ID),
zap.Int64("target_session_id", device.SessionID),
zap.String("delivery", "skipped_invalid_device_anchor"),
)
}
return
}
updates := r.phoneCallUpdates(ctx, call, targetUserID)
if err := r.deps.Sessions.PushToSessionForAuthKey(ctx, device.RawAuthKeyID, device.SessionID, proto.MessageFromServer, updates); err != nil && r.log != nil {
r.log.Debug(logMessage,
zap.Int64("target_user_id", targetUserID),
zap.Int64("call_id", call.ID),
zap.Int64("target_session_id", device.SessionID),
zap.Error(err),
)
}
}
// pushPhoneCallStopRinging 向被叫其它设备推合成 phoneCallDiscarded 停振铃（P0-1 修正）。
// ctx 必须是接听设备的请求上下文：except 语义恰好把赢家排除在外。
func (r *Router) pushPhoneCallStopRinging(ctx context.Context, call domain.PhoneCall) int {

80
internal/rpc/phone_rpc_test.go
View file

@ -25,6 +25,7 @@ import (
// phonePushRecord 记录一次定向推送（目标用户、被排除的 session、载荷）。
type phonePushRecord struct {
userID int64
rawAuthKeyID [8]byte
targetSession int64
excludeSession int64
msg bin.Encoder
@ -32,8 +33,9 @@ type phonePushRecord struct {
// phoneCaptureSessions 是带完整推送日志的 SessionBinder fake（captureSessions 只留最后一条）。
type phoneCaptureSessions struct {
mu sync.Mutex
log []phonePushRecord
mu sync.Mutex
log []phonePushRecord
pushErr error
}
func (s *phoneCaptureSessions) BindAuthKeyForSession([8]byte, int64, [8]byte) {}
@ -47,11 +49,11 @@ func (s *phoneCaptureSessions) UserIDResolvedForAuthKey([8]byte, int64) (int64,
func (s *phoneCaptureSessions) UnbindAuthKey([8]byte) int { return 0 }
func (s *phoneCaptureSessions) SetReceivesUpdatesForAuthKey([8]byte, int64, bool) {}
func (s *phoneCaptureSessions) PushToSessionForAuthKey(_ context.Context, _ [8]byte, sessionID int64, _ proto.MessageType, msg tg.UpdatesClass) error {
func (s *phoneCaptureSessions) PushToSessionForAuthKey(_ context.Context, rawAuthKeyID [8]byte, sessionID int64, _ proto.MessageType, msg tg.UpdatesClass) error {
s.mu.Lock()
defer s.mu.Unlock()
s.log = append(s.log, phonePushRecord{targetSession: sessionID, msg: msg})
return nil
s.log = append(s.log, phonePushRecord{rawAuthKeyID: rawAuthKeyID, targetSession: sessionID, msg: msg})
return s.pushErr
}
func (s *phoneCaptureSessions) PushToUserExceptAuthKeySession(_ context.Context, userID int64, _ [8]byte, excludeSessionID int64, _ proto.MessageType, msg tg.UpdatesClass) (int, error) {
@ -73,6 +75,12 @@ func (s *phoneCaptureSessions) reset() {
s.log = nil
}
func (s *phoneCaptureSessions) setPushError(err error) {
s.mu.Lock()
defer s.mu.Unlock()
s.pushErr = err
}
// stubPrivacy 只为 CanSee 服务；其余接口方法不在通话链路使用。
type stubPrivacy struct {
deny map[domain.PrivacyKey]bool
@ -104,8 +112,15 @@ type phoneFixture struct {
}
const (
phoneCallerSession = int64(101)
phoneCalleeSession = int64(202)
phoneCallerSession = int64(101)
phoneCalleeSession = int64(202)
phoneOtherCalleeSession = int64(303)
)
var (
phoneCallerRawAuthKey = [8]byte{0x11, 0x01}
phoneCalleeRawAuthKey = [8]byte{0x22, 0x02}
phoneOtherCalleeRawAuthKey = [8]byte{0x33, 0x03}
)
func newPhoneFixture(t *testing.T, privacy PrivacyService) *phoneFixture {
@ -133,11 +148,16 @@ func newPhoneFixture(t *testing.T, privacy PrivacyService) *phoneFixture {
}
func (f *phoneFixture) callerCtx() context.Context {
return WithSessionID(WithUserID(f.ctx, f.caller.ID), phoneCallerSession)
return WithSessionID(WithRawAuthKeyID(WithUserID(f.ctx, f.caller.ID), phoneCallerRawAuthKey), phoneCallerSession)
}
func (f *phoneFixture) calleeCtx() context.Context {
return WithSessionID(WithUserID(f.ctx, f.callee.ID), phoneCalleeSession)
return WithSessionID(WithRawAuthKeyID(WithUserID(f.ctx, f.callee.ID), phoneCalleeRawAuthKey), phoneCalleeSession)
}
func (f *phoneFixture) otherCalleeCtx() context.Context {
return WithSessionID(WithRawAuthKeyID(WithUserID(f.ctx, f.callee.ID), phoneOtherCalleeRawAuthKey), phoneOtherCalleeSession)
}
func phoneTestProtocol() tg.PhoneCallProtocol {
@ -226,8 +246,8 @@ func TestPhoneCallRPCHappyPath(t *testing.T) {
t.Fatalf("receivedCall = %v err=%v", ok, err)
}
pushes = f.sessions.records()
if len(pushes) != 1 || pushes[0].userID != f.caller.ID {
t.Fatalf("receivedCall pushes = %+v, want one to caller", pushes)
if len(pushes) != 1 || pushes[0].rawAuthKeyID != phoneCallerRawAuthKey || pushes[0].targetSession != phoneCallerSession {
t.Fatalf("receivedCall pushes = %+v, want caller device %x/%d", pushes, phoneCallerRawAuthKey, phoneCallerSession)
}
ringing, ok := phoneCallPayload(t, pushes[0]).(*tg.PhoneCallWaiting)
if !ok || ringing.ReceiveDate == 0 {
@ -236,6 +256,15 @@ func TestPhoneCallRPCHappyPath(t *testing.T) {
}
f.sessions.reset()
// 其它被叫设备晚到的 receivedCall 幂等成功，但不得再次推 ringing。
if ok, err := f.router.onPhoneReceivedCall(f.otherCalleeCtx(), tg.InputPhoneCall{ID: callID, AccessHash: accessHash}); err != nil || !ok {
t.Fatalf("duplicate receivedCall = %v err=%v", ok, err)
}
if pushes := f.sessions.records(); len(pushes) != 0 {
t.Fatalf("duplicate receivedCall pushes = %+v, want none", pushes)
}
f.sessions.reset()
// --- acceptCall（被叫赢家设备） ---
acceptRes, err := f.router.onPhoneAcceptCall(f.calleeCtx(), &tg.PhoneAcceptCallRequest{
Peer: tg.InputPhoneCall{ID: callID, AccessHash: accessHash},
@ -388,6 +417,35 @@ func TestPhoneCallRPCHappyPath(t *testing.T) {
}
}
func TestPhoneReceivedCallDevicePushFailureDoesNotBroadcast(t *testing.T) {
f := newPhoneFixture(t, stubPrivacy{})
_, gaHash, _ := phoneTestKeys()
res, err := f.router.onPhoneRequestCall(f.callerCtx(), &tg.PhoneRequestCallRequest{
UserID: &tg.InputUser{UserID: f.callee.ID, AccessHash: f.callee.AccessHash},
RandomID: 43,
GAHash: gaHash,
Protocol: phoneTestProtocol(),
})
if err != nil {
t.Fatalf("requestCall: %v", err)
}
waiting := res.PhoneCall.(*tg.PhoneCallWaiting)
f.sessions.reset()
f.sessions.setPushError(errors.New("caller session gone"))
if ok, err := f.router.onPhoneReceivedCall(f.calleeCtx(), tg.InputPhoneCall{ID: waiting.ID, AccessHash: waiting.AccessHash}); err != nil || !ok {
t.Fatalf("receivedCall = %v err=%v", ok, err)
}
pushes := f.sessions.records()
if len(pushes) != 1 || pushes[0].rawAuthKeyID != phoneCallerRawAuthKey || pushes[0].targetSession != phoneCallerSession {
t.Fatalf("receivedCall pushes = %+v, want one failed attempt to caller device", pushes)
}
if pushes[0].userID != 0 {
t.Fatalf("receivedCall failure fell back to user broadcast: %+v", pushes)
}
}
func TestPhoneCallRPCValidation(t *testing.T) {
f := newPhoneFixture(t, stubPrivacy{})
_, gaHash, gb := phoneTestKeys()