[RFC] Split reflists to share their contents across snapshots

.github/workflows/ci.yml

@@ -30,7 +30,7 @@ jobs:
       BOTO_CONFIG: /dev/null
       GO111MODULE: "on"
       GOPROXY: "https://proxy.golang.org"
-      GOVER: '1.21'
+      GOVER: '1.19'
 
     steps:
       - name: Checkout repository
@@ -99,7 +99,7 @@ jobs:
           - name: Ubuntu-20
             image: "ubuntu:20.04"
             suite: focal
-            GOVER: '1.21'
+            GOVER: '1.19'
             install: "make ca-certificates git curl"
           - name: Ubuntu-22
             image: "ubuntu:22.04"
@@ -108,12 +108,12 @@ jobs:
           - name: Debian-10
             image: "debian:buster"
             suite: buster
-            GOVER: '1.21'
+            GOVER: '1.19'
             install: "make ca-certificates git curl"
           - name: Debian-11
             image: "debian:bullseye"
             suite: bullseye
-            GOVER: '1.21'
+            GOVER: '1.19'
             install: "make ca-certificates git curl"
           - name: Debian-12
             image: "debian:bookworm"

api/api.go

@@ -189,7 +189,7 @@ func maybeRunTaskInBackground(c *gin.Context, name string, resources []string, p
 
 // Common piece of code to show list of packages,
 // with searching & details if requested
-func showPackages(c *gin.Context, reflist *deb.PackageRefList, collectionFactory *deb.CollectionFactory) {
+func showPackages(c *gin.Context, reflist deb.AnyRefList, collectionFactory *deb.CollectionFactory) {
 	result := []*deb.Package{}
 
 	list, err := deb.NewPackageListFromRefList(reflist, collectionFactory.PackageCollection(), nil)

api/db.go

@@ -5,6 +5,7 @@
 	"sort"
 
 	"github.com/aptly-dev/aptly/aptly"
+	"github.com/aptly-dev/aptly/database"
 	"github.com/aptly-dev/aptly/deb"
 	"github.com/aptly-dev/aptly/task"
 	"github.com/aptly-dev/aptly/utils"
@@ -20,18 +21,22 @@
 
 		collectionFactory := context.NewCollectionFactory()
 
-		// collect information about referenced packages...
-		existingPackageRefs := deb.NewPackageRefList()
+		// collect information about referenced packages and their reflist buckets...
+		existingPackageRefs := deb.NewSplitRefList()
+		existingBuckets := deb.NewRefListDigestSet()
+
+		reflistMigration := collectionFactory.RefListCollection().NewMigration()
 
 		out.Printf("Loading mirrors, local repos, snapshots and published repos...")
 		err = collectionFactory.RemoteRepoCollection().ForEach(func(repo *deb.RemoteRepo) error {
-			e := collectionFactory.RemoteRepoCollection().LoadComplete(repo)
-			if e != nil {
+			sl := deb.NewSplitRefList()
+			e := collectionFactory.RefListCollection().LoadCompleteAndMigrate(sl, repo.RefKey(), reflistMigration)
+			if e != nil && e != database.ErrNotFound {
 				return e
 			}
-			if repo.RefList() != nil {
-				existingPackageRefs = existingPackageRefs.Merge(repo.RefList(), false, true)
-			}
+
+			existingPackageRefs = existingPackageRefs.Merge(sl, false, true)
+			existingBuckets.AddAllInRefList(sl)
 
 			return nil
 		})
@@ -40,14 +45,14 @@
 		}
 
 		err = collectionFactory.LocalRepoCollection().ForEach(func(repo *deb.LocalRepo) error {
-			e := collectionFactory.LocalRepoCollection().LoadComplete(repo)
-			if e != nil {
+			sl := deb.NewSplitRefList()
+			e := collectionFactory.RefListCollection().LoadCompleteAndMigrate(sl, repo.RefKey(), reflistMigration)
+			if e != nil && e != database.ErrNotFound {
 				return e
 			}
 
-			if repo.RefList() != nil {
-				existingPackageRefs = existingPackageRefs.Merge(repo.RefList(), false, true)
-			}
+			existingPackageRefs = existingPackageRefs.Merge(sl, false, true)
+			existingBuckets.AddAllInRefList(sl)
 
 			return nil
 		})
@@ -56,12 +61,14 @@
 		}
 
 		err = collectionFactory.SnapshotCollection().ForEach(func(snapshot *deb.Snapshot) error {
-			e := collectionFactory.SnapshotCollection().LoadComplete(snapshot)
+			sl := deb.NewSplitRefList()
+			e := collectionFactory.RefListCollection().LoadCompleteAndMigrate(sl, snapshot.RefKey(), reflistMigration)
 			if e != nil {
 				return e
 			}
 
-			existingPackageRefs = existingPackageRefs.Merge(snapshot.RefList(), false, true)
+			existingPackageRefs = existingPackageRefs.Merge(sl, false, true)
+			existingBuckets.AddAllInRefList(sl)
 
 			return nil
 		})
@@ -73,25 +80,37 @@
 			if published.SourceKind != deb.SourceLocalRepo {
 				return nil
 			}
-			e := collectionFactory.PublishedRepoCollection().LoadComplete(published, collectionFactory)
-			if e != nil {
-				return e
-			}
 
 			for _, component := range published.Components() {
-				existingPackageRefs = existingPackageRefs.Merge(published.RefList(component), false, true)
+				sl := deb.NewSplitRefList()
+				e := collectionFactory.RefListCollection().LoadCompleteAndMigrate(sl, published.RefKey(component), reflistMigration)
+				if e != nil {
+					return e
+				}
+
+				existingPackageRefs = existingPackageRefs.Merge(sl, false, true)
+				existingBuckets.AddAllInRefList(sl)
 			}
 			return nil
 		})
 		if err != nil {
 			return nil, err
 		}
 
+		err = reflistMigration.Flush()
+		if err != nil {
+			return nil, err
+		}
+		if stats := reflistMigration.Stats(); stats.Reflists > 0 {
+			out.Printf("Split %d reflist(s) into %d bucket(s) (%d segment(s))",
+				stats.Reflists, stats.Buckets, stats.Segments)
+		}
+
 		// ... and compare it to the list of all packages
 		out.Printf("Loading list of all packages...")
 		allPackageRefs := collectionFactory.PackageCollection().AllPackageRefs()
 
-		toDelete := allPackageRefs.Subtract(existingPackageRefs)
+		toDelete := allPackageRefs.Subtract(existingPackageRefs.Flatten())
 
 		// delete packages that are no longer referenced
 		out.Printf("Deleting unreferenced packages (%d)...", toDelete.Len())
@@ -112,6 +131,28 @@
 			}
 		}
 
+		bucketsToDelete, err := collectionFactory.RefListCollection().AllBucketDigests()
+		if err != nil {
+			return nil, err
+		}
+
+		bucketsToDelete.RemoveAll(existingBuckets)
+
+		out.Printf("Deleting unreferenced reflist buckets (%d)...", bucketsToDelete.Len())
+		if bucketsToDelete.Len() > 0 {
+			batch := db.CreateBatch()
+			err := bucketsToDelete.ForEach(func(digest []byte) error {
+				return collectionFactory.RefListCollection().UnsafeDropBucket(digest, batch)
+			})
+			if err != nil {
+				return nil, err
+			}
+
+			if err := batch.Write(); err != nil {
+				return nil, err
+			}
+		}
+
 		// now, build a list of files that should be present in Repository (package pool)
 		out.Printf("Building list of files referenced by packages...")
 		referencedFiles := make([]string, 0, existingPackageRefs.Len())

api/files.go

@@ -125,11 +125,7 @@ func apiFilesListFiles(c *gin.Context) {
 	listLock := &sync.Mutex{}
 	root := filepath.Join(context.UploadPath(), c.Params.ByName("dir"))
 
-	err := filepath.Walk(root, func(path string, _ os.FileInfo, err error) error {
-		if err != nil {
-			return err
-		}
-
+	err := walker.Walk(root, func(path string, _ os.FileInfo) error {
 		if path == root {
 			return nil
 		}

api/metrics.go

@@ -102,7 +102,7 @@ func countPackagesByRepos() {
 
 		components := repo.Components()
 		for _, c := range components {
-			count := float64(len(repo.RefList(c).Refs))
+			count := float64(repo.RefList(c).Len())
 			apiReposPackageCountGauge.WithLabelValues(fmt.Sprintf("%s", (repo.SourceNames())), repo.Distribution, c).Set(count)
 		}
 

api/mirror.go

@@ -124,7 +124,7 @@ func apiMirrorsCreate(c *gin.Context) {
 		return
 	}
 
-	err = collection.Add(repo)
+	err = collection.Add(repo, collectionFactory.RefListCollection())
 	if err != nil {
 		AbortWithJSONError(c, 500, fmt.Errorf("unable to add mirror: %s", err))
 		return
@@ -184,7 +184,7 @@ func apiMirrorsShow(c *gin.Context) {
 		return
 	}
 
-	err = collection.LoadComplete(repo)
+	err = collection.LoadComplete(repo, collectionFactory.RefListCollection())
 	if err != nil {
 		AbortWithJSONError(c, 500, fmt.Errorf("unable to show: %s", err))
 	}
@@ -204,7 +204,7 @@ func apiMirrorsPackages(c *gin.Context) {
 		return
 	}
 
-	err = collection.LoadComplete(repo)
+	err = collection.LoadComplete(repo, collectionFactory.RefListCollection())
 	if err != nil {
 		AbortWithJSONError(c, 500, fmt.Errorf("unable to show: %s", err))
 	}
@@ -402,12 +402,12 @@ func apiMirrorsUpdate(c *gin.Context) {
 			e := context.ReOpenDatabase()
 			if e == nil {
 				remote.MarkAsIdle()
-				collection.Update(remote)
+				collection.Update(remote, collectionFactory.RefListCollection())
 			}
 		}()
 
 		remote.MarkAsUpdating()
-		err = collection.Update(remote)
+		err = collection.Update(remote, collectionFactory.RefListCollection())
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 		}
@@ -565,7 +565,7 @@ func apiMirrorsUpdate(c *gin.Context) {
 
 		log.Info().Msgf("%s: Finalizing download...", b.Name)
 		remote.FinalizeDownload(collectionFactory, out)
-		err = collectionFactory.RemoteRepoCollection().Update(remote)
+		err = collectionFactory.RemoteRepoCollection().Update(remote, collectionFactory.RefListCollection())
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 		}

api/publish.go

@@ -141,7 +141,7 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
 			}
 
 			resources = append(resources, string(snapshot.ResourceKey()))
-			err = snapshotCollection.LoadComplete(snapshot)
+			err = snapshotCollection.LoadComplete(snapshot, collectionFactory.RefListCollection())
 			if err != nil {
 				AbortWithJSONError(c, 500, fmt.Errorf("unable to publish: %s", err))
 				return
@@ -165,7 +165,7 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
 			}
 
 			resources = append(resources, string(localRepo.Key()))
-			err = localCollection.LoadComplete(localRepo)
+			err = localCollection.LoadComplete(localRepo, collectionFactory.RefListCollection())
 			if err != nil {
 				AbortWithJSONError(c, 500, fmt.Errorf("unable to publish: %s", err))
 			}
@@ -232,7 +232,7 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to publish: %s", err)
 		}
 
-		err = collection.Add(published)
+		err = collection.Add(published, collectionFactory.RefListCollection())
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to save to DB: %s", err)
 		}
@@ -313,7 +313,7 @@ func apiPublishUpdateSwitch(c *gin.Context) {
 				return
 			}
 
-			err2 = snapshotCollection.LoadComplete(snapshot)
+			err2 = snapshotCollection.LoadComplete(snapshot, collectionFactory.RefListCollection())
 			if err2 != nil {
 				AbortWithJSONError(c, 500, err2)
 				return
@@ -348,7 +348,7 @@ func apiPublishUpdateSwitch(c *gin.Context) {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 		}
 
-		err = collection.Update(published)
+		err = collection.Update(published, collectionFactory.RefListCollection())
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to save to DB: %s", err)
 		}

api/repos.go

@@ -82,7 +82,7 @@
 
 	collectionFactory := context.NewCollectionFactory()
 	collection := collectionFactory.LocalRepoCollection()
-	err := collection.Add(repo)
+	err := collection.Add(repo, collectionFactory.RefListCollection())
 	if err != nil {
 		AbortWithJSONError(c, 400, err)
 		return
@@ -132,7 +132,7 @@
 		repo.DefaultComponent = *b.DefaultComponent
 	}
 
-	err = collection.Update(repo)
+	err = collection.Update(repo, collectionFactory.RefListCollection())
 	if err != nil {
 		AbortWithJSONError(c, 500, err)
 		return
@@ -201,7 +201,7 @@
 		return
 	}
 
-	err = collection.LoadComplete(repo)
+	err = collection.LoadComplete(repo, collectionFactory.RefListCollection())
 	if err != nil {
 		AbortWithJSONError(c, 500, err)
 		return
@@ -229,7 +229,7 @@
 		return
 	}
 
-	err = collection.LoadComplete(repo)
+	err = collection.LoadComplete(repo, collectionFactory.RefListCollection())
 	if err != nil {
 		AbortWithJSONError(c, 500, err)
 		return
@@ -261,9 +261,9 @@
 			}
 		}
 
-		repo.UpdateRefList(deb.NewPackageRefListFromPackageList(list))
+		repo.UpdateRefList(deb.NewSplitRefListFromPackageList(list))
 
-		err = collectionFactory.LocalRepoCollection().Update(repo)
+		err = collectionFactory.LocalRepoCollection().Update(repo, collectionFactory.RefListCollection())
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to save: %s", err)
 		}
@@ -320,7 +320,7 @@
 		return
 	}
 
-	err = collection.LoadComplete(repo)
+	err = collection.LoadComplete(repo, collectionFactory.RefListCollection())
 	if err != nil {
 		AbortWithJSONError(c, 500, err)
 		return
@@ -369,9 +369,9 @@
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to import package files: %s", err)
 		}
 
-		repo.UpdateRefList(deb.NewPackageRefListFromPackageList(list))
+		repo.UpdateRefList(deb.NewSplitRefListFromPackageList(list))
 
-		err = collectionFactory.LocalRepoCollection().Update(repo)
+		err = collectionFactory.LocalRepoCollection().Update(repo, collectionFactory.RefListCollection())
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to save: %s", err)
 		}
@@ -489,7 +489,7 @@
 		_, failedFiles2, err = deb.ImportChangesFiles(
 			changesFiles, reporter, acceptUnsigned, ignoreSignature, forceReplace, noRemoveFiles, verifier,
 			repoTemplate, context.Progress(), collectionFactory.LocalRepoCollection(), collectionFactory.PackageCollection(),
-			context.PackagePool(), collectionFactory.ChecksumCollection, nil, query.Parse)
+			collectionFactory.RefListCollection(), context.PackagePool(), collectionFactory.ChecksumCollection, nil, query.Parse)
 		failedFiles = append(failedFiles, failedFiles2...)
 
 		if err != nil {