Wire up setting min epoch duration (#4799)

Add the required wiring for SDK consumers to set the min epoch duration
(and eventually other persistent config info). This is just boiler-plate
with some tests at the various layers to make sure that layer is run as
expected. More nuanced tests about setting various config values are
left to the code in `internal/kopia/conn_test.go`

---

#### Does this PR need a docs update or release note?

- [ ] ✅ Yes, it's included
- [ ] 🕐 Yes, but in a later PR
- [x] ⛔ No

#### Type of change

- [x] 🌻 Feature
- [ ] 🐛 Bugfix
- [ ] 🗺️ Documentation
- [ ] 🤖 Supportability/Tests
- [ ] 💻 CI/Deployment
- [ ] 🧹 Tech Debt/Cleanup

#### Issue(s)

* #4782

#### Test Plan

- [ ] 💪 Manual
- [x] ⚡ Unit test
- [ ] 💚 E2E

src/internal/kopia/wrapper.go

@@ -739,3 +739,10 @@ func (w *Wrapper) SetRetentionParameters(
 ) error {
 	return clues.Stack(w.c.setRetentionParameters(ctx, retention)).OrNil()
 }
+
+func (w *Wrapper) UpdatePersistentConfig(
+	ctx context.Context,
+	config repository.PersistentConfig,
+) error {
+	return clues.Stack(w.c.updatePersistentConfig(ctx, config)).OrNil()
+}

src/internal/kopia/wrapper_test.go

@@ -336,6 +336,45 @@ func (suite *BasicKopiaIntegrationSuite) TestSetRetentionParameters_NoChangesOnF
 		assert.False)
 }
 
+func (suite *BasicKopiaIntegrationSuite) TestUpdatePersistentConfig() {
+	t := suite.T()
+	repoNameHash := strTD.NewHashForRepoConfigName()
+
+	ctx, flush := tester.NewContext(t)
+	defer flush()
+
+	k, err := openLocalKopiaRepo(t, ctx)
+	require.NoError(t, err, clues.ToCore(err))
+
+	config := repository.PersistentConfig{
+		MinEpochDuration: ptr.To(8 * time.Hour),
+	}
+	w := &Wrapper{k}
+
+	err = w.UpdatePersistentConfig(ctx, config)
+	require.NoError(t, err, clues.ToCore(err))
+
+	// Close and reopen the repo to make sure it's the same.
+	err = w.Close(ctx)
+	require.NoError(t, err, clues.ToCore(err))
+
+	k.Close(ctx)
+	require.NoError(t, err, clues.ToCore(err))
+
+	err = k.Connect(ctx, repository.Options{}, repoNameHash)
+	require.NoError(t, err, clues.ToCore(err))
+
+	defer k.Close(ctx)
+
+	mutableParams, _, err := k.getPersistentConfig(ctx)
+	require.NoError(t, err, clues.ToCore(err))
+
+	assert.Equal(
+		t,
+		ptr.Val(config.MinEpochDuration),
+		mutableParams.EpochParameters.MinEpochDuration)
+}
+
 // ---------------
 // integration tests that require object locking to be enabled on the bucket.
 // ---------------

src/internal/operations/persistent_config.go

@@ -0,0 +1,78 @@
+package operations
+
+import (
+	"context"
+	"time"
+
+	"github.com/alcionai/clues"
+
+	"github.com/alcionai/corso/src/internal/common/crash"
+	"github.com/alcionai/corso/src/internal/events"
+	"github.com/alcionai/corso/src/internal/kopia"
+	"github.com/alcionai/corso/src/internal/stats"
+	"github.com/alcionai/corso/src/pkg/control"
+	"github.com/alcionai/corso/src/pkg/control/repository"
+	"github.com/alcionai/corso/src/pkg/count"
+)
+
+// PersistentConfig wraps an operation that deals with repo configuration.
+type PersistentConfigOperation struct {
+	operation
+	Results    PersistentConfigResults
+	configOpts repository.PersistentConfig
+}
+
+// PersistentConfigResults aggregate the details of the results of the operation.
+type PersistentConfigResults struct {
+	stats.StartAndEndTime
+}
+
+// NewPersistentConfigOperation constructs and validates an operation to change
+// various persistent config parameters like the minimum epoch duration for the
+// kopia index.
+func NewPersistentConfigOperation(
+	ctx context.Context,
+	opts control.Options,
+	kw *kopia.Wrapper,
+	configOpts repository.PersistentConfig,
+	bus events.Eventer,
+) (PersistentConfigOperation, error) {
+	op := PersistentConfigOperation{
+		operation:  newOperation(opts, bus, count.New(), kw, nil),
+		configOpts: configOpts,
+	}
+
+	// Don't run validation because we don't populate the model store.
+
+	return op, nil
+}
+
+func (op *PersistentConfigOperation) Run(ctx context.Context) (err error) {
+	defer func() {
+		if crErr := crash.Recovery(ctx, recover(), "persistent_config"); crErr != nil {
+			err = crErr
+		}
+	}()
+
+	// TODO(ashmrtn): Send telemetry?
+
+	return op.do(ctx)
+}
+
+func (op *PersistentConfigOperation) do(ctx context.Context) error {
+	op.Results.StartedAt = time.Now()
+
+	defer func() {
+		op.Results.CompletedAt = time.Now()
+	}()
+
+	err := op.operation.kopia.UpdatePersistentConfig(ctx, op.configOpts)
+	if err != nil {
+		op.Status = Failed
+		return clues.Wrap(err, "running update persistent config operation")
+	}
+
+	op.Status = Completed
+
+	return nil
+}

src/internal/operations/persistent_config_test.go

@@ -0,0 +1,77 @@
+package operations
+
+import (
+	"testing"
+	"time"
+
+	"github.com/alcionai/clues"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+
+	"github.com/alcionai/corso/src/internal/common/ptr"
+	strTD "github.com/alcionai/corso/src/internal/common/str/testdata"
+	evmock "github.com/alcionai/corso/src/internal/events/mock"
+	"github.com/alcionai/corso/src/internal/kopia"
+	"github.com/alcionai/corso/src/internal/tester"
+	"github.com/alcionai/corso/src/pkg/control"
+	"github.com/alcionai/corso/src/pkg/control/repository"
+	storeTD "github.com/alcionai/corso/src/pkg/storage/testdata"
+)
+
+type PersistentConfigOpIntegrationSuite struct {
+	tester.Suite
+}
+
+func TestPersistentConfigOpIntegrationSuite(t *testing.T) {
+	suite.Run(t, &PersistentConfigOpIntegrationSuite{
+		Suite: tester.NewIntegrationSuite(
+			t,
+			[][]string{storeTD.AWSStorageCredEnvs}),
+	})
+}
+
+func (suite *PersistentConfigOpIntegrationSuite) TestRepoPersistentConfig() {
+	var (
+		t = suite.T()
+		// need to initialize the repository before we can test connecting to it.
+		st           = storeTD.NewPrefixedS3Storage(t)
+		k            = kopia.NewConn(st)
+		repoNameHash = strTD.NewHashForRepoConfigName()
+	)
+
+	ctx, flush := tester.NewContext(t)
+	defer flush()
+
+	err := k.Initialize(ctx, repository.Options{}, repository.Retention{}, repoNameHash)
+	require.NoError(t, err, clues.ToCore(err))
+
+	kw, err := kopia.NewWrapper(k)
+	// kopiaRef comes with a count of 1 and Wrapper bumps it again so safe
+	// to close here.
+	k.Close(ctx)
+
+	require.NoError(t, err, clues.ToCore(err))
+
+	defer kw.Close(ctx)
+
+	// Only set extend locks parameter as other retention options require a bucket
+	// with object locking enabled. There's more complete tests in the kopia
+	// package.
+	rco, err := NewPersistentConfigOperation(
+		ctx,
+		control.DefaultOptions(),
+		kw,
+		repository.PersistentConfig{
+			MinEpochDuration: ptr.To(8 * time.Hour),
+		},
+		evmock.NewBus())
+	require.NoError(t, err, clues.ToCore(err))
+
+	err = rco.Run(ctx)
+	assert.NoError(t, err, clues.ToCore(err))
+	assert.Equal(t, Completed, rco.Status)
+	assert.NotZero(t, rco.Results.StartedAt)
+	assert.NotZero(t, rco.Results.CompletedAt)
+	assert.NotEqual(t, rco.Results.StartedAt, rco.Results.CompletedAt)
+}

src/pkg/repository/repository.go

@@ -60,6 +60,10 @@ type Repositoryer interface {
 		ctx context.Context,
 		rcOpts ctrlRepo.Retention,
 	) (operations.RetentionConfigOperation, error)
+	NewPersistentConfig(
+		ctx context.Context,
+		configOpts ctrlRepo.PersistentConfig,
+	) (operations.PersistentConfigOperation, error)
 
 	Counter() *count.Bus
 }
@@ -282,6 +286,18 @@ func (r repository) NewRetentionConfig(
 		r.Bus)
 }
 
+func (r repository) NewPersistentConfig(
+	ctx context.Context,
+	configOpts ctrlRepo.PersistentConfig,
+) (operations.PersistentConfigOperation, error) {
+	return operations.NewPersistentConfigOperation(
+		ctx,
+		r.Opts,
+		r.dataLayer,
+		configOpts,
+		r.Bus)
+}
+
 func (r repository) Counter() *count.Bus {
 	return r.counter
 }