Add doctor command for full GC of LFS (#21978)

The recent PR adding orphaned checks to the LFS storage is not
sufficient to completely GC LFS, as it is possible for LFSMetaObjects to
remain associated with repos but still need to be garbage collected.

Imagine a situation where a branch is uploaded containing LFS files but
that branch is later completely deleted. The LFSMetaObjects will remain
associated with the Repository but the Repository will no longer contain
any pointers to the object.

This PR adds a second doctor command to perform a full GC.

Signed-off-by: Andrew Thornton <art27@cantab.net>

1 files changed, 54 insertions, 0 deletions

diff --git a/models/git/lfs.go b/models/git/lfs.go
index a86e84c050..8d418b928d 100644
--- a/models/git/lfs.go
+++ b/models/git/lfs.go
@@ -6,6 +6,7 @@ package git
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"code.gitea.io/gitea/models/db"
 	"code.gitea.io/gitea/models/perm"
@@ -14,6 +15,7 @@ import (
 	user_model "code.gitea.io/gitea/models/user"
 	"code.gitea.io/gitea/modules/lfs"
 	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/setting"
 	"code.gitea.io/gitea/modules/timeutil"
 	"code.gitea.io/gitea/modules/util"
 
@@ -180,6 +182,12 @@ func GetLFSMetaObjectByOid(repoID int64, oid string) (*LFSMetaObject, error) {
 // RemoveLFSMetaObjectByOid removes a LFSMetaObject entry from database by its OID.
 // It may return ErrLFSObjectNotExist or a database error.
 func RemoveLFSMetaObjectByOid(repoID int64, oid string) (int64, error) {
+	return RemoveLFSMetaObjectByOidFn(repoID, oid, nil)
+}
+
+// RemoveLFSMetaObjectByOidFn removes a LFSMetaObject entry from database by its OID.
+// It may return ErrLFSObjectNotExist or a database error. It will run Fn with the current count within the transaction
+func RemoveLFSMetaObjectByOidFn(repoID int64, oid string, fn func(count int64) error) (int64, error) {
 	if len(oid) == 0 {
 		return 0, ErrLFSObjectNotExist
 	}
@@ -200,6 +208,12 @@ func RemoveLFSMetaObjectByOid(repoID int64, oid string) (int64, error) {
 		return count, err
 	}
 
+	if fn != nil {
+		if err := fn(count); err != nil {
+			return count, err
+		}
+	}
+
 	return count, committer.Commit()
 }
 
@@ -319,3 +333,43 @@ func GetRepoLFSSize(ctx context.Context, repoID int64) (int64, error) {
 	}
 	return lfsSize, nil
 }
+
+type IterateLFSMetaObjectsForRepoOptions struct {
+	OlderThan time.Time
+}
+
+// IterateLFSMetaObjectsForRepo provides a iterator for LFSMetaObjects per Repo
+func IterateLFSMetaObjectsForRepo(ctx context.Context, repoID int64, f func(context.Context, *LFSMetaObject, int64) error, opts *IterateLFSMetaObjectsForRepoOptions) error {
+	var start int
+	batchSize := setting.Database.IterateBufferSize
+	engine := db.GetEngine(ctx)
+	type CountLFSMetaObject struct {
+		Count int64
+		LFSMetaObject
+	}
+
+	for {
+		beans := make([]*CountLFSMetaObject, 0, batchSize)
+		// SELECT `lfs_meta_object`.*, COUNT(`l1`.id) as `count` FROM lfs_meta_object INNER JOIN lfs_meta_object AS l1 ON l1.oid = lfs_meta_object.oid WHERE lfs_meta_object.repository_id = ? GROUP BY lfs_meta_object.id
+		sess := engine.Select("`lfs_meta_object`.*, COUNT(`l1`.oid) AS `count`").
+			Join("INNER", "`lfs_meta_object` AS l1", "`lfs_meta_object`.oid = `l1`.oid").
+			Where("`lfs_meta_object`.repository_id = ?", repoID)
+		if !opts.OlderThan.IsZero() {
+			sess.And("`lfs_meta_object`.created_unix < ?", opts.OlderThan)
+		}
+		sess.GroupBy("`lfs_meta_object`.id")
+		if err := sess.Limit(batchSize, start).Find(&beans); err != nil {
+			return err
+		}
+		if len(beans) == 0 {
+			return nil
+		}
+		start += len(beans)
+
+		for _, bean := range beans {
+			if err := f(ctx, &bean.LFSMetaObject, bean.Count); err != nil {
+				return err
+			}
+		}
+	}
+}