Fix endurance test (#1254)

cli/command_content_verify.go

@@ -94,6 +94,8 @@ func (c *commandContentVerify) run(ctx context.Context, rep repo.DirectRepositor
 
 	log(ctx).Infof("Verifying all contents...")
 
+	rep.DisableIndexRefresh()
+
 	throttle := new(timetrack.Throttle)
 	est := timetrack.Start()
 

cli/command_snapshot_verify.go

@@ -202,6 +202,10 @@ func (c *commandSnapshotVerify) run(ctx context.Context, rep repo.Repository) er
 		log(ctx).Errorf("DEPRECATED: --all-sources flag has no effect and is the default when no sources are provided.")
 	}
 
+	if dr, ok := rep.(repo.DirectRepositoryWriter); ok {
+		dr.DisableIndexRefresh()
+	}
+
 	v := &verifier{
 		rep:                  rep,
 		tt:                   timetrack.Start(),

internal/gather/gather_write_buffer_chunk.go

@@ -78,7 +78,7 @@ func (a *chunkAllocator) dumpStats(ctx context.Context, prefix string) {
 	a.mu.Lock()
 	defer a.mu.Unlock()
 
-	log(ctx).Infof("%v (%v) - allocated %v chunks freed %v alive %v max %v free list high water mark: %v",
+	log(ctx).Debugf("%v (%v) - allocated %v chunks freed %v alive %v max %v free list high water mark: %v",
 		prefix,
 		units.Base2Bytes(int64(a.chunkSize)),
 		a.allocated, a.freed, a.allocated-a.freed, a.allocHighWaterMark, a.freeListHighWaterMark)

repo/content/committed_read_manager.go

@@ -494,16 +494,6 @@ func (sm *SharedManager) shouldRefreshIndexes() bool {
 	return sm.timeNow().After(sm.refreshIndexesAfter)
 }
 
-func (sm *SharedManager) maybeRefreshIndexes(ctx context.Context) error {
-	if sm.shouldRefreshIndexes() {
-		if err := sm.Refresh(ctx); err != nil {
-			return errors.Wrap(err, "error refreshing indexes")
-		}
-	}
-
-	return nil
-}
-
 // NewSharedManager returns SharedManager that is used by SessionWriteManagers on top of a repository.
 func NewSharedManager(ctx context.Context, st blob.Storage, f *FormattingOptions, caching *CachingOptions, opts *ManagerOptions) (*SharedManager, error) {
 	opts = opts.CloneOrDefault()

repo/content/content_manager.go

@@ -76,7 +76,8 @@ type IndexBlobInfo struct {
 
 // WriteManager builds content-addressable storage with encryption, deduplication and packaging on top of BLOB store.
 type WriteManager struct {
-	revision int64 // changes on each local write
+	revision            int64 // changes on each local write
+	disableIndexRefresh int32
 
 	mu       *sync.RWMutex
 	cond     *sync.Cond
@@ -162,6 +163,16 @@ func (bm *WriteManager) DeleteContent(ctx context.Context, contentID ID) error {
 	return bm.deletePreexistingContent(ctx, bi)
 }
 
+func (bm *WriteManager) maybeRefreshIndexes(ctx context.Context) error {
+	if atomic.LoadInt32(&bm.disableIndexRefresh) == 0 && bm.shouldRefreshIndexes() {
+		if err := bm.Refresh(ctx); err != nil {
+			return errors.Wrap(err, "error refreshing indexes")
+		}
+	}
+
+	return nil
+}
+
 // Intentionally passing bi by value.
 func (bm *WriteManager) deletePreexistingContent(ctx context.Context, ci Info) error {
 	if ci.GetDeleted() {
@@ -773,6 +784,11 @@ func (bm *WriteManager) ContentInfo(ctx context.Context, contentID ID) (Info, er
 	return bi, err
 }
 
+// DisableIndexRefresh disables index refresh for the remainder of this session.
+func (bm *WriteManager) DisableIndexRefresh() {
+	atomic.StoreInt32(&bm.disableIndexRefresh, 1)
+}
+
 func (bm *WriteManager) lock() {
 	bm.mu.Lock()
 }

repo/maintenance/blob_gc.go

@@ -2,6 +2,7 @@
 
 import (
 	"context"
+	"time"
 
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
@@ -15,13 +16,14 @@
 
 // DeleteUnreferencedBlobsOptions provides option for blob garbage collection algorithm.
 type DeleteUnreferencedBlobsOptions struct {
-	Parallel int
-	Prefix   blob.ID
-	DryRun   bool
+	Parallel     int
+	Prefix       blob.ID
+	DryRun       bool
+	NotAfterTime time.Time
 }
 
-// DeleteUnreferencedBlobs deletes old blobs that are no longer referenced by index entries.
-// nolint:gocyclo
+// DeleteUnreferencedBlobs deletes o was created after maintenance startederenced by index entries.
+// nolint:gocyclo,funlen
 func DeleteUnreferencedBlobs(ctx context.Context, rep repo.DirectRepositoryWriter, opt DeleteUnreferencedBlobsOptions, safety SafetyParameters) (int, error) {
 	if opt.Parallel == 0 {
 		opt.Parallel = 16
@@ -69,17 +71,34 @@ func DeleteUnreferencedBlobs(ctx context.Context, rep repo.DirectRepositoryWrite
 		return 0, errors.Wrap(err, "unable to load active sessions")
 	}
 
+	cutoffTime := opt.NotAfterTime
+	if cutoffTime.IsZero() {
+		cutoffTime = rep.Time()
+	}
+
+	// move the cutoff time a bit forward, because on Windows clock does not reliably move forward so we may end
+	// up not deleting some blobs - this only really affects tests, since BlobDeleteMinAge provides real
+	// protection here.
+	const cutoffTimeSlack = 1 * time.Second
+
+	cutoffTime = cutoffTime.Add(cutoffTimeSlack)
+
 	// iterate all pack blobs + session blobs and keep ones that are too young or
 	// belong to alive sessions.
 	if err := rep.ContentManager().IterateUnreferencedBlobs(ctx, prefixes, opt.Parallel, func(bm blob.Metadata) error {
-		if age := rep.Time().Sub(bm.Timestamp); age < safety.BlobDeleteMinAge {
+		if bm.Timestamp.After(cutoffTime) {
+			log(ctx).Debugf("  preserving %v because it was created after maintenance started", bm.BlobID)
+			return nil
+		}
+
+		if age := cutoffTime.Sub(bm.Timestamp); age < safety.BlobDeleteMinAge {
 			log(ctx).Debugf("  preserving %v because it's too new (age: %v<%v)", bm.BlobID, age, safety.BlobDeleteMinAge)
 			return nil
 		}
 
 		sid := content.SessionIDFromBlobID(bm.BlobID)
 		if s, ok := activeSessions[sid]; ok {
-			if age := rep.Time().Sub(s.CheckpointTime); age < safety.SessionExpirationAge {
+			if age := cutoffTime.Sub(s.CheckpointTime); age < safety.SessionExpirationAge {
 				log(ctx).Debugf("  preserving %v because it's part of an active session (%v)", bm.BlobID, sid)
 				return nil
 			}

repo/maintenance/maintenance_run.go

@@ -11,7 +11,6 @@
 
 	"github.com/kopia/kopia/internal/clock"
 	"github.com/kopia/kopia/repo"
-	"github.com/kopia/kopia/repo/blob"
 	"github.com/kopia/kopia/repo/content"
 	"github.com/kopia/kopia/repo/logging"
 )
@@ -124,6 +123,9 @@ type RunParameters struct {
 	Mode Mode
 
 	Params *Params
+
+	// timestamp of the last update of maintenance schedule blob
+	MaintenanceStartTime time.Time
 }
 
 // NotOwnedError is returned when maintenance cannot run because it is owned by another user.
@@ -139,6 +141,8 @@ func (e NotOwnedError) Error() string {
 // lock can be acquired. Lock is passed to the function, which ensures that every call to Run()
 // is within the exclusive context.
 func RunExclusive(ctx context.Context, rep repo.DirectRepositoryWriter, mode Mode, force bool, cb func(runParams RunParameters) error) error {
+	rep.DisableIndexRefresh()
+
 	p, err := GetParams(ctx, rep)
 	if err != nil {
 		return errors.Wrap(err, "unable to get maintenance params")
@@ -160,7 +164,7 @@ func RunExclusive(ctx context.Context, rep repo.DirectRepositoryWriter, mode Mod
 		return nil
 	}
 
-	runParams := RunParameters{rep, mode, p}
+	runParams := RunParameters{rep, mode, p, time.Time{}}
 
 	// update schedule so that we don't run the maintenance again immediately if
 	// this process crashes.
@@ -168,7 +172,14 @@ func RunExclusive(ctx context.Context, rep repo.DirectRepositoryWriter, mode Mod
 		return errors.Wrap(err, "error updating maintenance schedule")
 	}
 
-	if err = ensureNoClockSkew(ctx, runParams.rep.BlobReader(), runParams.rep.Time()); err != nil {
+	bm, err := runParams.rep.BlobReader().GetMetadata(ctx, maintenanceScheduleBlobID)
+	if err != nil {
+		return errors.Wrap(err, "error getting maintenance blob time")
+	}
+
+	runParams.MaintenanceStartTime = bm.Timestamp
+
+	if err = ensureNoClockSkew(runParams); err != nil {
 		return errors.Wrap(err, "error checking for clock skew")
 	}
 
@@ -200,21 +211,17 @@ func RunExclusive(ctx context.Context, rep repo.DirectRepositoryWriter, mode Mod
 	return cb(runParams)
 }
 
-func ensureNoClockSkew(ctx context.Context, st blob.Reader, now time.Time) error {
-	bm, err := st.GetMetadata(ctx, maintenanceScheduleBlobID)
-	if err != nil {
-		return errors.Wrap(err, "error getting maintenance blob time")
-	}
+func ensureNoClockSkew(rp RunParameters) error {
+	localTime := rp.rep.Time()
+	repoTime := rp.MaintenanceStartTime
 
-	repoTime := bm.Timestamp
-
-	clockSkew := bm.Timestamp.Sub(now)
+	clockSkew := repoTime.Sub(localTime)
 	if clockSkew < 0 {
 		clockSkew = -clockSkew
 	}
 
 	if clockSkew > maxClockSkew {
-		return errors.Errorf("Clock skew detected: local clock is out of sync with repository timestamp by more than allowed %v (local: %v repository: %v). Refusing to run maintenance.", maxClockSkew, now, repoTime)
+		return errors.Errorf("Clock skew detected: local clock is out of sync with repository timestamp by more than allowed %v (local: %v repository: %v). Refusing to run maintenance.", maxClockSkew, localTime, repoTime)
 	}
 
 	return nil
@@ -357,7 +364,9 @@ func runTaskRewriteContentsFull(ctx context.Context, runParams RunParameters, s
 
 func runTaskDeleteOrphanedBlobsFull(ctx context.Context, runParams RunParameters, s *Schedule, safety SafetyParameters) error {
 	return ReportRun(ctx, runParams.rep, TaskDeleteOrphanedBlobsFull, s, func() error {
-		_, err := DeleteUnreferencedBlobs(ctx, runParams.rep, DeleteUnreferencedBlobsOptions{}, safety)
+		_, err := DeleteUnreferencedBlobs(ctx, runParams.rep, DeleteUnreferencedBlobsOptions{
+			NotAfterTime: runParams.MaintenanceStartTime,
+		}, safety)
 		return err
 	})
 }
@@ -365,7 +374,8 @@ func runTaskDeleteOrphanedBlobsFull(ctx context.Context, runParams RunParameters
 func runTaskDeleteOrphanedBlobsQuick(ctx context.Context, runParams RunParameters, s *Schedule, safety SafetyParameters) error {
 	return ReportRun(ctx, runParams.rep, TaskDeleteOrphanedBlobsQuick, s, func() error {
 		_, err := DeleteUnreferencedBlobs(ctx, runParams.rep, DeleteUnreferencedBlobsOptions{
-			Prefix: content.PackBlobIDPrefixSpecial,
+			NotAfterTime: runParams.MaintenanceStartTime,
+			Prefix:       content.PackBlobIDPrefixSpecial,
 		}, safety)
 		return err
 	})

repo/repository.go

@@ -64,6 +64,8 @@ type DirectRepository interface {
 	ConfigFilename() string
 	DeriveKey(purpose []byte, keyLength int) []byte
 	Token(password string) (string, error)
+
+	DisableIndexRefresh()
 }
 
 // DirectRepositoryWriter provides low-level write access to the repository.
@@ -143,6 +145,11 @@ func (r *directRepository) NewObjectWriter(ctx context.Context, opt object.Write
 	return r.omgr.NewWriter(ctx, opt)
 }
 
+// DisableIndexRefresh disables index refresh for the duration of the write session.
+func (r *directRepository) DisableIndexRefresh() {
+	r.cmgr.DisableIndexRefresh()
+}
+
 // OpenObject opens the reader for a given object, returns object.ErrNotFound.
 func (r *directRepository) OpenObject(ctx context.Context, id object.ID) (object.Reader, error) {
 	// nolint:wrapcheck