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

fix(phone): sync confirmed call timeout handling

Browse source
This commit is contained in:
iamxvbaba 2026-07-26 15:59:20 +08:00
parent 7ce15f68ab
commit c3c079edf3
9 changed files with 152 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

1
cmd/telesrv/main.go
View file

@ -727,6 +727,7 @@ func run(logger *zap.Logger) error {
RingTimeout: cfg.CallRingTimeout,
TombstoneTTL: cfg.CallTombstoneTTL,
MaxActivePerUser: cfg.CallMaxActivePerUser,
MaxRegistryEntries: cfg.CallRegistryMaxEntries,
SignalingRatePerSecond: cfg.CallSignalingRate,
})
// 私聊端对端加密（Secret Chat）握手状态机 + qts 投递队列（盲中继）。

1
docs/configuration.en.md
View file

@ -545,6 +545,7 @@ path. `TELESRV_PUBLIC_BASE_URL` must resolve to that proxy for moderation freeze
| `TELESRV_CALL_RING_TIMEOUT` | duration / `90s` | Server fallback timeout for ringing/accepted private calls; should remain aligned with the client `callRingTimeoutMs`. |
| `TELESRV_CALL_TOMBSTONE_TTL` | duration / `60s` | Terminal-call tombstone window for idempotency and late RPC absorption. |
| `TELESRV_CALL_MAX_ACTIVE_PER_USER` | int / `4` | Maximum non-terminal private calls per user. Non-positive values are normalized by the phone service. |
| `TELESRV_CALL_REGISTRY_MAX_ENTRIES` | int / `10000` | Process-wide private-call registry hard limit. At capacity, new calls fail with `CALL_OCCUPY_FAILED`; established calls are never evicted by age. |
| `TELESRV_CALL_SIGNALING_MAX_BYTES` | int bytes / `65536` | Maximum payload for one `phone.sendSignalingData`. |
| `TELESRV_CALL_SIGNALING_RATE` | int / `50` | Signaling forwards per call per second; excess is silently dropped. |
| `TELESRV_CALL_EXPIRY_INTERVAL` | duration / `1s` | Call-expiry dispatcher polling interval. |

1
docs/configuration.zh-CN.md
View file

@ -522,6 +522,7 @@ active key。不要手工编辑 manifest 或 PEM，不要在各实例上分别
| `TELESRV_CALL_RING_TIMEOUT` | duration / `90s` | 私聊通话 ringing/accepted 服务端兜底超时，应与客户端 `callRingTimeoutMs` 保持一致。 |
| `TELESRV_CALL_TOMBSTONE_TTL` | duration / `60s` | 终态通话 tombstone 的幂等/晚到 RPC 吸收窗口。 |
| `TELESRV_CALL_MAX_ACTIVE_PER_USER` | int / `4` | 单用户非终态私聊通话上限；非正值由 phone service 归一。 |
| `TELESRV_CALL_REGISTRY_MAX_ENTRIES` | int / `10000` | 进程级私聊通话 registry 硬上限；满载返回 `CALL_OCCUPY_FAILED`，不按年龄驱逐已建立通话。 |
| `TELESRV_CALL_SIGNALING_MAX_BYTES` | int bytes / `65536` | 单条 `phone.sendSignalingData` 载荷上限。 |
| `TELESRV_CALL_SIGNALING_RATE` | int / `50` | 单通话每秒信令转发上限，超限静默丢弃。 |
| `TELESRV_CALL_EXPIRY_INTERVAL` | duration / `1s` | 通话 expiry dispatcher 轮询间隔。 |

27
internal/app/phone/registry.go
View file

@ -46,28 +46,27 @@ func newRegistry() *registry {
}
}
// sweepLocked 是 P1 的纯年龄 GC（调用方持有 r.mu）：
// - 终态 tombstone 超过 tombstoneTTL → 回收（密钥材料随之销毁）；
// - 非终态超过 2×ringTimeout → 直接回收（双端同时崩溃的兜底，防僵尸通话
// 吃满并发上限；不推送、不落历史，正常超时由客户端定时器与 P2 dispatcher 处理）。
func (r *registry) sweepLocked(nowUnix int64, ringTimeoutSec, tombstoneTTLSec int64) {
// sweepTombstonesLocked 只回收超过保留期的终态 tombstone（调用方持有 r.mu）。
//
// 非终态绝不能在 GC 中按 Date 直接删除：Requested/Ringing/Accepted 的超时必须由
// Service.ExpireDue 完成状态迁移、双端推送和历史落库；Confirmed 没有服务端时长
// 上限，必须一直可供 signaling/discard 寻址，直到显式挂断或进程重启。
func (r *registry) sweepTombstonesLocked(nowUnix, tombstoneTTLSec int64) {
for id, e := range r.byID {
switch {
case e.call.Terminal():
if nowUnix-int64(e.call.DiscardedAt) > tombstoneTTLSec {
if e.call.Terminal() && nowUnix-int64(e.call.DiscardedAt) > tombstoneTTLSec {
r.removeLocked(id, e, false)
}
default:
if nowUnix-int64(e.call.Date) > 2*ringTimeoutSec {
r.removeLocked(id, e, true)
}
}
}
}
func (r *registry) removeLocked(id int64, e *entry, wasActive bool) {
delete(r.byID, id)
delete(r.byRandom, randomKey{callerID: e.call.AdminID, randomID: e.call.RandomID})
key := randomKey{callerID: e.call.AdminID, randomID: e.call.RandomID}
// 终态后允许客户端复用 random_id 创建新通话；旧 tombstone 到期时不能
// 把已指向新 call 的幂等索引一并删掉。
if indexedID, ok := r.byRandom[key]; ok && indexedID == id {
delete(r.byRandom, key)
}
if wasActive {
r.decActiveLocked(e.call.AdminID)
}

13
internal/app/phone/service.go
View file

@ -45,6 +45,9 @@ type Config struct {
TombstoneTTL time.Duration
// MaxActivePerUser 是单用户并发非终态通话上限（防呼叫轰炸自锁）。
MaxActivePerUser int
// MaxRegistryEntries 是进程内 registry 的硬上限。达到上限时拒绝新通话，
// 不驱逐可能仍在进行的 Confirmed 通话。
MaxRegistryEntries int
// SignalingRatePerSecond 是单通话每秒信令转发上限；超限静默丢弃（不破坏客户端状态机）。
SignalingRatePerSecond int
}
@ -59,6 +62,9 @@ func (c Config) withDefaults() Config {
if c.MaxActivePerUser <= 0 {
c.MaxActivePerUser = 4
}
if c.MaxRegistryEntries <= 0 {
c.MaxRegistryEntries = 10_000
}
if c.SignalingRatePerSecond <= 0 {
c.SignalingRatePerSecond = 50
}
@ -103,7 +109,7 @@ func (s *Service) RequestCall(ctx context.Context, callerID int64, in domain.Pho
s.reg.mu.Lock()
defer s.reg.mu.Unlock()
s.reg.sweepLocked(nowUnix, int64(s.cfg.RingTimeout/time.Second), int64(s.cfg.TombstoneTTL/time.Second))
s.reg.sweepTombstonesLocked(nowUnix, int64(s.cfg.TombstoneTTL/time.Second))
// 幂等：同一 (callerID, randomID) 的未终结通话直接返回快照，吸收客户端重试。
key := randomKey{callerID: callerID, randomID: in.RandomID}
@ -112,6 +118,9 @@ func (s *Service) RequestCall(ctx context.Context, callerID int64, in domain.Pho
return e.call, nil
}
}
if len(s.reg.byID) >= s.cfg.MaxRegistryEntries {
return domain.PhoneCall{}, ErrOccupyFailed
}
if s.reg.active[callerID] >= s.cfg.MaxActivePerUser {
return domain.PhoneCall{}, ErrOccupyFailed
}
@ -306,7 +315,7 @@ func (s *Service) ExpireDue(ctx context.Context, now time.Time) []domain.PhoneCa
s.reg.markDiscardedLocked(e, reason, 0, int(nowUnix))
expired = append(expired, e.call)
}
s.reg.sweepLocked(nowUnix, ringSec, int64(s.cfg.TombstoneTTL/time.Second))
s.reg.sweepTombstonesLocked(nowUnix, int64(s.cfg.TombstoneTTL/time.Second))
return expired
}

67
internal/app/phone/service_test.go
View file

@ -72,6 +72,7 @@ func newTestService(clk clock.Clock, mutate ...func(*Config)) *Service {
RingTimeout: 90 * time.Second,
TombstoneTTL: 60 * time.Second,
MaxActivePerUser: 4,
MaxRegistryEntries: 10_000,
SignalingRatePerSecond: 50,
}
for _, fn := range mutate {
@ -265,9 +266,15 @@ func TestPhoneCallRandomIDIdempotent(t *testing.T) {
if err != nil || third.ID == first.ID {
t.Fatalf("post-discard request id = %d err=%v, want fresh call", third.ID, err)
}
// 旧 tombstone 到期回收时，不得误删已改指向新 call 的 random_id 索引。
clk.Advance(61 * time.Second)
retry, err := s.RequestCall(ctx, 1, req)
if err != nil || retry.ID != third.ID {
t.Fatalf("retry after old tombstone GC id = %d err=%v, want %d", retry.ID, err, third.ID)
}
}
func TestPhoneCallQuotaAndSweep(t *testing.T) {
func TestPhoneCallQuotaAndExpiry(t *testing.T) {
clk := newTestClock()
s := newTestService(clk, func(c *Config) { c.MaxActivePerUser = 2 })
ctx := context.Background()
@ -281,10 +288,53 @@ func TestPhoneCallQuotaAndSweep(t *testing.T) {
if _, err := s.RequestCall(ctx, 1, domain.PhoneCallRequest{CalleeID: 99, RandomID: 99, GAHash: gaHash, Protocol: testProtocol()}); !errors.Is(err, ErrOccupyFailed) {
t.Fatalf("over quota err = %v, want ErrOccupyFailed", err)
}
// 双端崩溃兜底：超过 2×RingTimeout 的僵尸通话被纯年龄 GC 回收，配额释放。
clk.Advance(181 * time.Second)
// 未建立通话只能由 ExpireDue 迁入终态，确保 dispatcher 能推送并落历史；
// registry GC 不得静默删除 active call。
clk.Advance(91 * time.Second)
expired := s.ExpireDue(ctx, clk.Now())
if len(expired) != 2 {
t.Fatalf("expired = %d, want 2", len(expired))
}
if _, err := s.RequestCall(ctx, 1, domain.PhoneCallRequest{CalleeID: 99, RandomID: 99, GAHash: gaHash, Protocol: testProtocol()}); err != nil {
t.Fatalf("request after sweep: %v", err)
t.Fatalf("request after expiry: %v", err)
}
}
func TestPhoneCallRegistryCapacityDoesNotEvictConfirmedCall(t *testing.T) {
clk := newTestClock()
s := newTestService(clk, func(c *Config) { c.MaxRegistryEntries = 1 })
ctx := context.Background()
ga, gaHash := testGA()
confirmed := mustRequest(t, s, 1, 2, gaHash)
if _, err := s.AcceptCall(ctx, 2, confirmed.ID, confirmed.AccessHash, testGB(), testProtocol(), domain.SessionRef{}); err != nil {
t.Fatalf("accept: %v", err)
}
if _, _, err := s.ConfirmCall(ctx, 1, confirmed.ID, confirmed.AccessHash, ga, 1, testProtocol()); err != nil {
t.Fatalf("confirm: %v", err)
}
clk.Advance(365 * 24 * time.Hour)
if got := s.ExpireDue(ctx, clk.Now()); len(got) != 0 {
t.Fatalf("confirmed call expired after one year: %+v", got)
}
if _, err := s.RequestCall(ctx, 3, domain.PhoneCallRequest{
CalleeID: 4, RandomID: 2, GAHash: gaHash, Protocol: testProtocol(),
}); !errors.Is(err, ErrOccupyFailed) {
t.Fatalf("request at registry capacity err = %v, want ErrOccupyFailed", err)
}
if snap, ok := s.Lookup(ctx, confirmed.ID, confirmed.AccessHash); !ok || snap.State != domain.PhoneCallStateConfirmed {
t.Fatalf("confirmed call = %+v ok=%v, want preserved", snap, ok)
}
if _, _, err := s.DiscardCall(ctx, 1, confirmed.ID, confirmed.AccessHash, domain.PhoneCallDiscardReasonHangup, 1); err != nil {
t.Fatalf("discard: %v", err)
}
clk.Advance(61 * time.Second)
if _, err := s.RequestCall(ctx, 3, domain.PhoneCallRequest{
CalleeID: 4, RandomID: 2, GAHash: gaHash, Protocol: testProtocol(),
}); err != nil {
t.Fatalf("request after tombstone GC: %v", err)
}
}
@ -476,4 +526,13 @@ func TestPhoneCallExpireDue(t *testing.T) {
if got := s.ExpireDue(ctx, clk.Now()); len(got) != 0 {
t.Fatalf("second ExpireDue = %d, want 0", len(got))
}
// 回归：旧 registry GC 会在 2×RingTimeout 后静默删除 Confirmed，导致后续
// sendSignalingData/discardCall 返回 CALL_PEER_INVALID。
clk.Advance(91 * time.Second)
if got := s.ExpireDue(ctx, clk.Now()); len(got) != 0 {
t.Fatalf("confirmed call expired after 2×RingTimeout: %+v", got)
}
if _, ok := s.Lookup(ctx, confirmedCall.ID, confirmedCall.AccessHash); !ok {
t.Fatal("confirmed call must remain addressable after 2×RingTimeout")
}
}

3
internal/config/config.go
View file

@ -319,6 +319,8 @@ type Config struct {
CallTombstoneTTL time.Duration
// CallMaxActivePerUser 是单用户并发非终态通话上限。
CallMaxActivePerUser int
// CallRegistryMaxEntries 是进程内通话 registry 的全局硬上限。
CallRegistryMaxEntries int
// CallSignalingMaxBytes 是 phone.sendSignalingData 单条载荷上限。
CallSignalingMaxBytes int
// CallSignalingRate 是单通话每秒信令转发上限（超限静默丢弃）。
@ -621,6 +623,7 @@ func Load() (Config, error) {
CallRingTimeout: envDurationOr("TELESRV_CALL_RING_TIMEOUT", 90*time.Second),
CallTombstoneTTL: envDurationOr("TELESRV_CALL_TOMBSTONE_TTL", 60*time.Second),
CallMaxActivePerUser: envIntOr("TELESRV_CALL_MAX_ACTIVE_PER_USER", 4),
CallRegistryMaxEntries: envIntOr("TELESRV_CALL_REGISTRY_MAX_ENTRIES", 10_000),
CallSignalingMaxBytes: envIntOr("TELESRV_CALL_SIGNALING_MAX_BYTES", 65536),
CallSignalingRate: envIntOr("TELESRV_CALL_SIGNALING_RATE", 50),
CallExpiryInterval: envDurationOr("TELESRV_CALL_EXPIRY_INTERVAL", time.Second),

3
internal/config/config_test.go
View file

@ -34,6 +34,9 @@ func TestLoadDefaultsAdvertiseIPToLoopback(t *testing.T) {
if cfg.PublicAppName != "telesrv" {
t.Fatalf("PublicAppName = %q, want telesrv", cfg.PublicAppName)
}
if cfg.CallRegistryMaxEntries != 10_000 {
t.Fatalf("CallRegistryMaxEntries = %d, want 10000", cfg.CallRegistryMaxEntries)
}
}
func TestLoadUsesExplicitAdvertiseIP(t *testing.T) {

49
internal/rpc/phone_rpc_test.go
View file

@ -782,6 +782,55 @@ func TestPhoneExpiryDispatcherMissedCall(t *testing.T) {
}
}
func TestPhoneExpiryDispatcherPreservesConfirmedCallPastRingTimeout(t *testing.T) {
clk := &phoneTestClock{now: time.Unix(1_700_000_000, 0)}
messages := &phoneCaptureMessages{}
f := newPhoneFixtureFull(t, clk, messages)
peer := establishPhoneCall(t, f)
// 回归旧的 registry GC：2×RingTimeout 后触发 dispatcher 时，已建立通话
// 仍须可供后续 signaling/discard 寻址，不能返回 CALL_PEER_INVALID。
clk.Advance(181 * time.Second)
dispatcher := NewPhoneExpiryDispatcher(f.router, zaptest.NewLogger(t), time.Second)
dispatcher.DispatchOnce(f.ctx)
if pushes := f.sessions.records(); len(pushes) != 0 {
t.Fatalf("confirmed call expiry pushes = %+v, want none", pushes)
}
if sent := messages.records(); len(sent) != 0 {
t.Fatalf("confirmed call expiry history = %+v, want none", sent)
}
ok, err := f.router.onPhoneSendSignalingData(f.callerCtx(), &tg.PhoneSendSignalingDataRequest{
Peer: peer,
Data: []byte("after-ring-timeout"),
})
if err != nil || !ok {
t.Fatalf("sendSignalingData after 2×RingTimeout = %v err=%v", ok, err)
}
pushes := f.sessions.records()
if len(pushes) != 1 || pushes[0].rawAuthKeyID != phoneCalleeRawAuthKey || pushes[0].targetSession != phoneCalleeSession {
t.Fatalf("signaling pushes = %+v, want callee accepted session", pushes)
}
updates, ok := pushes[0].msg.(*tg.Updates)
if !ok || len(updates.Updates) != 1 {
t.Fatalf("signaling payload = %T %+v, want single-update tg.Updates", pushes[0].msg, pushes[0].msg)
}
signal, ok := updates.Updates[0].(*tg.UpdatePhoneCallSignalingData)
if !ok || signal.PhoneCallID != peer.ID || string(signal.Data) != "after-ring-timeout" {
t.Fatalf("signaling payload = %+v", updates.Updates[0])
}
f.sessions.reset()
if _, err := f.router.onPhoneDiscardCall(f.callerCtx(), &tg.PhoneDiscardCallRequest{
Peer: peer, Duration: 181, Reason: &tg.PhoneCallDiscardReasonHangup{},
}); err != nil {
t.Fatalf("discardCall after 2×RingTimeout: %v", err)
}
if sent := messages.records(); len(sent) != 1 {
t.Fatalf("discard history = %d, want 1", len(sent))
}
}
// TestPhoneCallHistoryThroughRealPipeline 走真实 messages 管线（memory store）验证
// 通话历史端到端：新 action kind 经媒体 JSONB 序列化、读路径 TL 转换后完整存活。
func TestPhoneCallHistoryThroughRealPipeline(t *testing.T) {