Fix debian package clean up (#32351) (#32590)

Partially backport #32351

models/packages/debian/search.go

@@ -75,26 +75,27 @@ func ExistPackages(ctx context.Context, opts *PackageSearchOptions) (bool, error
 }
 
 // SearchPackages gets the packages matching the search options
-func SearchPackages(ctx context.Context, opts *PackageSearchOptions, iter func(*packages.PackageFileDescriptor)) error {
-	return db.GetEngine(ctx).
+func SearchPackages(ctx context.Context, opts *PackageSearchOptions) ([]*packages.PackageFileDescriptor, error) {
+	var pkgFiles []*packages.PackageFile
+	err := db.GetEngine(ctx).
 		Table("package_file").
 		Select("package_file.*").
 		Join("INNER", "package_version", "package_version.id = package_file.version_id").
 		Join("INNER", "package", "package.id = package_version.package_id").
 		Where(opts.toCond()).
-		Asc("package.lower_name", "package_version.created_unix").
-		Iterate(new(packages.PackageFile), func(_ int, bean any) error {
-			pf := bean.(*packages.PackageFile)
-
-			pfd, err := packages.GetPackageFileDescriptor(ctx, pf)
-			if err != nil {
-				return err
-			}
-
-			iter(pfd)
-
-			return nil
-		})
+		Asc("package.lower_name", "package_version.created_unix").Find(&pkgFiles)
+	if err != nil {
+		return nil, err
+	}
+	pfds := make([]*packages.PackageFileDescriptor, 0, len(pkgFiles))
+	for _, pf := range pkgFiles {
+		pfd, err := packages.GetPackageFileDescriptor(ctx, pf)
+		if err != nil {
+			return nil, err
+		}
+		pfds = append(pfds, pfd)
+	}
+	return pfds, nil
 }
 
 // GetDistributions gets all available distributions

services/packages/debian/repository.go

@@ -206,7 +206,11 @@ func buildPackagesIndices(ctx context.Context, ownerID int64, repoVersion *packa
 	w := io.MultiWriter(packagesContent, gzw, xzw)
 
 	addSeparator := false
-	if err := debian_model.SearchPackages(ctx, opts, func(pfd *packages_model.PackageFileDescriptor) {
+	pfds, err := debian_model.SearchPackages(ctx, opts)
+	if err != nil {
+		return err
+	}
+	for _, pfd := range pfds {
 		if addSeparator {
 			fmt.Fprintln(w)
 		}
@@ -220,10 +224,7 @@ func buildPackagesIndices(ctx context.Context, ownerID int64, repoVersion *packa
 		fmt.Fprintf(w, "SHA1: %s\n", pfd.Blob.HashSHA1)
 		fmt.Fprintf(w, "SHA256: %s\n", pfd.Blob.HashSHA256)
 		fmt.Fprintf(w, "SHA512: %s\n", pfd.Blob.HashSHA512)
-	}); err != nil {
-		return err
 	}
-
 	gzw.Close()
 	xzw.Close()
 

tests/integration/api_packages_debian_test.go

@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"strconv"
 	"strings"
 	"testing"
 
@@ -19,6 +20,7 @@ import (
 	user_model "code.gitea.io/gitea/models/user"
 	"code.gitea.io/gitea/modules/base"
 	debian_module "code.gitea.io/gitea/modules/packages/debian"
+	packages_cleanup_service "code.gitea.io/gitea/services/packages/cleanup"
 	"code.gitea.io/gitea/tests"
 
 	"github.com/blakesmith/ar"
@@ -263,4 +265,37 @@ func TestPackageDebian(t *testing.T) {
 		assert.Contains(t, body, "Components: "+strings.Join(components, " ")+"\n")
 		assert.Contains(t, body, "Architectures: "+architectures[1]+"\n")
 	})
+
+	t.Run("Cleanup", func(t *testing.T) {
+		defer tests.PrintCurrentTest(t)()
+
+		rule := &packages.PackageCleanupRule{
+			Enabled:       true,
+			RemovePattern: `.*`,
+			MatchFullName: true,
+			OwnerID:       user.ID,
+			Type:          packages.TypeDebian,
+		}
+
+		_, err := packages.InsertCleanupRule(db.DefaultContext, rule)
+		assert.NoError(t, err)
+
+		// When there were a lot of packages (> 50 or 100) and the code used "Iterate" to get all packages, it ever caused bugs,
+		// because "Iterate" keeps a dangling SQL session but the callback function still uses the same session to execute statements.
+		// The "Iterate" problem has been checked by TestContextSafety now, so here we only need to check the cleanup logic with a small number
+		packagesCount := 2
+		for i := 0; i < packagesCount; i++ {
+			uploadURL := fmt.Sprintf("%s/pool/%s/%s/upload", rootURL, "test", "main")
+			req := NewRequestWithBody(t, "PUT", uploadURL, createArchive(packageName, "1.0."+strconv.Itoa(i), "all")).AddBasicAuth(user.Name)
+			MakeRequest(t, req, http.StatusCreated)
+		}
+		req := NewRequest(t, "GET", fmt.Sprintf("%s/dists/%s/Release", rootURL, "test"))
+		MakeRequest(t, req, http.StatusOK)
+
+		err = packages_cleanup_service.CleanupTask(db.DefaultContext, 0)
+		assert.NoError(t, err)
+
+		req = NewRequest(t, "GET", fmt.Sprintf("%s/dists/%s/Release", rootURL, "test"))
+		MakeRequest(t, req, http.StatusNotFound)
+	})
 }