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

fix: sync streamline self username cache convergence

Browse source
This commit is contained in:
iamxvbaba 2026-08-02 19:32:18 +08:00
parent 8bc1560c16
commit 8356989e01
12 changed files with 294 additions and 89 deletions
Download patch file Download diff file Expand all files Collapse all files

2
internal/rpc/account.go
View file

@ -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()),
})
}

14
internal/rpc/auth.go
View file

@ -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 的授权。

4
internal/rpc/collectible_usernames_rpc_test.go
View file

@ -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) {

6
internal/rpc/deps.go
View file

@ -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 即时下推到连接（可选能力）。

2
internal/rpc/premium_sweeper.go
View file

@ -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()),
})
}

172
internal/rpc/receives_updates_mark_test.go
View file

@ -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)
}
}

7
internal/rpc/router.go
View file

@ -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 调用，不缓存则每次一发

47
internal/rpc/self_profile_refresh.go
View file

@ -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,

31
internal/rpc/updates.go
View file

@ -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 }

55
internal/rpc/updates_delivery.go
View file

@ -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
})
}