Repository: kopia/kopia
Pull request #3709 (2024-03-06 23:20:44)

refactor(general): add epoch.Manager.MaybeAdvanceEpoch helper
Adds `epoch.Manager.MaybeAdvanceEpoch helper` and tests.
The helper is not yet invoked in maintenance, a separate change
will add all the epoch management cleanup tasks to maintenance.

Ref:
- #3638
- #3645
- #3651

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/kopia/kopia/pull/3709?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 `80.00000%` with `2 lines` in your changes are missing coverage. Please review.
> Project coverage is 77.03%. Comparing base [(`cb455c6`)](https://app.codecov.io/gh/kopia/kopia/commit/cb455c699a04e54cb47e25a676d7d65063ce883e?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) to head [(`2844f78`)](https://app.codecov.io/gh/kopia/kopia/pull/3709?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia).
> Report is 45 commits behind head on master.

| [Files](https://app.codecov.io/gh/kopia/kopia/pull/3709?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/3709?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia#diff-aW50ZXJuYWwvZXBvY2gvZXBvY2hfbWFuYWdlci5nbw==) | 80.00% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/kopia/kopia/pull/3709?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    #3709      +/-   ##
==========================================
+ Coverage   75.86%   77.03%   +1.17%     
==========================================
  Files         470      476       +6     
  Lines       37301    28753    -8548     
==========================================
- Hits        28299    22151    -6148     
+ Misses       7071     4675    -2396     
+ Partials     1931     1927       -4     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/kopia/kopia/pull/3709?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 4fc61752..1546a71f 100644
--- a/internal/epoch/epoch_manager.go
+++ b/internal/epoch/epoch_manager.go
@@ -723,6 +723,25 @@ func (e *Manager) refreshAttemptLocked(ctx context.Context) error {
 	return nil
 }
 
+// MaybeAdvanceWriteEpoch writes a new write epoch marker when a new write
+// epoch should be started, otherwise it does not do anything.
+func (e *Manager) MaybeAdvanceWriteEpoch(ctx context.Context) error {
+	p, err := e.getParameters(ctx)
+	if err != nil {
+		return err
+	}
+
+	e.mu.Lock()
+	cs := e.lastKnownState
+	e.mu.Unlock()
+
+	if shouldAdvance(cs.UncompactedEpochSets[cs.WriteEpoch], p.MinEpochDuration, p.EpochAdvanceOnCountThreshold, p.EpochAdvanceOnTotalSizeBytesThreshold) {
+		return errors.Wrap(e.advanceEpochMarker(ctx, cs), "error advancing epoch")
+	}
+
+	return nil
+}
+
 func (e *Manager) advanceEpochMarker(ctx context.Context, cs CurrentSnapshot) error {
 	blobID := blob.ID(fmt.Sprintf("%v%v", string(EpochMarkerIndexBlobPrefix), cs.WriteEpoch+1))
 


Test patch:
diff --git a/internal/epoch/epoch_manager_test.go b/internal/epoch/epoch_manager_test.go
index 30ad25d9..2c15c851 100644
--- a/internal/epoch/epoch_manager_test.go
+++ b/internal/epoch/epoch_manager_test.go
@@ -529,6 +529,141 @@ func TestSlowWrite_MovesToNextEpochTwice(t *testing.T) {
 	require.Contains(t, err.Error(), "slow index write")
 }
 
+func TestMaybeAdvanceEpoch_Empty(t *testing.T) {
+	t.Parallel()
+
+	te := newTestEnv(t)
+	ctx := testlogging.Context(t)
+
+	te.verifyCurrentWriteEpoch(t, 0)
+
+	// this should be a no-op
+	err := te.mgr.MaybeAdvanceWriteEpoch(ctx)
+
+	require.NoError(t, err)
+
+	// check current epoch again
+	te.verifyCurrentWriteEpoch(t, 0)
+}
+
+func TestMaybeAdvanceEpoch(t *testing.T) {
+	t.Parallel()
+
+	ctx := testlogging.Context(t)
+	te := newTestEnv(t)
+
+	// Disable automatic epoch advancement and compaction to build up state
+	te.mgr.allowCleanupWritesOnIndexLoad = false
+	te.mgr.compact = func(context.Context, []blob.ID, blob.ID) error {
+		return nil
+	}
+
+	te.verifyCurrentWriteEpoch(t, 0)
+
+	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 i := 0; i < idxCount; i++ {
+		te.mustWriteIndexFiles(ctx, t, newFakeIndexWithEntries(i))
+	}
+
+	te.verifyCurrentWriteEpoch(t, 0)
+
+	// Advance the time so that the difference in times for writes will force
+	// new epochs.
+	te.ft.Advance(p.MinEpochDuration + 1*time.Hour)
+
+	err = te.mgr.Refresh(ctx)
+	require.NoError(t, err)
+
+	te.verifyCurrentWriteEpoch(t, 0)
+
+	// one more to go over the threshold
+	te.mustWriteIndexFiles(ctx, t, newFakeIndexWithEntries(idxCount))
+	err = te.mgr.Refresh(ctx)
+
+	require.NoError(t, err)
+	te.verifyCurrentWriteEpoch(t, 0)
+
+	err = te.mgr.MaybeAdvanceWriteEpoch(ctx)
+
+	require.NoError(t, err)
+
+	err = te.mgr.Refresh(ctx) // force state refresh
+
+	require.NoError(t, err)
+	te.verifyCurrentWriteEpoch(t, 1)
+}
+
+type faultyParamsProvider struct {
+	err error
+}
+
+func (p faultyParamsProvider) GetParameters(ctx context.Context) (*Parameters, error) {
+	return nil, p.err
+}
+
+func TestMaybeAdvanceEpoch_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.MaybeAdvanceWriteEpoch(ctx)
+
+	require.Error(t, err)
+	require.ErrorIs(t, err, paramsError)
+}
+
+func TestMaybeAdvanceEpoch_Error(t *testing.T) {
+	t.Parallel()
+
+	ctx := testlogging.Context(t)
+	te := newTestEnv(t)
+
+	// Disable automatic epoch advancement and compaction to build up state
+	te.mgr.allowCleanupWritesOnIndexLoad = false
+	te.mgr.compact = func(context.Context, []blob.ID, blob.ID) error {
+		return nil
+	}
+
+	te.verifyCurrentWriteEpoch(t, 0)
+
+	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 i := 0; i < idxCount; i++ {
+		te.mustWriteIndexFiles(ctx, t, newFakeIndexWithEntries(i))
+	}
+
+	// Advance the time so that the difference in times for writes will force
+	// new epochs.
+	te.ft.Advance(p.MinEpochDuration + 1*time.Hour)
+
+	// one more to go over the threshold
+	te.mustWriteIndexFiles(ctx, t, newFakeIndexWithEntries(idxCount))
+	err = te.mgr.Refresh(ctx)
+
+	require.NoError(t, err)
+	te.verifyCurrentWriteEpoch(t, 0)
+
+	berr := errors.New("advance epoch put blob error")
+	te.faultyStorage.AddFaults(blobtesting.MethodPutBlob,
+		fault.New().ErrorInstead(berr))
+
+	err = te.mgr.MaybeAdvanceWriteEpoch(ctx)
+
+	require.Error(t, err)
+	require.ErrorIs(t, err, berr)
+}
+
 func TestForceAdvanceEpoch(t *testing.T) {
 	te := newTestEnv(t)
 
@@ -825,3 +960,28 @@ func (e *Manager) forceAdvanceEpoch(ctx context.Context) error {
 
 	return nil
 }
+
+func (te *epochManagerTestEnv) verifyCurrentWriteEpoch(t *testing.T, expectedEpoch int) {
+	t.Helper()
+
+	// load current epoch directly from index blobs in the backend storage
+	cs := CurrentSnapshot{
+		WriteEpoch:                0,
+		EpochStartTime:            map[int]time.Time{},
+		UncompactedEpochSets:      map[int][]blob.Metadata{},
+		SingleEpochCompactionSets: map[int][]blob.Metadata{},
+	}
+
+	ctx := testlogging.Context(t)
+	err := te.mgr.loadWriteEpoch(ctx, &cs)
+
+	require.NoError(t, err)
+	require.Equal(t, expectedEpoch, cs.WriteEpoch)
+
+	// check current epoch via the epoch manager, this may or may not cause
+	// a refresh from storage.
+	cs, err = te.mgr.Current(ctx)
+
+	require.NoError(t, err)
+	require.Equal(t, expectedEpoch, cs.WriteEpoch)
+}