From 331c7f6298fc0930484fa06a98d72ba7b8e5db16 Mon Sep 17 00:00:00 2001 From: Jarek Kowalski Date: Wed, 15 Dec 2021 07:06:28 -0800 Subject: [PATCH] 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 | 156 ++++++++++++++++++--------- internal/epoch/epoch_manager_test.go | 155 ++++++++++++++++++++++---- internal/epoch/epoch_utils.go | 14 --- internal/epoch/epoch_utils_test.go | 30 ------ repo/content/content_manager_test.go | 6 +- 5 files changed, 245 insertions(+), 116 deletions(-) diff --git a/internal/epoch/epoch_manager.go b/internal/epoch/epoch_manager.go index a9d663331..e1a273acc 100644 --- a/internal/epoch/epoch_manager.go +++ b/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) diff --git a/internal/epoch/epoch_manager_test.go b/internal/epoch/epoch_manager_test.go index 5fea85e33..a0ada90ea 100644 --- a/internal/epoch/epoch_manager_test.go +++ b/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) } diff --git a/internal/epoch/epoch_utils.go b/internal/epoch/epoch_utils.go index ea463f1c6..3f1c6cb1e 100644 --- a/internal/epoch/epoch_utils.go +++ b/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{} diff --git a/internal/epoch/epoch_utils_test.go b/internal/epoch/epoch_utils_test.go index 50c31361a..1d2bdf370 100644 --- a/internal/epoch/epoch_utils_test.go +++ b/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 diff --git a/repo/content/content_manager_test.go b/repo/content/content_manager_test.go index 70c5ccf82..5f7f3495e 100644 --- a/repo/content/content_manager_test.go +++ b/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