Pass BackupBases directly to ConsumeBackupCollections (#3876)

Shift things so BackupBases is passed directly to
the kopia package which then extracts information
from it. This allows for fine-grained control
over kopia-assisted incremental bases and merge
bases. Generating subtree paths from Reasons is
also shifted to the kopia package

Also expands tests for better coverage of various
incremental backup situations

Viewing by commit may help and individual commit
comments usually contain reasons for changes,
especially for test removal

---

#### Does this PR need a docs update or release note?

- [ ] ✅ Yes, it's included
- [ ] 🕐 Yes, but in a later PR
- [x] ⛔ No

#### Type of change

- [ ] 🌻 Feature
- [ ] 🐛 Bugfix
- [ ] 🗺️ Documentation
- [ ] 🤖 Supportability/Tests
- [ ] 💻 CI/Deployment
- [x] 🧹 Tech Debt/Cleanup

#### Issue(s)

* #2068

#### Test Plan

- [x] 💪 Manual
- [x] ⚡ Unit test
- [x] 💚 E2E

src/internal/kopia/inject/inject.go

@@ -16,7 +16,7 @@ type (
 		ConsumeBackupCollections(
 			ctx context.Context,
 			backupReasons []kopia.Reasoner,
-			bases []kopia.IncrementalBase,
+			bases kopia.BackupBases,
 			cs []data.BackupCollection,
 			pmr prefixmatcher.StringSetReader,
 			tags map[string]string,

src/internal/kopia/upload.go

@@ -20,6 +20,7 @@ import (
 	"github.com/kopia/kopia/fs/virtualfs"
 	"github.com/kopia/kopia/repo/manifest"
 	"github.com/kopia/kopia/snapshot/snapshotfs"
+	"golang.org/x/exp/maps"
 
 	"github.com/alcionai/corso/src/internal/common/prefixmatcher"
 	"github.com/alcionai/corso/src/internal/data"
@@ -970,10 +971,32 @@ func traverseBaseDir(
 	return nil
 }
 
+func logBaseInfo(ctx context.Context, m ManifestEntry) {
+	svcs := map[string]struct{}{}
+	cats := map[string]struct{}{}
+
+	for _, r := range m.Reasons {
+		svcs[r.Service().String()] = struct{}{}
+		cats[r.Category().String()] = struct{}{}
+	}
+
+	mbID, _ := m.GetTag(TagBackupID)
+	if len(mbID) == 0 {
+		mbID = "no_backup_id_tag"
+	}
+
+	logger.Ctx(ctx).Infow(
+		"using base for backup",
+		"base_snapshot_id", m.ID,
+		"services", maps.Keys(svcs),
+		"categories", maps.Keys(cats),
+		"base_backup_id", mbID)
+}
+
 func inflateBaseTree(
 	ctx context.Context,
 	loader snapshotLoader,
-	snap IncrementalBase,
+	snap ManifestEntry,
 	updatedPaths map[string]path.Path,
 	roots map[string]*treeMap,
 ) error {
@@ -996,13 +1019,25 @@ func inflateBaseTree(
 		return clues.New("snapshot root is not a directory").WithClues(ctx)
 	}
 
+	// Some logging to help track things.
+	logBaseInfo(ctx, snap)
+
 	// For each subtree corresponding to the tuple
 	// (resource owner, service, category) merge the directories in the base with
 	// what has been reported in the collections we got.
-	for _, subtreePath := range snap.SubtreePaths {
+	for _, r := range snap.Reasons {
+		ictx := clues.Add(
+			ctx,
+			"subtree_service", r.Service().String(),
+			"subtree_category", r.Category().String())
+
+		subtreePath, err := r.SubtreePath()
+		if err != nil {
+			return clues.Wrap(err, "building subtree path").WithClues(ictx)
+		}
+
 		// We're starting from the root directory so don't need it in the path.
 		pathElems := encodeElements(subtreePath.PopFront().Elements()...)
-		ictx := clues.Add(ctx, "subtree_path", subtreePath)
 
 		ent, err := snapshotfs.GetNestedEntry(ictx, dir, pathElems)
 		if err != nil {
@@ -1022,7 +1057,7 @@ func inflateBaseTree(
 		// This ensures that a migration on the directory prefix can complete.
 		// The prefix is the tenant/service/owner/category set, which remains
 		// otherwise unchecked in tree inflation below this point.
-		newSubtreePath := subtreePath
+		newSubtreePath := subtreePath.ToBuilder()
 		if p, ok := updatedPaths[subtreePath.String()]; ok {
 			newSubtreePath = p.ToBuilder()
 		}
@@ -1031,7 +1066,7 @@ func inflateBaseTree(
 			ictx,
 			0,
 			updatedPaths,
-			subtreePath.Dir(),
+			subtreePath.ToBuilder().Dir(),
 			newSubtreePath.Dir(),
 			subtreeDir,
 			roots,
@@ -1059,7 +1094,7 @@ func inflateBaseTree(
 func inflateDirTree(
 	ctx context.Context,
 	loader snapshotLoader,
-	baseSnaps []IncrementalBase,
+	baseSnaps []ManifestEntry,
 	collections []data.BackupCollection,
 	globalExcludeSet prefixmatcher.StringSetReader,
 	progress *corsoProgress,

src/internal/kopia/upload_test.go

@@ -946,21 +946,22 @@ func (msw *mockSnapshotWalker) SnapshotRoot(*snapshot.Manifest) (fs.Entry, error
 	return msw.snapshotRoot, nil
 }
 
-func mockIncrementalBase(
+func makeManifestEntry(
 	id, tenant, resourceOwner string,
 	service path.ServiceType,
 	categories ...path.CategoryType,
-) IncrementalBase {
+) ManifestEntry {
-	stps := []*path.Builder{}
+	var reasons []Reasoner
+
 	for _, c := range categories {
-		stps = append(stps, path.Builder{}.Append(tenant, service.String(), resourceOwner, c.String()))
+		reasons = append(reasons, NewReason(tenant, resourceOwner, service, c))
 	}
 
-	return IncrementalBase{
+	return ManifestEntry{
 		Manifest: &snapshot.Manifest{
 			ID: manifest.ID(id),
 		},
-		SubtreePaths: stps,
+		Reasons: reasons,
 	}
 }
 
@@ -1331,8 +1332,8 @@ func (suite *HierarchyBuilderUnitSuite) TestBuildDirectoryTreeSingleSubtree() {
 			dirTree, err := inflateDirTree(
 				ctx,
 				msw,
-				[]IncrementalBase{
+				[]ManifestEntry{
-					mockIncrementalBase("", testTenant, testUser, path.ExchangeService, path.EmailCategory),
+					makeManifestEntry("", testTenant, testUser, path.ExchangeService, path.EmailCategory),
 				},
 				test.inputCollections(),
 				pmMock.NewPrefixMap(nil),
@@ -2260,8 +2261,8 @@ func (suite *HierarchyBuilderUnitSuite) TestBuildDirectoryTreeMultipleSubdirecto
 			dirTree, err := inflateDirTree(
 				ctx,
 				msw,
-				[]IncrementalBase{
+				[]ManifestEntry{
-					mockIncrementalBase("", testTenant, testUser, path.ExchangeService, path.EmailCategory),
+					makeManifestEntry("", testTenant, testUser, path.ExchangeService, path.EmailCategory),
 				},
 				test.inputCollections(t),
 				ie,
@@ -2425,8 +2426,8 @@ func (suite *HierarchyBuilderUnitSuite) TestBuildDirectoryTreeSkipsDeletedSubtre
 	dirTree, err := inflateDirTree(
 		ctx,
 		msw,
-		[]IncrementalBase{
+		[]ManifestEntry{
-			mockIncrementalBase("", testTenant, testUser, path.ExchangeService, path.EmailCategory),
+			makeManifestEntry("", testTenant, testUser, path.ExchangeService, path.EmailCategory),
 		},
 		collections,
 		pmMock.NewPrefixMap(nil),
@@ -2531,8 +2532,8 @@ func (suite *HierarchyBuilderUnitSuite) TestBuildDirectoryTree_HandleEmptyBase()
 	dirTree, err := inflateDirTree(
 		ctx,
 		msw,
-		[]IncrementalBase{
+		[]ManifestEntry{
-			mockIncrementalBase("", testTenant, testUser, path.ExchangeService, path.EmailCategory),
+			makeManifestEntry("", testTenant, testUser, path.ExchangeService, path.EmailCategory),
 		},
 		collections,
 		pmMock.NewPrefixMap(nil),
@@ -2782,9 +2783,9 @@ func (suite *HierarchyBuilderUnitSuite) TestBuildDirectoryTreeSelectsCorrectSubt
 	dirTree, err := inflateDirTree(
 		ctx,
 		msw,
-		[]IncrementalBase{
+		[]ManifestEntry{
-			mockIncrementalBase("id1", testTenant, testUser, path.ExchangeService, path.ContactsCategory),
+			makeManifestEntry("id1", testTenant, testUser, path.ExchangeService, path.ContactsCategory),
-			mockIncrementalBase("id2", testTenant, testUser, path.ExchangeService, path.EmailCategory),
+			makeManifestEntry("id2", testTenant, testUser, path.ExchangeService, path.EmailCategory),
 		},
 		collections,
 		pmMock.NewPrefixMap(nil),
@@ -2948,8 +2949,8 @@ func (suite *HierarchyBuilderUnitSuite) TestBuildDirectoryTreeSelectsMigrateSubt
 	dirTree, err := inflateDirTree(
 		ctx,
 		msw,
-		[]IncrementalBase{
+		[]ManifestEntry{
-			mockIncrementalBase("id1", testTenant, testUser, path.ExchangeService, path.EmailCategory, path.ContactsCategory),
+			makeManifestEntry("id1", testTenant, testUser, path.ExchangeService, path.EmailCategory, path.ContactsCategory),
 		},
 		[]data.BackupCollection{mce, mcc},
 		pmMock.NewPrefixMap(nil),

src/internal/kopia/wrapper.go

@@ -143,7 +143,7 @@ type IncrementalBase struct {
 func (w Wrapper) ConsumeBackupCollections(
 	ctx context.Context,
 	backupReasons []Reasoner,
-	previousSnapshots []IncrementalBase,
+	bases BackupBases,
 	collections []data.BackupCollection,
 	globalExcludeSet prefixmatcher.StringSetReader,
 	additionalTags map[string]string,
@@ -172,15 +172,23 @@ func (w Wrapper) ConsumeBackupCollections(
 	// When running an incremental backup, we need to pass the prior
 	// snapshot bases into inflateDirTree so that the new snapshot
 	// includes historical data.
-	var base []IncrementalBase
+	var (
+		mergeBase  []ManifestEntry
+		assistBase []ManifestEntry
+	)
+
+	if bases != nil {
 		if buildTreeWithBase {
-		base = previousSnapshots
+			mergeBase = bases.MergeBases()
+		}
+
+		assistBase = bases.AssistBases()
 	}
 
 	dirTree, err := inflateDirTree(
 		ctx,
 		w.c,
-		base,
+		mergeBase,
 		collections,
 		globalExcludeSet,
 		progress)
@@ -203,7 +211,7 @@ func (w Wrapper) ConsumeBackupCollections(
 
 	s, err := w.makeSnapshotWithRoot(
 		ctx,
-		previousSnapshots,
+		assistBase,
 		dirTree,
 		tags,
 		progress)
@@ -216,7 +224,7 @@ func (w Wrapper) ConsumeBackupCollections(
 
 func (w Wrapper) makeSnapshotWithRoot(
 	ctx context.Context,
-	prevSnapEntries []IncrementalBase,
+	prevSnapEntries []ManifestEntry,
 	root fs.Directory,
 	addlTags map[string]string,
 	progress *corsoProgress,
@@ -236,8 +244,8 @@ func (w Wrapper) makeSnapshotWithRoot(
 
 	ctx = clues.Add(
 		ctx,
-		"len_prev_base_snapshots", len(prevSnapEntries),
+		"num_assist_snapshots", len(prevSnapEntries),
-		"assist_snap_ids", snapIDs,
+		"assist_snapshot_ids", snapIDs,
 		"additional_tags", addlTags)
 
 	if len(snapIDs) > 0 {

src/internal/kopia/wrapper_test.go

@@ -696,6 +696,24 @@ func (suite *KopiaIntegrationSuite) TestBackupCollections() {
 			42),
 	}
 
+	c1 := exchMock.NewCollection(
+		suite.storePath1,
+		suite.locPath1,
+		0)
+	c1.ColState = data.NotMovedState
+	c1.PrevPath = suite.storePath1
+
+	c2 := exchMock.NewCollection(
+		suite.storePath2,
+		suite.locPath2,
+		0)
+	c2.ColState = data.NotMovedState
+	c2.PrevPath = suite.storePath2
+
+	// Make empty collections at the same locations to force a backup with no
+	// changes. Needed to ensure we force a backup even if nothing has changed.
+	emptyCollections := []data.BackupCollection{c1, c2}
+
 	// tags that are supplied by the caller. This includes basic tags to support
 	// lookups and extra tags the caller may want to apply.
 	tags := map[string]string{
@@ -730,86 +748,219 @@ func (suite *KopiaIntegrationSuite) TestBackupCollections() {
 
 	expectedTags = normalizeTagKVs(expectedTags)
 
-	table := []struct {
+	type testCase struct {
 		name                  string
+		baseBackups           func(base ManifestEntry) BackupBases
+		collections           []data.BackupCollection
 		expectedUploadedFiles int
 		expectedCachedFiles   int
+		// We're either going to get details entries or entries in the details
+		// merger. Details is populated when there's entries in the collection. The
+		// details merger is populated for cached entries. The details merger
+		// doesn't count folders, only items.
+		//
+		// Setting this to true looks for details merger entries. Setting it to
+		// false looks for details entries.
+		expectMerge bool
 		// Whether entries in the resulting details should be marked as updated.
-		deetsUpdated bool
+		deetsUpdated     assert.BoolAssertionFunc
-	}{
+		hashedBytesCheck assert.ValueAssertionFunc
-		{
+		// Range of bytes (inclusive) to expect as uploaded. A little fragile, but
-			name:                  "Uncached",
+		// allows us to differentiate between content that wasn't uploaded due to
-			expectedUploadedFiles: 47,
+		// being cached/deduped/skipped due to existing dir entries and stuff that
-			expectedCachedFiles:   0,
+		// was actually pushed to S3.
-			deetsUpdated:          true,
+		uploadedBytes []int64
-		},
-		{
-			name:                  "Cached",
-			expectedUploadedFiles: 0,
-			expectedCachedFiles:   47,
-			deetsUpdated:          false,
-		},
 	}
 
-	prevSnaps := []IncrementalBase{}
+	// Initial backup. All files should be considered new by kopia.
+	baseBackupCase := testCase{
+		name: "Uncached",
+		baseBackups: func(ManifestEntry) BackupBases {
+			return NewMockBackupBases()
+		},
+		collections:           collections,
+		expectedUploadedFiles: 47,
+		expectedCachedFiles:   0,
+		deetsUpdated:          assert.True,
+		hashedBytesCheck:      assert.NotZero,
+		uploadedBytes:         []int64{8000, 10000},
+	}
+
+	runAndTestBackup := func(test testCase, base ManifestEntry) ManifestEntry {
+		var res ManifestEntry
 
-	for _, test := range table {
 		suite.Run(test.name, func() {
 			t := suite.T()
 
-			stats, deets, _, err := suite.w.ConsumeBackupCollections(
+			ctx, flush := tester.NewContext(t)
-				suite.ctx,
+			defer flush()
+
+			bbs := test.baseBackups(base)
+
+			stats, deets, deetsMerger, err := suite.w.ConsumeBackupCollections(
+				ctx,
 				reasons,
-				prevSnaps,
+				bbs,
-				collections,
+				test.collections,
 				nil,
 				tags,
 				true,
 				fault.New(true))
-			assert.NoError(t, err, clues.ToCore(err))
+			require.NoError(t, err, clues.ToCore(err))
 
 			assert.Equal(t, test.expectedUploadedFiles, stats.TotalFileCount, "total files")
 			assert.Equal(t, test.expectedUploadedFiles, stats.UncachedFileCount, "uncached files")
 			assert.Equal(t, test.expectedCachedFiles, stats.CachedFileCount, "cached files")
-			assert.Equal(t, 6, stats.TotalDirectoryCount)
+			assert.Equal(t, 4+len(test.collections), stats.TotalDirectoryCount, "directory count")
 			assert.Equal(t, 0, stats.IgnoredErrorCount)
 			assert.Equal(t, 0, stats.ErrorCount)
 			assert.False(t, stats.Incomplete)
+			test.hashedBytesCheck(t, stats.TotalHashedBytes, "hashed bytes")
+			assert.LessOrEqual(
+				t,
+				test.uploadedBytes[0],
+				stats.TotalUploadedBytes,
+				"low end of uploaded bytes")
+			assert.GreaterOrEqual(
+				t,
+				test.uploadedBytes[1],
+				stats.TotalUploadedBytes,
+				"high end of uploaded bytes")
+
+			if test.expectMerge {
+				assert.Empty(t, deets.Details().Entries, "details entries")
+				assert.Equal(
+					t,
+					test.expectedUploadedFiles+test.expectedCachedFiles,
+					deetsMerger.ItemsToMerge(),
+					"details merger entries")
+			} else {
+				assert.Zero(t, deetsMerger.ItemsToMerge(), "details merger entries")
 
-			// 47 file and 2 folder entries.
 				details := deets.Details().Entries
 				assert.Len(
 					t,
 					details,
+					// 47 file and 2 folder entries.
 					test.expectedUploadedFiles+test.expectedCachedFiles+2,
 				)
 
 				for _, entry := range details {
-				assert.Equal(t, test.deetsUpdated, entry.Updated)
+					test.deetsUpdated(t, entry.Updated)
+				}
 			}
 
 			checkSnapshotTags(
 				t,
-				suite.ctx,
+				ctx,
 				suite.w.c,
 				expectedTags,
 				stats.SnapshotID,
 			)
 
 			snap, err := snapshot.LoadSnapshot(
-				suite.ctx,
+				ctx,
 				suite.w.c,
 				manifest.ID(stats.SnapshotID),
 			)
 			require.NoError(t, err, clues.ToCore(err))
 
-			prevSnaps = append(prevSnaps, IncrementalBase{
+			res = ManifestEntry{
 				Manifest: snap,
-				SubtreePaths: []*path.Builder{
+				Reasons:  reasons,
-					suite.storePath1.ToBuilder().Dir(),
+			}
+		})
+
+		return res
+	}
+
+	base := runAndTestBackup(baseBackupCase, ManifestEntry{})
+
+	table := []testCase{
+		{
+			name: "Kopia Assist And Merge All Files Changed",
+			baseBackups: func(base ManifestEntry) BackupBases {
+				return NewMockBackupBases().WithMergeBases(base)
 			},
-			})
+			collections:           collections,
-		})
+			expectedUploadedFiles: 0,
+			expectedCachedFiles:   47,
+			deetsUpdated:          assert.False,
+			hashedBytesCheck:      assert.Zero,
+			uploadedBytes:         []int64{4000, 6000},
+		},
+		{
+			name: "Kopia Assist And Merge No Files Changed",
+			baseBackups: func(base ManifestEntry) BackupBases {
+				return NewMockBackupBases().WithMergeBases(base)
+			},
+			// Pass in empty collections to force a backup. Otherwise we'll skip
+			// actually trying to do anything because we'll see there's nothing that
+			// changed. The real goal is to get it to deal with the merged collections
+			// again though.
+			collections: emptyCollections,
+			// Should hit cached check prior to dir entry check so we see them as
+			// cached.
+			expectedUploadedFiles: 0,
+			expectedCachedFiles:   47,
+			// Entries go into the details merger because we never materialize details
+			// info for the items since they're from the base.
+			expectMerge: true,
+			// Not used since there's no details entries.
+			deetsUpdated:     assert.False,
+			hashedBytesCheck: assert.Zero,
+			uploadedBytes:    []int64{4000, 6000},
+		},
+		{
+			name: "Kopia Assist Only",
+			baseBackups: func(base ManifestEntry) BackupBases {
+				return NewMockBackupBases().WithAssistBases(base)
+			},
+			collections:           collections,
+			expectedUploadedFiles: 0,
+			expectedCachedFiles:   47,
+			deetsUpdated:          assert.False,
+			hashedBytesCheck:      assert.Zero,
+			uploadedBytes:         []int64{4000, 6000},
+		},
+		{
+			name: "Merge Only",
+			baseBackups: func(base ManifestEntry) BackupBases {
+				return NewMockBackupBases().WithMergeBases(base).ClearMockAssistBases()
+			},
+			// Pass in empty collections to force a backup. Otherwise we'll skip
+			// actually trying to do anything because we'll see there's nothing that
+			// changed. The real goal is to get it to deal with the merged collections
+			// again though.
+			collections:           emptyCollections,
+			expectedUploadedFiles: 47,
+			expectedCachedFiles:   0,
+			expectMerge:           true,
+			// Not used since there's no details entries.
+			deetsUpdated: assert.False,
+			// Kopia still counts these bytes as "hashed" even though it shouldn't
+			// read the file data since they already have dir entries it can reuse.
+			hashedBytesCheck: assert.NotZero,
+			uploadedBytes:    []int64{4000, 6000},
+		},
+		{
+			name: "Content Hash Only",
+			baseBackups: func(base ManifestEntry) BackupBases {
+				return NewMockBackupBases()
+			},
+			collections:           collections,
+			expectedUploadedFiles: 47,
+			expectedCachedFiles:   0,
+			// Marked as updated because we still fall into the uploadFile handler in
+			// kopia instead of the cachedFile handler.
+			deetsUpdated:     assert.True,
+			hashedBytesCheck: assert.NotZero,
+			uploadedBytes:    []int64{4000, 6000},
+		},
+	}
+
+	for _, test := range table {
+		runAndTestBackup(test, base)
 	}
 }
 
@@ -938,7 +1089,7 @@ func (suite *KopiaIntegrationSuite) TestBackupCollections_NoDetailsForMeta() {
 		},
 	}
 
-	prevSnaps := []IncrementalBase{}
+	prevSnaps := NewMockBackupBases()
 
 	for _, test := range table {
 		suite.Run(test.name, func() {
@@ -1000,12 +1151,12 @@ func (suite *KopiaIntegrationSuite) TestBackupCollections_NoDetailsForMeta() {
 				manifest.ID(stats.SnapshotID))
 			require.NoError(t, err, clues.ToCore(err))
 
-			prevSnaps = append(prevSnaps, IncrementalBase{
+			prevSnaps.WithMergeBases(
+				ManifestEntry{
 					Manifest: snap,
-				SubtreePaths: []*path.Builder{
+					Reasons:  reasons,
-					storePath.ToBuilder().Dir(),
 				},
-			})
+			)
 		})
 	}
 }
@@ -1424,17 +1575,6 @@ func (c *i64counter) Count(i int64) {
 func (suite *KopiaSimpleRepoIntegrationSuite) TestBackupExcludeItem() {
 	r := NewReason(testTenant, testUser, path.ExchangeService, path.EmailCategory)
 
-	subtreePathTmp, err := path.Build(
-		testTenant,
-		testUser,
-		path.ExchangeService,
-		path.EmailCategory,
-		false,
-		"tmp")
-	require.NoError(suite.T(), err, clues.ToCore(err))
-
-	subtreePath := subtreePathTmp.ToBuilder().Dir()
-
 	man, err := suite.w.c.LoadSnapshot(suite.ctx, suite.snapshotID)
 	require.NoError(suite.T(), err, "getting base snapshot: %v", clues.ToCore(err))
 
@@ -1527,14 +1667,12 @@ func (suite *KopiaSimpleRepoIntegrationSuite) TestBackupExcludeItem() {
 			stats, _, _, err := suite.w.ConsumeBackupCollections(
 				suite.ctx,
 				[]Reasoner{r},
-				[]IncrementalBase{
+				NewMockBackupBases().WithMergeBases(
-					{
+					ManifestEntry{
 						Manifest: man,
-						SubtreePaths: []*path.Builder{
+						Reasons:  []Reasoner{r},
-							subtreePath,
-						},
-					},
 					},
+				),
 				test.cols(),
 				excluded,
 				nil,

src/internal/operations/backup.go

@@ -6,7 +6,6 @@ import (
 
 	"github.com/alcionai/clues"
 	"github.com/google/uuid"
-	"github.com/kopia/kopia/repo/manifest"
 
 	"github.com/alcionai/corso/src/internal/common/crash"
 	"github.com/alcionai/corso/src/internal/common/dttm"
@@ -449,26 +448,6 @@ func selectorToReasons(
 	return reasons
 }
 
-func builderFromReason(ctx context.Context, tenant string, r kopia.Reasoner) (*path.Builder, error) {
-	ctx = clues.Add(ctx, "category", r.Category().String())
-
-	// This is hacky, but we want the path package to format the path the right
-	// way (e.x. proper order for service, category, etc), but we don't care about
-	// the folders after the prefix.
-	p, err := path.Build(
-		tenant,
-		r.ProtectedResource(),
-		r.Service(),
-		r.Category(),
-		false,
-		"tmp")
-	if err != nil {
-		return nil, clues.Wrap(err, "building path").WithClues(ctx)
-	}
-
-	return p.ToBuilder().Dir(), nil
-}
-
 // calls kopia to backup the collections of data
 func consumeBackupCollections(
 	ctx context.Context,
@@ -495,85 +474,10 @@ func consumeBackupCollections(
 		kopia.TagBackupCategory: "",
 	}
 
-	// AssistBases should be the upper bound for how many snapshots we pass in.
-	bases := make([]kopia.IncrementalBase, 0, len(bbs.AssistBases()))
-	// Track IDs we've seen already so we don't accidentally duplicate some
-	// manifests. This can be removed when we move the code below into the kopia
-	// package.
-	ids := map[manifest.ID]struct{}{}
-
-	var mb []kopia.ManifestEntry
-
-	if bbs != nil {
-		mb = bbs.MergeBases()
-	}
-
-	// TODO(ashmrtn): Make a wrapper for Reson that allows adding a tenant and
-	// make a function that will spit out a prefix that includes the tenant. With
-	// that done this code can be moved to kopia wrapper since it's really more
-	// specific to that.
-	for _, m := range mb {
-		paths := make([]*path.Builder, 0, len(m.Reasons))
-		services := map[string]struct{}{}
-		categories := map[string]struct{}{}
-
-		for _, reason := range m.Reasons {
-			pb, err := builderFromReason(ctx, tenantID, reason)
-			if err != nil {
-				return nil, nil, nil, clues.Wrap(err, "getting subtree paths for bases")
-			}
-
-			paths = append(paths, pb)
-			services[reason.Service().String()] = struct{}{}
-			categories[reason.Category().String()] = struct{}{}
-		}
-
-		ids[m.ID] = struct{}{}
-
-		bases = append(bases, kopia.IncrementalBase{
-			Manifest:     m.Manifest,
-			SubtreePaths: paths,
-		})
-
-		svcs := make([]string, 0, len(services))
-		for k := range services {
-			svcs = append(svcs, k)
-		}
-
-		cats := make([]string, 0, len(categories))
-		for k := range categories {
-			cats = append(cats, k)
-		}
-
-		mbID, ok := m.GetTag(kopia.TagBackupID)
-		if !ok {
-			mbID = "no_backup_id_tag"
-		}
-
-		logger.Ctx(ctx).Infow(
-			"using base for backup",
-			"base_snapshot_id", m.ID,
-			"services", svcs,
-			"categories", cats,
-			"base_backup_id", mbID)
-	}
-
-	// At the moment kopia assisted snapshots are in the same set as merge bases.
-	// When we fixup generating subtree paths we can remove this.
-	if bbs != nil {
-		for _, ab := range bbs.AssistBases() {
-			if _, ok := ids[ab.ID]; ok {
-				continue
-			}
-
-			bases = append(bases, kopia.IncrementalBase{Manifest: ab.Manifest})
-		}
-	}
-
 	kopiaStats, deets, itemsSourcedFromBase, err := bc.ConsumeBackupCollections(
 		ctx,
 		reasons,
-		bases,
+		bbs,
 		cs,
 		pmr,
 		tags,
@@ -581,7 +485,7 @@ func consumeBackupCollections(
 		errs)
 	if err != nil {
 		if kopiaStats == nil {
-			return nil, nil, nil, err
+			return nil, nil, nil, clues.Stack(err)
 		}
 
 		return nil, nil, nil, clues.Stack(err).With(

src/internal/operations/backup_test.go

@@ -108,7 +108,7 @@ func checkPaths(t *testing.T, expected, got []path.Path) {
 type mockBackupConsumer struct {
 	checkFunc func(
 		backupReasons []kopia.Reasoner,
-		bases []kopia.IncrementalBase,
+		bases kopia.BackupBases,
 		cs []data.BackupCollection,
 		tags map[string]string,
 		buildTreeWithBase bool)
@@ -117,7 +117,7 @@ type mockBackupConsumer struct {
 func (mbu mockBackupConsumer) ConsumeBackupCollections(
 	ctx context.Context,
 	backupReasons []kopia.Reasoner,
-	bases []kopia.IncrementalBase,
+	bases kopia.BackupBases,
 	cs []data.BackupCollection,
 	excluded prefixmatcher.StringSetReader,
 	tags map[string]string,
@@ -390,120 +390,35 @@ func (suite *BackupOpUnitSuite) TestBackupOperation_PersistResults() {
 
 func (suite *BackupOpUnitSuite) TestBackupOperation_ConsumeBackupDataCollections_Paths() {
 	var (
+		t = suite.T()
+
 		tenant        = "a-tenant"
 		resourceOwner = "a-user"
 
-		emailBuilder = path.Builder{}.Append(
-			tenant,
-			path.ExchangeService.String(),
-			resourceOwner,
-			path.EmailCategory.String())
-		contactsBuilder = path.Builder{}.Append(
-			tenant,
-			path.ExchangeService.String(),
-			resourceOwner,
-			path.ContactsCategory.String())
-
 		emailReason = kopia.NewReason(
-			"",
+			tenant,
 			resourceOwner,
 			path.ExchangeService,
 			path.EmailCategory)
 		contactsReason = kopia.NewReason(
-			"",
+			tenant,
 			resourceOwner,
 			path.ExchangeService,
 			path.ContactsCategory)
 
+		reasons = []kopia.Reasoner{
+			emailReason,
+			contactsReason,
+		}
+
 		manifest1 = &snapshot.Manifest{
 			ID: "id1",
 		}
 		manifest2 = &snapshot.Manifest{
 			ID: "id2",
 		}
-	)
 
-	table := []struct {
+		bases = kopia.NewMockBackupBases().WithMergeBases(
-		name string
-		// Backup model is untouched in this test so there's no need to populate it.
-		input    kopia.BackupBases
-		expected []kopia.IncrementalBase
-	}{
-		{
-			name: "SingleManifestSingleReason",
-			input: kopia.NewMockBackupBases().WithMergeBases(
-				kopia.ManifestEntry{
-					Manifest: manifest1,
-					Reasons: []kopia.Reasoner{
-						emailReason,
-					},
-				}).ClearMockAssistBases(),
-			expected: []kopia.IncrementalBase{
-				{
-					Manifest: manifest1,
-					SubtreePaths: []*path.Builder{
-						emailBuilder,
-					},
-				},
-			},
-		},
-		{
-			name: "SingleManifestMultipleReasons",
-			input: kopia.NewMockBackupBases().WithMergeBases(
-				kopia.ManifestEntry{
-					Manifest: manifest1,
-					Reasons: []kopia.Reasoner{
-						emailReason,
-						contactsReason,
-					},
-				}).ClearMockAssistBases(),
-			expected: []kopia.IncrementalBase{
-				{
-					Manifest: manifest1,
-					SubtreePaths: []*path.Builder{
-						emailBuilder,
-						contactsBuilder,
-					},
-				},
-			},
-		},
-		{
-			name: "MultipleManifestsMultipleReasons",
-			input: kopia.NewMockBackupBases().WithMergeBases(
-				kopia.ManifestEntry{
-					Manifest: manifest1,
-					Reasons: []kopia.Reasoner{
-						emailReason,
-						contactsReason,
-					},
-				},
-				kopia.ManifestEntry{
-					Manifest: manifest2,
-					Reasons: []kopia.Reasoner{
-						emailReason,
-						contactsReason,
-					},
-				}).ClearMockAssistBases(),
-			expected: []kopia.IncrementalBase{
-				{
-					Manifest: manifest1,
-					SubtreePaths: []*path.Builder{
-						emailBuilder,
-						contactsBuilder,
-					},
-				},
-				{
-					Manifest: manifest2,
-					SubtreePaths: []*path.Builder{
-						emailBuilder,
-						contactsBuilder,
-					},
-				},
-			},
-		},
-		{
-			name: "Single Manifest Single Reason With Assist Base",
-			input: kopia.NewMockBackupBases().WithMergeBases(
 			kopia.ManifestEntry{
 				Manifest: manifest1,
 				Reasons: []kopia.Reasoner{
@@ -515,54 +430,44 @@ func (suite *BackupOpUnitSuite) TestBackupOperation_ConsumeBackupDataCollections
 				Reasons: []kopia.Reasoner{
 					contactsReason,
 				},
-				}),
+			})
-			expected: []kopia.IncrementalBase{
+
-				{
+		backupID     = model.StableID("foo")
-					Manifest: manifest1,
+		expectedTags = map[string]string{
-					SubtreePaths: []*path.Builder{
+			kopia.TagBackupID:       string(backupID),
-						emailBuilder,
+			kopia.TagBackupCategory: "",
-					},
-				},
-				{
-					Manifest: manifest2,
-				},
-			},
-		},
 		}
-
+	)
-	for _, test := range table {
-		suite.Run(test.name, func() {
-			t := suite.T()
-
-			ctx, flush := tester.NewContext(t)
-			defer flush()
 
 	mbu := &mockBackupConsumer{
 		checkFunc: func(
 			backupReasons []kopia.Reasoner,
-					bases []kopia.IncrementalBase,
+			gotBases kopia.BackupBases,
 			cs []data.BackupCollection,
-					tags map[string]string,
+			gotTags map[string]string,
 			buildTreeWithBase bool,
 		) {
-					assert.ElementsMatch(t, test.expected, bases)
+			kopia.AssertBackupBasesEqual(t, bases, gotBases)
+			assert.Equal(t, expectedTags, gotTags)
+			assert.ElementsMatch(t, reasons, backupReasons)
 		},
 	}
 
+	ctx, flush := tester.NewContext(t)
+	defer flush()
+
 	//nolint:errcheck
 	consumeBackupCollections(
 		ctx,
 		mbu,
 		tenant,
-				nil,
+		reasons,
-				test.input,
+		bases,
 		nil,
 		nil,
-				model.StableID(""),
+		backupID,
 		true,
 		fault.New(true))
-		})
-	}
 }
 
 func (suite *BackupOpUnitSuite) TestBackupOperation_MergeBackupDetails_AddsItems() {