fix(server): limit implicit large group subscription meta

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.

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()