Add Pagination to Listing Projects for Housekeeping

To avoid all project info being loaded from DB, pagination is added for projects and perform housekeeping in smaller portions.

* Add HousekeepingProjectFetchSize in config

server/backend/database/database.go

@@ -138,6 +138,7 @@ type Database interface {
 	FindDeactivateCandidates(
 		ctx context.Context,
 		candidatesLimitPerProject int,
+		projectFetchSize int,
 	) ([]*ClientInfo, error)
 
 	// FindDocInfoByKey finds the document of the given key.

server/backend/database/memory/database.go

@@ -36,7 +36,8 @@ import (
 
 // DB is an in-memory database for testing or temporarily.
 type DB struct {
-	db *memdb.MemDB
+	db                      *memdb.MemDB
+	housekeepingProjectPage int
 }
 
 // New returns a new in-memory database.
@@ -47,7 +48,8 @@ func New() (*DB, error) {
 	}
 
 	return &DB{
-		db: memDB,
+		db:                      memDB,
+		housekeepingProjectPage: 0,
 	}, nil
 }
 
@@ -224,16 +226,15 @@ func (d *DB) CreateProjectInfo(
 	return info, nil
 }
 
-// ListAllProjectInfos returns all project infos.
-func (d *DB) listAllProjectInfos(
+// listProjectInfos returns all project infos rotationally.
+func (d *DB) listProjectInfos(
 	ctx context.Context,
+	pageSize int,
 ) ([]*database.ProjectInfo, error) {
 	txn := d.db.Txn(false)
 	defer txn.Abort()
 
-	// TODO(krapie): txn.Get() loads all projects in memory,
-	// which will cause performance issue as number of projects in DB grows.
-	// Therefore, pagination of projects is needed to avoid this issue.
+	offset := d.housekeepingProjectPage * pageSize
 	iter, err := txn.Get(
 		tblProjects,
 		"id",
@@ -243,11 +244,22 @@ func (d *DB) listAllProjectInfos(
 	}
 
 	var infos []*database.ProjectInfo
-	for raw := iter.Next(); raw != nil; raw = iter.Next() {
+
+	for i := 0; i < offset; i++ {
+		iter.Next()
+	}
+
+	for i := 0; i < pageSize; i++ {
+		raw := iter.Next()
+		if raw == nil {
+			d.housekeepingProjectPage = 0
+			break
+		}
 		info := raw.(*database.ProjectInfo).DeepCopy()
 		infos = append(infos, info)
 	}
 
+	d.housekeepingProjectPage++
 	return infos, nil
 }
 
@@ -599,8 +611,9 @@ func (d *DB) findDeactivateCandidatesPerProject(
 func (d *DB) FindDeactivateCandidates(
 	ctx context.Context,
 	candidatesLimitPerProject int,
+	projectFetchSize int,
 ) ([]*database.ClientInfo, error) {
-	projects, err := d.listAllProjectInfos(ctx)
+	projects, err := d.listProjectInfos(ctx, projectFetchSize)
 	if err != nil {
 		return nil, err
 	}

server/backend/database/memory/housekeeping_test.go

@@ -62,6 +62,7 @@ func TestHousekeeping(t *testing.T) {
 		candidates, err := memdb.FindDeactivateCandidates(
 			ctx,
 			10,
+			10,
 		)
 		assert.NoError(t, err)
 		assert.Len(t, candidates, 2)

server/backend/database/mongo/client.go

@@ -42,8 +42,9 @@ import (
 
 // Client is a client that connects to Mongo DB and reads or saves Yorkie data.
 type Client struct {
-	config *Config
-	client *mongo.Client
+	config                  *Config
+	client                  *mongo.Client
+	housekeepingProjectPage int
 }
 
 // Dial creates an instance of Client and dials the given MongoDB.
@@ -76,8 +77,9 @@ func Dial(conf *Config) (*Client, error) {
 	logging.DefaultLogger().Infof("MongoDB connected, URI: %s, DB: %s", conf.ConnectionURI, conf.YorkieDatabase)
 
 	return &Client{
-		config: conf,
-		client: client,
+		config:                  conf,
+		client:                  client,
+		housekeepingProjectPage: 0,
 	}, nil
 }
 
@@ -235,23 +237,30 @@ func (c *Client) CreateProjectInfo(
 	return info, nil
 }
 
-// ListAllProjectInfos returns all project infos.
-func (c *Client) listAllProjectInfos(
+// listProjectInfos returns all project infos rotationally.
+func (c *Client) listProjectInfos(
 	ctx context.Context,
+	pageSize int,
 ) ([]*database.ProjectInfo, error) {
-	// TODO(krapie): Find(ctx, bson.D{{}}) loads all projects in memory,
-	// which will cause performance issue as number of projects in DB grows.
-	// Therefore, pagination of projects is needed to avoid this issue.
-	cursor, err := c.collection(colProjects).Find(ctx, bson.D{{}})
+	opts := options.Find()
+	opts.SetSkip(int64(c.housekeepingProjectPage * pageSize))
+	opts.SetLimit(int64(pageSize))
+
+	cursor, err := c.collection(colProjects).Find(ctx, bson.D{{}}, opts)
 	if err != nil {
-		return nil, fmt.Errorf("fetch all project infos: %w", err)
+		return nil, fmt.Errorf("find project infos: %w", err)
 	}
 
 	var infos []*database.ProjectInfo
 	if err := cursor.All(ctx, &infos); err != nil {
-		return nil, fmt.Errorf("fetch all project infos: %w", err)
+		return nil, fmt.Errorf("fetch project infos: %w", err)
+	}
+
+	if cursor.RemainingBatchLength() == 0 {
+		c.housekeepingProjectPage = 0
 	}
 
+	c.housekeepingProjectPage++
 	return infos, nil
 }
 
@@ -657,8 +666,9 @@ func (c *Client) findDeactivateCandidatesPerProject(
 func (c *Client) FindDeactivateCandidates(
 	ctx context.Context,
 	candidatesLimitPerProject int,
+	projectFetchSize int,
 ) ([]*database.ClientInfo, error) {
-	projects, err := c.listAllProjectInfos(ctx)
+	projects, err := c.listProjectInfos(ctx, projectFetchSize)
 	if err != nil {
 		return nil, err
 	}

server/backend/housekeeping/housekeeping.go

@@ -41,6 +41,9 @@ type Config struct {
 
 	// CandidatesLimitPerProject is the maximum number of candidates to be returned per project.
 	CandidatesLimitPerProject int `yaml:"CandidatesLimitPerProject"`
+
+	// HousekeepingProjectFetchSize is the maximum number of projects to be returned to deactivate candidates.
+	HousekeepingProjectFetchSize int `yaml:"HousekeepingProjectFetchSize"`
 }
 
 // Validate validates the configuration.
@@ -65,6 +68,7 @@ type Housekeeping struct {
 
 	interval                  time.Duration
 	candidatesLimitPerProject int
+	projectFetchSize          int
 
 	ctx        context.Context
 	cancelFunc context.CancelFunc
@@ -106,6 +110,7 @@ func New(
 
 		interval:                  interval,
 		candidatesLimitPerProject: conf.CandidatesLimitPerProject,
+		projectFetchSize:          conf.HousekeepingProjectFetchSize,
 
 		ctx:        ctx,
 		cancelFunc: cancelFunc,
@@ -162,6 +167,7 @@ func (h *Housekeeping) deactivateCandidates(ctx context.Context) error {
 	candidates, err := h.database.FindDeactivateCandidates(
 		ctx,
 		h.candidatesLimitPerProject,
+		h.projectFetchSize,
 	)
 	if err != nil {
 		return err

server/config.go

@@ -42,6 +42,7 @@ const (
 
 	DefaultHousekeepingInterval                  = 30 * time.Second
 	DefaultHousekeepingCandidatesLimitPerProject = 500
+	DefaultHousekeepingProjectFetchSize          = 100
 
 	DefaultMongoConnectionURI     = "mongodb://localhost:27017"
 	DefaultMongoConnectionTimeout = 5 * time.Second
@@ -229,8 +230,9 @@ func newConfig(port int, profilingPort int) *Config {
 			Port: profilingPort,
 		},
 		Housekeeping: &housekeeping.Config{
-			Interval:                  DefaultHousekeepingInterval.String(),
-			CandidatesLimitPerProject: DefaultHousekeepingCandidatesLimitPerProject,
+			Interval:                     DefaultHousekeepingInterval.String(),
+			CandidatesLimitPerProject:    DefaultHousekeepingCandidatesLimitPerProject,
+			HousekeepingProjectFetchSize: DefaultHousekeepingProjectFetchSize,
 		},
 		Backend: &backend.Config{
 			ClientDeactivateThreshold:  DefaultClientDeactivateThreshold,

server/config.sample.yml

@@ -36,6 +36,9 @@ Housekeeping:
   # CandidatesLimitPerProject is the maximum number of candidates to be returned per project (default: 100).
   CandidatesLimitPerProject: 100
 
+  # HousekeepingProjectFetchSize is the maximum number of projects to be returned to deactivate candidates. (default: 100)
+  HousekeepingProjectFetchSize: 10
+
 # Backend is the configuration for the backend of Yorkie.
 Backend:
   # UseDefaultProject is whether to use the default project (default: true).