Repository: kopia/kopia
Pull request #3726 (2024-03-13 01:22:50)

refactor(general): add epoch.Manager.CleanupMarkers


Ref:
- Extracted from #3651
- #3638


codecov[bot]: ## [Codecov](https://app.codecov.io/gh/kopia/kopia/pull/3726?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 `8 lines` in your changes are missing coverage. Please review.
> Project coverage is 77.08%. 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 [(`2290a51`)](https://app.codecov.io/gh/kopia/kopia/pull/3726?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/3726?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/3726?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia#diff-aW50ZXJuYWwvZXBvY2gvZXBvY2hfbWFuYWdlci5nbw==) | 0.00% | [8 Missing :warning: ](https://app.codecov.io/gh/kopia/kopia/pull/3726?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    #3726      +/-   ##
==========================================
+ Coverage   75.86%   77.08%   +1.21%     
==========================================
  Files         470      476       +6     
  Lines       37301    28792    -8509     
==========================================
- Hits        28299    22194    -6105     
+ Misses       7071     4678    -2393     
+ Partials     1931     1920      -11     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/kopia/kopia/pull/3726?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 52f19508..98810601 100644
--- a/internal/epoch/epoch_manager.go
+++ b/internal/epoch/epoch_manager.go
@@ -288,6 +288,21 @@ func (e *Manager) maxCleanupTime(cs CurrentSnapshot) time.Time {
 	return maxTime
 }
 
+// CleanupMarkers removes superseded watermarks and epoch markers.
+func (e *Manager) CleanupMarkers(ctx context.Context) error {
+	cs, err := e.committedState(ctx, 0)
+	if err != nil {
+		return err
+	}
+
+	p, err := e.getParameters(ctx)
+	if err != nil {
+		return err
+	}
+
+	return e.cleanupInternal(ctx, cs, p)
+}
+
 func (e *Manager) cleanupInternal(ctx context.Context, cs CurrentSnapshot, p *Parameters) error {
 	eg, ctx := errgroup.WithContext(ctx)
 


Test patch:
diff --git a/internal/epoch/epoch_manager_test.go b/internal/epoch/epoch_manager_test.go
index b7c615cc..f669c042 100644
--- a/internal/epoch/epoch_manager_test.go
+++ b/internal/epoch/epoch_manager_test.go
@@ -1015,6 +1015,117 @@ func TestValidateParameters(t *testing.T) {
 	}
 }
 
+func TestCleanupMarkers_Empty(t *testing.T) {
+	t.Parallel()
+
+	te := newTestEnv(t)
+	te.mgr.allowCleanupWritesOnIndexLoad = false
+	ctx := testlogging.Context(t)
+
+	// this should be a no-op
+	err := te.mgr.CleanupMarkers(ctx)
+
+	require.NoError(t, err)
+}
+
+func TestCleanupMarkers_GetParametersError(t *testing.T) {
+	t.Parallel()
+
+	te := newTestEnv(t)
+	ctx := testlogging.Context(t)
+	te.mgr.allowCleanupWritesOnIndexLoad = false
+
+	paramsError := errors.New("no parameters error")
+	te.mgr.paramProvider = faultyParamsProvider{err: paramsError}
+
+	err := te.mgr.CleanupMarkers(ctx)
+
+	require.Error(t, err)
+	require.ErrorIs(t, err, paramsError)
+}
+
+func TestCleanupMarkers_FailToReadState(t *testing.T) {
+	t.Parallel()
+
+	te := newTestEnv(t)
+	te.mgr.allowCleanupWritesOnIndexLoad = false
+	ctx, cancel := context.WithCancel(testlogging.Context(t))
+
+	te.ft.Advance(1 * time.Hour) // force state refresh in CleanupMarkers
+
+	cancel()
+	err := te.mgr.CleanupMarkers(ctx)
+
+	require.Error(t, err)
+}
+
+func TestCleanupMarkers_AvoidCleaningUpSingleEpochMarker(t *testing.T) {
+	t.Parallel()
+
+	te := newTestEnv(t)
+	te.mgr.allowCleanupWritesOnIndexLoad = false
+	ctx := testlogging.Context(t)
+
+	te.mgr.forceAdvanceEpoch(ctx)
+	te.ft.Advance(1 * time.Hour)
+
+	require.NoError(t, te.mgr.Refresh(ctx))
+
+	cs, err := te.mgr.Current(ctx)
+	require.NoError(t, err)
+	require.Equal(t, 1, cs.WriteEpoch)
+
+	err = te.mgr.CleanupMarkers(ctx)
+	require.NoError(t, err)
+
+	require.NoError(t, te.mgr.Refresh(ctx))
+
+	// is the epoch marker preserved?
+	te.verifyCurrentWriteEpoch(t, 1)
+
+	cs, err = te.mgr.Current(ctx)
+	require.NoError(t, err)
+	require.Len(t, cs.EpochMarkerBlobs, 1)
+}
+
+func TestCleanupMarkers_CleanUpManyMarkers(t *testing.T) {
+	t.Parallel()
+
+	te := newTestEnv(t)
+	te.mgr.allowCleanupWritesOnIndexLoad = false
+	ctx := testlogging.Context(t)
+
+	p, err := te.mgr.getParameters(ctx)
+	require.NoError(t, err)
+
+	const epochsToAdvance = 5
+
+	te.mustWriteIndexFiles(ctx, t, newFakeIndexWithEntries(0))
+	for i := 0; i < epochsToAdvance; i++ {
+		te.ft.Advance(p.MinEpochDuration + 1*time.Hour)
+		te.mgr.forceAdvanceEpoch(ctx)
+		te.mustWriteIndexFiles(ctx, t, newFakeIndexWithEntries(i+1))
+	}
+
+	require.NoError(t, te.mgr.Refresh(ctx))
+	te.verifyCurrentWriteEpoch(t, epochsToAdvance)
+
+	cs, err := te.mgr.Current(ctx)
+	require.NoError(t, err)
+	require.Len(t, cs.EpochMarkerBlobs, epochsToAdvance)
+
+	err = te.mgr.CleanupMarkers(ctx)
+	require.NoError(t, err)
+
+	// is the epoch marker preserved?
+	require.NoError(t, te.mgr.Refresh(ctx))
+	te.verifyCurrentWriteEpoch(t, epochsToAdvance)
+
+	cs, err = te.mgr.Current(ctx)
+	require.NoError(t, err)
+	require.Len(t, cs.EpochMarkerBlobs, 2) // at least 2 epoch markers are kept
+}
+
 func randomTime(min, max time.Duration) time.Duration {
 	return time.Duration(float64(max-min)*rand.Float64() + float64(min))
 }