diff --git a/internal/mtprotoedge/session_manager.go b/internal/mtprotoedge/session_manager.go index 07dbec49..cd7691f1 100644 --- a/internal/mtprotoedge/session_manager.go +++ b/internal/mtprotoedge/session_manager.go @@ -1307,6 +1307,24 @@ func (m *SessionManager) ReceivesUpdatesForAuthKey(authKeyID [8]byte, sessionID return hasProfile && c.receivesUpdates.Load() && c.membershipsSynced.Load() } +// UpdatesActivationStartedForAuthKey reports whether the session has completed +// membership synchronization and either became ready or is draining its older +// pending FIFO. Unlike ReceivesUpdatesForAuthKey, flushing is a positive result: +// another delivery callback must not rebuild memberships or enqueue a duplicate +// cache-convergence update behind the same pending batch. +func (m *SessionManager) UpdatesActivationStartedForAuthKey(authKeyID [8]byte, sessionID int64) bool { + m.mu.RLock() + key := sessionKey{authKeyID: authKeyID, sessionID: sessionID} + c, ok := m.bySession[key] + flushing := m.flushing[key] + m.mu.RUnlock() + if !ok { + return false + } + _, hasProfile := c.LayerProfile() + return hasProfile && c.membershipsSynced.Load() && (c.receivesUpdates.Load() || flushing) +} + // SetReceivesUpdatesForAuthKey 标记指定 raw auth_key_id + session_id 是否接收主动 updates。 func (m *SessionManager) SetReceivesUpdatesForAuthKey(authKeyID [8]byte, sessionID int64, receives bool) { m.mu.Lock() diff --git a/internal/mtprotoedge/session_ready_test.go b/internal/mtprotoedge/session_ready_test.go index dc0a2725..38c9f7f4 100644 --- a/internal/mtprotoedge/session_ready_test.go +++ b/internal/mtprotoedge/session_ready_test.go @@ -26,11 +26,36 @@ func TestReceivesUpdatesForAuthKeyRequiresMembershipSync(t *testing.T) { if sm.ReceivesUpdatesForAuthKey(raw, 42) { t.Fatal("ready before membership sync — a failed sync would never be retried") } + if sm.UpdatesActivationStartedForAuthKey(raw, 42) { + t.Fatal("activation started before membership sync") + } sm.SetSessionChannelMemberships(raw, 42, 100, []int64{7}, sm.ChannelMembershipGeneration(raw, 42)) if !sm.ReceivesUpdatesForAuthKey(raw, 42) { t.Fatal("not ready after successful membership sync") } + if !sm.UpdatesActivationStartedForAuthKey(raw, 42) { + t.Fatal("activation not started after successful membership sync") + } + + // Pending FIFO drain counts as an activation already in progress even though + // the session is not fully ready yet. A stale RPC callback must not enqueue a + // duplicate cache-convergence update behind that same FIFO. + key := sessionKey{authKeyID: raw, sessionID: 42} + sm.mu.Lock() + c.receivesUpdates.Store(false) + sm.flushing[key] = true + sm.mu.Unlock() + if sm.ReceivesUpdatesForAuthKey(raw, 42) { + t.Fatal("flushing session reported fully ready") + } + if !sm.UpdatesActivationStartedForAuthKey(raw, 42) { + t.Fatal("flushing session did not report activation started") + } + sm.mu.Lock() + delete(sm.flushing, key) + c.receivesUpdates.Store(true) + sm.mu.Unlock() // 没有任何频道的账号：空列表的成功同步同样算就绪。 sm.SetSessionChannelMemberships(raw, 42, 100, nil, sm.ChannelMembershipGeneration(raw, 42)) diff --git a/internal/rpc/account.go b/internal/rpc/account.go index 49a9407f..d17ad220 100644 --- a/internal/rpc/account.go +++ b/internal/rpc/account.go @@ -1962,7 +1962,7 @@ func (r *Router) pushSelfUserChangedUpdate(ctx context.Context, u domain.User) { } r.pushUserUpdates(ctx, u.ID, &tg.Updates{ Updates: []tg.UpdateClass{&tg.UpdateUser{UserID: u.ID}}, - Users: []tg.UserClass{r.tgSelfUserWithReadModels(ctx, u)}, + Users: []tg.UserClass{r.tgSelfUserWithUsernames(ctx, u)}, Date: int(r.clock.Now().Unix()), }) } diff --git a/internal/rpc/auth.go b/internal/rpc/auth.go index 5fa52603..db278be0 100644 --- a/internal/rpc/auth.go +++ b/internal/rpc/auth.go @@ -319,7 +319,7 @@ func (r *Router) authLoginTokenSuccess(ctx context.Context, a domain.Authorizati return nil, internalErr() } return &tg.AuthLoginTokenSuccess{ - Authorization: &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(ctx, u)}, + Authorization: &tg.AuthAuthorization{User: r.tgSelfUserWithUsernames(ctx, u)}, }, nil } @@ -528,7 +528,7 @@ func (r *Router) finishAuthSignIn(ctx context.Context, u domain.User, needSignUp } r.bindSessionUser(ctx, u.ID) r.pushSignInServiceNotificationToOthers(ctx, u) - return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(ctx, u)}, nil + return &tg.AuthAuthorization{User: r.tgSelfUserWithUsernames(ctx, u)}, nil } func (r *Router) onAuthResendCode(ctx context.Context, req *tg.AuthResendCodeRequest) (tg.AuthSentCodeClass, error) { @@ -637,7 +637,7 @@ func (r *Router) onAuthCheckPassword(ctx context.Context, password tg.InputCheck if err != nil { return nil, internalErr() } - return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(ctx, u)}, nil + return &tg.AuthAuthorization{User: r.tgSelfUserWithUsernames(ctx, u)}, nil } func (r *Router) onAuthRequestPasswordRecovery(ctx context.Context) (*tg.AuthPasswordRecovery, error) { @@ -678,7 +678,7 @@ func (r *Router) onAuthRecoverPassword(ctx context.Context, req *tg.AuthRecoverP if err != nil { return nil, internalErr() } - return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(ctx, u)}, nil + return &tg.AuthAuthorization{User: r.tgSelfUserWithUsernames(ctx, u)}, nil } func (r *Router) onAuthCheckRecoveryPassword(ctx context.Context, code string) (bool, error) { @@ -809,7 +809,7 @@ func (r *Router) onAuthFinishPasskeyLogin(ctx context.Context, req *tg.AuthFinis r.setAuthUserCache(id, u.ID, true) } r.bindSessionUser(ctx, u.ID) - return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(ctx, u)}, nil + return &tg.AuthAuthorization{User: r.tgSelfUserWithUsernames(ctx, u)}, nil } func emailVerificationCode(v tg.EmailVerificationClass) string { @@ -840,7 +840,7 @@ func (r *Router) onAuthImportBotAuthorization(ctx context.Context, req *tg.AuthI r.setAuthUserCache(id, u.ID, true) } r.bindSessionUser(ctx, u.ID) - return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(ctx, u)}, nil + return &tg.AuthAuthorization{User: r.tgSelfUserWithUsernames(ctx, u)}, nil } // onAuthSignUp 处理 auth.signUp：创建用户并绑定授权。 @@ -854,7 +854,7 @@ func (r *Router) onAuthSignUp(ctx context.Context, req *tg.AuthSignUpRequest) (t } r.bindSessionUser(ctx, u.ID) r.enqueueLoginMessageBootstrap(ctx, loginMessage) - return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(ctx, u)}, nil + return &tg.AuthAuthorization{User: r.tgSelfUserWithUsernames(ctx, u)}, nil } // onAuthLogOut 处理 auth.logOut：解绑当前 auth_key 的授权。 diff --git a/internal/rpc/collectible_usernames_rpc_test.go b/internal/rpc/collectible_usernames_rpc_test.go index cfa4b004..52f96088 100644 --- a/internal/rpc/collectible_usernames_rpc_test.go +++ b/internal/rpc/collectible_usernames_rpc_test.go @@ -430,6 +430,10 @@ func TestAuthLoginTokenSuccessProjectsCompleteSelfUsernames(t *testing.T) { t.Fatalf("decoded authorization usernames = %v (set %v), want %v", usernameStrings(decodedVector), decodedSet, want) } + if registry.peerCalls != 1 || registry.batchCalls != 0 { + t.Fatalf("authorization username reads = peer:%d batch:%d, want 1/0", + registry.peerCalls, registry.batchCalls) + } } func TestMessageEchoProjectsCompleteUsernamesInOneBatch(t *testing.T) { diff --git a/internal/rpc/deps.go b/internal/rpc/deps.go index 6cda9794..8d827ef9 100644 --- a/internal/rpc/deps.go +++ b/internal/rpc/deps.go @@ -93,10 +93,12 @@ type ImmediateSessionPusher interface { PushToSessionForAuthKeyImmediate(ctx context.Context, rawAuthKeyID [8]byte, sessionID int64, t proto.MessageType, msg tg.UpdatesClass) error } -// SessionUpdatesStateProvider 暴露连接当前的 updates 接收状态（可选能力）。 -// 用于按 RPC 置位 receivesUpdates 时的幂等短路；不实现时每次都走完整置位（幂等，仅多余开销）。 +// SessionUpdatesStateProvider 暴露连接当前的 updates 激活状态（可选能力）。 +// started 同时覆盖 ready 与 pending FIFO 正在 flushing；两者都证明当前物理 session +// 已完成 membership 建立并开始激活，后续陈旧 post-response callback 不得重复查库或推送。 type SessionUpdatesStateProvider interface { ReceivesUpdatesForAuthKey(rawAuthKeyID [8]byte, sessionID int64) bool + UpdatesActivationStartedForAuthKey(rawAuthKeyID [8]byte, sessionID int64) bool } // ClientLayerBinder 把协商 TL layer 即时下推到连接（可选能力）。 diff --git a/internal/rpc/premium_sweeper.go b/internal/rpc/premium_sweeper.go index 7c3ee3b8..29d89f36 100644 --- a/internal/rpc/premium_sweeper.go +++ b/internal/rpc/premium_sweeper.go @@ -160,7 +160,7 @@ func (r *Router) pushPremiumStatusUpdate(ctx context.Context, u domain.User) { defer cancel() r.pushUserUpdates(pushCtx, u.ID, &tg.Updates{ Updates: []tg.UpdateClass{&tg.UpdateUser{UserID: u.ID}}, - Users: []tg.UserClass{r.tgSelfUserWithReadModels(pushCtx, u)}, + Users: []tg.UserClass{r.tgSelfUserWithUsernames(pushCtx, u)}, Date: int(r.clock.Now().Unix()), }) } diff --git a/internal/rpc/receives_updates_mark_test.go b/internal/rpc/receives_updates_mark_test.go index 1051ad6f..e8785d7d 100644 --- a/internal/rpc/receives_updates_mark_test.go +++ b/internal/rpc/receives_updates_mark_test.go @@ -5,7 +5,9 @@ import ( "errors" "reflect" "sync" + "sync/atomic" "testing" + "time" "github.com/iamxvbaba/td/bin" "github.com/iamxvbaba/td/clock" @@ -69,10 +71,24 @@ type updatesStateCaptureSessions struct { *captureSessions } +type selfCountingUsersService struct { + staticUsersService + selfCalls atomic.Int32 +} + +func (s *selfCountingUsersService) Self(ctx context.Context, userID int64) (domain.User, error) { + s.selfCalls.Add(1) + return s.staticUsersService.Self(ctx, userID) +} + func (s *updatesStateCaptureSessions) ReceivesUpdatesForAuthKey([8]byte, int64) bool { return s.snapshot().receives } +func (s *updatesStateCaptureSessions) UpdatesActivationStartedForAuthKey([8]byte, int64) bool { + return s.snapshot().receives +} + func TestDispatchPushesCompleteSelfProfileOnceWhenSessionBecomesReady(t *testing.T) { const ( userID = int64(1000000311) @@ -93,9 +109,10 @@ func TestDispatchPushesCompleteSelfProfileOnceWhenSessionBecomesReady(t *testing {Username: "aliceCollect0728a", Active: true, SortOrder: 2, CollectibleID: 1}, } sessions := &updatesStateCaptureSessions{captureSessions: &captureSessions{}} + users := &selfCountingUsersService{staticUsersService: staticUsersService{user: self}} r := New(Config{DC: 2, IP: "127.0.0.1", Port: 2398}, Deps{ Sessions: sessions, - Users: staticUsersService{user: self}, + Users: users, Usernames: registry, }, zaptest.NewLogger(t), clock.System) @@ -113,10 +130,12 @@ func TestDispatchPushesCompleteSelfProfileOnceWhenSessionBecomesReady(t *testing } ctx := dispatch() + staleCtx := dispatch() // staged before the first delivery makes the session ready if got := sessions.snapshot(); got.receives || got.sessionPushCalls != 0 { t.Fatalf("pre-delivery readiness = receives:%v pushes:%d, want false/0", got.receives, got.sessionPushCalls) } postresponse.Run(ctx) + postresponse.Run(staleCtx) got := sessions.snapshot() if !got.receives || got.receivesCalls != 1 || got.sessionPushCalls != 1 { @@ -124,8 +143,8 @@ func TestDispatchPushesCompleteSelfProfileOnceWhenSessionBecomesReady(t *testing got.receives, got.receivesCalls, got.sessionPushCalls) } updates, ok := got.message.(*tg.Updates) - if !ok || len(updates.Updates) != 2 || len(updates.Users) != 1 { - t.Fatalf("self refresh = %T %+v, want two updates and one user", got.message, got.message) + if !ok || len(updates.Updates) != 1 || len(updates.Users) != 1 { + t.Fatalf("self refresh = %T %+v, want one updateUserName and one user", got.message, got.message) } nameRefresh, ok := updates.Updates[0].(*tg.UpdateUserName) if !ok || nameRefresh.UserID != userID || nameRefresh.FirstName != "Alice" || nameRefresh.LastName != "" { @@ -135,10 +154,6 @@ func TestDispatchPushesCompleteSelfProfileOnceWhenSessionBecomesReady(t *testing if !reflect.DeepEqual(usernameStrings(nameRefresh.Usernames), wantUsernames) { t.Fatalf("self updateUserName usernames = %v, want %v", usernameStrings(nameRefresh.Usernames), wantUsernames) } - refresh, ok := updates.Updates[1].(*tg.UpdateUser) - if !ok || refresh.UserID != userID { - t.Fatalf("self refresh update = %T %+v, want updateUser(%d)", updates.Updates[1], updates.Updates[1], userID) - } projected, ok := updates.Users[0].(*tg.User) if !ok { t.Fatalf("self refresh user = %T, want *tg.User", updates.Users[0]) @@ -154,37 +169,130 @@ func TestDispatchPushesCompleteSelfProfileOnceWhenSessionBecomesReady(t *testing t.Fatalf("self refresh seq = %d, want 0", updates.Seq) } - var wire bin.Buffer - if err := tlprofile.EncodeObject(tlprofile.Profile228, updates, &wire); err != nil { - t.Fatalf("encode Layer 228 self refresh: %v", err) - } - decoded, err := tlprofile.DecodeObject(tlprofile.Profile228, &bin.Buffer{Buf: wire.Copy()}, tlprofile.Limits{}) - if err != nil { - t.Fatalf("decode Layer 228 self refresh: %v", err) - } - decodedUpdates, ok := decoded.(*tg.Updates) - if !ok || len(decodedUpdates.Updates) != 2 || len(decodedUpdates.Users) != 1 { - t.Fatalf("decoded Layer 228 self refresh = %T %+v", decoded, decoded) - } - decodedNameRefresh, ok := decodedUpdates.Updates[0].(*tg.UpdateUserName) - if !ok || !reflect.DeepEqual(usernameStrings(decodedNameRefresh.Usernames), wantUsernames) { - t.Fatalf("decoded Layer 228 updateUserName = %T usernames=%v, want %v", - decodedUpdates.Updates[0], usernameStrings(decodedNameRefresh.Usernames), wantUsernames) - } - decodedUser := decodedUpdates.Users[0].(*tg.User) - decodedVector, decodedSet := decodedUser.GetUsernames() - if !decodedSet || !reflect.DeepEqual(usernameStrings(decodedVector), wantUsernames) { - t.Fatalf("decoded Layer 228 usernames = %v (set %v), want %v", - usernameStrings(decodedVector), decodedSet, wantUsernames) + for _, profile := range []tlprofile.Profile{ + tlprofile.Profile225, + tlprofile.Profile226, + tlprofile.Profile227, + tlprofile.Profile228, + } { + var wire bin.Buffer + if err := tlprofile.EncodeObject(profile, updates, &wire); err != nil { + t.Fatalf("encode Layer %d self refresh: %v", profile, err) + } + decoded, err := tlprofile.DecodeObject(profile, &bin.Buffer{Buf: wire.Copy()}, tlprofile.Limits{}) + if err != nil { + t.Fatalf("decode Layer %d self refresh: %v", profile, err) + } + decodedUpdates, ok := decoded.(*tg.Updates) + if !ok || len(decodedUpdates.Updates) != 1 || len(decodedUpdates.Users) != 1 { + t.Fatalf("decoded Layer %d self refresh = %T %+v", profile, decoded, decoded) + } + decodedNameRefresh, ok := decodedUpdates.Updates[0].(*tg.UpdateUserName) + if !ok || !reflect.DeepEqual(usernameStrings(decodedNameRefresh.Usernames), wantUsernames) { + t.Fatalf("decoded Layer %d updateUserName = %T usernames=%v, want %v", + profile, decodedUpdates.Updates[0], usernameStrings(decodedNameRefresh.Usernames), wantUsernames) + } + decodedUser := decodedUpdates.Users[0].(*tg.User) + decodedVector, decodedSet := decodedUser.GetUsernames() + if !decodedSet || !reflect.DeepEqual(usernameStrings(decodedVector), wantUsernames) { + t.Fatalf("decoded Layer %d usernames = %v (set %v), want %v", + profile, usernameStrings(decodedVector), decodedSet, wantUsernames) + } } // A session that is already fully ready must not receive the bootstrap again // on every ordinary RPC. postresponse.Run(dispatch()) got = sessions.snapshot() - if got.receivesCalls != 1 || got.sessionPushCalls != 1 || registry.peerCalls != 1 { - t.Fatalf("repeat dispatch effects = ready_calls:%d pushes:%d registry_reads:%d, want 1/1/1", - got.receivesCalls, got.sessionPushCalls, registry.peerCalls) + if got.receivesCalls != 1 || got.sessionPushCalls != 1 || users.selfCalls.Load() != 1 || registry.peerCalls != 1 { + t.Fatalf("repeat dispatch effects = ready_calls:%d pushes:%d self_reads:%d registry_reads:%d, want 1/1/1/1", + got.receivesCalls, got.sessionPushCalls, users.selfCalls.Load(), registry.peerCalls) + } +} + +type blockingActivationSessions struct { + *updatesStateCaptureSessions + setEntered chan struct{} + releaseSet chan struct{} + blockOnce sync.Once +} + +func (s *blockingActivationSessions) SetReceivesUpdatesForAuthKey(rawAuthKeyID [8]byte, sessionID int64, receives bool) { + s.blockOnce.Do(func() { + close(s.setEntered) + <-s.releaseSet + }) + s.captureSessions.SetReceivesUpdatesForAuthKey(rawAuthKeyID, sessionID, receives) +} + +func TestConcurrentDeliveredRPCsClaimSessionActivationOnce(t *testing.T) { + const ( + userID = int64(1000000313) + sessionID = int64(313) + ) + rawAuthKeyID := [8]byte{33} + peer := domain.Peer{Type: domain.PeerTypeUser, ID: userID} + registry := newFakeUsernameRegistry() + registry.byPeer[peer] = []domain.Username{ + {Username: "Alice", Active: true, Editable: true, SortOrder: 0}, + {Username: "aliceCollect0728b", Active: true, SortOrder: 1, CollectibleID: 2}, + } + sessions := &blockingActivationSessions{ + updatesStateCaptureSessions: &updatesStateCaptureSessions{captureSessions: &captureSessions{}}, + setEntered: make(chan struct{}), + releaseSet: make(chan struct{}), + } + users := &selfCountingUsersService{staticUsersService: staticUsersService{user: domain.User{ + ID: userID, FirstName: "Alice", Username: "Alice", + }}} + r := New(Config{DC: 2, IP: "127.0.0.1", Port: 2398}, Deps{ + Sessions: sessions, + Users: users, + Usernames: registry, + }, zaptest.NewLogger(t), clock.System) + + dispatch := func() context.Context { + t.Helper() + var in bin.Buffer + if err := (&tg.HelpGetConfigRequest{}).Encode(&in); err != nil { + t.Fatalf("encode help.getConfig: %v", err) + } + ctx := postresponse.WithCallbacks(WithUserID(context.Background(), userID)) + if _, err := r.Dispatch(ctx, rawAuthKeyID, sessionID, &in); err != nil { + t.Fatalf("dispatch help.getConfig: %v", err) + } + return ctx + } + first, second := dispatch(), dispatch() + firstDone := make(chan struct{}) + go func() { + postresponse.Run(first) + close(firstDone) + }() + <-sessions.setEntered + + secondDone := make(chan struct{}) + go func() { + postresponse.Run(second) + close(secondDone) + }() + select { + case <-secondDone: + close(sessions.releaseSet) + <-firstDone + t.Fatal("second activation callback overtook the in-flight activation") + case <-time.After(50 * time.Millisecond): + // The waiter shares the owner's activation result, preserving the later + // bootstrap phase ordering without repeating membership work. + } + close(sessions.releaseSet) + <-firstDone + <-secondDone + + got := sessions.snapshot() + if !got.receives || got.receivesCalls != 1 || got.sessionPushCalls != 1 || users.selfCalls.Load() != 1 || registry.peerCalls != 1 { + t.Fatalf("concurrent activation effects = receives:%v ready_calls:%d pushes:%d self_reads:%d registry_reads:%d, want true/1/1/1/1", + got.receives, got.receivesCalls, got.sessionPushCalls, users.selfCalls.Load(), registry.peerCalls) } } diff --git a/internal/rpc/router.go b/internal/rpc/router.go index badcc97c..5ff3f843 100644 --- a/internal/rpc/router.go +++ b/internal/rpc/router.go @@ -149,6 +149,13 @@ type Router struct { // online 过滤前），按 userID 短 TTL；零值 sync.Map 即可用，无需构造器初始化。候选集变动 // 很慢（加好友/开新私聊），短 TTL 内复用避免 updateStatus 每次重跑 ~25-30 条 hydration 查询。 presenceCandidateCache sync.Map // userID(int64) -> *presenceCandidateEntry + // updatesReadySF serializes the delivery-gated activation of one physical + // raw auth-key + session. Cold-start RPCs are concurrent and can all be staged + // before the first rpc_result is delivered; only the claim owner may rebuild + // memberships and emit the one-shot self cache-convergence update. Waiters + // share completion so their later bootstrap phase cannot overtake activation; + // SessionUpdatesStateProvider retains ready/flushing for stale callbacks. + updatesReadySF singleflight.Group // botStatus 永久缓存 userID->是否 bot。bot 标志按账号不可变（BotFather 注册即定，普通用户永不变 bot）， // 故可无 TTL 缓存。userIsBot 在 PFS 连接上被 announceSessionOnline 每 RPC 调用，不缓存则每次一发 diff --git a/internal/rpc/self_profile_refresh.go b/internal/rpc/self_profile_refresh.go index 0a6918b5..d1ca31c3 100644 --- a/internal/rpc/self_profile_refresh.go +++ b/internal/rpc/self_profile_refresh.go @@ -10,22 +10,24 @@ import ( "telesrv/internal/domain" ) -// tgSelfUserWithReadModels is the single-object response-boundary projection -// used by authorization results and self-profile updates. tgSelfUser itself is -// intentionally a pure domain -> TL conversion because many list paths call it -// in loops; the read-model pass belongs here, where it stays one query per -// response. -func (r *Router) tgSelfUserWithReadModels(ctx context.Context, u domain.User) *tg.User { +// tgSelfUserWithUsernames is the narrow single-object projection used by +// authorization results and self-profile updates. Those constructors already +// carry a User, so it must include the complete username registry vector and +// never downgrade a client's collectible usernames to the legacy scalar. Story +// and bot-verification read models are unrelated and deliberately not queried. +func (r *Router) tgSelfUserWithUsernames(ctx context.Context, u domain.User) *tg.User { self := r.tgSelfUser(u) users := []tg.UserClass{self} - r.applyPeerReadModels(ctx, u.ID, users, nil) + r.applyUsernamesToPeerObjects(ctx, users, nil) return self } // pushUpdatesReadySelfProfile repairs the current session's cached self user -// when it first becomes eligible for proactive updates. Telegram's updateUser -// contract requires the complete User to be carried by the outer updates.users -// vector; TDLib applies that vector before invalidating userFull for updateUser. +// when it first becomes eligible for proactive updates. updateUserName is the +// authoritative basic-name/username cache write in TDLib and DrKLO. A companion +// username-complete User remains in updates.users so DrKLO's normal peer merge +// persists the vector, but no updateUser is sent because it only invalidates +// userFull (and makes DrKLO clear photo caches). // // This is an ephemeral cache-convergence update: it allocates no PTS, writes no // durable update event and targets only the physical session that just became @@ -69,30 +71,17 @@ func (r *Router) updatesReadySelfProfile(ctx context.Context, userID int64) (*tg applyUsernamesFromRegistry(users, nil, map[domain.Peer][]domain.Username{peer: list}) } } - // These read models are optional projections. Their existing response-boundary - // contract degrades independently; the username registry above is handled - // strictly because losing that vector is the cache corruption fixed here. - r.applyStoryMaxIDsToPeerObjects(ctx, userID, users, nil) - r.applyBotVerificationIconsToPeerObjects(ctx, users, nil) - usernames := tgUsernames(u.Username) if vector, ok := self.GetUsernames(); ok && len(vector) != 0 { usernames = vector } return &tg.Updates{ - Updates: []tg.UpdateClass{ - // updateUserName is the authoritative basic-name/username cache write. - // TDLib handles it by calling on_update_user_usernames directly. - &tg.UpdateUserName{ - UserID: userID, - FirstName: u.FirstName, - LastName: u.LastName, - Usernames: usernames, - }, - // updateUser additionally invalidates userFull while the complete basic - // User remains bundled in the outer users vector. - &tg.UpdateUser{UserID: userID}, - }, + Updates: []tg.UpdateClass{&tg.UpdateUserName{ + UserID: userID, + FirstName: u.FirstName, + LastName: u.LastName, + Usernames: usernames, + }}, Users: users, Date: int(r.clock.Now().Unix()), Seq: 0, diff --git a/internal/rpc/updates.go b/internal/rpc/updates.go index eab8fd6a..b292ed50 100644 --- a/internal/rpc/updates.go +++ b/internal/rpc/updates.go @@ -210,26 +210,39 @@ func (r *Router) maybeMarkSessionReceivesUpdates(ctx context.Context) { if !ok { return } - if provider, ok := r.deps.Sessions.(SessionUpdatesStateProvider); ok { - rawAuthKeyID, okRaw := RawAuthKeyIDFrom(ctx) - sessionID, okSess := SessionIDFrom(ctx) - if okRaw && okSess && provider.ReceivesUpdatesForAuthKey(rawAuthKeyID, sessionID) { - return - } + if r.sessionUpdatesActivationStarted(ctx) { + return } r.stageSessionUpdatesReadyAfterDelivery(ctx, userID) } -func (r *Router) markSessionReceivesUpdatesNow(ctx context.Context, userID int64) { +func (r *Router) sessionUpdatesActivationStarted(ctx context.Context) bool { + provider, ok := r.deps.Sessions.(SessionUpdatesStateProvider) + if !ok { + return false + } + rawAuthKeyID, okRaw := RawAuthKeyIDFrom(ctx) + sessionID, okSession := SessionIDFrom(ctx) + return okRaw && okSession && provider.UpdatesActivationStartedForAuthKey(rawAuthKeyID, sessionID) +} + +func (r *Router) markSessionReceivesUpdatesNow(ctx context.Context, userID int64) bool { if r.deps.Sessions == nil { - return + return false } r.syncSessionChannelMemberships(ctx, userID) sessionID, ok := SessionIDFrom(ctx) if !ok { - return + return false } r.deps.Sessions.SetReceivesUpdatesForAuthKey(rawAuthKeyIDForOrigin(ctx), sessionID, true) + if _, ok := r.deps.Sessions.(SessionUpdatesStateProvider); !ok { + // Alternate lightweight binders cannot expose flushing/membership state. + // Preserve their historical behavior; the Router claim still collapses + // callbacks which overlap this activation attempt. + return true + } + return r.sessionUpdatesActivationStarted(ctx) } func ptr[T any](v T) *T { return &v } diff --git a/internal/rpc/updates_delivery.go b/internal/rpc/updates_delivery.go index c723327b..d4932303 100644 --- a/internal/rpc/updates_delivery.go +++ b/internal/rpc/updates_delivery.go @@ -2,6 +2,7 @@ package rpc import ( "context" + "strconv" "time" "go.uber.org/zap" @@ -202,7 +203,7 @@ func (r *Router) registerUpdatesDeliveryPlan(ctx context.Context, plan *updatesD // 1. commit the exact account cursor carried by the delivered result; // 2. the just-delivered difference may retire its projected secret-chat events; // 3. membership routing is rebuilt before SetReceivesUpdates starts FIFO flush; -// 4. the current session receives one complete self profile after becoming ready; +// 4. one claim owner activates the session and emits its username convergence update; // 5. bootstrap jobs are published last, so they queue behind older pending updates. // // Each phase gets an independent timeout so one failed side effect cannot starve @@ -236,15 +237,53 @@ func (r *Router) runUpdatesDeliveryPlan(plan updatesDeliveryPlan) { } } if plan.markSessionReady { - ctx, cancel := context.WithTimeout(baseCtx, updatesDeliveryPhaseTimeout) - r.markSessionReceivesUpdatesNow(ctx, plan.readyUserID) - cancel() - - ctx, cancel = context.WithTimeout(baseCtx, updatesDeliveryPhaseTimeout) - r.pushUpdatesReadySelfProfile(ctx, plan.readyUserID) - cancel() + r.activateSessionUpdates(baseCtx, plan.readyUserID) } if plan.publishBootstrap { r.publishBootstrapAfterBaseline(baseCtx, plan.bootstrapUserID) } } + +func (r *Router) activateSessionUpdates(baseCtx context.Context, userID int64) { + rawAuthKeyID, okRaw := RawAuthKeyIDFrom(baseCtx) + sessionID, okSession := SessionIDFrom(baseCtx) + if userID == 0 { + return + } + if !okRaw || !okSession { + // Direct handler tests and lightweight embeddings may not carry the full + // physical identity. They cannot benefit from a cross-request claim, but + // must retain the historical delivery-gated activation behavior. + ctx, cancel := context.WithTimeout(baseCtx, updatesDeliveryPhaseTimeout) + started := r.markSessionReceivesUpdatesNow(ctx, userID) + cancel() + if started { + ctx, cancel = context.WithTimeout(baseCtx, updatesDeliveryPhaseTimeout) + r.pushUpdatesReadySelfProfile(ctx, userID) + cancel() + } + return + } + if r.sessionUpdatesActivationStarted(baseCtx) { + return + } + key := string(rawAuthKeyID[:]) + strconv.FormatInt(sessionID, 10) + _, _, _ = r.updatesReadySF.Do(key, func() (any, error) { + // A callback may have been staged before another callback completed the + // same activation. Re-check under singleflight before touching storage. + if r.sessionUpdatesActivationStarted(baseCtx) { + return nil, nil + } + ctx, cancel := context.WithTimeout(baseCtx, updatesDeliveryPhaseTimeout) + started := r.markSessionReceivesUpdatesNow(ctx, userID) + cancel() + if !started { + return nil, nil + } + + ctx, cancel = context.WithTimeout(baseCtx, updatesDeliveryPhaseTimeout) + r.pushUpdatesReadySelfProfile(ctx, userID) + cancel() + return nil, nil + }) +}