Minor refactor for checking backup type when finding bases (#4592)

Use the new helper function to get the backup type and rewrite logic
to streamline checks/actions on the backup type

Hoists a check on the backup's snapshot ID higher in the code and
adds a test case for that

---

#### 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

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

#### Test Plan

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

src/internal/kopia/base_finder.go

@@ -197,6 +197,14 @@ func (b *baseFinder) findBasesInSet(
 
 		ictx = clues.Add(ictx, "ssid", ssid)
 
+		if bup.SnapshotID != string(man.ID) {
+			logger.Ctx(ictx).Infow(
+				"retrieved backup has empty or different snapshot ID from provided manifest",
+				"backup_snapshot_id", bup.SnapshotID)
+
+			continue
+		}
+
 		// If we've made it to this point then we're considering the backup
 		// complete as it has both an item data snapshot and a backup details
 		// snapshot.
@@ -207,31 +215,39 @@ func (b *baseFinder) findBasesInSet(
 		// 2. at most one assist base per reason.
 		// 3. it must be more recent than the merge backup for the reason, if
 		// a merge backup exists.
-
-		if b.isAssistBackupModel(ictx, bup) {
+		switch bup.Type() {
+		case model.AssistBackup:
+			// Only add an assist base if we haven't already found one.
 			if assistBase == nil {
+				logger.Ctx(ictx).Info("found assist base")
+
 				assistBase = &BackupBase{
 					Backup:           bup,
 					ItemDataSnapshot: man,
 					Reasons:          []identity.Reasoner{reason},
 				}
-
-				logger.Ctx(ictx).Info("found assist base")
 			}
 
-			// Skip if an assist base has already been selected.
-			continue
+		case model.MergeBackup:
+			logger.Ctx(ictx).Info("found merge base")
+
+			mergeBase = &BackupBase{
+				Backup:           bup,
+				ItemDataSnapshot: man,
+				Reasons:          []identity.Reasoner{reason},
+			}
+
+		default:
+			logger.Ctx(ictx).Infow(
+				"skipping backup with empty or invalid type for incremental backups",
+				"backup_type", bup.Type())
 		}
 
-		logger.Ctx(ictx).Info("found merge base")
-
-		mergeBase = &BackupBase{
-			Backup:           bup,
-			ItemDataSnapshot: man,
-			Reasons:          []identity.Reasoner{reason},
+		// Need to check here if we found a merge base because adding a break in the
+		// case-statement will just leave the case not the for-loop.
+		if mergeBase != nil {
+			break
 		}
-
-		break
 	}
 
 	if mergeBase == nil && assistBase == nil {
@@ -241,42 +257,6 @@ func (b *baseFinder) findBasesInSet(
 	return mergeBase, assistBase, nil
 }
 
-// isAssistBackupModel checks if the provided backup is an assist backup.
-func (b *baseFinder) isAssistBackupModel(
-	ctx context.Context,
-	bup *backup.Backup,
-) bool {
-	allTags := map[string]string{
-		model.BackupTypeTag: model.AssistBackup,
-	}
-
-	for k, v := range allTags {
-		if bup.Tags[k] != v {
-			// This is not an assist backup so we can just exit here.
-			logger.Ctx(ctx).Debugw(
-				"assist backup model missing tags",
-				"backup_id", bup.ID,
-				"tag", k,
-				"expected_value", v,
-				"actual_value", bup.Tags[k])
-
-			return false
-		}
-	}
-
-	// Check if it has a valid streamstore id and snapshot id.
-	if len(bup.StreamStoreID) == 0 || len(bup.SnapshotID) == 0 {
-		logger.Ctx(ctx).Infow(
-			"nil ssid or snapshot id in assist base",
-			"ssid", bup.StreamStoreID,
-			"snapshot_id", bup.SnapshotID)
-
-		return false
-	}
-
-	return true
-}
-
 func (b *baseFinder) getBase(
 	ctx context.Context,
 	r identity.Reasoner,

src/internal/kopia/base_finder_test.go

@@ -279,6 +279,13 @@ func (builder *baseInfoBuilder) legacyBackupDetails() *baseInfoBuilder {
 	return builder
 }
 
+func (builder *baseInfoBuilder) setBackupItemSnapshotID(
+	id string,
+) *baseInfoBuilder {
+	builder.info.backup.b.SnapshotID = id
+	return builder
+}
+
 func (builder *baseInfoBuilder) clearBackupDetails() *baseInfoBuilder {
 	builder.info.backup.b.DetailsID = ""
 	builder.info.backup.b.StreamStoreID = ""
@@ -444,6 +451,20 @@ func (suite *BaseFinderUnitSuite) TestGetBases() {
 				1: testUser1Mail,
 			},
 		},
+		{
+			name:  "Return Older Base If Snapshot ID Mismatch",
+			input: testUser1Mail,
+			data: []baseInfo{
+				newBaseInfoBuilder(2, testT2, testUser1Mail...).
+					setBackupItemSnapshotID("foo").
+					build(),
+				newBaseInfoBuilder(1, testT1, testUser1Mail...).
+					build(),
+			},
+			expectedMergeReasons: map[int][]identity.Reasoner{
+				1: testUser1Mail,
+			},
+		},
 		{
 			name:  "Old Backup Details Pointer",
 			input: testUser1Mail,