Repository: kopia/kopia
Pull request #3728 (2024-03-13 04:00:09)

refactor(general): add epoch.Manager.MaybeCompactSingleEpoch
Subset and dependency of #3651
Depends on #3735

Ref:
- #3638
- #3639


codecov[bot]: ## [Codecov](https://app.codecov.io/gh/kopia/kopia/pull/3728?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) Report
Attention: Patch coverage is `0%` with `31 lines` in your changes are missing coverage. Please review.
> Project coverage is 76.97%. Comparing base [(`cb455c6`)](https://app.codecov.io/gh/kopia/kopia/commit/cb455c699a04e54cb47e25a676d7d65063ce883e?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) to head [(`3865bae`)](https://app.codecov.io/gh/kopia/kopia/pull/3728?dropdown=coverage&src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia).
> Report is 56 commits behind head on master.

| [Files](https://app.codecov.io/gh/kopia/kopia/pull/3728?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) | Patch % | Lines |
|---|---|---|
| [internal/epoch/epoch\_manager.go](https://app.codecov.io/gh/kopia/kopia/pull/3728?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia#diff-aW50ZXJuYWwvZXBvY2gvZXBvY2hfbWFuYWdlci5nbw==) | 0.00% | [31 Missing :warning: ](https://app.codecov.io/gh/kopia/kopia/pull/3728?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #3728      +/-   ##
==========================================
+ Coverage   75.86%   76.97%   +1.10%     
==========================================
  Files         470      476       +6     
  Lines       37301    28815    -8486     
==========================================
- Hits        28299    22180    -6119     
+ Misses       7071     4709    -2362     
+ Partials     1931     1926       -5     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/kopia/kopia/pull/3728?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia).


Patch:
diff --git a/internal/epoch/epoch_manager.go b/internal/epoch/epoch_manager.go
index 6f8c085d..52f19508 100644
--- a/internal/epoch/epoch_manager.go
+++ b/internal/epoch/epoch_manager.go
@@ -970,6 +970,45 @@ func (e *Manager) getCompleteIndexSetForCommittedState(ctx context.Context, cs C
 	return result, nil
 }
 
+// MaybeCompactSingleEpoch compacts the oldest epoch that is eligible for
+// compaction if there is one.
+func (e *Manager) MaybeCompactSingleEpoch(ctx context.Context) error {
+	cs, err := e.committedState(ctx, 0)
+	if err != nil {
+		return err
+	}
+
+	uncompacted, err := oldestUncompactedEpoch(cs)
+	if err != nil {
+		return err
+	}
+
+	if !cs.isSettledEpochNumber(uncompacted) {
+		e.log.Debugw("there are no uncompacted epochs eligible for compaction", "oldestUncompactedEpoch", uncompacted)
+
+		return nil
+	}
+
+	uncompactedBlobs, ok := cs.UncompactedEpochSets[uncompacted]
+	if !ok {
+		// blobs for this epoch were not loaded in the current snapshot, get the list of blobs for this epoch
+		ue, err := blob.ListAllBlobs(ctx, e.st, UncompactedEpochBlobPrefix(uncompacted))
+		if err != nil {
+			return errors.Wrapf(err, "error listing uncompacted indexes for epoch %v", uncompacted)
+		}
+
+		uncompactedBlobs = ue
+	}
+
+	e.log.Debugf("starting single-epoch compaction of %v")
+
+	if err := e.compact(ctx, blob.IDsFromMetadata(uncompactedBlobs), compactedEpochBlobPrefix(uncompacted)); err != nil {
+		return errors.Wrapf(err, "unable to compact blobs for epoch %v: performance will be affected", uncompacted)
+	}
+
+	return nil
+}
+
 func (e *Manager) getIndexesFromEpochInternal(ctx context.Context, cs CurrentSnapshot, epoch int) ([]blob.Metadata, error) {
 	// check if the epoch is old enough to possibly have compacted blobs
 	epochSettled := cs.isSettledEpochNumber(epoch)
diff --git a/internal/epoch/epoch_utils.go b/internal/epoch/epoch_utils.go
index 0e184934..4334f9f0 100644
--- a/internal/epoch/epoch_utils.go
+++ b/internal/epoch/epoch_utils.go
@@ -160,3 +160,53 @@ func getContiguousKeyRange[E any](m map[int]E) (closedIntRange, error) {
 
 	return r, nil
 }
+
+func getCompactedEpochRange(cs CurrentSnapshot) (closedIntRange, error) {
+	return getContiguousKeyRange(cs.SingleEpochCompactionSets)
+}
+
+var errInvalidCompactedRange = errors.New("invalid compacted epoch range")
+
+func getRangeCompactedRange(cs CurrentSnapshot) closedIntRange {
+	rangeSetsLen := len(cs.LongestRangeCheckpointSets)
+
+	if rangeSetsLen == 0 {
+		return closedIntRange{lo: 0, hi: -1}
+	}
+
+	return closedIntRange{
+		lo: cs.LongestRangeCheckpointSets[0].MinEpoch,
+		hi: cs.LongestRangeCheckpointSets[rangeSetsLen-1].MaxEpoch,
+	}
+}
+
+func oldestUncompactedEpoch(cs CurrentSnapshot) (int, error) {
+	rangeCompacted := getRangeCompactedRange(cs)
+
+	var oldestUncompacted int
+
+	if !rangeCompacted.isEmpty() {
+		if rangeCompacted.lo != 0 {
+			// range compactions are expected to cover the 0 epoch
+			return -1, errors.Wrapf(errInvalidCompactedRange, "Epoch 0 not included in range compaction, lowest epoch in range compactions: %v", rangeCompacted.lo)
+		}
+
+		oldestUncompacted = rangeCompacted.hi + 1
+	}
+
+	singleCompacted, err := getCompactedEpochRange(cs)
+	if err != nil {
+		return -1, errors.Wrap(err, "could not get latest single-compacted epoch")
+	}
+
+	if singleCompacted.isEmpty() || oldestUncompacted < singleCompacted.lo {
+		return oldestUncompacted, nil
+	}
+
+	// singleCompacted is not empty
+	if oldestUncompacted > singleCompacted.hi {
+		return oldestUncompacted, nil
+	}
+
+	return singleCompacted.hi + 1, nil
+}


Test patch:
diff --git a/internal/epoch/epoch_manager_test.go b/internal/epoch/epoch_manager_test.go
index 343e7d48..b7c615cc 100644
--- a/internal/epoch/epoch_manager_test.go
+++ b/internal/epoch/epoch_manager_test.go
@@ -800,6 +800,147 @@ func TestIndexEpochManager_RefreshContextCanceled(t *testing.T) {
 	require.ErrorIs(t, err, ctx.Err())
 }
 
+func TestMaybeCompactSingleEpoch_Empty(t *testing.T) {
+	t.Parallel()
+
+	te := newTestEnv(t)
+	ctx := testlogging.Context(t)
+
+	// this should be a no-op
+	err := te.mgr.MaybeCompactSingleEpoch(ctx)
+
+	require.NoError(t, err)
+}
+
+func TestMaybeCompactSingleEpoch_GetParametersError(t *testing.T) {
+	t.Parallel()
+
+	te := newTestEnv(t)
+	ctx := testlogging.Context(t)
+
+	paramsError := errors.New("no parameters error")
+	te.mgr.paramProvider = faultyParamsProvider{err: paramsError}
+
+	err := te.mgr.MaybeCompactSingleEpoch(ctx)
+
+	require.Error(t, err)
+	require.ErrorIs(t, err, paramsError)
+}
+
+func TestMaybeCompactSingleEpoch_CompactionError(t *testing.T) {
+	t.Parallel()
+
+	te := newTestEnv(t)
+	ctx := testlogging.Context(t)
+
+	p, err := te.mgr.getParameters(ctx)
+	require.NoError(t, err)
+
+	idxCount := p.GetEpochAdvanceOnCountThreshold()
+	// Create sufficient indexes blobs and move clock forward to advance epoch.
+	for j := 0; j < 3; j++ {
+		for i := 0; i < idxCount; i++ {
+			if i == idxCount-1 {
+				// Advance the time so that the difference in times for writes will force
+				// new epochs.
+				te.ft.Advance(p.MinEpochDuration + 1*time.Hour)
+			}
+
+			te.mustWriteIndexFiles(ctx, t, newFakeIndexWithEntries(i))
+		}
+	}
+
+	compactionError := errors.New("test compaction error")
+	te.mgr.compact = func(context.Context, []blob.ID, blob.ID) error {
+		return compactionError
+	}
+
+	err = te.mgr.MaybeCompactSingleEpoch(ctx)
+
+	require.Error(t, err)
+	require.ErrorIs(t, err, compactionError)
+}
+
+func TestMaybeCompactSingleEpoch(t *testing.T) {
+	const epochsToWrite = 5
+
+	t.Parallel()
+
+	te := newTestEnv(t)
+	ctx := testlogging.Context(t)
+	te.mgr.allowCleanupWritesOnIndexLoad = false
+
+	p, err := te.mgr.getParameters(ctx)
+	require.NoError(t, err)
+
+	idxCount := p.GetEpochAdvanceOnCountThreshold()
+
+	var k int
+
+	// Create sufficient indexes blobs and move clock forward to advance current epoch
+	for j := 0; j < epochsToWrite; j++ {
+		for i := 0; i < idxCount; i++ {
+			if i == idxCount-1 {
+				// Advance the time so that the difference in times for writes will force
+				// new epochs.
+				te.ft.Advance(p.MinEpochDuration + 1*time.Hour)
+			}
+
+			te.mustWriteIndexFiles(ctx, t, newFakeIndexWithEntries(k))
+			k++
+		}
+
+		te.verifyCurrentWriteEpoch(t, j)
+
+		err = te.mgr.MaybeAdvanceWriteEpoch(ctx)
+		require.NoError(t, err)
+
+		err = te.mgr.Refresh(ctx) // force state refresh
+
+		require.NoError(t, err)
+		te.verifyCurrentWriteEpoch(t, j+1)
+	}
+
+	cs, err := te.mgr.Current(ctx)
+
+	require.NoError(t, err)
+	require.Equal(t, epochsToWrite, cs.WriteEpoch)
+
+	// no epochs have been compacted, so the compacted set should be empty and
+	// the uncompacted epoch set should have all the epochs
+	require.Empty(t, cs.LongestRangeCheckpointSets)
+	require.Empty(t, cs.SingleEpochCompactionSets)
+
+	// perform single-epoch compaction for settled epochs
+	newestEpochToCompact := cs.WriteEpoch - numUnsettledEpochs + 1
+	for j := 0; j < newestEpochToCompact; j++ {
+		err = te.mgr.MaybeCompactSingleEpoch(ctx)
+		require.NoError(t, err)
+
+		err = te.mgr.Refresh(ctx) // force state refresh
+		require.NoError(t, err)
+
+		cs, err = te.mgr.Current(ctx)
+		require.NoError(t, err)
+
+		require.Len(t, cs.SingleEpochCompactionSets, j+1)
+	}
+
+	require.Len(t, cs.SingleEpochCompactionSets, newestEpochToCompact)
+
+	// no more epochs should be compacted at this point
+	err = te.mgr.MaybeCompactSingleEpoch(ctx)
+	require.NoError(t, err)
+
+	err = te.mgr.Refresh(ctx)
+	require.NoError(t, err)
+
+	cs, err = te.mgr.Current(ctx)
+	require.NoError(t, err)
+
+	require.Len(t, cs.SingleEpochCompactionSets, newestEpochToCompact)
+}
+
 func TestValidateParameters(t *testing.T) {
 	cases := []struct {
 		p       Parameters
diff --git a/internal/epoch/epoch_utils_test.go b/internal/epoch/epoch_utils_test.go
index 02579abe..5dcf9171 100644
--- a/internal/epoch/epoch_utils_test.go
+++ b/internal/epoch/epoch_utils_test.go
@@ -312,3 +312,225 @@ func TestGetKeyRange(t *testing.T) {
 		})
 	}
 }
+
+func TestOldestUncompactedEpoch(t *testing.T) {
+	cases := []struct {
+		input         CurrentSnapshot
+		expectedEpoch int
+		wantErr       error
+	}{
+		{
+			input: CurrentSnapshot{
+				SingleEpochCompactionSets: map[int][]blob.Metadata{},
+			},
+		},
+		{
+			input: CurrentSnapshot{
+				WriteEpoch: 0,
+				SingleEpochCompactionSets: map[int][]blob.Metadata{
+					0: {blob.Metadata{BlobID: compactedEpochBlobPrefix(0) + "foo0"}},
+				},
+			},
+			expectedEpoch: 1,
+		},
+		{
+			input: CurrentSnapshot{
+				SingleEpochCompactionSets: map[int][]blob.Metadata{
+					0: {blob.Metadata{BlobID: compactedEpochBlobPrefix(0) + "foo0"}},
+					1: {blob.Metadata{BlobID: compactedEpochBlobPrefix(1) + "foo1"}},
+				},
+			},
+			expectedEpoch: 2,
+		},
+		{
+			input: CurrentSnapshot{
+				SingleEpochCompactionSets: map[int][]blob.Metadata{
+					1: {blob.Metadata{BlobID: compactedEpochBlobPrefix(1) + "foo1"}},
+				},
+			},
+			expectedEpoch: 0,
+		},
+		{
+			input: CurrentSnapshot{
+				SingleEpochCompactionSets: map[int][]blob.Metadata{
+					0: {blob.Metadata{BlobID: compactedEpochBlobPrefix(0) + "foo0"}},
+					2: {blob.Metadata{BlobID: compactedEpochBlobPrefix(2) + "foo2"}},
+				},
+			},
+			expectedEpoch: -1,
+			wantErr:       errNonContiguousRange,
+		},
+		{
+			input: CurrentSnapshot{
+				LongestRangeCheckpointSets: []*RangeMetadata{
+					{
+						MinEpoch: 0,
+						MaxEpoch: 2,
+						Blobs: []blob.Metadata{
+							{BlobID: rangeCheckpointBlobPrefix(0, 2) + "foo-0-2"},
+						},
+					},
+				},
+			},
+			expectedEpoch: 3,
+		},
+		{
+			input: CurrentSnapshot{
+				LongestRangeCheckpointSets: []*RangeMetadata{
+					{
+						MinEpoch: 0,
+						MaxEpoch: 2,
+						Blobs: []blob.Metadata{
+							{BlobID: rangeCheckpointBlobPrefix(0, 2) + "foo-0-2"},
+						},
+					},
+				},
+				SingleEpochCompactionSets: map[int][]blob.Metadata{
+					0: {blob.Metadata{BlobID: compactedEpochBlobPrefix(0) + "foo0"}},
+					1: {blob.Metadata{BlobID: compactedEpochBlobPrefix(1) + "foo1"}},
+				},
+			},
+			expectedEpoch: 3,
+		},
+		{
+			input: CurrentSnapshot{
+				LongestRangeCheckpointSets: []*RangeMetadata{
+					{
+						MinEpoch: 0,
+						MaxEpoch: 2,
+						Blobs: []blob.Metadata{
+							{BlobID: rangeCheckpointBlobPrefix(0, 2) + "foo-0-2"},
+						},
+					},
+				},
+				SingleEpochCompactionSets: map[int][]blob.Metadata{
+					1: {blob.Metadata{BlobID: compactedEpochBlobPrefix(1) + "foo1"}},
+					2: {blob.Metadata{BlobID: compactedEpochBlobPrefix(2) + "foo2"}},
+				},
+			},
+			expectedEpoch: 3,
+		},
+		{
+			input: CurrentSnapshot{
+				LongestRangeCheckpointSets: []*RangeMetadata{
+					{
+						MinEpoch: 0,
+						MaxEpoch: 2,
+						Blobs: []blob.Metadata{
+							{BlobID: rangeCheckpointBlobPrefix(0, 2) + "foo-0-2"},
+						},
+					},
+				},
+				SingleEpochCompactionSets: map[int][]blob.Metadata{
+					1: {blob.Metadata{BlobID: compactedEpochBlobPrefix(1) + "foo1"}},
+				},
+			},
+			expectedEpoch: 3,
+		},
+
+		{
+			input: CurrentSnapshot{
+				LongestRangeCheckpointSets: []*RangeMetadata{
+					{
+						MinEpoch: 0,
+						MaxEpoch: 2,
+						Blobs: []blob.Metadata{
+							{BlobID: rangeCheckpointBlobPrefix(0, 2) + "foo-0-2"},
+						},
+					},
+				},
+				SingleEpochCompactionSets: map[int][]blob.Metadata{
+					4: {blob.Metadata{BlobID: compactedEpochBlobPrefix(4) + "foo4"}},
+					5: {blob.Metadata{BlobID: compactedEpochBlobPrefix(5) + "foo5"}},
+				},
+			},
+			expectedEpoch: 3,
+		},
+		{
+			input: CurrentSnapshot{
+				LongestRangeCheckpointSets: []*RangeMetadata{
+					{
+						MinEpoch: 0,
+						MaxEpoch: 2,
+						Blobs: []blob.Metadata{
+							{BlobID: rangeCheckpointBlobPrefix(0, 2) + "foo-0-2"},
+						},
+					},
+				},
+				SingleEpochCompactionSets: map[int][]blob.Metadata{
+					2: {blob.Metadata{BlobID: compactedEpochBlobPrefix(2) + "foo2"}},
+					3: {blob.Metadata{BlobID: compactedEpochBlobPrefix(3) + "foo3"}},
+				},
+			},
+			expectedEpoch: 4,
+		},
+		{
+			input: CurrentSnapshot{
+				LongestRangeCheckpointSets: []*RangeMetadata{
+					{
+						MinEpoch: 0,
+						MaxEpoch: 2,
+						Blobs: []blob.Metadata{
+							{BlobID: rangeCheckpointBlobPrefix(0, 2) + "foo-0-2"},
+						},
+					},
+				},
+				SingleEpochCompactionSets: map[int][]blob.Metadata{
+					3: {blob.Metadata{BlobID: compactedEpochBlobPrefix(3) + "foo3"}},
+					4: {blob.Metadata{BlobID: compactedEpochBlobPrefix(4) + "foo4"}},
+				},
+			},
+			expectedEpoch: 5,
+		},
+		{
+			input: CurrentSnapshot{
+				LongestRangeCheckpointSets: []*RangeMetadata{
+					{
+						MinEpoch: 1,
+						MaxEpoch: 2,
+						Blobs: []blob.Metadata{
+							{BlobID: rangeCheckpointBlobPrefix(1, 2) + "foo-1-2"},
+						},
+					},
+				},
+				SingleEpochCompactionSets: map[int][]blob.Metadata{
+					3: {blob.Metadata{BlobID: compactedEpochBlobPrefix(3) + "foo3"}},
+					4: {blob.Metadata{BlobID: compactedEpochBlobPrefix(4) + "foo4"}},
+				},
+			},
+			expectedEpoch: -1,
+			wantErr:       errInvalidCompactedRange,
+		},
+		{
+			input: CurrentSnapshot{
+				LongestRangeCheckpointSets: []*RangeMetadata{
+					{
+						MinEpoch: 0,
+						MaxEpoch: 2,
+						Blobs: []blob.Metadata{
+							{BlobID: rangeCheckpointBlobPrefix(0, 2) + "foo-0-2"},
+						},
+					},
+				},
+				SingleEpochCompactionSets: map[int][]blob.Metadata{
+					3: {blob.Metadata{BlobID: compactedEpochBlobPrefix(3) + "foo3"}},
+					5: {blob.Metadata{BlobID: compactedEpochBlobPrefix(5) + "foo5"}},
+				},
+			},
+			expectedEpoch: -1,
+			wantErr:       errNonContiguousRange,
+		},
+	}
+
+	for i, tc := range cases {
+		t.Run(fmt.Sprint("case: ", i), func(t *testing.T) {
+			got, err := oldestUncompactedEpoch(tc.input)
+
+			if tc.wantErr != nil {
+				require.Error(t, err)
+			}
+
+			require.Equal(t, tc.expectedEpoch, got, "input: %#v", tc.input)
+		})
+	}
+}