From 8a2e799d29224e4de94c49982cd84522e1fc480a Mon Sep 17 00:00:00 2001 From: DeliciousBuding Date: Tue, 1 Sep 2026 22:13:22 +0800 Subject: [PATCH] =?UTF-8?q?perf(presence):=20=E8=81=94=E7=B3=BB=E4=BA=BA?= =?UTF-8?q?=E5=88=97=E8=A1=A8=E4=B8=8E=E5=9C=A8=E7=BA=BF=E5=B9=BF=E6=92=AD?= =?UTF-8?q?=20presence=20=E6=89=B9=E9=87=8F=E5=8C=96=EF=BC=88Redis=20pipel?= =?UTF-8?q?ine=20=E5=8D=95=E6=AC=A1=E5=BE=80=E8=BF=94=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Cursor --- hub-server/internal/app/events.go | 8 ++- hub-server/internal/app/events_test.go | 58 +++++++++++++++++++ hub-server/internal/cache/client_noop.go | 7 ++- hub-server/internal/cache/client_routes.go | 32 ++++++++++ hub-server/internal/cache/client_test.go | 31 ++++++++++ .../internal/service/contact/service.go | 9 ++- .../internal/service/contact/service_test.go | 54 ++++++++++++++++- 7 files changed, 193 insertions(+), 6 deletions(-) diff --git a/hub-server/internal/app/events.go b/hub-server/internal/app/events.go index daf39013c..27f5a9e4f 100644 --- a/hub-server/internal/app/events.go +++ b/hub-server/internal/app/events.go @@ -670,8 +670,14 @@ func (a *App) broadcastOnlineStatus(ctx context.Context, userID string, online b } frame := ws.NewFrame(eventType, map[string]string{"user_id": userID}) + // One pipelined presence round trip for all friends instead of one per + // friend (#2154 perf lane); errors degrade to no fanout. + onlineSet, err := a.CacheClient.AreOnline(ctx, friendIDs) + if err != nil { + return + } for _, friendID := range friendIDs { - if online, _ := a.CacheClient.IsOnline(ctx, friendID); online { + if onlineSet[friendID] { a.mgr.PushToUser(friendID, frame) } } diff --git a/hub-server/internal/app/events_test.go b/hub-server/internal/app/events_test.go index e6055214b..16262e0f9 100644 --- a/hub-server/internal/app/events_test.go +++ b/hub-server/internal/app/events_test.go @@ -18,6 +18,7 @@ import ( "github.com/agenthub/hub-server/internal/cache" "github.com/agenthub/hub-server/internal/config" + "github.com/agenthub/hub-server/internal/service/contact" "github.com/agenthub/hub-server/internal/service/session" "github.com/agenthub/hub-server/internal/ws" ) @@ -160,3 +161,60 @@ func TestHandleTypingFrameRejectsNonMember(t *testing.T) { require.EqualValues(t, 1, queryCount.Load(), "admission check resolves members once") require.NoError(t, mock.ExpectationsWereMet()) } + +// TestBroadcastOnlineStatusUsesBatchedPresence proves the online/offline +// fanout resolves friend presence with one pipelined AreOnline call instead +// of one IsOnline round trip per friend (#2154 perf lane). +func TestBroadcastOnlineStatusUsesBatchedPresence(t *testing.T) { + sqlDB, mock, err := sqlmock.New() + require.NoError(t, err) + t.Cleanup(func() { _ = sqlDB.Close() }) + + gormDB, err := gorm.Open(postgres.New(postgres.Config{ + Conn: sqlDB, + PreferSimpleProtocol: true, + }), &gorm.Config{ + DisableAutomaticPing: true, + Logger: gormlogger.Default.LogMode(gormlogger.Silent), + }) + require.NoError(t, err) + + rows := sqlmock.NewRows([]string{"friend_id"}). + AddRow("friend-online"). + AddRow("friend-offline") + mock.ExpectQuery(`FROM "friendships"`).WillReturnRows(rows) + + mr := miniredis.RunT(t) + rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { _ = rdb.Close() }) + cacheClient := cache.NewClient(rdb) + + ctx := context.Background() + require.NoError(t, cacheClient.SetRoute(ctx, "friend-online", "desktop", "conn-x")) + + mgr := ws.NewManager() + onlinePeer := &ws.Conn{Send: make(chan []byte, 4)} + require.NoError(t, mgr.Register(onlinePeer)) + mgr.SetAuth(onlinePeer.ID, "friend-online", "desktop", "dev-on") + offlinePeer := &ws.Conn{Send: make(chan []byte, 4)} + require.NoError(t, mgr.Register(offlinePeer)) + mgr.SetAuth(offlinePeer.ID, "friend-offline", "desktop", "dev-off") + + a := &App{ + CacheClient: cacheClient, + ContactService: contact.NewService(gormDB, nil, cacheClient), + mgr: mgr, + } + a.broadcastOnlineStatus(ctx, "user-1", true) + + select { + case raw := <-onlinePeer.Send: + var f ws.Frame + require.NoError(t, json.Unmarshal(raw, &f)) + require.Equal(t, ws.TypeDeviceOnline, f.Type) + case <-time.After(time.Second): + t.Fatal("expected the online friend to receive the presence frame") + } + require.Empty(t, offlinePeer.Send, "offline friend must not receive the presence frame") + require.NoError(t, mock.ExpectationsWereMet()) +} diff --git a/hub-server/internal/cache/client_noop.go b/hub-server/internal/cache/client_noop.go index 06093c40d..e492a4cf6 100644 --- a/hub-server/internal/cache/client_noop.go +++ b/hub-server/internal/cache/client_noop.go @@ -16,8 +16,11 @@ import ( // *Client — passing nil will panic. type NoOpCache struct{} -func (NoOpCache) Invalidate(ctx context.Context, keys ...string) error { return nil } -func (NoOpCache) IsOnline(ctx context.Context, userID string) (bool, error) { return false, nil } +func (NoOpCache) Invalidate(ctx context.Context, keys ...string) error { return nil } +func (NoOpCache) IsOnline(ctx context.Context, userID string) (bool, error) { return false, nil } +func (NoOpCache) AreOnline(ctx context.Context, userIDs []string) (map[string]bool, error) { + return map[string]bool{}, nil +} func (NoOpCache) InitSeqIfAbsent(ctx context.Context, sessionID string, seq int64) error { return nil } func (NoOpCache) SetSeq(ctx context.Context, sessionID string, seq int64) error { return nil } func (NoOpCache) AllocateSeq(ctx context.Context, sessionID string) (int64, error) { diff --git a/hub-server/internal/cache/client_routes.go b/hub-server/internal/cache/client_routes.go index 6fc408eb6..35a688311 100644 --- a/hub-server/internal/cache/client_routes.go +++ b/hub-server/internal/cache/client_routes.go @@ -93,6 +93,38 @@ func (c *Client) IsOnline(ctx context.Context, userID string) (bool, error) { return n > 0, nil } +// AreOnline reports, for every given user, whether at least one active +// device route exists. The per-user HLEN commands are bundled into a single +// Redis pipeline round trip instead of N sequential ones — ListContacts and +// online-status fanout previously paid one round trip per friend (#2154 +// perf lane). Users missing from the result map are offline; a pipeline +// error is returned to the caller (callers today ignore it and treat the +// batch as offline, matching the per-item IsOnline convention). +func (c *Client) AreOnline(ctx context.Context, userIDs []string) (map[string]bool, error) { + online := make(map[string]bool, len(userIDs)) + if len(userIDs) == 0 { + return online, nil + } + pipe := c.rdb.Pipeline() + cmds := make([]*redis.IntCmd, 0, len(userIDs)) + ids := make([]string, 0, len(userIDs)) + for _, userID := range userIDs { + if _, dup := online[userID]; dup { + continue + } + online[userID] = false + cmds = append(cmds, pipe.HLen(ctx, routeKey(userID))) + ids = append(ids, userID) + } + if _, err := pipe.Exec(ctx); err != nil && err != redis.Nil { + return nil, err + } + for i, cmd := range cmds { + online[ids[i]] = cmd.Val() > 0 + } + return online, nil +} + // GetAllRoutes returns all device routes for a user. func (c *Client) GetAllRoutes(ctx context.Context, userID string) (map[string]string, error) { return c.rdb.HGetAll(ctx, routeKey(userID)).Result() diff --git a/hub-server/internal/cache/client_test.go b/hub-server/internal/cache/client_test.go index 2f2a34426..ab91a3e89 100644 --- a/hub-server/internal/cache/client_test.go +++ b/hub-server/internal/cache/client_test.go @@ -377,6 +377,37 @@ func TestIsOnline(t *testing.T) { assert.False(t, online) } +// TestAreOnline covers the batched presence lookup (#2154 perf lane): +// one pipelined round trip replaces one HLEN per user. +func TestAreOnline(t *testing.T) { + c, _ := testClient(t) + ctx := context.Background() + + // Empty input returns an empty map. + onlineSet, err := c.AreOnline(ctx, nil) + require.NoError(t, err) + assert.Empty(t, onlineSet) + + // friend-a has a route (online), friend-b has none (offline). + require.NoError(t, c.SetRoute(ctx, "friend-a", "desktop", "conn-1")) + onlineSet, err = c.AreOnline(ctx, []string{"friend-a", "friend-b"}) + require.NoError(t, err) + assert.True(t, onlineSet["friend-a"]) + assert.False(t, onlineSet["friend-b"]) + + // Duplicate IDs are deduplicated. + onlineSet, err = c.AreOnline(ctx, []string{"friend-a", "friend-a"}) + require.NoError(t, err) + assert.Len(t, onlineSet, 1) + assert.True(t, onlineSet["friend-a"]) + + // Route removal flips presence back to offline. + require.NoError(t, c.DeleteRoute(ctx, "friend-a", "desktop")) + onlineSet, err = c.AreOnline(ctx, []string{"friend-a"}) + require.NoError(t, err) + assert.False(t, onlineSet["friend-a"]) +} + func TestGetAllRoutes(t *testing.T) { c, _ := testClient(t) ctx := context.Background() diff --git a/hub-server/internal/service/contact/service.go b/hub-server/internal/service/contact/service.go index 1eaecb704..debcea3f9 100644 --- a/hub-server/internal/service/contact/service.go +++ b/hub-server/internal/service/contact/service.go @@ -24,6 +24,8 @@ type Bus interface { type Cache interface { Invalidate(ctx context.Context, keys ...string) error IsOnline(ctx context.Context, userID string) (bool, error) + // AreOnline batches presence lookups in one round trip (#2154 perf lane). + AreOnline(ctx context.Context, userIDs []string) (map[string]bool, error) } // Service owns contact/friendship orchestration: user search, friend request @@ -310,13 +312,18 @@ func (s *Service) ListContacts(ctx context.Context, userID string) ([]ContactInf return nil, err } + // Presence in one pipelined round trip instead of one per friend (#2154 + // perf lane); errors degrade to all-offline, matching the previous + // per-item error swallowing. + onlineSet, _ := resolveCache(s.cacheClient).AreOnline(ctx, friendIDs) + result := make([]ContactInfo, 0, len(friends)) for _, f := range friends { friend, ok := users[f.FriendID] if !ok { continue } - online, _ := resolveCache(s.cacheClient).IsOnline(ctx, friend.ID) + online := onlineSet[friend.ID] result = append(result, ContactInfo{ UserID: friend.ID, Username: friend.Username, diff --git a/hub-server/internal/service/contact/service_test.go b/hub-server/internal/service/contact/service_test.go index f0c8cb726..1ca99c417 100644 --- a/hub-server/internal/service/contact/service_test.go +++ b/hub-server/internal/service/contact/service_test.go @@ -24,8 +24,10 @@ import ( // mockContactCache implements Cache for testing. type mockContactCache struct { - invalidated []string - online map[string]bool + invalidated []string + online map[string]bool + isOnlineCalls int + areOnlineCalls [][]string } func (m *mockContactCache) Invalidate(ctx context.Context, keys ...string) error { @@ -34,12 +36,24 @@ func (m *mockContactCache) Invalidate(ctx context.Context, keys ...string) error } func (m *mockContactCache) IsOnline(ctx context.Context, userID string) (bool, error) { + m.isOnlineCalls++ if m.online == nil { return false, nil } return m.online[userID], nil } +func (m *mockContactCache) AreOnline(ctx context.Context, userIDs []string) (map[string]bool, error) { + m.areOnlineCalls = append(m.areOnlineCalls, append([]string(nil), userIDs...)) + out := make(map[string]bool, len(userIDs)) + for _, id := range userIDs { + if m.online != nil { + out[id] = m.online[id] + } + } + return out, nil +} + // recordingContactBus is a Bus test double that records Publish calls. type recordingContactBus struct { events []bus.Event @@ -789,6 +803,42 @@ func TestListContacts_BatchesFriendUserLookup(t *testing.T) { assert.NoError(t, mock.ExpectationsWereMet()) } +func TestListContacts_BatchesPresenceLookups(t *testing.T) { + db, mock, sqlDB := newMockDBContact(t) + defer sqlDB.Close() + + mock.ExpectQuery(sqlcFriendshipsByUser). + WithArgs("user-1", model.StatusAccepted, 500). + WillReturnRows(sqlmock.NewRows([]string{"id", "user_id", "friend_id", "status", "remark"}). + AddRow("f-1", "user-1", "friend-a", model.StatusAccepted, "A"). + AddRow("f-2", "user-1", "friend-b", model.StatusAccepted, "B"). + AddRow("f-3", "user-1", "friend-c", model.StatusAccepted, "C")) + + mock.ExpectQuery(sqlcUsersByIDs). + WithArgs("friend-a", "friend-b", "friend-c"). + WillReturnRows(sqlmock.NewRows([]string{"id", "username", "password_hash", "nickname", "avatar_url"}). + AddRow("friend-a", "friendA", "hash-a", "Friend A", ""). + AddRow("friend-b", "friendB", "hash-b", "Friend B", ""). + AddRow("friend-c", "friendC", "hash-c", "Friend C", "")) + + mc := &mockContactCache{online: map[string]bool{"friend-b": true}} + svc := NewService(db, nil, mc) + contacts, err := svc.ListContacts(context.Background(), "user-1") + require.NoError(t, err) + require.Len(t, contacts, 3) + + // Presence resolves in exactly one batched call covering all friends; the + // per-user IsOnline path must not be used anymore (#2154 perf lane). + require.Len(t, mc.areOnlineCalls, 1) + assert.ElementsMatch(t, []string{"friend-a", "friend-b", "friend-c"}, mc.areOnlineCalls[0]) + assert.Zero(t, mc.isOnlineCalls) + + assert.False(t, contacts[0].Online) + assert.True(t, contacts[1].Online) + assert.False(t, contacts[2].Online) + assert.NoError(t, mock.ExpectationsWereMet()) +} + // ==================== UpdateRemark ==================== func TestUpdateRemark(t *testing.T) {