backup: fix elide skipped layers to account for new incremental path suffix

Due to the changes introduced in #143226, the order of the backup
manifests passed to `ElideSkippedLayers` is now changed. Whereas
previously a compacted backup would always appear *after* any backup
with the same end time, compacted backups now appear *before*. This
meant that during restore, compacted backups ended up always being
skipped and never used.

To fix this, prior to running the `ElideSkippedLayers` algorithm, we now
remove duplicate backups (i.e. same end time) and keep the backup with
the earliest start time (aka the most compacted backup).

Epic: None

Release note: None

Author: kev-cao

pkg/backup/backupinfo/manifest_handling.go

@@ -887,11 +887,19 @@ func ValidateEndTimeAndTruncate(
 	endTime hlc.Timestamp,
 	includeSkipped bool,
 ) ([]string, []backuppb.BackupManifest, []jobspb.RestoreDetails_BackupLocalityInfo, error) {
+
 	if endTime.IsEmpty() {
 		if includeSkipped {
 			return defaultURIs, mainBackupManifests, localityInfo, nil
 		}
-		return ElideSkippedLayers(defaultURIs, mainBackupManifests, localityInfo)
+		uris, manifests, locality, err := ElideSkippedLayers(defaultURIs, mainBackupManifests, localityInfo)
+		if err != nil {
+			return nil, nil, nil, err
+		}
+		if err := validateContinuity(manifests, endTime); err != nil {
+			return nil, nil, nil, err
+		}
+		return uris, manifests, locality, nil
 	}
 	for i, b := range mainBackupManifests {
 		// Find the backup that covers the requested time.
@@ -930,7 +938,16 @@ func ValidateEndTimeAndTruncate(
 		if includeSkipped {
 			return defaultURIs[:i+1], mainBackupManifests[:i+1], localityInfo[:i+1], nil
 		}
-		return ElideSkippedLayers(defaultURIs[:i+1], mainBackupManifests[:i+1], localityInfo[:i+1])
+		uris, manifests, locality, err := ElideSkippedLayers(
+			defaultURIs[:i+1], mainBackupManifests[:i+1], localityInfo[:i+1],
+		)
+		if err != nil {
+			return nil, nil, nil, err
+		}
+		if err := validateContinuity(manifests, endTime); err != nil {
+			return nil, nil, nil, err
+		}
+		return uris, manifests, locality, nil
 
 	}
 
@@ -939,10 +956,42 @@ func ValidateEndTimeAndTruncate(
 	)
 }
 
-// ElideSkippedLayers removes backups that are skipped in the backup chain.
+// ValidateContinuity checks that the backups are continuous and cover the
+// requested end time.
+func validateContinuity(manifests []backuppb.BackupManifest, endTime hlc.Timestamp) error {
+	if len(manifests) == 0 {
+		return errors.AssertionFailedf("an empty chain of backups cannot cover an end time")
+	}
+	for i := range len(manifests) - 1 {
+		if !manifests[i].EndTime.Equal(manifests[i+1].StartTime) {
+			return errors.AssertionFailedf(
+				"backups are not continuous: %dth backup ends at %+v, %dth backup starts at %+v",
+				i, manifests[i].EndTime,
+				i+1, manifests[i+1].StartTime,
+			)
+		}
+	}
+	if !endTime.IsEmpty() {
+		lastManifest := manifests[len(manifests)-1]
+		if !lastManifest.StartTime.Less(endTime) || !endTime.LessEq(lastManifest.EndTime) {
+			return errors.AssertionFailedf(
+				"requested time %s is not covered by the last backup",
+				endTime,
+			)
+		}
+	}
+	return nil
+}
+
+// ElideSkippedLayers removes backups that are skipped in the backup chain and
+// ensures only backups that will be used in the restore are returned.
+//
+// Note: This assumes that the provided backups are sorted in increasing order
+// by end time, and then sorted in increasing order by start time to break ties.
 func ElideSkippedLayers(
 	uris []string, backups []backuppb.BackupManifest, loc []jobspb.RestoreDetails_BackupLocalityInfo,
 ) ([]string, []backuppb.BackupManifest, []jobspb.RestoreDetails_BackupLocalityInfo, error) {
+	uris, backups, loc = elideDuplicateEndTimes(uris, backups, loc)
 	i := len(backups) - 1
 	for i > 0 {
 		// Find j such that backups[j] is parent of backups[i].
@@ -959,10 +1008,37 @@ func ElideSkippedLayers(
 		// Move up to check the chain from j now.
 		i = j
 	}
-
 	return uris, backups, loc, nil
 }
 
+// elideDuplicateEndTimes ensures that backups in a list of backups are
+// functionally unique by removing any duplicates that have the same end time,
+// choosing backups with earlier start times and eliding the rest.
+//
+// Note: This assumes that the provided backups are sorted in increasing order
+// by end time, and then sorted in increasing order by start time to break ties.
+// This is the case for backups being returned by storage clients due to us
+// encoding backup paths in a way that ensures this order.
+func elideDuplicateEndTimes(
+	uris []string, backups []backuppb.BackupManifest, loc []jobspb.RestoreDetails_BackupLocalityInfo,
+) ([]string, []backuppb.BackupManifest, []jobspb.RestoreDetails_BackupLocalityInfo) {
+	for i := range len(backups) - 1 {
+		j := i + 1
+		// Find j such that backups[j] no longer shares the same end time as
+		// backups[i].
+		for j < len(backups) && backups[i].EndTime.Equal(backups[j].EndTime) {
+			j++
+		}
+		// If there exists backups between i and j, remove them.
+		if j > i+1 {
+			uris = slices.Delete(uris, i+1, j)
+			backups = slices.Delete(backups, i+1, j)
+			loc = slices.Delete(loc, i+1, j)
+		}
+	}
+	return uris, backups, loc
+}
+
 // GetBackupIndexAtTime returns the index of the latest backup in
 // `backupManifests` with a StartTime >= asOf.
 func GetBackupIndexAtTime(

pkg/backup/backupinfo/manifest_handling_test.go

@@ -379,18 +379,46 @@ func TestMakeBackupCodec(t *testing.T) {
 func TestElideSkippedLayers(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 
+	// Note: The tests here work under the assumption that the input lists are
+	// always sorted in ascending order by end time, and then sorted in ascending
+	// order by start time.
 	for _, tc := range []struct {
 		name     string
 		times    [][]int // len 2 slices of start and end time.
-		expected []int   // expected end times.
+		expected [][]int // expected start and end times
 	}{
-		{"single", [][]int{{0, 1}}, []int{1}},
-		{"double", [][]int{{0, 1}, {1, 2}}, []int{1, 2}},
-		{"simple chain", [][]int{{0, 1}, {1, 2}, {2, 3}, {3, 5}, {5, 8}}, []int{1, 2, 3, 5, 8}},
-		{"skip one", [][]int{{0, 1}, {1, 2}, {1, 3}, {3, 5}, {5, 8}}, []int{1, 3, 5, 8}},
-		{"skip all", [][]int{{0, 1}, {1, 2}, {1, 3}, {3, 5}, {1, 8}}, []int{1, 8}},
-		{"skip twice to first", [][]int{{0, 1}, {1, 2}, {1, 3}, {3, 5}, {3, 8}}, []int{1, 3, 8}},
-		{"skip twice to second", [][]int{{0, 1}, {1, 2}, {1, 3}, {3, 5}, {2, 8}}, []int{1, 2, 8}},
+		{"single", [][]int{{0, 1}}, [][]int{{0, 1}}},
+		{"double", [][]int{{0, 1}, {1, 2}}, [][]int{{0, 1}, {1, 2}}},
+		{
+			"simple chain, no skips",
+			[][]int{{0, 1}, {1, 2}, {2, 3}, {3, 5}, {5, 8}},
+			[][]int{{0, 1}, {1, 2}, {2, 3}, {3, 5}, {5, 8}},
+		},
+		{
+			"compaction of two backups",
+			[][]int{{0, 1}, {1, 2}, {1, 3}, {2, 3}, {3, 5}, {5, 8}},
+			[][]int{{0, 1}, {1, 3}, {3, 5}, {5, 8}},
+		},
+		{
+			"compaction of entire chain",
+			[][]int{{0, 1}, {1, 2}, {2, 3}, {3, 5}, {0, 8}, {5, 8}},
+			[][]int{{0, 8}},
+		},
+		{
+			"two compactions of two backups",
+			[][]int{{0, 1}, {1, 2}, {1, 3}, {2, 3}, {3, 5}, {3, 8}, {5, 8}},
+			[][]int{{0, 1}, {1, 3}, {3, 8}},
+		},
+		{
+			"compaction includes a compacted backup in the middle",
+			[][]int{{0, 1}, {1, 2}, {1, 3}, {2, 3}, {3, 5}, {1, 8}, {5, 8}},
+			[][]int{{0, 1}, {1, 8}},
+		},
+		{
+			"two compactions with the same end time",
+			[][]int{{0, 1}, {1, 2}, {2, 3}, {3, 5}, {1, 8}, {3, 8}, {5, 8}},
+			[][]int{{0, 1}, {1, 8}},
+		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			chain := make([]backuppb.BackupManifest, len(tc.times))
@@ -408,7 +436,8 @@ func TestElideSkippedLayers(t *testing.T) {
 			require.Equal(t, len(tc.expected), len(locs))
 			require.Equal(t, len(tc.expected), len(res))
 			for i := range tc.expected {
-				require.Equal(t, tc.expected[i], int(res[i].EndTime.WallTime), "expected %q\ngot: %q")
+				actual := []int{int(res[i].StartTime.WallTime), int(res[i].EndTime.WallTime)}
+				require.Equal(t, tc.expected[i], actual)
 			}
 		})
 	}