From c3c079edf3248b40259af6ab1b6db129bacc4267 Mon Sep 17 00:00:00 2001 From: iamxvbaba <28732408+iamxvbaba@users.noreply.github.com> Date: Sun, 26 Jul 2026 15:59:20 +0800 Subject: [PATCH] fix(phone): sync confirmed call timeout handling --- cmd/telesrv/main.go | 1 + docs/configuration.en.md | 1 + docs/configuration.zh-CN.md | 1 + internal/app/phone/registry.go | 29 +++++++------ internal/app/phone/service.go | 13 +++++- internal/app/phone/service_test.go | 67 ++++++++++++++++++++++++++++-- internal/config/config.go | 15 ++++--- internal/config/config_test.go | 3 ++ internal/rpc/phone_rpc_test.go | 49 ++++++++++++++++++++++ 9 files changed, 152 insertions(+), 27 deletions(-) diff --git a/cmd/telesrv/main.go b/cmd/telesrv/main.go index 53da4161..c1c8b82d 100644 --- a/cmd/telesrv/main.go +++ b/cmd/telesrv/main.go @@ -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 投递队列（盲中继）。 diff --git a/docs/configuration.en.md b/docs/configuration.en.md index 522b6673..a6dc2af0 100644 --- a/docs/configuration.en.md +++ b/docs/configuration.en.md @@ -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. | diff --git a/docs/configuration.zh-CN.md b/docs/configuration.zh-CN.md index 3e27d6cc..34edc6ea 100644 --- a/docs/configuration.zh-CN.md +++ b/docs/configuration.zh-CN.md @@ -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 轮询间隔。 | diff --git a/internal/app/phone/registry.go b/internal/app/phone/registry.go index d464f2fd..c6fcdc80 100644 --- a/internal/app/phone/registry.go +++ b/internal/app/phone/registry.go @@ -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 { - r.removeLocked(id, e, false) - } - default: - if nowUnix-int64(e.call.Date) > 2*ringTimeoutSec { - r.removeLocked(id, e, true) - } + if e.call.Terminal() && nowUnix-int64(e.call.DiscardedAt) > tombstoneTTLSec { + r.removeLocked(id, e, false) } } } 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) } diff --git a/internal/app/phone/service.go b/internal/app/phone/service.go index 8420cd4f..3717de7a 100644 --- a/internal/app/phone/service.go +++ b/internal/app/phone/service.go @@ -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 } diff --git a/internal/app/phone/service_test.go b/internal/app/phone/service_test.go index 8f99239d..c7e1a7be 100644 --- a/internal/app/phone/service_test.go +++ b/internal/app/phone/service_test.go @@ -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") + } } diff --git a/internal/config/config.go b/internal/config/config.go index ff79edf9..0035f3c4 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -319,6 +319,8 @@ type Config struct { CallTombstoneTTL time.Duration // CallMaxActivePerUser 是单用户并发非终态通话上限。 CallMaxActivePerUser int + // CallRegistryMaxEntries 是进程内通话 registry 的全局硬上限。 + CallRegistryMaxEntries int // CallSignalingMaxBytes 是 phone.sendSignalingData 单条载荷上限。 CallSignalingMaxBytes int // CallSignalingRate 是单通话每秒信令转发上限（超限静默丢弃）。 @@ -618,12 +620,13 @@ func Load() (Config, error) { UploadInFlightMaxParts: envIntOr("TELESRV_UPLOAD_INFLIGHT_MAX_PARTS", 8000), UploadInFlightMaxFiles: envIntOr("TELESRV_UPLOAD_INFLIGHT_MAX_FILES", 64), - 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), - CallSignalingMaxBytes: envIntOr("TELESRV_CALL_SIGNALING_MAX_BYTES", 65536), - CallSignalingRate: envIntOr("TELESRV_CALL_SIGNALING_RATE", 50), - CallExpiryInterval: envDurationOr("TELESRV_CALL_EXPIRY_INTERVAL", time.Second), + 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), PremiumGrantMonths: envIntOr("TELESRV_PREMIUM_GRANT_MONTHS", 3), PasskeyRPID: envOr("TELESRV_PASSKEY_RP_ID", "telesrv.net"), diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 79fa69b6..64a345d1 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -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) { diff --git a/internal/rpc/phone_rpc_test.go b/internal/rpc/phone_rpc_test.go index f96e579f..825bc351 100644 --- a/internal/rpc/phone_rpc_test.go +++ b/internal/rpc/phone_rpc_test.go @@ -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) {