Split reflists to share their contents across snapshots [take 2]

api/api.go

@@ -233,7 +233,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 @@ import (
 	"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"
@@ -28,18 +29,22 @@ func apiDBCleanup(c *gin.Context) {
 
 		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(deb.RefListMigrationOptions{})
 
 		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
 		})
@@ -48,14 +53,14 @@ func apiDBCleanup(c *gin.Context) {
 		}
 
 		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
 		})
@@ -64,12 +69,14 @@ func apiDBCleanup(c *gin.Context) {
 		}
 
 		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
 		})
@@ -81,25 +88,43 @@ func apiDBCleanup(c *gin.Context) {
 			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 {
+					// < 0.6 saving w/o component name
+					if e == database.ErrNotFound && len(published.Sources) == 1 {
+						e = collectionFactory.RefListCollection().LoadCompleteAndMigrate(sl, published.RefKey(""), 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())
@@ -120,6 +145,28 @@ func apiDBCleanup(c *gin.Context) {
 			}
 		}
 
+		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/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

@@ -74,7 +74,7 @@ func apiMirrorsList(c *gin.Context) {
 
 	result := []remoteRepoResponse{}
 	err := collection.ForEach(func(repo *deb.RemoteRepo) error {
-		err := collection.LoadComplete(repo)
+		err := collection.LoadComplete(repo, collectionFactory.RefListCollection())
 		if err != nil {
 			return err
 		}
@@ -191,7 +191,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
@@ -283,7 +283,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))
 		return
@@ -315,7 +315,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))
 	}
@@ -498,7 +498,7 @@ func apiMirrorsEdit(c *gin.Context) {
 		}
 	}
 
-	err = collection.Update(repo)
+	err = collection.Update(repo, collectionFactory.RefListCollection())
 	if err != nil {
 		AbortWithJSONError(c, 500, fmt.Errorf("unable to edit: %s", err))
 		return
@@ -584,6 +584,7 @@ func apiMirrorsUpdate(c *gin.Context) {
 		// Phase 2: Inside task lock - create fresh factory
 		taskCollectionFactory := context.NewCollectionFactory()
 		taskCollection := taskCollectionFactory.RemoteRepoCollection()
+		taskRefListCollection := taskCollectionFactory.RefListCollection()
 
 		// Fresh load after lock acquired (use captured `name` variable, not gin context)
 		remote, err := taskCollection.ByName(name)
@@ -650,12 +651,12 @@ func apiMirrorsUpdate(c *gin.Context) {
 			e := context.ReOpenDatabase()
 			if e == nil {
 				remote.MarkAsIdle()
-				_ = taskCollection.Update(remote)
+				_ = taskCollection.Update(remote, taskRefListCollection)
 			}
 		}()
 
 		remote.MarkAsUpdating()
-		err = taskCollection.Update(remote)
+		err = taskCollection.Update(remote, taskRefListCollection)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 		}
@@ -813,7 +814,7 @@ func apiMirrorsUpdate(c *gin.Context) {
 
 		log.Info().Msgf("%s: Finalizing download...", b.Name)
 		_ = remote.FinalizeDownload(taskCollectionFactory, out)
-		err = taskCollection.Update(remote)
+		err = taskCollection.Update(remote, taskRefListCollection)
 		if err != nil {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to update: %s", err)
 		}

api/mirror_test.go

@@ -35,7 +35,7 @@ func (s *MirrorSuite) TestCreateMirrorFlatWithAppStream(c *C) {
 		"Name":              "test-flat-appstream",
 		"ArchiveURL":        "http://example.com/repo/",
 		"Distribution":      "./",
-		"DownloadAppStream":  true,
+		"DownloadAppStream": true,
 	})
 	c.Assert(err, IsNil)
 
@@ -59,14 +59,17 @@ func (s *MirrorSuite) TestCreateMirror(c *C) {
 }
 
 func (s *MirrorSuite) TestGetMirrorsIncludesNumPackages(c *C) {
-	collection := s.context.NewCollectionFactory().RemoteRepoCollection()
+	collectionFactory := s.context.NewCollectionFactory()
+	collection := collectionFactory.RemoteRepoCollection()
+	reflistCollection := collectionFactory.RefListCollection()
 
 	repo, err := deb.NewRemoteRepo("count-mirror", "http://example.com/debian", "stable", []string{"main"}, []string{}, false, false, false, false)
 	c.Assert(err, IsNil)
 
-	err = collection.Add(repo)
+	err = collection.Add(repo, reflistCollection)
+	c.Assert(err, IsNil)
+	err = reflistCollection.Update(makePackageRefList(c), repo.RefKey())
 	c.Assert(err, IsNil)
-	putRawDBValue(c, &s.APISuite, repo.RefKey(), makePackageRefList(c).Encode())
 
 	response, err := s.HTTPRequest("GET", "/api/mirrors", nil)
 	c.Assert(err, IsNil)
@@ -91,11 +94,13 @@ func (s *MirrorSuite) TestGetMirrorsIncludesNumPackages(c *C) {
 }
 
 func (s *MirrorSuite) TestGetMirrorsReturns500OnCorruptRefList(c *C) {
-	collection := s.context.NewCollectionFactory().RemoteRepoCollection()
+	collectionFactory := s.context.NewCollectionFactory()
+	collection := collectionFactory.RemoteRepoCollection()
+	reflistCollection := collectionFactory.RefListCollection()
 
 	repo, err := deb.NewRemoteRepo("broken-mirror", "http://example.com/debian", "stable", []string{"main"}, []string{}, false, false, false, false)
 	c.Assert(err, IsNil)
-	c.Assert(collection.Add(repo), IsNil)
+	c.Assert(collection.Add(repo, reflistCollection), IsNil)
 	putRawDBValue(c, &s.APISuite, repo.RefKey(), []byte("not-msgpack"))
 
 	response, err := s.HTTPRequest("GET", "/api/mirrors", nil)

api/package_count_test_helpers_test.go

@@ -5,15 +5,15 @@ import (
 	. "gopkg.in/check.v1"
 )
 
-func makePackageRefList(c *C) *deb.PackageRefList {
+func makePackageRefList(c *C) *deb.SplitRefList {
 	list := deb.NewPackageList()
 	c.Assert(list.Add(&deb.Package{Name: "libcount", Version: "1.0", Architecture: "amd64"}), IsNil)
 	c.Assert(list.Add(&deb.Package{Name: "appcount", Version: "2.0", Architecture: "all"}), IsNil)
-	return deb.NewPackageRefListFromPackageList(list)
+	return deb.NewSplitRefListFromPackageList(list)
 }
 
 func putRawDBValue(c *C, s *APISuite, key []byte, value []byte) {
 	db, err := s.context.Database()
 	c.Assert(err, IsNil)
 	c.Assert(db.Put(key, value), IsNil)
-}
+}