From e30b70a3a753d1a70f5be15fca5fa69645b1bc0f Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Wed, 7 Oct 2026 23:13:47 +0800 Subject: [PATCH] fix(server): limit implicit large group subscription meta --- server/topic.go | 24 ++++++++- server/topic_test.go | 115 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 138 insertions(+), 1 deletion(-) diff --git a/server/topic.go b/server/topic.go index 172de61..0f4dfb3 100644 --- a/server/topic.go +++ b/server/topic.go @@ -202,6 +202,8 @@ const ( StopRehashing ) +const implicitGroupSubSelfOnlyThreshold = 1000 + // Topic shutdown type shutDown struct { // Channel to report back completion of topic shutdown. Could be nil @@ -2894,7 +2896,6 @@ func (t *Topic) replyGetDesc(sess *Session, asUid types.Uid, _ bool, opts *MsgGe if t.cat == types.TopicCatGrp { desc.IsChan = t.isChan desc.SubCnt = t.subCnt - logs.Info.Println("replyGetDesc: grp topic", t.name, "subs", t.subCnt) } if ifUpdated { @@ -3353,6 +3354,12 @@ func (t *Topic) replyGetSub(sess *Session, asUid types.Uid, authLevel auth.Level // Channel subscribers are using chnXXX topic name rather than grpXXX. topicName = msg.Original } + if req = t.limitImplicitLargeGroupSubQuery(req, asUid, msg); req != nil { + ifModified = time.Time{} + if req.IfModifiedSince != nil { + ifModified = *req.IfModifiedSince + } + } // Include sub.Public. if ifModified.IsZero() { // No cache management. Skip deleted subscriptions. @@ -3557,6 +3564,21 @@ func (t *Topic) replyGetSub(sess *Session, asUid types.Uid, authLevel auth.Level return nil } +func (t *Topic) limitImplicitLargeGroupSubQuery(req *MsgGetOpts, asUid types.Uid, msg *ClientComMessage) *MsgGetOpts { + if t.cat != types.TopicCatGrp || msg == nil || msg.Sub == nil || t.subCnt < implicitGroupSubSelfOnlyThreshold { + return req + } + if req != nil && (req.User != "" || req.Topic != "" || req.Limit != 0 || req.IfModifiedSince != nil) { + return req + } + limited := &MsgGetOpts{} + if req != nil { + *limited = *req + } + limited.User = asUid.UserId() + return limited +} + // replySetSub is a response to new subscription request or an update to a subscription {set.sub}: // update topic metadata cache, save/update subs, reply to the caller as {ctrl} message, // generate a presence notification, if appropriate. diff --git a/server/topic_test.go b/server/topic_test.go index 60567c6..d5309e0 100644 --- a/server/topic_test.go +++ b/server/topic_test.go @@ -2254,6 +2254,121 @@ func TestReplyGetDescInvalidOpts(t *testing.T) { } } +func TestReplyGetSubLimitsImplicitLargeGroupSubscriptionQuery(t *testing.T) { + helper := TopicTestHelper{} + helper.setUp(t, 2, types.TopicCatGrp, "grpLarge", true) + defer helper.tearDown() + + uid := helper.uids[0] + helper.topic.subCnt = implicitGroupSubSelfOnlyThreshold + helper.topic.perUser[uid] = perUserData{ + modeWant: types.ModeCReadOnly, + modeGiven: types.ModeCReadOnly, + } + + helper.tt.EXPECT(). + GetUsers(types.TenantID(1), "grpLarge", gomock.AssignableToTypeOf(&types.QueryOpt{})). + DoAndReturn(func(_ types.TenantID, _ string, opts *types.QueryOpt) ([]types.Subscription, error) { + if opts == nil { + t.Fatal("expected limited query options") + } + if opts.User != uid { + t.Fatalf("expected query to be limited to %s, got %s", uid.UserId(), opts.User.UserId()) + } + return []types.Subscription{{ + User: uid.String(), + Topic: "grpLarge", + ModeWant: types.ModeCReadOnly, + ModeGiven: types.ModeCReadOnly, + }}, nil + }) + + msg := &ClientComMessage{ + Id: "sub1", + Original: "grpLarge", + AsUser: uid.UserId(), + TenantID: 1, + Sub: &MsgClientSub{Get: &MsgGetQuery{ + What: "sub", + Sub: nil, + }}, + } + + if err := helper.topic.replyGetSub(helper.sessions[0], uid, auth.LevelAuth, false, msg); err != nil { + t.Fatalf("replyGetSub failed: %v", err) + } + helper.finish() + + if len(helper.results[0].messages) != 1 { + t.Fatalf("expected one meta response, got %d", len(helper.results[0].messages)) + } + resp := helper.results[0].messages[0].(*ServerComMessage) + if resp.Meta == nil { + t.Fatalf("expected meta response") + } + if len(resp.Meta.Sub) != 1 { + t.Fatalf("expected one subscription, got %d", len(resp.Meta.Sub)) + } + if got := resp.Meta.Sub[0].User; got != uid.UserId() { + t.Fatalf("expected self subscription %s, got %s", uid.UserId(), got) + } +} + +func TestReplyGetSubKeepsExplicitLargeGroupQuery(t *testing.T) { + helper := TopicTestHelper{} + helper.setUp(t, 2, types.TopicCatGrp, "grpLarge", true) + defer helper.tearDown() + + uid := helper.uids[0] + peer := helper.uids[1] + helper.topic.subCnt = implicitGroupSubSelfOnlyThreshold + helper.topic.perUser[uid] = perUserData{ + modeWant: types.ModeCReadOnly, + modeGiven: types.ModeCReadOnly, + } + + helper.tt.EXPECT(). + GetUsers(types.TenantID(1), "grpLarge", gomock.Nil()). + Return([]types.Subscription{ + { + User: uid.String(), + Topic: "grpLarge", + ModeWant: types.ModeCReadOnly, + ModeGiven: types.ModeCReadOnly, + }, + { + User: peer.String(), + Topic: "grpLarge", + ModeWant: types.ModeCReadOnly, + ModeGiven: types.ModeCReadOnly, + }, + }, nil) + + msg := &ClientComMessage{ + Id: "get1", + Original: "grpLarge", + AsUser: uid.UserId(), + TenantID: 1, + Get: &MsgClientGet{MsgGetQuery: MsgGetQuery{What: "sub"}}, + } + + if err := helper.topic.replyGetSub(helper.sessions[0], uid, auth.LevelAuth, false, msg); err != nil { + t.Fatalf("replyGetSub failed: %v", err) + } + helper.finish() + + if len(helper.results[0].messages) != 1 { + t.Fatalf("expected one meta response, got %d", len(helper.results[0].messages)) + } + resp := helper.results[0].messages[0].(*ServerComMessage) + if resp.Meta == nil { + t.Fatalf("expected meta response") + } + if len(resp.Meta.Sub) != 2 { + t.Fatalf("expected explicit query to return all subscriptions, got %d", len(resp.Meta.Sub)) + } +} + // Verifies ctrl codes in session outputs. func registerSessionVerifyOutputs(t *testing.T, sessionOutput *responses, expectedCtrlCodes []int) { t.Helper()