refactor(general): allow disabling writes on index index loads (#3645)

Conditionally disables epoch index maintenance operations when loading indexes.
This prevents (potentially expensive) cleanup write operations on the index
read path.
The behavior is controlled via the `epoch.Manager.allowCleanupWritesOnIndexLoad`
field, which can be temporarily overridden via an environment variable.
This override mechanism will be removed in the near future.

Refs:

- #3174
- #3224
- #3225
- #3638
- #3639

internal/epoch/epoch_manager.go

@@ -6,6 +6,8 @@
 import (
 	"context"
 	"fmt"
+	"os"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -186,6 +188,8 @@ type Manager struct {
 	log      logging.Logger
 	timeFunc func() time.Time
 
+	allowCleanupWritesOnIndexLoad bool
+
 	// wait group that waits for all compaction and cleanup goroutines.
 	backgroundWork sync.WaitGroup
 
@@ -485,7 +489,7 @@ func (e *Manager) maybeCompactAndCleanupLocked(ctx context.Context, p *Parameter
 // These index cleanup operations are disabled when using read-only storage
 // since they will fail when they try to mutate the underlying storage.
 func (e *Manager) allowWritesOnLoad() bool {
-	return !e.st.IsReadOnly()
+	return e.allowCleanupWritesOnIndexLoad && !e.st.IsReadOnly()
 }
 
 func (e *Manager) loadWriteEpoch(ctx context.Context, cs *CurrentSnapshot) error {
@@ -1015,16 +1019,28 @@ func rangeCheckpointBlobPrefix(epoch1, epoch2 int) blob.ID {
 	return blob.ID(fmt.Sprintf("%v%v_%v_", RangeCheckpointIndexBlobPrefix, epoch1, epoch2))
 }
 
+func allowWritesOnIndexLoad() bool {
+	v := strings.ToLower(os.Getenv("KOPIA_ALLOW_WRITE_ON_INDEX_LOAD"))
+
+	if v == "" {
+		// temporary default to be changed once index cleanup is performed on maintenance
+		return true
+	}
+
+	return v == "true" || v == "1"
+}
+
 // NewManager creates new epoch manager.
 func NewManager(st blob.Storage, paramProvider ParametersProvider, compactor CompactionFunc, log logging.Logger, timeNow func() time.Time) *Manager {
 	return &Manager{
-		st:                           st,
-		log:                          log,
-		compact:                      compactor,
-		timeFunc:                     timeNow,
-		paramProvider:                paramProvider,
-		getCompleteIndexSetTooSlow:   new(int32),
-		committedStateRefreshTooSlow: new(int32),
-		writeIndexTooSlow:            new(int32),
+		st:                            st,
+		log:                           log,
+		compact:                       compactor,
+		timeFunc:                      timeNow,
+		paramProvider:                 paramProvider,
+		allowCleanupWritesOnIndexLoad: allowWritesOnIndexLoad(),
+		getCompleteIndexSetTooSlow:    new(int32),
+		committedStateRefreshTooSlow:  new(int32),
+		writeIndexTooSlow:             new(int32),
 	}
 }