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

fix: sync refresh complete self profile on session ready

Browse source
This commit is contained in:
iamxvbaba 2026-08-02 19:32:17 +08:00
parent 2512eab51d
commit f67e87689c
8 changed files with 405 additions and 28 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.tgSelfUser(u)},
Users: []tg.UserClass{r.tgSelfUserWithReadModels(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.tgSelfUser(u)},
Authorization: &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(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.tgSelfUser(u)}, nil
return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(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.tgSelfUser(u)}, nil
return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(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.tgSelfUser(u)}, nil
return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(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.tgSelfUser(u)}, nil
return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(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.tgSelfUser(u)}, nil
return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(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.tgSelfUser(u)}, nil
return &tg.AuthAuthorization{User: r.tgSelfUserWithReadModels(ctx, u)}, nil
}
// onAuthLogOut 处理 auth.logOut：解绑当前 auth_key 的授权。

144
internal/rpc/collectible_usernames_rpc_test.go
View file

@ -12,6 +12,7 @@ import (
"github.com/iamxvbaba/td/clock"
"github.com/iamxvbaba/td/tg"
"github.com/iamxvbaba/td/tgerr"
"github.com/iamxvbaba/td/tlprofile"
"go.uber.org/zap/zaptest"
appchannels "telesrv/internal/app/channels"
@ -288,6 +289,149 @@ func TestUsersGetUsersProjectsCollectibleUsernamesInOneBatch(t *testing.T) {
}
}
func TestResolveUsernamePreservesCompleteUsernamesThroughLayer228(t *testing.T) {
registry := newFakeUsernameRegistry()
f := newUsernameProjectionFixture(t, registry)
registry.byPeer[domain.Peer{Type: domain.PeerTypeUser, ID: f.owner.ID}] = []domain.Username{
{Username: "owner_slot", Editable: true, Active: true, SortOrder: 0},
{Username: "owner_collectible_b", Active: true, SortOrder: 1, CollectibleID: 22},
{Username: "owner_collectible_a", Active: true, SortOrder: 2, CollectibleID: 21},
}
resolved, err := f.router.onContactsResolveUsername(
WithUserID(context.Background(), f.friend.ID),
&tg.ContactsResolveUsernameRequest{Username: "@owner_slot"},
)
if err != nil {
t.Fatalf("resolve username: %v", err)
}
assertResolvedUsernames := func(stage string, value *tg.ContactsResolvedPeer) {
t.Helper()
if value == nil || len(value.Users) != 1 {
t.Fatalf("%s resolved users = %+v, want one user", stage, value)
}
user, ok := value.Users[0].(*tg.User)
if !ok {
t.Fatalf("%s resolved user = %T, want *tg.User", stage, value.Users[0])
}
if scalar, set := user.GetUsername(); !set || scalar != "owner_slot" {
t.Fatalf("%s scalar username = %q (set %v), want owner_slot", stage, scalar, set)
}
vector, set := user.GetUsernames()
want := []string{"owner_slot", "owner_collectible_b", "owner_collectible_a"}
if !set || !reflect.DeepEqual(usernameStrings(vector), want) {
t.Fatalf("%s usernames = %v (set %v), want %v", stage, usernameStrings(vector), set, want)
}
}
assertResolvedUsernames("canonical", resolved)
var wire bin.Buffer
if err := tlprofile.EncodeObject(tlprofile.Profile228, resolved, &wire); err != nil {
t.Fatalf("encode Layer 228 resolved peer: %v", err)
}
decoded, err := tlprofile.DecodeObject(
tlprofile.Profile228,
&bin.Buffer{Buf: wire.Copy()},
tlprofile.Limits{},
)
if err != nil {
t.Fatalf("decode Layer 228 resolved peer: %v", err)
}
decodedResolved, ok := decoded.(*tg.ContactsResolvedPeer)
if !ok {
t.Fatalf("decoded Layer 228 object = %T, want *tg.ContactsResolvedPeer", decoded)
}
assertResolvedUsernames("Layer 228", decodedResolved)
requestBody := encodeExactLayerRPC(t, tlprofile.Profile228, &tg.ContactsResolveUsernameRequest{
Username: "@owner_slot",
})
admitted, err := f.router.AdmitLayer(tlprofile.Profile228, &requestBody, tlprofile.Limits{})
if err != nil {
t.Fatalf("admit Layer 228 contacts.resolveUsername: %v", err)
}
result, method, err := f.router.DispatchAdmitted(
WithUserID(context.Background(), f.friend.ID),
[8]byte{1},
1,
1,
1,
admitted,
)
if err != nil || method != "contacts.resolveUsername" {
t.Fatalf("dispatch Layer 228 method=%q err=%v", method, err)
}
var resultWire bin.Buffer
if err := result.Encode(&resultWire); err != nil {
t.Fatalf("encode Layer 228 method result: %v", err)
}
decodedResult, err := tlprofile.DecodeObject(
tlprofile.Profile228,
&bin.Buffer{Buf: resultWire.Copy()},
tlprofile.Limits{},
)
if err != nil {
t.Fatalf("decode Layer 228 method result: %v", err)
}
methodResolved, ok := decodedResult.(*tg.ContactsResolvedPeer)
if !ok {
t.Fatalf("decoded Layer 228 method result = %T, want *tg.ContactsResolvedPeer", decodedResult)
}
assertResolvedUsernames("Layer 228 method result", methodResolved)
}
func TestAuthLoginTokenSuccessProjectsCompleteSelfUsernames(t *testing.T) {
registry := newFakeUsernameRegistry()
f := newUsernameProjectionFixture(t, registry)
registry.byPeer[domain.Peer{Type: domain.PeerTypeUser, ID: f.owner.ID}] = []domain.Username{
{Username: "owner_slot", Editable: true, Active: true, SortOrder: 0},
{Username: "owner_collectible_b", Active: true, SortOrder: 1, CollectibleID: 22},
{Username: "owner_collectible_a", Active: true, SortOrder: 2, CollectibleID: 21},
}
result, err := f.router.authLoginTokenSuccess(context.Background(), domain.Authorization{UserID: f.owner.ID})
if err != nil {
t.Fatalf("auth login token success: %v", err)
}
success, ok := result.(*tg.AuthLoginTokenSuccess)
if !ok {
t.Fatalf("auth login token result = %T, want *tg.AuthLoginTokenSuccess", result)
}
authorization, ok := success.Authorization.(*tg.AuthAuthorization)
if !ok {
t.Fatalf("authorization = %T, want *tg.AuthAuthorization", success.Authorization)
}
self, ok := authorization.User.(*tg.User)
if !ok {
t.Fatalf("authorization user = %T, want *tg.User", authorization.User)
}
want := []string{"owner_slot", "owner_collectible_b", "owner_collectible_a"}
vector, set := self.GetUsernames()
if !set || !reflect.DeepEqual(usernameStrings(vector), want) {
t.Fatalf("authorization usernames = %v (set %v), want %v", usernameStrings(vector), set, want)
}
var wire bin.Buffer
if err := tlprofile.EncodeObject(tlprofile.Profile228, success, &wire); err != nil {
t.Fatalf("encode Layer 228 login token success: %v", err)
}
decoded, err := tlprofile.DecodeObject(tlprofile.Profile228, &bin.Buffer{Buf: wire.Copy()}, tlprofile.Limits{})
if err != nil {
t.Fatalf("decode Layer 228 login token success: %v", err)
}
decodedSuccess, ok := decoded.(*tg.AuthLoginTokenSuccess)
if !ok {
t.Fatalf("decoded login token success = %T", decoded)
}
decodedAuthorization := decodedSuccess.Authorization.(*tg.AuthAuthorization)
decodedSelf := decodedAuthorization.User.(*tg.User)
decodedVector, decodedSet := decodedSelf.GetUsernames()
if !decodedSet || !reflect.DeepEqual(usernameStrings(decodedVector), want) {
t.Fatalf("decoded authorization usernames = %v (set %v), want %v",
usernameStrings(decodedVector), decodedSet, want)
}
}
func TestMessageEchoProjectsCompleteUsernamesInOneBatch(t *testing.T) {
registry := newFakeUsernameRegistry()
f := newUsernameProjectionFixture(t, registry)

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.tgSelfUser(u)},
Users: []tg.UserClass{r.tgSelfUserWithReadModels(pushCtx, u)},
Date: int(r.clock.Now().Unix()),
})
}

140
internal/rpc/receives_updates_mark_test.go
View file

@ -10,6 +10,7 @@ import (
"github.com/iamxvbaba/td/bin"
"github.com/iamxvbaba/td/clock"
"github.com/iamxvbaba/td/tg"
"github.com/iamxvbaba/td/tlprofile"
"go.uber.org/zap/zaptest"
appchannels "telesrv/internal/app/channels"
@ -64,6 +65,145 @@ func TestDispatchMarksSessionReceivesUpdates(t *testing.T) {
}
}
type updatesStateCaptureSessions struct {
*captureSessions
}
func (s *updatesStateCaptureSessions) ReceivesUpdatesForAuthKey([8]byte, int64) bool {
return s.snapshot().receives
}
func TestDispatchPushesCompleteSelfProfileOnceWhenSessionBecomesReady(t *testing.T) {
const (
userID = int64(1000000311)
sessionID = int64(311)
)
rawAuthKeyID := [8]byte{31}
self := domain.User{
ID: userID,
AccessHash: 3111,
FirstName: "Alice",
Username: "Alice",
}
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},
{Username: "aliceCollect0728a", Active: true, SortOrder: 2, CollectibleID: 1},
}
sessions := &updatesStateCaptureSessions{captureSessions: &captureSessions{}}
r := New(Config{DC: 2, IP: "127.0.0.1", Port: 2398}, Deps{
Sessions: sessions,
Users: staticUsersService{user: self},
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
}
ctx := dispatch()
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)
got := sessions.snapshot()
if !got.receives || got.receivesCalls != 1 || got.sessionPushCalls != 1 {
t.Fatalf("post-delivery readiness = receives:%v ready_calls:%d pushes:%d, want true/1/1",
got.receives, got.receivesCalls, got.sessionPushCalls)
}
updates, ok := got.message.(*tg.Updates)
if !ok || len(updates.Updates) != 1 || len(updates.Users) != 1 {
t.Fatalf("self refresh = %T %+v, want one update and one user", got.message, got.message)
}
refresh, ok := updates.Updates[0].(*tg.UpdateUser)
if !ok || refresh.UserID != userID {
t.Fatalf("self refresh update = %T %+v, want updateUser(%d)", updates.Updates[0], updates.Updates[0], userID)
}
projected, ok := updates.Users[0].(*tg.User)
if !ok {
t.Fatalf("self refresh user = %T, want *tg.User", updates.Users[0])
}
vector, set := projected.GetUsernames()
wantUsernames := []string{"Alice", "aliceCollect0728b", "aliceCollect0728a"}
if !set || !reflect.DeepEqual(usernameStrings(vector), wantUsernames) {
t.Fatalf("self refresh usernames = %v (set %v), want %v", usernameStrings(vector), set, wantUsernames)
}
if scalar, set := projected.GetUsername(); !set || scalar != "Alice" {
t.Fatalf("self refresh scalar username = %q (set %v), want Alice", scalar, set)
}
if updates.Seq != 0 {
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.Users) != 1 {
t.Fatalf("decoded Layer 228 self refresh = %T %+v", decoded, decoded)
}
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)
}
// 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)
}
}
func TestDispatchSuppressesSelfProfileWhenUsernameRegistryFails(t *testing.T) {
const userID = int64(1000000312)
registry := newFakeUsernameRegistry()
registry.err = errors.New("registry unavailable")
sessions := &updatesStateCaptureSessions{captureSessions: &captureSessions{}}
r := New(Config{DC: 2, IP: "127.0.0.1", Port: 2398}, Deps{
Sessions: sessions,
Users: staticUsersService{user: domain.User{
ID: userID, FirstName: "Alice", Username: "Alice",
}},
Usernames: registry,
}, zaptest.NewLogger(t), clock.System)
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, [8]byte{32}, 312, &in); err != nil {
t.Fatalf("dispatch help.getConfig: %v", err)
}
postresponse.Run(ctx)
got := sessions.snapshot()
if !got.receives || got.sessionPushCalls != 0 {
t.Fatalf("registry failure effects = receives:%v pushes:%d, want true/0", got.receives, got.sessionPushCalls)
}
}
// TestDispatchSkipsReceivesUpdatesForInvokeWithoutUpdates 验证 invokeWithoutUpdates
// 包装的请求（media/temp 连接）不会把该 session 标记为 updates 接收者。
func TestDispatchSkipsReceivesUpdatesForInvokeWithoutUpdates(t *testing.T) {

40
internal/rpc/rpc_testkit_sessions_test.go
View file

@ -20,6 +20,7 @@ type captureSessions struct {
authKeyResolved bool
receives bool
receivesCalls int
sessionPushCalls int
messageType proto.MessageType
message bin.Encoder
userMessage bin.Encoder // 最近一次 PushToUser* 的消息（与 message 区分：message 也被 PushToSession 覆盖）
@ -33,30 +34,32 @@ type captureSessions struct {
}
type captureSessionsSnapshot struct {
sessionID int64
userID int64
userResolved bool
authKeyID [8]byte
authKeyResolved bool
receives bool
receivesCalls int
messageType proto.MessageType
message bin.Encoder
sessionID int64
userID int64
userResolved bool
authKeyID [8]byte
authKeyResolved bool
receives bool
receivesCalls int
sessionPushCalls int
messageType proto.MessageType
message bin.Encoder
}
func (s *captureSessions) snapshot() captureSessionsSnapshot {
s.mu.Lock()
defer s.mu.Unlock()
return captureSessionsSnapshot{
sessionID: s.sessionID,
userID: s.userID,
userResolved: s.userResolved,
authKeyID: s.authKeyID,
authKeyResolved: s.authKeyResolved,
receives: s.receives,
receivesCalls: s.receivesCalls,
messageType: s.messageType,
message: s.message,
sessionID: s.sessionID,
userID: s.userID,
userResolved: s.userResolved,
authKeyID: s.authKeyID,
authKeyResolved: s.authKeyResolved,
receives: s.receives,
receivesCalls: s.receivesCalls,
sessionPushCalls: s.sessionPushCalls,
messageType: s.messageType,
message: s.message,
}
}
@ -156,6 +159,7 @@ func (s *captureSessions) PushToSessionForAuthKey(_ context.Context, rawAuthKeyI
s.sessionID = sessionID
s.messageType = t
s.message = msg
s.sessionPushCalls++
return nil
}

84
internal/rpc/self_profile_refresh.go Normal file
View file

@ -0,0 +1,84 @@
package rpc
import (
"context"
"fmt"
"github.com/iamxvbaba/td/tg"
"go.uber.org/zap"
"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 {
self := r.tgSelfUser(u)
users := []tg.UserClass{self}
r.applyPeerReadModels(ctx, u.ID, 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.
//
// 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
// ready. A username-registry read failure suppresses the refresh instead of
// replacing a possibly richer client cache with the legacy scalar-only shape.
func (r *Router) pushUpdatesReadySelfProfile(ctx context.Context, userID int64) {
updates, err := r.updatesReadySelfProfile(ctx, userID)
if err != nil {
r.log.Warn("build updates-ready self profile",
zap.Int64("user_id", userID),
zap.Error(err))
return
}
if updates == nil {
return
}
r.pushCurrentSessionMessage(ctx, "push updates-ready self profile", updates)
}
func (r *Router) updatesReadySelfProfile(ctx context.Context, userID int64) (*tg.Updates, error) {
if userID == 0 || r.deps.Users == nil {
return nil, nil
}
u, err := r.deps.Users.Self(ctx, userID)
if err != nil {
return nil, fmt.Errorf("load self user: %w", err)
}
if u.ID != userID || u.Deleted {
return nil, fmt.Errorf("invalid self user: requested %d, got %d deleted=%v", userID, u.ID, u.Deleted)
}
self := r.tgSelfUser(u)
users := []tg.UserClass{self}
if r.deps.Usernames != nil {
peer := domain.Peer{Type: domain.PeerTypeUser, ID: userID}
list, err := r.deps.Usernames.PeerUsernames(ctx, peer)
if err != nil {
return nil, fmt.Errorf("load self usernames: %w", err)
}
if len(list) != 0 {
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)
return &tg.Updates{
Updates: []tg.UpdateClass{&tg.UpdateUser{UserID: userID}},
Users: users,
Date: int(r.clock.Now().Unix()),
Seq: 0,
}, nil
}

7
internal/rpc/updates_delivery.go
View file

@ -202,7 +202,8 @@ 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. bootstrap jobs are published last, so they queue behind older pending updates.
// 4. the current session receives one complete self profile after becoming ready;
// 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
// the remaining delivery-safe transitions.
@ -238,6 +239,10 @@ func (r *Router) runUpdatesDeliveryPlan(plan updatesDeliveryPlan) {
ctx, cancel := context.WithTimeout(baseCtx, updatesDeliveryPhaseTimeout)
r.markSessionReceivesUpdatesNow(ctx, plan.readyUserID)
cancel()
ctx, cancel = context.WithTimeout(baseCtx, updatesDeliveryPhaseTimeout)
r.pushUpdatesReadySelfProfile(ctx, plan.readyUserID)
cancel()
}
if plan.publishBootstrap {
r.publishBootstrapAfterBaseline(baseCtx, plan.bootstrapUserID)