Add document hard deletion functionality to housekeeping tasks.

cmd/yorkie/server.go

@@ -40,6 +40,7 @@ var (
 
 	adminTokenDuration        time.Duration
 	housekeepingInterval      time.Duration
+	deleteAfterTime           time.Duration
 	clientDeactivateThreshold string
 
 	mongoConnectionURI     string
@@ -70,6 +71,7 @@ func newServerCmd() *cobra.Command {
 			conf.Backend.ProjectInfoCacheTTL = projectInfoCacheTTL.String()
 
 			conf.Housekeeping.Interval = housekeepingInterval.String()
+			conf.Housekeeping.DeleteAfterTime = deleteAfterTime.String()
 
 			if mongoConnectionURI != "" {
 				conf.Mongo = &mongo.Config{
@@ -205,6 +207,18 @@ func init() {
 		server.DefaultHousekeepingCandidatesLimitPerProject,
 		"candidates limit per project for a single housekeeping run",
 	)
+	cmd.Flags().IntVar(
+		&conf.Housekeeping.DocumentHardDeletionLimitPerProject,
+		"housekeeping-DocumentHardDeletion-limit-per-project",
+		server.DefaultHousekeepingDocumentHardDeletionLimitPerProject,
+		"DocumentHardDeletion limit per project for a single housekeeping run",
+	)
+	cmd.Flags().DurationVar(
+		&deleteAfterTime,
+		"housekeeping-DocumentHardDeletion-delete-after-time",
+		server.DefaultHousekeepingDeleteAfterTime,
+		"DocumentHardDeletion delete after time for a single housekeeping run",
+	)
 	cmd.Flags().IntVar(
 		&conf.Housekeeping.ProjectFetchSize,
 		"housekeeping-project-fetch-size",

server/backend/database/database.go

@@ -20,6 +20,7 @@ package database
 import (
 	"context"
 	"errors"
+	gotime "time"
 
 	"github.com/yorkie-team/yorkie/api/types"
 	"github.com/yorkie-team/yorkie/pkg/document"
@@ -157,6 +158,21 @@ type Database interface {
 		candidatesLimit int,
 	) ([]*ClientInfo, error)
 
+	// FindDocumentHardDeletionCandidates finds the housekeeping deletion candidates.
+	FindDocumentHardDeletionCandidates(
+		ctx context.Context,
+		candidatesLimitPerProject int,
+		projectFetchSize int,
+		deleteAfterTime gotime.Duration,
+		lastProjectID types.ID,
+	) (types.ID, []*DocInfo, error)
+
+	// DocumentHardDeletion Document complete deletion in progress
+	DocumentHardDeletion(
+		ctx context.Context,
+		candidate []*DocInfo,
+	) error
+
 	// FindDocInfoByKey finds the document of the given key.
 	FindDocInfoByKey(
 		ctx context.Context,

server/backend/database/memory/database.go

@@ -619,6 +619,44 @@ func (d *DB) UpdateClientInfoAfterPushPull(
 	return nil
 }
 
+// FindDocumentHardDeletionCandidatesPerProject finds the documents that need housekeeping per project.
+func (d *DB) FindDocumentHardDeletionCandidatesPerProject(
+	_ context.Context,
+	project *database.ProjectInfo,
+	candidatesLimit int,
+	deleteAfterTime gotime.Duration,
+) ([]*database.DocInfo, error) {
+
+	txn := d.db.Txn(false)
+	defer txn.Abort()
+
+	var documents []*database.DocInfo
+	iterator, err := txn.Get(
+		tblDocuments,
+		"project_id",
+		project.ID.String(),
+	)
+
+	if err != nil {
+		return nil, fmt.Errorf("fetch hard deletion candidates: %w", err)
+	}
+
+	currentTime := gotime.Now()
+	conditionDeleteAfterTime := currentTime.Add(deleteAfterTime)
+	for raw := iterator.Next(); raw != nil; raw = iterator.Next() {
+		document := raw.(*database.DocInfo)
+		if candidatesLimit <= len(documents) && candidatesLimit != 0 {
+			break
+		}
+
+		if !document.RemovedAt.After(conditionDeleteAfterTime) {
+			documents = append(documents, document)
+		}
+	}
+
+	return documents, nil
+}
+
 // FindDeactivateCandidatesPerProject finds the clients that need housekeeping per project.
 func (d *DB) FindDeactivateCandidatesPerProject(
 	_ context.Context,
@@ -663,6 +701,66 @@ func (d *DB) FindDeactivateCandidatesPerProject(
 	return infos, nil
 }
 
+// FindDocumentHardDeletionCandidates finds the clients that need housekeeping.
+func (d *DB) FindDocumentHardDeletionCandidates(
+	ctx context.Context,
+	candidatesLimitPerProject int,
+	projectFetchSize int,
+	deletedAfterTime gotime.Duration,
+	lastProjectID types.ID,
+) (types.ID, []*database.DocInfo, error) {
+	projects, err := d.FindNextNCyclingProjectInfos(ctx, projectFetchSize, lastProjectID)
+	if err != nil {
+		return database.DefaultProjectID, nil, err
+	}
+
+	var candidates []*database.DocInfo
+	for _, project := range projects {
+		infos, err := d.FindDocumentHardDeletionCandidatesPerProject(
+			ctx,
+			project,
+			candidatesLimitPerProject,
+			deletedAfterTime,
+		)
+		if err != nil {
+			return database.DefaultProjectID, nil, err
+		}
+
+		candidates = append(candidates, infos...)
+	}
+
+	var topProjectID types.ID
+	if len(projects) < projectFetchSize {
+		topProjectID = database.DefaultProjectID
+	} else {
+		topProjectID = projects[len(projects)-1].ID
+	}
+
+	return topProjectID, candidates, nil
+}
+
+// DocumentHardDeletion Deletes the documents completely.
+func (d *DB) DocumentHardDeletion(
+	_ context.Context,
+	candidates []*database.DocInfo,
+) error {
+	if len(candidates) <= 0 {
+		return nil
+	}
+
+	txn := d.db.Txn(true)
+	defer txn.Abort()
+
+	for _, candidate := range candidates {
+		if err := txn.Delete(tblDocuments, candidate); err != nil {
+			return fmt.Errorf("fetch hard deletion candidates: %w", err)
+		}
+	}
+	txn.Commit()
+
+	return nil
+}
+
 // FindDocInfoByKeyAndOwner finds the document of the given key. If the
 // createDocIfNotExist condition is true, create the document if it does not
 // exist.

server/backend/database/memory/database_test.go

@@ -103,4 +103,8 @@ func TestDB(t *testing.T) {
 	t.Run("IsDocumentAttached test", func(t *testing.T) {
 		testcases.RunIsDocumentAttachedTest(t, db, projectID)
 	})
+
+	t.Run("DocumentHardDeletion test", func(t *testing.T) {
+		testcases.RunDocumentHardDeletion(t, db)
+	})
 }

server/backend/database/memory/indexes.go

@@ -117,6 +117,11 @@ var schema = &memdb.DBSchema{
 					Unique:  true,
 					Indexer: &memdb.StringFieldIndex{Field: "ID"},
 				},
+				"project_id": {
+					Name:    "project_id",
+					Unique:  true,
+					Indexer: &memdb.StringFieldIndex{Field: "ProjectID"},
+				},
 				"project_id_id": {
 					Name:   "project_id_id",
 					Unique: true,

server/backend/database/mongo/client.go

@@ -227,6 +227,36 @@ func (c *Client) CreateProjectInfo(
 	return info, nil
 }
 
+// DocumentHardDeletion Deletes the documents completely.
+func (c *Client) DocumentHardDeletion(
+	ctx context.Context,
+	candidates []*database.DocInfo,
+) error {
+	if len(candidates) <= 0 {
+		return nil
+	}
+
+	var idList []primitive.ObjectID
+	for _, docInfo := range candidates {
+		encodedID, err := encodeID(docInfo.ID)
+		if err != nil {
+			return err
+		}
+		idList = append(idList, encodedID)
+	}
+
+	_, err := c.collection(ColDocuments).DeleteMany(
+		ctx,
+		bson.M{"_id": bson.M{"$in": idList}},
+	)
+
+	if err != nil {
+		return fmt.Errorf("deletion Error : %w", err)
+	}
+
+	return nil
+}
+
 // FindNextNCyclingProjectInfos finds the next N cycling projects from the given projectID.
 func (c *Client) FindNextNCyclingProjectInfos(
 	ctx context.Context,
@@ -619,6 +649,38 @@ func (c *Client) UpdateClientInfoAfterPushPull(
 	return nil
 }
 
+// FindDocumentHardDeletionCandidatesPerProject finds the clients that need housekeeping per project.
+func (c *Client) FindDocumentHardDeletionCandidatesPerProject(
+	ctx context.Context,
+	project *database.ProjectInfo,
+	candidatesLimit int,
+	deleteAfterTime gotime.Duration,
+) ([]*database.DocInfo, error) {
+	encodedProjectID, err := encodeID(project.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	currentTime := gotime.Now()
+	conditionDeleteAfterTime := currentTime.Add(deleteAfterTime)
+
+	var DocInfos []*database.DocInfo
+	cursor, err := c.collection(ColDocuments).Find(ctx, bson.M{
+		"project_id": encodedProjectID,
+		"removed_at": bson.M{"$lt": conditionDeleteAfterTime},
+	}, options.Find().SetLimit(int64(candidatesLimit)))
+
+	if err != nil {
+		return nil, err
+	}
+
+	if err := cursor.All(ctx, &DocInfos); err != nil {
+		return nil, fmt.Errorf("fetch hard deletion candidates: %w", err)
+	}
+
+	return DocInfos, nil
+}
+
 // FindDeactivateCandidatesPerProject finds the clients that need housekeeping per project.
 func (c *Client) FindDeactivateCandidatesPerProject(
 	ctx context.Context,
@@ -650,6 +712,73 @@ func (c *Client) FindDeactivateCandidatesPerProject(
 	return clientInfos, nil
 }
 
+// FindDeactivateCandidates finds the clients that need housekeeping.
+func (c *Client) FindDeactivateCandidates(
+	ctx context.Context,
+	candidatesLimitPerProject int,
+	projectFetchSize int,
+	lastProjectID types.ID,
+) (types.ID, []*database.ClientInfo, error) {
+	projects, err := c.FindNextNCyclingProjectInfos(ctx, projectFetchSize, lastProjectID)
+	if err != nil {
+		return database.DefaultProjectID, nil, err
+	}
+
+	var candidates []*database.ClientInfo
+	for _, project := range projects {
+		clientInfos, err := c.FindDeactivateCandidatesPerProject(ctx, project, candidatesLimitPerProject)
+		if err != nil {
+			return database.DefaultProjectID, nil, err
+		}
+
+		candidates = append(candidates, clientInfos...)
+	}
+
+	var topProjectID types.ID
+	if len(projects) < projectFetchSize {
+		topProjectID = database.DefaultProjectID
+	} else {
+		topProjectID = projects[len(projects)-1].ID
+	}
+	return topProjectID, candidates, nil
+}
+
+// FindDocumentHardDeletionCandidates finds the clients that need housekeeping.
+func (c *Client) FindDocumentHardDeletionCandidates(
+	ctx context.Context,
+	candidatesLimitPerProject int,
+	projectFetchSize int,
+	deletedAfterTime gotime.Duration,
+	lastProjectID types.ID,
+) (types.ID, []*database.DocInfo, error) {
+	projects, err := c.FindNextNCyclingProjectInfos(ctx, projectFetchSize, lastProjectID)
+	if err != nil {
+		return database.DefaultProjectID, nil, err
+	}
+
+	var candidates []*database.DocInfo
+	for _, project := range projects {
+		docInfos, err := c.FindDocumentHardDeletionCandidatesPerProject(
+			ctx,
+			project,
+			candidatesLimitPerProject,
+			deletedAfterTime,
+		)
+		if err != nil {
+			return database.DefaultProjectID, nil, err
+		}
+		candidates = append(candidates, docInfos...)
+	}
+
+	var topProjectID types.ID
+	if len(projects) < projectFetchSize {
+		topProjectID = database.DefaultProjectID
+	} else {
+		topProjectID = projects[len(projects)-1].ID
+	}
+	return topProjectID, candidates, nil
+}
+
 // FindDocInfoByKeyAndOwner finds the document of the given key. If the
 // createDocIfNotExist condition is true, create the document if it does not
 // exist.

server/backend/database/mongo/client_test.go

@@ -120,4 +120,8 @@ func TestClient(t *testing.T) {
 	t.Run("IsDocumentAttached test", func(t *testing.T) {
 		testcases.RunIsDocumentAttachedTest(t, cli, dummyProjectID)
 	})
+
+	t.Run("DocumentHardDeletion test", func(t *testing.T) {
+		testcases.RunDocumentHardDeletion(t, cli)
+	})
 }