epoch: changed how very slow index writes are handled (#1587)

Previously epoch manager would discard (on read) any index blobs
associated with epoch N written after epoch N+2 has already started.

There was no signal to the writer that the write is actually bad. This
change reports slow writes as a failure and disables read-time filtering.

This is important for repositories copied using sync-to because it
may not necessarily preserve the modification time leading to problems
with restore.

internal/epoch/epoch_manager.go

@@ -29,6 +29,10 @@ const (
 	maxRefreshAttemptSleepExponent = 1.5
 )
 
+// ErrVerySlowIndexWrite is returned by WriteIndex if a write takes more than 2 epochs (usually >48h).
+// This is theoretically possible with laptops going to sleep, etc.
+var ErrVerySlowIndexWrite = errors.Errorf("extremely slow index write - index write took more than two epochs")
+
 // Parameters encapsulates all parameters that influence the behavior of epoch manager.
 type Parameters struct {
 	// whether epoch manager is enabled, must be true.
@@ -172,13 +176,13 @@ func (e *Manager) Flush() {
 
 // Current retrieves current snapshot.
 func (e *Manager) Current(ctx context.Context) (CurrentSnapshot, error) {
-	return e.committedState(ctx)
+	return e.committedState(ctx, 0)
 }
 
 // AdvanceDeletionWatermark moves the deletion watermark time to a given timestamp
 // this causes all deleted content entries before given time to be treated as non-existent.
 func (e *Manager) AdvanceDeletionWatermark(ctx context.Context, ts time.Time) error {
-	cs, err := e.committedState(ctx)
+	cs, err := e.committedState(ctx, 0)
 	if err != nil {
 		return err
 	}
@@ -202,7 +206,7 @@ func (e *Manager) AdvanceDeletionWatermark(ctx context.Context, ts time.Time) er
 
 // ForceAdvanceEpoch advances current epoch unconditionally.
 func (e *Manager) ForceAdvanceEpoch(ctx context.Context) error {
-	cs, err := e.committedState(ctx)
+	cs, err := e.committedState(ctx, 0)
 	if err != nil {
 		return err
 	}
@@ -226,7 +230,7 @@ func (e *Manager) Refresh(ctx context.Context) error {
 
 // Cleanup cleans up the old indexes for which there's a compacted replacement.
 func (e *Manager) Cleanup(ctx context.Context) error {
-	cs, err := e.committedState(ctx)
+	cs, err := e.committedState(ctx, 0)
 	if err != nil {
 		return err
 	}
@@ -583,10 +587,6 @@ func (e *Manager) refreshAttemptLocked(ctx context.Context) error {
 		return errors.Wrap(err, "error loading uncompacted epochs")
 	}
 
-	for epoch := range ues {
-		ues[epoch] = blobsWrittenBefore(ues[epoch], cs.EpochStartTime[epoch+numUnsettledEpochs])
-	}
-
 	cs.UncompactedEpochSets = ues
 
 	e.log.Debugf("current epoch %v, uncompacted epoch sets %v %v %v, valid until %v",
@@ -627,11 +627,11 @@ func (e *Manager) advanceEpoch(ctx context.Context, cs CurrentSnapshot) error {
 	return nil
 }
 
-func (e *Manager) committedState(ctx context.Context) (CurrentSnapshot, error) {
+func (e *Manager) committedState(ctx context.Context, ensureMinTime time.Duration) (CurrentSnapshot, error) {
 	e.mu.Lock()
 	defer e.mu.Unlock()
 
-	if now := e.timeFunc(); now.After(e.lastKnownState.ValidUntil) {
+	if now := e.timeFunc().Add(ensureMinTime); now.After(e.lastKnownState.ValidUntil) {
 		e.log.Debugf("refreshing committed state because it's no longer valid (now %v, valid until %v)", now, e.lastKnownState.ValidUntil.Format(time.RFC3339Nano))
 
 		if err := e.refreshLocked(ctx); err != nil {
@@ -645,7 +645,7 @@ func (e *Manager) committedState(ctx context.Context) (CurrentSnapshot, error) {
 // GetCompleteIndexSet returns the set of blobs forming a complete index set up to the provided epoch number.
 func (e *Manager) GetCompleteIndexSet(ctx context.Context, maxEpoch int) ([]blob.Metadata, time.Time, error) {
 	for {
-		cs, err := e.committedState(ctx)
+		cs, err := e.committedState(ctx, 0)
 		if err != nil {
 			return nil, time.Time{}, err
 		}
@@ -673,48 +673,119 @@ func (e *Manager) GetCompleteIndexSet(ctx context.Context, maxEpoch int) ([]blob
 	}
 }
 
+var errWriteIndexTryAgain = errors.Errorf("try again")
+
 // WriteIndex writes new index blob by picking the appropriate prefix based on current epoch.
 func (e *Manager) WriteIndex(ctx context.Context, dataShards map[blob.ID]blob.Bytes) ([]blob.Metadata, error) {
+	written := map[blob.ID]blob.Metadata{}
+	writtenForEpoch := -1
+
 	for {
-		cs, err := e.committedState(ctx)
+		// make sure we have at least 75% of remaining time
+		// nolint:gomnd
+		cs, err := e.committedState(ctx, 3*e.Params.EpochRefreshFrequency/4)
 		if err != nil {
 			return nil, errors.Wrap(err, "error getting committed state")
 		}
 
-		e.log.Debugf("writing %v index shard(s) - valid until %v (%v remaining)", len(dataShards), cs.ValidUntil.Format(time.RFC3339Nano), cs.ValidUntil.Sub(e.timeFunc()))
-
-		var results []blob.Metadata
-
-		for unprefixedBlobID, data := range dataShards {
-			blobID := UncompactedEpochBlobPrefix(cs.WriteEpoch) + unprefixedBlobID
-
-			if err := e.st.PutBlob(ctx, blobID, data, blob.PutOptions{}); err != nil {
-				return nil, errors.Wrap(err, "error writing index blob")
+		if cs.WriteEpoch != writtenForEpoch {
+			if err = e.deletePartiallyWrittenShards(ctx, written); err != nil {
+				return nil, errors.Wrap(err, "unable to delete partially written shard")
 			}
 
-			bm, err := e.st.GetMetadata(ctx, blobID)
-			if err != nil {
-				return nil, errors.Wrap(err, "error getting index metadata")
-			}
-
-			e.log.Debugf("wrote-index %v", bm)
-
-			results = append(results, bm)
+			writtenForEpoch = cs.WriteEpoch
+			written = map[blob.ID]blob.Metadata{}
 		}
 
-		if now := e.timeFunc(); !now.Before(cs.ValidUntil) {
-			e.log.Debugf("write was too slow, retrying (now %v, valid until %v)", now, cs.ValidUntil.Format(time.RFC3339Nano))
-			atomic.AddInt32(e.writeIndexTooSlow, 1)
-
+		err = e.writeIndexShards(ctx, dataShards, written, cs)
+		if errors.Is(err, errWriteIndexTryAgain) {
 			continue
 		}
 
-		e.log.Debugf("index-write-success, valid until %v", cs.ValidUntil.Format(time.RFC3339Nano))
+		if err != nil {
+			e.log.Debugw("index-write-error", "error", err)
+			return nil, err
+		}
 
 		e.Invalidate()
 
-		return results, nil
+		break
 	}
+
+	cs, err := e.committedState(ctx, 0)
+	if err != nil {
+		return nil, errors.Wrap(err, "error getting committed state")
+	}
+
+	if cs.WriteEpoch >= writtenForEpoch+2 {
+		e.log.Debugw("index-write-extremely-slow")
+
+		if err = e.deletePartiallyWrittenShards(ctx, written); err != nil {
+			e.log.Debugw("index-write-extremely-slow-cleanup-failed", "error", err)
+		}
+
+		return nil, ErrVerySlowIndexWrite
+	}
+
+	var results []blob.Metadata
+
+	for _, v := range written {
+		results = append(results, v)
+	}
+
+	e.log.Debugw("index-write-success", "results", results)
+
+	return results, nil
+}
+
+func (e *Manager) deletePartiallyWrittenShards(ctx context.Context, blobs map[blob.ID]blob.Metadata) error {
+	for blobID := range blobs {
+		if err := e.st.DeleteBlob(ctx, blobID); err != nil {
+			return errors.Wrap(err, "error deleting partially written shard")
+		}
+	}
+
+	return nil
+}
+
+func (e *Manager) writeIndexShards(ctx context.Context, dataShards map[blob.ID]blob.Bytes, written map[blob.ID]blob.Metadata, cs CurrentSnapshot) error {
+	e.log.Debugw("writing index shards",
+		"shardCount", len(dataShards),
+		"validUntil", cs.ValidUntil,
+		"remaining", cs.ValidUntil.Sub(e.timeFunc()))
+
+	for unprefixedBlobID, data := range dataShards {
+		blobID := UncompactedEpochBlobPrefix(cs.WriteEpoch) + unprefixedBlobID
+		if _, ok := written[blobID]; ok {
+			e.log.Debugw("already written",
+				"blobID", blobID)
+			continue
+		}
+
+		if now := e.timeFunc(); !now.Before(cs.ValidUntil) {
+			e.log.Debugw("write was too slow, retrying",
+				"validUntil", cs.ValidUntil)
+			atomic.AddInt32(e.writeIndexTooSlow, 1)
+
+			return errWriteIndexTryAgain
+		}
+
+		if err := e.st.PutBlob(ctx, blobID, data, blob.PutOptions{}); err != nil {
+			return errors.Wrap(err, "error writing index blob")
+		}
+
+		bm, err := e.st.GetMetadata(ctx, blobID)
+		if err != nil {
+			return errors.Wrap(err, "error getting index metadata")
+		}
+
+		e.log.Debugw("wrote-index-shard",
+			"metadata", bm)
+
+		written[bm.BlobID] = bm
+	}
+
+	return nil
 }
 
 // Invalidate ensures that all cached index information is discarded.
@@ -789,21 +860,6 @@ func (e *Manager) getIndexesFromEpochInternal(ctx context.Context, cs CurrentSna
 		uncompactedBlobs = ue
 	}
 
-	// Ignore blobs written after the epoch has been settled.
-	//
-	// Epochs N is 'settled' after epoch N+2 has been started and that makes N subject to compaction,
-	// because at this point all clients will agree that we're in epoch N+1 or N+2.
-	//
-	// In a pathological case it's possible for client to write a blob for a 'settled' epoch if they:
-	//
-	// 1. determine current epoch number (N).
-	// 2. go to sleep for a very long time, enough for epoch >=N+2 to become current.
-	// 3. write blob for the epoch number N
-	uncompactedBlobs = blobsWrittenBefore(
-		uncompactedBlobs,
-		cs.EpochStartTime[epoch+numUnsettledEpochs],
-	)
-
 	if epochSettled {
 		// we're starting background work, ignore parent cancelation signal.
 		ctx = ctxutil.Detach(ctx)

internal/epoch/epoch_manager_test.go

@@ -2,7 +2,6 @@ package epoch
 
 import (
 	"context"
-	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"math/rand"
@@ -18,6 +17,7 @@ import (
 	"github.com/kopia/kopia/internal/blobtesting"
 	"github.com/kopia/kopia/internal/clock"
 	"github.com/kopia/kopia/internal/faketime"
+	"github.com/kopia/kopia/internal/fault"
 	"github.com/kopia/kopia/internal/gather"
 	"github.com/kopia/kopia/internal/testlogging"
 	"github.com/kopia/kopia/repo/blob"
@@ -180,6 +180,11 @@ func TestIndexEpochManager_Parallel(t *testing.T) {
 				if _, err := te2.mgr.WriteIndex(ctx, map[blob.ID]blob.Bytes{
 					blob.ID(fmt.Sprintf("w%vr%x", worker, rnd)): gather.FromSlice(ndx.Bytes()),
 				}); err != nil {
+					if errors.Is(err, ErrVerySlowIndexWrite) {
+						indexNum--
+						continue
+					}
+
 					return errors.Wrap(err, "error writing")
 				}
 
@@ -381,21 +386,75 @@ func TestGetCompleteIndexSetRetriesIfTookTooLong(t *testing.T) {
 	require.EqualValues(t, 1, *te.mgr.getCompleteIndexSetTooSlow)
 }
 
-func TestSlowWrite(t *testing.T) {
+func TestSlowWrite_RefreshesCurrentState(t *testing.T) {
 	te := newTestEnv(t)
 
 	ctx := testlogging.Context(t)
 
-	te.faultyStorage.AddFault(blobtesting.MethodPutBlob).
-		Repeat(10).
-		Before(func() {
-			te.ft.Advance(24 * time.Hour)
-		})
+	// on first write, advance time enough to lose current context, but not to go to the next epoch.
+	te.faultyStorage.AddFaults(blobtesting.MethodPutBlob,
+		fault.New().Before(func() { te.ft.Advance(1 * time.Hour) }))
 
-	te.mustWriteIndexFile(ctx, t, newFakeIndexWithEntries(1))
-	require.EqualValues(t, 11, *te.mgr.writeIndexTooSlow)
-	te.mustWriteIndexFile(ctx, t, newFakeIndexWithEntries(2))
-	te.verifyCompleteIndexSet(ctx, t, LatestEpoch, newFakeIndexWithEntries(1, 2), time.Time{})
+	te.mustWriteIndexFiles(ctx, t,
+		newFakeIndexWithEntries(1),
+		newFakeIndexWithEntries(2),
+		newFakeIndexWithEntries(3),
+	)
+	require.EqualValues(t, 1, *te.mgr.writeIndexTooSlow)
+	cs, err := te.mgr.Current(ctx)
+	require.NoError(t, err)
+	require.Equal(t, 0, cs.WriteEpoch)
+	te.verifyCompleteIndexSet(ctx, t, LatestEpoch, newFakeIndexWithEntries(1, 2, 3), time.Time{})
+}
+
+func TestSlowWrite_MovesToNextEpoch(t *testing.T) {
+	te := newTestEnv(t)
+
+	ctx := testlogging.Context(t)
+
+	// on first write, advance time enough to lose current context and go to the next epoch.
+	te.faultyStorage.AddFaults(blobtesting.MethodPutBlob,
+		fault.New().Before(func() {
+			te.ft.Advance(1 * time.Hour)
+			te.mgr.ForceAdvanceEpoch(ctx)
+		}),
+		fault.New().Before(func() { te.ft.Advance(1 * time.Hour) }))
+
+	te.mustWriteIndexFiles(ctx, t,
+		newFakeIndexWithEntries(1),
+		newFakeIndexWithEntries(2),
+		newFakeIndexWithEntries(3),
+	)
+	require.EqualValues(t, 1, *te.mgr.writeIndexTooSlow)
+	cs, err := te.mgr.Current(ctx)
+	require.NoError(t, err)
+	require.Equal(t, 1, cs.WriteEpoch)
+	te.verifyCompleteIndexSet(ctx, t, LatestEpoch, newFakeIndexWithEntries(1, 2, 3), time.Time{})
+}
+
+func TestSlowWrite_MovesToNextEpochTwice(t *testing.T) {
+	te := newTestEnv(t)
+
+	ctx := testlogging.Context(t)
+
+	// on first write, advance time enough to lose current context and go to the next epoch.
+	te.faultyStorage.AddFaults(blobtesting.MethodPutBlob,
+		fault.New().Before(func() {
+			te.ft.Advance(24 * time.Hour)
+		}),
+		fault.New().Before(func() {
+			te.mgr.ForceAdvanceEpoch(ctx)
+			te.mgr.ForceAdvanceEpoch(ctx)
+		}))
+
+	_, err := te.writeIndexFiles(ctx,
+		newFakeIndexWithEntries(1),
+		newFakeIndexWithEntries(2),
+		newFakeIndexWithEntries(3),
+	)
+
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "slow index write")
 }
 
 func TestForceAdvanceEpoch(t *testing.T) {
@@ -419,6 +478,51 @@ func TestForceAdvanceEpoch(t *testing.T) {
 	require.Equal(t, 2, cs.WriteEpoch)
 }
 
+func TestInvalid_WriteIndex(t *testing.T) {
+	te := newTestEnv(t)
+
+	ctx, cancel := context.WithCancel(testlogging.Context(t))
+	defer cancel()
+
+	// on first write, advance time enough to lose current context and go to the next epoch.
+	te.faultyStorage.AddFault(blobtesting.MethodListBlobs).Repeat(100).Before(cancel).ErrorInstead(errors.Errorf("canceled"))
+
+	_, err := te.writeIndexFiles(ctx,
+		newFakeIndexWithEntries(1),
+		newFakeIndexWithEntries(2),
+		newFakeIndexWithEntries(3),
+	)
+
+	require.ErrorIs(t, err, ctx.Err())
+}
+
+func TestInvalid_ForceAdvanceEpoch(t *testing.T) {
+	te := newTestEnv(t)
+
+	ctx, cancel := context.WithCancel(testlogging.Context(t))
+	defer cancel()
+
+	err := te.mgr.ForceAdvanceEpoch(ctx)
+	require.ErrorIs(t, err, ctx.Err())
+
+	ctx = testlogging.Context(t)
+	someError := errors.Errorf("failed")
+	te.faultyStorage.AddFault(blobtesting.MethodPutBlob).ErrorInstead(someError)
+
+	err = te.mgr.ForceAdvanceEpoch(ctx)
+	require.ErrorIs(t, err, someError)
+}
+
+func TestInvalid_Cleanup(t *testing.T) {
+	te := newTestEnv(t)
+
+	ctx, cancel := context.WithCancel(testlogging.Context(t))
+	cancel()
+
+	err := te.mgr.Cleanup(ctx)
+	require.ErrorIs(t, err, ctx.Err())
+}
+
 // nolint:thelper
 func verifySequentialWrites(t *testing.T, te *epochManagerTestEnv) {
 	ctx := testlogging.Context(t)
@@ -432,7 +536,7 @@ func verifySequentialWrites(t *testing.T, te *epochManagerTestEnv) {
 	for te.ft.NowFunc()().Before(endTime) {
 		indexNum++
 
-		te.mustWriteIndexFile(ctx, t, newFakeIndexWithEntries(indexNum))
+		te.mustWriteIndexFiles(ctx, t, newFakeIndexWithEntries(indexNum))
 
 		expected.Entries = append(expected.Entries, indexNum)
 		te.verifyCompleteIndexSet(ctx, t, LatestEpoch, expected, lastDeletionWatermark)
@@ -605,15 +709,28 @@ func (te *epochManagerTestEnv) getMergedIndexContents(ctx context.Context, blobI
 	return result, nil
 }
 
-func (te *epochManagerTestEnv) mustWriteIndexFile(ctx context.Context, t *testing.T, ndx *fakeIndex) {
+func (te *epochManagerTestEnv) writeIndexFiles(ctx context.Context, ndx ...*fakeIndex) ([]blob.Metadata, error) {
+	shards := map[blob.ID]blob.Bytes{}
+
+	var sessionID [8]byte
+
+	rand.Read(sessionID[:])
+
+	for _, n := range ndx {
+		var rnd [8]byte
+
+		rand.Read(rnd[:])
+
+		shards[blob.ID(fmt.Sprintf("%x-c%v-s%x", rnd[:], len(ndx), sessionID))] = gather.FromSlice(n.Bytes())
+	}
+
+	return te.mgr.WriteIndex(ctx, shards)
+}
+
+func (te *epochManagerTestEnv) mustWriteIndexFiles(ctx context.Context, t *testing.T, ndx ...*fakeIndex) {
 	t.Helper()
 
-	var rnd [8]byte
+	_, err := te.writeIndexFiles(ctx, ndx...)
 
-	rand.Read(rnd[:])
-
-	_, err := te.mgr.WriteIndex(ctx, map[blob.ID]blob.Bytes{
-		blob.ID(hex.EncodeToString(rnd[:])): gather.FromSlice(ndx.Bytes()),
-	})
 	require.NoError(t, err)
 }

internal/epoch/epoch_utils.go

@@ -53,20 +53,6 @@ func epochRangeFromBlobID(blobID blob.ID) (min, max int, ok bool) {
 	return n1, n2, err1 == nil && err2 == nil
 }
 
-func blobsWrittenBefore(bms []blob.Metadata, maxTime time.Time) []blob.Metadata {
-	var result []blob.Metadata
-
-	for _, bm := range bms {
-		if !maxTime.IsZero() && bm.Timestamp.After(maxTime) {
-			continue
-		}
-
-		result = append(result, bm)
-	}
-
-	return result
-}
-
 func groupByEpochNumber(bms []blob.Metadata) map[int][]blob.Metadata {
 	result := map[int][]blob.Metadata{}
 

internal/epoch/epoch_utils_test.go

@@ -2,7 +2,6 @@ package epoch
 
 import (
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -43,35 +42,6 @@ func TestEpochNumberFromBlobID_Invalid(t *testing.T) {
 	}
 }
 
-func TestBlobsWrittenBefore(t *testing.T) {
-	t0 := time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC)
-	bm0 := blob.Metadata{BlobID: "a", Timestamp: t0}
-
-	t1 := time.Date(2000, 1, 2, 0, 0, 0, 0, time.UTC)
-	bm1 := blob.Metadata{BlobID: "b", Timestamp: t1}
-
-	t2 := time.Date(2000, 1, 3, 0, 0, 0, 0, time.UTC)
-	bm2 := blob.Metadata{BlobID: "c", Timestamp: t2}
-
-	cases := []struct {
-		bms    []blob.Metadata
-		cutoff time.Time
-		want   []blob.Metadata
-	}{
-		{[]blob.Metadata{bm0, bm1, bm2}, time.Time{}, []blob.Metadata{bm0, bm1, bm2}},
-		{[]blob.Metadata{bm0, bm1, bm2}, t0.Add(-1), nil},
-		{[]blob.Metadata{bm0, bm1, bm2}, t0, []blob.Metadata{bm0}},
-		{[]blob.Metadata{bm0, bm1, bm2}, t1.Add(-1), []blob.Metadata{bm0}},
-		{[]blob.Metadata{bm0, bm1, bm2}, t1, []blob.Metadata{bm0, bm1}},
-		{[]blob.Metadata{bm0, bm1, bm2}, t2.Add(-1), []blob.Metadata{bm0, bm1}},
-		{[]blob.Metadata{bm0, bm1, bm2}, t2, []blob.Metadata{bm0, bm1, bm2}},
-	}
-
-	for i, tc := range cases {
-		require.Equal(t, tc.want, blobsWrittenBefore(tc.bms, tc.cutoff), "#%v blobsWrittenBefore(%v,%v)", i, tc.bms, tc.cutoff)
-	}
-}
-
 func TestGroupByEpochNumber(t *testing.T) {
 	cases := []struct {
 		input []blob.Metadata

repo/content/content_manager_test.go

@@ -1269,10 +1269,10 @@ func (s *contentManagerSuite) TestHandleWriteErrors(t *testing.T) {
 				cids = append(cids, cid)
 			}
 			if got, want := flushWithRetries(ctx, t, bm), tc.expectedFlushRetries; got != want {
-				t.Errorf("invalid # of flush retries %v, wanted %v", got, want)
+				t.Fatalf("invalid # of flush retries %v, wanted %v", got, want)
 			}
 			if diff := cmp.Diff(writeRetries, tc.expectedWriteRetries); diff != "" {
-				t.Errorf("invalid # of write retries (-got,+want): %v", diff)
+				t.Fatalf("invalid # of write retries (-got,+want): %v", diff)
 			}
 
 			bm2 := s.newTestContentManagerWithTweaks(t, st, nil)
@@ -2179,7 +2179,7 @@ func flushWithRetries(ctx context.Context, t *testing.T, bm *WriteManager) int {
 	}
 
 	if err != nil {
-		t.Errorf("err: %v", err)
+		t.Fatalf("err: %v", err)
 	}
 
 	return retryCount