fix: sync correct collectible username projection
This commit is contained in:
iamxvbaba
parent
8356989e01
commit
9041abd3d3
42 changed files with 601 additions and 758 deletions
|
|
@ -5,14 +5,11 @@ import (
|
|||
"errors"
|
||||
"reflect"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"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"
|
||||
|
|
@ -65,262 +62,8 @@ func TestDispatchMarksSessionReceivesUpdates(t *testing.T) {
|
|||
if sessions.sessionID != 42 {
|
||||
t.Fatalf("marked session_id = %d, want 42", sessions.sessionID)
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
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{}}
|
||||
users := &selfCountingUsersService{staticUsersService: staticUsersService{user: self}}
|
||||
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
|
||||
}
|
||||
|
||||
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 {
|
||||
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 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 != "" {
|
||||
t.Fatalf("self name refresh = %T %+v, want updateUserName(%d, Alice)", updates.Updates[0], updates.Updates[0], userID)
|
||||
}
|
||||
wantUsernames := []string{"Alice", "aliceCollect0728b", "aliceCollect0728a"}
|
||||
if !reflect.DeepEqual(usernameStrings(nameRefresh.Usernames), wantUsernames) {
|
||||
t.Fatalf("self updateUserName usernames = %v, want %v", usernameStrings(nameRefresh.Usernames), wantUsernames)
|
||||
}
|
||||
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()
|
||||
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)
|
||||
}
|
||||
|
||||
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 || 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)
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
if sessions.message != nil {
|
||||
t.Fatalf("session readiness emitted unsolicited update %T; readiness must only open delivery", sessions.message)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue